fix(service-queue): compare the publish idempotency window as instants, not strings

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",

.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.

packages/services/service-queue/src/db-queue-adapter-13993-idempotency-window-materialisation.test.ts

@@ -0,0 +1,207 @@
+// 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 {
+ assertEngineDeleteDispatch,
+ assertEngineUpdateDispatch,
+} from '@objectstack/objectql';
+import { DbQueueAdapter } from './db-queue-adapter.js';
+
+/**
+ * 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 = {}) {
+ 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];
+ // 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(_table: string, data: any, options?: any): Promise<never> {
+ assertEngineUpdateDispatch(data, options);
+ throw new Error('not reachable from publish()');
+ },
+ async delete(_table: string, options?: any): Promise<never> {
+ assertEngineDeleteDispatch(options);
+ 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);
+ });
+ });
+});

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);
  }

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",