diff --git a/.changeset/17634-http-ack-claim-credential.md b/.changeset/17634-http-ack-claim-credential.md new file mode 100644 index 0000000000..738a2744bc --- /dev/null +++ b/.changeset/17634-http-ack-claim-credential.md @@ -0,0 +1,7 @@ +--- +'@objectstack/service-messaging': minor +--- + +`IHttpOutbox.ack()` takes an optional third argument, the claim credential, and `HttpDispatcher` now always passes it (#17634). A late ack from a claim the visibility-timeout reap had taken back — a send that outran `claimTtlMs` while another dispatcher re-claimed the row — used to write its outcome by row id over that dispatcher's live attempt: a delivery still in progress could be marked `dead`, or one attempt's outcome overwrite another's. Handed the credential, `SqlHttpOutbox` and `MemoryHttpOutbox` perform the compare-and-set `INotificationOutbox.ack()` has performed since #11859: the outcome is written only while the row is still `in_flight` under the same (`claimedBy`, `claimedAt`) pair `claim()` stamped on it. A lost claim writes nothing and throws the new `HttpAckError` (`DELIVERY_NOT_ELIGIBLE`, the code this package already raises for a delivery row in the wrong state); the dispatcher logs `http-dispatcher: ack refused, claim no longer held`, carries on with the rest of its batch, and whoever holds the row re-drives the delivery. + +Nothing written against the two-argument `ack(id, result)` has to change. An `IHttpOutbox` implementation that does not read the third argument compiles and works as before, and a caller that does not pass it gets the by-id write it always got — that arity is deprecated, because it checks no ownership. New exports: `HttpClaimCredential` and `HttpAckError`. A subclass that overrides a built-in store's `ack()` should forward the third argument to `super.ack()`, or its dispatcher acks keep the old unchecked write. diff --git a/content/docs/automation/webhooks.mdx b/content/docs/automation/webhooks.mdx index f94b92ab59..77abaa8d34 100644 --- a/content/docs/automation/webhooks.mdx +++ b/content/docs/automation/webhooks.mdx @@ -622,6 +622,7 @@ A precise table of what the runtime promises and what it does not. | Producer node crashes mid-emit | **Not durable today.** The realtime bus (`InMemoryRealtimeAdapter`) is an unpersisted, in-process pub/sub — an event lost before Stage 3's INSERT is gone, not redelivered (see §4.1). | | Subscriber node crashes after persist | Row exists in `sys_http_delivery`, another node picks it up. | | Dispatcher node crashes mid-HTTP | Row stays `in_flight` with `claimed_by`; the first dispatcher tick after the claim TTL reverts it to `pending` and it is re-posted. The TTL derives from the dispatcher tick (`intervalMs`, default 500ms): `lockTtlMs = 5 × intervalMs`, `claimTtlMs = 2 × lockTtlMs` (so ~5s at defaults), all configurable via `HttpDispatcherOptions`. An idle surviving dispatcher ticks at least every `maxIdleIntervalMs` (default 30s), so recovery takes at most `claimTtlMs + maxIdleIntervalMs` (~35s at defaults). | +| A send outlasts the claim TTL and another dispatcher re-claims the row | The re-claiming dispatcher posts the delivery again (at-least-once — the receiver can see it twice; see the last row). The first dispatcher's late outcome is **not** written over the live attempt: its ack carries the claim credential (`claimed_by`, `claimed_at`) its claim stamped, the conditional update matches nothing, and it logs `http-dispatcher: ack refused, claim no longer held`. The row records the outcome of the attempt that still holds it. | | Receiver returns 5xx | Retry per backoff schedule until the fixed 8-attempt budget is exhausted (§4.5). | | Receiver returns 4xx | Treated as terminal — no retry, status `dead` immediately. Exception: 408 / 429 are retried. | | Receiver returns 2xx | `status = success`, no more attempts. | diff --git a/content/docs/permissions/tenant-audit-census.mdx b/content/docs/permissions/tenant-audit-census.mdx index 611e972c22..afccd682f0 100644 --- a/content/docs/permissions/tenant-audit-census.mdx +++ b/content/docs/permissions/tenant-audit-census.mdx @@ -98,7 +98,7 @@ are reported as `undecidable` rather than assumed either way. The same holds twice over for the context. An options argument spelled as a literal can be read; one spelled `options`, `{ ...opts }`, or handed through a -forwarding shim cannot, and **66 of the 222 sites are spelled that way**. A +forwarding shim cannot, and **67 of the 223 sites are spelled that way**. A context resolved from an inline literal or a local `const` can be tested for `isSystem`; one arriving from a helper call cannot. @@ -126,8 +126,8 @@ now **0**: nothing on this surface threads a context that provably lacks the fla **"No tenant context" counted sites it had not read.** An options argument the walker could not parse was folded into the same bucket as one it had read and -found empty. That published **83 sites "carrying no tenant context at all"** -when 17 said so and 66 were simply unread — an over-claim in the *alarming* +found empty. That published **84 sites "carrying no tenant context at all"** +when 17 said so and 67 were simply unread — an over-claim in the *alarming* direction, on the very figure this page tells other cards to cite. `carries` is now three-valued, and an unreadable argument can never contribute to the provable count. @@ -147,10 +147,10 @@ reproduce them. Where it disagrees, it disagrees on the page: | carried figure | where it survives | this census | | :--- | :--- | ---: | -| 175 write call sites | quoted in the merged changeset | **222** | +| 175 write call sites | quoted in the merged changeset | **223** | | 24 carrying no tenant context | quoted in the merged changeset | **9** provable and tenancy-enabled; **32** more whose options argument is unreadable | -| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **149 of 222** decidable, **73** undecidable | -| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 105 decidably elevated, 0 decidably not, 100 undecidable | +| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **149 of 223** decidable, **74** undecidable | +| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 105 decidably elevated, 0 decidably not, 101 undecidable | | 141 and 132, two independent re-derivations | the card that filed this work | — | **The differences are not reconciled, and deliberately so.** The old census's @@ -167,11 +167,11 @@ would report a smaller number and would not say so. The fourth row is the one worth flagging to anyone citing it. **The 135 / 77% figure has no surviving corroboration anywhere in the tree.** This census reads -105 of 222 (47%) as decidably elevated, with 100 more whose elevation is a +105 of 223 (47%) as decidably elevated, with 101 more whose elevation is a run-time fact — so the claim is neither confirmed nor refuted, and the honest answer is that a static reading cannot settle it. -⇒ **Cite `9 / 222`, and say what it is**: the sites whose options argument was +⇒ **Cite `9 / 223`, and say what it is**: the sites whose options argument was READ and holds no tenant context, against a decidably tenancy-enabled object. That is the control's provable yield surface. ⛔ Do not cite it as "the sites without tenant context" — **32 further sites** have an options argument this @@ -183,23 +183,23 @@ cannot read, and they are neither in nor out. | what | count | | :--- | ---: | -| write call sites on the application surface | **222** | +| write call sites on the application surface | **223** | | …whose object name is statically decidable | 149 | -| …whose object name is chosen at run time | 73 | +| …whose object name is chosen at run time | 74 | | …against an object with tenancy ENABLED | 149 | | …against an object that declares tenancy off | 0 | | threading a tenant context | 139 | | PROVABLY carrying none (options read, no context key) | **17** | | …of those, against a decidably tenancy-enabled object | **9** | -| options argument UNREADABLE — may or may not carry one | 66 | +| options argument UNREADABLE — may or may not carry one | 67 | | …of those, against a decidably tenancy-enabled object | 32 | | threading a decidably ELEVATED (`isSystem`) context | 105 | | threading a context that is decidably NOT elevated | 0 | -| threading a context whose elevation is a run-time fact | 100 | +| threading a context whose elevation is a run-time fact | 101 | | how the instrument reached the site | count | | :--- | ---: | -| receiver carried a readable engine type | 177 | +| receiver carried a readable engine type | 178 | | receiver erased, placed by the object NAME | 19 | | receiver erased, placed by an `object: string` PARAMETER | 15 | | receiver erased, placed by an `UNTYPED_RECEIVERS` row | 11 | @@ -207,7 +207,7 @@ cannot read, and they are neither in nor out. | object name spelled inline | 109 | | object name spelled through a `const` | 40 | | object name is an `object: string` parameter | 19 | -| object name is some other run-time expression | 54 | +| object name is some other run-time expression | 55 | The corpus walked is every tracked non-test source under `packages/services/` and `packages/plugins/`; calls to a same-named method on something that is not @@ -224,11 +224,11 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-11 at `f368fd0aa`. +Measured on 2026-09-11 at `06215b074`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 563 | +| tracked non-test sources scanned | 564 | | engine-shaped types recognised | 59 | | declared objects in the registry | 300 | | same-named calls subtracted as non-engine | 137 | diff --git a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md index 7f3bed3901..ef136b4152 100644 --- a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md +++ b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md @@ -29,19 +29,19 @@ silent, and `node scripts/tenant-audit-census.mjs --write` is the resolution. | Measure | Value | |---|---:| -| Write call sites | 222 | +| Write call sites | 223 | | Object name statically decidable | 149 | -| Object name chosen at run time | 73 | +| Object name chosen at run time | 74 | | Against a tenancy-enabled object | 149 | | Against an object declaring tenancy off | 0 | | Threading a tenant context | 139 | | Provably carrying none | 17 | | …and decidably tenancy-enabled | 9 | -| Options argument unreadable | 66 | +| Options argument unreadable | 67 | | …and decidably tenancy-enabled | 32 | | Threading a decidably elevated context | 105 | | Threading a decidably non-elevated context | 0 | -| Threading a context of undecidable elevation | 100 | +| Threading a context of undecidable elevation | 101 | ## Corpus scale — present and dated, ⛔ NOT enforced @@ -52,11 +52,11 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-11 at `f368fd0aa`. +Measured on 2026-09-11 at `06215b074`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 563 | +| tracked non-test sources scanned | 564 | | engine-shaped types recognised | 59 | | declared objects in the registry | 300 | | same-named calls subtracted as non-engine | 137 | @@ -179,7 +179,7 @@ Measured on 2026-09-11 at `f368fd0aa`. | `packages/services/service-messaging/src/messaging-service.ts` | `update` | `RECEIPT_OBJECT` | undecidable | options unreadable | 1 | | `packages/services/service-messaging/src/messaging-service.ts` | `insert` | `sys_notification` | enabled | options unreadable | 1 | | `packages/services/service-messaging/src/sql-http-outbox.ts` | `insert` | `this.objectName` | undecidable | options unreadable | 1 | -| `packages/services/service-messaging/src/sql-http-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 4 | +| `packages/services/service-messaging/src/sql-http-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 5 | | `packages/services/service-messaging/src/sql-outbox.ts` | `insert` | `this.objectName` | undecidable | options unreadable | 1 | | `packages/services/service-messaging/src/sql-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 4 | | `packages/services/service-queue/src/db-queue-adapter.ts` | `delete` | `sys_job_queue` | enabled | context, elevation undecidable | 2 | 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 a286d5b1ab..5815f5c73b 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 @@ -25,10 +25,15 @@ * `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. + * separately: the op moved, the warrant did not. * - * The `ack` pair is declared global (`dispatcherAckOptions`, warrant in + * [#17634] `SqlHttpOutbox.ack` has since made the same move: handed the claim + * credential — which `HttpDispatcher` always hands it — its ownership test is + * a compare-and-set, so its op is `updateMany`, via `dispatcherAckCasOptions`, + * with the classification unchanged. Only its deprecated credential-less arity + * still writes by id, and no dispatcher tick reaches it. + * + * The `ack` pair is declared global (`dispatcherAckCasOptions`, 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 * `bypassTenantAudit` on that third site would convert a detectable hole into @@ -56,9 +61,9 @@ * ## The vacuity traps closed explicitly * 1. **"the audit was never armed."** Every silence assertion is followed by * a positive control on the SAME object through the SAME driver: an - * unscoped by-id `update` that MUST produce the line. The gate throttles - * one warning per `${object}:${op}`, so the control runs last and only - * fires if the production path consumed no `update` warning of its own. + * unscoped write on the SAME op that MUST produce the line. The gate + * throttles one warning per `${object}:${op}`, so the control runs last and + * only fires if the production path consumed no warning on that op itself. * 2. **"a fix that touches nothing."** Row state is pinned after every write. * 3. **"a refusal that refuses everything."** The cross-tenant refusal is * paired with a still-works leg: an in-tenant redeliver still succeeds. @@ -84,13 +89,9 @@ 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, and the notification `ack`'s since #11453. */ +/** Every `options` bag that reached `SqlDriver.updateMany` — `redeliver`'s op since #11009, the notification `ack`'s since #11453, and the HTTP `ack`'s since #17634. */ 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 notification `ack`'s since #11453. */ const auditedUpdateMany = (object: string): boolean => warns.some((w) => w.msg.includes(`[tenant-audit] updateMany on tenant-scoped object "${object}"`)); @@ -145,30 +146,6 @@ afterEach(async () => { else process.env.OS_TENANT_AUDIT = OLD_AUDIT; }); -/** - * The positive control for the `update` op. A by-id write with no `tenantId` - * and no bypass MUST produce the audit line on `object`, or this file cannot - * tell "classified" from "the gate was never armed". Run AFTER the assertion - * it guards — the gate throttles one warning per `${object}:${op}`. - */ -async function controlUnscopedUpdate(object: string, existingId: string): Promise { - // `where: { id }` with a scalar id routes through `driver.update` - // (`resolveEngineUpdateDispatch` → `by-id`), exactly as the production - // paths under test do. - // - // ⚠️ It must name a row that EXISTS. The engine's by-id branch raises - // `Record not found` before it ever reaches the driver, so a control - // pointed at a missing id never arms the gate it is meant to prove is - // armed — it fails as an error rather than reporting a vacuous suite, - // which is the only reason that mistake was visible here. - await engine.update(object, { attempts: 99 }, { where: { id: existingId } } as any); - expect( - auditedUpdate(object), - `positive control failed: an unscoped by-id update on ${object} produced no [tenant-audit] ` - + 'line, so every "no finding" assertion in this file is vacuous', - ).toBe(true); -} - /** * The positive control for the `updateMany` op — `redeliver`'s op since * #11009. An unscoped predicate write with no bypass MUST produce the @@ -221,7 +198,7 @@ async function seedDeadRow(id: string, org: string): Promise { } // ─────────────────────────────────────────────────────────────────────────── -describe('ack — the two dispatcher sites are a classified global sweep (update + updateMany ops)', () => { +describe('ack — the two dispatcher sites are a classified global sweep (updateMany op)', () => { 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'); @@ -246,14 +223,30 @@ describe('ack — the two dispatcher sites are a classified global sweep (update ]); // ② Both organizations' rows were acked by one dispatcher — the // cross-organization reach is the operation's semantics. - const ackWrites = driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY); - expect(ackWrites.map((u) => u.id).sort()).toEqual(['h_a', 'h_b']); + // + // [#17634] The dispatcher's ack hands the claim credential, so it is a + // compare-and-set on the predicate path and the reading moves to the + // `updateMany` spy — the move the notification ack made in #11453. The + // claim path writes there too (its reap and its atomic claim), so the + // filter names what an ACK write looks like: a scalar id bound to + // `in_flight` AND to the claiming node. That predicate 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 === SYS_HTTP_DELIVERY + && typeof (u.where as any)?.id === 'string' + && (u.where as any)?.status === 'in_flight' + && (u.where as any)?.claimed_by === 'n1', + ); + expect(ackWrites.map((u) => (u.where as any).id).sort()).toEqual(['h_a', 'h_b']); // ③ …under the DECLARED classification, not an accidental silence. expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true); expect(ackWrites.every((u) => u.options?.tenantId === undefined)).toBe(true); - expect(auditedUpdate(SYS_HTTP_DELIVERY)).toBe(false); + // …and no by-id write reached the driver: the dispatcher never takes the deprecated arity. + expect(driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY)).toEqual([]); + expect(auditedUpdateMany(SYS_HTTP_DELIVERY)).toBe(false); - await controlUnscopedUpdate(SYS_HTTP_DELIVERY, 'h_a'); + await controlUnscopedUpdateMany(SYS_HTTP_DELIVERY, 'h_a'); }); it('SqlNotificationOutbox.ack records a REAL delivery in every organization, without a finding', async () => { diff --git a/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts b/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts index b798c59cee..1896d5422d 100644 --- a/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts +++ b/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts @@ -23,7 +23,7 @@ import { HttpDispatcher } from './http-dispatcher.js'; import { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatcher.js'; import { MessagingService } from './messaging-service.js'; import type { FetchImpl } from './http-sender.js'; -import type { EnqueueHttpInput, HttpAckResult, HttpClaimOptions, HttpDelivery } from './http-outbox.js'; +import type { EnqueueHttpInput, HttpAckResult, HttpClaimCredential, HttpClaimOptions, HttpDelivery } from './http-outbox.js'; const BASE = 500; const CAP = 30_000; @@ -39,9 +39,10 @@ class TickRecordingOutbox extends MemoryHttpOutbox { this.tickStarts.push(Date.now()); return super.claim(opts); } - override async ack(id: string, result: HttpAckResult): Promise { + override async ack(id: string, result: HttpAckResult, claimed?: HttpClaimCredential): Promise { if (!result.success && result.nextRetryAt !== undefined) this.retriesDueAt.push(result.nextRetryAt); - return super.ack(id, result); + // [#17634] Forward the claim credential, so the dispatcher's acks keep their ownership check. + return super.ack(id, result, claimed); } } diff --git a/packages/services/service-messaging/src/http-dispatcher.ts b/packages/services/service-messaging/src/http-dispatcher.ts index fdec977f13..f8e5db21f4 100644 --- a/packages/services/service-messaging/src/http-dispatcher.ts +++ b/packages/services/service-messaging/src/http-dispatcher.ts @@ -3,7 +3,7 @@ import type { DispatchCluster, DispatchLockHandle } from './dispatcher.js'; import { DispatchLoop } from './dispatch-loop.js'; import { classifyAttempt, sendOnce, type FetchImpl } from './http-sender.js'; -import type { HttpDelivery, IHttpOutbox } from './http-outbox.js'; +import type { HttpAckResult, HttpClaimCredential, HttpDelivery, IHttpOutbox } from './http-outbox.js'; /** * HttpDispatcher (ADR-0018 M3) — drains the generic outbound-HTTP outbox @@ -20,6 +20,12 @@ import type { HttpDelivery, IHttpOutbox } from './http-outbox.js'; * to pending after the claim TTL and is re-posted. Receivers MUST be idempotent * on the `X-Objectstack-Delivery` (== row id) header. * + * A send that outruns the claim TTL can be reaped and re-claimed — by another + * node, or by this one — while it is still on the wire. Its ack hands the store + * the claim credential this node's `claim()` stamped, so the outcome is written + * only while that claim still holds the row; a lost claim's ack is refused, + * logged and absorbed, never written over the live attempt (#17634). + * * ## What an idle tick costs (#17623) * * Against an empty outbox one tick is `1 + partitionCount` store round trips: @@ -245,7 +251,7 @@ export class HttpDispatcher { | undefined; if (!fetchImpl) { this.opts.logger?.warn?.('http-dispatcher: no fetch impl available', { rowId: row.id }); - await this.opts.outbox.ack(row.id, { + await this.ackAttempt(row, { success: false, error: 'no fetch implementation', durationMs: 0, @@ -255,9 +261,43 @@ export class HttpDispatcher { } const outcome = await sendOnce(row, fetchImpl); const result = classifyAttempt(outcome, row.attempts, this.opts.now?.() ?? Date.now(), this.opts.rng); - await this.opts.outbox.ack(row.id, result); + await this.ackAttempt(row, result); this.opts.onAttempt?.(row, result.success); } + + /** + * [#17634] Record one attempt's outcome with the claim credential this + * node's `claim()` stamped on the row, tolerating the ONE refusal a correct + * dispatcher can legitimately provoke — `NotificationDispatcher.ackAttempt`'s + * shape (#11453, #11859). + * + * A send slower than `claimTtlMs` lets the visibility-timeout reap return the + * row to `pending`, and another node — or this one, on a later tick — + * re-claims it. By the time this ack runs the row is not ours: a race we are + * ALLOWED to lose, because whoever holds the row now re-drives the delivery + * (at-least-once), and the store's refusal is it correctly declining to write + * our outcome over that live attempt. So the refusal is logged and absorbed. + * ⛔ It must not abort the tick: the rows still validly claimed by this node + * come after this one, and unwinding the partition loop would strand each of + * them `in_flight` until its own timeout. Only `DELIVERY_NOT_ELIGIBLE` is + * absorbed; a store fault still propagates. + */ + private async ackAttempt(row: HttpDelivery, result: HttpAckResult): Promise { + try { + // The row IS the record `claim()` returned, so its (claimedBy, + // claimedAt) pair is the credential — handed back whole, never + // rebuilt from this node's id. A store written before the parameter + // ignores it; a store that honours it re-checks it at run time. + await this.opts.outbox.ack(row.id, result, row as HttpClaimCredential); + } catch (err) { + if ((err as { code?: string })?.code !== 'DELIVERY_NOT_ELIGIBLE') throw err; + this.opts.logger?.warn?.('http-dispatcher: ack refused, claim no longer held', { + nodeId: this.opts.nodeId, + deliveryId: row.id, + error: (err as Error)?.message ?? String(err), + }); + } + } } /** Spread starting partition per node so nodes don't serialise on partition 0. */ diff --git a/packages/services/service-messaging/src/http-outbox-ack-claim-ownership.integration.test.ts b/packages/services/service-messaging/src/http-outbox-ack-claim-ownership.integration.test.ts new file mode 100644 index 0000000000..810c0c718e --- /dev/null +++ b/packages/services/service-messaging/src/http-outbox-ack-claim-ownership.integration.test.ts @@ -0,0 +1,402 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17634 — an HTTP ack proves OWNERSHIP of the claim it completes, not just + * that a row id exists: a late ack from a claim the visibility-timeout reap + * took back must not overwrite the live re-claim on `sys_http_delivery`. The + * notification outbox closed the same shape in #11859; this file pins the HTTP + * outbox against the same sequence. + * + * ## The reachable sequence this file replays — for real + * + * 1. node A claims row R and starts a POST; + * 2. the POST outruns `claimTtlMs`; + * 3. node B's tick reaps R back to `pending` and re-claims it — R is + * `in_flight` again, claimed by B, and B is POSTing; + * 4. A's POST finishes and A acks. Without the claim credential both stores + * write by id, so A's outcome (here a permanent 410 → `dead`) lands over + * B's live attempt. + * + * Two blocks, one table of stores: + * + * - **Through the dispatcher.** Two real `HttpDispatcher`s share one store, + * with an injected clock and gated fetches: the defect lives in the + * interaction of the reap, the re-claim and the late ack, and in what the + * dispatcher hands the store at ack time. No step pokes the store by hand. + * - **Through the store contract.** `claim()` with an explicit `now`, then + * `ack(id, result, claimed)` with the credential the claim returned — the + * refusals a pair of dispatchers cannot reach deterministically: the same + * node's own re-claim, a claim reaped and not re-claimed, a credential + * missing a member. + * + * ## The vacuity traps closed explicitly + * + * - **"Refused" cannot be told from "never ran" by the refusal alone.** Each + * refusal leg pins what the ack did NOT do — the row still reads as the live + * claim, with no attempt and no response code recorded — and, where a live + * claim exists, that its own ack then lands with its own outcome. + * - **A store that refuses EVERY ack passes the refusal legs.** Each block has + * a negative control: the same sequence WITHOUT the reap, whose ack MUST + * land. + * - **`toThrow()` alone proves nothing** — an unfixed store 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. + * - **A fix that stops honouring the two-argument `ack` breaks every outbox + * written against it.** The compatibility leg runs the dispatcher over an + * outbox whose `ack` reads `(id, result)` only, and requires the delivery to + * land. + * + * ## Why both stores, one table + * + * The guarantee is a property of {@link IHttpOutbox}. The SQL leg runs on a + * REAL engine (`ObjectQL` + `SqlDriver`, better-sqlite3 `:memory:`) because the + * fix IS a conditional UPDATE and a fake engine cannot refuse a write; the + * memory leg keeps every test built on `MemoryHttpOutbox` 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 { MemoryHttpOutbox } from './memory-http-outbox.js'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; +import { HttpDelivery } from './objects/http-delivery.object.js'; +import { HttpDispatcher } from './http-dispatcher.js'; +import type { FetchImpl } from './http-sender.js'; +import type { + EnqueueHttpInput, + HttpClaimCredential, + HttpDelivery as HttpDeliveryRow, + IHttpOutbox, +} from './http-outbox.js'; + +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 controls: inside the timeout. */ +const T_WITHIN_TTL = T0 + TTL - 1; + +/** What the dispatcher logs when the store refuses a lost claim's ack — the notification dispatcher's words. */ +const ACK_REFUSED_WARN = 'http-dispatcher: ack refused, claim no longer held'; +/** The refusal identity — `name` + ADR-0112 `code`, never a bare throw. */ +const REFUSAL = { name: 'HttpAckError', code: 'DELIVERY_NOT_ELIGIBLE' }; + +const DELIVERY: EnqueueHttpInput = { + source: 'flow', refId: 'r1', dedupKey: 'd_r1', url: 'https://receiver.example/r1', payload: { n: 1 }, +}; + +type BuiltInHttpOutbox = MemoryHttpOutbox | SqlHttpOutbox; + +interface Backend { + readonly name: string; + create(): Promise; + destroy(): Promise; +} + +function memoryBackend(): Backend { + return { + name: 'MemoryHttpOutbox', + async create() { return new MemoryHttpOutbox(); }, + async destroy() { /* nothing to tear down */ }, + }; +} + +function sqlBackend(): Backend { + let engine: ObjectQL | undefined; + return { + name: 'SqlHttpOutbox', + 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(HttpDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); + return new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + }, + async destroy() { + try { await engine?.destroy(); } catch { /* noop */ } + engine = undefined; + }, + }; +} + +function respond(status: number) { + return { ok: status >= 200 && status < 300, status, async text() { return `status ${status}`; } }; +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((r) => { resolve = r; }); + return { promise, resolve }; +} + +/** One line a human can diff: status, holder, claim instant, attempts, and WHOSE response was recorded. */ +function fingerprint(r: HttpDeliveryRow): string { + return `${r.status}:${r.claimedBy ?? '-'}:${r.claimedAt ?? '-'}:${r.attempts}:${r.responseCode ?? '-'}`; +} + +function claimOpts(nodeId: string, now: number) { + return { nodeId, limit: 10, claimTtlMs: TTL, now }; +} + +/** The credential on a row `claim()` returned — asserted present, because handing it back IS the contract. */ +function credentialOf(row: HttpDeliveryRow): HttpClaimCredential { + const { claimedBy, claimedAt } = row; + if (typeof claimedBy !== 'string' || typeof claimedAt !== 'number') { + throw new Error(`claim() returned row '${row.id}' without its claim credential`); + } + return { claimedBy, claimedAt }; +} + +describe.each([memoryBackend(), sqlBackend()])('$name — HttpDispatcher acks with the claim it holds (#17634)', (backend) => { + let outbox: BuiltInHttpOutbox; + /** Held-open POSTs to release before the store goes away, so a red leg cannot strand one. */ + let cleanups: Array<() => Promise>; + + beforeEach(async () => { + outbox = await backend.create(); + cleanups = []; + }); + afterEach(async () => { + for (const cleanup of cleanups) await cleanup(); + await backend.destroy(); + }); + + async function readRow(id: string): Promise { + const row = (await outbox.list()).find((r) => r.id === id); + if (!row) throw new Error(`row '${id}' vanished — the harness, not the contract, is broken`); + return row; + } + + /** + * Two dispatchers on one store and one injected clock. Node A claims the + * row at `T0`; its POST moves the clock to `secondTickAt` and runs node B's + * tick, then answers 410 — a permanent failure, so A's outcome is `dead`, + * the card's shape — once B is either POSTing the row it re-claimed or has + * finished a tick that took nothing. B's POST is held open until the leg + * calls `finishB()`, so A's ack lands while B is still sending. + */ + async function raceTwoDispatchers(secondTickAt: number) { + const id = await outbox.enqueue(DELIVERY); + let clock = T0; + const now = () => clock; + const posts: string[] = []; + const warnsA: Array<{ msg: string; meta: unknown }> = []; + const bSending = deferred(); + const releaseB = deferred(); + let bTick: Promise = Promise.resolve(); + cleanups.push(async () => { releaseB.resolve(); await bTick; }); + + const bFetch: FetchImpl = async () => { + posts.push('node-b'); + bSending.resolve(); + await releaseB.promise; + return respond(200); + }; + const b = new HttpDispatcher({ + nodeId: 'node-b', outbox, fetchImpl: bFetch, partitionCount: 1, claimTtlMs: TTL, intervalMs: 10_000, now, + }); + + const aFetch: FetchImpl = async () => { + posts.push('node-a'); + // 2. A's POST takes until `secondTickAt`… + clock = secondTickAt; + // 3. …and node B ticks meanwhile. + bTick = b.tick(); + await Promise.race([bSending.promise, bTick]); + return respond(410); + }; + const a = new HttpDispatcher({ + nodeId: 'node-a', outbox, fetchImpl: aFetch, partitionCount: 1, claimTtlMs: TTL, intervalMs: 10_000, now, + logger: { warn: (msg, meta) => { warnsA.push({ msg, meta }); } }, + }); + + // 1. + 4. A claims, POSTs and acks: one whole tick. + await a.tick(); + return { + id, + posts, + warnsA, + async finishB() { releaseB.resolve(); await bTick; }, + }; + } + + it("replays the card: A's late ack leaves B's live re-claim alone, and B's own ack lands", async () => { + const race = await raceTwoDispatchers(T_AFTER_TTL); + + // Both nodes really POSTed: B re-claimed the row A's POST outran. + expect(race.posts).toEqual(['node-a', 'node-b']); + + // A's ack did NOT land: the row still reads as B's claim — B's claim + // instant, no attempt recorded, and A's 410 nowhere on it. + expect(fingerprint(await readRow(race.id))).toBe(`in_flight:node-b:${T_AFTER_TTL}:0:-`); + // …and A's dispatcher said so, once, naming itself and the row. + expect(race.warnsA).toEqual([ + { msg: ACK_REFUSED_WARN, meta: expect.objectContaining({ nodeId: 'node-a', deliveryId: race.id }) }, + ]); + + // B finishes: its ack — the live attempt A would have overwritten — + // records exactly one attempt, with B's outcome. + await race.finishB(); + expect(fingerprint(await readRow(race.id))).toBe('success:-:-:1:200'); + }); + + it("negative control: the same two dispatchers WITHOUT the reap — A's ack lands", async () => { + const race = await raceTwoDispatchers(T_WITHIN_TTL); + + // B's tick found the row still inside A's claim and took nothing. + expect(race.posts).toEqual(['node-a']); + // A's ack, completing the claim it still holds, MUST land. + expect(fingerprint(await readRow(race.id))).toBe('dead:-:-:1:410'); + expect(race.warnsA).toEqual([]); + }); + + it('an outbox whose ack() reads (id, result) only keeps working: the dispatcher still records its attempts', async () => { + // The shape of an outbox written before the credential existed: the same + // store underneath, reached through an `ack` that reads two parameters. + const acked: string[] = []; + const legacy: IHttpOutbox = { + enqueue: (input) => outbox.enqueue(input), + recordUndeliverable: (input) => outbox.recordUndeliverable(input), + reap: (opts) => outbox.reap(opts), + claim: (opts) => outbox.claim(opts), + ack: (id, result) => { acked.push(id); return outbox.ack(id, result); }, + list: (filter) => outbox.list(filter), + redeliver: (id, options) => outbox.redeliver(id, options), + }; + const id = await outbox.enqueue(DELIVERY); + const fetchImpl: FetchImpl = async () => respond(200); + + await new HttpDispatcher({ + nodeId: 'node-a', outbox: legacy, fetchImpl, partitionCount: 1, claimTtlMs: TTL, intervalMs: 10_000, + now: () => T0, + }).tick(); + + expect(acked).toEqual([id]); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:200'); + }); +}); + +describe.each([memoryBackend(), sqlBackend()])('$name — ack() with a claim credential proves the claim (#17634)', (backend) => { + let outbox: BuiltInHttpOutbox; + + beforeEach(async () => { outbox = await backend.create(); }); + afterEach(async () => { await backend.destroy(); }); + + async function readRow(id: string): Promise { + const row = (await outbox.list()).find((r) => r.id === id); + if (!row) throw new Error(`row '${id}' vanished — the harness, not the contract, is broken`); + return row; + } + + it('replays the card: a late ack from a reaped claim is refused, and touches NOTHING', async () => { + const id = await outbox.enqueue(DELIVERY); + + // 1. node A claims R and "starts a send". + const byA = await outbox.claim(claimOpts('node-a', T0)); + expect(byA.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 — `in_flight` AGAIN, the + // state a status-only predicate cannot tell from step 1. + const byB = await outbox.claim(claimOpts('node-b', T_AFTER_TTL)); + expect(byB.map((r) => `${r.id}:${r.claimedBy}:${r.claimedAt}`)).toEqual([`${id}:node-b:${T_AFTER_TTL}`]); + + // 4. node A finishes and acks with the credential its own claim returned. + await expect( + outbox.ack(id, { success: false, httpStatus: 410, error: 'gone', durationMs: TTL + 5, dead: true }, credentialOf(byA[0])), + ).rejects.toMatchObject(REFUSAL); + // What the refusal did NOT do: the row still belongs to B's claim. + 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 — lands. + await expect( + outbox.ack(id, { success: true, httpStatus: 200, durationMs: 3 }, credentialOf(byB[0])), + ).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:200'); + }); + + it('negative control: the SAME sequence without the reap still acks', async () => { + const id = await outbox.enqueue(DELIVERY); + const byA = await outbox.claim(claimOpts('node-a', T0)); + expect(byA.map((r) => r.id)).toEqual([id]); + + // 2'. Slow, but INSIDE the visibility timeout: node B's claim() reaps + // nothing and takes nothing — proven, not assumed. + expect(await outbox.claim(claimOpts('node-b', T_WITHIN_TTL))).toEqual([]); + expect(fingerprint(await readRow(id))).toBe(`in_flight:node-a:${T0}:0:-`); + + // 4'. A's ack with its own credential MUST land — a predicate that + // refused every ack fails here. + await expect( + outbox.ack(id, { success: false, httpStatus: 410, error: 'gone', durationMs: 5, dead: true }, credentialOf(byA[0])), + ).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('dead:-:-:1:410'); + }); + + 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 outbox.enqueue(DELIVERY); + const stale = await outbox.claim(claimOpts('node-a', T0)); + expect(stale.map((r) => r.id)).toEqual([id]); + + // A ITSELF reaps and re-claims on a later tick. Same node id — a + // claimed_by-only predicate would match. + const fresh = await outbox.claim(claimOpts('node-a', T_AFTER_TTL)); + expect(fresh.map((r) => `${r.claimedBy}:${r.claimedAt}`)).toEqual([`node-a:${T_AFTER_TTL}`]); + + // The FIRST attempt's late ack is refused — `claimedAt` is what tells two + // claims by one node apart. + await expect( + outbox.ack( + id, + { success: false, httpStatus: 503, error: 'unavailable', durationMs: 1, nextRetryAt: T_AFTER_TTL + 1_000 }, + credentialOf(stale[0]), + ), + ).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(id, { success: true, httpStatus: 200, durationMs: 1 }, credentialOf(fresh[0])), + ).resolves.toBeUndefined(); + expect(fingerprint(await readRow(id))).toBe('success:-:-:1:200'); + }); + + it('a claim reaped and NOT re-claimed: the late ack is refused, and the row stays queued with no attempt recorded', async () => { + const id = await outbox.enqueue(DELIVERY); + const byA = await outbox.claim(claimOpts('node-a', T0)); + await outbox.reap({ claimTtlMs: TTL, now: T_AFTER_TTL }); + expect(fingerprint(await readRow(id))).toBe('pending:-:-:0:-'); + + // A retry outcome, deliberately: its post-state would ALSO read + // `pending`, so only the attempt counter could expose a write that landed. + await expect( + outbox.ack( + id, + { success: false, httpStatus: 503, error: 'unavailable', durationMs: 1, nextRetryAt: T_AFTER_TTL + 1_000 }, + credentialOf(byA[0]), + ), + ).rejects.toMatchObject(REFUSAL); + expect(fingerprint(await readRow(id))).toBe('pending:-:-:0:-'); + }); + + it('a credential missing a member is refused before anything is written', async () => { + const id = await outbox.enqueue(DELIVERY); + await outbox.claim(claimOpts('node-a', T0)); + + // A JS caller or a cast: `claimedAt` absent. + const partial = { claimedBy: 'node-a' } as unknown as HttpClaimCredential; + await expect( + outbox.ack(id, { success: true, httpStatus: 200, durationMs: 1 }, partial), + ).rejects.toMatchObject(REFUSAL); + expect(fingerprint(await readRow(id))).toBe(`in_flight:node-a:${T0}:0:-`); + }); +}); diff --git a/packages/services/service-messaging/src/http-outbox.ts b/packages/services/service-messaging/src/http-outbox.ts index cc98ae84bf..31184c15d8 100644 --- a/packages/services/service-messaging/src/http-outbox.ts +++ b/packages/services/service-messaging/src/http-outbox.ts @@ -120,7 +120,8 @@ export interface HttpDelivery { * Number of attempts made so far (0 before first attempt). * * [#8069] **Load-bearing beyond diagnostics.** `ack()` — the only writer of - * a terminal status — increments this unconditionally, so a row that is + * a terminal status — increments this with every outcome it records + * (#17634: never for a claim it no longer holds), so a row that is * terminal (`success` / `failed` / `dead`) with `attempts === 0` never came * from the dispatch path: it was **parked** by * {@link IHttpOutbox.recordUndeliverable} and has never existed on the @@ -357,6 +358,100 @@ export interface HttpAckFailure { export type HttpAckResult = HttpAckSuccess | HttpAckFailure; +/** + * [#17634] The claim credential — the (`claimedBy`, `claimedAt`) pair a store + * stamps on a row when {@link IHttpOutbox.claim} takes it — handed back to + * {@link IHttpOutbox.ack} so the outcome is written only while that claim still + * holds the row. It is exactly the pair the notification outbox's + * `ClaimedDeliveryRecord` guarantees for `INotificationOutbox.ack` (#11859), with + * the same meaning: + * + * - ownership is proven by ROUND-TRIPPING what `claim()` returned, never by the + * caller supplying an identity it had to know — hand back the claimed row + * itself, which is structurally one; + * - 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. + * + * A credential rather than a claimed-record type because `claim()` keeps + * declaring `HttpDelivery[]`, whose `claimedBy` / `claimedAt` are optional: + * narrowing that declared return type would stop every subclass override of a + * built-in store's `claim()` that declares the old one from compiling. + */ +export interface HttpClaimCredential { + /** Node id the claim stamped — {@link HttpDelivery.claimedBy} on the claimed row. */ + claimedBy: string; + /** Claim instant (ms) the claim stamped — {@link HttpDelivery.claimedAt} on the claimed row. */ + claimedAt: number; +} + +/** + * [#17634] Error raised by {@link IHttpOutbox.ack} when it is handed a claim + * credential and the row is no longer held by that claim — or was never + * claimed — so the outcome was NOT recorded. + * + * `DELIVERY_NOT_ELIGIBLE` is this package's registered ADR-0112 code for "this + * delivery row's state does not permit the requested operation", already raised + * by {@link HttpRedeliverError} and by the notification outbox's + * `NotificationAckError`. Reused, not minted: one concept, one spelling for a + * caller to match on. + */ +export class HttpAckError extends Error { + constructor( + message: string, + readonly code: 'DELIVERY_NOT_ELIGIBLE', + ) { + super(message); + this.name = 'HttpAckError'; + } +} + +/** + * [#17634] Refuse a claim credential that is not one — the runtime half of + * {@link HttpClaimCredential}, for JS callers and casts. In ONE place both + * stores call, before any read or write. + */ +export function assertHttpClaimCredential(id: string, claimed: HttpClaimCredential): void { + if (typeof claimed?.claimedBy !== 'string' || typeof claimed?.claimedAt !== 'number') { + throw new HttpAckError( + `Delivery row '${id}': the claim credential passed to ack() is incomplete (claimedBy + claimedAt). ` + + 'ack() proves the claim it completes by handing back the pair claim() stamped on the row. ' + + 'Nothing was written.', + 'DELIVERY_NOT_ELIGIBLE', + ); + } +} + +/** + * [#17634] The refusal message for a row that is not `in_flight` when a + * credentialed ack runs — reaped back to `pending` and not re-claimed yet, + * already terminal, or never claimed. In ONE place both stores call, so the two + * cannot drift into two wordings for one refusal. + */ +export function httpAckNotClaimedMessage(id: string, status: HttpDeliveryStatus | 'unknown'): string { + return ( + `Delivery row '${id}' is '${status}', not 'in_flight': ack() was handed a claim credential and ` + + 'this row is not claimed, so NOTHING was written. Expected when a slow send outruns `claimTtlMs` ' + + 'and the visibility-timeout reap returns the row to the queue: whoever claims it next sends it ' + + 'again.' + ); +} + +/** + * [#17634] The refusal message for a row that IS `in_flight`, but under a + * different claim — reaped and re-claimed while the send ran, by another node + * or by this node's own later claim — or that changed between the ownership + * read and the conditional write. In ONE place both stores call. + */ +export function httpAckLostClaimMessage(id: string, status: HttpDeliveryStatus | 'unknown'): string { + return ( + `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.' + ); +} + /** * Error raised by `IHttpOutbox.redeliver` when the requested row is either * missing or in a non-terminal state. @@ -377,8 +472,8 @@ export class HttpRedeliverError extends Error { * ## What it refuses, and why that predicate * `redeliver` means *send this again*. A terminal row with `attempts === 0` was * never sent a first time — {@link IHttpOutbox.ack}, the only writer of a - * terminal status, increments `attempts` unconditionally, so the pair - * (terminal, 0 attempts) is reachable only through + * terminal status, increments `attempts` with every outcome it records, so + * the pair (terminal, 0 attempts) is reachable only through * {@link IHttpOutbox.recordUndeliverable}. Such a row is a **record of a * delivery that was refused before it ever existed on the wire**, and resetting * it to `pending` is not a replay: it is a FIRST delivery conjured by an @@ -540,11 +635,48 @@ export interface IHttpOutbox { * authored header. An implementation whose storage redacts the column * recovers it through a privileged read (see `SqlHttpOutbox`) or fails the * claim loudly; it must not return the row with the map silently absent. + * + * [#17634] Every returned row carries its claim credential — + * `claimedBy` = `opts.nodeId`, `claimedAt` = the claim instant — exactly as + * the claiming write stamped it. {@link ack} takes that pair back to prove + * the claim it completes, so an implementation that honours `ack`'s + * `claimed` argument MUST stamp it on every row. */ claim(opts: HttpClaimOptions): Promise; - /** Record the outcome of an attempt. */ - ack(id: string, result: HttpAckResult): Promise; + /** + * Record the outcome of ONE dispatch attempt. + * + * [#17634] **Pass `claimed`** — the claim credential on the row {@link claim} + * returned (the row itself will do). With it, `ack` is the ownership-checked + * completion `INotificationOutbox.ack` performs (#11453, #11859): + * + * ⛔ **Precondition: the row MUST still be held by that claim.** Two tests, + * both re-stated IN the conditional write: the row is `in_flight`, AND its + * (`claimed_by`, `claimed_at`) pair equals `claimed`. Otherwise `ack` throws + * {@link HttpAckError} (`DELIVERY_NOT_ELIGIBLE`), writes nothing and leaves + * `attempts` untouched. That covers the late ack whose claim the + * visibility-timeout reap took back while the send ran — re-claimed by + * another node that is still sending, or by this node's own later claim + * (the credential is the PAIR, so `claimedAt` tells two claims by one node + * apart) — and a claim reaped and not yet re-claimed. A `claimed` missing + * either member is refused the same way, before any read. An `id` matching + * no row stays a silent no-op: no state to corrupt, no claim to lose. + * Implementations MUST re-read what they need (e.g. `attempts`) from the + * store, and MUST make the ownership test and the write one atomic + * operation, never a read-then-write. + * + * **Without `claimed`** — the arity this method had before #17634 — both + * built-in stores write by id with no ownership check, exactly as they always + * did, so a late ack overwrites whatever holds the row now. That arity is + * deprecated, kept only so callers written against it keep working; + * `HttpDispatcher` always passes `claimed`. The parameter is optional — + * rather than a new method or a required argument — so an `IHttpOutbox` + * implementation written before it keeps compiling and working unchanged (it + * simply ignores the dispatcher's extra argument), and so a subclass that + * overrides a built-in store's `ack()` keeps seeing every dispatcher ack. + */ + ack(id: string, result: HttpAckResult, claimed?: HttpClaimCredential): Promise; /** * Snapshot accessor for tests / admin tooling. [#8118] Not a dispatch diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 40fac79407..07ea1a05ab 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -139,12 +139,17 @@ export type { HttpAckResult, HttpAckSuccess, HttpAckFailure, + // [#17634] The (claimedBy, claimedAt) pair `claim()` stamps and `ack()` + // takes back — the credential its compare-and-set binds. + HttpClaimCredential, UndeliverableHttpInput, RedeliverGuard, RedeliverOptions, } from './http-outbox.js'; export { HttpRedeliverError, + // [#17634] `ack()` refuses a claim it no longer holds with this. + HttpAckError, assertHttpRedeliverable, assertRedeliverAllowed, assertEnqueueDeliverable, diff --git a/packages/services/service-messaging/src/memory-http-outbox.ts b/packages/services/service-messaging/src/memory-http-outbox.ts index 69c7fb6762..b7ed88f5f1 100644 --- a/packages/services/service-messaging/src/memory-http-outbox.ts +++ b/packages/services/service-messaging/src/memory-http-outbox.ts @@ -4,11 +4,16 @@ import { randomUUID } from 'node:crypto'; import { hashPartition } from './backoff.js'; import { deliveryBody, signBody } from './http-sender.js'; import { + HttpAckError, HttpRedeliverError, assertEnqueueDeliverable, + assertHttpClaimCredential, assertRedeliverAllowed, + httpAckLostClaimMessage, + httpAckNotClaimedMessage, type EnqueueHttpInput, type HttpAckResult, + type HttpClaimCredential, type HttpClaimOptions, type HttpDelivery, type HttpDeliveryStatus, @@ -135,9 +140,29 @@ export class MemoryHttpOutbox implements IHttpOutbox { } } - async ack(id: string, result: HttpAckResult): Promise { + /** + * Record one attempt's outcome — see {@link IHttpOutbox.ack}. + * + * [#17634] Handed `claimed`, the row must still be held by that claim — + * `in_flight`, under the same (`claimedBy`, `claimedAt`) pair — or the ack + * is refused with {@link HttpAckError} and nothing is written, so a late ack + * from a reaped claim cannot land on the live re-claim. Single-threaded, so + * each test and the mutation below are one atomic step; `SqlHttpOutbox` + * spells the same guard as a conditional UPDATE because it is not. Without + * `claimed` — the deprecated arity — this is the by-id write it always was. + */ + async ack(id: string, result: HttpAckResult, claimed?: HttpClaimCredential): Promise { + if (claimed !== undefined) assertHttpClaimCredential(id, claimed); const row = this.rows.get(id); if (!row) return; + if (claimed !== undefined) { + if (row.status !== 'in_flight') { + throw new HttpAckError(httpAckNotClaimedMessage(id, row.status), 'DELIVERY_NOT_ELIGIBLE'); + } + if (row.claimedBy !== claimed.claimedBy || row.claimedAt !== claimed.claimedAt) { + throw new HttpAckError(httpAckLostClaimMessage(id, row.status), 'DELIVERY_NOT_ELIGIBLE'); + } + } const now = Date.now(); row.attempts += 1; row.lastAttemptedAt = now; diff --git a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts index 4ef650e7dc..2a8646124a 100644 --- a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -50,9 +50,11 @@ 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 `SqlHttpOutbox.ack` - * and {@link dispatcherAckCasOptions} carries it to `SqlNotificationOutbox.ack` - * (a `multi: true` compare-and-set since #11453), while `SqlHttpOutbox.redeliver` + * {@link dispatcherAckOptions} carries the sweep warrant to the deprecated + * credential-less arity of `SqlHttpOutbox.ack`, and + * {@link dispatcherAckCasOptions} carries it to `SqlNotificationOutbox.ack` (a + * `multi: true` compare-and-set since #11453) and to `SqlHttpOutbox.ack` handed + * a claim credential (since #17634), 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- @@ -81,6 +83,12 @@ export function dispatcherSweepOptions( * grew a status precondition, and a precondition on the by-id path is silently * discarded (#11009), so it rides {@link dispatcherAckCasOptions} instead. * + * [#17634] Nor does `SqlHttpOutbox.ack` when it is handed a claim credential — + * which `HttpDispatcher` always does: its ownership test is a compare-and-set + * too, so it rides {@link dispatcherAckCasOptions}. What is left here is that + * method's deprecated credential-less arity, a by-id write kept unchanged for + * callers written against it. + * * ## Why a second helper instead of {@link dispatcherSweepOptions} * These are audited under the driver's **`update`** op, not `updateMany`, and * the two ops are separate keys in `auditMissingTenant`'s throttle — so a @@ -131,9 +139,10 @@ export function dispatcherAckOptions( /** - * [#11453] The write options for **`SqlNotificationOutbox.ack`** — the same + * [#11453] The write options for **`SqlNotificationOutbox.ack`** — and, since + * #17634, for **`SqlHttpOutbox.ack`** handed a claim credential — the same * warrant as {@link dispatcherAckOptions} above, spelled as a PREDICATE write - * because that ack is now a compare-and-set. + * because each of those acks is a compare-and-set. * * ## Why `multi: true` for a write that still targets ONE row * @@ -160,6 +169,14 @@ 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. * + * [#17634] The HTTP site, re-derived rather than inherited: the credentialed + * `SqlHttpOutbox.ack` has one caller, `HttpDispatcher.ackAttempt`, reached only + * from `runPartition()` — a timer tick under the `http.dispatcher.partition.` + * cluster lock, with no HTTP request, session or active organization to thread — + * and the row it acks was claimed by the same environment-wide sweep. + * `redeliver` stays the one request-reachable write on `sys_http_delivery`, and + * it threads the caller's tenant. + * * ## [#11859] Ownership joined the predicate * * `status = 'in_flight'` can prove a claim EXISTS but not WHOSE: after a diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index 0f5f6bec13..ba8f3c188c 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -4,14 +4,19 @@ import { randomUUID } from 'node:crypto'; import type { IDataEngine } from '@objectstack/spec/contracts'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; -import { dispatcherAckOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; +import { dispatcherAckCasOptions, dispatcherAckOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; import { deliveryBody, signBody } from './http-sender.js'; import { + HttpAckError, HttpRedeliverError, assertEnqueueDeliverable, + assertHttpClaimCredential, assertRedeliverAllowed, + httpAckLostClaimMessage, + httpAckNotClaimedMessage, type EnqueueHttpInput, type HttpAckResult, + type HttpClaimCredential, type HttpClaimOptions, type HttpDelivery, type HttpDeliveryStatus, @@ -248,7 +253,11 @@ export class SqlHttpOutbox implements IHttpOutbox { // claim now owns — the one read that must see the authored map. const headerColumns = await this.readClaimedHeaderColumns(claimed.map((r) => r.id)); - return claimed.map((r) => this.toDelivery(r, headerColumns)); + // [#17634] The credential `ack()` takes back, stated explicitly: the + // read-back WHERE just proved (claimed_by, claimed_at) = (nodeId, now), + // so this restates what the query established — in the exact values the + // claiming UPDATE wrote, whatever form a dialect reads the column back in. + return claimed.map((r) => ({ ...this.toDelivery(r, headerColumns), claimedBy: opts.nodeId, claimedAt: now })); } /** @@ -321,48 +330,91 @@ export class SqlHttpOutbox implements IHttpOutbox { return engine.resolveInternalField(this.objectName, ids, 'headers_json'); } - async ack(id: string, result: HttpAckResult): Promise { + /** + * Record one attempt's outcome — see {@link IHttpOutbox.ack}. + * + * [#17634] Handed `claimed` — as `HttpDispatcher` always hands it — this is + * the compare-and-set `SqlNotificationOutbox.ack` performs (#11453, #11859): + * two deterministic refusals read before any write, the same two tests + * re-stated IN a conditional UPDATE (the half that holds under the race), + * and a read-back that reports a write which matched nothing instead of a + * silent success. Without `claimed` — the deprecated arity — it is the by-id + * write it always was ({@link ackById}). + */ + async ack(id: string, result: HttpAckResult, claimed?: HttpClaimCredential): Promise { + if (claimed === undefined) return this.ackById(id, result); + // The runtime half of the credential contract, for JS callers and + // casts: refused before any IO. + assertHttpClaimCredential(id, claimed); const current = (await this.engine.findOne(this.objectName, { where: { id }, - fields: ['attempts'], - })) as { attempts?: number } | null; + fields: ['status', 'attempts', 'claimed_by', 'claimed_at'], + })) as Pick | null; + // An id matching no row: no state to corrupt, no claim to lose. if (!current) return; + // Not claimed at all — reaped back to the queue, already terminal, or + // never claimed. Refused BEFORE any write: the row is left byte-identical. + if (current.status !== 'in_flight') { + throw new HttpAckError(httpAckNotClaimedMessage(id, current.status ?? 'unknown'), 'DELIVERY_NOT_ELIGIBLE'); + } + // Ownership, read half: claimed, but not by the claim being completed — + // reaped and re-claimed while the send ran. `status = 'in_flight'` alone + // matches the re-claiming node's live attempt, which is exactly the + // overwrite #17634 measured. The SAME test is re-stated in the 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 HttpAckError(httpAckLostClaimMessage(id, current.status), 'DELIVERY_NOT_ELIGIBLE'); + } - const now = Date.now(); - let status: HttpDeliveryStatus; - let nextRetryAt: number | null; - let error: string | null; - - if (result.success) { - status = 'success'; - nextRetryAt = null; - error = null; - } else if (result.dead) { - status = 'dead'; - nextRetryAt = null; - error = result.error ?? null; - } else { - status = 'pending'; - nextRetryAt = result.nextRetryAt ?? null; - error = result.error ?? null; + // Ownership, write half: the row transitions only if it is STILL + // `in_flight` AND still held by THIS claim. A row reaped and re-claimed + // between the read above and here matches nothing and is left alone, + // whoever re-claimed it. `attempts` moves only inside that condition, so + // it counts real dispatch attempts and nothing else. + const attempts = (current.attempts ?? 0) + 1; + const patch = attemptPatch(result, attempts); + await this.engine.update( + this.objectName, + patch, + // Predicate write (`updateMany`): on the by-id path every predicate + // but the id is silently discarded (#11009). 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', claimed.claimedBy, claimed.claimedAt), + ); + + // Did the conditional write land? `IDataEngine.update` declares its + // return as `any`, so the row itself is the only contract-safe answer. + // The detector is the pair (status, attempts), not status alone: a retry + // ack's post-state IS `pending`, the status a reaped row already has, so + // only the recorded attempt tells the two apart. + const after = (await this.engine.findOne(this.objectName, { + where: { id }, + fields: ['status', 'attempts'], + })) as Pick | null; + if (!after || after.status !== patch.status || (after.attempts ?? 0) !== attempts) { + throw new HttpAckError(httpAckLostClaimMessage(id, after?.status ?? 'unknown'), 'DELIVERY_NOT_ELIGIBLE'); } + } + + /** + * The deprecated credential-less ack (#17634): a by-id write with no + * ownership check, unchanged — kept so a caller written against the + * two-argument `ack()` keeps working. `HttpDispatcher` never takes this path. + */ + private async ackById(id: string, result: HttpAckResult): Promise { + const current = (await this.engine.findOne(this.objectName, { + where: { id }, + fields: ['attempts'], + })) as { attempts?: number } | null; + if (!current) return; await this.engine.update( this.objectName, - { - status, - attempts: (current.attempts ?? 0) + 1, - last_attempted_at: now, - claimed_by: null, - claimed_at: null, - response_code: result.httpStatus ?? null, - response_body: result.responseBody ?? null, - next_retry_at: nextRetryAt, - 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`. + attemptPatch(result, (current.attempts ?? 0) + 1), + // Single-record write, audited under the `update` op. Declared a + // global-sweep site — no request context reaches it. Warrant in + // `outbox-dispatcher-scope.ts`. dispatcherAckOptions(id), ); } @@ -485,3 +537,41 @@ export class SqlHttpOutbox implements IHttpOutbox { }; } } + +/** + * The row patch that records one attempt's outcome — shared by both arities of + * {@link SqlHttpOutbox.ack}, so the conditional write and the by-id write cannot + * drift into two readings of one {@link HttpAckResult}. No `updated_at` (#4765 — + * see the class docs). + */ +function attemptPatch(result: HttpAckResult, attempts: number) { + let status: HttpDeliveryStatus; + let nextRetryAt: number | null; + let error: string | null; + + if (result.success) { + status = 'success'; + nextRetryAt = null; + error = null; + } else if (result.dead) { + status = 'dead'; + nextRetryAt = null; + error = result.error ?? null; + } else { + status = 'pending'; + nextRetryAt = result.nextRetryAt ?? null; + error = result.error ?? null; + } + + return { + status, + attempts, + last_attempted_at: Date.now(), + claimed_by: null, + claimed_at: null, + response_code: result.httpStatus ?? null, + response_body: result.responseBody ?? null, + next_retry_at: nextRetryAt, + error, + }; +}