perf(logs): order public log pages on an indexed column and retire a superseded document index (#8913)

* perf(logs): order the public log keyset on the raw started_at column

The public logs list ordered by date_trunc('milliseconds', started_at), which
no index serves, so every page read and sorted every run in scope and computed
the handled-error flag for each. Order and compare on the raw column instead,
carrying a full-precision cursor value (to_char ... US"Z") so the keyset can
name the exact stored instant. Cursors minted before this keep the truncated
keyset they were read under until their walk ends, so in-flight pagination
neither skips nor repeats rows.

Also:
- the precise timestamp key opts into stating its implied inclusive bound in
  keysetAfter, so a resumed page starts its index scan at the cursor instead
  of filtering down to it; every other keyset list renders unchanged SQL
- the workflowIds filter matches workflow_execution_logs.workflow_id rather
  than the joined workflow.id (equivalent under the SET NULL FK), which lets a
  multi-workflow list reach the log indexes

* perf(db): drop the superseded reconciliation index and index the sync-log prune

- Drop doc_connector_reconciliation_idx (contract of 0388): it keys on
  source_seen_at, so every connector listing's seen stamp is a non-HOT update
  that rewrites every document index; nothing orders by it since v0.9.4.
- Add kcsl_terminal_started_at_idx, partial on the retention prune's status
  list, so the sync-log prune is a range scan on started_at.

* fix(logs): classify timestamp cursors in one place so v1 never 500s on one

The v1 decoder admitted any date a Date could parse, while the precise key
bound only the six-digit form, so a no-fraction or four-digit cursor passed
the decoder and failed in the query as a 500. classifyTimestampCursor now
decides for both: a value in the precise form must name a real instant or is
invalid (a 400 at the v1 decoder), and any other parseable date in years
1-9999 resumes under the millisecond keyset exactly as before.

* test(logs): assert the precise cursor decodes before paging from it

Author: waleedlatif1

apps/sim/lib/api/list-query.test.ts

@@ -14,6 +14,7 @@ import {
   keysetAfter,
   listOrderBy,
   numberKey,
+  preciseTimestampKey,
   searchFilter,
   textKey,
   timestampKey,
@@ -108,6 +109,21 @@ describe('timestampKey', () => {
   })
 })
 
