factory: continuous worker pool (10 workers, one seed each) instead of lock-step batches; per-organization wall-clock deadline (120 s); test updated for canada.ca departments

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 11 days ago (Sep 13, 2026) parent 587aaf0

7 changed files +85 −35

modified .env.example +2 −1

		`@@ -46,7 +46,8 @@ WS_MONITORS_PER_OWNER=5`
46	46	# Source Factory (apps/engine/src/factory-main.ts — separate `websensor-factory` process). Seeds come from
47	47	# config/coverage/.yaml (members not yet monitored, or carrying hints) and config/factory/seeds/.yaml.
48	48	WS_FACTORY=1
49		−WS_FACTORY_CONCURRENCY=6 # organizations discovered in parallel
	49	+WS_FACTORY_CONCURRENCY=10 # organizations discovered in parallel (continuous worker pool)
	50	+WS_FACTORY_DEADLINE_MS=120000 # wall-clock deadline per organization
50	51	WS_FACTORY_BUDGET=90 # HTTP requests per organization
51	52	WS_FACTORY_MIN_SCORE=0.35 # candidate score needed to enter shadow monitoring
52	53	WS_FACTORY_SHADOW_MIN_CHECKS=5 # checks before a shadow sensor can be accepted

modified apps/engine/src/config.ts +3 −1

		`@@ -61,9 +61,11 @@ export const log = pino({`
