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.

Author: waleedlatif1

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

@@ -70,11 +70,13 @@ const {
   isFeatureEnabledMock,
   setLastStartedBlockMock,
   setLastCompletedBlockMock,
+  getProgressMarkersMock,
   clearProgressMarkersMock,
 } = vi.hoisted(() => ({
   isFeatureEnabledMock: vi.fn().mockResolvedValue(false),
-  setLastStartedBlockMock: vi.fn().mockResolvedValue(undefined),
-  setLastCompletedBlockMock: vi.fn().mockResolvedValue(undefined),
+  setLastStartedBlockMock: vi.fn().mockResolvedValue(false),
+  setLastCompletedBlockMock: vi.fn().mockResolvedValue(false),
+  getProgressMarkersMock: vi.fn().mockResolvedValue({}),
   clearProgressMarkersMock: vi.fn().mockResolvedValue(undefined),
 }))
 
@@ -85,6 +87,7 @@ vi.mock('@/lib/core/config/feature-flags', () => ({
 vi.mock('@/lib/logs/execution/progress-markers', () => ({
   setLastStartedBlock: setLastStartedBlockMock,
   setLastCompletedBlock: setLastCompletedBlockMock,
+  getProgressMarkers: getProgressMarkersMock,
   clearProgressMarkers: clearProgressMarkersMock,
 }))
 
@@ -674,6 +677,28 @@ describe('LoggingSession.markExecutionAsFailed workflowId scoping', () => {
     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')
+  })
 })
 
 describe('LoggingSession progress-marker write path', () => {
@@ -689,8 +714,10 @@ describe('LoggingSession progress-marker write path', () => {
     dbMocks.execute.mockResolvedValue(undefined)
   })
 
-  it('writes markers to Redis (not the row) when the flag is on', async () => {
+  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' })
 
@@ -708,6 +735,18 @@ describe('LoggingSession progress-marker write path', () => {
     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')

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

@@ -16,6 +16,7 @@ import {
 } from '@/lib/logs/execution/logging-factory'
 import {
   clearProgressMarkers,
+  getProgressMarkers,
   setLastCompletedBlock,
   setLastStartedBlock,
 } from '@/lib/logs/execution/progress-markers'
@@ -193,8 +194,9 @@ export class LoggingSession {
   }
 
   private async persistLastStartedBlock(marker: ExecutionLastStartedBlock): Promise<void> {
-    if (this.useRedisMarkers) {
-      await setLastStartedBlock(this.executionId, marker)
+    // Redis is the primary path when enabled; fall back to the durable SQL write
+    // when Redis is unavailable or the write fails, so a marker is never dropped.
+    if (this.useRedisMarkers && (await setLastStartedBlock(this.executionId, marker))) {
       return
     }
     try {
@@ -215,8 +217,9 @@ export class LoggingSession {
   }
 
   private async persistLastCompletedBlock(marker: ExecutionLastCompletedBlock): Promise<void> {
-    if (this.useRedisMarkers) {
-      await setLastCompletedBlock(this.executionId, marker)
+    // Redis is the primary path when enabled; fall back to the durable SQL write
+    // when Redis is unavailable or the write fails, so a marker is never dropped.
+    if (this.useRedisMarkers && (await setLastCompletedBlock(this.executionId, marker))) {
       return
     }
     try {
@@ -1027,12 +1030,15 @@ export class LoggingSession {
   ): Promise<void> {
     try {
       const message = errorMessage || 'Run failed'
-      await db
-        .update(workflowExecutionLogs)
-        .set({
-          level: 'error',
-          status: 'failed',
-          executionData: sql`jsonb_set(
+
+      // This terminal boundary bypasses completeWorkflowExecution, so fold any
+      // live Redis markers into the row here too — otherwise a run whose markers
+      // only ever lived in Redis would be saved with error data but no
+      // last-started/last-completed breadcrumb. Empty when the standard
+      // completion path already folded and cleared them.
+      const markers = await getProgressMarkers(executionId)
+
+      let executionData = sql`jsonb_set(
             jsonb_set(
               jsonb_set(
                 COALESCE(execution_data, '{}'::jsonb),
@@ -1044,17 +1050,26 @@ export class LoggingSession {
             ),
             ARRAY['finalizationPath'],
             to_jsonb('force_failed'::text)
-          )`,
-        })
+          )`
+      if (markers.lastStartedBlock) {
+        executionData = sql`jsonb_set(${executionData}, ARRAY['lastStartedBlock'], ${JSON.stringify(markers.lastStartedBlock)}::jsonb)`
+      }
+      if (markers.lastCompletedBlock) {
+        executionData = sql`jsonb_set(${executionData}, ARRAY['lastCompletedBlock'], ${JSON.stringify(markers.lastCompletedBlock)}::jsonb)`
+      }
+
+      await db
+        .update(workflowExecutionLogs)
+        .set({ level: 'error', status: 'failed', executionData })
         .where(
           and(
             eq(workflowExecutionLogs.executionId, executionId),
             eq(workflowExecutionLogs.workflowId, workflowId)
           )
         )
 
-      // Terminal boundary that bypasses completeWorkflowExecution; clear markers
-      // so this path matches the standard clear-at-every-boundary invariant.
+      // Markers are now durable on the row; drop the Redis key to match the
+      // standard clear-at-every-boundary invariant.
       void clearProgressMarkers(executionId)
 
       logger.info(`[${requestId || 'unknown'}] Marked execution ${executionId} as failed`)

apps/sim/lib/logs/execution/progress-markers.test.ts

@@ -72,14 +72,18 @@ describe('progress-markers', () => {
       expect(args[7]).toBe(EXPECTED_TTL_MS)
     })
 
-    it('swallows eval errors (fire-and-forget)', async () => {
+    it('returns true when the Redis write succeeds', async () => {
+      await expect(setLastStartedBlock(EXECUTION_ID, startedMarker)).resolves.toBe(true)
+    })
+
+    it('returns false (caller falls back to SQL) when the eval fails', async () => {
       mockRedis.eval.mockRejectedValueOnce(new Error('redis down'))
-      await expect(setLastStartedBlock(EXECUTION_ID, startedMarker)).resolves.toBeUndefined()
+      await expect(setLastStartedBlock(EXECUTION_ID, startedMarker)).resolves.toBe(false)
     })
 
-    it('no-ops when Redis is unavailable', async () => {
+    it('returns false and no-ops when Redis is unavailable', async () => {
       mockGetRedisClient.mockReturnValue(null)
-      await setLastStartedBlock(EXECUTION_ID, startedMarker)
+      await expect(setLastStartedBlock(EXECUTION_ID, startedMarker)).resolves.toBe(false)
       expect(mockRedis.eval).not.toHaveBeenCalled()
     })
   })
@@ -127,6 +131,21 @@ describe('progress-markers', () => {
       expect(await getProgressMarkers(EXECUTION_ID)).toEqual({})
     })
 
+    it('drops wrong-shaped JSON so malformed markers never reach clients', async () => {
+      mockRedis.hgetall.mockResolvedValueOnce({
+        started: JSON.stringify('just a string'),
+        completed: JSON.stringify({ blockId: 123, blockName: 'x', blockType: 'api', endedAt: 'z' }),
+      })
+      expect(await getProgressMarkers(EXECUTION_ID)).toEqual({})
+    })
+
+    it('strips extra fields, returning only the validated marker shape', async () => {
+      mockRedis.hgetall.mockResolvedValueOnce({
+        started: JSON.stringify({ ...startedMarker, secret: 'leak', extra: 1 }),
+      })
+      expect(await getProgressMarkers(EXECUTION_ID)).toEqual({ lastStartedBlock: startedMarker })
+    })
+
     it('returns {} when Redis is unavailable', async () => {
       mockGetRedisClient.mockReturnValue(null)
       expect(await getProgressMarkers(EXECUTION_ID)).toEqual({})

apps/sim/lib/logs/execution/progress-markers.ts

@@ -61,17 +61,18 @@ return 1
  * TTL is a backstop for executions that die without a terminal/pause boundary
  * (deterministic cleanup is {@link clearProgressMarkers}); it mirrors the
  * admission-reservation TTL so a crashed run's marker key and slot expire
- * together.
+ * together. Returns `true` only when the marker was durably written to Redis,
+ * so callers can fall back to the SQL path on a missing client or a failure.
  */
 async function setMarker(
   executionId: string,
   field: string,
   timestampField: 'startedAt' | 'endedAt',
   timestamp: string,
   marker: ExecutionLastStartedBlock | ExecutionLastCompletedBlock
-): Promise<void> {
+): Promise<boolean> {
   const redis = getMarkerClient()
-  if (!redis) return
+  if (!redis) return false
 
   try {
     await redis.eval(
@@ -84,44 +85,95 @@ async function setMarker(
       JSON.stringify(marker),
       getExecutionReservationTtlMs().toString()
     )
+    return true
   } catch (error) {
     logger.error(`Failed to persist progress marker for execution ${executionId}`, {
       field,
       error: toError(error).message,
     })
+    return false
   }
 }
 
-/** Persist the last-started-block marker. No-op when Redis is unavailable. */
+/**
+ * Persist the last-started-block marker. Returns `false` (caller should fall
+ * back to the durable SQL path) when Redis is unavailable or the write fails.
+ */
 export async function setLastStartedBlock(
   executionId: string,
   marker: ExecutionLastStartedBlock
-): Promise<void> {
-  await setMarker(executionId, STARTED_FIELD, 'startedAt', marker.startedAt, marker)
+): Promise<boolean> {
+  return setMarker(executionId, STARTED_FIELD, 'startedAt', marker.startedAt, marker)
 }
 
-/** Persist the last-completed-block marker. No-op when Redis is unavailable. */
+/**
+ * Persist the last-completed-block marker. Returns `false` (caller should fall
+ * back to the durable SQL path) when Redis is unavailable or the write fails.
+ */
 export async function setLastCompletedBlock(
   executionId: string,
   marker: ExecutionLastCompletedBlock
-): Promise<void> {
-  await setMarker(executionId, COMPLETED_FIELD, 'endedAt', marker.endedAt, marker)
+): Promise<boolean> {
+  return setMarker(executionId, COMPLETED_FIELD, 'endedAt', marker.endedAt, marker)
 }
 
 export interface ExecutionProgressMarkers {
   lastStartedBlock?: ExecutionLastStartedBlock
   lastCompletedBlock?: ExecutionLastCompletedBlock
 }
 
-function parseMarker<T>(raw: string | undefined): T | undefined {
+function safeJsonParse(raw: string | undefined): unknown {
   if (!raw) return undefined
   try {
-    return JSON.parse(raw) as T
+    return JSON.parse(raw)
   } catch {
     return undefined
   }
 }
 
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+/**
+ * Parse a stored last-started marker, rebuilding it from validated fields so a
+ * stale or wrong-shaped Redis value can never reach API consumers.
+ */
+function parseStartedMarker(raw: string | undefined): ExecutionLastStartedBlock | undefined {
+  const v = safeJsonParse(raw)
+  if (!isRecord(v)) return undefined
+  const { blockId, blockName, blockType, startedAt } = v
+  if (
+    typeof blockId === 'string' &&
+    typeof blockName === 'string' &&
+    typeof blockType === 'string' &&
+    typeof startedAt === 'string'
+  ) {
+    return { blockId, blockName, blockType, startedAt }
+  }
+  return undefined
+}
+
+/**
+ * Parse a stored last-completed marker, rebuilding it from validated fields so a
+ * stale or wrong-shaped Redis value can never reach API consumers.
+ */
+function parseCompletedMarker(raw: string | undefined): ExecutionLastCompletedBlock | undefined {
+  const v = safeJsonParse(raw)
+  if (!isRecord(v)) return undefined
+  const { blockId, blockName, blockType, endedAt, success } = v
+  if (
+    typeof blockId === 'string' &&
+    typeof blockName === 'string' &&
+    typeof blockType === 'string' &&
+    typeof endedAt === 'string' &&
+    typeof success === 'boolean'
+  ) {
+    return { blockId, blockName, blockType, endedAt, success }
+  }
+  return undefined
+}
+
 /**
  * Read both markers for an execution. Returns an empty object when Redis is
  * unavailable, the key has expired, or nothing has been written yet.
@@ -135,9 +187,9 @@ export async function getProgressMarkers(executionId: string): Promise<Execution
     if (!fields || Object.keys(fields).length === 0) return {}
 
     const result: ExecutionProgressMarkers = {}
-    const started = parseMarker<ExecutionLastStartedBlock>(fields[STARTED_FIELD])
+    const started = parseStartedMarker(fields[STARTED_FIELD])
     if (started) result.lastStartedBlock = started
-    const completed = parseMarker<ExecutionLastCompletedBlock>(fields[COMPLETED_FIELD])
+    const completed = parseCompletedMarker(fields[COMPLETED_FIELD])
     if (completed) result.lastCompletedBlock = completed
     return result
   } catch (error) {