+describe('preciseTimestampKey', () => {
+  const preciseKey = preciseTimestampKey<{ at: string }>(thing.createdAt, (r) => r.at)
+
+  /** A cursor is caller-controlled, and each of these would reach the cast and 500. */
+  it.each([
+    ['a millisecond cursor', '2024-01-01T00:00:00.123Z'],
+    ['an impossible calendar day', '2024-02-30T00:00:00.000000Z'],
+    ['an impossible clock time', '2024-01-01T24:00:00.000000Z'],
+    ['year zero', '0000-01-01T00:00:00.000000Z'],
+    ['a non-string value', 1700000000000],
+  ])('rejects %s', (_case, value) => {
+    expect(preciseKey.bind(value)).toBeNull()
+  })
+})
+
 describe('decimalKey', () => {
   const costKey = decimalKey<{ cost: string }>(thing.cost, (r) => r.cost)
 

apps/sim/lib/api/list-query.ts

@@ -6,8 +6,10 @@ import {
   desc,
   eq,
   gt,
+  gte,
   ilike,
   lt,
+  lte,
   or,
   type SQL,
   type SQLWrapper,
@@ -62,6 +64,11 @@ export interface KeysetKey<Row> {
   encode: (row: Row) => CursorKey
   /** The cursor value as bindable SQL, or `null` when this key cannot hold it. */
   bind: (value: CursorKey) => SQL | null
+  /**
+   * Whether a cursor leading with this key also states its inclusive bound — see
+   * {@link keysetAfter}. Set only on a key an index serves in order.
+   */
+  boundsScan?: boolean
 }
 
 /** A text key — names, titles, ids. */
@@ -128,9 +135,9 @@ export function decimalKey<Row>(column: SQLWrapper, read: (row: Row) => string):
  * can express, so the `id` tiebreaker is what actually separates rows inside a
  * millisecond.
  *
- * `date_trunc` rules out an index-ordered scan, but none of the timestamp
- * columns sorted here are indexed, so it costs nothing today. Adding an index
- * to serve one of these sorts means indexing this same expression.
+ * `date_trunc` rules out an index-ordered scan on the column. A list whose
+ * sort an index on the raw column should serve uses {@link preciseTimestampKey}
+ * instead, which orders on the column itself.
  *
  * The column serializes the `Date` (drizzle's own timestamp encoder) and also
  * supplies the cast. A bare placeholder arrives as `unknown` and `date_trunc` is
@@ -154,6 +161,71 @@ export function timestampKey<Row>(column: Column, read: (row: Row) => Date): Key
   }
 }
 
+/**
+ * The text spelling a {@link preciseTimestampKey} cursor carries: ISO 8601 in
+ * UTC with all six fractional digits Postgres stores.
+ */
+const PRECISE_TIMESTAMP_CURSOR_PATTERN = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$/
+
+/**
+ * What a timestamp cursor value is: a {@link preciseTimestampKey} position, a
+ * date for the millisecond {@link timestampKey}, or neither.
+ *
+ * The one place a timestamp cursor is classified, so a codec that validates a
+ * caller's token and the keyset that later binds it cannot disagree — a token
+ * the codec admits but the key refuses would surface as a 500 on a surface that
+ * renders its own errors.
+ *
+ * A value spelled in the precise form is precise or invalid, never a date: it
+ * must name a real instant, because the pattern admits `2026-02-30`, which a
+ * `Date` silently rolls into March and Postgres refuses outright. Anything else a
+ * `Date` parses is a date. Both classes stay within years 1 to 9999, the range
+ * whose ISO spelling Postgres reads back.
+ */
+export function classifyTimestampCursor(value: CursorKey): 'precise' | 'date' | 'invalid' {
+  if (typeof value !== 'string') return 'invalid'
+  const date = new Date(value)
+  const year = date.getUTCFullYear()
+  if (Number.isNaN(date.getTime()) || year < 1 || year > 9999) return 'invalid'
+  if (!PRECISE_TIMESTAMP_CURSOR_PATTERN.test(value)) return 'date'
+  return date.toISOString().slice(0, 23) === value.slice(0, 23) ? 'precise' : 'invalid'
+}
+
+/**
+ * A `timestamp` column rendered as a {@link preciseTimestampKey} cursor value.
+ *
+ * Select it next to the column so each row carries its own full-precision
+ * position. Only for a `timestamp` (without time zone) column, which holds UTC
+ * wall-clock time: `to_char` renders a `timestamptz` in the session time zone,
+ * which the trailing `Z` would misstate.
+ */
+export function preciseTimestampText(column: Column): SQL<string> {
+  return sql<string>`to_char(${column}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`
+}
+
+/**
+ * A `timestamp` key ordered and compared on the raw column at full microsecond
+ * precision, so an index on the column serves the sort.
+ *
+ * {@link timestampKey} truncates to milliseconds because its cursor value comes
+ * from a JS `Date`; this key instead reads the row's {@link preciseTimestampText}
+ * value, so the cursor names the exact stored instant and the raw ordering needs
+ * no truncation. `timestamp` input ignores the trailing `Z`. It binds only a
+ * value {@link classifyTimestampCursor} calls precise.
+ */
+export function preciseTimestampKey<Row>(
+  column: Column,
+  read: (row: Row) => string
+): KeysetKey<Row> {
+  return {
+    expr: column,
+    encode: read,
+    boundsScan: true,
+    bind: (value) =>
+      classifyTimestampCursor(value) === 'precise' ? sql`cast(${value} as timestamp)` : null,
+  }
+}
+
 export function sortDirection(order: ListSortOrder): typeof asc {
   return order === 'asc' ? asc : desc
 }
@@ -233,6 +305,12 @@ export function keysetPage<Row>(
  * direction, expanded lexicographically so ties on a leading key fall through
  * to the next one.
  *
+ * When the leading key {@link KeysetKey.boundsScan | bounds the scan}, the
+ * cursor also carries that key's inclusive bound, which the expansion already
+ * implies. Postgres cannot derive a range from an `OR`, so without it an index
+ * on the leading key starts every page at the top of the list and filters its
+ * way down to the cursor.
+ *
  * Returns `null` when the cursor does not fit this sort — wrong number of keys,
  * or a value the key cannot hold. Callers render that as a 400 rather than
  * paging from a nonsense position.
@@ -255,5 +333,9 @@ export function keysetAfter<Row>(
   const clauses = keys.map((key, i) =>
     and(...keys.slice(0, i).map((prior, j) => eq(prior.expr, bound[j])), beyond(key.expr, bound[i]))
   )
-  return or(...clauses) ?? null
+  const after = or(...clauses)
+  if (!after) return null
+  if (keys.length === 1 || !keys[0].boundsScan) return after
+  const atOrBeyond = order === 'asc' ? gte : lte
+  return and(atOrBeyond(keys[0].expr, bound[0]), after) ?? null
 }

apps/sim/lib/logs/public-filters.ts

@@ -64,9 +64,10 @@ export function buildLogFilters(filters: LogFilters): SQL<unknown> {
 
   conditions.push(eq(workflowExecutionLogs.workspaceId, filters.workspaceId))
 
-  // Workflow IDs filter
+  // The log's own column, not the joined `workflow.id`: the FK nulls it on delete, so
+  // both select the same rows, but only this side reaches the log indexes.
   if (filters.workflowIds && filters.workflowIds.length > 0) {
-    conditions.push(inArray(workflow.id, filters.workflowIds))
+    conditions.push(inArray(workflowExecutionLogs.workflowId, filters.workflowIds))
   }
 
   // Folder IDs filter

apps/sim/lib/logs/public-queries.integration.ts

@@ -0,0 +1,203 @@
+/**
+ * Public log paging against real PostgreSQL: the `(started_at, id)` keyset walks
+ * rows that share a millisecond in their stored microsecond order, across both
+ * log tables, without repeating or dropping one.
+ */
+import { db } from '@sim/db'
+import {
+  jobExecutionLogs,
+  user,
+  workflow,
+  workflowExecutionLogs,
+  workflowExecutionSnapshots,
+  workspace,
+} from '@sim/db/schema'
+import { generateId } from '@sim/utils/id'
+import { eq, sql } from 'drizzle-orm'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import type { CursorKey, ListSortOrder } from '@/lib/api/list-query'
+import type { LogFilters } from '@/lib/logs/public-filters'
+import {
+  decodePublicLogCursor,
+  encodePublicLogCursor,
+  readPublicLogPage,
+} from '@/lib/logs/public-queries'
+
+const prefix = generateId()
+const ids = {
+  owner: `public-log-paging-${prefix}`,
+  workspace: generateId(),
+  workflow: generateId(),
+  snapshot: generateId(),
+}
+
+/** Run id → microseconds past `12:00:00.123`, inverted against id order inside the millisecond. */
+const WORKFLOW_RUNS = {
+  [`${prefix}-0`]: -1000,
+  [`${prefix}-1`]: 100,
+  [`${prefix}-2`]: 500,
+  [`${prefix}-3`]: 500,
+  [`${prefix}-4`]: 900,
+}
+const JOB_RUNS = { [`${prefix}-5`]: 700 }
+
+/** Stored order, newest first: microseconds decide, and `id` only splits the exact tie. */
+const NEWEST_FIRST = [4, 5, 3, 2, 1, 0].map((n) => `${prefix}-${n}`)
+
+function startedAt(offsetMicroseconds: number) {
+  return sql`timestamp '2026-01-01 12:00:00.123' + ${offsetMicroseconds} * interval '1 microsecond'`
+}
+
+async function walk(
+  filters: Omit<LogFilters, 'workspaceId'>,
+  order: ListSortOrder,
+  options: { includeJobRuns: boolean; cursorKeys?: CursorKey[] }
+) {
+  const seen: string[] = []
+  const cursors: CursorKey[][] = []
+  let cursorKeys = options.cursorKeys
+  for (let page = 0; page <= NEWEST_FIRST.length; page++) {
+    const result = await readPublicLogPage({
+      filters: { ...filters, workspaceId: ids.workspace },
+      limit: 1,
+      includeExecutionData: false,
+      includeJobRuns: options.includeJobRuns,
+      sortBy: 'startedAt',
+      sortOrder: order,
+      cursorKeys,
+    })
+    seen.push(...result.data.map((row) => row.id))
+    if (!result.nextCursorKeys) break
+    cursors.push(result.nextCursorKeys)
+    cursorKeys = result.nextCursorKeys
+  }
+  return { seen, cursors }
+}
+
+beforeAll(async () => {
+  const now = new Date()
+  await db.insert(user).values({
+    id: ids.owner,
+    name: 'Public Log Paging',
+    email: `${ids.owner}@public-log-paging.test`,
+    emailVerified: true,
+    createdAt: now,
+    updatedAt: now,
+  })
+  await db.insert(workspace).values({
+    id: ids.workspace,
+    name: 'Public Log Paging',
+    ownerId: ids.owner,
+    billedAccountUserId: ids.owner,
+  })
+  await db.insert(workflow).values({
+    id: ids.workflow,
+    userId: ids.owner,
+    workspaceId: ids.workspace,
+    name: 'Public Log Paging',
+    lastSynced: now,
+    createdAt: now,
+    updatedAt: now,
+  })
+  await db.insert(workflowExecutionSnapshots).values({
+    id: ids.snapshot,
+    workflowId: ids.workflow,
+    stateHash: ids.snapshot,
+    stateData: {},
+  })
+  await db.insert(workflowExecutionLogs).values(
+    Object.entries(WORKFLOW_RUNS).map(([id, offset]) => ({
+      id,
+      workflowId: ids.workflow,
+      workspaceId: ids.workspace,
+      executionId: generateId(),
+      stateSnapshotId: ids.snapshot,
+      level: 'info',
+      status: 'completed',
+      trigger: 'api',
+      startedAt: startedAt(offset),
+    }))
+  )
+  await db.insert(jobExecutionLogs).values(
+    Object.entries(JOB_RUNS).map(([id, offset]) => ({
+      id,
+      workspaceId: ids.workspace,
+      executionId: generateId(),
+      level: 'info',
+      status: 'completed',
+      trigger: 'mothership',
+      startedAt: startedAt(offset),
+    }))
+  )
+}, 30_000)
+
+afterAll(async () => {
+  await db.delete(workflowExecutionLogs).where(eq(workflowExecutionLogs.workspaceId, ids.workspace))
+  await db.delete(jobExecutionLogs).where(eq(jobExecutionLogs.workspaceId, ids.workspace))
+  await db.delete(workflowExecutionSnapshots).where(eq(workflowExecutionSnapshots.id, ids.snapshot))
+  await db.delete(workflow).where(eq(workflow.id, ids.workflow))
+  await db.delete(workspace).where(eq(workspace.id, ids.workspace))
+  await db.delete(user).where(eq(user.id, ids.owner))
+})
+
+describe('public log keyset paging', () => {
+  it.each([
+    ['desc', NEWEST_FIRST],
+    ['asc', [...NEWEST_FIRST].reverse()],
+  ] as const)(
+    'pages one run at a time in stored %s order across a shared millisecond and both tables',
+    async (order, expected) => {
+      const { seen } = await walk({}, order, { includeJobRuns: true })
+
+      expect(seen).toEqual(expected)
+    }
+  )
+
+  it('pages a workflow filter in stored order', async () => {
+    const { seen } = await walk({ workflowIds: [ids.workflow] }, 'desc', {
+      includeJobRuns: false,
+    })
+
+    expect(seen).toEqual(NEWEST_FIRST.filter((id) => id in WORKFLOW_RUNS))
+  })
+
+  /**
+   * Earlier releases accepted any date a v1 caller sent and resumed it under
+   * millisecond ordering, which `-0` (`.122`) then the `.123` runs in `id` order.
+   */
+  it.each([
+    ['without a fraction', '2026-01-01T12:00:00Z', [0, 1, 2, 3, 4, 5]],
+    ['with a four-digit fraction', '2026-01-01T12:00:00.1225Z', [1, 2, 3, 4, 5]],
+  ])('resumes a v1 date cursor %s under millisecond ordering', async (_case, startedAt, rest) => {
+    const cursor = decodePublicLogCursor(
+      encodePublicLogCursor({ startedAt, id: `${prefix}-0`, order: 'asc' }),
+      'asc'
+    )
+    expect(cursor).not.toBeNull()
+
+    const { seen } = await walk({}, 'asc', {
+      includeJobRuns: true,
+      cursorKeys: [cursor?.startedAt ?? '', cursor?.id ?? ''],
+    })
+
+    expect(seen).toEqual(rest.map((n) => `${prefix}-${n}`))
+  })
+
+  /**
+   * A cursor minted before raw ordering names a millisecond and was read under
+   * millisecond-truncated ordering, where the shared millisecond fell in `id`
+   * order: `-5`, then `-4`, … Resuming after `-4` must serve exactly the runs
+   * that order had not reached, and keep minting cursors it can resume.
+   */
+  it('finishes a walk begun under a millisecond cursor in the order it began', async () => {
+    const { seen, cursors } = await walk({}, 'desc', {
+      includeJobRuns: true,
+      cursorKeys: ['2026-01-01T12:00:00.123Z', `${prefix}-4`],
+    })
+
+    expect(seen).toEqual([3, 2, 1, 0].map((n) => `${prefix}-${n}`))
+    for (const [startedAtKey] of cursors) {
+      expect(startedAtKey).toMatch(/\.\d{3}Z$/)
+    }
+  })
+})