fix(metadata-protocol): batchData upsert fork asks existence, not caller visibility (#5099)

.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.

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 });
                             }

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<string, any>([
+        ['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<any>, 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);
+    });
+});