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

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

The idempotency check deduped terminal rows with a lexicographic
String(row.created_at) compare against ISO text. On Postgres/MySQL the
builtin audit column materialises as a JS Date whose String() starts with
a weekday letter, unconditionally above the ISO window-start's digit, so
the predicate was always true: terminal rows blocked re-publish forever
and publish() silently enqueued nothing. Normalise created_at to an
instant (the canonicalVersionInstant shape) and compare epoch ms; the
pending/running arm and SQLite verdicts are unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ZC5rNQj3WEet5HAmmAkMs

* test(service-queue): conform the fake engine to the double gates and register its pins

The new double's update()/delete() now open with the engine's own dispatch
predicates, find() bounds by presence and refuses combinators, and the
engine-double-contract RETAINED ledger records the new (file, verb) pins.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ZC5rNQj3WEet5HAmmAkMs

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: claude[bot]

.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

@@ -3256,6 +3256,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",