From 6e3a6f7c26c3d1eeba7fe7d4c8d8295e5b53a67f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 11:00:16 +0000 Subject: [PATCH 1/2] fix(service-queue): compare the publish idempotency window as instants, not strings (#13993) The idempotency check deduped terminal rows with a lexicographic String(row.created_at) compare against ISO text. On Postgres/MySQL the builtin audit column materialises as a JS Date whose String() starts with a weekday letter, unconditionally above the ISO window-start's digit, so the predicate was always true: terminal rows blocked re-publish forever and publish() silently enqueued nothing. Normalise created_at to an instant (the canonicalVersionInstant shape) and compare epoch ms; the pending/running arm and SQLite verdicts are unchanged. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016ZC5rNQj3WEet5HAmmAkMs --- ...ueue-idempotency-window-instant-compare.md | 24 +++ ...idempotency-window-materialisation.test.ts | 192 ++++++++++++++++++ .../service-queue/src/db-queue-adapter.ts | 64 +++++- 3 files changed, 278 insertions(+), 2 deletions(-) create mode 100644 .changeset/queue-idempotency-window-instant-compare.md create mode 100644 packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts diff --git a/.changeset/queue-idempotency-window-instant-compare.md b/.changeset/queue-idempotency-window-instant-compare.md new file mode 100644 index 0000000000..20dc405433 --- /dev/null +++ b/.changeset/queue-idempotency-window-instant-compare.md @@ -0,0 +1,24 @@ +--- +"@objectstack/service-queue": patch +--- + +fix(service-queue): compare the publish idempotency window as instants, not strings (#13993) + +`DbQueueAdapter#publish` deduped terminal rows with +`String(row.created_at) >= windowStart` — a lexicographic compare of the raw +driver value against canonical ISO text. On Postgres/MySQL the builtin audit +column `created_at` comes out of the record read door as a JS `Date`, whose +`String()` begins with a weekday letter, unconditionally above the ISO +window-start's leading digit — so the predicate was always true: any terminal +(`completed`/`dlq`) row with that idempotency key blocked re-publish forever, +and `publish()` returned the old id having enqueued nothing. Silent message +loss on the production default drivers; SQLite (ISO text on both sides) was +always correct, which is why every existing test stayed green. + +The check now normalises `created_at` to an instant (the #13382 +`canonicalVersionInstant` shape: `Date`, epoch-ms number, or absolute ISO +text) and compares epoch milliseconds, so every dialect gets the declared +window semantics. The `pending`/`running` arm — which blocks regardless of +age — is untouched, and SQLite verdicts are unchanged. A `created_at` that +denotes no instant cannot be inside a window measured on the `created_at` +axis and no longer blocks. diff --git a/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts b/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts new file mode 100644 index 0000000000..32d4509910 --- /dev/null +++ b/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts @@ -0,0 +1,192 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#13993] The publish idempotency window, driven through every `created_at` + * materialisation a driver actually hands out of the record read door. + * + * The defect: `DbQueueAdapter#publish` compared + * `String(row.created_at) >= windowStart` — lexicographic text against + * canonical ISO text. On Postgres/MySQL the builtin audit column `created_at` + * comes back as a JS `Date` (pinned in `driver-sql`'s + * `sql-driver-13567-audit-stamp-materialisation.test.ts`), whose `String()` + * starts with a weekday LETTER (0x41–0x5A), unconditionally above the ISO + * text's leading digit `'2'` (0x32) — so the predicate was TRUE for every + * terminal row, the window never expired, and `publish()` returned the old id + * having enqueued nothing: silent message loss on the production default + * drivers. SQLite hands back ISO-Z text on both sides, so the SQLite arm was + * always correct — which is why every existing test stayed green, and why the + * ISO cases below are the CONTROL group: they must keep passing unchanged. + * + * The discriminating `Date` input exists in CI only inside + * `@objectstack/driver-sql` today (#13973 point 4), so this pin lives in THIS + * package driving hand-made `Date`s — deliberately NOT by widening any + * required job's package set (#13567, maintainer decision). + * + * Assertions are DIRECTIONAL, not literal: out-of-window must stop blocking, + * in-window must keep blocking, and `pending`/`running` rows must block + * regardless of age (that arm bypasses the time compare entirely). + */ + +import { describe, it, expect } from 'vitest'; +import { DbQueueAdapter } from './db-queue-adapter.js'; + +/** Minimal engine double — only the surface `publish()` touches. */ +function makeFakeEngine(seed: any[] = []) { + const rows: any[] = [...seed]; + return { + rows, + async find(_table: string, opts: any = {}) { + let out = opts?.where + ? rows.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v)) + : [...rows]; + if (opts?.limit) out = out.slice(0, opts.limit); + return out; + }, + async insert(_table: string, data: any) { + rows.push({ ...data }); + return { id: data.id }; + }, + async update(): Promise { + throw new Error('not reachable from publish()'); + }, + async delete(): Promise { + throw new Error('not reachable from publish()'); + }, + }; +} + +/** Frozen "now" so window edges are deterministic. */ +const NOW_MS = Date.parse('2026-08-30T10:00:00.000Z'); +const WINDOW_MS = 60_000; + +function makeAdapter(seed: any[]) { + const engine = makeFakeEngine(seed); + const adapter = new DbQueueAdapter({ + engine, + clock: { now: () => new Date(NOW_MS) }, + options: { autoStart: false, idempotencyWindowMs: WINDOW_MS }, + }); + return { engine, adapter }; +} + +function terminalRow(id: string, status: 'completed' | 'dlq', createdAt: unknown) { + return { + id, + queue: 'q', + idempotency_key: 'k', + status, + created_at: createdAt, + }; +} + +describe('[#13993] publish idempotency window vs created_at materialisation', () => { + describe('Date side (Postgres/MySQL/Mongo hand the audit column back as a JS Date)', () => { + it('an OUT-OF-WINDOW terminal Date row no longer blocks — publish enqueues a NEW message', async () => { + // Pre-fix this row blocked FOREVER: String(Date) begins with a weekday + // letter, lexicographically above the ISO windowStart's digit. + const { engine, adapter } = makeAdapter([ + terminalRow('row_old', 'completed', new Date(NOW_MS - 2 * WINDOW_MS)), + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).not.toBe('row_old'); + const inserted = engine.rows.find((r) => r.id === id); + expect(inserted).toBeDefined(); + expect(inserted.status).toBe('pending'); + expect(engine.rows).toHaveLength(2); + }); + + it('an IN-WINDOW terminal Date row still blocks — old id back, nothing enqueued', async () => { + const { engine, adapter } = makeAdapter([ + terminalRow('row_recent', 'dlq', new Date(NOW_MS - WINDOW_MS / 2)), + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).toBe('row_recent'); + expect(engine.rows).toHaveLength(1); + }); + }); + + describe('ISO-text side (SQLite/turso/wasm/memory status quo — the CONTROL group)', () => { + // The lexicographic compare was CORRECT on ISO-Z text (order = chronology). + // These two must hold before AND after the fix; a red here is a regression + // in the only arm that ever worked. + it('an OUT-OF-WINDOW terminal ISO row does not block (as it never did on SQLite)', async () => { + const { engine, adapter } = makeAdapter([ + terminalRow('row_old_iso', 'completed', new Date(NOW_MS - 2 * WINDOW_MS).toISOString()), + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).not.toBe('row_old_iso'); + expect(engine.rows).toHaveLength(2); + }); + + it('an IN-WINDOW terminal ISO row still blocks', async () => { + const { engine, adapter } = makeAdapter([ + terminalRow('row_recent_iso', 'completed', new Date(NOW_MS - WINDOW_MS / 2).toISOString()), + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).toBe('row_recent_iso'); + expect(engine.rows).toHaveLength(1); + }); + }); + + describe('epoch-ms number (pre-canonical / hand-migrated SQLite column)', () => { + it('windowed verdicts hold for a numeric created_at too', async () => { + const outOfWindow = makeAdapter([ + terminalRow('row_old_num', 'completed', NOW_MS - 2 * WINDOW_MS), + ]); + const idA = await outOfWindow.adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(idA).not.toBe('row_old_num'); + + const inWindow = makeAdapter([ + terminalRow('row_recent_num', 'completed', NOW_MS - WINDOW_MS / 2), + ]); + const idB = await inWindow.adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(idB).toBe('row_recent_num'); + }); + }); + + describe('reverse control: the non-terminal arm bypasses the time compare', () => { + // pending/running block REGARDLESS of age — prove the fix did not narrow + // that arm. Both materialisations, both statuses, absurdly old stamps. + it('a pending row blocks however old, Date and ISO alike', async () => { + for (const createdAt of [ + new Date(NOW_MS - 1000 * WINDOW_MS), + new Date(NOW_MS - 1000 * WINDOW_MS).toISOString(), + ]) { + const { engine, adapter } = makeAdapter([ + { id: 'row_pending', queue: 'q', idempotency_key: 'k', status: 'pending', created_at: createdAt }, + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).toBe('row_pending'); + expect(engine.rows).toHaveLength(1); + } + }); + + it('a running row blocks however old, Date and ISO alike', async () => { + for (const createdAt of [ + new Date(NOW_MS - 1000 * WINDOW_MS), + new Date(NOW_MS - 1000 * WINDOW_MS).toISOString(), + ]) { + const { engine, adapter } = makeAdapter([ + { id: 'row_running', queue: 'q', idempotency_key: 'k', status: 'running', created_at: createdAt }, + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).toBe('row_running'); + expect(engine.rows).toHaveLength(1); + } + }); + }); + + describe('a created_at that denotes no instant', () => { + it('cannot be inside a window measured on the created_at axis — does not block', async () => { + // Documented decision (createdAtInstantMs): duplicate delivery is + // tolerated by contract; "suppress forever" is the defect. Pre-fix this + // very value DID block forever ('n' is above '2' lexicographically). + const { engine, adapter } = makeAdapter([ + terminalRow('row_opaque', 'completed', 'not-an-instant'), + ]); + const id = await adapter.publish('q', { x: 1 }, { idempotencyKey: 'k' }); + expect(id).not.toBe('row_opaque'); + expect(engine.rows).toHaveLength(2); + }); + }); +}); diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts index 1acc3816bb..85f0e450b0 100644 --- a/packages/services/service-queue/src/db-queue-adapter.ts +++ b/packages/services/service-queue/src/db-queue-adapter.ts @@ -21,6 +21,57 @@ import { const QUEUE_TABLE = 'sys_job_queue'; +/** + * [#13993] An ISO-8601 date-time that denotes an ABSOLUTE instant — it + * carries an explicit `Z` or a numeric offset, so reading it never consults + * the process timezone. Same shape as the OCC seam's `ABSOLUTE_ISO_INSTANT` + * (#13382, `packages/metadata-protocol/src/protocol.ts`): a string this + * pattern rejects does not denote an instant and is not guessed at. + */ +const ABSOLUTE_ISO_INSTANT = + /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:\d{2})$/; + +/** + * [#13993] A `created_at` as a driver hands it out of the record read door, + * read as epoch milliseconds — or null when the value does not denote an + * instant. + * + * `created_at` is a builtin audit column: it is not in `datetimeFields`, so no + * declared-field coercion reaches it, and the dialects genuinely disagree on + * its materialisation (the domain below is the one #13382 measured and #13973 + * re-measured, pinned in `driver-sql`'s + * `sql-driver-13567-audit-stamp-materialisation.test.ts`): + * + * - **JS `Date`** — Postgres / MySQL (`timestamptz` / `DATETIME(3)` are + * instants and the driver materialises them as `Date` on purpose; + * `SqlDriver.withPostgresCalendarDayAsText` says so in as many words). + * - **canonical ISO-8601 UTC text** — SQLite and its turso / sqlite-wasm + * siblings, and `driver-memory`. This adapter writes `toISOString()`. + * - **`number`, epoch milliseconds** — a pre-canonical or hand-migrated + * SQLite column; the legacy datetime repair is keyed on declared + * `Field.datetime` columns, and the engine-injected audit columns are not + * in that set. + * - **anything else** — not an instant. Returns null, and the caller treats + * the row as OUTSIDE the window: the dedup window is measured on the + * `created_at` axis, so a row that cannot be placed on that axis cannot be + * inside it (and duplicate delivery is tolerated by contract — see + * `claimBatch` — while "suppress forever" is the very defect #13993 + * removes). + */ +function createdAtInstantMs(value: unknown): number | null { + let ms: number; + if (value instanceof Date) { + ms = value.getTime(); + } else if (typeof value === 'number') { + ms = value; + } else if (typeof value === 'string' && ABSOLUTE_ISO_INSTANT.test(value.trim())) { + ms = Date.parse(value.trim()); + } else { + return null; + } + return Number.isFinite(ms) ? ms : null; +} + /** * How long a `completed` row survives before the platform Reaper deletes it. * @@ -247,7 +298,7 @@ export class DbQueueAdapter implements IQueueService { // constructor — which makes "the reaper deleted a row the dedup check // needed" unrepresentable rather than merely unlikely. if (opts.idempotencyKey) { - const windowStart = new Date(now.getTime() - this.opts.idempotencyWindowMs).toISOString(); + const windowStartMs = now.getTime() - this.opts.idempotencyWindowMs; const existing = await this.engine.find(QUEUE_TABLE, { where: { queue, @@ -259,7 +310,16 @@ export class DbQueueAdapter implements IQueueService { }); const blocking = (existing ?? []).find((row: any) => { if (row.status === 'pending' || row.status === 'running') return true; - return String(row.created_at ?? '') >= windowStart; + // [#13993] Compare INSTANTS, not strings (#13382's shape). The old + // `String(row.created_at) >= windowStart` was a lexicographic compare + // whose left side, on Postgres/MySQL, is a `Date.toString()` starting + // with a weekday LETTER — unconditionally above the ISO text's digit — + // so every terminal row blocked forever and publish() silently + // enqueued nothing. An instant compare gives every materialisation the + // same verdict; a row whose created_at denotes no instant cannot be + // inside the window (see createdAtInstantMs). + const createdAtMs = createdAtInstantMs(row.created_at); + return createdAtMs !== null && createdAtMs >= windowStartMs; }); if (blocking) return String(blocking.id); } From 51b3804a0b304c7e15adbc39f31f719065885d0c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 11:32:17 +0000 Subject: [PATCH 2/2] test(service-queue): conform the fake engine to the double gates and register its pins The new double's update()/delete() now open with the engine's own dispatch predicates, find() bounds by presence and refuses combinators, and the engine-double-contract RETAINED ledger records the new (file, verb) pins. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016ZC5rNQj3WEet5HAmmAkMs --- ...idempotency-window-materialisation.test.ts | 29 ++++++++++++++----- scripts/engine-double-contract.pinned.json | 10 +++++++ 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts b/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts index 32d4509910..b9ce549d1e 100644 --- a/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts +++ b/packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts @@ -28,28 +28,43 @@ */ import { describe, it, expect } from 'vitest'; +import { + assertEngineDeleteDispatch, + assertEngineUpdateDispatch, +} from '@objectstack/objectql'; import { DbQueueAdapter } from './db-queue-adapter.js'; -/** Minimal engine double — only the surface `publish()` touches. */ +/** + * Minimal engine double — only the surface `publish()` touches. `update()` and + * `delete()` are unreachable from `publish()`, but they still open with the + * engine's own dispatch predicates so this fake can never drift looser than + * ObjectQL's contract (`check:engine-double-contract`). + */ function makeFakeEngine(seed: any[] = []) { const rows: any[] = [...seed]; return { rows, async find(_table: string, opts: any = {}) { - let out = opts?.where - ? rows.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v)) + const out = opts?.where + ? rows.filter((r) => Object.entries(opts.where).every(([k, v]) => { + // Refuse combinators rather than reading them as field names. + if (k.startsWith('$')) throw new Error(`fake engine: unsupported operator ${k}`); + return r[k] === v; + })) : [...rows]; - if (opts?.limit) out = out.slice(0, opts.limit); - return out; + // The caller's bound, by PRESENCE — `limit: 0` must bound to zero rows. + return typeof opts?.limit === 'number' ? out.slice(0, opts.limit) : out; }, async insert(_table: string, data: any) { rows.push({ ...data }); return { id: data.id }; }, - async update(): Promise { + async update(_table: string, data: any, options?: any): Promise { + assertEngineUpdateDispatch(data, options); throw new Error('not reachable from publish()'); }, - async delete(): Promise { + async delete(_table: string, options?: any): Promise { + assertEngineDeleteDispatch(options); throw new Error('not reachable from publish()'); }, }; diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index c1d6016491..8d43aa123a 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3241,6 +3241,16 @@ "verb": "findOne", "pinned": 1 }, + { + "file": "packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-queue/src/db-queue-adapter.test.ts", "verb": "delete",