improvement(tables): read row counts and versions through an append-only change log

Author: TheodoreSpeaks

apps/sim/app/api/cron/fold-table-row-changes/route.ts

@@ -0,0 +1,26 @@
+import { createLogger } from '@sim/logger'
+import { type NextRequest, NextResponse } from 'next/server'
+import { verifyCronAuth } from '@/lib/auth/internal'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { foldPendingTableRowChanges } from '@/lib/table/row-changes'
+
+const logger = createLogger('FoldTableRowChangesApi')
+
+export const dynamic = 'force-dynamic'
+export const maxDuration = 60
+
+/** Leaves the route's `maxDuration` room for a fold started at the deadline to finish. */
+const FOLD_SWEEP_BUDGET_MS = 45_000
+
+/**
+ * Folds the table row-change log into each table's definition row. Each fold is one short
+ * statement, so the sweep runs in the request; overlapping sweeps skip each other's tables.
+ */
+export const GET = withRouteHandler(async (request: NextRequest) => {
+  const authError = verifyCronAuth(request, 'Table row-change fold')
+  if (authError) return authError
+
+  const result = await foldPendingTableRowChanges(FOLD_SWEEP_BUDGET_MS)
+  logger.info('Table row-change fold sweep completed', { ...result })
+  return NextResponse.json({ success: true, ...result })
+})

apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts

