perf(table): read a table and its latest job in one round trip

getTableById issued the table SELECT and then awaited latestJobForTable, so
every table request paid two sequential round trips. With prepared statements
disabled for PgBouncer transaction mode every await is a full round trip, and
this loader is on essentially every table route.

The job read cannot be skipped: a table's reported rowCount is the stored
count minus the job's pendingDeleteRemaining, so dropping it would overstate
the count during a pending delete and could wrongly reject inserts as over
capacity. An opt-out flag would have made that a caller's trap. Instead the
job is read in the same statement, as a correlated jsonb subquery in the
select list — the select-list form of a LEFT JOIN LATERAL, which is what
drizzle can type here. Output is unchanged for every input.

latestJobForTable is deleted rather than left dangling: getTableById was its
only caller, and keeping it would have carried a third copy of the
exports-excluded / newest-started_at / limit-one rule. mapJobRow is now
exported so the batch path and the lateral share one implementation of the
doomedCount and pendingDeleteRemaining logic. The batch DISTINCT ON path used
by the list endpoint is untouched.

Verified to fail: dropping the export filter, reversing or re-keying the sort,
dropping the limit, dropping the correlation, loosening either doomedCount
condition, and removing the rowCount subtraction each turn tests red. Dropping
the lateral from the projection initially survived, because the shared db mock
returns queued rows regardless of predicate; a projection assertion now
covers it.

Author: waleedlatif1

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

@@ -37,7 +37,8 @@ vi.mock('@/lib/table/billing', () => ({
 }))
 vi.mock('@/lib/table/jobs/service', () => ({
   EMPTY_JOB_FIELDS: {},
-  latestJobForTable: vi.fn(async () => null),
+  latestNonExportJobJson: vi.fn(() => null),
+  mapJobRow: vi.fn(() => ({})),
   latestJobsForTables: vi.fn(async () => new Map()),
 }))
 vi.mock('@/lib/table/events', () => ({ appendTableEvent: vi.fn() }))

apps/sim/lib/table/jobs/service.test.ts

@@ -0,0 +1,110 @@
+/**
+ * @vitest-environment node
+ */
+import { schemaMock } from '@sim/testing'
+import { describe, expect, it } from 'vitest'
+import {
+  EMPTY_JOB_FIELDS,
+  type LatestJobRow,
+  latestNonExportJobJson,
+  mapJobRow,
+} from '@/lib/table/jobs/service'
+
+function job(overrides: Partial<LatestJobRow>): LatestJobRow {
+  return {
+    id: 'job-1',
+    type: 'delete',
+    status: 'running',
+    rowsProcessed: 0,
+    error: null,
+    payload: null,
+    ...overrides,
+  }
+}
+
+describe('mapJobRow', () => {
+  it('returns the empty fields when the table has no job row', () => {
+    expect(mapJobRow(null)).toEqual(EMPTY_JOB_FIELDS)
+    expect(mapJobRow(undefined)).toEqual(EMPTY_JOB_FIELDS)
+  })
+
+  it('projects a running delete job and its remaining doomed rows', () => {
+    expect(mapJobRow(job({ rowsProcessed: 4, payload: { doomedCount: 10 } }))).toEqual({
+      jobStatus: 'running',
+      jobId: 'job-1',
+      jobType: 'delete',
+      jobError: null,
+      jobRowsProcessed: 4,
+      pendingDeleteRemaining: 6,
+    })
+  })
+
+  it('ignores doomedCount once the delete job is terminal', () => {
+    expect(
+      mapJobRow(job({ status: 'ready', rowsProcessed: 4, payload: { doomedCount: 10 } }))
+        .pendingDeleteRemaining
+    ).toBe(0)
+  })
+
+  it('ignores doomedCount for a running job that is not a delete', () => {
+    expect(
+      mapJobRow(job({ type: 'import', rowsProcessed: 4, payload: { doomedCount: 10 } }))
+        .pendingDeleteRemaining
+    ).toBe(0)
+  })
+
+  it('treats a missing doomedCount as zero and never goes negative', () => {
+    expect(mapJobRow(job({ rowsProcessed: 4 })).pendingDeleteRemaining).toBe(0)
+    expect(
+      mapJobRow(job({ rowsProcessed: 25, payload: { doomedCount: 10 } })).pendingDeleteRemaining
+    ).toBe(0)
+  })
+
+  it('carries a failed job error through', () => {
+    expect(mapJobRow(job({ status: 'failed', error: 'boom' }))).toMatchObject({
+      jobStatus: 'failed',
+      jobError: 'boom',
+    })
+  })
+})
+
+/**
+ * The lateral is a raw `sql` fragment, so the mocked drizzle `sql` tag is the only
+ * place its text is observable — and the text IS the contract (`getTableById` would
+ * otherwise silently return a different job than `latestJobsForTables` does).
+ */
+function renderLateral(): { text: string; values: unknown[] } {
+  // double-cast-allowed: the mocked drizzle `sql` tag exposes the raw template parts
+  const fragment = latestNonExportJobJson(schemaMock.userTableDefinitions.id) as unknown as {
+    strings: string[]
+    values: unknown[]
+  }
+  return { text: fragment.strings.join(' ? ').replace(/\s+/g, ' '), values: fragment.values }
+}
+
+describe('latestNonExportJobJson', () => {
+  it('excludes export jobs', () => {
+    expect(renderLateral().text).toContain("<> 'export'")
+  })
+
+  it('takes the single newest job by started_at', () => {
+    const { text, values } = renderLateral()
+    expect(text).toContain('order by ? desc')
+    expect(text).toContain('limit 1')
+    expect(values).toContain(schemaMock.tableJobs.startedAt)
+  })
+
+  it('correlates the subquery to the outer table id', () => {
+    const { text, values } = renderLateral()
+    expect(text).toContain('where ? = ?')
+    expect(values).toContain(schemaMock.tableJobs.tableId)
+    expect(values).toContain(schemaMock.userTableDefinitions.id)
+  })
+
+  it('projects every field mapJobRow reads', () => {
+    const { text } = renderLateral()
+    for (const key of ['id', 'type', 'status', 'rowsProcessed', 'error', 'payload']) {
+      expect(text).toContain(`'${key}',`)
+    }
+  })
+})

