improvement(logs): move per-block progress markers to Redis to cut write amplification (#5248)

* improvement(logs): move per-block progress markers to Redis to cut write amplification

Per-block lastStartedBlock/lastCompletedBlock markers were persisted via a
jsonb_set UPDATE on workflow_execution_logs on every block start and complete
(~2N UPDATEs per run) — the heaviest write query in the DB. These are live
progress breadcrumbs with no DB-polling consumer (live progress comes from the
executor over WebSocket); their only durable value is a breadcrumb folded into
the final record.

Behind the redis-progress-markers flag, markers now live in Redis during the run
and are folded into the single terminal UPDATE at completion, dropping per-run
row UPDATEs from ~2N+1 to 1.

- New progress-markers module: HASH execution:progress:{id}, atomic Lua
  monotonic-guard writes preserving the existing <= ordering, reservation-aligned
  TTL backstop, graceful no-op when Redis is unavailable
- Deterministic GC: cleared at every terminal/pause boundary; TTL covers crashes
- Flag resolved once per logging session so a run never mixes write paths
- Fold markers into the completion record (Redis wins, falls back to row markers)
- Merge live markers for in-flight detail reads
- Extract shared getExecutionReservationTtlMs so marker and admission-slot TTLs
  share one source of truth

* fix(logs): SQL fallback when Redis marker write fails, fold markers on force-fail, validate marker shape

Addresses review feedback on the redis-progress-markers PR:
- persistLast* now falls back to the jsonb_set UPDATE when Redis is unavailable or the write fails (setLast* returns whether it persisted), so a marker is never dropped when the flag is on without a healthy Redis.
- markExecutionAsFailed folds live Redis markers into execution_data before clearing, so the last-started/last-completed breadcrumb survives the force-fail path.
- getProgressMarkers validates marker shape (rebuilds from typed fields), so a stale or wrong-shaped Redis value can never reach API consumers.

* chore(logs): convert inline marker comments to TSDoc

* fix(logs): preserve markers when the completion read fails

getProgressMarkers now returns null on a Redis read error (vs {} for genuinely empty). completeWorkflowExecution and markExecutionAsFailed skip clearProgressMarkers when the read returns null, so a transient read error at completion no longer wipes markers that are still durably in Redis — the TTL reclaims them instead.

* fix(logs): resolve marker store split-brain by latest-timestamp-wins + drain on force-fail

- When a Redis marker write falls back to SQL, Redis and the row can each hold a marker for a different block; reads/folds previously preferred Redis unconditionally and could pick a stale value. Now the completion fold, the in-flight detail read, and the force-fail fold all pick the marker with the later timestamp (pickLatestStartedMarker/pickLatestCompletedMarker; markExecutionAsFailed uses a monotonic SQL guard).
- markAsFailed now drains pending per-block marker writes (not just the completion promise) before folding, so a force-fail racing onBlockStart/onBlockComplete still captures the latest breadcrumb.

* fix(logs): harden Lua marker guard against non-table decoded values

Guard the monotonic-check index with type(decoded) == 'table' so a corrupted Redis field that decodes to a non-table (e.g. a number) can't error the eval; our write path only ever stores JSON objects, so this is defense-in-depth.

* perf(logs): skip completion Redis read/clear when markers went to SQL

completeWorkflowExecution now takes readProgressMarkers (the session's resolved marker mode); when the flag is off it skips the per-completion HGETALL+DEL entirely instead of probing a key that was never written. Sticky to the session so it stays flip-safe (an execution that wrote to Redis always folds+clears Redis). Non-session callers default to true (safe read-and-fold). Also hardened the Lua guard with type(decoded)=='table'.

Author: waleedlatif1

apps/sim/lib/billing/calculations/usage-reservation.ts

@@ -5,7 +5,7 @@ import { getPlanTypeForLimits } from '@/lib/billing/plan-helpers'
 import { isOrgScopedSubscription } from '@/lib/billing/subscriptions/utils'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { getRedisClient } from '@/lib/core/config/redis'
-import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
+import { getExecutionReservationTtlMs } from '@/lib/core/execution-limits'
 import type { SubscriptionPlan } from '@/lib/core/rate-limiter/types'
 
 const logger = createLogger('UsageReservation')
@@ -41,9 +41,6 @@ const MAX_CONCURRENT_EXECUTIONS: Record<SubscriptionPlan, number> = {
  */
 const SLOT_COST_ESTIMATE = BASE_EXECUTION_CHARGE
 
-/** Safety buffer added to the reservation TTL beyond the max execution timeout. */
-const RESERVATION_TTL_BUFFER_MS = 60_000
-
 const INFLIGHT_KEY_PREFIX = 'usage:inflight:'
 const POINTER_KEY_PREFIX = 'usage:reservation:'
 
@@ -135,7 +132,7 @@ export async function reserveExecutionSlot(
   const maxConcurrency = getMaxConcurrentExecutions(subscription?.plan)
   const headroom = Math.max(0, limit - currentUsage)
   const headroomSlots = Math.floor(headroom / SLOT_COST_ESTIMATE)
-  const ttlMs = getMaxExecutionTimeout() + RESERVATION_TTL_BUFFER_MS
+  const ttlMs = getExecutionReservationTtlMs()
   const now = Date.now()
   const expiryScore = now + ttlMs
 

apps/sim/lib/core/config/env.ts

@@ -77,6 +77,7 @@ export const env = createEnv({
     TABLE_SNAPSHOT_CACHE:                  z.boolean().optional(),                 // Mount tables into sandboxes by reference via a version-keyed CSV snapshot in object storage instead of draining the whole table into web-process heap
     PII_REDACTION:                         z.boolean().optional(),                 // Redact PII from workflow logs via configurable Data Retention rules (Presidio at the logger persist choke point) and expose the Data Retention config UI
     TRIGGER_EU_REGION:                     z.boolean().optional(),                 // Route Trigger.dev runs to eu-central-1 instead of the default us-east-1 (fallback for the trigger-eu-region flag when AppConfig is not the source of truth)
+    REDIS_PROGRESS_MARKERS:                z.boolean().optional(),                 // Write per-block live progress markers to Redis instead of jsonb_set UPDATEs on workflow_execution_logs (fallback for the redis-progress-markers flag when AppConfig is not the source of truth)
 
     // Table feature limits (per plan). Apply when billing is disabled (free tier defaults) or for billed plans.
     FREE_TABLES_LIMIT:                     z.number().optional(),                  // Max user tables per workspace on free tier (default: 5)

apps/sim/lib/core/config/feature-flags.ts

@@ -97,6 +97,14 @@ const FEATURE_FLAGS = {
       'resolveTriggerRegion, so the whole deployment switches regions together.',
     fallback: 'TRIGGER_EU_REGION',
   },
+  'redis-progress-markers': {
+    description:
+      'Write per-block live progress markers (lastStartedBlock/lastCompletedBlock) to Redis ' +
+      'instead of jsonb_set UPDATEs on workflow_execution_logs, folding them into the single ' +
+      'terminal UPDATE at completion. Eliminates the heaviest write query. Resolved once per ' +
+      'logging session (no user/org context) so an execution never mixes write paths.',
+    fallback: 'REDIS_PROGRESS_MARKERS',
+  },
 } satisfies Record<string, FeatureFlagDefinition>
 
 /**

apps/sim/lib/core/execution-limits/types.ts

@@ -75,6 +75,19 @@ export function getMaxExecutionTimeout(): number {
   return EXECUTION_TIMEOUTS.enterprise.async
 }
 
+/** Safety buffer added beyond the max execution timeout for execution-lifetime TTLs. */
+export const RESERVATION_TTL_BUFFER_MS = 60_000
+
+/**
+ * TTL (ms) bounding how long a single execution can remain in flight: the max
+ * execution timeout plus a safety buffer. Shared source of truth for the
+ * admission-reservation key and the live progress-marker key so they expire on
+ * the same timeline.
+ */
+export function getExecutionReservationTtlMs(): number {
+  return getMaxExecutionTimeout() + RESERVATION_TTL_BUFFER_MS
+}
+
 export const DEFAULT_EXECUTION_TIMEOUT_MS = EXECUTION_TIMEOUTS.free.sync
 
 export function isTimeoutError(error: unknown): boolean {

apps/sim/lib/logs/execution/logger.ts

@@ -37,6 +37,13 @@ import {
   replaceLargeValueReferenceKeysWithClient,
 } from '@/lib/execution/payloads/large-value-metadata'
 import { type RedactablePayload, redactPIIFromExecution } from '@/lib/logs/execution/pii-redaction'
+import {
+  clearProgressMarkers,
+  type ExecutionProgressMarkers,
+  getProgressMarkers,
+  pickLatestCompletedMarker,
+  pickLatestStartedMarker,
+} from '@/lib/logs/execution/progress-markers'
 import { snapshotService } from '@/lib/logs/execution/snapshot/service'
 import {
   externalizeExecutionData,
@@ -417,8 +424,15 @@ export class ExecutionLogger implements IExecutionLoggerService {
     return minimalWithSize.executionData
   }
 
+  /**
+   * Assemble the final `execution_data` for the terminal UPDATE. Live progress
+   * markers are sourced from `progressMarkers` (Redis, current run) and fall back
+   * to markers already on the row — the legacy SQL path and resumed rows that
+   * folded markers in at their prior pause boundary.
+   */
   private buildCompletedExecutionData(params: {
     existingExecutionData?: WorkflowExecutionLog['executionData']
+    progressMarkers?: ExecutionProgressMarkers
     traceSpans?: TraceSpan[]
     finalOutput: BlockOutputData
     finalizationPath?: ExecutionFinalizationPath
@@ -436,6 +450,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
   }): WorkflowExecutionLog['executionData'] {
     const {
       existingExecutionData,
+      progressMarkers,
       traceSpans,
       finalOutput,
       finalizationPath,
@@ -446,6 +461,15 @@ export class ExecutionLogger implements IExecutionLoggerService {
     } = params
     const traceSpanCount = countTraceSpans(traceSpans)
 
+    const lastStartedBlock = pickLatestStartedMarker(
+      progressMarkers?.lastStartedBlock,
+      existingExecutionData?.lastStartedBlock
+    )
+    const lastCompletedBlock = pickLatestCompletedMarker(
+      progressMarkers?.lastCompletedBlock,
+      existingExecutionData?.lastCompletedBlock
+    )
+
     return {
       ...(existingExecutionData?.environment
         ? { environment: existingExecutionData.environment }
@@ -459,12 +483,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
           }
         : {}),
       ...(existingExecutionData?.error ? { error: existingExecutionData.error } : {}),
-      ...(existingExecutionData?.lastStartedBlock
-        ? { lastStartedBlock: existingExecutionData.lastStartedBlock }
-        : {}),
-      ...(existingExecutionData?.lastCompletedBlock
-        ? { lastCompletedBlock: existingExecutionData.lastCompletedBlock }
-        : {}),
+      ...(lastStartedBlock ? { lastStartedBlock } : {}),
+      ...(lastCompletedBlock ? { lastCompletedBlock } : {}),
       ...(completionFailure ? { completionFailure } : {}),
       ...(finalizationPath ? { finalizationPath } : {}),
       hasTraceSpans: traceSpanCount > 0,
@@ -659,6 +679,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
     isResume?: boolean
     level?: 'info' | 'error'
     status?: 'completed' | 'failed' | 'cancelled' | 'pending'
+    readProgressMarkers?: boolean
   }): Promise<WorkflowExecutionLog> {
     const {
       executionId,
@@ -674,6 +695,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
       isResume,
       level: levelOverride,
       status: statusOverride,
+      readProgressMarkers = true,
     } = params
 
     let execLog = logger.withMetadata({ executionId })
@@ -731,8 +753,11 @@ export class ExecutionLogger implements IExecutionLoggerService {
       models: costSummary.models,
     }
 
+    const progressMarkers = readProgressMarkers ? await getProgressMarkers(executionId) : null
+
     const builtExecutionData = this.buildCompletedExecutionData({
       existingExecutionData,
+      progressMarkers: progressMarkers ?? undefined,
       traceSpans: mergedTraceSpans,
       finalOutput,
       finalizationPath,
@@ -893,6 +918,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
       return log
     })
 
+    if (progressMarkers !== null) void clearProgressMarkers(executionId)
+
     try {
       // Skip workflow lookup if workflow was deleted.
       const wf = updatedLog.workflowId

apps/sim/lib/logs/execution/logging-session.test.ts

@@ -66,6 +66,31 @@ vi.mock('@/lib/logs/execution/logger', () => ({
   },
 }))
 
+const {
+  isFeatureEnabledMock,
+  setLastStartedBlockMock,
+  setLastCompletedBlockMock,
+  getProgressMarkersMock,
+  clearProgressMarkersMock,
+} = vi.hoisted(() => ({
+  isFeatureEnabledMock: vi.fn().mockResolvedValue(false),
+  setLastStartedBlockMock: vi.fn().mockResolvedValue(false),
+  setLastCompletedBlockMock: vi.fn().mockResolvedValue(false),
+  getProgressMarkersMock: vi.fn().mockResolvedValue({}),
+  clearProgressMarkersMock: vi.fn().mockResolvedValue(undefined),
+}))
+
+vi.mock('@/lib/core/config/feature-flags', () => ({
+  isFeatureEnabled: isFeatureEnabledMock,
+}))
+
+vi.mock('@/lib/logs/execution/progress-markers', () => ({
+  setLastStartedBlock: setLastStartedBlockMock,
+  setLastCompletedBlock: setLastCompletedBlockMock,
+  getProgressMarkers: getProgressMarkersMock,
+  clearProgressMarkers: clearProgressMarkersMock,
+}))
+
 vi.mock('@/lib/logs/execution/logging-factory', () => ({
   calculateCostSummary: vi.fn().mockReturnValue({
     totalCost: 0,
@@ -647,4 +672,126 @@ describe('LoggingSession.markExecutionAsFailed workflowId scoping', () => {
     const combined = String(Array.from(strings)).toLowerCase() + values.join(' ').toLowerCase()
     expect(combined).toContain('force_failed')
   })
+
+  it('clears Redis markers when marking failed (terminal boundary outside completeWorkflowExecution)', async () => {
+    await LoggingSession.markExecutionAsFailed('exec-3', 'boom', undefined, 'wf-3')
+    expect(clearProgressMarkersMock).toHaveBeenCalledWith('exec-3')
+  })
+
+  it('folds live Redis markers into the row before clearing on force-fail', async () => {
+    getProgressMarkersMock.mockResolvedValueOnce({
+      lastStartedBlock: { blockId: 'b1', blockName: 'Fetch', blockType: 'api', startedAt: 't1' },
+      lastCompletedBlock: {
+        blockId: 'b1',
+        blockName: 'Fetch',
+        blockType: 'api',
+        endedAt: 't2',
+        success: false,
+      },
+    })
+
+    await LoggingSession.markExecutionAsFailed('exec-9', 'boom', undefined, 'wf-9')
+
+    const folded = dbMocks.sql.mock.calls
+      .map((c) => String(Array.from(c[0] as TemplateStringsArray)))
+      .join(' ')
+    expect(folded).toContain('lastStartedBlock')
+    expect(folded).toContain('lastCompletedBlock')
+    expect(clearProgressMarkersMock).toHaveBeenCalledWith('exec-9')
+  })
+
+  it('does not clear markers when the Redis read fails (avoids wiping the only copy)', async () => {
+    getProgressMarkersMock.mockResolvedValueOnce(null)
+    await LoggingSession.markExecutionAsFailed('exec-readfail', 'boom', undefined, 'wf-x')
+    expect(clearProgressMarkersMock).not.toHaveBeenCalled()
+  })
+})
+
+describe('LoggingSession progress-marker write path', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    startWorkflowExecutionMock.mockResolvedValue({})
+    loadWorkflowStateForExecutionMock.mockResolvedValue({
+      blocks: {},
+      edges: [],
+      loops: {},
+      parallels: {},
+    })
+    dbMocks.execute.mockResolvedValue(undefined)
+  })
+
+  it('writes markers to Redis (not the row) when the flag is on and Redis accepts the write', async () => {
+    isFeatureEnabledMock.mockResolvedValue(true)
+    setLastStartedBlockMock.mockResolvedValue(true)
+    setLastCompletedBlockMock.mockResolvedValue(true)
+    const session = new LoggingSession('wf-1', 'exec-redis', 'manual', 'req-1')
+    await session.start({ workspaceId: 'ws-1' })
+
+    await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
+    await session.onBlockComplete('b1', 'Fetch', 'api', { endedAt: '2026-06-27T10:00:01.000Z' })
+
+    expect(setLastStartedBlockMock).toHaveBeenCalledWith(
+      'exec-redis',
+      expect.objectContaining({ blockId: 'b1', startedAt: '2026-06-27T10:00:00.000Z' })
+    )
+    expect(setLastCompletedBlockMock).toHaveBeenCalledWith(
+      'exec-redis',
+      expect.objectContaining({ blockId: 'b1', success: true })
+    )
+    expect(dbMocks.execute).not.toHaveBeenCalled()
+  })
+
+  it('falls back to the SQL UPDATE when the flag is on but the Redis write fails', async () => {
+    isFeatureEnabledMock.mockResolvedValue(true)
+    setLastStartedBlockMock.mockResolvedValue(false)
+    const session = new LoggingSession('wf-1', 'exec-redis-down', 'manual', 'req-1')
+    await session.start({ workspaceId: 'ws-1' })
+
+    await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
+
+    expect(setLastStartedBlockMock).toHaveBeenCalled()
+    expect(dbMocks.execute).toHaveBeenCalledTimes(1)
+  })
+
+  it('writes markers via jsonb_set UPDATE when the flag is off', async () => {
+    isFeatureEnabledMock.mockResolvedValue(false)
+    const session = new LoggingSession('wf-1', 'exec-sql', 'manual', 'req-1')
+    await session.start({ workspaceId: 'ws-1' })
+
+    await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
+
+    expect(dbMocks.execute).toHaveBeenCalledTimes(1)
+    expect(setLastStartedBlockMock).not.toHaveBeenCalled()
+  })
+
+  it('falls back to the SQL path when flag resolution throws', async () => {
+    isFeatureEnabledMock.mockRejectedValue(new Error('appconfig unavailable'))
+    const session = new LoggingSession('wf-1', 'exec-fallback', 'manual', 'req-1')
+    await session.start({ workspaceId: 'ws-1' })
+
+    await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
+
+    expect(dbMocks.execute).toHaveBeenCalledTimes(1)
+    expect(setLastStartedBlockMock).not.toHaveBeenCalled()
+  })
+
+  it('tells completion to read Redis markers only when the flag is on (no wasted ops when off)', async () => {
+    completeWorkflowExecutionMock.mockResolvedValue({})
+
+    isFeatureEnabledMock.mockResolvedValue(true)
+    const onSession = new LoggingSession('wf-1', 'exec-on', 'manual', 'req-1')
+    await onSession.start({ workspaceId: 'ws-1' })
+    await onSession.safeComplete({ finalOutput: { ok: true } })
+    expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
+      expect.objectContaining({ executionId: 'exec-on', readProgressMarkers: true })
+    )
+
+    isFeatureEnabledMock.mockResolvedValue(false)
+    const offSession = new LoggingSession('wf-1', 'exec-off', 'manual', 'req-1')
+    await offSession.start({ workspaceId: 'ws-1' })
+    await offSession.safeComplete({ finalOutput: { ok: true } })
+    expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
+      expect.objectContaining({ executionId: 'exec-off', readProgressMarkers: false })
+    )
+  })
 })