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.

Author: waleedlatif1

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

@@ -41,6 +41,8 @@ import {
   clearProgressMarkers,
   type ExecutionProgressMarkers,
   getProgressMarkers,
+  pickLatestCompletedMarker,
+  pickLatestStartedMarker,
 } from '@/lib/logs/execution/progress-markers'
 import { snapshotService } from '@/lib/logs/execution/snapshot/service'
 import {
@@ -459,10 +461,14 @@ export class ExecutionLogger implements IExecutionLoggerService {
     } = params
     const traceSpanCount = countTraceSpans(traceSpans)
 
-    const lastStartedBlock =
-      progressMarkers?.lastStartedBlock ?? existingExecutionData?.lastStartedBlock
-    const lastCompletedBlock =
-      progressMarkers?.lastCompletedBlock ?? existingExecutionData?.lastCompletedBlock
+    const lastStartedBlock = pickLatestStartedMarker(
+      progressMarkers?.lastStartedBlock,
+      existingExecutionData?.lastStartedBlock
+    )
+    const lastCompletedBlock = pickLatestCompletedMarker(
+      progressMarkers?.lastCompletedBlock,
+      existingExecutionData?.lastCompletedBlock
+    )
 
     return {
       ...(existingExecutionData?.environment

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

@@ -1018,8 +1018,14 @@ export class LoggingSession {
     }
   }
 
+  /**
+   * Force-fail the execution. Waits for any in-flight completion and drains
+   * pending per-block marker writes first, so a force-fail racing
+   * onBlockStart/onBlockComplete still captures the latest breadcrumb in the fold.
+   */
   async markAsFailed(errorMessage?: string): Promise<void> {
     await this.waitForCompletion()
+    await this.drainPendingProgressWrites()
     await LoggingSession.markExecutionAsFailed(
       this.executionId,
       errorMessage,
@@ -1060,10 +1066,18 @@ export class LoggingSession {
             to_jsonb('force_failed'::text)
           )`
       if (markers?.lastStartedBlock) {
-        executionData = sql`jsonb_set(${executionData}, ARRAY['lastStartedBlock'], ${JSON.stringify(markers.lastStartedBlock)}::jsonb)`
+        const startedAt = markers.lastStartedBlock.startedAt
+        const startedJson = JSON.stringify(markers.lastStartedBlock)
+        executionData = sql`CASE WHEN COALESCE(jsonb_extract_path_text(execution_data, 'lastStartedBlock', 'startedAt'), '') <= ${startedAt}
+            THEN jsonb_set(${executionData}, ARRAY['lastStartedBlock'], ${startedJson}::jsonb)
+            ELSE ${executionData} END`
       }
       if (markers?.lastCompletedBlock) {
-        executionData = sql`jsonb_set(${executionData}, ARRAY['lastCompletedBlock'], ${JSON.stringify(markers.lastCompletedBlock)}::jsonb)`
+        const endedAt = markers.lastCompletedBlock.endedAt
+        const completedJson = JSON.stringify(markers.lastCompletedBlock)
+        executionData = sql`CASE WHEN COALESCE(jsonb_extract_path_text(execution_data, 'lastCompletedBlock', 'endedAt'), '') <= ${endedAt}
+            THEN jsonb_set(${executionData}, ARRAY['lastCompletedBlock'], ${completedJson}::jsonb)
+            ELSE ${executionData} END`
       }
 
       await db

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

@@ -24,6 +24,8 @@ vi.mock('@/lib/core/execution-limits', () => ({
 import {
   clearProgressMarkers,
   getProgressMarkers,
+  pickLatestCompletedMarker,
+  pickLatestStartedMarker,
   setLastCompletedBlock,
   setLastStartedBlock,
 } from '@/lib/logs/execution/progress-markers'
@@ -175,4 +177,27 @@ describe('progress-markers', () => {
       expect(mockRedis.del).not.toHaveBeenCalled()
     })
   })
+
+  describe('latest-wins pickers (stale-store safety)', () => {
+    const older = { ...startedMarker, blockId: 'old', startedAt: '2026-06-27T10:00:00.000Z' }
+    const newer = { ...startedMarker, blockId: 'new', startedAt: '2026-06-27T10:00:05.000Z' }
+
+    it('returns the defined side when the other is undefined', () => {
+      expect(pickLatestStartedMarker(older, undefined)).toBe(older)
+      expect(pickLatestStartedMarker(undefined, newer)).toBe(newer)
+      expect(pickLatestStartedMarker(undefined, undefined)).toBeUndefined()
+    })
+
+    it('picks the later startedAt regardless of argument order (row newer than Redis still wins)', () => {
+      expect(pickLatestStartedMarker(older, newer)).toBe(newer)
+      expect(pickLatestStartedMarker(newer, older)).toBe(newer)
+    })
+
+    it('picks the later endedAt for completed markers', () => {
+      const c1 = { ...completedMarker, endedAt: '2026-06-27T10:00:01.000Z' }
+      const c2 = { ...completedMarker, endedAt: '2026-06-27T10:00:09.000Z' }
+      expect(pickLatestCompletedMarker(c1, c2)).toBe(c2)
+      expect(pickLatestCompletedMarker(c2, c1)).toBe(c2)
+    })
+  })
 })

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

@@ -122,6 +122,32 @@ export interface ExecutionProgressMarkers {
   lastCompletedBlock?: ExecutionLastCompletedBlock
 }
 
+/**
+ * Pick the later of two last-started markers by `startedAt`. Markers can split
+ * across stores — a failed Redis write falls back to the row, so an earlier
+ * successful Redis write may coexist with a newer row marker (or vice versa).
+ * Choosing by timestamp keeps the freshest breadcrumb regardless of which store
+ * holds it. ISO UTC timestamps compare correctly lexicographically.
+ */
+export function pickLatestStartedMarker(
+  a: ExecutionLastStartedBlock | undefined,
+  b: ExecutionLastStartedBlock | undefined
+): ExecutionLastStartedBlock | undefined {
+  if (!a) return b
+  if (!b) return a
+  return a.startedAt >= b.startedAt ? a : b
+}
+
+/** Pick the later of two last-completed markers by `endedAt`. See {@link pickLatestStartedMarker}. */
+export function pickLatestCompletedMarker(
+  a: ExecutionLastCompletedBlock | undefined,
+  b: ExecutionLastCompletedBlock | undefined
+): ExecutionLastCompletedBlock | undefined {
+  if (!a) return b
+  if (!b) return a
+  return a.endedAt >= b.endedAt ? a : b
+}
+
 function safeJsonParse(raw: string | undefined): unknown {
   if (!raw) return undefined
   try {

apps/sim/lib/logs/fetch-log-detail.ts

@@ -9,7 +9,12 @@ import {
 } from '@sim/db/schema'
 import { and, eq, type SQL } from 'drizzle-orm'
 import type { CostLedger } from '@/lib/api/contracts/logs'
-import { getProgressMarkers } from '@/lib/logs/execution/progress-markers'
+import {
+  type ExecutionProgressMarkers,
+  getProgressMarkers,
+  pickLatestCompletedMarker,
+  pickLatestStartedMarker,
+} from '@/lib/logs/execution/progress-markers'
 import { materializeExecutionData } from '@/lib/logs/execution/trace-store'
 import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 
@@ -174,6 +179,15 @@ export async function fetchLogDetail({
       log.status === 'running' || log.status === 'pending'
         ? ((await getProgressMarkers(log.executionId)) ?? {})
         : {}
+    const rowMarkers = (executionData ?? {}) as ExecutionProgressMarkers
+    const mergedStartedBlock = pickLatestStartedMarker(
+      liveMarkers.lastStartedBlock,
+      rowMarkers.lastStartedBlock
+    )
+    const mergedCompletedBlock = pickLatestCompletedMarker(
+      liveMarkers.lastCompletedBlock,
+      rowMarkers.lastCompletedBlock
+    )
 
     return {
       id: log.id,
@@ -200,7 +214,8 @@ export async function fetchLogDetail({
       executionData: {
         totalDuration: log.totalDurationMs,
         ...executionData,
-        ...liveMarkers,
+        ...(mergedStartedBlock ? { lastStartedBlock: mergedStartedBlock } : {}),
+        ...(mergedCompletedBlock ? { lastCompletedBlock: mergedCompletedBlock } : {}),
         enhanced: true as const,
       },
       files: log.files ?? null,