spb/market-atlas
Public
TypeScript 96.7%
SQL 1.6%
CSS 0.8%
JavaScript 0.5%
1import type { ConnectorHealth, ConnectorState } from "@market-atlas/market-model";2import { pool } from "../db/pool.js";3import { logger } from "../logger.js";4import { bus } from "./bus.js";5import { Reservoir, telemetry } from "./telemetry.js";67interface Tracker {8 connectorId: string;9 sourceId: string;10 state: ConnectorState;11 startedAt: number | null;12 lastMessageAt: number | null;13 lastSuccessAt: number | null;14 lastErrorAt: number | null;15 lastError: string | null;16 messagesTotal: number;17 errorsTotal: number;18 parseOk: number;19 parseFail: number;20 reconnects: number;21 latency: Reservoir;22 instruments: Set<string>;23 schemaFingerprints: Map<string, Set<string>>;24 nextPollAt: number | null;25 /** Rolling availability samples (1 per minute, 1 = healthy). */26 availability: number[];27}2829/**30 * Operational health and reliability scoring per connector/source. The score measures the31 * connector's operational behaviour (availability, latency, parse success, stability) — never32 * financial truth.33 */34export class HealthEngine {35 private trackers = new Map<string, Tracker>();36 private sourceOfConnector = new Map<string, string>();3738 register(connectorId: string, sourceId: string): void {39 this.sourceOfConnector.set(connectorId, sourceId);40 if (!this.trackers.has(connectorId)) {41 this.trackers.set(connectorId, {42 connectorId,43 sourceId,44 state: "STARTING",45 startedAt: null,46 lastMessageAt: null,47 lastSuccessAt: null,48 lastErrorAt: null,49 lastError: null,50 messagesTotal: 0,51 errorsTotal: 0,52 parseOk: 0,53 parseFail: 0,54 reconnects: 0,55 latency: new Reservoir(256),56 instruments: new Set(),57 schemaFingerprints: new Map(),58 nextPollAt: null,59 availability: [],60 });61 }62 }6364 private t(id: string): Tracker {65 const t = this.trackers.get(id);66 if (!t) throw new Error(`unknown connector ${id}`);67 return t;68 }6970 setState(id: string, state: ConnectorState) {71 const t = this.t(id);72 if (t.state === state) return;73 const prev = t.state;74 t.state = state;75 if (state === "HEALTHY" && !t.startedAt) t.startedAt = Date.now();76 telemetry.gauge("connector_state", stateNum(state), { connector: id });77 bus.publish("connector.health", this.snapshot(id));78 logger.info({ connector: id, from: prev, to: state }, "connector state");79 }8081 state(id: string): ConnectorState {82 return this.trackers.get(id)?.state ?? "DISABLED";83 }8485 message(id: string, latencyMs: number | null, instrumentIds: string[] = []) {86 const t = this.t(id);87 t.messagesTotal++;88 t.lastMessageAt = Date.now();89 t.lastSuccessAt = t.lastMessageAt;90 if (latencyMs != null && latencyMs >= 0 && latencyMs < 3_600_000) t.latency.add(latencyMs);91 for (const i of instrumentIds) t.instruments.add(i);92 telemetry.inc("connector_messages_total", 1, { connector: id });93 telemetry.inc("observations_received_total");94 }9596 parse(id: string, ok: boolean) {97 const t = this.t(id);98 if (ok) t.parseOk++;99 else t.parseFail++;100 }101102 error(id: string, err: unknown) {103 const t = this.t(id);104 t.errorsTotal++;105 t.lastErrorAt = Date.now();106 t.lastError = (err instanceof Error ? err.message : String(err)).slice(0, 500);107 telemetry.inc("connector_errors_total", 1, { connector: id });108 }109110 reconnect(id: string) {111 this.t(id).reconnects++;112 }113114 poll(id: string, nextAt: number | null) {115 this.t(id).nextPollAt = nextAt;116 }117118 recordFingerprint(id: string, kind: string, fp: string): "known" | "new" | "first" {119 const t = this.t(id);120 let set = t.schemaFingerprints.get(kind);121 if (!set) {122 set = new Set();123 t.schemaFingerprints.set(kind, set);124 }125 if (set.has(fp)) return "known";126 const first = set.size === 0;127 set.add(fp);128 return first ? "first" : "new";129 }130131 /** 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 }149150 /** 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>();152153 loadFingerprints(id: string, fps: Record<string, string[]>) {154 const t = this.t(id);155 for (const [kind, arr] of Object.entries(fps)) t.schemaFingerprints.set(kind, new Set(arr));156 }157158 fingerprints(id: string): Record<string, string[]> {159 const t = this.t(id);160 return Object.fromEntries([...t.schemaFingerprints].map(([k, v]) => [k, [...v]]));161 }162163 /** 0–100 operational reliability. */164 score(id: string): number | null {165 const t = this.trackers.get(id);166 if (!t || t.messagesTotal === 0) return null;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;168 const parseTotal = t.parseOk + t.parseFail;169 const parse = parseTotal ? t.parseOk / parseTotal : 1;170 const p95 = t.latency.quantile(0.95);171 const latencyScore = p95 == null ? 0.8 : p95 < 1000 ? 1 : p95 < 5000 ? 0.85 : p95 < 60_000 ? 0.7 : 0.5;172 const stability = Math.max(0.4, 1 - t.reconnects / Math.max(50, t.messagesTotal / 100));173 const errorRate = t.errorsTotal / Math.max(1, t.messagesTotal + t.errorsTotal);174 const errScore = Math.max(0, 1 - errorRate * 10);175 return Math.round(100 * (0.4 * avail + 0.2 * parse + 0.15 * latencyScore + 0.15 * stability + 0.1 * errScore));176 }177178 /** Reliability 0..1 for the consensus engine, keyed by source id (best connector of the source). */179 sourceReliability(sourceId: string): number {180 const scores: number[] = [];181 for (const t of this.trackers.values()) {182 if (t.sourceId !== sourceId) continue;183 const s = this.score(t.connectorId);184 if (s != null) scores.push(s / 100);185 }186 return scores.length ? Math.max(...scores) : 0.6;187 }188189 snapshot(id: string): ConnectorHealth {190 const t = this.t(id);191 const parseTotal = t.parseOk + t.parseFail;192 return {193 connectorId: id,194 state: t.state,195 lastMessageAt: t.lastMessageAt,196 lastSuccessAt: t.lastSuccessAt,197 lastErrorAt: t.lastErrorAt,198 lastError: t.lastError,199 messages1m: telemetry.lastMinute("connector_messages_total", { connector: id }),200 messagesTotal: t.messagesTotal,201 errorsTotal: t.errorsTotal,202 reconnects: t.reconnects,203 medianLatencyMs: t.latency.quantile(0.5),204 p95LatencyMs: t.latency.quantile(0.95),205 parseSuccessRate: parseTotal ? t.parseOk / parseTotal : null,206 instrumentsCovered: Math.max(t.instruments.size, this.coverageFloor.get(id) ?? 0),207 schemaFingerprints: [...t.schemaFingerprints.values()].flatMap((s) => [...s]),208 reliabilityScore: this.score(id),209 startedAt: t.startedAt,210 nextPollAt: t.nextPollAt,211 };212 }213214 all(): ConnectorHealth[] {215 return [...this.trackers.keys()].map((id) => this.snapshot(id));216 }217218 /** Called every minute: staleness detection, availability sample, DB snapshot. */219 async tick(staleAfterMs: (id: string) => number): Promise<void> {220 const now = Date.now();221 const rows: unknown[][] = [];222 for (const t of this.trackers.values()) {223 if (t.state === "HEALTHY" && t.lastMessageAt && now - t.lastMessageAt > staleAfterMs(t.connectorId)) this.setState(t.connectorId, "STALE");224 const healthy = t.state === "HEALTHY" ? 1 : t.state === "DEGRADED" || t.state === "STALE" ? 0.5 : t.state === "PAUSED" || t.state === "DISABLED" ? NaN : 0;225 if (!Number.isNaN(healthy)) {226 t.availability.push(healthy);227 if (t.availability.length > 24 * 60) t.availability.shift();228 }229 const s = this.snapshot(t.connectorId);230 rows.push([new Date(now).toISOString(), t.connectorId, t.state, s.messages1m, telemetry.lastMinute("connector_errors_total", { connector: t.connectorId }), s.medianLatencyMs, s.p95LatencyMs, s.parseSuccessRate, s.instrumentsCovered, t.reconnects, s.reliabilityScore]);231 }232 if (!rows.length) return;233 const cols = 11;234 const tuples = rows.map((_, i) => `(${Array.from({ length: cols }, (_, j) => `$${i * cols + j + 1}`).join(",")})`);235 try {236 await pool.query(237 `insert into connector_health (ts, connector_id, status, messages_1m, errors_1m, median_latency_ms, p95_latency_ms, parse_success_rate, instruments_covered, reconnects, reliability_score) values ${tuples.join(",")} on conflict do nothing`,238 rows.flat(),239 );240 for (const t of this.trackers.values()) {241 await pool.query(242 `update connectors set status = $2, last_message_at = $3, last_success_at = $4, last_error_at = $5, last_error = $6, messages_total = $7, errors_total = $8, reconnects = $9, reliability_score = $10, schema_fingerprints = $11, updated_at = now() where id = $1`,243 [t.connectorId, t.state, iso(t.lastMessageAt), iso(t.lastSuccessAt), iso(t.lastErrorAt), t.lastError, t.messagesTotal, t.errorsTotal, t.reconnects, this.score(t.connectorId), JSON.stringify(this.fingerprints(t.connectorId))],244 );245 }246 } catch (err) {247 logger.error({ err }, "health snapshot failed");248 }249 }250}251252const iso = (ms: number | null) => (ms == null ? null : new Date(ms).toISOString());253const stateNum = (s: ConnectorState) => ({ HEALTHY: 1, DEGRADED: 2, STALE: 3, RECONNECTING: 4, FAILED: 5, PAUSED: 6, DISABLED: 7, STARTING: 0 })[s];254255export const health = new HealthEngine();256