fix(service-queue,platform-objects): the claim path's sort keys join the declared index, and due-ness becomes a SQL predicate

.changeset/17612-job-queue-claim-index.md

@@ -0,0 +1,18 @@
+---
+"@objectstack/platform-objects": patch
+"@objectstack/service-queue": patch
+---
+
+`sys_job_queue`'s claim path no longer sorts the whole queue on every poll, and a job's due time is now a SQL predicate instead of a filter applied after `LIMIT` (#17612).
+
+`DbQueueAdapter.claimBatch` — the 1s poll every `DbQueueAdapter` runs — read the queue as `WHERE queue = ? AND status = 'pending' ORDER BY priority ASC, scheduled_for ASC`, while `sys_job_queue` declared `['queue','status','scheduled_for']`. The sort's **first** key, `priority`, was in no declared index at all, so the equality prefix seeked and the planner then built a sorter over every pending row in the queue, every tick. Measured on both Turso faces:
+
+```
+SEARCH sys_job_queue USING INDEX idx_sys_job_queue_queue_status_scheduled_for (queue=? AND status=?)
+USE TEMP B-TREE FOR ORDER BY
+```
+
+- **The declared index becomes `['queue','status','priority','scheduled_for']`**, replacing `['queue','status','scheduled_for']` — the table still declares three. The full-queue sort is gone on both faces; what remains is a sorter bounded to rows tying on the whole indexed prefix, because a paged read carries one ORDER BY term the caller never writes — the unique tie-breaker of the deterministic-paging contract (ADR-0053 D-A1), here `id`. ⛔ That last term is deliberately **not** closed by appending `id` to the index: `id` is an unbounded `Field.text`, and a text column a declared index keys on without a `maxLength` makes MySQL reject the index DDL outright (`check:keyed-text-bounds`, ER_BLOB_KEY_WITHOUT_LENGTH).
+- **Due-ness moved into `where`** as `$or: [{ scheduled_for: null }, { scheduled_for: { $lte: now } }]`, the same shape `SqlOutboxStore.claim` uses. It had been a JS filter applied to rows `LIMIT` had already chosen, so a window full of not-yet-due high-priority jobs hid already-due work behind it indefinitely: at the default `batchSize: 10` (candidate window 30), 30 future-dated `priority: 1` rows plus one due `priority: 100` row claimed **0** per poll, forever. It now claims 1.
+- **`priority` still decides claim order.** The alternative — dropping it from the sort — would have left a declared, documented field (`Lower = higher priority`) with no runtime effect at all.
+- ⚠️ **On an existing database the superseded index is not dropped.** The retrofit adds `idx_sys_job_queue_queue_status_priority_scheduled_for` and leaves `idx_sys_job_queue_queue_status_scheduled_for` in place (measured: 3 indexes before, 4 after, no row touched), so a provisioned table carries one redundant index until an operator drops it through the migrate-plan path. A freshly created table gets three.

packages/drivers/driver-turso/src/turso-local-remote-declared-index-parity.test.ts

@@ -124,7 +124,9 @@ const JOB_QUEUE: ObjectDef = {
     updated_at: Field.datetime({ label: 'Updated At', required: false }),
   },
   indexes: [
-    { fields: ['queue', 'status', 'scheduled_for'] },
+    // [#17612] Widened from `['queue','status','scheduled_for']` — verbatim
+    // from `sys-job-queue.object.ts`, which is where the reasoning lives.
+    { fields: ['queue', 'status', 'priority', 'scheduled_for'] },
     { fields: ['idempotency_key', 'queue'] },
     { fields: ['status'] },
   ],
@@ -168,7 +170,7 @@ const EXPECTED_NAMES: Record<string, string[]> = {
     `sqlite_autoindex_${DELIVERY.name}_1`,
   ].sort(),
   [JOB_QUEUE.name]: [
-    buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'scheduled_for'], false),
+    buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'priority', 'scheduled_for'], false),
     buildIndexName(JOB_QUEUE.name, ['idempotency_key', 'queue'], false),
     buildIndexName(JOB_QUEUE.name, ['status'], false),
     `sqlite_autoindex_${JOB_QUEUE.name}_1`,
