From 8deb984b295b073eacf35b3fbbb5a31759110e4a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 15:33:55 +0000 Subject: [PATCH] feat(service-messaging): bind the claim credential in ack()'s compare-and-set (#11859) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit INotificationOutbox.ack() takes back the claimed record; claim()/claimDigest() declare ClaimedDeliveryRecord[] (the (claimedBy, claimedAt) pair the store stamps, guaranteed present). The ack predicate now includes ownership, so a late ack from a node whose claim was reaped and re-claimed matches nothing — refused with the existing NotificationAckError DELIVERY_NOT_ELIGIBLE, writing nothing. The caller never supplies an identity: ownership is proven by round-tripping what claim() returned. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01UQgPSniH1GFM9ZDeGyuGUa --- .changeset/outbox-ack-claim-credential.md | 11 + .../service-messaging/src/dispatcher.ts | 18 +- .../services/service-messaging/src/index.ts | 3 + .../service-messaging/src/memory-outbox.ts | 43 +++- ...ox-ack-claim-ownership.integration.test.ts | 209 ++++++++++++++++++ ...utbox-ack-precondition.integration.test.ts | 31 ++- .../src/outbox-dispatcher-scope.ts | 22 +- .../services/service-messaging/src/outbox.ts | 91 ++++++-- .../src/sql-outbox-audit-columns.test.ts | 12 +- .../service-messaging/src/sql-outbox.ts | 72 ++++-- 10 files changed, 444 insertions(+), 68 deletions(-) create mode 100644 .changeset/outbox-ack-claim-credential.md create mode 100644 packages/services/service-messaging/src/outbox-ack-claim-ownership.integration.test.ts diff --git a/.changeset/outbox-ack-claim-credential.md b/.changeset/outbox-ack-claim-credential.md new file mode 100644 index 0000000000..1bb04022ee --- /dev/null +++ b/.changeset/outbox-ack-claim-credential.md @@ -0,0 +1,11 @@ +--- +'@objectstack/service-messaging': minor +--- + +**BREAKING (interface member signature):** `INotificationOutbox.ack()` now takes back the claimed record instead of a bare row id, and its compare-and-set binds the claim credential the record carries (#11859). `claim()` / `claimDigest()` declare their true return type, `ClaimedDeliveryRecord[]` — the same rows as before, with the (`claimedBy`, `claimedAt`) pair the store stamps guaranteed present — so reads of claim results do not change; the one breaking edit is at ack call sites, which hand the whole record back where they previously handed `record.id` (the caller already holds it: `ack` completes a claim, and the record is what `claim()` returned). + +Why: `status = 'in_flight'` could prove a claim exists but not whose. In the reachable sequence — node A claims a row, the send outruns `claimTtlMs`, another node's `claim()` reaps and re-claims the row, A finishes late — A's ack still matched and wrote its outcome over B's live attempt. With the credential in the predicate a late ack matches nothing, is refused with the existing `NotificationAckError` (`DELIVERY_NOT_ELIGIBLE`, ADR-0112), and writes nothing; the caller never needs to know its own `nodeId`, because ownership is proven by round-tripping what `claim()` returned. Both implementations (`SqlNotificationOutbox`, `MemoryNotificationOutbox`) enforce it identically. + +Breaking ships as `minor` per the launch-window convention (`scripts/check-changeset-no-major.mjs`). + + diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index a948adccd0..81b4cdb5d9 100644 --- a/packages/services/service-messaging/src/dispatcher.ts +++ b/packages/services/service-messaging/src/dispatcher.ts @@ -1,7 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import type { MessagingChannel, MessagingChannelContext, Notification, SendResult } from './channel.js'; -import type { AckResult, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; +import type { AckResult, ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; import { classifyDeliveryAttempt } from './backoff.js'; import { renderDigest } from './digest-render.js'; @@ -201,7 +201,7 @@ export class NotificationDispatcher { * ack every row in it with that one outcome. On failure the whole group * re-defers together (each row keeps its own backoff via its `attempts`). */ - private async processDigestGroup(rows: NotificationDeliveryRecord[]): Promise { + private async processDigestGroup(rows: ClaimedDeliveryRecord[]): Promise { const channelName = rows[0].channel; const recipient = rows[0].recipientId; const channel = this.opts.channels.getChannel(channelName); @@ -242,7 +242,7 @@ export class NotificationDispatcher { } } - private async processRow(row: NotificationDeliveryRecord): Promise { + private async processRow(row: ClaimedDeliveryRecord): Promise { const channel = this.opts.channels.getChannel(row.channel); if (!channel) { // No transport for this channel → terminal, observable on the row. @@ -307,9 +307,13 @@ export class NotificationDispatcher { * Only `DELIVERY_NOT_ELIGIBLE` is absorbed. A store fault is not a lost * race and still propagates to `runTick`'s handler. */ - private async ackAttempt(row: NotificationDeliveryRecord, result: AckResult): Promise { + private async ackAttempt(row: ClaimedDeliveryRecord, result: AckResult): Promise { try { - await this.opts.outbox.ack(row.id, result); + // [#11859] The record is handed back WHOLE: its (claimedBy, + // claimedAt) pair is the claim credential the store stamped, and + // the ack's compare-and-set binds it — this loop never needs to + // know or repeat its own nodeId. + await this.opts.outbox.ack(row, result); } catch (err) { if ((err as { code?: string })?.code !== 'DELIVERY_NOT_ELIGIBLE') throw err; this.opts.logger?.warn?.('notification-dispatcher: ack refused, claim no longer held', { @@ -322,8 +326,8 @@ export class NotificationDispatcher { } /** Group claimed digest rows by their `digestKey` (insertion order preserved). */ -function groupByDigestKey(rows: NotificationDeliveryRecord[]): NotificationDeliveryRecord[][] { - const groups = new Map(); +function groupByDigestKey(rows: ClaimedDeliveryRecord[]): ClaimedDeliveryRecord[][] { + const groups = new Map(); for (const r of rows) { const key = r.digestKey ?? r.id; // defensive — claimDigest only returns keyed rows let g = groups.get(key); diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 3b4b8468da..f29c7852d5 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -91,6 +91,9 @@ export type { export type { INotificationOutbox, NotificationDeliveryRecord, + // [#11859] What claim()/claimDigest() hand out and ack() takes back — the + // record carrying the claim credential the compare-and-set binds. + ClaimedDeliveryRecord, DeliveryStatus, DeliveryPayload, EnqueueDeliveryInput, diff --git a/packages/services/service-messaging/src/memory-outbox.ts b/packages/services/service-messaging/src/memory-outbox.ts index 0185818997..f2a9467861 100644 --- a/packages/services/service-messaging/src/memory-outbox.ts +++ b/packages/services/service-messaging/src/memory-outbox.ts @@ -3,13 +3,19 @@ import { randomUUID } from 'node:crypto'; import type { AckResult, + ClaimedDeliveryRecord, ClaimOptions, DeliveryStatus, EnqueueDeliveryInput, INotificationOutbox, NotificationDeliveryRecord, } from './outbox.js'; -import { NotificationAckError, notificationAckNotClaimedMessage } from './outbox.js'; +import { + NotificationAckError, + notificationAckLostClaimMessage, + notificationAckNoCredentialMessage, + notificationAckNotClaimedMessage, +} from './outbox.js'; import { hashPartition } from './backoff.js'; /** @@ -61,7 +67,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox { return id; } - async claim(opts: ClaimOptions): Promise { + async claim(opts: ClaimOptions): Promise { const now = opts.now ?? this.clock(); // Reap stale in_flight. for (const r of this.rows.values()) { @@ -72,7 +78,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox { r.updatedAt = now; } } - const out: NotificationDeliveryRecord[] = []; + const out: ClaimedDeliveryRecord[] = []; for (const r of this.rows.values()) { if (out.length >= opts.limit) break; if (r.status !== 'pending') continue; @@ -83,12 +89,14 @@ export class MemoryNotificationOutbox implements INotificationOutbox { r.claimedBy = opts.nodeId; r.claimedAt = now; r.updatedAt = now; - out.push({ ...r }); + // [#11859] The copy handed out carries the claim credential the + // two lines above just stamped — the record IS the credential. + out.push({ ...r, claimedBy: opts.nodeId, claimedAt: now }); } return out; } - async claimDigest(opts: ClaimOptions): Promise { + async claimDigest(opts: ClaimOptions): Promise { const now = opts.now ?? this.clock(); // Reap stale in_flight (same as claim). for (const r of this.rows.values()) { @@ -101,7 +109,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox { } // Claim every DUE batched row in the partition — a window must be taken // whole, so `limit` does not truncate a group here. - const out: NotificationDeliveryRecord[] = []; + const out: ClaimedDeliveryRecord[] = []; for (const r of this.rows.values()) { if (r.status !== 'pending') continue; if (r.digestKey == null) continue; @@ -111,12 +119,19 @@ export class MemoryNotificationOutbox implements INotificationOutbox { r.claimedBy = opts.nodeId; r.claimedAt = now; r.updatedAt = now; - out.push({ ...r }); + out.push({ ...r, claimedBy: opts.nodeId, claimedAt: now }); } return out; } - async ack(id: string, result: AckResult): Promise { + async ack(claimed: ClaimedDeliveryRecord, result: AckResult): Promise { + const id = claimed.id; + // [#11859] The runtime half of the ClaimedDeliveryRecord contract, for + // JS callers and casts: a record with no claim credential was not + // handed out by claim()/claimDigest() and is refused before any read. + if (typeof claimed.claimedBy !== 'string' || typeof claimed.claimedAt !== 'number') { + throw new NotificationAckError(notificationAckNoCredentialMessage(id), 'DELIVERY_NOT_ELIGIBLE'); + } const r = this.rows.get(id); // An id matching no row is not a contract violation: there is no state // to corrupt and no claim to lose. Unchanged, and declared on the @@ -135,6 +150,18 @@ export class MemoryNotificationOutbox implements INotificationOutbox { 'DELIVERY_NOT_ELIGIBLE', ); } + // [#11859] Ownership: the row is claimed, but not by the claim this + // record came from — it was reaped and re-claimed while the send ran + // (possibly by this same store handing it to this same node again: the + // credential is the PAIR, so a later claim's `claimedAt` refuses the + // earlier claim's ack). Refused with nothing written, so the live + // attempt it would have overwritten stays intact. + if (r.claimedBy !== claimed.claimedBy || r.claimedAt !== claimed.claimedAt) { + throw new NotificationAckError( + notificationAckLostClaimMessage(id, r.status), + 'DELIVERY_NOT_ELIGIBLE', + ); + } const now = this.clock(); // Reached only for a genuinely claimed row, so this counts a real // dispatch attempt and nothing else (#11453). diff --git a/packages/services/service-messaging/src/outbox-ack-claim-ownership.integration.test.ts b/packages/services/service-messaging/src/outbox-ack-claim-ownership.integration.test.ts new file mode 100644 index 0000000000..f8e7598c1f --- /dev/null +++ b/packages/services/service-messaging/src/outbox-ack-claim-ownership.integration.test.ts @@ -0,0 +1,209 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #11859 — `ack()` proves OWNERSHIP, not just "a claim exists": the claim + * credential rides the record `claim()` returns, and the compare-and-set + * binds it (ruling C on the card; option A's caller-supplied identity and + * option B's required `nodeId` parameter were both refused). + * + * ## The reachable sequence this file replays — for real + * + * 1. node A claims row R and starts a send; + * 2. the send outruns `claimTtlMs`; + * 3. another node's `claim()` reaps R back to `pending` and re-claims it — + * R is `in_flight` again, claimed by B; + * 4. node A finishes and acks. Before #11859, `status = 'in_flight'` MATCHED + * and A's outcome was written over B's live attempt. + * + * Every step is driven through the public contract (`claim` with an explicit + * `now`, never a hand-set `claimed_by`), because the defect lives in the + * interaction of the reap, the re-claim and the late ack — a fixture that + * fakes step 3 by poking the store would pin the poke, not the race. + * + * ## The vacuity traps closed explicitly + * + * - **Refusal alone cannot tell "refused" from "landed, then errored".** + * Each refusal leg also asserts what the ack did NOT do: the row still + * belongs to B's claim, B's attempt counter is untouched, and B's own ack + * then lands — the outcome the late ack would have overwritten. + * - **A backend that refuses EVERY ack passes the refusal legs.** The + * negative control runs the same sequence WITHOUT the reap (B's claim + * finds nothing to take) and requires A's ack to succeed. + * - **`toThrow()` alone proves nothing** (an unfixed backend throws nothing; + * a broken one could throw anything): refusals assert the ERROR IDENTITY — + * `name` + the ADR-0112 `code`. There is no HTTP envelope on this surface, + * so `code` is the whole machine-readable identity + * (`NotificationAckError` carries no HTTP status). + * + * ## Why both backends, one table + * + * Same warrant as the #11453 file beside this one: the guarantee is a + * property of {@link INotificationOutbox}, the SQL leg runs on a REAL engine + * (`ObjectQL` + `SqlDriver`, better-sqlite3 `:memory:` — the #5704 ruled test + * backend) because the fix IS an atomic conditional UPDATE and a fake engine + * cannot refuse a write, and the memory leg keeps every unit test in the repo + * honest about the same contract. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { MemoryNotificationOutbox } from './memory-outbox.js'; +import { SqlNotificationOutbox } from './sql-outbox.js'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; +import type { INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; + +/** The refusal identity — `name` + ADR-0112 `code`, never a bare throw. */ +const REFUSAL = { name: 'NotificationAckError', code: 'DELIVERY_NOT_ELIGIBLE' }; + +const TTL = 60_000; +/** Step-1 instant: node A's claim. */ +const T0 = 1_000_000; +/** Step-3 instant: one past the visibility timeout, so the reap fires. */ +const T_AFTER_TTL = T0 + TTL + 1; +/** The no-reap instant for the negative control: inside the timeout. */ +const T_WITHIN_TTL = T0 + TTL - 1; + +function claimOpts(nodeId: string, now: number) { + return { nodeId, limit: 10, claimTtlMs: TTL, now }; +} + +interface Backend { + readonly name: string; + create(): Promise; + destroy(): Promise; +} + +function memoryBackend(): Backend { + return { + name: 'MemoryNotificationOutbox', + async create() { return new MemoryNotificationOutbox(1); }, + async destroy() { /* nothing to tear down */ }, + }; +} + +function sqlBackend(): Backend { + let engine: ObjectQL | undefined; + return { + name: 'SqlNotificationOutbox', + async create() { + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(NotificationDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); + return new SqlNotificationOutbox(engine as any, { partitionCount: 1 }); + }, + async destroy() { + try { await engine?.destroy(); } catch { /* noop */ } + engine = undefined; + }, + }; +} + +describe.each([memoryBackend(), sqlBackend()])('$name — ack() claim ownership (#11859)', (backend) => { + let outbox: INotificationOutbox; + + beforeEach(async () => { outbox = await backend.create(); }); + afterEach(async () => { await backend.destroy(); }); + + async function enqueueOne(notificationId = 'n1'): Promise { + return outbox.enqueue({ + notificationId, + recipientId: 'u1', + channel: 'inbox', + payload: { title: 'hello', body: 'world' }, + }); + } + + async function readRow(id: string): Promise { + const rows = await outbox.list(); + const row = rows.find((r) => r.id === id); + if (!row) throw new Error(`row '${id}' vanished — the harness, not the contract, is broken`); + return row; + } + + /** One line a human can diff: status, holder, claim instant, attempts, error. */ + function fingerprint(r: NotificationDeliveryRecord): string { + return `${r.status}:${r.claimedBy ?? '-'}:${r.claimedAt ?? '-'}:${r.attempts}:${r.error ?? ''}`; + } + + it('replays the card: a late ack from a reaped claim is refused, and touches NOTHING', async () => { + const id = await enqueueOne(); + + // 1. node A claims R and "starts a send". + const claimedByA = await outbox.claim(claimOpts('node-a', T0)); + expect(claimedByA.map((r) => `${r.id}:${r.claimedBy}:${r.claimedAt}`)).toEqual([`${id}:node-a:${T0}`]); + + // 2.–3. The send outruns claimTtlMs; node B's claim() reaps R back to + // pending and re-claims it in the same call. R is in_flight AGAIN — + // the state #11453's status-only predicate cannot tell from step 1. + const claimedByB = await outbox.claim(claimOpts('node-b', T_AFTER_TTL)); + expect(claimedByB.map((r) => `${r.id}:${r.claimedBy}:${r.claimedAt}`)).toEqual([`${id}:node-b:${T_AFTER_TTL}`]); + + // 4. node A finishes and acks — with the record its own claim returned. + // Refused by identity, not by accident. + await expect( + outbox.ack(claimedByA[0], { success: true, durationMs: TTL + 5 }), + ).rejects.toMatchObject(REFUSAL); + + // What the refusal did NOT do: the row still belongs to B's claim, + // B's attempt is intact (attempts untouched, no error, no outcome). + expect(fingerprint(await readRow(id))).toBe(`in_flight:node-b:${T_AFTER_TTL}:0:`); + + // …and B's own ack — the live attempt A would have overwritten — + // still lands, recording exactly one real attempt. + await expect(outbox.ack(claimedByB[0], { success: true, durationMs: 3 })).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:'); + }); + + it('negative control: the SAME sequence without the reap still acks successfully', async () => { + const id = await enqueueOne(); + + // 1. node A claims R. + const claimedByA = await outbox.claim(claimOpts('node-a', T0)); + expect(claimedByA.map((r) => r.id)).toEqual([id]); + + // 2'. The send is slow but INSIDE the visibility timeout, so node B's + // claim() reaps nothing and takes nothing — proven, not assumed. + const claimedByB = await outbox.claim(claimOpts('node-b', T_WITHIN_TTL)); + expect(claimedByB).toEqual([]); + expect(fingerprint(await readRow(id))).toBe(`in_flight:node-a:${T0}:0:`); + + // 4'. A's ack with the record its claim returned MUST land — a + // predicate that refused every ack would fail here, not just pass + // the refusal legs above. + await expect(outbox.ack(claimedByA[0], { success: true, durationMs: 5 })).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:'); + }); + + it('the credential is the CLAIM, not the node: a stale ack loses to the same node\'s own re-claim', async () => { + const id = await enqueueOne(); + + // 1. node A claims R… + const staleClaim = await outbox.claim(claimOpts('node-a', T0)); + expect(staleClaim.map((r) => r.id)).toEqual([id]); + + // 2.–3. …outruns the TTL, and A ITSELF reaps and re-claims on a later + // tick. Same node id — a claimed_by-only predicate would match. + const freshClaim = await outbox.claim(claimOpts('node-a', T_AFTER_TTL)); + expect(freshClaim.map((r) => `${r.claimedBy}:${r.claimedAt}`)).toEqual([`node-a:${T_AFTER_TTL}`]); + + // 4. The FIRST attempt's late ack is refused — `claimedAt` is what + // tells two claims by one node apart. The outcome belongs to the + // attempt, and a re-claim is a new attempt. + await expect( + outbox.ack(staleClaim[0], { success: false, error: 'timed out', nextAttemptAt: T_AFTER_TTL + 1 }), + ).rejects.toMatchObject(REFUSAL); + expect(fingerprint(await readRow(id))).toBe(`in_flight:node-a:${T_AFTER_TTL}:0:`); + + // The fresh claim's ack still lands. + await expect(outbox.ack(freshClaim[0], { success: true })).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:'); + }); +}); diff --git a/packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts b/packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts index 4059419f37..43988d41e2 100644 --- a/packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts +++ b/packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts @@ -57,7 +57,7 @@ import { SqlDriver } from '@objectstack/driver-sql'; import { MemoryNotificationOutbox } from './memory-outbox.js'; import { SqlNotificationOutbox } from './sql-outbox.js'; import { NotificationDelivery } from './objects/notification-delivery.object.js'; -import type { INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; +import type { ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; /** * The refusal identity both backends must produce. `DELIVERY_NOT_ELIGIBLE` is @@ -133,9 +133,13 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit const id = await enqueueOne(); expect(`${(await readRow(id)).status}:${(await readRow(id)).attempts}`).toBe('pending:0'); - // The card's trap, verbatim: ack-as-cancel on a row no dispatcher holds. + // The card's trap, verbatim: ack-as-cancel on a row no dispatcher + // holds. [#11859] `ack` now takes the claimed record back, so the + // literal spelling of the trap is handing it a `list()` row — which + // carries NO claim credential; the cast is the JS caller/miscast this + // pin keeps refused at runtime, not just at compile time. await expect( - outbox.ack(id, { success: false, suppressed: true }), + outbox.ack((await readRow(id)) as ClaimedDeliveryRecord, { success: false, suppressed: true }), ).rejects.toMatchObject(REFUSAL); // A refused ack is not a partial one: the row keeps its identity, its @@ -150,7 +154,7 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit const claimed = await outbox.claim(CLAIM); expect(claimed.map((r) => r.id)).toEqual([id]); - await expect(outbox.ack(id, { success: true, durationMs: 5 })).resolves.toBeUndefined(); + await expect(outbox.ack(claimed[0], { success: true, durationMs: 5 })).resolves.toBeUndefined(); const after = await readRow(id); expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:success:1`); @@ -158,11 +162,14 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit it('refuses a SECOND ack on a terminal row, leaving the first outcome intact', async () => { const id = await enqueueOne(); - await outbox.claim(CLAIM); - await outbox.ack(id, { success: true }); + const [rec] = await outbox.claim(CLAIM); + await outbox.ack(rec, { success: true }); + // The SAME once-genuine record, handed back a second time: its + // credential was real, but the row is terminal now and the first + // outcome must stand. await expect( - outbox.ack(id, { success: false, suppressed: true }), + outbox.ack(rec, { success: false, suppressed: true }), ).rejects.toMatchObject(REFUSAL); // `attempts` is the assertion that matters: an unconditional increment @@ -173,10 +180,10 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit it('re-arms a retried row, then refuses an ack on the re-armed (pending) row', async () => { const id = await enqueueOne(); - await outbox.claim(CLAIM); + const [rec] = await outbox.claim(CLAIM); // A real failed attempt: the row goes back to `pending` for a later try. - await outbox.ack(id, { success: false, error: 'transport blip', nextAttemptAt: 1 }); + await outbox.ack(rec, { success: false, error: 'transport blip', nextAttemptAt: 1 }); const retried = await readRow(id); expect(`${retried.id}:${retried.status}:${retried.attempts}`).toBe(`${id}:pending:1`); @@ -186,7 +193,7 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit // and the refused row's state (`pending`) are the SAME status, so only // a real conditional write tells them apart. await expect( - outbox.ack(id, { success: false, suppressed: true }), + outbox.ack(rec, { success: false, suppressed: true }), ).rejects.toMatchObject(REFUSAL); const after = await readRow(id); @@ -199,9 +206,9 @@ describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondit // state happens to be `suppressed`. The precondition must not confuse // that legitimate dispatcher outcome with ack-as-cancel. const id = await enqueueOne(); - await outbox.claim(CLAIM); + const [rec] = await outbox.claim(CLAIM); - await outbox.ack(id, { success: false, suppressed: true, error: 'no such recipient' }); + await outbox.ack(rec, { success: false, suppressed: true, error: 'no such recipient' }); const after = await readRow(id); expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:suppressed:1`); diff --git a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts index d4f65bd28c..4ef650e7dc 100644 --- a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -160,15 +160,35 @@ export function dispatcherAckOptions( * the one write on these objects that must never reach for a bypass: it is * request-reachable and threads the caller's tenant instead. * + * ## [#11859] Ownership joined the predicate + * + * `status = 'in_flight'` can prove a claim EXISTS but not WHOSE: after a + * visibility-timeout reap plus a re-claim, the row is `in_flight` again under + * another claim, and the reaped node's late ack still matched — writing its + * outcome over the live attempt. The predicate therefore also binds the claim + * credential (`claimed_by`, `claimed_at`) round-tripped from the record + * `claim()` returned, so the compare-and-set asks "is this row still held by + * the claim being completed", and a late ack matches nothing. The PAIR is the + * credential, not `claimed_by` alone: `claimed_at` distinguishes two claims by + * the SAME node, so a node's stale ack cannot land on its own later re-claim. + * * ⚠️ Diagnostics only, exactly as above: `bypassTenantAudit` never changes * what the write touches. * * @param id Primary key of the delivery row this attempt outcome belongs to. * @param expectedStatus The status the row MUST still hold for the write to land. + * @param claimedBy The node id stamped by the claim this ack completes. + * @param claimedAt The claim instant (ms) stamped by that same claim. */ export function dispatcherAckCasOptions( id: string, expectedStatus: 'in_flight', + claimedBy: string, + claimedAt: number, ): EngineUpdateOptions & { multi: true; bypassTenantAudit: true } { - return { where: { id, status: expectedStatus }, multi: true, bypassTenantAudit: true }; + return { + where: { id, status: expectedStatus, claimed_by: claimedBy, claimed_at: claimedAt }, + multi: true, + bypassTenantAudit: true, + }; } diff --git a/packages/services/service-messaging/src/outbox.ts b/packages/services/service-messaging/src/outbox.ts index 5a5fb0690f..3ca0c1a9a2 100644 --- a/packages/services/service-messaging/src/outbox.ts +++ b/packages/services/service-messaging/src/outbox.ts @@ -54,6 +54,23 @@ export interface NotificationDeliveryRecord { digestKey?: string; } +/** + * [#11859] A delivery row as handed out by {@link INotificationOutbox.claim} / + * {@link INotificationOutbox.claimDigest}: the **claim credential** — the + * (`claimedBy`, `claimedAt`) pair the store stamped when it took the row — is + * guaranteed present. {@link INotificationOutbox.ack} takes this record back, + * and the credential joins the compare-and-set predicate, so ownership is + * proven by ROUND-TRIPPING what `claim()` returned rather than by the caller + * supplying an identity it had to know (the option-A shape the #11859 ruling + * refused). The pair identifies one CLAIM, not one node: `claimedAt` is what + * refuses a late ack even when the SAME node re-claimed its own reaped row — + * the outcome belongs to the attempt, and a re-claim is a new attempt. + */ +export interface ClaimedDeliveryRecord extends NotificationDeliveryRecord { + claimedBy: string; + claimedAt: number; +} + export interface EnqueueDeliveryInput { notificationId: string; recipientId: string; @@ -144,9 +161,11 @@ export function notificationAckNotClaimedMessage(id: string, status: DeliverySta /** * The refusal message for the OTHER half of the precondition: the row WAS - * `in_flight` when ack() read it and had stopped being so by the time the - * conditional write ran — a claim lost to the visibility-timeout reap plus a - * re-claim by another node. + * claimed by this caller and had stopped being so by the time the outcome was + * recorded — a claim lost to the visibility-timeout reap. [#11859] Covers BOTH + * post-reap states: the row moved out of `in_flight`, and the row re-claimed + * (still `in_flight`, but under a different claim credential — possibly the + * same node's LATER claim, which is still not the claim this ack completes). * * Distinguished from {@link notificationAckNotClaimedMessage} because the two * say different things to whoever reads the log: the first is a CALLER using @@ -155,10 +174,30 @@ export function notificationAckNotClaimedMessage(id: string, status: DeliverySta */ export function notificationAckLostClaimMessage(id: string, status: DeliveryStatus | 'unknown'): string { return ( - `Delivery row '${id}' stopped being 'in_flight' while ack() was recording its outcome ` - + `(it now reads '${status}'), so the conditional update matched no row and NOTHING was ` + `Delivery row '${id}' is no longer held by the claim this ack completes (it now reads ` + + `'${status}'), so the ownership-checked conditional update matched no row and NOTHING was ` + 'written — this attempt was not recorded and the row belongs to whoever holds it now. ' - + 'Expected when a slow send outruns `claimTtlMs` and the row is reaped and re-claimed (#11453).' + + 'Expected when a slow send outruns `claimTtlMs` and the row is reaped and re-claimed ' + + '(#11453, #11859).' + ); +} + +/** + * [#11859] The refusal message for a record that carries no claim credential + * at all. `ack()` takes back the exact record {@link INotificationOutbox.claim} + * / {@link INotificationOutbox.claimDigest} returned; a row read via `list()` + * while unclaimed, or a hand-built record, has no (`claimedBy`, `claimedAt`) + * pair and is refused before any read or write — possession of a row's id was + * never evidence of a claim, and the compile-time contract + * ({@link ClaimedDeliveryRecord}) is enforced here at runtime for JS callers + * and casts. + */ +export function notificationAckNoCredentialMessage(id: string): string { + return ( + `Delivery row '${id}': the record passed to ack() carries no claim credential ` + + '(claimedBy + claimedAt). ack() records the outcome of a delivery the caller CLAIMED, ' + + 'and proves the claim by handing back the record claim()/claimDigest() returned. ' + + 'Nothing was written.' ); } @@ -170,24 +209,36 @@ export function notificationAckLostClaimMessage(id: string, status: DeliveryStat */ export interface INotificationOutbox { enqueue(input: EnqueueDeliveryInput): Promise; - claim(opts: ClaimOptions): Promise; + claim(opts: ClaimOptions): Promise; /** - * Record the outcome of ONE dispatch attempt on a row this caller claimed. + * Record the outcome of ONE dispatch attempt on a row this caller claimed, + * by handing back the record {@link claim} / {@link claimDigest} returned. + * + * ⛔ **Precondition: the row MUST still be held by the claim `claimed` + * came from.** That is two tests, both re-stated IN the conditional write: + * the row is `in_flight`, AND its (`claimed_by`, `claimed_at`) pair equals + * the credential on the record handed back. Acking a row in any other + * status — an unclaimed `pending` row, or one already terminal — throws + * {@link NotificationAckError}, writes nothing, and leaves `attempts` + * untouched; so does a late ack whose claim was reaped and re-claimed + * (#11859): `status = 'in_flight'` alone could not tell "claimed" from + * "claimed by the caller", so a node whose send outran `claimTtlMs` wrote + * its outcome over the re-claiming node's live attempt. `ack` is the + * dispatcher's completion callback, NOT a cancellation primitive: using it + * to flip a `pending` row to `suppressed` raced the dispatcher and + * recorded an attempt that never happened (#11453). A record whose id + * matches no row is not a contract violation and stays a silent no-op — + * an absent row has no state to corrupt and no claim to lose. * - * ⛔ **Precondition: the row MUST be `in_flight`** (i.e. claimed). Acking a - * row in any other status — an unclaimed `pending` row, or one already - * terminal — throws {@link NotificationAckError}, writes nothing, and - * leaves `attempts` untouched. `ack` is the dispatcher's completion - * callback, NOT a cancellation primitive: using it to flip a `pending` row - * to `suppressed` raced the dispatcher and recorded an attempt that never - * happened (#11453). An id that matches no row is not a contract - * violation and stays a silent no-op — an absent row has no state to - * corrupt and no claim to lose. + * Only `claimed.id` and the credential are trusted; every other field on + * the record may be stale by the time the ack runs, and implementations + * MUST re-read what they need (e.g. `attempts`) from the store. * * Implementations MUST make the transition atomic against {@link claim}: - * the status test and the write are one operation, never a read-then-write. + * the ownership test and the write are one operation, never a + * read-then-write. */ - ack(id: string, result: AckResult): Promise; + ack(claimed: ClaimedDeliveryRecord, result: AckResult): Promise; list(filter?: { status?: DeliveryStatus; notificationId?: string }): Promise; /** * P3b-2: atomically claim **all** due batched rows (those with a `digestKey`) @@ -197,5 +248,5 @@ export interface INotificationOutbox { * `limit`-bounded per group (a window must be claimed whole). Normal `claim` * MUST exclude digest rows so they are never sent individually. */ - claimDigest(opts: ClaimOptions): Promise; + claimDigest(opts: ClaimOptions): Promise; } diff --git a/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts b/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts index aa483a750a..c25802ca70 100644 --- a/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts +++ b/packages/services/service-messaging/src/sql-outbox-audit-columns.test.ts @@ -47,7 +47,9 @@ function makeEngine(findResults: Array>> = []) { const updates: RecordedUpdate[] = []; const inserts: Array<{ object: string; data: Record }> = []; let findCall = 0; - const row: Record = { status: 'in_flight', attempts: 2 }; + // [#11859] The row carries the claim credential ack()'s ownership check + // reads back; the record handed to ack() below round-trips the same pair. + const row: Record = { status: 'in_flight', attempts: 2, claimed_by: 'n1', claimed_at: 111 }; const engine = { async insert(object: string, data: Record) { @@ -130,7 +132,13 @@ describe('SqlNotificationOutbox — audit columns on UPDATE (#4765)', () => { const { engine, updates } = makeEngine(); const outbox = new SqlNotificationOutbox(engine, { partitionCount: 8 }); - await outbox.ack('d1', { success: false, error: 'boom', nextAttemptAt: 123 }); + // [#11859] ack() takes the claimed record back; the credential here + // matches what the fake row carries, as a real claim's would. + await outbox.ack({ + id: 'd1', notificationId: 'n1', recipientId: 'u1', channel: 'inbox', + payload: {}, partitionKey: 0, status: 'in_flight', attempts: 2, + claimedBy: 'n1', claimedAt: 111, createdAt: 0, updatedAt: 0, + }, { success: false, error: 'boom', nextAttemptAt: 123 }); expectNoPlatformAuditColumns(updates); expect(updates[0].data).toMatchObject({ diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index 5559898dea..72cfa57419 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -4,6 +4,7 @@ import { randomUUID } from 'node:crypto'; import type { IDataEngine } from '@objectstack/spec/contracts'; import type { AckResult, + ClaimedDeliveryRecord, ClaimOptions, DeliveryStatus, EnqueueDeliveryInput, @@ -13,7 +14,12 @@ import type { import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; import { dispatcherAckCasOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; -import { NotificationAckError, notificationAckLostClaimMessage, notificationAckNotClaimedMessage } from './outbox.js'; +import { + NotificationAckError, + notificationAckLostClaimMessage, + notificationAckNoCredentialMessage, + notificationAckNotClaimedMessage, +} from './outbox.js'; export const DELIVERY_OBJECT = 'sys_notification_delivery'; @@ -122,7 +128,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { } } - async claim(opts: ClaimOptions): Promise { + async claim(opts: ClaimOptions): Promise { const now = opts.now ?? Date.now(); // 1. Reap stale in_flight rows (visibility-timeout recovery). @@ -159,14 +165,17 @@ export class SqlNotificationOutbox implements INotificationOutbox { dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), ); - // 4. Read back only the rows we own. + // 4. Read back only the rows we own. [#11859] The read-back WHERE just + // proved (claimed_by, claimed_at) — the claim credential — so the + // explicit stamp below narrows to ClaimedDeliveryRecord without a + // cast, and states nothing the query did not already establish. const claimed = (await this.engine.find(this.objectName, { where: { id: { $in: ids }, claimed_by: opts.nodeId, claimed_at: now, status: 'in_flight' }, })) as DeliveryRow[]; - return claimed.map((r) => this.toRecord(r)); + return claimed.map((r) => ({ ...this.toRecord(r), claimedBy: opts.nodeId, claimedAt: now })); } - async claimDigest(opts: ClaimOptions): Promise { + async claimDigest(opts: ClaimOptions): Promise { const now = opts.now ?? Date.now(); // 1. Reap stale in_flight (same as claim). @@ -203,18 +212,30 @@ export class SqlNotificationOutbox implements INotificationOutbox { dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), ); - // 4. Read back the rows we own. + // 4. Read back the rows we own — same credential stamp as claim(). const claimed = (await this.engine.find(this.objectName, { where: { id: { $in: ids }, claimed_by: opts.nodeId, claimed_at: now, status: 'in_flight' }, })) as DeliveryRow[]; - return claimed.map((r) => this.toRecord(r)); + return claimed.map((r) => ({ ...this.toRecord(r), claimedBy: opts.nodeId, claimedAt: now })); } - async ack(id: string, result: AckResult): Promise { + async ack(claimed: ClaimedDeliveryRecord, result: AckResult): Promise { + const id = claimed.id; + // [#11859] The runtime half of the ClaimedDeliveryRecord contract, for + // JS callers and casts: a record with no claim credential was not + // handed out by claim()/claimDigest() and is refused before any IO. + if (typeof claimed.claimedBy !== 'string' || typeof claimed.claimedAt !== 'number') { + throw new NotificationAckError(notificationAckNoCredentialMessage(id), 'DELIVERY_NOT_ELIGIBLE'); + } const current = (await this.engine.findOne(this.objectName, { where: { id }, - fields: ['status', 'attempts'], - })) as { status?: DeliveryStatus; attempts?: number } | null; + fields: ['status', 'attempts', 'claimed_by', 'claimed_at'], + })) as { + status?: DeliveryStatus; + attempts?: number; + claimed_by?: string | null; + claimed_at?: number | null; + } | null; // An id matching no row is not a contract violation: no state to // corrupt, no claim to lose. Declared on the interface, unchanged. if (!current) return; @@ -227,6 +248,18 @@ export class SqlNotificationOutbox implements INotificationOutbox { 'DELIVERY_NOT_ELIGIBLE', ); } + // [#11859] Ownership, read half: the row is claimed, but not by the + // claim this record came from — reaped and re-claimed while the send + // ran (`status = 'in_flight'` alone matches B's live attempt, which is + // exactly the overwrite the card measured). Deterministic refusal + // before any write; the SAME test is re-stated in the conditional + // write below, which is the half that actually holds under the race. + if (current.claimed_by !== claimed.claimedBy || current.claimed_at !== claimed.claimedAt) { + throw new NotificationAckError( + notificationAckLostClaimMessage(id, current.status ?? 'unknown'), + 'DELIVERY_NOT_ELIGIBLE', + ); + } const now = Date.now(); let status: DeliveryStatus; @@ -247,13 +280,16 @@ export class SqlNotificationOutbox implements INotificationOutbox { error = result.error ?? null; } - // [#11453] Precondition, half two: the ATOMIC one. The status test - // above is a read, and a read cannot hold a row still — `claim()` is - // atomic by contract and this call was never part of that atom, which - // is the race the card describes. So the requirement is re-stated IN - // the write: the row is transitioned only if it is STILL `in_flight`. - // A row reaped by the visibility timeout and re-claimed between the - // read and here matches nothing and is left entirely alone. + // [#11453] Precondition, half two: the ATOMIC one. The tests above are + // reads, and a read cannot hold a row still — `claim()` is atomic by + // contract and this call was never part of that atom, which is the + // race the card describes. So the requirement is re-stated IN the + // write: the row is transitioned only if it is STILL `in_flight` AND + // [#11859] still held by THIS claim — the (`claimed_by`, `claimed_at`) + // credential round-tripped from the record `claim()` returned. A row + // reaped by the visibility timeout and re-claimed between the read and + // here matches nothing and is left entirely alone, whoever re-claimed + // it — another node, or this node's own later claim. // // `attempts` is incremented HERE and only here, inside that condition, // so the counter can only move for a row that was genuinely claimed — @@ -275,7 +311,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { // Predicate write (`updateMany`), audited under that op. Declared a // global-sweep site — no request context exists on the tick that // reaches here. Warrant in `outbox-dispatcher-scope.ts`. - dispatcherAckCasOptions(id, 'in_flight') as any, + dispatcherAckCasOptions(id, 'in_flight', claimed.claimedBy, claimed.claimedAt) as any, ); // Did the conditional write land? `IDataEngine.update` declares its