apps/sim/lib/table/jobs/service.ts

@@ -13,8 +13,8 @@
 
 import { db } from '@sim/db'
 import { tableJobs, userTableDefinitions, userTableRows } from '@sim/db/schema'
+import type { Column, SQL } from 'drizzle-orm'
 import { and, asc, desc, eq, gt, inArray, ne, or, sql } from 'drizzle-orm'
-import type { DbOrTx } from '@/lib/db/types'
 import { pendingDeleteMask } from '@/lib/table/rows/pending-delete-mask'
 import type {
   RowData,
@@ -25,7 +25,7 @@ import type {
 } from '@/lib/table/types'
 
 /** Job fields projected onto a {@link TableDefinition}, derived from its latest `table_jobs` row. */
-interface DerivedJobFields {
+export interface DerivedJobFields {
   jobStatus: TableDefinition['jobStatus']
   jobId: string | null
   jobType: TableDefinition['jobType']
@@ -49,18 +49,22 @@ export const EMPTY_JOB_FIELDS: DerivedJobFields = {
   pendingDeleteRemaining: 0,
 }
 
-function mapJobRow(
-  row:
-    | {
-        id: string
-        type: string
-        status: string
-        rowsProcessed: number
-        error: string | null
-        payload: unknown
-      }
-    | undefined
-): DerivedJobFields {
+/**
+ * The shape every latest-job read produces, whether it comes back as query columns
+ * (the batch `DISTINCT ON`) or as one jsonb object (the correlated lateral folded
+ * into the table SELECT). The single source of truth for the doomed-count rule is
+ * {@link mapJobRow} — never re-derive `pendingDeleteRemaining` at a call site.
+ */
+export interface LatestJobRow {
+  id: string
+  type: string
+  status: string
+  rowsProcessed: number
+  error: string | null
+  payload: unknown
+}
+
+export function mapJobRow(row: LatestJobRow | null | undefined): DerivedJobFields {
   if (!row) return EMPTY_JOB_FIELDS
   const doomedCount =
     row.type === 'delete' && row.status === 'running'
@@ -86,22 +90,36 @@ const JOB_PROJECTION = {
 } as const
 
 /**
- * The latest job for one table (the running one if present, else the most recent terminal).
- * Exports are excluded: they're read-only, run concurrently with other jobs, and have their own
- * client surface — surfacing one here would clobber the import/delete/backfill status the tray
- * and SSE consumer derive from these fields.
+ * The latest non-export job for one table, as a single jsonb value correlated to
+ * `outerTableId` — i.e. a `LEFT JOIN LATERAL (... LIMIT 1) ON true` expressed in the
+ * select list, which is the form drizzle can type without `leftJoinLateral`.
+ *
+ * It exists so {@link getTableById} stays ONE database round trip. With prepared
+ * statements disabled (PgBouncer transaction mode) every extra `await` is a full
+ * round trip, and `getTableById` is on essentially every table request. The job row
+ * cannot simply be skipped: a table's reported `rowCount` is the stored count minus
+ * this job's `pendingDeleteRemaining`, so the count and the job row are one read.
+ *
+ * Semantics match the batch {@link latestJobsForTables} exactly — exports excluded
+ * (they run concurrently and have their own client surface), newest `started_at`
+ * first, one row. `NULL` when the table has no such job; feed the result straight to
+ * {@link mapJobRow}.
  */
-export async function latestJobForTable(
-  tableId: string,
-  executor: DbOrTx = db
-): Promise<DerivedJobFields> {
-  const [row] = await executor
-    .select(JOB_PROJECTION)
-    .from(tableJobs)
-    .where(and(eq(tableJobs.tableId, tableId), ne(tableJobs.type, 'export')))
-    .orderBy(desc(tableJobs.startedAt))
-    .limit(1)
-  return mapJobRow(row)
+export function latestNonExportJobJson(outerTableId: Column | SQL): SQL<LatestJobRow | null> {
+  return sql<LatestJobRow | null>`(
+    select jsonb_build_object(
+      'id', ${tableJobs.id},
+      'type', ${tableJobs.type},
+      'status', ${tableJobs.status},
+      'rowsProcessed', ${tableJobs.rowsProcessed},
+      'error', ${tableJobs.error},
+      'payload', ${tableJobs.payload}
+    )
+    from ${tableJobs}
+    where ${tableJobs.tableId} = ${outerTableId} and ${tableJobs.type} <> 'export'
+    order by ${tableJobs.startedAt} desc
+    limit 1
+  )`
 }
 
 /** Latest non-export job per table for a batch of ids, via `DISTINCT ON (table_id)`. */