SPB Git forge
spboucher.ai ↗

spb/market-atlas

Public
12commits 1branches 0releases
1.1 MBsize
maindefault branch
10 days agolast push
TypeScript 96.7% SQL 1.6% CSS 0.8% JavaScript 0.5%

connectors: quick retry after failed poll, HEALTHY on first payload, hydrate counters/coverage from DB at boot, longer network backoff

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed 11 days ago (Sep 13, 2026) parent c633aac

4 changed files +33 −7

modified apps/api/src/core/connector-manager.ts +4 −2
@@ -267,7 +267,8 @@ export class ConnectorManager {
267 267 } catch (err) {
268 268 rt.consecutiveFailures++;
269 269 health.error(id, err);
270 − const wait = Math.max(this.nextInterval(rt), backoffMs(rt.consecutiveFailures, 5000, 30 * 60_000));
270 + // Retry sooner than the regular cadence (a daily connector must not wait 24 h after a transient network error).
271 + const wait = Math.max(5000, Math.min(this.nextInterval(rt), backoffMs(rt.consecutiveFailures, 15_000, 30 * 60_000)));
271 272 logger.warn({ connector: id, err: err instanceof Error ? err.message : String(err), failures: rt.consecutiveFailures, retryInMs: wait }, "poll failed");
272 273 if (rt.consecutiveFailures >= 3) this.markFailed(rt, err instanceof Error ? err.message : String(err));
273 274 else health.setState(id, "DEGRADED");
@@ -321,7 +322,8 @@ export class ConnectorManager {
321 322 }
322 323 if (produced > 0) rt.driftStrikes = Math.max(0, rt.driftStrikes - 1);
323 324 const rawRef = rawArchive.store(raw, forceArchive || rt.def.metadata.sourceType !== "WEBSOCKET");
324 − if (health.state(id) === "STALE" || health.state(id) === "DEGRADED") health.setState(id, "HEALTHY");
325 + const st = health.state(id);
326 + if (st === "STALE" || st === "DEGRADED" || st === "STARTING" || st === "RECONNECTING") health.setState(id, "HEALTHY");
325 327 return pipeline.process(rt.def, raw, batch, rawRef);
326 328 }
327 329
modified apps/api/src/core/health.ts +24 −2
@@ -128,6 +128,28 @@ export class HealthEngine {
128 128 return first ? "first" : "new";
129 129 }
130 130
131 + /** Restore lifetime counters and 24 h instrument coverage from the database so restarts do not zero the directory. */
132 + async hydrate(): Promise<void> {
133 + const ids = [...this.trackers.keys()];
134 + if (!ids.length) return;
135 + const [rows, cov] = await Promise.all([
136 + pool.query<{ id: string; messages_total: number; errors_total: number; reconnects: number }>("select id, messages_total, errors_total, reconnects from connectors where id = any($1)", [ids]),
137 + pool.query<{ connector_id: string; n: number }>("select connector_id, count(distinct instrument_id)::int as n from observations where received_at > now() - interval '24 hours' group by 1"),
138 + ]);
139 + for (const r of rows.rows) {
140 + const t = this.trackers.get(r.id);
141 + if (!t) continue;
142 + t.messagesTotal = Math.max(t.messagesTotal, Number(r.messages_total) || 0);
143 + t.errorsTotal = Math.max(t.errorsTotal, Number(r.errors_total) || 0);
144 + t.reconnects = Math.max(t.reconnects, Number(r.reconnects) || 0);
145 + }
146 + this.coverageFloor.clear();
147 + for (const c of cov.rows) this.coverageFloor.set(c.connector_id, c.n);
148 + }
149 +
150 + /** Coverage observed in the DB over the last 24 h (floor for the live set, which starts empty at boot). */
151 + private coverageFloor = new Map<string, number>();
152 +
131 153 loadFingerprints(id: string, fps: Record<string, string[]>) {
132 154 const t = this.t(id);
133 155 for (const [kind, arr] of Object.entries(fps)) t.schemaFingerprints.set(kind, new Set(arr));
@@ -142,7 +164,7 @@ export class HealthEngine {
142 164 score(id: string): number | null {
143 165 const t = this.trackers.get(id);
144 166 if (!t || t.messagesTotal === 0) return null;
145 − const avail = t.availability.length ? t.availability.reduce((a, b) => a + b, 0) / t.availability.length : t.state === "HEALTHY" ? 1 : 0.5;
167 + const avail = t.availability.length ? t.availability.reduce((a, b) => a + b, 0) / t.availability.length : t.state === "HEALTHY" || t.state === "STARTING" ? 1 : 0.5;
146 168 const parseTotal = t.parseOk + t.parseFail;
147 169 const parse = parseTotal ? t.parseOk / parseTotal : 1;
148 170 const p95 = t.latency.quantile(0.95);
@@ -181,7 +203,7 @@ export class HealthEngine {
181 203 medianLatencyMs: t.latency.quantile(0.5),
182 204 p95LatencyMs: t.latency.quantile(0.95),
183 205 parseSuccessRate: parseTotal ? t.parseOk / parseTotal : null,
184 − instrumentsCovered: t.instruments.size,
206 + instrumentsCovered: Math.max(t.instruments.size, this.coverageFloor.get(id) ?? 0),
185 207 schemaFingerprints: [...t.schemaFingerprints.values()].flatMap((s) => [...s]),
186 208 reliabilityScore: this.score(id),
187 209 startedAt: t.startedAt,
modified apps/api/src/main.ts +2 −0
@@ -7,6 +7,7 @@ import { instruments } from "./core/instruments.js";
7 7 import { calendar } from "./core/calendar.js";
8 8 import { quoteStore } from "./core/quotes.js";
9 9 import { eventEngine } from "./core/events.js";
10 +import { health } from "./core/health.js";
10 11 import { syncRegistry } from "./core/registry.js";
11 12 import { connectorManager } from "./core/connector-manager.js";
12 13 import { scheduler } from "./core/jobs.js";
@@ -37,6 +38,7 @@ async function main() {
37 38
38 39 if (runsWorker()) {
39 40 await syncRegistry();
41 + await health.hydrate();
40 42 scheduler.start();
41 43 await connectorManager.startAll();
42 44 }
modified packages/connector-sdk/src/http.ts +3 −3
@@ -119,12 +119,12 @@ export class HttpClient {
119 119 return { status: res.status, ok: true, notModified: false, headers: resHeaders, text, url, durationMs, fromCache: false };
120 120 } catch (err) {
121 121 if (err instanceof HttpError) throw err;
122 − if (attempt < retries) {
122 + if (attempt < Math.max(retries, 3)) {
123 123 attempt++;
124 − await sleep(backoffMs(attempt, 500, 15_000));
124 + await sleep(1000 + backoffMs(attempt, 1500, 20_000)); // DNS/TLS hiccups at boot need more than a few hundred ms
125 125 continue;
126 126 }
127 − const msg = err instanceof Error ? err.message : String(err);
127 + const msg = err instanceof Error ? `${err.message}${(err as { cause?: { code?: string } }).cause?.code ? ` [${(err as { cause?: { code?: string } }).cause?.code}]` : ""}` : String(err);
128 128 throw new HttpError(`request failed: ${msg} (${redactUrl(url)})`, 0, redactUrl(url));
129 129 } finally {
130 130 clearTimeout(timer);
131 131