fix(data-drains): add stability window to time cursors so late-visible rows are never skipped

Author: waleedlatif1

apps/sim/lib/data-drains/sources/audit-logs.ts

@@ -6,6 +6,7 @@ import {
   encodeTimeCursor,
   timeCursorOrderBy,
   timeCursorPredicate,
+  timeCursorStabilityBound,
 } from '@/lib/data-drains/sources/cursor'
 import { getOrganizationWorkspaceIds } from '@/lib/data-drains/sources/helpers'
 import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/types'
@@ -38,7 +39,7 @@ async function* pages(input: SourcePageInput): AsyncIterable<AuditLogRow[]> {
     const rows = await dbReplica
       .select()
       .from(auditLog)
-      .where(and(scopeClause, cursorClause))
+      .where(and(scopeClause, timeCursorStabilityBound(auditLog.createdAt), cursorClause))
       .orderBy(...timeCursorOrderBy(auditLog.createdAt, auditLog.id))
       .limit(input.chunkSize)
 

apps/sim/lib/data-drains/sources/copilot-chats.ts

@@ -6,6 +6,7 @@ import {
   encodeTimeCursor,
   timeCursorOrderBy,
   timeCursorPredicate,
+  timeCursorStabilityBound,
 } from '@/lib/data-drains/sources/cursor'
 import { getOrganizationWorkspaceIds } from '@/lib/data-drains/sources/helpers'
 import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/types'
@@ -58,7 +59,13 @@ async function* pages(input: SourcePageInput): AsyncIterable<CopilotChatRow[]> {
     const metaRows = await dbReplica
       .select(chatColumns)
       .from(copilotChats)
-      .where(and(inArray(copilotChats.workspaceId, workspaceIds), cursorClause))
+      .where(
+        and(
+          inArray(copilotChats.workspaceId, workspaceIds),
+          timeCursorStabilityBound(copilotChats.createdAt),
+          cursorClause
+        )
+      )
       .orderBy(...timeCursorOrderBy(copilotChats.createdAt, copilotChats.id))
       .limit(input.chunkSize)
 

apps/sim/lib/data-drains/sources/copilot-runs.ts

@@ -6,6 +6,7 @@ import {
   encodeTimeCursor,
   timeCursorOrderBy,
   timeCursorPredicate,
+  timeCursorStabilityBound,
 } from '@/lib/data-drains/sources/cursor'
 import { getOrganizationWorkspaceIds } from '@/lib/data-drains/sources/helpers'
 import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/types'
@@ -31,6 +32,7 @@ async function* pages(input: SourcePageInput): AsyncIterable<CopilotRunRow[]> {
         and(
           inArray(copilotRuns.workspaceId, workspaceIds),
           isNotNull(copilotRuns.completedAt),
+          timeCursorStabilityBound(copilotRuns.completedAt),
           cursorClause
         )
       )

apps/sim/lib/data-drains/sources/cursor.ts

@@ -54,3 +54,14 @@ export function timeCursorPredicate(
 export function timeCursorOrderBy(timestampCol: PgColumn, idCol: PgColumn): [SQL, SQL] {
   return [sql`date_trunc('milliseconds', ${timestampCol}) asc`, sql`${idCol} asc`]
 }
+
+/**
+ * Excludes rows newer than a short stability window. Timestamp cursors assume
+ * rows become visible in timestamp order, but out-of-order commits and replica
+ * lag can surface an earlier-stamped row after the cursor has advanced past it
+ * — permanently skipping it. Leaving the freshest rows for the next run bounds
+ * both.
+ */
+export function timeCursorStabilityBound(timestampCol: PgColumn): SQL {
+  return sql`${timestampCol} <= now() - interval '5 minutes'`
+}

apps/sim/lib/data-drains/sources/job-logs.ts

@@ -6,6 +6,7 @@ import {
   encodeTimeCursor,
   timeCursorOrderBy,
   timeCursorPredicate,
+  timeCursorStabilityBound,
 } from '@/lib/data-drains/sources/cursor'
 import { getOrganizationWorkspaceIds } from '@/lib/data-drains/sources/helpers'
 import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/types'
@@ -31,6 +32,7 @@ async function* pages(input: SourcePageInput): AsyncIterable<JobLogRow[]> {
         and(
           inArray(jobExecutionLogs.workspaceId, workspaceIds),
           isNotNull(jobExecutionLogs.endedAt),
+          timeCursorStabilityBound(jobExecutionLogs.endedAt),
           cursorClause
         )
       )

apps/sim/lib/data-drains/sources/workflow-logs.ts

@@ -7,6 +7,7 @@ import {
   encodeTimeCursor,
   timeCursorOrderBy,
   timeCursorPredicate,
+  timeCursorStabilityBound,
 } from '@/lib/data-drains/sources/cursor'
 import { getOrganizationWorkspaceIds } from '@/lib/data-drains/sources/helpers'
 import type { Cursor, DrainSource, SourcePageInput } from '@/lib/data-drains/types'
@@ -40,6 +41,7 @@ async function* pages(input: SourcePageInput): AsyncIterable<WorkflowLogRow[]> {
         and(
           inArray(workflowExecutionLogs.workspaceId, workspaceIds),
           isNotNull(workflowExecutionLogs.endedAt),
+          timeCursorStabilityBound(workflowExecutionLogs.endedAt),
           cursorClause
         )
       )