From e2af29fd9d1fbb4cb07a1ddcc09c62eb71b6b251 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:19:07 +0000 Subject: [PATCH 1/5] fix(service-messaging): read the outbox claim credential back from the store claim()/claimDigest() on SqlNotificationOutbox and SqlHttpOutbox now hand out claimedAt as the claimed_at value the store returned after the claiming UPDATE, not the in-memory now, and identify the claimed rows by exact predicates only (candidate id, holder, status). On a Postgres table whose epoch-ms columns are still `real`, the stored instant is a float32 rounding of now, so the old credential never equalled the stored value: every ack() was refused as a lost claim and the row was re-sent after each reap. Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../service-messaging/src/http-outbox.ts | 8 +-- .../src/outbox-dispatcher-scope.ts | 4 +- .../service-messaging/src/sql-http-outbox.ts | 35 +++++++++--- .../service-messaging/src/sql-outbox.ts | 54 +++++++++++++++---- 4 files changed, 78 insertions(+), 23 deletions(-) diff --git a/packages/services/service-messaging/src/http-outbox.ts b/packages/services/service-messaging/src/http-outbox.ts index 29c08310ce1..32c0d01ac5e 100644 --- a/packages/services/service-messaging/src/http-outbox.ts +++ b/packages/services/service-messaging/src/http-outbox.ts @@ -640,9 +640,11 @@ export interface IHttpOutbox { * * [#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. + * the store holds it after the claiming write: a store whose numeric column + * does not round-trip epoch-ms exactly hands back the value it read, not the + * `now` it was given. {@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; diff --git a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts index 53764aa1c6d..0d3ca1b067d 100644 --- a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -278,7 +278,9 @@ export function dispatcherAckOptions( * @param id Primary key of the delivery row this attempt outcome belongs to. * @param expectedStatus The status the row MUST still hold for the write to land. * @param claimedBy The node id stamped by the claim this ack completes. - * @param claimedAt The claim instant (ms) stamped by that same claim. + * @param claimedAt The claim instant (ms) stamped by that same claim, as the + * store read it back — bound here, it matches the stored value whatever + * numeric type holds the column. */ export function dispatcherAckCasOptions( id: string, diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index 7fcc7954d33..ceb42436327 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -254,20 +254,36 @@ export class SqlHttpOutbox implements IHttpOutbox { { ...dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), context: DISPATCHER_SYSTEM_CONTEXT }, ); - // 4. Read back the rows we actually own. + // 4. Read back the rows we actually own. `claimed_at` is deliberately + // not in this WHERE — the same reading as + // `SqlNotificationOutbox.readBackClaimed`: an equality between the + // column and the client's `now` holds only when the store coerces the + // comparand to the column's own type. Every candidate was `pending` + // at step 2, and a node runs one claim at a time, so a candidate now + // `in_flight` under `nodeId` was taken by the UPDATE above. const claimed = (await this.engine.find(this.objectName, { - where: { id: { $in: ids }, claimed_by: opts.nodeId, claimed_at: now, status: 'in_flight' }, + where: { id: { $in: ids }, claimed_by: opts.nodeId, status: 'in_flight' }, }, { context: DISPATCHER_SYSTEM_CONTEXT })) as DeliveryRow[]; // 5. [#8118] Recover the redacted header column for the rows this // claim now owns — the one read that must see the authored map. const headerColumns = await this.readClaimedHeaderColumns(claimed.map((r) => r.id)); - // [#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 })); + // [#17634] The credential `ack()` takes back: the holder the WHERE just + // proved, and the claim instant AS THE STORE HOLDS IT — the `claimed_at` + // read back from the row, never the `now` this process wrote. A numeric + // column need not round-trip epoch-ms exactly (a Postgres `real` column + // left by the old `number` mapping stores a float32 rounding of it), and + // a credential holding `now` would then never equal the stored value: + // every ack refused, the row re-sent after each reap. The read-back value + // equals every later read of the cell and binds back to the same stored + // value in the ack's compare-and-set. `claimed_at` is non-null on an + // `in_flight` row: every write that sets `in_flight` sets it. + return claimed.map((r) => ({ + ...this.toDelivery(r, headerColumns), + claimedBy: opts.nodeId, + claimedAt: r.claimed_at as number, + })); } /** @@ -376,7 +392,10 @@ export class SqlHttpOutbox implements IHttpOutbox { // 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. + // below, which is the half that actually holds under the race. Both + // sides of the `claimed_at` test are reads of the same cell — + // `claimedAt` is what `claim()` read back — so it is exact on any + // numeric column, a lossy `real` one included. if (current.claimed_by !== claimed.claimedBy || current.claimed_at !== claimed.claimedAt) { throw new HttpAckError(httpAckLostClaimMessage(id, current.status), 'DELIVERY_NOT_ELIGIBLE'); } diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index 60f597aa658..8821ed0806e 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -181,14 +181,8 @@ export class SqlNotificationOutbox implements INotificationOutbox { { ...dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), context: DISPATCHER_SYSTEM_CONTEXT }, ); - // 4. Read back only the rows we own. [commit d9cf78eaa] The read-back WHERE just - // proved (claimed_by, claimed_at) — the claim credential — so the - // explicit stamp below narrows to ClaimedDeliveryRecord without a - // cast, and states nothing the query did not already establish. - const claimed = (await this.engine.find(this.objectName, { - where: { id: { $in: ids }, claimed_by: opts.nodeId, claimed_at: now, status: 'in_flight' }, - }, { context: DISPATCHER_SYSTEM_CONTEXT })) as DeliveryRow[]; - return claimed.map((r) => ({ ...this.toRecord(r), claimedBy: opts.nodeId, claimedAt: now })); + // 4. Read back only the rows we own, with the credential as stored. + return this.readBackClaimed(ids, opts.nodeId); } async claimDigest(opts: ClaimOptions): Promise { @@ -222,11 +216,46 @@ export class SqlNotificationOutbox implements INotificationOutbox { { ...dispatcherSweepOptions({ id: { $in: ids }, status: 'pending' }), context: DISPATCHER_SYSTEM_CONTEXT }, ); - // 4. Read back the rows we own — same credential stamp as claim(). + // 4. Read back the rows we own — same credential as claim(). + return this.readBackClaimed(ids, opts.nodeId); + } + + /** + * Step 4 of {@link claim} / {@link claimDigest}: the rows the claiming + * UPDATE just took, each carrying its claim credential **as the store + * holds it**. + * + * `claimedAt` is the `claimed_at` value read back from the row, never the + * `now` this process wrote. A numeric column does not have to round-trip + * an epoch-ms value exactly: a Postgres table created while the `number` + * arm still mapped to `real` (IEEE-754 binary32) stores `now` rounded to a + * float32 step (131 072 ms at today's magnitude) and reads back a nearby + * decimal, so a credential holding `now` never equals the stored value, + * every `ack()` is refused as a lost claim, and the row is re-sent after + * each reap. The value the store returned compares equal to every later + * read of the same cell, and binds back to the same stored value in the + * compare-and-set, whatever numeric type holds it. + * + * That is also why `claimed_at` is not in the WHERE: an equality between + * the column and the client's `now` holds only when the store coerces the + * comparand to the column's own type. The rows are identified by exact + * predicates instead — candidate id, holder, status. Every candidate was + * `pending` when step 2 read it, so a candidate now `in_flight` under + * `nodeId` was taken by this call's UPDATE: a node runs one claim at a + * time (the dispatcher's ticks never overlap, and `claim` / `claimDigest` + * split the rows on `digest_key`). And the credential read here still + * tells this claim from any earlier one on the same row: a re-claim needs + * a reap first, the reap needs the stored instant to be older than + * `now - claimTtlMs` on the same column, and the re-claim then stores a + * later instant. + */ + private async readBackClaimed(ids: string[], nodeId: string): Promise { const claimed = (await this.engine.find(this.objectName, { - where: { id: { $in: ids }, claimed_by: opts.nodeId, claimed_at: now, status: 'in_flight' }, + where: { id: { $in: ids }, claimed_by: nodeId, status: 'in_flight' }, }, { context: DISPATCHER_SYSTEM_CONTEXT })) as DeliveryRow[]; - return claimed.map((r) => ({ ...this.toRecord(r), claimedBy: opts.nodeId, claimedAt: now })); + // `claimed_at` is non-null on an `in_flight` row: every write that sets + // `in_flight` sets it, and every write that clears it leaves `in_flight`. + return claimed.map((r) => ({ ...this.toRecord(r), claimedBy: nodeId, claimedAt: r.claimed_at as number })); } async ack(claimed: ClaimedDeliveryRecord, result: AckResult): Promise { @@ -266,6 +295,9 @@ export class SqlNotificationOutbox implements INotificationOutbox { // exactly the overwrite the card measured). Deterministic refusal // before any write; the SAME test is re-stated in the conditional // write below, which is the half that actually holds under the race. + // Both sides of the `claimed_at` test are reads of the same cell — + // `claimedAt` is the value `claim()` read back (`readBackClaimed`) — + // so it is exact on any numeric column, a lossy `real` one included. if (current.claimed_by !== claimed.claimedBy || current.claimed_at !== claimed.claimedAt) { throw new NotificationAckError( notificationAckLostClaimMessage(id, current.status ?? 'unknown'), From 1c39df18d64d08304df8100987ad658dc20df8b1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:24:02 +0000 Subject: [PATCH 2/5] test(service-messaging): pin the claim credential on a lossy real column A real engine behind a double that gives the epoch-ms columns Postgres real semantics (binary32 store, comparand coercion, shortest-decimal read-back, pinned to three live PostgreSQL 16.15 readings). Both outboxes: one send and one ack reach success, a claim lost to another node still refuses the late ack, and a row stuck in_flight behind an ack that could never land is sent once more and acked with no operator act. Exact-column and memory stores are the controls. Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- ...redential-lossy-column.integration.test.ts | 454 ++++++++++++++++++ 1 file changed, 454 insertions(+) create mode 100644 packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts diff --git a/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts new file mode 100644 index 00000000000..1bda2d23ee3 --- /dev/null +++ b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts @@ -0,0 +1,454 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The claim credential survives the column that holds it. + * + * A Postgres table created while the `number` arm still mapped to `real` + * keeps its epoch-ms columns as IEEE-754 binary32 — new tables get an exact + * numeric column, existing ones are not migrated. At today's magnitude a + * binary32 step is 131 072 ms, so `claimed_at` stores a rounding of the + * claim's `now` and reads back a nearby decimal that is neither. A credential + * holding the in-memory `now` then never equals the stored value: every + * `ack()` is refused as a lost claim, the dispatcher absorbs the refusal, the + * row stays `in_flight` with `attempts = 0`, and each reap hands it to the + * next claim — one more send every 131 s, forever. + * + * ## The double, and what it models + * + * {@link realColumnEngine} wraps a REAL engine (`ObjectQL` + `SqlDriver`, + * better-sqlite3 `:memory:`) and gives the epoch-ms columns the three + * behaviours a Postgres `real` column has, each measured on PostgreSQL 16.15 + * through `SqlDriver` + node-postgres: + * + * - a written number is STORED as its binary32 rounding (`Math.fround`); + * - a comparand in a predicate is coerced to the column's type before it is + * compared — node-postgres sends parameters untyped, so `claimed_at = $1` + * and `claimed_at < $1` compare binary32 with binary32; + * - a READ hands back the shortest decimal that round-trips to the stored + * binary32 (`float4out`, `extra_float_digits = 1`, the default since + * PostgreSQL 12), parsed as a double. The fidelity leg pins three readings + * the live server returned. + * + * Every write still goes through the real engine, so the compare-and-set in + * `ack()` is evaluated by a real conditional UPDATE, never by the double. + * + * ## The legs + * + * - **One send, one ack** on the lossy column: the row reaches `success` and + * no later pass sends it again. The same leg on an exact column and on the + * memory store is the control — the fix must not change what already + * worked. + * - **A genuinely lost claim is still refused.** Another node reaping and + * re-claiming the row while the first send runs must keep the late ack + * out (`DELIVERY_NOT_ELIGIBLE`), on every store — the fix may not turn a + * real lost claim into a success. + * - **The stuck row heals.** A row left `in_flight` by an ack that could never + * land (the credential claim() handed out before this fix) is reaped, + * re-claimed, sent once and acked by the next passes, with no operator act. + * + * Refusals assert the error identity — `name` + the ADR-0112 `code` — never a + * bare throw. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import type { IDataEngine } from '@objectstack/spec/contracts'; +import { MemoryNotificationOutbox } from './memory-outbox.js'; +import { MemoryHttpOutbox } from './memory-http-outbox.js'; +import { SqlNotificationOutbox } from './sql-outbox.js'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; +import { HttpDelivery } from './objects/http-delivery.object.js'; +import type { ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; +import type { HttpDelivery as HttpDeliveryRow, IHttpOutbox } from './http-outbox.js'; + +// ── The `real` column double ──────────────────────────────────────────────── + +/** The epoch-ms columns the two delivery objects declare as `number`. */ +const EPOCH_MS_COLUMNS: ReadonlySet = new Set([ + 'claimed_at', + 'next_attempt_at', + 'next_retry_at', + 'last_attempted_at', +]); + +/** What a binary32 column stores for a number written into it. */ +function storeReal(value: unknown): unknown { + return typeof value === 'number' ? Math.fround(value) : value; +} + +/** What Postgres hands back for a stored binary32: the shortest round-tripping decimal. */ +function readReal(value: unknown): unknown { + if (typeof value !== 'number') return value; + const stored = Math.fround(value); + for (let digits = 1; digits <= 9; digits++) { + const decimal = Number(stored.toPrecision(digits)); + if (Math.fround(decimal) === stored) return decimal; + } + return stored; +} + +function coerceOperand(operand: unknown): unknown { + if (typeof operand === 'number') return Math.fround(operand); + if (Array.isArray(operand)) return operand.map(coerceOperand); + if (operand && typeof operand === 'object') { + return Object.fromEntries(Object.entries(operand).map(([op, v]) => [op, coerceOperand(v)])); + } + return operand; +} + +function coerceComparands(where: unknown): unknown { + if (Array.isArray(where)) return where.map(coerceComparands); + if (!where || typeof where !== 'object') return where; + return Object.fromEntries(Object.entries(where).map(([key, value]) => [ + key, + EPOCH_MS_COLUMNS.has(key) ? coerceOperand(value) : key.startsWith('$') ? coerceComparands(value) : value, + ])); +} + +function mapEpochColumns(row: unknown, fn: (value: unknown) => unknown): unknown { + if (!row || typeof row !== 'object') return row; + return Object.fromEntries(Object.entries(row).map(([k, v]) => [k, EPOCH_MS_COLUMNS.has(k) ? fn(v) : v])); +} + +function withComparands(query: Q): Q { + const q = query as { where?: unknown } | undefined; + return (q && q.where !== undefined ? { ...q, where: coerceComparands(q.where) } : query) as Q; +} + +/** + * A real engine whose epoch-ms columns behave like a Postgres `real` column — + * see the file header. Reads and writes are forwarded to `engine`; only the + * values crossing the column boundary are changed. + */ +function realColumnEngine(engine: ObjectQL): IDataEngine { + return new Proxy(engine, { + get(target, prop, receiver) { + switch (prop) { + case 'find': + return async (object: string, query: unknown, options?: unknown) => { + const rows = await target.find(object, withComparands(query) as any, options as any); + return (rows as unknown[]).map((r) => mapEpochColumns(r, readReal)); + }; + case 'findOne': + return async (object: string, query: unknown, options?: unknown) => + mapEpochColumns(await target.findOne(object, withComparands(query) as any, options as any), readReal); + case 'insert': + return (object: string, data: unknown, options?: unknown) => + target.insert(object, mapEpochColumns(data, storeReal) as any, options as any); + case 'update': + return (object: string, data: unknown, options?: unknown) => + target.update(object, mapEpochColumns(data, storeReal) as any, withComparands(options) as any); + default: { + const value = Reflect.get(target, prop, receiver); + return typeof value === 'function' ? value.bind(target) : value; + } + } + }, + }) as unknown as IDataEngine; +} + +// ── Instants ──────────────────────────────────────────────────────────────── + +const TTL = 60_000; +/** One binary32 step at today's epoch-ms magnitude (2^40 ≤ t < 2^41 → 2^17 ms). */ +const STEP = 131_072; +/** The first claim: 2026-10-12T13:00:12.345Z — not a binary32 value. */ +const T0 = 1_791_810_012_345; +/** + * A later claim instant past the visibility timeout by more than one binary32 + * step. On a `real` column the reap compares binary32 with binary32, so it + * fires once `now - claimTtlMs` reaches the next step — up to one step late. + */ +const T_REAPED = T0 + TTL + 2 * STEP; +/** The k-th dispatcher pass after T0, each one reap window past the last. */ +const passAt = (k: number) => T0 + k * (TTL + 2 * STEP); + +async function sqliteEngine(...objects: unknown[]): Promise { + const engine = new ObjectQL(); + engine.registerDriver(new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }), true); + await engine.init(); + for (const object of objects) engine.registry.registerObject(object as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); + return engine; +} + +/** How each store reads back a claim instant it was handed. */ +type ColumnModel = 'real' | 'exact'; +const readBack = (model: ColumnModel, now: number) => (model === 'real' ? readReal(now) as number : now); + +interface Store { + readonly name: string; + readonly column: ColumnModel; + create(): Promise; + destroy(): Promise; +} + +function sqlStore(name: string, column: ColumnModel, object: unknown, build: (engine: IDataEngine) => O): Store { + let engine: ObjectQL | undefined; + return { + name, + column, + async create() { + engine = await sqliteEngine(object); + return build(column === 'real' ? realColumnEngine(engine) : engine as unknown as IDataEngine); + }, + async destroy() { + try { await engine?.destroy(); } catch { /* noop */ } + engine = undefined; + }, + }; +} + +describe('the real-column double reproduces the live Postgres readings', () => { + it('stores binary32, reads back the shortest decimal (PostgreSQL 16.15, extra_float_digits = 1)', () => { + // Measured: written through SqlDriver into a `real` column, read back + // through it. Neither reading is the written value, nor its fround. + expect(readReal(storeReal(1_791_810_012_345))).toBe(1_791_810_100_000); + expect(readReal(storeReal(1_791_813_456_789))).toBe(1_791_813_500_000); + expect(readReal(storeReal(1_791_857_407_123))).toBe(1_791_857_400_000); + // So the legs below exercise a lossy column, not an exact one. + expect(readBack('real', T0)).not.toBe(T0); + expect(Math.fround(T0)).not.toBe(T0); + }); +}); + +// ── SqlNotificationOutbox ─────────────────────────────────────────────────── + +const NOTIFY_REFUSAL = { name: 'NotificationAckError', code: 'DELIVERY_NOT_ELIGIBLE' }; + +const notificationStores: Store[] = [ + sqlStore('SqlNotificationOutbox on a real column', 'real', NotificationDelivery, + (engine) => new SqlNotificationOutbox(engine, { partitionCount: 1 })), + sqlStore('SqlNotificationOutbox on an exact column (control)', 'exact', NotificationDelivery, + (engine) => new SqlNotificationOutbox(engine, { partitionCount: 1 })), + { + name: 'MemoryNotificationOutbox (control)', + column: 'exact', + async create() { return new MemoryNotificationOutbox(1); }, + async destroy() { /* nothing to tear down */ }, + }, +]; + +function notifyClaim(nodeId: string, now: number) { + return { nodeId, limit: 10, claimTtlMs: TTL, now }; +} + +/** One state line a human can diff: status, holder, claim instant, attempts. */ +function notifyFingerprint(r: NotificationDeliveryRecord): string { + return `${r.status}:${r.claimedBy ?? '-'}:${r.claimedAt ?? '-'}:${r.attempts}`; +} + +/** + * One dispatcher pass, the shape `NotificationDispatcher.runTick` runs: the + * reap, a claim, a send per claimed row, its ack — with the dispatcher's + * tolerance of a refused ack (`DELIVERY_NOT_ELIGIBLE` is absorbed, the row + * left to the next reap). Returns the ids it sent. + */ +async function notifyPass(outbox: INotificationOutbox, nodeId: string, now: number): Promise { + await outbox.reap?.({ claimTtlMs: TTL, now }); + const claimed = await outbox.claim({ ...notifyClaim(nodeId, now), skipReap: true }); + for (const row of claimed) { + try { + await outbox.ack(row, { success: true }); + } catch (err) { + if ((err as { code?: string }).code !== 'DELIVERY_NOT_ELIGIBLE') throw err; + } + } + return claimed.map((r) => r.id); +} + +describe.each(notificationStores)('$name — the claim credential round-trips the column', (store) => { + let outbox: INotificationOutbox; + + beforeEach(async () => { outbox = await store.create(); }); + afterEach(async () => { await store.destroy(); }); + + function enqueueOne(): Promise { + return outbox.enqueue({ notificationId: 'n1', recipientId: 'u1', channel: 'inbox', payload: { title: 'hello' } }); + } + + 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('one send, one ack: the row reaches success, and no later pass sends it again', async () => { + const id = await enqueueOne(); + + const claimed = await outbox.claim(notifyClaim('node-a', T0)); + expect(claimed.map((r) => r.id)).toEqual([id]); + // The credential is the claim instant AS THE STORE HOLDS IT — what + // every later read of the cell returns. + expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); + expect(claimed[0].claimedAt).toBe((await readRow(id)).claimedAt); + + await expect(outbox.ack(claimed[0], { success: true, durationMs: 5 })).resolves.toBeUndefined(); + expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + + // The loop is closed: passes over three later reap windows send nothing. + const sent = [ + ...await notifyPass(outbox, 'node-a', passAt(1)), + ...await notifyPass(outbox, 'node-a', passAt(2)), + ...await notifyPass(outbox, 'node-a', passAt(3)), + ]; + expect(sent).toEqual([]); + expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); + + it('a claim lost to another node still refuses the late ack, and the holder\'s ack lands', async () => { + const id = await enqueueOne(); + + // node A claims and starts a send that outruns the visibility timeout; + // node B's claim reaps the row and re-claims it. + const [claimA] = await outbox.claim(notifyClaim('node-a', T0)); + const [claimB] = await outbox.claim(notifyClaim('node-b', T_REAPED)); + expect([claimA?.id, claimB?.id]).toEqual([id, id]); + expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); + + // A's late ack is refused by identity and writes nothing. + await expect(outbox.ack(claimA, { success: true, durationMs: TTL + 5 })).rejects.toMatchObject(NOTIFY_REFUSAL); + expect(notifyFingerprint(await readRow(id))).toBe(`in_flight:node-b:${claimB.claimedAt}:0`); + + // B's ack — the live attempt A would have overwritten — lands. + await expect(outbox.ack(claimB, { success: true, durationMs: 3 })).resolves.toBeUndefined(); + expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); + + // The stuck state exists only where the old credential failed: the lossy column. + if (store.column === 'real') it('a row stuck in_flight behind an ack that could never land is sent once more, acked, and left alone', async () => { + const id = await enqueueOne(); + + // The production state: claim() stored the row in_flight, and the ack + // carried the in-memory `now` as its credential — what claim() handed + // out before the credential was read back. That ack can never land. + const [stale] = await outbox.claim(notifyClaim('node-old', T0)); + const preFixCredential: ClaimedDeliveryRecord = { ...stale, claimedAt: T0 }; + await expect(outbox.ack(preFixCredential, { success: true })).rejects.toMatchObject(NOTIFY_REFUSAL); + expect(notifyFingerprint(await readRow(id))).toBe(`in_flight:node-old:${readBack('real', T0)}:0`); + + // The dispatcher's passes from here on, with no operator act: the reap + // returns the row, the next claim takes it, the send's ack lands — and + // every pass after that finds nothing. + const sent: string[] = []; + for (let k = 1; k <= 5; k++) sent.push(...await notifyPass(outbox, 'node-a', passAt(k))); + expect(sent).toEqual([id]); + expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); +}); + +// ── SqlHttpOutbox ─────────────────────────────────────────────────────────── + +const HTTP_REFUSAL = { name: 'HttpAckError', code: 'DELIVERY_NOT_ELIGIBLE' }; +const HTTP_OK = { success: true as const, httpStatus: 200, durationMs: 4 }; + +const httpStores: Store[] = [ + sqlStore('SqlHttpOutbox on a real column', 'real', HttpDelivery, + (engine) => new SqlHttpOutbox(engine, { partitionCount: 1 })), + sqlStore('SqlHttpOutbox on an exact column (control)', 'exact', HttpDelivery, + (engine) => new SqlHttpOutbox(engine, { partitionCount: 1 })), + { + name: 'MemoryHttpOutbox (control)', + column: 'exact', + async create() { return new MemoryHttpOutbox(); }, + async destroy() { /* nothing to tear down */ }, + }, +]; + +function httpClaim(nodeId: string, now: number) { + return { nodeId, limit: 10, claimTtlMs: TTL, now }; +} + +function httpFingerprint(r: HttpDeliveryRow): string { + return `${r.status}:${r.claimedBy ?? '-'}:${r.claimedAt ?? '-'}:${r.attempts}`; +} + +/** The `HttpDispatcher` pass — same shape as {@link notifyPass}. */ +async function httpPass(outbox: IHttpOutbox, nodeId: string, now: number): Promise { + await outbox.reap?.({ claimTtlMs: TTL, now }); + const claimed = await outbox.claim({ ...httpClaim(nodeId, now), skipReap: true }); + for (const row of claimed) { + try { + await outbox.ack(row.id, HTTP_OK, row as { claimedBy: string; claimedAt: number }); + } catch (err) { + if ((err as { code?: string }).code !== 'DELIVERY_NOT_ELIGIBLE') throw err; + } + } + return claimed.map((r) => r.id); +} + +describe.each(httpStores)('$name — the claim credential round-trips the column', (store) => { + let outbox: IHttpOutbox; + + beforeEach(async () => { outbox = await store.create(); }); + afterEach(async () => { await store.destroy(); }); + + function enqueueOne(): Promise { + return outbox.enqueue({ source: 'flow', refId: 'r1', dedupKey: 'd_r1', url: 'https://receiver.example/r1', payload: { n: 1 } }); + } + + 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; + } + + function credentialOf(row: HttpDeliveryRow) { + return { claimedBy: row.claimedBy as string, claimedAt: row.claimedAt as number }; + } + + it('one send, one ack: the row reaches success, and no later pass sends it again', async () => { + const id = await enqueueOne(); + + const claimed = await outbox.claim(httpClaim('node-a', T0)); + expect(claimed.map((r) => r.id)).toEqual([id]); + expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); + expect(claimed[0].claimedAt).toBe((await readRow(id)).claimedAt); + + await expect(outbox.ack(id, HTTP_OK, credentialOf(claimed[0]))).resolves.toBeUndefined(); + expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + + const sent = [ + ...await httpPass(outbox, 'node-a', passAt(1)), + ...await httpPass(outbox, 'node-a', passAt(2)), + ...await httpPass(outbox, 'node-a', passAt(3)), + ]; + expect(sent).toEqual([]); + expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); + + it('a claim lost to another node still refuses the late ack, and the holder\'s ack lands', async () => { + const id = await enqueueOne(); + + const [claimA] = await outbox.claim(httpClaim('node-a', T0)); + const [claimB] = await outbox.claim(httpClaim('node-b', T_REAPED)); + expect([claimA?.id, claimB?.id]).toEqual([id, id]); + expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); + + await expect(outbox.ack(id, HTTP_OK, credentialOf(claimA))).rejects.toMatchObject(HTTP_REFUSAL); + expect(httpFingerprint(await readRow(id))).toBe(`in_flight:node-b:${claimB.claimedAt}:0`); + + await expect(outbox.ack(id, HTTP_OK, credentialOf(claimB))).resolves.toBeUndefined(); + expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); + + if (store.column === 'real') it('a row stuck in_flight behind an ack that could never land is sent once more, acked, and left alone', async () => { + const id = await enqueueOne(); + + const [stale] = await outbox.claim(httpClaim('node-old', T0)); + await expect(outbox.ack(id, HTTP_OK, { claimedBy: stale.claimedBy as string, claimedAt: T0 })) + .rejects.toMatchObject(HTTP_REFUSAL); + expect(httpFingerprint(await readRow(id))).toBe(`in_flight:node-old:${readBack('real', T0)}:0`); + + const sent: string[] = []; + for (let k = 1; k <= 5; k++) sent.push(...await httpPass(outbox, 'node-a', passAt(k))); + expect(sent).toEqual([id]); + expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + }); +}); From dee1e11da85a0c4156140f3b271fe4ad1519b39a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:26:23 +0000 Subject: [PATCH 3/5] test(service-messaging): assert the ack before the credential value in the lossy-column pins Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- ...redential-lossy-column.integration.test.ts | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts index 1bda2d23ee3..0b122c889b1 100644 --- a/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts +++ b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts @@ -284,13 +284,14 @@ describe.each(notificationStores)('$name — the claim credential round-trips th const claimed = await outbox.claim(notifyClaim('node-a', T0)); expect(claimed.map((r) => r.id)).toEqual([id]); - // The credential is the claim instant AS THE STORE HOLDS IT — what - // every later read of the cell returns. - expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); - expect(claimed[0].claimedAt).toBe((await readRow(id)).claimedAt); + const stored = (await readRow(id)).claimedAt; await expect(outbox.ack(claimed[0], { success: true, durationMs: 5 })).resolves.toBeUndefined(); expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + // The credential that landed is the claim instant AS THE STORE HOLDS + // IT — what a read of the cell returned — never the `now` written. + expect(claimed[0].claimedAt).toBe(stored); + expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); // The loop is closed: passes over three later reap windows send nothing. const sent = [ @@ -310,15 +311,15 @@ describe.each(notificationStores)('$name — the claim credential round-trips th const [claimA] = await outbox.claim(notifyClaim('node-a', T0)); const [claimB] = await outbox.claim(notifyClaim('node-b', T_REAPED)); expect([claimA?.id, claimB?.id]).toEqual([id, id]); - expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); // A's late ack is refused by identity and writes nothing. await expect(outbox.ack(claimA, { success: true, durationMs: TTL + 5 })).rejects.toMatchObject(NOTIFY_REFUSAL); - expect(notifyFingerprint(await readRow(id))).toBe(`in_flight:node-b:${claimB.claimedAt}:0`); + expect(notifyFingerprint(await readRow(id))).toBe(`in_flight:node-b:${readBack(store.column, T_REAPED)}:0`); // B's ack — the live attempt A would have overwritten — lands. await expect(outbox.ack(claimB, { success: true, durationMs: 3 })).resolves.toBeUndefined(); expect(notifyFingerprint(await readRow(id))).toBe('success:-:-:1'); + expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); }); // The stuck state exists only where the old credential failed: the lossy column. @@ -408,11 +409,12 @@ describe.each(httpStores)('$name — the claim credential round-trips the column const claimed = await outbox.claim(httpClaim('node-a', T0)); expect(claimed.map((r) => r.id)).toEqual([id]); - expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); - expect(claimed[0].claimedAt).toBe((await readRow(id)).claimedAt); + const stored = (await readRow(id)).claimedAt; await expect(outbox.ack(id, HTTP_OK, credentialOf(claimed[0]))).resolves.toBeUndefined(); expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + expect(claimed[0].claimedAt).toBe(stored); + expect(claimed[0].claimedAt).toBe(readBack(store.column, T0)); const sent = [ ...await httpPass(outbox, 'node-a', passAt(1)), @@ -429,13 +431,13 @@ describe.each(httpStores)('$name — the claim credential round-trips the column const [claimA] = await outbox.claim(httpClaim('node-a', T0)); const [claimB] = await outbox.claim(httpClaim('node-b', T_REAPED)); expect([claimA?.id, claimB?.id]).toEqual([id, id]); - expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); await expect(outbox.ack(id, HTTP_OK, credentialOf(claimA))).rejects.toMatchObject(HTTP_REFUSAL); - expect(httpFingerprint(await readRow(id))).toBe(`in_flight:node-b:${claimB.claimedAt}:0`); + expect(httpFingerprint(await readRow(id))).toBe(`in_flight:node-b:${readBack(store.column, T_REAPED)}:0`); await expect(outbox.ack(id, HTTP_OK, credentialOf(claimB))).resolves.toBeUndefined(); expect(httpFingerprint(await readRow(id))).toBe('success:-:-:1'); + expect(claimB.claimedAt).toBe(readBack(store.column, T_REAPED)); }); if (store.column === 'real') it('a row stuck in_flight behind an ack that could never land is sent once more, acked, and left alone', async () => { From b8b1dedb341048ed09a61784fc4c22cf9bdcc824 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:29:34 +0000 Subject: [PATCH 4/5] chore(changeset): service-messaging patch for the read-back claim credential Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../23122-outbox-claim-credential-read-back.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/23122-outbox-claim-credential-read-back.md diff --git a/.changeset/23122-outbox-claim-credential-read-back.md b/.changeset/23122-outbox-claim-credential-read-back.md new file mode 100644 index 00000000000..b630c5020db --- /dev/null +++ b/.changeset/23122-outbox-claim-credential-read-back.md @@ -0,0 +1,13 @@ +--- +'@objectstack/service-messaging': patch +--- + +The notification and HTTP outboxes read their claim credential back from the store, so `ack()` lands on a Postgres table whose epoch-ms columns are still `real` + +Clause-②: no + +Tables created while the `number` field type still mapped to `real` keep those columns as IEEE-754 binary32. Newer tables get an exact numeric column, and existing columns are not migrated. A binary32 `claimed_at` stores a rounding of the claim's `now`, to the nearest 131 072 ms at today's magnitude, and reads back a nearby decimal. `SqlNotificationOutbox` and `SqlHttpOutbox` handed out the in-memory `now` as the claim credential, so it never equalled the stored value. Every `ack()` was refused with `DELIVERY_NOT_ELIGIBLE` as a lost claim, and the dispatcher absorbed the refusal. The row stayed `in_flight` with `attempts = 0`, and every reap handed it to the next claim. The notification was re-delivered about every 131 s, with no end. + +- **What changes.** `claim()` and `claimDigest()` return `claimedAt` as the `claimed_at` value the store returned after the claiming update. The value written is unchanged. The rows a claim took are identified by candidate id, holder and status. `claimed_at` is no longer compared with the client's `now` in that read. `ack()`'s ownership check and its compare-and-set are unchanged. They now compare that stored value with the stored value, which is exact on any numeric column. +- **A stuck row heals by itself.** A row left `in_flight` by a refused ack is reaped by the visibility timeout, claimed again, sent once and acked. No operator act is needed. On a `real` column the reap still runs at binary32 granularity, up to about 131 s after `claimTtlMs`. +- **What is unchanged.** A genuinely lost claim is still refused. If another node, or this node's own later claim, reaped and re-claimed the row while a send ran, the late ack gets `DELIVERY_NOT_ELIGIBLE` and writes nothing. The `ClaimedDeliveryRecord` and `HttpClaimCredential` types are unchanged. Existing `real` columns are not migrated, and duplicate inbox rows that were already delivered stay as they are. From 82c0b1457f17b73b1cb91fc25180e3c79f1791f2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 02:56:08 +0000 Subject: [PATCH 5/5] test(service-messaging): type the real-column double's engine calls instead of erasing their options Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- ...redential-lossy-column.integration.test.ts | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts index 0b122c889b1..59b3464df64 100644 --- a/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts +++ b/packages/services/service-messaging/src/outbox-claim-credential-lossy-column.integration.test.ts @@ -107,9 +107,9 @@ function coerceComparands(where: unknown): unknown { ])); } -function mapEpochColumns(row: unknown, fn: (value: unknown) => unknown): unknown { +function mapEpochColumns(row: R, fn: (value: unknown) => unknown): R { if (!row || typeof row !== 'object') return row; - return Object.fromEntries(Object.entries(row).map(([k, v]) => [k, EPOCH_MS_COLUMNS.has(k) ? fn(v) : v])); + return Object.fromEntries(Object.entries(row).map(([k, v]) => [k, EPOCH_MS_COLUMNS.has(k) ? fn(v) : v])) as R; } function withComparands(query: Q): Q { @@ -127,19 +127,19 @@ function realColumnEngine(engine: ObjectQL): IDataEngine { get(target, prop, receiver) { switch (prop) { case 'find': - return async (object: string, query: unknown, options?: unknown) => { - const rows = await target.find(object, withComparands(query) as any, options as any); - return (rows as unknown[]).map((r) => mapEpochColumns(r, readReal)); + return async (...[object, query, options]: Parameters) => { + const rows = await target.find(object, withComparands(query), options); + return rows.map((r) => mapEpochColumns(r, readReal)); }; case 'findOne': - return async (object: string, query: unknown, options?: unknown) => - mapEpochColumns(await target.findOne(object, withComparands(query) as any, options as any), readReal); + return async (...[object, query, options]: Parameters) => + mapEpochColumns(await target.findOne(object, withComparands(query), options), readReal); case 'insert': - return (object: string, data: unknown, options?: unknown) => - target.insert(object, mapEpochColumns(data, storeReal) as any, options as any); + return (...[object, data, options]: Parameters) => + target.insert(object, mapEpochColumns(data, storeReal), options); case 'update': - return (object: string, data: unknown, options?: unknown) => - target.update(object, mapEpochColumns(data, storeReal) as any, withComparands(options) as any); + return (...[object, data, options]: Parameters) => + target.update(object, mapEpochColumns(data, storeReal), withComparands(options)); default: { const value = Reflect.get(target, prop, receiver); return typeof value === 'function' ? value.bind(target) : value;