Skip to content

Commit f81bc80

Browse files
committed
test(service-queue,driver-turso): pin the claim plan on both faces and the due-set LIMIT
- driver-turso parity: the job-claim SELECT is CAPTURED from the driver rather than retyped, so the `id` tie-breaker the deterministic-paging contract appends is visible in the pin; the plan must carry no sorter on either face, with the pre-#17612 index set as the negative control. - service-queue: the starvation reproduction (30 future-dated priority-1 rows + 1 due priority-100 row) and its 29-row control, a NULL `scheduled_for` leg, and a pin that `priority` still orders the due set. - both fake engines in service-queue learn `$or`/`$lte`/`IS NULL`; every other `$` key stays a loud failure. Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude <noreply@anthropic.com>
1 parent 5aa11cd commit f81bc80

3 files changed

Lines changed: 218 additions & 12 deletions

File tree

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

Lines changed: 92 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,9 @@ const JOB_QUEUE: ObjectDef = {
124124
updated_at: Field.datetime({ label: 'Updated At', required: false }),
125125
},
126126
indexes: [
127-
{ fields: ['queue', 'status', 'scheduled_for'] },
127+
// [#17612] Widened from `['queue','status','scheduled_for']` — verbatim
128+
// from `sys-job-queue.object.ts`, which is where the reasoning lives.
129+
{ fields: ['queue', 'status', 'priority', 'scheduled_for', 'id'] },
128130
{ fields: ['idempotency_key', 'queue'] },
129131
{ fields: ['status'] },
130132
],
@@ -168,7 +170,7 @@ const EXPECTED_NAMES: Record<string, string[]> = {
168170
`sqlite_autoindex_${DELIVERY.name}_1`,
169171
].sort(),
170172
[JOB_QUEUE.name]: [
171-
buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'scheduled_for'], false),
173+
buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'priority', 'scheduled_for', 'id'], false),
172174
buildIndexName(JOB_QUEUE.name, ['idempotency_key', 'queue'], false),
173175
buildIndexName(JOB_QUEUE.name, ['status'], false),
174176
`sqlite_autoindex_${JOB_QUEUE.name}_1`,
@@ -185,6 +187,18 @@ const EXPECTED_NAMES: Record<string, string[]> = {
185187
};
186188

187189
const CLAIM_INDEX = buildIndexName(DELIVERY.name, ['status', 'partition_key', 'next_attempt_at'], false);
190+
/** [#17612] The index `sys_job_queue`'s claim path must be served by, end to end. */
191+
const JOB_CLAIM_INDEX = buildIndexName(
192+
JOB_QUEUE.name,
193+
['queue', 'status', 'priority', 'scheduled_for', 'id'],
194+
false,
195+
);
196+
/** The shape the remote face provisioned BEFORE #17612 — this file's negative control. */
197+
const JOB_QUEUE_PRE_17612: ObjectDef['indexes'] = [
198+
{ fields: ['queue', 'status', 'scheduled_for'] },
199+
{ fields: ['idempotency_key', 'queue'] },
200+
{ fields: ['status'] },
201+
];
188202
const DEDUP_INDEX = buildIndexName(DELIVERY.name, ['notification_id', 'recipient_id', 'channel'], true);
189203

