diff --git a/.changeset/batch-upsert-existence-fork.md b/.changeset/batch-upsert-existence-fork.md new file mode 100644 index 0000000000..c312ae7d32 --- /dev/null +++ b/.changeset/batch-upsert-existence-fork.md @@ -0,0 +1,35 @@ +--- +"@objectstack/metadata-protocol": patch +--- + +fix(metadata-protocol): `batchData`'s upsert fork decides update-or-insert by EXISTENCE, not caller visibility (#5099) + +The fork asked `findOne` under the CALLER's context — the read RLS/sharing +narrows (#3455). An existing row outside the caller's read scope therefore +answered `null` and took the INSERT arm: on a store with a unique id constraint +the insert duplicate-keyed (an authorization/update scenario reported as a key +collision — the same misdirection class as #5088), and on a store without one +it wrote a **second row** for an id that already exists. + +The fork now uses the same existence probe (`probeRecord`, system context) as +the single-record path and the update/delete bulk faces (#4620: one reading per +file). Whether the caller may WRITE the row it proves stays exactly where it +was — #1994's pre-image check inside `engine.update` — so the row's outcome is +the write policy's own answer instead of a spurious `duplicate key` error. + +**Observable change under row-level visibility**: upserting an id that exists +outside your read scope no longer attempts an insert. The row now answers +whatever the by-id update path answers for that record (for a masked pre-image +check, the same 404 a direct update returns). The existence oracle is not +widened: the previous duplicate-key failure already revealed that the id +exists. + +The non-atomic fallback (update threw → blind insert) is removed with it, on +both arms. With existence decided before the fork, the fallback could only +bury a real update failure under the duplicate-key error of inserting a row +just proven to exist — the same masking ADR-0119 D4 already forbade inside the +atomic arm. A row whose update fails now reports that failure. + +Cost note: each by-id upsert row now performs one existence read before the +write — the same probe cost #4435 accepted for the single-record path and +#5088 accepted for the update/delete bulk faces. diff --git a/packages/metadata-protocol/src/protocol.ts b/packages/metadata-protocol/src/protocol.ts index 99801072d8..75a6f74a9a 100644 --- a/packages/metadata-protocol/src/protocol.ts +++ b/packages/metadata-protocol/src/protocol.ts @@ -5582,11 +5582,14 @@ export class ObjectStackProtocolImplementation implements /** * The per-record loop, shared by both arms of {@link batchData} (ADR-0119 * D4) so atomic and non-atomic cannot drift apart. `atomic` changes exactly - * two things: it aborts on the first failure regardless of + * one thing: it aborts on the first failure regardless of * `continueOnError` (whose own contract text already scopes it to - * `atomic=false`), and it forbids the upsert fallback — inside an aborted - * transaction a fallback insert can only fail with a secondary error that - * masks the real cause. + * `atomic=false`). It used to change a second — forbidding the upsert + * fallback insert, whose failure inside an aborted transaction could only + * mask the real cause — until #5099 removed that fallback from BOTH arms: + * the upsert fork is decided by an existence probe before any write, so a + * fallback insert could only bury a real update failure under the + * duplicate-key error of inserting a row just proven to exist. */ private async runBatchDataLoop(args: { object: string; @@ -5640,25 +5643,34 @@ export class ObjectStackProtocolImplementation implements break; } case 'upsert': { - // Try update first, then create if not found if (record.id) { - try { - const existing = await this.engine.findOne(object, { where: { id: record.id }, ...ctxOpt } as any); - if (existing) { - const dropped: DroppedFieldsEvent[] = []; - const updated = await this.engine.update(object, record.data || {}, { where: { id: record.id }, onFieldsDropped: (e: DroppedFieldsEvent) => { dropped.push(e); }, ...ctxOpt } as any); - results.push({ id: record.id, success: true, data: updated, index, ...(dropped.length > 0 ? { droppedFields: dropped } : {}) }); - } else { - const created = await this.engine.insert(object, { id: record.id, ...(record.data || {}) }, insertCtx as any); - results.push({ id: created.id, success: true, data: created, index }); - } - } catch (err) { - // ADR-0119 D4 — no blind fallback inside a - // transaction: once the failing statement has - // aborted it, this insert can only fail with a - // secondary error ("current transaction is - // aborted") that buries the real cause. - if (atomic) throw err; + // [#5099] The update-or-insert fork asks EXISTENCE, + // not visibility. `findOne` under the CALLER's + // context is the read RLS/sharing narrows (#3455), + // so an existing row outside the caller's scope + // answered null, took the insert arm, and either + // duplicate-keyed — an authorization/update + // scenario reported as a key collision — or, on a + // store without a unique id constraint, wrote a + // second row. Same probe as the update branch above + // and the single-record path (#4620): it asks the + // database a fact, and whether the caller may WRITE + // the row it proves stays #1994's decision inside + // `engine.update`. + // + // The old fallback (update threw → blind insert) + // is gone with the fork's visibility read: with + // existence decided BEFORE the write, a fallback + // insert could only bury a real update failure + // under the duplicate-key error of inserting a row + // just proven to exist — the same masking ADR-0119 + // D4 forbade inside the atomic arm. + const existing = await this.probeRecord(object, record.id); + if (existing) { + const dropped: DroppedFieldsEvent[] = []; + const updated = await this.engine.update(object, record.data || {}, { where: { id: record.id }, onFieldsDropped: (e: DroppedFieldsEvent) => { dropped.push(e); }, ...ctxOpt } as any); + results.push({ id: record.id, success: true, data: updated, index, ...(dropped.length > 0 ? { droppedFields: dropped } : {}) }); + } else { const created = await this.engine.insert(object, { id: record.id, ...(record.data || {}) }, insertCtx as any); results.push({ id: created.id, success: true, data: created, index }); } diff --git a/packages/metadata-protocol/src/protocol.upsert-existence.test.ts b/packages/metadata-protocol/src/protocol.upsert-existence.test.ts new file mode 100644 index 0000000000..70be250204 --- /dev/null +++ b/packages/metadata-protocol/src/protocol.upsert-existence.test.ts @@ -0,0 +1,234 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#5099] `batchData`'s upsert fork asks EXISTENCE, not visibility. + * + * The fork used to ask `findOne` under the CALLER's context — the read that + * RLS/sharing narrows (#3455). An existing row outside the caller's read scope + * therefore answered `null` and took the INSERT arm: on a store with a unique + * id constraint the insert duplicate-keyed (an authorization/update scenario + * reported as a key collision — the same misdirection class as #5088), and on + * a store without one it wrote a second row. `probeRecord`'s own comment + * (#4435) calls the distinction load-bearing; #5088 put it on the update and + * delete faces of this file; the upsert fork was the one branch still reading + * it backwards (#4620: one reading per file, not three that happen to agree). + * + * So the fork now uses the same existence probe as every other by-id face, and + * whether the caller may WRITE the row it proves stays #1994's decision inside + * `engine.update` — the fake engine below models that pre-image check masking + * an out-of-scope by-id write as the same 404 the read path answers. + * + * The old fallback (update threw → blind insert) goes with it, on BOTH arms: + * existence is now decided before the fork, so a fallback insert could only + * bury a real update failure under the duplicate-key error of inserting a row + * just proven to exist — the exact masking ADR-0119 D4 already forbade inside + * the atomic arm. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { ObjectStackProtocolImplementation } from './protocol.js'; + +const SCHEMA = { + name: 'showcase_task', + fields: { + progress: { name: 'progress', type: 'number' }, + owner_id: { name: 'owner_id', type: 'text' }, + }, +}; + +/** + * In-memory store with a caller-scoped read path and a #1994-style by-id + * write policy: reads under a non-system context only see the caller's own + * rows, and a by-id write to a row outside the caller's scope is masked as + * the same 404 the read path answers. Snapshot/rollback transaction semantics + * as in the #4620 / #5088 harnesses. + */ +function makeRlsEngine() { + const rows = new Map([ + ['mine_1', { id: 'mine_1', owner_id: 'u1', progress: 0 }], + ['theirs_1', { id: 'theirs_1', owner_id: 'u2', progress: 0 }], + ]); + const handle = { id: 'trx-1' }; + + const visible = (row: any, context: any) => + context?.isSystem === true || row.owner_id === context?.userId; + + const findOne = vi.fn(async (_object: string, options?: any) => { + const row = rows.get(options?.where?.id) ?? null; + if (!row) return null; + return visible(row, options?.context) ? row : null; + }); + const update = vi.fn(async (_object: string, data: any, options?: any) => { + const id = options?.where?.id; + const row = rows.get(id); + if (!row || !visible(row, options?.context)) { + // #1994's pre-image check: an unwritable/unknown row masks as the + // read path's 404 — THIS answer must reach the response row. + throw Object.assign(new Error(`Record ${id} not found in showcase_task`), { + code: 'RECORD_NOT_FOUND', status: 404, + }); + } + const next = { ...row, ...data }; + rows.set(id, next); + return next; + }); + const insert = vi.fn(async (_object: string, data: any) => { + if (data.id && rows.has(data.id)) { + // Unclassified, as a driver throws it → `toRowApiError` renders + // INTERNAL_ERROR: the misdirection the probe exists to prevent. + throw new Error('duplicate key value violates unique constraint "showcase_task_pkey"'); + } + const rec = { id: data.id ?? `new-${rows.size + 1}`, ...data }; + rows.set(rec.id, rec); + return rec; + }); + + const engine: any = { + registry: { getObject: (n: string) => (n === 'showcase_task' ? SCHEMA : undefined) }, + findOne, + update, + insert, + delete: vi.fn(), + getDefaultDriverName: () => 'default', + getDriverByName: () => ({ beginTransaction: async () => handle }), + transaction: vi.fn(async (callback: (ctx: any) => Promise, baseContext?: any) => { + const snapshot = new Map(rows); + try { + return await callback({ ...(baseContext ?? {}), transaction: handle }); + } catch (err) { + rows.clear(); + for (const [k, v] of snapshot) rows.set(k, v); + throw err; + } + }), + }; + return { engine, rows, findOne, update, insert }; +} + +const upsert = (p: any, records: any[], options?: any) => + p.batchData({ + object: 'showcase_task', + request: { operation: 'upsert', records, ...(options ? { options } : {}) }, + context: { userId: 'u1' }, + } as any); + +describe('[#5099] batchData upsert — the fork asks EXISTENCE, not visibility', () => { + it('an existing row outside the caller`s read scope takes the UPDATE arm — never a duplicate insert', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [{ id: 'theirs_1', data: { progress: 5 } }], { continueOnError: true }); + + // THE CRUX. The row EXISTS, so the fork must not try to create it — + // under the old visibility read this row inserted, duplicate-keyed, + // and the non-atomic fallback inserted AGAIN. + expect(t.insert).not.toHaveBeenCalled(); + expect(t.update).toHaveBeenCalledTimes(1); + expect(t.update.mock.calls[0][2].where.id).toBe('theirs_1'); + + // What the caller sees is the WRITE POLICY's answer (#1994), not a + // primary-key collision. + expect(res.succeeded).toBe(0); + expect(res.failed).toBe(1); + expect(res.results[0].errors?.[0]?.code).toBe('RECORD_NOT_FOUND'); + expect(res.results[0].errors?.[0]?.httpStatus).toBe(404); + expect(res.results[0].errors?.[0]?.message).not.toContain('duplicate key'); + + // And the store is untouched — no second row, no clobbered value. + expect(t.rows.get('theirs_1')).toMatchObject({ progress: 0, owner_id: 'u2' }); + expect(t.rows.size).toBe(2); + }); + + it('the probe asks with SYSTEM context — visibility stays the write policy`s decision', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + await upsert(p, [ + { id: 'mine_1', data: { progress: 1 } }, + { id: 'theirs_1', data: { progress: 5 } }, + { id: 'brand_new', data: { progress: 7 } }, + ], { continueOnError: true }); + + expect(t.findOne).toHaveBeenCalledTimes(3); + expect(t.findOne.mock.calls.every((c: any[]) => c[1]?.context?.isSystem === true)).toBe(true); + }); + + it('a real update failure surfaces ITSELF — no fallback insert to mask it (non-atomic)', async () => { + const t = makeRlsEngine(); + t.engine.update = vi.fn(async () => { throw new Error('update exploded'); }); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [{ id: 'mine_1', data: { progress: 1 } }], { continueOnError: true }); + + // The old non-atomic fallback inserted here, buried 'update exploded' + // under a duplicate-key error, and reported THAT to the caller. + expect(t.insert).not.toHaveBeenCalled(); + expect(res.results[0].success).toBe(false); + expect(res.results[0].errors?.[0]?.message).toContain('update exploded'); + expect(res.results[0].errors?.[0]?.message).not.toContain('duplicate key'); + }); + + it('a missing id still INSERTS — the #5088 pin is unchanged', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [{ id: 'brand_new', data: { progress: 7 } }]); + + expect(res.succeeded).toBe(1); + expect(res.results[0]).toMatchObject({ id: 'brand_new', success: true }); + expect(t.rows.get('brand_new')).toMatchObject({ progress: 7 }); + expect(t.update).not.toHaveBeenCalled(); + }); + + it('a visible existing id still UPDATES, with the CALLER`s context on the write (#3455)', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [{ id: 'mine_1', data: { progress: 9 } }]); + + expect(res.succeeded).toBe(1); + expect(res.results[0]).toMatchObject({ id: 'mine_1', success: true }); + expect(t.rows.get('mine_1')).toMatchObject({ progress: 9 }); + expect(t.insert).not.toHaveBeenCalled(); + // Only the PROBE is system — the write still runs as the caller. + expect(t.update.mock.calls[0][2].context?.userId).toBe('u1'); + expect(t.update.mock.calls[0][2].context?.isSystem).not.toBe(true); + }); + + it('atomic: the out-of-scope row is CAUSAL, no fallback fires, the earlier write rolls back', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [ + { id: 'mine_1', data: { progress: 1 } }, + { id: 'theirs_1', data: { progress: 5 } }, + ], { atomic: true }); + + expect(res.results.map((r: any) => r.errors?.[0]?.code)).toEqual([ + 'ROLLED_BACK', 'RECORD_NOT_FOUND', + ]); + expect(t.insert).not.toHaveBeenCalled(); + expect(t.rows.get('mine_1')).toMatchObject({ progress: 0 }); + }); + + it('mixed batch, continueOnError: update / policy-denied / insert each answer for themselves', async () => { + const t = makeRlsEngine(); + const p = new ObjectStackProtocolImplementation(t.engine); + + const res: any = await upsert(p, [ + { id: 'mine_1', data: { progress: 1 } }, + { id: 'theirs_1', data: { progress: 5 } }, + { id: 'brand_new', data: { progress: 7 } }, + ], { continueOnError: true }); + + expect(res.succeeded).toBe(2); + expect(res.failed).toBe(1); + expect(res.results[0]).toMatchObject({ id: 'mine_1', success: true }); + expect(res.results[1].errors?.[0]?.code).toBe('RECORD_NOT_FOUND'); + expect(res.results[2]).toMatchObject({ id: 'brand_new', success: true }); + expect(t.rows.get('mine_1')).toMatchObject({ progress: 1 }); + expect(t.rows.get('theirs_1')).toMatchObject({ progress: 0 }); + expect(t.rows.get('brand_new')).toMatchObject({ progress: 7 }); + expect(t.insert).toHaveBeenCalledTimes(1); + }); +});