61	61	export const factoryConfig = {
62	62	enabled: (env.WS_FACTORY ?? "1") !== "0",
63	63	/** seeds discovered in parallel */
64		− concurrency: Number(env.WS_FACTORY_CONCURRENCY ?? 6),
	64	+ concurrency: Number(env.WS_FACTORY_CONCURRENCY ?? 10),
65	65	/** HTTP requests per organization during discovery */
66	66	budget: Number(env.WS_FACTORY_BUDGET ?? 90),
	67	+ /** wall-clock deadline per organization (ms) */
	68	+ deadlineMs: Number(env.WS_FACTORY_DEADLINE_MS ?? 120_000),
67	69	/** minimum candidate score to enter shadow monitoring */
68	70	minScore: Number(env.WS_FACTORY_MIN_SCORE ?? 0.35),
69	71	/** shadow evaluation: checks needed, minimum and maximum shadow duration */

modified apps/engine/src/factory-main.ts +50 −19

		`@@ -1,15 +1,16 @@`
1	1	import { closeDb, db, migrate, sql } from "@websensor/db";
2	2	import { closeDispatcher } from "@websensor/connectors";
3	3	import { config, factoryConfig, log } from "./config";
4		−import { evaluateShadows, factoryStats, runFactoryBatch, seedFromCoverage, seedFromFiles } from "./factory";
	4	+import { claimSeeds, evaluateShadows, factoryStats, processSeedSafe, seedFromCoverage, seedFromFiles, type ProcessOutcome } from "./factory";
5	5	import { closeRedis, getRedis } from "./redis";
6	6
7	7	/**
8	8	* Source Factory process (`websensor-factory`): a slow, polite background worker separate from the poller.
9		− * every tick — claim queued seeds (systemic organizations first) and run deep discovery + shadow creation
	9	+ * N workers — each claims one queued seed at a time (systemic organizations first) and runs deep discovery +
	10	+ * shadow creation; a slow organization never stalls the others (per-organization deadline).
10	11	* every 15 min — evaluate shadow sensors (accept / reject / defer)
11	12	* every 24 h — re-seed from the coverage universes (new members become seeds automatically)
12		− * Heartbeat: Redis `ws:factory:status`.
	13	+ * Heartbeat: Redis `ws:factory:status` (processed, queued, in-flight, last outcomes).
13	14	*/
14	15	async function heartbeat(extra: Record<string, unknown>): Promise<void> {
15	16	try {
		`@@ -20,7 +21,7 @@ async function heartbeat(extra: Record<string, unknown>): Promise<void> {`
20	21	}
21	22
22	23	async function main(): Promise<void> {
23		− log.info({ env: config.env, version: config.version, concurrency: factoryConfig.concurrency, budget: factoryConfig.budget }, "websensor factory starting");
	24	+ log.info({ env: config.env, version: config.version, concurrency: factoryConfig.concurrency, budget: factoryConfig.budget, deadlineMs: factoryConfig.deadlineMs }, "websensor factory starting");
24	25	const applied = await migrate(config.databaseUrl);
25	26	if (applied.length) log.info({ applied }, "migrations applied");
26	27	if (!factoryConfig.enabled) {
		`@@ -29,10 +30,17 @@ async function main(): Promise<void> {`
29	30	return;
30	31	}
31	32	let stopping = false;
32		− let busy = false;
33	33	let processed = 0;
	34	+ let inflight = 0;
34	35	let lastSeed = 0;
35	36	let lastEval = 0;
	37	+ let maintenanceBusy = false;
	38	+ const recent: { seed: string; status: string; candidates: number; shadow: number; requests: number; ms: number }[] = [];
	39	+ const note = (o: ProcessOutcome): void => {
	40	+ processed++;
	41	+ recent.unshift({ seed: o.seedId, status: o.status, candidates: o.candidates, shadow: o.shadow, requests: o.requests, ms: o.durationMs });
	42	+ if (recent.length > 12) recent.pop();
	43	+ };
36	44
37	45	const seed = async (): Promise<void> => {
38	46	try {
		`@@ -45,9 +53,31 @@ async function main(): Promise<void> {`
45	53	}
46	54	};
47	55
48		− const tick = async (): Promise<void> => {
49		− if (stopping \|\| busy) return;
50		− busy = true;
	56	+ // Continuous worker pool: each worker claims one seed, processes it, and immediately claims the next.
	57	+ const worker = async (n: number): Promise<void> => {
	58	+ while (!stopping) {
	59	+ let seeds: Awaited<ReturnType<typeof claimSeeds>> = [];
	60	+ try {
	61	+ seeds = await claimSeeds(1);
	62	+ } catch (e) {
	63	+ log.warn({ worker: n, err: (e as Error).message }, "factory: claim failed");
	64	+ }
	65	+ if (!seeds.length) {
	66	+ await new Promise((r) => setTimeout(r, factoryConfig.tickSeconds * 1000 + Math.random() * 2000));
	67	+ continue;
	68	+ }
	69	+ inflight++;
	70	+ try {
	71	+ note(await processSeedSafe(seeds[0]!));
	72	+ } finally {
	73	+ inflight--;
	74	+ }
	75	+ }
	76	+ };
	77	+
	78	+ const maintenance = async (): Promise<void> => {
	79	+ if (stopping \|\| maintenanceBusy) return;
	80	+ maintenanceBusy = true;
51	81	try {
52	82	if (Date.now() - lastSeed > 24 * 3600e3) {
53	83	lastSeed = Date.now();
		`@@ -57,32 +87,32 @@ async function main(): Promise<void> {`
57	87	lastEval = Date.now();
58	88	await evaluateShadows().catch((e) => log.warn({ err: (e as Error).message }, "factory: shadow evaluation failed"));
59	89	}
60		− const out = await runFactoryBatch(factoryConfig.concurrency);
61		− processed += out.length;
62	90	const queued = Number((await db.execute<{ n: string }>(sql`select count(*)::text as n from factory_seeds where status = 'queued'`)).rows[0]?.n ?? 0);
63		− await heartbeat({ processed, queued, lastBatch: out.map((o) => ({ seed: o.seedId, status: o.status, candidates: o.candidates, shadow: o.shadow, requests: o.requests, ms: o.durationMs })), concurrency: factoryConfig.concurrency });
64		− if (!out.length && processed % 50 === 0) {
	91	+ await heartbeat({ processed, queued, inflight, lastBatch: recent, concurrency: factoryConfig.concurrency });
	92	+ if (!inflight && !queued && processed % 50 === 0) {
65	93	const st = await factoryStats();
66	94	log.info({ seeds: st.seeds, candidates: st.candidates, shadow: st.shadow }, "factory: idle");
67	95	}
68	96	} catch (e) {
69		− log.error({ err: (e as Error).stack ?? (e as Error).message }, "factory tick failed");
	97	+ log.error({ err: (e as Error).stack ?? (e as Error).message }, "factory maintenance failed");
70	98	} finally {
71		− busy = false;
	99	+ maintenanceBusy = false;
72	100	}
73	101	};
74	102
75	103	await heartbeat({ starting: true });
76		− void tick();
77		− const timer = setInterval(() => void tick(), factoryConfig.tickSeconds * 1000);
	104	+ await maintenance();
	105	+ const timer = setInterval(() => void maintenance(), 15_000);
	106	+ const workers = Array.from({ length: factoryConfig.concurrency }, (_, i) => worker(i));
78	107
79	108	const shutdown = async (signal: string): Promise<void> => {
80	109	if (stopping) return;
81	110	stopping = true;
82		− log.info({ signal }, "factory shutting down");
	111	+ log.info({ signal, inflight }, "factory shutting down");
83	112	clearInterval(timer);
84		− const deadline = Date.now() + 60_000;
85		− while (busy && Date.now() < deadline) await new Promise((r) => setTimeout(r, 250));
	113	+ const deadline = Date.now() + 90_000;
	114	+ while (inflight > 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 250));
	115	+ // Seeds still marked `discovering` are reclaimed automatically after 2 h (claim query).
86	116	await closeDispatcher();
87	117	await closeRedis();
88	118	await closeDb();
		`@@ -91,6 +121,7 @@ async function main(): Promise<void> {`
91	121	process.on("SIGINT", () => void shutdown("SIGINT"));
92	122	process.on("SIGTERM", () => void shutdown("SIGTERM"));
93	123	process.on("unhandledRejection", (e) => log.error({ err: e instanceof Error ? e.stack : String(e) }, "unhandled rejection"));
	124	+ await Promise.all(workers);
94	125	}
95	126
96	127	main().catch((e) => {

modified apps/engine/src/factory/factory.test.ts +2 −1

		`@@ -12,7 +12,8 @@ describe("registrable domain (coverage key)", () => {`
12	12	expect(registrableDomain("www.ons.gov.uk")).toBe("ons.gov.uk");
13	13	expect(registrableDomain("www.bankofengland.co.uk")).toBe("bankofengland.co.uk");
14	14	expect(registrableDomain("statcan.gc.ca")).toBe("statcan.gc.ca");
15		− expect(registrableDomain("open.canada.ca")).toBe("canada.ca");
	15	+ expect(registrableDomain("open.canada.ca")).toBe("open.canada.ca"); // federal departments are distinct organizations
	16	+ expect(registrableDomain("www.canada.ca")).toBe("canada.ca");
16	17	expect(registrableDomain("boj.or.jp")).toBe("boj.or.jp");
17	18	expect(coverageKey("https://WWW.Example.COM/path")).toBe("example.com");
18	19	});

modified apps/engine/src/factory/run.ts +21 −10

		`@@ -55,6 +55,7 @@ export async function processSeed(seed: FactorySeed): Promise<ProcessOutcome> {`
55	55	const disc = await discoverOrganization(seed.domain, {
56	56	hints,
57	57	budget: factoryConfig.budget,
	58	+ deadlineMs: factoryConfig.deadlineMs,
58	59	concurrency: 4,
59	60	sensorIdForLogs: `factory_${seed.id}`,
60	61	probePages: true,
		`@@ -213,8 +214,8 @@ export async function bumpFactoryDaily(delta: Partial<Record<"seeds_processed" \|`
213	214	on conflict (day) do update set seeds_processed = factory_daily.seeds_processed + ${d.seeds_processed}, requests = factory_daily.requests + ${d.requests}, candidates = factory_daily.candidates + ${d.candidates}, shadow_created = factory_daily.shadow_created + ${d.shadow_created}, accepted = factory_daily.accepted + ${d.accepted}, rejected = factory_daily.rejected + ${d.rejected}, blocked = factory_daily.blocked + ${d.blocked}`).catch(() => undefined);
214	215	}
215	216
216		−/** Claim up to `n` queued seeds (systemic organizations first) and process them with bounded concurrency. */
217		−export async function runFactoryBatch(n = factoryConfig.concurrency, opts: { seedIds?: string[] } = {}): Promise<ProcessOutcome[]> {
	217	+/** Claim up to `n` queued seeds (systemic organizations first). */
	218	+export async function claimSeeds(n: number, opts: { seedIds?: string[] } = {}): Promise<FactorySeed[]> {
218	219	const claimed = await db.execute<Record<string, unknown>>(sql`
219	220	update factory_seeds set status = 'discovering', attempts = attempts + 1, updated_at = now()
220	221	where id in (
		`@@ -222,20 +223,30 @@ export async function runFactoryBatch(n = factoryConfig.concurrency, opts: { see`
222	223	where ${opts.seedIds ? sql`id = any(${sql.raw("array[" + opts.seedIds.map((i) => "'" + i.replace(/'/g, "''") + "'").join(",") + "]::text[]")})` : sql`(status = 'queued' or (status = 'error' and attempts < 3 and updated_at < now() - interval '6 hours') or (status = 'discovering' and updated_at < now() - interval '2 hours'))`}
223	224	order by importance desc, created_at asc limit ${n} for update skip locked)
224	225	returning *`);
225		− const seeds = claimed.rows.map(normalizeSeed);
	226	+ return claimed.rows.map(normalizeSeed);
	227	+}
	228	+
	229	+/** Run a seed with error capture (status → error, retried up to 3 times by the claim query). */
	230	+export async function processSeedSafe(seed: FactorySeed): Promise<ProcessOutcome> {
	231	+ try {
	232	+ return await processSeed(seed);
	233	+ } catch (e) {
	234	+ log.error({ seed: seed.id, err: (e as Error).stack ?? (e as Error).message }, "factory: seed failed");
	235	+ await db.execute(sql`update factory_seeds set status = 'error', last_error = ${(e as Error).message.slice(0, 500)}, updated_at = now() where id = ${seed.id}`).catch(() => undefined);
	236	+ return { seedId: seed.id, sourceId: null, status: "error", candidates: 0, shadow: 0, requests: 0, durationMs: 0, notes: [(e as Error).message] };
	237	+ }
	238	+}
	239	+
	240	+/** One batch: claim `n` seeds and process them with bounded concurrency (CLI / tests; the process uses a continuous pool). */
	241	+export async function runFactoryBatch(n = factoryConfig.concurrency, opts: { seedIds?: string[] } = {}): Promise<ProcessOutcome[]> {
	242	+ const seeds = await claimSeeds(n, opts);
226	243	const out: ProcessOutcome[] = [];
227	244	let i = 0;
228	245	await Promise.all(
229	246	Array.from({ length: Math.min(factoryConfig.concurrency, seeds.length) }, async () => {
230	247	while (i < seeds.length) {
231	248	const seed = seeds[i++]!;
232		− try {
233		− out.push(await processSeed(seed));
234		− } catch (e) {
235		− log.error({ seed: seed.id, err: (e as Error).stack ?? (e as Error).message }, "factory: seed failed");
236		− await db.execute(sql`update factory_seeds set status = 'error', last_error = ${(e as Error).message.slice(0, 500)}, updated_at = now() where id = ${seed.id}`).catch(() => undefined);
237		− out.push({ seedId: seed.id, sourceId: null, status: "error", candidates: 0, shadow: 0, requests: 0, durationMs: 0, notes: [(e as Error).message] });
238		− }
	249	+ out.push(await processSeedSafe(seed));
239	250	}
240	251	}),
241	252	);

modified deploy/websensor.mld.json.example +3 −2

		`@@ -111,7 +111,7 @@`
111	111	"WS_USER_AGENT": "WebSensorBot/0.3 (+https://www.websensor.io/bot; contact@websensor.io)",
112	112	"LOG_LEVEL": "info",
113	113	"WS_FACTORY": "1",
114		− "WS_FACTORY_CONCURRENCY": "6",
	114	+ "WS_FACTORY_CONCURRENCY": "10",
115	115	"WS_FACTORY_BUDGET": "90",
116	116	"WS_FACTORY_MIN_SCORE": "0.35",
117	117	"WS_FACTORY_SHADOW_MIN_CHECKS": "5",
		`@@ -120,7 +120,8 @@`
120	120	"WS_FACTORY_TICK_SECONDS": "20",
121	121	"WS_COVERAGE_DIR": "./config/coverage",
122	122	"WS_EDGAR_USER_AGENT": "WebSensor (contact@websensor.io)",
123		− "WS_APP_NAME": "websensor-factory"
	123	+ "WS_APP_NAME": "websensor-factory",
	124	+ "WS_FACTORY_DEADLINE_MS": "120000"
124	125	},
125	126	"cron_restart": null,
126	127	"autorestart": true,

modified packages/connectors/src/discovery-deep.ts +4 −1

		`@@ -58,6 +58,8 @@ export interface DeepDiscoveryOptions {`
58	58	hints?: DeepDiscoveryHints;
59	59	/** maximum number of HTTP requests for this organization (default 90) */
60	60	budget?: number;
	61	+ /** wall-clock deadline for the whole organization (default 120 s); optional stages are skipped once reached */
	62	+ deadlineMs?: number;
61	63	concurrency?: number;
62	64	sensorIdForLogs?: string;
63	65	probePages?: boolean;
		`@@ -126,7 +128,8 @@ export async function discoverOrganization(domain: string, opts: DeepDiscoveryOp`
126	128	const prev = found.get(k);
127	129	if (!prev \|\| prev.value < c.value) found.set(k, c);
128	130	};
129		− const budgetLeft = (): boolean => requests < budget;
	131	+ const deadline = started + (opts.deadlineMs ?? 120_000);
	132	+ const budgetLeft = (): boolean => requests < budget && Date.now() < deadline;
130	133	const get = async (url: string, o: { timeoutMs?: number; maxBytes?: number; accept?: string; headers?: Record<string, string>; userAgent?: string } = {}) => {
131	134	requests++;
132	135	return httpFetch(id, url, { timeoutMs: o.timeoutMs ?? 15_000, maxBytes: o.maxBytes ?? 3 * 1024 * 1024, accept: o.accept, headers: o.headers, userAgent: o.userAgent });
133	136