| 26 | } catch (e) { |
26 | } catch (e) { |
| 27 | log.error({ err: (e as Error).message }, "scheduler tick failed"); |
27 | log.error({ err: (e as Error).message }, "scheduler tick failed"); |
| 28 | } |
28 | } |
| 29 | − this.timer = setTimeout(tick, this.inflight >= config.fetchConcurrency ? 500 : 1500); |
29 | + this.timer = setTimeout(tick, this.inflight >= config.fetchConcurrency ? 500 : 800); |
| 30 | }; |
30 | }; |
| 31 | void tick(); |
31 | void tick(); |
| 32 | } |
32 | } |
| 78 | } |
78 | } |
| 79 | const slots = config.fetchConcurrency - this.inflight; |
79 | const slots = config.fetchConcurrency - this.inflight; |
| 80 | if (slots <= 0) return; |
80 | if (slots <= 0) return; |
| 81 | − // Priority-aware claim: critical sensors (priority 0) go first among what is due; within a priority the most overdue first. |
81 | + // Priority-aware claim: critical sensors (priority 0) go first among what is due; within a priority the most |
| |
82 | + // overdue first. We claim more candidates than free slots because many sensors share a few big hosts |
| |
83 | + // (github.com, data.sec.gov…) and the per-host limit would otherwise leave slots idle. |
| 82 | const claimed = await db.execute<Sensor>(sql` |
84 | const claimed = await db.execute<Sensor>(sql` |
| 83 | update sensors set next_check_at = now() + interval '10 minutes' |
85 | update sensors set next_check_at = now() + interval '10 minutes' |
| 84 | − where id in (select id from sensors where enabled and next_check_at <= now() order by priority asc, next_check_at asc limit ${slots} for update skip locked) |
86 | + where id in (select id from sensors where enabled and next_check_at <= now() order by priority asc, next_check_at asc limit ${slots * 4} for update skip locked) |
| 85 | returning *`); |
87 | returning *`); |
| |
88 | + let started = 0; |
| 86 | for (const row of claimed.rows) { |
89 | for (const row of claimed.rows) { |
| 87 | const sensor = normalizeRow(row as unknown as Record<string, unknown>); |
90 | const sensor = normalizeRow(row as unknown as Record<string, unknown>); |
| 88 | const host = safeHost(sensor.url); |
91 | const host = safeHost(sensor.url); |
| 89 | − const hostBusy = (this.perHost.get(host) ?? 0) >= config.perHostConcurrency; |
92 | + const hostLimit = config.perHostOverrides[host] ?? config.perHostConcurrency; |
| |
93 | + const hostBusy = started >= slots || (this.perHost.get(host) ?? 0) >= hostLimit; |
| 90 | if (hostBusy) { |
94 | if (hostBusy) { |
| 91 | − await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 5000) }).where(sql`id = ${sensor.id}`); |
95 | + // give back the lease quickly (5–20 s, jittered so a big host drains smoothly) |
| |
96 | + await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 5000 + Math.random() * 15000) }).where(sql`id = ${sensor.id}`); |
| 92 | continue; |
97 | continue; |
| 93 | } |
98 | } |
| |
99 | + started++; |
| 94 | const trip = this.hostFailures.get(host); |
100 | const trip = this.hostFailures.get(host); |
| 95 | if (trip && trip.until > Date.now()) { |
101 | if (trip && trip.until > Date.now()) { |
| 96 | await db.update(sensors).set({ nextCheckAt: new Date(trip.until + Math.random() * 30_000) }).where(sql`id = ${sensor.id}`); |
102 | await db.update(sensors).set({ nextCheckAt: new Date(trip.until + Math.random() * 30_000) }).where(sql`id = ${sensor.id}`); |
| 97 | |
103 | |