From 0b765a45295768cf589a081007129197caecef9c Mon Sep 17 00:00:00 2001 From: os-sam Date: Mon, 24 Aug 2026 21:24:44 +0000 Subject: [PATCH] fix(service-messaging): enforce ack()'s claimed-row precondition in both outbox implementations (#11453) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ack()` is the dispatcher's completion callback for a row it CLAIMED, and neither implementation checked that, so `ack(id, { success: false, suppressed: true })` on an unclaimed `pending` row succeeded — flipping the row terminal and recording an attempt that never went on the wire. That made `ack` read like the cancellation primitive this interface deliberately does not have, and it raced `claim()` (atomic by contract; `ack` was never part of that atom). Both implementations now refuse a row that is not `in_flight`, with `NotificationAckError` / `DELIVERY_NOT_ELIGIBLE` — this package's already registered ADR-0112 code, the same refusal `SqlHttpOutbox.redeliver` raises when its own compare-and-set misses. A refused ack writes nothing. `SqlNotificationOutbox` does it as an atomic conditional update, not a read-then-write: the precondition is re-stated in the write, which per #11009 must ride the predicate path (the by-id path silently discards it). `attempts` increments inside that condition and nowhere else, so it can only move for a row that was genuinely claimed. The sibling HTTP outbox is untouched: `assertHttpRedeliverable` depends on `IHttpOutbox.ack` incrementing unconditionally, so `attempts === 0` on a terminal row still means "parked, never sent". Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01APWX2AwT3a4xDcjPCe8bk4 --- .changeset/outbox-ack-status-precondition.md | 53 +++++ ...ry-update-tenant-audit.integration.test.ts | 33 ++- .../service-messaging/src/dispatcher.ts | 43 +++- .../services/service-messaging/src/index.ts | 3 + .../service-messaging/src/memory-outbox.ts | 19 ++ ...utbox-ack-precondition.integration.test.ts | 209 ++++++++++++++++++ .../src/outbox-dispatcher-scope.ts | 59 ++++- .../services/service-messaging/src/outbox.ts | 72 ++++++ .../src/sql-outbox-audit-columns.test.ts | 22 +- .../service-messaging/src/sql-outbox.ts | 63 +++++- 10 files changed, 548 insertions(+), 28 deletions(-) create mode 100644 .changeset/outbox-ack-status-precondition.md create mode 100644 packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts diff --git a/.changeset/outbox-ack-status-precondition.md b/.changeset/outbox-ack-status-precondition.md new file mode 100644 index 0000000000..43ee0fc6f8 --- /dev/null +++ b/.changeset/outbox-ack-status-precondition.md @@ -0,0 +1,53 @@ +--- +'@objectstack/service-messaging': minor +--- + +`INotificationOutbox.ack()` enforces its declared precondition — the row must be claimed — in both implementations, and `attempts` moves only for a real dispatch attempt + +`ack()` is the dispatcher's completion callback for a row it CLAIMED, and +neither implementation checked that. `MemoryNotificationOutbox.ack` looked the +row up by id and mutated it; `SqlNotificationOutbox.ack` read only `attempts` +by id. So `ack(id, { success: false, suppressed: true })` on an unclaimed +`pending` row succeeded, flipped the row to terminal `suppressed`, and +incremented `attempts` — which made `ack` read like the cancellation primitive +this interface deliberately does not have. + +That was a trap in two directions. It **raced the dispatcher**: between a +caller's `list()` and its `ack()`, `claim()` could take the row — `claim` is +atomic by contract and `ack` was never part of that atom — so a suppression +could land on a delivery already on the wire, or a dispatcher's real outcome +could be overwritten by a caller that thought it was cancelling. And it +**corrupted `attempts`**: the counter feeds the retry schedule +(`classifyDeliveryAttempt(result, errorClass, row.attempts, …)`), so a row +"cancelled" this way arrived at its next real attempt with the backoff already +advanced by an attempt that never went out. + +Both implementations now refuse an ack on a row that is not `in_flight`, +throwing `NotificationAckError` with this package's already-registered +ADR-0112 code `DELIVERY_NOT_ELIGIBLE` — the same refusal +`SqlHttpOutbox.redeliver` raises when its own compare-and-set misses. A refused +ack writes **nothing**: status, `attempts` and `error` are left exactly as they +were, so the row stays claimable and its backoff position stays honest. An id +matching no row remains a silent no-op — an absent row has no state to corrupt +and no claim to lose. + +`SqlNotificationOutbox` does it as an **atomic conditional update** rather than +a read-then-write, because a read cannot hold a row still and a read-then-write +is the same defect wearing a different hat. The precondition is re-stated in +the write (`where: { id, status: 'in_flight' }`), which — per #11009 — must +ride the predicate path: on the by-id path the driver binds only the primary +key and the extra predicate is silently discarded. `attempts` is incremented +inside that condition and nowhere else, so the counter can only move for a row +that was genuinely claimed. A conditional write that matches nothing is +reported rather than passed off as success. + +`NotificationDispatcher` absorbs exactly one refusal — `DELIVERY_NOT_ELIGIBLE` +— logs it and continues with the rest of the batch, because a send slower than +`claimTtlMs` legitimately loses its claim to the visibility-timeout reap, and +letting that unwind the partition loop would strand every still-valid row in +the batch `in_flight` until its own timeout expired. Any other error still +propagates. + +The sibling HTTP outbox is deliberately untouched: `assertHttpRedeliverable` +depends on `IHttpOutbox.ack` incrementing `attempts` unconditionally, so that +`attempts === 0` on a terminal row still means "parked, never sent". diff --git a/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts b/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts index 0a2fa910f1..a286d5b1ab 100644 --- a/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts +++ b/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts @@ -18,6 +18,16 @@ * tenant-classification contract this file pins (#10740) is unchanged — * threaded `tenantId`, never `bypassTenantAudit`. * + * [#11453] `SqlNotificationOutbox.ack` has since made the SAME move for the + * same reason: its new status precondition ("this row must still be + * `in_flight`") is a compare-and-set, and a predicate on the by-id path is + * silently discarded, so it rides `multi: true` too. Its audit op is + * `updateMany` and its spy below records `SqlDriver.updateMany`. Its + * CLASSIFICATION is unchanged — declared global, now via + * `dispatcherAckCasOptions` — which is the point of pinning the two + * separately: the op moved, the warrant did not. Of the three sites only + * `SqlHttpOutbox.ack` still writes by id. + * * The `ack` pair is declared global (`dispatcherAckOptions`, warrant in * `outbox-dispatcher-scope.ts`). `redeliver` is NOT: it is served to any * authenticated user, so it threads the caller's tenant instead. ⛔ A @@ -74,14 +84,14 @@ let driver: SqlDriver; let warns: Array<{ msg: string; meta: any }>; /** Every `options` bag that reached `SqlDriver.update` — the `update` op only. */ let driverUpdates: Array<{ object: string; id: unknown; options: any }>; -/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009. */ +/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009, and the notification `ack`'s since #11453. */ let driverUpdateManys: Array<{ object: string; where: unknown; options: any }>; /** The audit line for the SINGLE-RECORD op, matched on object + op. */ const auditedUpdate = (object: string): boolean => warns.some((w) => w.msg.includes(`[tenant-audit] update on tenant-scoped object "${object}"`)); -/** The audit line for the PREDICATE op — `redeliver`'s write since #11009. */ +/** The audit line for the PREDICATE op — `redeliver`'s write since #11009, the notification `ack`'s since #11453. */ const auditedUpdateMany = (object: string): boolean => warns.some((w) => w.msg.includes(`[tenant-audit] updateMany on tenant-scoped object "${object}"`)); @@ -211,7 +221,7 @@ async function seedDeadRow(id: string, org: string): Promise { } // ─────────────────────────────────────────────────────────────────────────── -describe('ack — the two dispatcher sites are a classified global sweep (update op)', () => { +describe('ack — the two dispatcher sites are a classified global sweep (update + updateMany ops)', () => { it('SqlHttpOutbox.ack records a REAL delivery in every organization, without a finding', async () => { // The gate's own precondition: this object really is tenant-scoped. expect((driver as any).resolveTenantField(SYS_HTTP_DELIVERY)).toBe('organization_id'); @@ -279,12 +289,23 @@ describe('ack — the two dispatcher sites are a classified global sweep (update 'org_b:success:1', ]); // ② Declared global, for both organizations' rows. - const ackWrites = driverUpdates.filter((u) => u.object === DELIVERY_OBJECT); + // + // [#11453] The ack's op is `updateMany` now, so the reading moves to + // that spy. The claim path writes there too (its reap and its atomic + // claim), so the filter names what an ACK write looks like — and that + // predicate is not incidental: `{ id: , status: 'in_flight' }` + // IS the compare-and-set, so matching on it pins that the ack reached + // the driver CONDITIONAL rather than as a blind by-id write. + const ackWrites = driverUpdateManys.filter( + (u) => u.object === DELIVERY_OBJECT + && typeof (u.where as any)?.id === 'string' + && (u.where as any)?.status === 'in_flight', + ); expect(ackWrites).toHaveLength(2); expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true); - expect(auditedUpdate(DELIVERY_OBJECT)).toBe(false); + expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false); - await controlUnscopedUpdate(DELIVERY_OBJECT, idA); + await controlUnscopedUpdateMany(DELIVERY_OBJECT, idA); }); }); diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index f242d50964..a948adccd0 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 { INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; +import type { AckResult, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; import { classifyDeliveryAttempt } from './backoff.js'; import { renderDigest } from './digest-render.js'; @@ -207,7 +207,7 @@ export class NotificationDispatcher { const channel = this.opts.channels.getChannel(channelName); if (!channel) { for (const row of rows) { - await this.opts.outbox.ack(row.id, { success: false, error: `channel '${channelName}' not registered`, dead: true }); + await this.ackAttempt(row, { success: false, error: `channel '${channelName}' not registered`, dead: true }); this.opts.onAttempt?.(row, false); } return; @@ -237,7 +237,7 @@ export class NotificationDispatcher { const now = this.opts.now?.() ?? Date.now(); for (const row of rows) { const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng); - await this.opts.outbox.ack(row.id, ack); + await this.ackAttempt(row, ack); this.opts.onAttempt?.(row, result.ok); } } @@ -246,7 +246,7 @@ export class NotificationDispatcher { const channel = this.opts.channels.getChannel(row.channel); if (!channel) { // No transport for this channel → terminal, observable on the row. - await this.opts.outbox.ack(row.id, { + await this.ackAttempt(row, { success: false, error: `channel '${row.channel}' not registered`, dead: true, @@ -283,9 +283,42 @@ export class NotificationDispatcher { const errorClass = !result.ok && channel.classifyError ? channel.classifyError(result.error) : undefined; const now = this.opts.now?.() ?? Date.now(); const ack = classifyDeliveryAttempt(result, errorClass, row.attempts, now, this.opts.rng); - await this.opts.outbox.ack(row.id, ack); + await this.ackAttempt(row, ack); this.opts.onAttempt?.(row, result.ok); } + + /** + * [#11453] Record one attempt's outcome, tolerating the ONE refusal a + * correct dispatcher can legitimately provoke. + * + * `ack()` now refuses a row that is not `in_flight`, and this loop can meet + * that honestly: a send slower than `claimTtlMs` lets the visibility-timeout + * reap return the row to `pending` for another node, so by the time we ack, + * the row is not ours. That is a race we are ALLOWED to lose — the delivery + * is re-driven by whoever holds the row now, which is what at-least-once + * means — and the refusal is the outbox correctly declining to overwrite + * someone else's state. + * + * ⛔ What it must not do is abort the tick. The rows still validly claimed + * by this node are processed after this one; letting a lost race unwind the + * partition loop would strand every one of them `in_flight` until their own + * timeouts expire, turning one lost race into a batch-wide delay. + * + * 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 { + try { + await this.opts.outbox.ack(row.id, 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', { + nodeId: this.opts.nodeId, + deliveryId: row.id, + error: (err as Error)?.message ?? String(err), + }); + } + } } /** Group claimed digest rows by their `digestKey` (insertion order preserved). */ diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 85cea966ab..3b4b8468da 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -97,6 +97,9 @@ export type { ClaimOptions, AckResult, } from './outbox.js'; +// [#11453] `ack()`'s status precondition refuses with this, so a caller that +// wants to distinguish "I lost the claim" from a transport fault can catch it. +export { NotificationAckError } from './outbox.js'; export { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; export type { SqlNotificationOutboxOptions } from './sql-outbox.js'; export { MemoryNotificationOutbox } from './memory-outbox.js'; diff --git a/packages/services/service-messaging/src/memory-outbox.ts b/packages/services/service-messaging/src/memory-outbox.ts index 84c9a9ca3b..0185818997 100644 --- a/packages/services/service-messaging/src/memory-outbox.ts +++ b/packages/services/service-messaging/src/memory-outbox.ts @@ -9,6 +9,7 @@ import type { INotificationOutbox, NotificationDeliveryRecord, } from './outbox.js'; +import { NotificationAckError, notificationAckNotClaimedMessage } from './outbox.js'; import { hashPartition } from './backoff.js'; /** @@ -117,8 +118,26 @@ export class MemoryNotificationOutbox implements INotificationOutbox { async ack(id: string, result: AckResult): Promise { 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 + // interface so the two backends agree about it. if (!r) return; + // [#11453] The status precondition. `ack` completes a delivery this + // caller CLAIMED; an unclaimed `pending` row (the ack-as-cancel trap) + // or an already-terminal one is refused, and nothing below runs — so a + // refused ack leaves status, attempts and error exactly as they were. + // Single-threaded, so this test and the mutation are already one atomic + // step; `SqlNotificationOutbox` spells the same guard as a conditional + // UPDATE because it is not. + if (r.status !== 'in_flight') { + throw new NotificationAckError( + notificationAckNotClaimedMessage(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). r.attempts += 1; r.lastAttemptedAt = now; r.claimedBy = undefined; 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 new file mode 100644 index 0000000000..4059419f37 --- /dev/null +++ b/packages/services/service-messaging/src/outbox-ack-precondition.integration.test.ts @@ -0,0 +1,209 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #11453 — `ack()` is the dispatcher's completion callback for a row IT + * CLAIMED, and both implementations must enforce that. + * + * ## The measured defect this file pins the fix for + * + * Neither implementation checked the row's status. Measured on `origin/main` + * @ `a1c804bc9` with this exact harness, on BOTH backends: + * `ack(id, { success: false, suppressed: true })` against a `pending` row that + * was never claimed RESOLVED, flipped the row to terminal `suppressed`, and + * incremented `attempts` to 1 — recording a delivery attempt that never + * happened, on a row no dispatcher ever held. + * + * That made `ack` read like the missing `cancel` primitive, and it is a trap: + * + * - **It races the dispatcher.** Between a caller's `list()` and its `ack()`, + * `claim()` can take the row — `claim` is atomic by contract and `ack` was + * never part of that atom — so a suppression could land on a delivery + * already on the wire, or a dispatcher's real outcome could be overwritten. + * - **It corrupts `attempts`.** The counter feeds the retry schedule + * (`classifyDeliveryAttempt(result, errorClass, row.attempts, …)`), and a + * row "cancelled" this way arrives at its next real attempt with the + * backoff already advanced by an attempt that never went out. + * + * ## Why this file is ONE table over BOTH backends + * + * The precondition is a property of {@link INotificationOutbox}, not of either + * implementation, and the two drifting apart is the specific failure this + * shape prevents: a memory-only pin would let the SQL path keep the hole (it + * is the production store), and a SQL-only pin would let every unit test in + * the repo keep exercising the trap. Every case below therefore runs + * identically against both, and the SQL leg runs on a REAL engine + * (`ObjectQL` + `SqlDriver`, better-sqlite3 `:memory:` — the #5704 ruled test + * backend) rather than a fake, because the SQL half of the fix is an ATOMIC + * conditional UPDATE and a fake engine cannot refuse a write. + * + * ## The vacuity traps closed explicitly + * + * 1. **"a refusal that refuses everything."** Every refusal case is paired + * with a still-works leg on the same backend: a claimed row still acks to + * `success`, and a retry ack still re-arms the row. A precondition that + * rejected unconditionally would fail those. + * 2. **"the assertion passes because nothing ran."** Identities, not counts: + * each case names the row, its expected terminal state AND its `attempts` + * value, so a no-op implementation cannot read as a pass. + * 3. **`toThrow()` alone is not a refusal test.** An un-fixed backend throws + * nothing and a broken one could throw anything, so the refusals below + * assert the ERROR IDENTITY (`name` + ADR-0112 `code`), not merely that + * something was thrown. + */ + +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 both backends must produce. `DELIVERY_NOT_ELIGIBLE` is + * this package's already-registered ADR-0112 code for "this delivery row's + * state does not permit the operation" — the same refusal + * `SqlHttpOutbox.redeliver` raises when its own compare-and-set misses. + */ +const REFUSAL = { name: 'NotificationAckError', code: 'DELIVERY_NOT_ELIGIBLE' }; + +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; + }, + }; +} + +const CLAIM = { nodeId: 'node-a', limit: 10, claimTtlMs: 60_000 }; + +describe.each([memoryBackend(), sqlBackend()])('$name — ack() status precondition (#11453)', (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; + } + + it('refuses ack on an unclaimed pending row, and writes NOTHING', async () => { + 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. + await expect( + outbox.ack(id, { success: false, suppressed: true }), + ).rejects.toMatchObject(REFUSAL); + + // A refused ack is not a partial one: the row keeps its identity, its + // status AND its attempt count, so it is still claimable and its + // backoff position is still honest. + const after = await readRow(id); + expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:pending:0`); + }); + + it('still acks a CLAIMED row — the precondition refuses the unclaimed, not everything', async () => { + const id = await enqueueOne(); + 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(); + + const after = await readRow(id); + expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:success:1`); + }); + + 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 }); + + await expect( + outbox.ack(id, { success: false, suppressed: true }), + ).rejects.toMatchObject(REFUSAL); + + // `attempts` is the assertion that matters: an unconditional increment + // would read 2 here for ONE delivery that went out once. + const after = await readRow(id); + expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:success:1`); + }); + + 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); + + // 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 }); + const retried = await readRow(id); + expect(`${retried.id}:${retried.status}:${retried.attempts}`).toBe(`${id}:pending:1`); + + // …and now it is unclaimed again, so the precondition applies to it + // exactly as it did before the first claim. This is the case a + // read-back implementation gets wrong: the CAS's post-state (`pending`) + // 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 }), + ).rejects.toMatchObject(REFUSAL); + + const after = await readRow(id); + expect(`${after.id}:${after.status}:${after.attempts}`).toBe(`${id}:pending:1`); + }); + + it('a claimed row acked as suppressed by a REAL attempt still reaches suppressed', async () => { + // `classifyDeliveryAttempt` returns `suppressed: true` for an + // `invalid_recipient` send outcome — a genuine attempt whose terminal + // 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); + + await outbox.ack(id, { 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 ca21e8788c..d4f65bd28c 100644 --- a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -50,9 +50,10 @@ import type { EngineUpdateOptions } from '@objectstack/spec/data'; * `multi: true` so they cannot, deliberately. The single-record writes on * these same objects are audited under a DIFFERENT op (`update`, not * `updateMany`) and they do **not** share one classification: - * {@link dispatcherAckOptions} carries the sweep warrant to the two `ack` - * sites, and `SqlHttpOutbox.redeliver` — request-reachable — carries a - * threaded tenant and no bypass at all. + * {@link dispatcherAckOptions} carries the sweep warrant to `SqlHttpOutbox.ack` + * and {@link dispatcherAckCasOptions} carries it to `SqlNotificationOutbox.ack` + * (a `multi: true` compare-and-set since #11453), while `SqlHttpOutbox.redeliver` + * — request-reachable — carries a threaded tenant and no bypass at all. * * ⛔ [#11009] `redeliver` is now ALSO a `multi: true` write (its terminal- * status compare-and-set must ride the predicate path to be evaluated at @@ -73,8 +74,12 @@ export function dispatcherSweepOptions( /** * The write options for a dispatcher **`ack`** — the single-record - * (`multi: false`) write that records one delivery attempt's outcome, on - * `SqlNotificationOutbox.ack` and `SqlHttpOutbox.ack`. + * (`multi: false`) write that records one delivery attempt's outcome on + * `SqlHttpOutbox.ack`. + * + * [#11453] `SqlNotificationOutbox.ack` no longer uses this helper: its ack + * grew a status precondition, and a precondition on the by-id path is silently + * discarded (#11009), so it rides {@link dispatcherAckCasOptions} instead. * * ## Why a second helper instead of {@link dispatcherSweepOptions} * These are audited under the driver's **`update`** op, not `updateMany`, and @@ -123,3 +128,47 @@ export function dispatcherAckOptions( ): EngineUpdateOptions & { multi: false; bypassTenantAudit: true } { return { where: { id }, multi: false, bypassTenantAudit: true }; } + + +/** + * [#11453] The write options for **`SqlNotificationOutbox.ack`** — the same + * warrant as {@link dispatcherAckOptions} above, spelled as a PREDICATE write + * because that ack is now a compare-and-set. + * + * ## Why `multi: true` for a write that still targets ONE row + * + * ⛔ [#11009] Not a preference — a requirement. `ack` re-states its + * precondition IN the write (`where: { id, status: 'in_flight' }`) so a row + * that stopped being claimed underneath is not written. On the by-id path + * (`multi: false`) `driver.update` binds only the primary key and the extra + * predicate is SILENTLY DISCARDED — the identical defect `redeliver` carried: + * the guard evaluates to nothing and the write lands unconditionally. The + * engine now REFUSES that spelling outright, so the predicate path + * (`driver.updateMany`, which compiles every `where` key) is the only spelling + * in which this compare-and-set exists at all. + * + * ## Why not {@link dispatcherSweepOptions}, now that both are `multi: true` + * + * Because that helper's warrant is the CLAIM path's, and this file's own rule + * is that a classification made for one site says nothing about another. The + * warrant here is re-derived and identical in substance to + * {@link dispatcherAckOptions}': `ack`'s only caller is + * `NotificationDispatcher` inside `runPartition()`, a `setInterval` tick under + * the `notify.dispatcher.partition.` cluster lock — no HTTP request, no + * session and no active organization exists to thread, and the row being acked + * was claimed by a deliberately environment-wide sweep. `redeliver` remains + * the one write on these objects that must never reach for a bypass: it is + * request-reachable and threads the caller's tenant instead. + * + * ⚠️ 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. + */ +export function dispatcherAckCasOptions( + id: string, + expectedStatus: 'in_flight', +): EngineUpdateOptions & { multi: true; bypassTenantAudit: true } { + return { where: { id, status: expectedStatus }, multi: true, bypassTenantAudit: true }; +} diff --git a/packages/services/service-messaging/src/outbox.ts b/packages/services/service-messaging/src/outbox.ts index c95f866ae1..5a5fb0690f 100644 --- a/packages/services/service-messaging/src/outbox.ts +++ b/packages/services/service-messaging/src/outbox.ts @@ -106,6 +106,62 @@ export interface AckFailure { export type AckResult = AckSuccess | AckFailure; +/** + * [#11453] Error raised by {@link INotificationOutbox.ack} when the delivery + * row's status does not permit the completion it was handed. + * + * `DELIVERY_NOT_ELIGIBLE` is this package's already-registered ADR-0112 code + * for "this delivery row's state does not permit the requested operation" — + * the same refusal `SqlHttpOutbox.redeliver` raises when its own + * compare-and-set misses. Reused deliberately rather than minted: the two + * refusals are one concept on two delivery surfaces, and a second spelling + * would be a second thing for a caller to match on. + */ +export class NotificationAckError extends Error { + constructor( + message: string, + readonly code: 'DELIVERY_NOT_ELIGIBLE', + ) { + super(message); + this.name = 'NotificationAckError'; + } +} + +/** + * The refusal message, in ONE place both implementations call — so the memory + * and SQL backends cannot drift into two different wordings for one contract + * violation (the drift this card's contract test exists to prevent). + */ +export function notificationAckNotClaimedMessage(id: string, status: DeliveryStatus | 'unknown'): string { + return ( + `Delivery row '${id}' is '${status}', not 'in_flight': ack() records the outcome of a delivery ` + + 'the caller CLAIMED, and this row is not claimed. Acking an unclaimed row would race the ' + + "dispatcher (claim() is atomic by contract and ack() is not part of that atom) and record an " + + 'attempt that never went on the wire. To stop a pending delivery, there is deliberately no ' + + 'ack() spelling — see #11453.' + ); +} + +/** + * 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. + * + * Distinguished from {@link notificationAckNotClaimedMessage} because the two + * say different things to whoever reads the log: the first is a CALLER using + * `ack` wrongly, the second is a dispatcher that lost a race it is allowed to + * lose. Both write nothing. + */ +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 ` + + '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).' + ); +} + /** * Pluggable storage for delivery rows. `claim()` MUST be atomic across * concurrent callers (the at-least-once guarantee), and `enqueue()` MUST treat @@ -115,6 +171,22 @@ export type AckResult = AckSuccess | AckFailure; export interface INotificationOutbox { enqueue(input: EnqueueDeliveryInput): Promise; claim(opts: ClaimOptions): Promise; + /** + * Record the outcome of ONE dispatch attempt on a row this caller claimed. + * + * ⛔ **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. + * + * Implementations MUST make the transition atomic against {@link claim}: + * the status test and the write are one operation, never a read-then-write. + */ ack(id: string, result: AckResult): Promise; list(filter?: { status?: DeliveryStatus; notificationId?: string }): 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 8ffd92346c..aa483a750a 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 @@ -30,11 +30,24 @@ interface RecordedUpdate { * Minimal recording `IDataEngine`. `find` replays a scripted queue of result * sets so a claim can be driven all the way through candidate-select → * atomic-claim → read-back; everything else is inert. + * + * [#11453] `SqlNotificationOutbox.ack` is a compare-and-set now — it reads the + * row's STATUS as well as its attempts, writes conditionally, then reads back + * to confirm the write landed. So the fake keeps one row's state and applies + * writes to it, rather than answering a bare `{ attempts }` forever: + * + * - the row starts `in_flight`, because acking a row is something a + * dispatcher does to a row IT CLAIMED, and a fixture that left it `pending` + * would be asserting audit columns on a call the contract now refuses; + * - `update` applies the payload, so the read-back sees what was written and + * the ack completes. A fake that could never report the write would make + * every ack look like a lost claim. */ 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 }; const engine = { async insert(object: string, data: Record) { @@ -43,21 +56,22 @@ function makeEngine(findResults: Array>> = []) { }, async update(object: string, data: Record) { updates.push({ object, data }); - return { matched: 0, modified: 0 }; + Object.assign(row, data); + return { matched: 1, modified: 1 }; }, async find(_object: string) { return findResults[findCall++] ?? []; }, async findOne(_object: string, opts?: { fields?: string[] }) { - // `ack()` reads the current attempt count; `enqueue()` probes for a + // `ack()` reads the row it is completing; `enqueue()` probes for a // dedup winner and must miss so the insert path runs. - if (opts?.fields?.includes('attempts')) return { attempts: 2 }; + if (opts?.fields?.includes('attempts')) return { ...row }; return null; }, async delete() { return { matched: 0, modified: 0 }; }, } as unknown as IDataEngine; - return { engine, updates, inserts }; + return { engine, updates, inserts, row }; } /** Every audit column an UPDATE payload must leave to the platform. */ diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index 9e9a5790ad..5559898dea 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -12,7 +12,8 @@ import type { } from './outbox.js'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; -import { dispatcherAckOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; +import { dispatcherAckCasOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; +import { NotificationAckError, notificationAckLostClaimMessage, notificationAckNotClaimedMessage } from './outbox.js'; export const DELIVERY_OBJECT = 'sys_notification_delivery'; @@ -212,9 +213,20 @@ export class SqlNotificationOutbox implements INotificationOutbox { async ack(id: string, result: AckResult): Promise { const current = (await this.engine.findOne(this.objectName, { where: { id }, - fields: ['attempts'], - })) as { attempts?: number } | null; + fields: ['status', 'attempts'], + })) as { status?: DeliveryStatus; attempts?: number } | 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; + // [#11453] Precondition, half one: the loud, deterministic refusal for + // a row that is not claimed at all — the ack-as-cancel trap. Refused + // BEFORE any write, so a refused ack leaves the row byte-identical. + if (current.status !== 'in_flight') { + throw new NotificationAckError( + notificationAckNotClaimedMessage(id, current.status ?? 'unknown'), + 'DELIVERY_NOT_ELIGIBLE', + ); + } const now = Date.now(); let status: DeliveryStatus; @@ -235,22 +247,57 @@ 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. + // + // `attempts` is incremented HERE and only here, inside that condition, + // so the counter can only move for a row that was genuinely claimed — + // i.e. for a real dispatch attempt. Previously this write was + // unconditional and by-id, so any caller could advance the retry + // schedule of a row no dispatcher ever held. + const attempts = (current.attempts ?? 0) + 1; await this.engine.update( this.objectName, { status, - attempts: (current.attempts ?? 0) + 1, + attempts, last_attempted_at: now, claimed_by: null, claimed_at: null, next_attempt_at: nextAttemptAt, error, }, - // Single-record dispatcher write, audited under the `update` op. - // Declared a global-sweep site — no request context exists on the - // tick that reaches here. Warrant in `outbox-dispatcher-scope.ts`. - dispatcherAckOptions(id) as any, + // 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, ); + + // Did the conditional write land? `IDataEngine.update` declares its + // return as `any`, so the row itself is the only contract-safe answer + // — the same read-back `SqlHttpOutbox.redeliver` uses to report its own + // compare-and-set miss. Without it a lost claim would write nothing and + // still report success, which is the silent-success family this card + // exists to close. + // + // The detector is the pair (status, attempts), not status alone: a + // retry ack's post-state IS `pending`, the same status a refused row + // already had, so only the recorded attempt tells the two apart. + const after = (await this.engine.findOne(this.objectName, { + where: { id }, + fields: ['status', 'attempts'], + })) as { status?: DeliveryStatus; attempts?: number } | null; + if (!after || after.status !== status || (after.attempts ?? 0) !== attempts) { + throw new NotificationAckError( + notificationAckLostClaimMessage(id, after?.status ?? 'unknown'), + 'DELIVERY_NOT_ELIGIBLE', + ); + } } async list(filter?: { status?: DeliveryStatus; notificationId?: string }): Promise {