@@ -185,6 +187,14 @@ const EXPECTED_NAMES: Record<string, string[]> = {
 };
 
 const CLAIM_INDEX = buildIndexName(DELIVERY.name, ['status', 'partition_key', 'next_attempt_at'], false);
+/** [#17612] The index `sys_job_queue`'s claim path must be served by, end to end. */
+const JOB_CLAIM_INDEX = buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'priority', 'scheduled_for'], false);
+/** The shape the remote face provisioned BEFORE #17612 — this file's negative control. */
+const JOB_QUEUE_PRE_17612: ObjectDef['indexes'] = [
+  { fields: ['queue', 'status', 'scheduled_for'] },
+  { fields: ['idempotency_key', 'queue'] },
+  { fields: ['status'] },
+];
 const DEDUP_INDEX = buildIndexName(DELIVERY.name, ['notification_id', 'recipient_id', 'channel'], true);
 
 /** A fresh copy per sync — neither face may see a definition the other one mutated. */
@@ -275,20 +285,27 @@ async function remoteFace(client: Client = createClient({ url: 'file::memory:' }
 }
 
 /** Every statement the driver hands the client, in order, with the call that carried it. */
-function countingClient(inner: Client): { client: Client; calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> } {
-  const calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> = [];
+function countingClient(inner: Client): {
+  client: Client;
+  calls: Array<{ via: 'execute' | 'batch'; sql: string[]; args: unknown[][] }>;
+} {
+  const calls: Array<{ via: 'execute' | 'batch'; sql: string[]; args: unknown[][] }> = [];
   const sqlOf = (s: InStatement) => (typeof s === 'string' ? s : s.sql);
+  // [#17612] The bound values, alongside the text — a captured statement is only
+  // re-runnable (`EXPLAIN QUERY PLAN`) if its placeholders can be filled again.
+  const argsOf = (s: InStatement): unknown[] =>
+    typeof s === 'string' ? [] : Array.isArray(s.args) ? [...s.args] : [];
   const client = new Proxy(inner, {
     get(target, prop) {
       if (prop === 'execute') {
         return (stmt: InStatement, ...rest: unknown[]) => {
-          calls.push({ via: 'execute', sql: [sqlOf(stmt)] });
+          calls.push({ via: 'execute', sql: [sqlOf(stmt)], args: [argsOf(stmt)] });
           return (target.execute as (...a: unknown[]) => unknown).call(target, stmt, ...rest);
         };
       }
       if (prop === 'batch') {
         return (stmts: InStatement[], ...rest: unknown[]) => {
-          calls.push({ via: 'batch', sql: stmts.map(sqlOf) });
+          calls.push({ via: 'batch', sql: stmts.map(sqlOf), args: stmts.map(argsOf) });
           return (target.batch as (...a: unknown[]) => unknown).call(target, stmts, ...rest);
         };
       }
@@ -299,6 +316,34 @@ function countingClient(inner: Client): { client: Client; calls: Array<{ via: 'e
   return { client, calls };
 }
 
+/**
+ * [#17612] The job-claim SELECT as the driver EMITS it, with its bound values —
+ * driven from `DbQueueAdapter.claimBatch`'s query shape through a throwaway
+ * remote face, so the statement under test is the driver's own rather than a
+ * second copy of it maintained here.
+ */
+async function capturedJobClaim(): Promise<{ sql: string; args: unknown[] }> {
+  const { client, calls } = countingClient(createClient({ url: 'file::memory:' }));
+  const face = await remoteFace(client);
+  await face.driver.initObjects([fresh(JOB_QUEUE)]);
+  calls.length = 0;
+  await face.driver.find(JOB_QUEUE.name, {
+    where: {
+      queue: 'q',
+      status: 'pending',
+      $or: [{ scheduled_for: null }, { scheduled_for: { $lte: '2026-09-14T00:00:00.000Z' } }],
+    },
+    limit: 30,
+    orderBy: [
+      { field: 'priority', order: 'asc' },
+      { field: 'scheduled_for', order: 'asc' },
+    ],
+  } as never);
+  const select = calls.find((c) => /^SELECT\b/i.test(c.sql[0] ?? ''));
+  if (!select) throw new Error('[#17612] no SELECT reached the client — the capture is void');
+  return { sql: select.sql[0]!, args: select.args[0]! };
+}
+
 const INDEX_DDL = /^\s*CREATE\s+(UNIQUE\s+)?INDEX\b/i;
 const INDEX_NAME_READ = /FROM\s+sqlite_master\s+WHERE\s+type\s*=\s*'index'/i;
 
@@ -432,6 +477,56 @@ describe('[#17609] declared object-level indexes land identically on both TursoD
     expect(remotePlan).not.toMatch(/SCAN sys_notification_delivery/);
     expect(remotePlan).toBe(await plan(local.query));
   });
+
+  it('[#17612] serves the job claim query without sorting the queue — the full ORDER BY sort is gone, both faces', async () => {
+    // The statement is CAPTURED from the driver, never retyped. What comes back
+    // carries one ORDER BY term `DbQueueAdapter.claimBatch` never wrote — `id`,
+    // the deterministic-paging tie-breaker `SqlDriver.orderKeysFor` appends to
+    // every paged read (ADR-0053 D-A1 / objectstack#4363) — and a retyped
+    // literal would have hidden it, along with the residue it leaves below.
+    const claim = await capturedJobClaim();
+    expect(claim.sql).toContain('ORDER BY "priority" ASC, "scheduled_for" ASC, "id" ASC');
+
+    const local = await localFace();
+    const remote = await remoteFace();
+    await local.driver.initObjects([fresh(JOB_QUEUE)]);
+    await remote.driver.initObjects([fresh(JOB_QUEUE)]);
+
+    const plan = async (query: Query) =>
+      (await query(`EXPLAIN QUERY PLAN ${claim.sql}`, claim.args)).map((r) => String(r.detail)).join('\n');
+    const remotePlan = await plan(remote.query);
+    const localPlan = await plan(local.query);
+
+    for (const [face, p] of [['remote', remotePlan], ['local', localPlan]] as const) {
+      expect(p, face).toContain(`SEARCH ${JOB_QUEUE.name} USING INDEX ${JOB_CLAIM_INDEX} (queue=? AND status=?)`);
+      expect(p, face).not.toMatch(new RegExp(`SCAN ${JOB_QUEUE.name}`));
+      // ⛔ The defect's own signature: a sorter over EVERY pending row in the
+      // queue, because the sort's first key was unindexed. Gone on both faces.
+      expect(p, face).not.toMatch(/USE TEMP B-TREE FOR ORDER BY/);
+      // What remains is bounded to rows tying on the whole indexed prefix — the
+      // `id` tie-breaker alone. ⚠️ The two faces WORD it differently because
+      // they run different SQLite builds (`@libsql/client` vs better-sqlite3):
+      // `RIGHT PART OF ORDER BY` remote, `LAST TERM OF ORDER BY` local. Same
+      // plan, two spellings — which is why this pin compares the two faces by
+      // index and by sort CLASS rather than by `EXPLAIN` text equality, the way
+      // the delivery pin above can (its plan has no sorter line to disagree on).
+      expect(p, face).toMatch(/USE TEMP B-TREE FOR (?:RIGHT PART|LAST TERM) OF ORDER BY/);
+    }
+
+    // NEGATIVE CONTROL — the SAME statement against the index set this table
+    // carried before #17612. Both faces must show the FULL sort; if this reads
+    // clean, the assertions above are measuring nothing.
+    const beforeRemote = await remoteFace();
+    const beforeLocal = await localFace();
+    const preFix = { ...fresh(JOB_QUEUE), indexes: JOB_QUEUE_PRE_17612!.map((i) => ({ ...i })) };
+    await beforeRemote.driver.initObjects([preFix]);
+    await beforeLocal.driver.initObjects([{ ...preFix, indexes: JOB_QUEUE_PRE_17612!.map((i) => ({ ...i })) }]);
+    for (const face of [beforeRemote, beforeLocal]) {
+      const p = await plan(face.query);
+      expect(p).toMatch(/USE TEMP B-TREE FOR ORDER BY/);
+      expect(p).not.toContain(JOB_CLAIM_INDEX);
+    }
+  });
 });
 
 describe('[#17609] a declared index the retrofit cannot create is reported at `error`, never forced', () => {

packages/platform-objects/src/audit/sys-job-queue.object.ts

@@ -196,8 +196,45 @@ export const SysJobQueue = ObjectSchema.create({
     updated_at: Field.datetime({ label: 'Updated At', required: false, group: 'System' }),
   },
 
+  // [#17612] The claim path's read order, materialized.
+  //
+  // `DbQueueAdapter.claimBatch` reads the queue as
+  //   WHERE queue = ? AND status = 'pending'
+  //         AND (scheduled_for IS NULL OR scheduled_for <= ?)
+  //   ORDER BY priority ASC, scheduled_for ASC
+  // and `['queue','status','scheduled_for']` served the equality prefix and
+  // then left the planner to sort: the sort's FIRST key, `priority`, appeared
+  // in no declared index at all, so every 1s poll built a temp B-tree over
+  // every pending row in the queue — `USE TEMP B-TREE FOR ORDER BY`, measured
+  // on both Turso faces. Putting `priority` between the prefix and
+  // `scheduled_for` is what removes that sort.
+  //
+  // ⚠️ It does NOT remove the sorter entirely, and the reason is worth knowing
+  // before anyone "finishes the job": a PAGED read carries one more ORDER BY
+  // term the caller never writes — the unique tie-breaker of the
+  // deterministic-paging contract (ADR-0053 D-A1 / objectstack#4363,
+  // `SqlDriver.orderKeysFor`), which here is `id`. So the planner's ORDER BY
+  // ends `…, id ASC` and a bounded per-tie-group sorter remains
+  // (`USE TEMP B-TREE FOR LAST TERM OF ORDER BY`).
+  //
+  // ⛔ Do not close that last term by appending `id` to this index. `id` is an
+  // unbounded `Field.text` — as it is on every platform object here — and
+  // `pnpm check:keyed-text-bounds` refuses a text-family column a declared
+  // index keys on without a `maxLength`, because `driver-sql` emits it TEXT and
+  // MySQL then rejects `ALTER TABLE … ADD INDEX` with
+  // ER_BLOB_KEY_WITHOUT_LENGTH, leaving the object registered with its index
+  // silently absent. Bounding a primary key's column type on a provisioned
+  // table is its own piece of work, not a rider on a sort fix.
+  //
+  // Four columns, and the table still carries three indexes: this REPLACES
+  // `['queue','status','scheduled_for']` rather than joining it. The equality
+  // prefix `queue, status` is unchanged, so `getQueueSize` and `purge` keep the
+  // same seek, and no reader in this repo uses `scheduled_for` as an index
+  // RANGE (the claim's due bound is a residual filter on index rows, and the
+  // retention reaper is keyed on `created_at`). The plan is pinned on both
+  // faces in `driver-turso/src/turso-local-remote-declared-index-parity.test.ts`.
   indexes: [
-    { fields: ['queue', 'status', 'scheduled_for'] },
+    { fields: ['queue', 'status', 'priority', 'scheduled_for'] },
     { fields: ['idempotency_key', 'queue'] },
     { fields: ['status'] },
   ],

packages/plugins/plugin-email/src/email-plugin.attachment-storage.test.ts

@@ -40,15 +40,29 @@ function fakeEngine() {
   const tables = new Map<string, any[]>();
   const hooks: Record<string, Array<(ctx: any) => any>> = {};
   const rowsOf = (t: string) => tables.get(t) ?? [];
-  const matches = (row: any, where: Record<string, any>) =>
+  const matches = (row: any, where: Record<string, any>): boolean =>
     Object.entries(where).every(([k, v]) => {
+      // [#17612] `$or` — the due bound `DbQueueAdapter.claimBatch` now pushes
+      // into SQL. Answered from INSIDE the entries callback so `every` still
+      // ANDs it with the sibling keys; an early `return` here would answer a
+      // NARROWER query than it was handed, which is shape (a) of
+      // `check:where-matcher`'s defect class. Every other `$` key still throws:
+      // that gate's criterion is answer-correctly-or-refuse, never silently wrong.
+      if (k === '$or') return (v as Array<Record<string, any>>).some((leg) => matches(row, leg));
       if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`);
       if (v && typeof v === 'object' && !Array.isArray(v)) {
         return Object.entries(v).every(([op, target]) => {
+          // `$lt`'s existing semantics are left exactly as they were — this
+          // change adds only what #17612 makes reachable.
           if (op === '$lt') return row[k] < (target as any);
+          // NULL-safe like SQL: a row with no value never satisfies `$lte`.
+          if (op === '$lte') return row[k] != null && row[k] <= (target as any);
           throw new Error(`fakeEngine: unsupported operator ${op}`);
         });
       }
+      // `where: { k: null }` is IS NULL — an absent column and an explicit
+      // null are the same absence, which `row[k] === v` could not say.
+      if (v === null) return row[k] == null;
       return row[k] === v;
     });
   const engine = {

packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts

@@ -49,15 +49,29 @@ function fakeEngine(opts: EngineOpts = {}) {
   const tables = new Map<string, any[]>();
   const hooks: Record<string, Array<(ctx: any) => any>> = {};
   const rowsOf = (t: string) => tables.get(t) ?? [];
-  const matches = (row: any, where: Record<string, any>) =>
+  const matches = (row: any, where: Record<string, any>): boolean =>
     Object.entries(where).every(([k, v]) => {
+      // [#17612] `$or` — the due bound `DbQueueAdapter.claimBatch` now pushes
+      // into SQL. Answered from INSIDE the entries callback so `every` still
+      // ANDs it with the sibling keys; an early `return` here would answer a
+      // NARROWER query than it was handed, which is shape (a) of
+      // `check:where-matcher`'s defect class. Every other `$` key still throws:
+      // that gate's criterion is answer-correctly-or-refuse, never silently wrong.
+      if (k === '$or') return (v as Array<Record<string, any>>).some((leg) => matches(row, leg));
       if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`);
       if (v && typeof v === 'object' && !Array.isArray(v)) {
         return Object.entries(v).every(([op, target]) => {
+          // `$lt`'s existing semantics are left exactly as they were — this
+          // change adds only what #17612 makes reachable.
           if (op === '$lt') return row[k] < (target as any);
+          // NULL-safe like SQL: a row with no value never satisfies `$lte`.
+          if (op === '$lte') return row[k] != null && row[k] <= (target as any);
           throw new Error(`fakeEngine: unsupported operator ${op}`);
         });
       }
+      // `where: { k: null }` is IS NULL — an absent column and an explicit
+      // null are the same absence, which `row[k] === v` could not say.
+      if (v === null) return row[k] == null;
       return row[k] === v;
     });
   const engine = {

packages/plugins/plugin-email/src/email-plugin.queue-delivery.test.ts

@@ -49,9 +49,26 @@ function fakeSettings(values: Record<string, Resolved>) {
 function fakeEngine() {
   const tables = new Map<string, any[]>();
   const rowsOf = (t: string) => tables.get(t) ?? [];
-  const matches = (row: any, where: Record<string, any>) =>
+  const matches = (row: any, where: Record<string, any>): boolean =>
     Object.entries(where).every(([k, v]) => {
+      // [#17612] `$or` — the due bound `DbQueueAdapter.claimBatch` now pushes
+      // into SQL. Answered from INSIDE the entries callback so `every` still
+      // ANDs it with the sibling keys; an early `return` here would answer a
+      // NARROWER query than it was handed, which is shape (a) of
+      // `check:where-matcher`'s defect class. Every other `$` key still throws:
+      // that gate's criterion is answer-correctly-or-refuse, never silently wrong.
+      if (k === '$or') return (v as Array<Record<string, any>>).some((leg) => matches(row, leg));
       if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`);
+      if (v && typeof v === 'object' && !Array.isArray(v)) {
+        return Object.entries(v).every(([op, target]) => {
+          // NULL-safe like SQL: a row with no value never satisfies `$lte`.
+          if (op === '$lte') return row[k] != null && row[k] <= (target as any);
+          throw new Error(`fakeEngine: unsupported operator ${op}`);
+        });
+      }
+      // `where: { k: null }` is IS NULL — an absent column and an explicit
+      // null are the same absence, which `row[k] === v` could not say.
+      if (v === null) return row[k] == null;
       return row[k] === v;
     });
   return {