Skip to content

Commit 873c4a5

Browse files
waleedlatif1waleedlatif1
committed
fix(tables): install the schema guard for db:push, refit rows from raw input
- user_table_schema_for_write moves from Drizzle migration 0391 to script migration 0026, which db:push runs too. A db:push database (local dev, the CI push provision) never applied 0391, so every guarded row write failed there. The journal ends at 0390 again; the function body and its comments are unchanged. - A writer that finds the schema moved rebuilds the row from the caller's raw input instead of the value it coerced against its snapshot, so a "007" sent while a column changed from number to text is stored as "007", not "7". This covers insert, upsert, update, batch update and the import batch. - The refit re-checks the row's size, which a coercion to a wider type can grow past the limit after the pre-lock check passed.
1 parent c687293 commit 873c4a5

12 files changed

Lines changed: 224 additions & 29447 deletions

‎apps/sim/lib/table/import-data.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ export async function bulkInsertImportBatch(
8484
// the caller's snapshot too would reject a since-cleared lock.
8585
if (!revalidate) assertRowInsert(table)
8686

87+
const rawRows = data.rows.map((row) => ({ ...row }))
8788
for (let i = 0; i < data.rows.length; i++) {
8889
const sizeValidation = validateRowSize(data.rows[i])
8990
if (!sizeValidation.valid) {
@@ -125,7 +126,7 @@ export async function bulkInsertImportBatch(
125126
const live = fresh ? withLiveSchema(table, fresh.schema) : await lockLiveTableSchema(trx, table)
126127
if (live !== table) {
127128
for (let i = 0; i < data.rows.length; i++) {
128-
const refit = refitRowToSchema(data.rows[i], table.schema, live.schema, 'null')
129+
const refit = refitRowToSchema(data.rows[i], rawRows[i], table.schema, live.schema, 'null')
129130
if (!refit.valid) {
130131
throw new OrchestrationError('validation', `Row ${i + 1}: ${refit.errors.join(', ')}`)
131132
}

‎apps/sim/lib/table/rows/live-schema.ts‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import {
2323
coerceRowToSchema,
2424
type PatchedKeys,
2525
type UncoercibleValuePolicy,
26+
validateRowSize,
2627
} from '@/lib/table/validation'
2728

2829
/** Compares schemas by content, whatever order their columns are listed in. */
@@ -44,9 +45,9 @@ export function withLiveSchema(table: TableDefinition, schema: TableSchema): Tab
4445
* the live schema. Call it first in the transaction, before its other locks, passing the
4546
* transaction's `timeouts` (see `setTableTxTimeouts`) in place of a separate timeouts statement.
4647
*
47-
* One statement: `user_table_schema_for_write` (migration 0391) takes the lock and then reads the
48-
* schema. It is VOLATILE, so under READ COMMITTED its read takes a fresh snapshot and sees a schema
49-
* change that committed while the lock waited. The timeouts are applied in a subquery the call
48+
* One statement: `user_table_schema_for_write` (script migration 0026) takes the lock and then
49+
* reads the schema. It is VOLATILE, so under READ COMMITTED its read takes a fresh snapshot and sees
50+
* a schema change that committed while the lock waited. The timeouts are applied in a subquery the call
5051
* reads from, first. The lock waits as long as the transaction's `statement_timeout` allows, not
5152
* its shorter `lock_timeout`, which the function leaves as it found it for the locks that follow.
5253
*/
@@ -81,18 +82,23 @@ export function dropDeletedColumns(
8182
}
8283

8384
/**
84-
* Re-validates, in place, a row prepared against `snapshot` once the schema has moved to `live`:
85-
* drops the cells of deleted columns, then coerces and validates against `live` exactly as the
86-
* write first did. Coercion leaves a value it already produced unchanged, the property every
87-
* merged-row write relies on when it re-coerces stored cells.
85+
* Rebuilds `row`, in place, for `live` once the schema has moved since `snapshot`: from `raw`, the
86+
* row as the caller wrote it before coercing it against `snapshot`, so a value that schema would
87+
* have reshaped (`"007"` read as a number) reaches the live column as it was sent. Then drops the
88+
* cells of deleted columns, coerces and validates against `live` exactly as the write first did,
89+
* and re-checks the row's size, which a coercion to a wider type can grow.
8890
*/
8991
export function refitRowToSchema(
9092
row: RowData,
93+
raw: RowData,
9194
snapshot: TableSchema,
9295
live: TableSchema,
9396
policy?: UncoercibleValuePolicy,
9497
patchedKeys?: PatchedKeys
9598
): ValidationResult {
99+
for (const key of Object.keys(row)) delete row[key]
100+
Object.assign(row, raw)
96101
dropDeletedColumns([row], snapshot, live)
97-
return coerceRowToSchema(row, live, policy, patchedKeys)
102+
const result = coerceRowToSchema(row, live, policy, patchedKeys)
103+
return result.valid ? validateRowSize(row) : result
98104
}

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

Lines changed: 130 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import {
2727
updateColumnConstraints,
2828
updateColumnOptions,
2929
} from '@/lib/table/columns/service'
30-
import { TABLE_LIMITS } from '@/lib/table/constants'
30+
import { getMaxRowSizeBytes, TABLE_LIMITS } from '@/lib/table/constants'
3131
import { bulkInsertImportBatch, importReplaceRows } from '@/lib/table/import-data'
3232
import { markTableJobRunningInWorkspace } from '@/lib/table/jobs/service'
3333
import type { DbTransaction } from '@/lib/table/planner'
@@ -921,6 +921,135 @@ describe('table row writes against real PostgreSQL', () => {
921921
expect(count).toBe(0)
922922
})
923923

924+
/** A table whose `code` column is text, and a snapshot from when it was a number. */
925+
async function retypedTable(): Promise<{ table: TableDefinition; stale: TableDefinition }> {
926+
const table = await createTable([
927+
{ id: 'key', name: 'key', type: 'string', unique: true },
928+
{ id: 'code', name: 'code', type: 'string' },
929+
{ id: 'filler', name: 'filler', type: 'string' },
930+
])
931+
await seedRows(table.id, [
932+
{ id: `${table.id}-2`, data: { key: 'k2', filler: '' }, orderKey: 'a0' },
933+
])
934+
const stale: TableDefinition = {
935+
...table,
936+
schema: {
937+
columns: table.schema.columns.map((column) =>
938+
column.id === 'code' ? { ...column, type: 'number' as const } : column
939+
),
940+
},
941+
}
942+
return { table, stale }
943+
}
944+
945+
/** Writes `row` (a new row, or a patch to row `k2`) through each writer holding `stale`. */
946+
const retypeWriters: Array<
947+
[string, (table: TableDefinition, row: RowData) => Promise<unknown>]
948+
> = [
949+
[
950+
'insertRow',
951+
(table, row) =>
952+
insertRow(
953+
{
954+
tableId: table.id,
955+
workspaceId,
956+
data: { key: 'k3', ...row },
957+
secretProvenance: undefined,
958+
capabilityGovernedUserId: null,
959+
},
960+
table,
961+
'retype'
962+
),
963+
],
964+
[
965+
'upsertRow',
966+
(table, row) =>
967+
upsertRow(
968+
{
969+
tableId: table.id,
970+
workspaceId,
971+
data: { key: 'k3', ...row },
972+
conflictTarget: 'key',
973+
secretProvenance: undefined,
974+
capabilityGovernedUserId: null,
975+
},
976+
table,
977+
'retype'
978+
),
979+
],
980+
[
981+
'bulkInsertImportBatch',
982+
(table, row) =>
983+
bulkInsertImportBatch(
984+
{ tableId: table.id, workspaceId, rows: [{ key: 'k3', ...row }], startPosition: 1 },
985+
table,
986+
'retype'
987+
),
988+
],
989+
[
990+
'updateRow',
991+
(table, row) =>
992+
updateRow(
993+
{
994+
tableId: table.id,
995+
rowId: `${table.id}-2`,
996+
workspaceId,
997+
data: row,
998+
secretProvenance: undefined,
999+
capabilityGovernedUserId: null,
1000+
},
1001+
table,
1002+
'retype'
1003+
),
1004+
],
1005+
[
1006+
'batchUpdateRows',
1007+
(table, row) =>
1008+
batchUpdateRows(
1009+
{
1010+
tableId: table.id,
1011+
workspaceId,
1012+
updates: [{ rowId: `${table.id}-2`, data: row }],
1013+
capabilityGovernedUserId: null,
1014+
},
1015+
table,
1016+
'retype'
1017+
),
1018+
],
1019+
]
1020+
1021+
it.each(retypeWriters)(
1022+
'%s stores the value it was sent in a column retyped since its snapshot',
1023+
async (_, write) => {
1024+
const { table, stale } = await retypedTable()
1025+
1026+
await write(stale, { code: '007' })
1027+
1028+
const [{ count }] = await control<{ count: number }[]>`SELECT count(*)::int AS count
1029+
FROM user_table_rows WHERE table_id = ${table.id} AND data->'code' = '"007"'`
1030+
expect(count).toBe(1)
1031+
}
1032+
)
1033+
1034+
it.each(retypeWriters)(
1035+
'%s refuses a row a retype since its snapshot grows past the size limit',
1036+
async (writer, write) => {
1037+
const { table, stale } = await retypedTable()
1038+
// Exactly at the limit with `code` a number; the live text column stores it with quotes.
1039+
const inserting = writer !== 'updateRow' && writer !== 'batchUpdateRows'
1040+
const shape = (filler: string) =>
1041+
inserting ? { key: 'k3', code: 7, filler } : { key: 'k2', filler, code: 7 }
1042+
const filler = 'x'.repeat(
1043+
getMaxRowSizeBytes() - Buffer.byteLength(JSON.stringify(shape('')))
1044+
)
1045+
1046+
await expect(write(stale, { code: 7, filler })).rejects.toThrow(/Row size exceeds limit/)
1047+
const [{ count }] = await control<{ count: number }[]>`SELECT count(*)::int AS count
1048+
FROM user_table_rows WHERE table_id = ${table.id} AND data->>'code' = '7'`
1049+
expect(count).toBe(0)
1050+
}
1051+
)
1052+
9241053
it('refuses a bulk update writing one value to rows of a column made unique since its snapshot', async () => {
9251054
const table = await seededTable()
9261055
await updateColumnConstraints(

‎apps/sim/lib/table/rows/service.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,7 @@ export async function insertRow(
178178
}
179179

180180
// Validate against schema
181+
const raw = { ...data.data }
181182
const schemaValidation = coerceRowToSchema(data.data, table.schema, options.uncoercibleValues)
182183
if (!schemaValidation.valid) {
183184
throw new OrchestrationError(
@@ -216,6 +217,7 @@ export async function insertRow(
216217
if (live !== table) {
217218
const refit = refitRowToSchema(
218219
data.data,
220+
raw,
219221
table.schema,
220222
live.schema,
221223
options.uncoercibleValues
@@ -794,6 +796,7 @@ export async function upsertRow(
794796
throw new OrchestrationError('validation', sizeValidation.errors.join(', '))
795797
}
796798

799+
const raw = { ...data.data }
797800
const schemaValidation = coerceRowToSchema(data.data, table.schema, options.uncoercibleValues)
798801
if (!schemaValidation.valid) {
799802
throw new OrchestrationError(
@@ -821,6 +824,7 @@ export async function upsertRow(
821824
if (live !== table) {
822825
const refit = refitRowToSchema(
823826
data.data,
827+
raw,
824828
table.schema,
825829
live.schema,
826830
options.uncoercibleValues
@@ -1886,6 +1890,7 @@ export async function updateRow(
18861890
if (live !== table) {
18871891
const refit = refitRowToSchema(
18881892
mergedData,
1893+
{ ...(existingRow.data as RowData), ...data.data },
18891894
table.schema,
18901895
live.schema,
18911896
options.uncoercibleValues,
@@ -2706,6 +2711,7 @@ export async function batchUpdateRows(
27062711
const existing = existingMap.get(update.rowId)!
27072712
const refit = refitRowToSchema(
27082713
update.mergedData,
2714+
{ ...existing.data, ...request.data },
27092715
table.schema,
27102716
live.schema,
27112717
options.uncoercibleValues,

‎packages/db/migrations/0391_user_table_schema_for_write.sql‎

Lines changed: 0 additions & 38 deletions
This file was deleted.

0 commit comments

Comments
 (0)