workers: atomic batch claiming (SKIP LOCKED) and concurrent resolvers

Simon-Pierre Boucher committed 17 days ago (Sep 7, 2026) parent f9a634e

2 changed files +23 −10

modified workers/entity-resolution/index.ts +16 −7

		`@@ -23,12 +23,21 @@ export interface ResolveResult {`
23	23	export async function resolveBatch(opts: { limit?: number; connectorId?: string } = {}): Promise<ResolveResult> {
24	24	const log = logger.child({ component: 'resolver' });
25	25	const limit = opts.limit ?? 500;
26		− const rows = await db()
27		− .select()
28		− .from(normalizedRecords)
29		− .where(and(eq(normalizedRecords.status, 'pending'), opts.connectorId ? eq(normalizedRecords.connectorId, opts.connectorId) : undefined))
30		− .orderBy(sql`case ${normalizedRecords.kind} when 'catalog_item' then 0 when 'population_report' then 1 when 'sale' then 2 when 'price_observation' then 3 when 'listing' then 4 else 5 end`, normalizedRecords.createdAt)
31		− .limit(limit);
	26	+ // Atomically claim a batch (status pending → processing) so several resolver workers can run
	27	+ // concurrently without processing the same rows; rows are released back to pending on crash.
	28	+ const connectorFilter = opts.connectorId ? sql` and connector_id = ${opts.connectorId}` : sql``;
	29	+ const rows = (await db().execute(sql`
	30	+ with claimed as (
	31	+ select id from normalized_records
	32	+ where status = 'pending'${connectorFilter}
	33	+ order by case kind when 'catalog_item' then 0 when 'population_report' then 1 when 'sale' then 2 when 'price_observation' then 3 when 'listing' then 4 else 5 end, created_at
	34	+ limit ${limit}
	35	+ for update skip locked
	36	+ )
	37	+ update normalized_records n set status = 'processing'
	38	+ from claimed where n.id = claimed.id
	39	+ returning n.id, n.raw_record_id as "rawRecordId", n.connector_id as "connectorId", n.source_id as "sourceId", n.kind, n.payload, n.seq, n.asset_id as "assetId", n.variant_id as "variantId", n.match_method as "matchMethod", n.match_confidence as "matchConfidence", n.status, n.reject_reason as "rejectReason", n.target_id as "targetId", n.created_at as "createdAt", n.processed_at as "processedAt"
	40	+ `)) as unknown as Array<typeof normalizedRecords.$inferSelect>;
32	41	const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] };
33	42	if (rows.length === 0) return res;
34	43	const touched = new Set<string>();
		`@@ -63,7 +72,7 @@ export async function resolveBatch(opts: { limit?: number; connectorId?: string`
63	72	res.rejected++;
64	73	updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `${err.message} (no rate after sync)`, processedAt: new Date() } });
65	74	} else {
66		− updates.push({ id: row.id, set: { rejectReason: err.message } }); // stays pending until fx sync
	75	+ updates.push({ id: row.id, set: { status: 'pending', rejectReason: err.message } }); // back to pending until fx sync
67	76	}
68	77	continue;
69	78	}

modified workers/main.ts +7 −3

		`@@ -9,6 +9,7 @@ import { flushCosts } from './lib/costs.ts';`
9	9	import { runCrawl } from './crawler/run.ts';
10	10	import { scheduleDueCrawls } from './crawler/scheduler.ts';
11	11	import { normalizeBatch } from './normalizer/index.ts';
	12	+import { sql } from 'drizzle-orm';
12	13	import { pendingCount, resolveBatch } from './entity-resolution/index.ts';
13	14	import { assetsNeedingValuation, computePremiums, valueMany } from './valuation/run.ts';
14	15	import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';
		`@@ -39,16 +40,19 @@ export async function startWorker(): Promise<() => Promise<void>> {`
39	40	total += r.processed;
40	41	if (r.processed < 500) break;
41	42	}
42		− if (total > 0) await queue.send(JOBS.resolveBatch, {}, { singletonKey: 'resolve', singletonSeconds: 30 });
	43	+ if (total > 0) for (let k = 0; k < Number(process.env.RESOLVE_CONCURRENCY ?? 3); k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30 });
43	44	});
44		− await queue.work<Record<string, never>>(JOBS.resolveBatch, { concurrency: 1, pollingIntervalSeconds: 5 }, async () => {
	45	+ const RESOLVE_WORKERS = Number(process.env.RESOLVE_CONCURRENCY ?? 3);
	46	+ // Release rows claimed by a previous worker process that died mid-batch.
	47	+ await db().execute(sql`update normalized_records set status = 'pending' where status = 'processing'`);
	48	+ await queue.work<Record<string, never>>(JOBS.resolveBatch, { concurrency: RESOLVE_WORKERS, pollingIntervalSeconds: 5 }, async () => {
45	49	const touched = new Set<string>();
46	50	for (let i = 0; i < 40; i++) {
47	51	const r = await resolveBatch({ limit: 500 });
48	52	for (const a of r.touchedAssets) touched.add(a);
49	53	if (r.processed < 500) break;
50	54	}
51		− if ((await pendingCount()) > 0) await queue.send(JOBS.resolveBatch, {}, { singletonKey: 'resolve', singletonSeconds: 30, startAfterSeconds: 10 });
	55	+ if ((await pendingCount()) > 0) for (let k = 0; k < RESOLVE_WORKERS; k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30, startAfterSeconds: 5 });
52	56	if (touched.size) await queue.send(JOBS.valuationAsset, { assetIds: [...touched].slice(0, 5000) }, { singletonKey: `value:${Date.now()}` });
53	57	});
54	58	await queue.work<{ assetIds: string[] }>(JOBS.valuationAsset, { concurrency: 1, pollingIntervalSeconds: 5 }, async (data) => {
55	59