190204
/** A fresh copy per sync — neither face may see a definition the other one mutated. */
@@ -275,20 +289,27 @@ async function remoteFace(client: Client = createClient({ url: 'file::memory:' }
275289
}
276290

277291
/** Every statement the driver hands the client, in order, with the call that carried it. */
278-
function countingClient(inner: Client): { client: Client; calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> } {
279-
const calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> = [];
292+
function countingClient(inner: Client): {
293+
client: Client;
294+
calls: Array<{ via: 'execute' | 'batch'; sql: string[]; args: unknown[][] }>;
295+
} {
296+
const calls: Array<{ via: 'execute' | 'batch'; sql: string[]; args: unknown[][] }> = [];
280297
const sqlOf = (s: InStatement) => (typeof s === 'string' ? s : s.sql);
298+
// [#17612] The bound values, alongside the text — a captured statement is only
299+
// re-runnable (`EXPLAIN QUERY PLAN`) if its placeholders can be filled again.
300+
const argsOf = (s: InStatement): unknown[] =>
301+
typeof s === 'string' ? [] : Array.isArray(s.args) ? [...s.args] : [];
281302
const client = new Proxy(inner, {
282303
get(target, prop) {
283304
if (prop === 'execute') {
284305
return (stmt: InStatement, ...rest: unknown[]) => {
285-
calls.push({ via: 'execute', sql: [sqlOf(stmt)] });
306+
calls.push({ via: 'execute', sql: [sqlOf(stmt)], args: [argsOf(stmt)] });
286307
return (target.execute as (...a: unknown[]) => unknown).call(target, stmt, ...rest);
287308
};
288309
}
289310
if (prop === 'batch') {
290311
return (stmts: InStatement[], ...rest: unknown[]) => {
291-
calls.push({ via: 'batch', sql: stmts.map(sqlOf) });
312+
calls.push({ via: 'batch', sql: stmts.map(sqlOf), args: stmts.map(argsOf) });
292313
return (target.batch as (...a: unknown[]) => unknown).call(target, stmts, ...rest);
293314
};
294315
}
@@ -299,6 +320,34 @@ function countingClient(inner: Client): { client: Client; calls: Array<{ via: 'e
299320
return { client, calls };
300321
}
301322

323+
/**
324+
* [#17612] The job-claim SELECT as the driver EMITS it, with its bound values —
325+
* driven from `DbQueueAdapter.claimBatch`'s query shape through a throwaway
326+
* remote face, so the statement under test is the driver's own rather than a
327+
* second copy of it maintained here.
328+
*/
329+
async function capturedJobClaim(): Promise<{ sql: string; args: unknown[] }> {
330+
const { client, calls } = countingClient(createClient({ url: 'file::memory:' }));
331+
const face = await remoteFace(client);
332+
await face.driver.initObjects([fresh(JOB_QUEUE)]);
333+
calls.length = 0;
334+
await face.driver.find(JOB_QUEUE.name, {
335+
where: {
336+
queue: 'q',
337+
status: 'pending',
338+
$or: [{ scheduled_for: null }, { scheduled_for: { $lte: '2026-09-14T00:00:00.000Z' } }],
339+
},
340+
limit: 30,
341+
orderBy: [
342+
{ field: 'priority', order: 'asc' },
343+
{ field: 'scheduled_for', order: 'asc' },
344+
],
345+
} as never);
346+
const select = calls.find((c) => /^SELECT\b/i.test(c.sql[0] ?? ''));
347+
if (!select) throw new Error('[#17612] no SELECT reached the client — the capture is void');
348+
return { sql: select.sql[0]!, args: select.args[0]! };
349+
}
350+
302351
const INDEX_DDL = /^\s*CREATE\s+(UNIQUE\s+)?INDEX\b/i;
303352
const INDEX_NAME_READ = /FROM\s+sqlite_master\s+WHERE\s+type\s*=\s*'index'/i;
304353

@@ -432,6 +481,43 @@ describe('[#17609] declared object-level indexes land identically on both TursoD
432481
expect(remotePlan).not.toMatch(/SCAN sys_notification_delivery/);
433482
expect(remotePlan).toBe(await plan(local.query));
434483
});
484+
485+
it('[#17612] serves the job claim query with NO sorter — every ORDER BY term indexed, both faces', async () => {
486+
// The statement is CAPTURED from the driver, never retyped: the ORDER BY
487+
// the planner sees carries one term `DbQueueAdapter.claimBatch` never
488+
// wrote — `id`, the deterministic-paging tie-breaker `SqlDriver.orderKeysFor`
489+
// appends to every paged read (ADR-0053 D-A1 / objectstack#4363) — and that
490+
// term is exactly why an index stopping at `scheduled_for` still left a
491+
// sorter in the plan. A retyped literal would have hidden it.
492+
const claim = await capturedJobClaim();
493+
expect(claim.sql).toContain('ORDER BY "priority" ASC, "scheduled_for" ASC, "id" ASC');
494+
495+
const local = await localFace();
496+
const remote = await remoteFace();
497+
await local.driver.initObjects([fresh(JOB_QUEUE)]);
498+
await remote.driver.initObjects([fresh(JOB_QUEUE)]);
499+
500+
const plan = async (query: Query) =>
501+
(await query(`EXPLAIN QUERY PLAN ${claim.sql}`, claim.args)).map((r) => String(r.detail)).join('\n');
502+
const remotePlan = await plan(remote.query);
503+
504+
expect(remotePlan).toContain(`SEARCH ${JOB_QUEUE.name} USING INDEX ${JOB_CLAIM_INDEX} (queue=? AND status=?)`);
505+
// ⛔ No sorter at all — not "a cheaper sorter". The whole ORDER BY is index order.
506+
expect(remotePlan).not.toMatch(/USE TEMP B-TREE/);
507+
expect(remotePlan).not.toMatch(new RegExp(`SCAN ${JOB_QUEUE.name}`));
508+
expect(remotePlan).toBe(await plan(local.query));
509+
510+
// NEGATIVE CONTROL — the SAME statement against the index set this table
511+
// carried before #17612. If this reads clean, the assertions above are
512+
// measuring nothing.
513+
const before = await remoteFace();
514+
await before.driver.initObjects([{ ...fresh(JOB_QUEUE), indexes: JOB_QUEUE_PRE_17612!.map((i) => ({ ...i })) }]);
515+
const beforePlan = (await before.query(`EXPLAIN QUERY PLAN ${claim.sql}`, claim.args))
516+
.map((r) => String(r.detail))
517+
.join('\n');
518+
expect(beforePlan).toMatch(/USE TEMP B-TREE FOR ORDER BY/);
519+
expect(beforePlan).not.toContain(JOB_CLAIM_INDEX);
520+
});
435521
});
436522

437523
describe('[#17609] a declared index the retrofit cannot create is reported at `error`, never forced', () => {

‎packages/services/service-queue/src/db-queue-adapter.test.ts‎

Lines changed: 110 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -346,9 +346,11 @@ describe('makeFakeEngine().delete conforms to ObjectQL.delete (#4550)', () => {
346346
// `where: { id: { $in: […] } }` only LOOKS like an id: it is a multi-row
347347
// predicate, so without `multi` the engine rejects it — the exact case a
348348
// hand-mirrored `if (opts?.where?.id == null)` waves through, and the reason
349-
// the mirror had to go rather than be corrected in place. (This fake's
350-
// `matches` is equality-only by design — no fixture here sends an operator
351-
// predicate — so what is pinned is the dispatch verdict, not `$in` matching.)
349+
// the mirror had to go rather than be corrected in place. (What is pinned
350+
// here is the DISPATCH verdict, not `$in` matching: the refusal happens in
351+
// `assertEngineDeleteDispatch` before `matches` is ever reached, so it
352+
// holds whether or not this fake understands the operator — and since
353+
// #17612 it does understand a few, `$in` among them.)
352354
engine.tables.set('sys_job_queue', [{ id: 'a' }, { id: 'b' }, { id: 'c' }]);
353355
await expect(
354356
engine.delete('sys_job_queue', { where: { id: { $in: ['a', 'b'] } } }),
@@ -365,3 +367,108 @@ describe('makeFakeEngine().delete conforms to ObjectQL.delete (#4550)', () => {
365367
expect(rows()).toHaveLength(3);
366368
});
367369
});
370+
371+
/**
372+
* [#17612] Head-of-line starvation: the due bound was applied in JS AFTER the
373+
* SQL `LIMIT`, so a window full of not-yet-due high-priority rows hid every
374+
* already-due row behind them — indefinitely, not just for one tick.
375+
*
376+
* The numbers below are the reproduction the card carries, re-measured here.
377+
* Default `batchSize` is 10, so the candidate window is `10 * 3 = 30`.
378+
*/
379+
describe('DbQueueAdapter — the due bound is a predicate, not a post-LIMIT filter', () => {
380+
let engine: ReturnType<typeof makeFakeEngine>;
381+
let adapter: DbQueueAdapter;
382+
383+
beforeEach(() => {
384+
engine = makeFakeEngine();
385+
adapter = new DbQueueAdapter({
386+
engine,
387+
options: { pollIntervalMs: 60_000, autoStart: false, defaultMaxAttempts: 3 },
388+
});
389+
});
390+
391+
/** `n` not-yet-due priority-1 rows, then one already-due priority-100 row. */
392+
async function seedStarvation(n: number): Promise<string> {
393+
for (let i = 0; i < n; i++) {
394+
await adapter.publish('starve', { i }, { priority: 1, delay: 600_000 });
395+
}
396+
return adapter.publish('starve', { due: true }, { priority: 100 });
397+
}
398+
399+
it('claims the due low-priority row even when the whole candidate window is future-dated work', async () => {
400+
const handled: any[] = [];
401+
await adapter.subscribe('starve', async (msg) => { handled.push(msg.data); });
402+
const dueId = await seedStarvation(30);
403+
404+
// Was 0 — the 30 priority-1 rows filled the window and the due row was
405+
// never a candidate. The queue looked healthy and drained nothing.
406+
expect(await adapter.pollOnce()).toBe(1);
407+
expect(handled).toEqual([{ due: true }]);
408+
expect((engine.tables.get('sys_job_queue') ?? []).find((r: any) => r.id === dueId).status)
409+
.toBe('completed');
410+
});
411+
412+
it('CONTROL — one row short of filling the window, the same claim already worked before the fix', async () => {
413+
const handled: any[] = [];
414+
await adapter.subscribe('starve', async (msg) => { handled.push(msg.data); });
415+
await seedStarvation(29);
416+
417+
// 29 future-dated rows + the due one is exactly 30 = the window, so the due
418+
// row rode in on the last slot. This case read 1 before the fix too — which
419+
// is what makes the case above a measurement rather than a coincidence.
420+
expect(await adapter.pollOnce()).toBe(1);
421+
expect(handled).toEqual([{ due: true }]);
422+
});
423+
424+
it('LIMIT now sees the DUE set — the engine hands back one candidate, not a window of thirty', async () => {
425+
const claims: Array<{ where: any; limit: number; rows: number }> = [];
426+
const inner = engine.find.bind(engine);
427+
engine.find = async (table: string, opts: any = {}) => {
428+
const rows = await inner(table, opts);
429+
if (opts?.orderBy?.[0]?.field === 'priority') {
430+
claims.push({ where: opts.where, limit: opts.limit, rows: rows.length });
431+
}
432+
return rows;
433+
};
434+
435+
await adapter.subscribe('starve', async () => {});
436+
await seedStarvation(30);
437+
await adapter.pollOnce();
438+
439+
expect(claims).toHaveLength(1);
440+
expect(claims[0].limit).toBe(30);
441+
// The over-fetch window is unchanged; what changed is WHICH rows fill it.
442+
expect(claims[0].rows).toBe(1);
443+
expect(claims[0].where.$or).toEqual([
444+
{ scheduled_for: null },
445+
{ scheduled_for: { $lte: expect.any(String) } },
446+
]);
447+
});
448+
449+
it('a row with no `scheduled_for` at all is due — `NULL <= now` is NULL, so it needs its own leg', async () => {
450+
const handled: any[] = [];
451+
await adapter.subscribe('nulls', async (msg) => { handled.push(msg.data); });
452+
await adapter.publish('nulls', { a: 1 });
453+
// Not a shape `publish` writes — `sys_job_queue.scheduled_for` is nullable
454+
// and a row can reach this table from a migration or a hand-written insert.
455+
(engine.tables.get('sys_job_queue') ?? [])[0].scheduled_for = null;
456+
457+
expect(await adapter.pollOnce()).toBe(1);
458+
expect(handled).toEqual([{ a: 1 }]);
459+
});
460+
461+
it('priority still decides the order among DUE rows — the fork this card did NOT take', async () => {
462+
const handled: any[] = [];
463+
await adapter.subscribe('ordered', async (msg) => { handled.push(msg.data); });
464+
await adapter.publish('ordered', { p: 100 }, { priority: 100 });
465+
await adapter.publish('ordered', { p: 1 }, { priority: 1 });
466+
await adapter.publish('ordered', { p: 50 }, { priority: 50 });
467+
468+
expect(await adapter.pollOnce()).toBe(3);
469+
// Lower number = higher priority (`sys_job_queue.priority`'s own description).
470+
// Dropping `priority` from the claim sort — the card's option 2 — would have
471+
// made this read in insert order and left the declared field inert.
472+
expect(handled).toEqual([{ p: 1 }, { p: 50 }, { p: 100 }]);
473+
});
474+
});

‎packages/services/service-queue/src/job-queue-retention.test.ts‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,20 +45,33 @@ type Row = Record<string, any>;
4545
* (same contract as `db-queue-adapter.test.ts`);
4646
* - the Reaper's `delete(table, { where, multi: true })` with a `$lt`
4747
* operator, which the row-by-row `where.id` delete of the adapter's own
48-
* `purge()` does not exercise.
48+
* `purge()` does not exercise;
49+
* - [#17612] the claim path's `$or` due bound, which `pollOnce()` carries in
50+
* here whether or not a fixture is about it.
4951
*/
5052
function makeFakeEngine() {
5153
const tables = new Map<string, Row[]>();
5254
function matches(row: Row, where: Record<string, any>): boolean {
5355
for (const [k, v] of Object.entries(where)) {
56+
// [#17612] `$or` — the claim path's due bound reaches this fake too, via
57+
// the `pollOnce()` these retention fixtures drive. Every OTHER `$` key
58+
// stays a loud failure: a double that silently passed an operator it
59+
// does not implement is the shape this package refuses.
60+
if (k === '$or') {
61+
if (!(v as Array<Record<string, any>>).some((leg) => matches(row, leg))) return false;
62+
continue;
63+
}
5464
if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`);
5565
if (v && typeof v === 'object' && !Array.isArray(v)) {
56-
// NULL-safe like SQL: a row with no value never satisfies `$lt`.
66+
// NULL-safe like SQL: a row with no value never satisfies `$lt`/`$lte`.
5767
if ('$lt' in v && (row[k] == null || !(String(row[k]) < String(v.$lt)))) return false;
68+
if ('$lte' in v && (row[k] == null || !(String(row[k]) <= String(v.$lte)))) return false;
5869
if ('$in' in v && !(v.$in as unknown[]).includes(row[k])) return false;
5970
continue;
6071
}
61-
if (row[k] !== v) return false;
72+
// `where: { k: null }` is IS NULL — an absent column and an explicit null
73+
// are the same absence.
74+
if (v === null ? row[k] != null : row[k] !== v) return false;
6275
}
6376
return true;
6477
}

0 commit comments

Comments
 (0)