Merge file ownership foundation staging alignment into Project files

Author: mzxchandra

apps/sim/background/cleanup-table-row-ttl.integration.ts

@@ -65,9 +65,10 @@ async function seedRows(tableId: string, count: number, value: string | null = e
 async function rowCount(tableId: string): Promise<number> {
   const [result] =
     await control`SELECT count(*)::int AS count FROM user_table_rows WHERE table_id = ${tableId}`
-  const [definition] =
-    await control`SELECT row_count FROM user_table_definitions WHERE id = ${tableId}`
-  expect(definition.row_count).toBe(result.count)
+  const [definition] = await control`SELECT d.row_count + coalesce(sum(c.row_delta), 0)::int AS live
+    FROM user_table_definitions d LEFT JOIN user_table_row_changes c ON c.table_id = d.id
+    WHERE d.id = ${tableId} GROUP BY d.id`
+  expect(definition.live).toBe(result.count)
   return result.count
 }
 

apps/sim/lib/projects/__integration__/foundation.integration.ts

@@ -121,7 +121,7 @@ async function enforce() {
   try {
     const migration = await readFile(
       new URL(
-        '../../../../../packages/db/migrations/0402_project_membership_enforcement.sql',
+        '../../../../../packages/db/migrations/0403_project_membership_enforcement.sql',
         import.meta.url
       ),
       'utf8'

apps/sim/lib/table/rows/ordering.ts

@@ -623,9 +623,8 @@ export async function guardBatch(
 
 /**
  * Deletes one page of rows for the async delete-job worker, committing each `DELETE_BATCH_SIZE`
- * chunk in its own short transaction. One statement per transaction bounds how long the
- * statement-level row_count trigger's lock on the definition row is held (a page-wide transaction
- * held it for the entire page, starving concurrent inserts and overrunning `statement_timeout`),
+ * chunk in its own short transaction. One statement per transaction keeps each transaction short
+ * (a page-wide transaction held its row locks for the entire page and overran `statement_timeout`),
  * and a mid-page failure loses at most one uncommitted batch — the keyset walker (or a task
  * retry) re-walks whatever remains. Skips legacy position compaction: under fractional ordering
  * it's unnecessary, and in the legacy path `position` gaps are harmless — rows still order by

apps/sim/lib/table/rows/row-writes.integration.ts

@@ -26,11 +26,13 @@ import { deleteColumn, updateColumnConstraints } from '@/lib/table/columns/servi
 import { getMaxRowSizeBytes } from '@/lib/table/constants'
 import { bulkInsertImportBatch, importReplaceRows } from '@/lib/table/import-data'
 import type { DbTransaction } from '@/lib/table/planner'
+import { readCurrentRowsVersion } from '@/lib/table/row-changes'
 import { lockLiveTableSchema } from '@/lib/table/rows/live-schema'
 import { acquireRowOrderLock } from '@/lib/table/rows/ordering'
 import {
   batchInsertRows,
   batchUpdateRows,
+  deleteRowsByFilter,
   insertRow,
   replaceTableRows,
   updateRow,
@@ -66,6 +68,12 @@ const [{ migrated }] = await control<{ migrated: boolean }[]>`SELECT EXISTS (
     AND p.proname = 'bump_user_table_rows_version_at_commit'
 ) AS migrated`
 
+/** Whether the row triggers log to `user_table_row_changes` (0402) rather than lock the definition. */
+const [{ logsRowChanges }] = await control<{ logsRowChanges: boolean }[]>`SELECT EXISTS (
+  SELECT 1 FROM pg_proc
+  WHERE proname = 'increment_user_table_row_count_stmt' AND prosrc LIKE '%user_table_row_changes%'
+) AS "logsRowChanges"`
+
 async function createTable(columns: ColumnDefinition[]): Promise<TableDefinition> {
   const id = generateId()
   await db
@@ -84,8 +92,9 @@ async function seedRows(
 }
 
 async function rowsVersion(tableId: string): Promise<number> {
-  const [row] = await control`SELECT rows_version FROM user_table_definitions WHERE id = ${tableId}`
-  return Number(row.rows_version)
+  const version = await readCurrentRowsVersion(tableId)
+  if (version === null) throw new Error('Fixture table missing')
+  return version
 }
 
 const textColumns = (...ids: string[]): ColumnDefinition[] =>
@@ -1389,6 +1398,160 @@ describe('table row writes against real PostgreSQL', () => {
     })
   })
 
+  describe.skipIf(!migrated || !logsRowChanges)('a held definition row', () => {
+    it('lets every row write path commit while another session holds the definition row', async () => {
+      const table = await createTable([
+        { id: 'key', name: 'key', type: 'string', unique: true },
+        { id: 'name', name: 'name', type: 'string' },
+      ])
+      const versionBefore = await rowsVersion(table.id)
+      const rowIdByKey = async (key: string) => {
+        const [row] = await control<{ id: string }[]>`SELECT id FROM user_table_rows
+          WHERE table_id = ${table.id} AND data->>'key' = ${key}`
+        return row.id
+      }
+      const writes: Array<[string, () => Promise<unknown>]> = [
+        [
+          'insert',
+          () =>
+            insertRow(
+              {
+                tableId: table.id,
+                workspaceId,
+                data: { key: 'a', name: 'a' },
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              table,
+              'held-insert'
+            ),
+        ],
+        [
+          'batch insert',
+          () =>
+            batchInsertRows(
+              {
+                tableId: table.id,
+                workspaceId,
+                rows: [
+                  { key: 'b', name: 'b' },
+                  { key: 'c', name: 'c' },
+                ],
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              table,
+              'held-batch-insert'
+            ),
+        ],
+        [
+          'upsert',
+          () =>
+            upsertRow(
+              {
+                tableId: table.id,
+                workspaceId,
+                data: { key: 'd', name: 'd' },
+                conflictTarget: 'key',
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              table,
+              'held-upsert'
+            ),
+        ],
+        [
+          'upsert of an existing key',
+          () =>
+            upsertRow(
+              {
+                tableId: table.id,
+                workspaceId,
+                data: { key: 'd', name: 'd2' },
+                conflictTarget: 'key',
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              table,
+              'held-upsert-existing'
+            ),
+        ],
+        [
+          'update by id',
+          async () =>
+            updateRow(
+              {
+                tableId: table.id,
+                rowId: await rowIdByKey('a'),
+                workspaceId,
+                data: { name: 'a2' },
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              table,
+              'held-update'
+            ),
+        ],
+        [
+          'update by filter',
+          () =>
+            updateRowsByFilter(
+              table,
+              {
+                filter: { name: 'b' },
+                data: { name: 'b2' },
+                limit: 10,
+                secretProvenance: undefined,
+                capabilityGovernedUserId: null,
+              },
+              'held-update-by-filter'
+            ),
+        ],
+        [
+          'delete by filter',
+          () => deleteRowsByFilter(table, { filter: { name: 'c' } }, 'held-delete-by-filter'),
+        ],
+        [
+          'replace',
+          () =>
+            replaceTableRows(
+              {
+                tableId: table.id,
+                workspaceId,
+                rows: [{ key: 'x', name: 'x' }],
+                secretProvenance: undefined,
+              },
+              table,
+              'held-replace'
+            ),
+        ],
+      ]
+
+      const holder = await control.reserve()
+      const elapsedMs: Record<string, number> = {}
+      try {
+        await holder`BEGIN`
+        await holder`SELECT 1 FROM user_table_definitions WHERE id = ${table.id} FOR NO KEY UPDATE`
+        for (const [name, write] of writes) {
+          const started = Date.now()
+          await write()
+          elapsedMs[name] = Date.now() - started
+        }
+      } finally {
+        await holder`ROLLBACK`.catch(() => {})
+        holder.release()
+      }
+
+      for (const [name] of writes) expect(elapsedMs[name], name).toBeLessThan(2_000)
+      const [{ count }] = await control<{ count: number }[]>`SELECT count(*)::int AS count
+        FROM user_table_rows WHERE table_id = ${table.id}`
+      expect(count).toBe(1)
+      expect((await getTableById(table.id))?.rowCount).toBe(count)
+      // One log entry per write, except replace, which logs its DELETE and its INSERT.
+      expect(await rowsVersion(table.id)).toBe(versionBefore + writes.length + 1)
+    })
+  })
+
   describe.skipIf(!migrated)('rows_version', () => {
     it('advances once for a transaction that edits cells across several statements', async () => {
       const table = await createTable(textColumns('name'))

apps/sim/lib/table/service.ts

@@ -123,9 +123,9 @@ function readLocks(row: {
  * validates and computes against the prior writer's committed columns.
  *
  * Uses an advisory lock (not `SELECT ... FOR UPDATE` on the definition row) so
- * it adds no edges to the row-lock graph — the row-count trigger (migration
- * 0198) locks the definition row from `insertRow`/`deleteRow`, and a FOR UPDATE
- * here would invert that order. Mirrors `acquireRowOrderLock`. The lock and
+ * it adds no edges to the row-lock graph — a FOR UPDATE here would also block the
+ * foreign-key check (KEY SHARE) of every row write to the table. Mirrors
+ * `acquireRowOrderLock`. The lock and
  * the read both release at COMMIT/ROLLBACK; the wait is bounded by the
  * `statement_timeout` set in `setTableTxTimeouts`.
  */

apps/sim/lib/workspace-files/search/README.md

@@ -69,7 +69,7 @@ The design uses PostgreSQL's documented [TOAST behavior](https://www.postgresql.
 
 ### File owner expansion
 
-Migration `0408_file_search_owner_scope.sql` adds canonical workspace/Project ownership to the existing build, revision, and chunk pipeline. Legacy workspace inserts are translated by triggers, and the existing workspace trigram index remains usable. Project rows use a separate owner-scoped trigram index on the same chunk table. Organizations and personal users are not enabled search owners.
+Migration `0409_file_search_owner_scope.sql` adds canonical workspace/Project ownership to the existing build, revision, and chunk pipeline. Legacy workspace inserts are translated by triggers, and the existing workspace trigram index remains usable. Project rows use a separate owner-scoped trigram index on the same chunk table. Organizations and personal users are not enabled search owners.
 
 The new `file_search_dispatch_queue` replaces the workspace-keyed scheduler because older workers cannot decode a Project without a workspace. During expansion, workspace mutations enqueue both queues, while the new dispatcher reads the owner queue. Both deployments share the dispatcher lock, revision claim token, build lease, and publication fence. Drain old app dispatchers and Trigger workers/retries before enabling Project dispatch. Retire the legacy queue and its dual enqueue in a later contract deploy; they are not a second indexing pipeline. The `file-search-owners-v3` cursor performs bounded owner/file keyset backfill and hourly reconciliation without rewriting extracted text in the migration.
 

apps/sim/lib/workspace-files/search/chunks.integration.ts

@@ -173,7 +173,7 @@ describe('chunked workspace file search on PostgreSQL', () => {
       '0359_workspace_file_search_chunks.sql',
       ginWriteMigration,
       '0382_workspace_file_search_dispatch_handoff.sql',
-      '0408_file_search_owner_scope.sql',
+      '0409_file_search_owner_scope.sql',
     ]) {
       await applyMigration(migration)
     }

apps/sim/lib/workspace-files/search/dispatcher.integration.ts

@@ -70,7 +70,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     })
   )
   const migration = readFileSync(
-    resolve(process.cwd(), '../../packages/db/migrations/0408_file_search_owner_scope.sql'),
+    resolve(process.cwd(), '../../packages/db/migrations/0409_file_search_owner_scope.sql'),
     'utf8'
   ).replaceAll('"public".', `"${schemaName}".`)
   const ownerIndexes = [

packages/db/file-creator-lifetime.integration.ts

@@ -9,9 +9,9 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'
 
 const databaseUrl = readTestDatabaseUrl()
 const migrations = [
-  '0403_file_entity_ownership.sql',
-  '0404_file_folder_version_ownership.sql',
-  '0405_file_creator_lifetime.sql',
+  '0404_file_entity_ownership.sql',
+  '0405_file_folder_version_ownership.sql',
+  '0406_file_creator_lifetime.sql',
 ].map((name) => readFileSync(new URL(`./migrations/${name}`, import.meta.url), 'utf8'))
 const checks: { name: string; status: 'passed' | 'failed'; durationMs: number; error?: string }[] =
   []

packages/db/file-entity-ownership.integration.ts

@@ -10,12 +10,12 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'
 
 const databaseUrl = readTestDatabaseUrl()
 const migration = readFileSync(
-  new URL('./migrations/0403_file_entity_ownership.sql', import.meta.url),
+  new URL('./migrations/0404_file_entity_ownership.sql', import.meta.url),
   'utf8'
 )
 
 const workspaceBindingMigration = readFileSync(
-  new URL('./migrations/0409_file_workspace_binding.sql', import.meta.url),
+  new URL('./migrations/0410_file_workspace_binding.sql', import.meta.url),
   'utf8'
 )