@@ -706,8 +706,8 @@ export async function copyForkResourceContainers(
         schema: remappedSchema,
         createdBy: userId,
         rowsVersion: 0,
-        // Start at 0 - the post-commit content copy raises it to the rows actually
-        // copied, so a failed/partial copy never advertises the source's count.
+        // Start at 0 - the row-count trigger counts the rows the content copy actually
+        // inserts, so a failed/partial copy never advertises the source's count.
         rowCount: 0,
         // Locks are workspace-local governance and never transit a fork edge —
         // mirrors `copy-workflows.ts` writing `locked: false`. Inheriting them
@@ -1343,10 +1343,6 @@ export async function copyForkResourceContent(params: {
         }
         if (rows.length < PROVENANCE_CONTENT_PAGE) break
       }
-      await db
-        .update(userTableDefinitions)
-        .set({ rowCount: copied })
-        .where(eq(userTableDefinitions.id, table.childId))
       await completeForkCopyResource(control, `table:${table.childId}`)
       copiedResources += 1
     } catch (error) {

apps/sim/lib/table/row-changes.integration.ts

@@ -0,0 +1,184 @@
+/**
+ * The table row-change log against real PostgreSQL: reads add the unfolded tail, and the fold
+ * moves it into the definition row without changing what a reader sees, under concurrent appends.
+ */
+import { db } from '@sim/db'
+import { userTableDefinitions } from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { generateId } from '@sim/utils/id'
+import postgres from 'postgres'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import {
+  foldPendingTableRowChanges,
+  foldTableRowChanges,
+  readCurrentRowsVersion,
+} from '@/lib/table/row-changes'
+import { getTableById, listTables } from '@/lib/table/service'
+
+const url = readTestDatabaseUrl()
+if (process.env.DATABASE_URL !== url) {
+  throw new Error('This suite requires only the disposable local test database')
+}
+const control = postgres(url, { max: 8, onnotice: () => {} })
+const workspaceId = generateId()
+const userId = generateId()
+
+async function createTable(): Promise<string> {
+  const id = generateId()
+  await db
+    .insert(userTableDefinitions)
+    .values({ id, workspaceId, name: id, schema: { columns: [] }, createdBy: userId })
+  return id
+}
+
+async function logChanges(tableId: string, ...deltas: number[]) {
+  for (const delta of deltas) {
+    await control`INSERT INTO user_table_row_changes (table_id, row_delta) VALUES (${tableId}, ${delta})`
+  }
+}
+
+async function stored(tableId: string) {
+  const [row] = await control<{ row_count: number; rows_version: string; updated_at: Date }[]>`
+    SELECT row_count, rows_version, updated_at FROM user_table_definitions WHERE id = ${tableId}`
+  return {
+    rowCount: row.row_count,
+    rowsVersion: Number(row.rows_version),
+    updatedAt: row.updated_at,
+  }
+}
+
+async function tailLength(tableId: string): Promise<number> {
+  const [{ n }] = await control<{ n: number }[]>`
+    SELECT count(*)::int AS n FROM user_table_row_changes WHERE table_id = ${tableId}`
+  return n
+}
+
+async function liveRowCount(tableId: string): Promise<number> {
+  const table = await getTableById(tableId)
+  if (!table) throw new Error('Fixture table missing')
+  return table.rowCount
+}
+
+describe('table row-change log against real PostgreSQL', () => {
+  beforeAll(async () => {
+    await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
+      VALUES (${userId}, 'Row change fixture', ${`${userId}@example.test`}, true, now(), now())`
+    await control`INSERT INTO workspace (id, name, owner_id, billed_account_user_id)
+      VALUES (${workspaceId}, 'Row change fixtures', ${userId}, ${userId})`
+  })
+
+  afterAll(async () => {
+    await control`DELETE FROM workspace WHERE id = ${workspaceId}`
+    await control`DELETE FROM "user" WHERE id = ${userId}`
+    await control.end()
+  })
+
+  it('reads the stored count and version plus the unfolded tail', async () => {
+    const tableId = await createTable()
+    await logChanges(tableId, 5, -2, 0)
+
+    expect(await liveRowCount(tableId)).toBe(3)
+    expect(await readCurrentRowsVersion(tableId)).toBe(3)
+    expect(await readCurrentRowsVersion(tableId, workspaceId)).toBe(3)
+    expect(await readCurrentRowsVersion(tableId, generateId())).toBeNull()
+    const listed = (await listTables(workspaceId)).find((table) => table.id === tableId)
+    expect(listed?.rowCount).toBe(3)
+  })
+
+  it('folds the tail into the definition row without changing what readers see', async () => {
+    const tableId = await createTable()
+    const before = await stored(tableId)
+    await logChanges(tableId, 4, 0, -1)
+
+    expect(await foldTableRowChanges(tableId)).toBe(true)
+
+    expect(await tailLength(tableId)).toBe(0)
+    const after = await stored(tableId)
+    expect(after.rowCount).toBe(3)
+    expect(after.rowsVersion).toBe(3)
+    expect(after.updatedAt.getTime()).toBeGreaterThanOrEqual(before.updatedAt.getTime())
+    expect(await liveRowCount(tableId)).toBe(3)
+    expect(await readCurrentRowsVersion(tableId)).toBe(3)
+  })
+
+  it('leaves updated_at alone when the tail holds only updates', async () => {
+    const tableId = await createTable()
+    const before = await stored(tableId)
+    await logChanges(tableId, 0, 0)
+
+    await foldTableRowChanges(tableId)
+
+    const after = await stored(tableId)
+    expect(after.rowsVersion).toBe(2)
+    expect(after.updatedAt.getTime()).toBe(before.updatedAt.getTime())
+  })
+
+  it('skips a table whose definition row is held instead of waiting on it', async () => {
+    const tableId = await createTable()
+    await logChanges(tableId, 2)
+
+    const held = control.begin(async (holder) => {
+      await holder`SELECT 1 FROM user_table_definitions WHERE id = ${tableId} FOR NO KEY UPDATE`
+      const started = Date.now()
+      const folded = await foldTableRowChanges(tableId)
+      return { folded, elapsedMs: Date.now() - started }
+    })
+    const { folded, elapsedMs } = await held
+
+    expect(folded).toBe(false)
+    expect(elapsedMs).toBeLessThan(1_000)
+    expect(await tailLength(tableId)).toBe(1)
+    expect(await liveRowCount(tableId)).toBe(2)
+  })
+
+  it('keeps the live version exact and non-decreasing while writers append during folds', async () => {
+    const tableId = await createTable()
+    const writers = 8
+    const appendsPerWriter = 40
+    let appending = true
+
+    const appenders = Array.from({ length: writers }, async () => {
+      for (let i = 0; i < appendsPerWriter; i++) await logChanges(tableId, 1)
+    })
+    const folder = (async () => {
+      while (appending) await foldTableRowChanges(tableId)
+    })()
+    const versionsSeen: number[] = []
+    const reader = (async () => {
+      while (appending) {
+        const version = await readCurrentRowsVersion(tableId)
+        if (version === null) throw new Error('Fixture table missing')
+        versionsSeen.push(version)
+      }
+    })()
+
+    await Promise.all(appenders)
+    appending = false
+    await Promise.all([folder, reader])
+
+    const total = writers * appendsPerWriter
+    expect(await readCurrentRowsVersion(tableId)).toBe(total)
+    expect(await liveRowCount(tableId)).toBe(total)
+    for (let i = 1; i < versionsSeen.length; i++) {
+      expect(versionsSeen[i]).toBeGreaterThanOrEqual(versionsSeen[i - 1])
+    }
+
+    await foldTableRowChanges(tableId)
+    expect(await tailLength(tableId)).toBe(0)
+    expect(await stored(tableId)).toMatchObject({ rowCount: total, rowsVersion: total })
+  })
+
+  it('sweeps every table with a tail', async () => {
+    const tableIds = [await createTable(), await createTable(), await createTable()]
+    for (const tableId of tableIds) await logChanges(tableId, 1, 1)
+
+    const result = await foldPendingTableRowChanges(30_000)
+
+    expect(result.budgetExhausted).toBe(false)
+    expect(result.folded).toBeGreaterThanOrEqual(tableIds.length)
+    for (const tableId of tableIds) {
+      expect(await tailLength(tableId)).toBe(0)
+      expect(await stored(tableId)).toMatchObject({ rowCount: 2, rowsVersion: 2 })
+    }
+  })
+})

apps/sim/lib/table/row-changes.ts

@@ -0,0 +1,137 @@
+/**
+ * A table's live `row_count` and `rows_version`: the values stored on `user_table_definitions`
+ * plus the tail of `user_table_row_changes` not yet folded into them.
+ *
+ * Row writes only append to the change log, so no writer waits on another for the shared
+ * definition row, which otherwise stays locked until a standby acknowledges the commit. The fold
+ * moves a table's log rows into the definition row and deletes them in one transaction, so stored
+ * value + tail is the same immediately before and after it. `updated_at` moves only at the fold, so
+ * it trails the last insert or delete by up to one sweep.
+ *
+ * Three rules keep `rows_version` exact for the snapshot cache and its mount-safety check:
+ * - Read stored value and tail in ONE statement on the primary, so both come from one snapshot.
+ *   Two statements could straddle a fold and count its rows twice or not at all.
+ * - Only the fold deletes log rows.
+ * - The version is a count of log rows, never `max(id)`: ids are allocated
+ *   before commit, so a row that commits late can carry a lower id than one already seen.
+ */
+
+import { db } from '@sim/db'
+import { userTableDefinitions, userTableRowChanges } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { and, eq, notInArray, sql } from 'drizzle-orm'
+import { setTableTxTimeouts } from '@/lib/table/tx'
+
+const logger = createLogger('TableRowChanges')
+
+/** Tables folded per sweep query; the sweep loops until the log is empty or its budget runs out. */
+const FOLD_SWEEP_PAGE_SIZE = 500
+
+/**
+ * Live row count, for a select over `user_table_definitions`. Spelled with explicit table names:
+ * drizzle prints bare column names in a subquery, where `id` would bind to the log's own column.
+ */
+export const currentRowCountSql = sql<number>`(user_table_definitions.row_count + coalesce((
+  select sum(c.row_delta) from user_table_row_changes c
+  where c.table_id = user_table_definitions.id
+), 0))::integer`.mapWith(Number)
+
+/** Live rows version, for a select over `user_table_definitions`. */
+const currentRowsVersionSql = sql<number>`(user_table_definitions.rows_version + (
+  select count(*) from user_table_row_changes c
+  where c.table_id = user_table_definitions.id
+))::bigint`.mapWith(Number)
+
+/**
+ * The table's live `rows_version`, or `null` when no such table exists (in `workspaceId`, when
+ * given). Always reads the primary.
+ */
+export async function readCurrentRowsVersion(
+  tableId: string,
+  workspaceId?: string
+): Promise<number | null> {
+  const [row] = await db
+    .select({ rowsVersion: currentRowsVersionSql })
+    .from(userTableDefinitions)
+    .where(
+      workspaceId
+        ? and(
+            eq(userTableDefinitions.id, tableId),
+            eq(userTableDefinitions.workspaceId, workspaceId)
+          )
+        : eq(userTableDefinitions.id, tableId)
+    )
+    .limit(1)
+  return row?.rowsVersion ?? null
+}
+
+/**
+ * Folds one table's change log into its definition row. Skips the table, returning `false`, when
+ * another transaction holds the definition row (a schema change, or a concurrent fold), so the fold
+ * never waits on it; the next sweep retries.
+ */
+export async function foldTableRowChanges(tableId: string): Promise<boolean> {
+  return db.transaction(async (tx) => {
+    await setTableTxTimeouts(tx)
+    const rows = await tx.execute<{ locked: boolean }>(sql`
+      WITH target AS (
+        SELECT id FROM user_table_definitions WHERE id = ${tableId}
+        FOR NO KEY UPDATE SKIP LOCKED
+      ), folded AS (
+        DELETE FROM user_table_row_changes c USING target
+        WHERE c.table_id = target.id
+        RETURNING c.row_delta
+      ), totals AS (
+        SELECT count(*) AS n,
+               coalesce(sum(row_delta), 0) AS delta,
+               bool_or(row_delta <> 0) AS rows_added_or_removed
+        FROM folded
+      ), applied AS (
+        UPDATE user_table_definitions d SET
+          rows_version = d.rows_version + totals.n,
+          row_count = greatest(d.row_count + totals.delta, 0),
+          updated_at = CASE WHEN totals.rows_added_or_removed
+            THEN timezone('UTC', now()) ELSE d.updated_at END
+        FROM target, totals
+        WHERE d.id = target.id AND totals.n > 0
+      )
+      SELECT EXISTS (SELECT 1 FROM target) AS locked`)
+    return rows[0]?.locked === true
+  })
+}
+
+export interface FoldSweepResult {
+  folded: number
+  skipped: number
+  budgetExhausted: boolean
+}
+
+/**
+ * Folds each table with unfolded log rows once, until none is left or `budgetMs` elapses. Rows a
+ * table logs after its fold wait for the next sweep.
+ */
+export async function foldPendingTableRowChanges(budgetMs: number): Promise<FoldSweepResult> {
+  const deadline = Date.now() + budgetMs
+  const visited = new Set<string>()
+  let folded = 0
+  let skipped = 0
+
+  for (;;) {
+    const pending = await db
+      .selectDistinct({ tableId: userTableRowChanges.tableId })
+      .from(userTableRowChanges)
+      .where(visited.size > 0 ? notInArray(userTableRowChanges.tableId, [...visited]) : undefined)
+      .limit(FOLD_SWEEP_PAGE_SIZE)
+    if (pending.length === 0) return { folded, skipped, budgetExhausted: false }
+
+    for (const { tableId } of pending) {
+      if (Date.now() >= deadline) {
+        logger.warn('Table row-change fold sweep ran out of budget', { folded, skipped, budgetMs })
+        return { folded, skipped, budgetExhausted: true }
+      }
+      visited.add(tableId)
+      if (await foldTableRowChanges(tableId)) folded++
+      else skipped++
+    }
+  }
+}

apps/sim/lib/table/rows/secret-provenance.integration.ts

@@ -189,6 +189,10 @@ describe('table provenance in PostgreSQL', () => {
     database.current = drizzle(connection, { schema })
     await connection.unsafe(`
       CREATE TABLE user_table_definitions (id text PRIMARY KEY, workspace_id text NOT NULL, rows_version integer NOT NULL, schema jsonb NOT NULL);
+      CREATE TABLE user_table_row_changes (
+        id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, table_id text NOT NULL,
+        row_delta integer NOT NULL
+      );
       CREATE TABLE user_table_rows (
         id text PRIMARY KEY, table_id text NOT NULL, workspace_id text NOT NULL,
         data jsonb NOT NULL, updated_at timestamp NOT NULL, secret_provenance_version integer,
@@ -226,7 +230,7 @@ describe('table provenance in PostgreSQL', () => {
 
   beforeEach(async () => {
     await connection.unsafe(
-      'TRUNCATE table_row_executions, user_table_rows, user_table_row_secret_provenance, user_table_definitions'
+      'TRUNCATE table_row_executions, user_table_rows, user_table_row_secret_provenance, user_table_row_changes, user_table_definitions'
     )
     await connection`INSERT INTO user_table_definitions VALUES ('table-1', 'workspace-1', 7, ${JSON.stringify(table.schema)}::jsonb)`
   })