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%

fix: serialize bars/quotes/master writes (Postgres deadlocks starved the pool → heal restarts), exempt loopback SSR from rate limit, calmer PRICE_CHANGE/DIVERGENCE thresholds, lock_timeout

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

10 changed files +65 −18

modified apps/api/src/api/server.ts +2 −0
@@ -44,6 +44,8 @@ export async function buildServer(): Promise<FastifyInstance> {
44 44 (req as FastifyRequest & { startedAt: number }).startedAt = Date.now();
45 45 if (req.url.startsWith("/v1/stream") || req.url.startsWith("/v1/sse")) return;
46 46 const ip = req.ip;
47 + // Server-side renders of the web app arrive from loopback without a client IP — never throttle them.
48 + if (ip === "127.0.0.1" || ip === "::1" || ip === "::ffff:127.0.0.1") return;
47 49 const now = Date.now();
48 50 let b = buckets.get(ip);
49 51 if (!b) {
modified apps/api/src/core/bars.ts +16 −2
@@ -3,6 +3,7 @@ import { CANONICAL_VERSIONS, RESOLUTION_MS, floorTo } from "@market-atlas/market
3 3 import { pool } from "../db/pool.js";
4 4 import { logger } from "../logger.js";
5 5 import { telemetry } from "./telemetry.js";
6 +import { barsMutex } from "./mutex.js";
6 7
7 8 /**
8 9 * Live 1-minute bar aggregation from canonical prices; higher resolutions are rolled up in SQL
@@ -56,8 +57,16 @@ export class BarAggregator {
56 57 }
57 58 }
58 59
59 −export async function upsertBars(bars: Bar[]): Promise<void> {
60 − if (!bars.length) return;
60 +export async function upsertBars(input: Bar[]): Promise<void> {
61 + if (!input.length) return;
62 + // One row per key (a late tick can re-open an already flushed minute) and a deterministic lock order.
63 + const byKey = new Map<string, Bar>();
64 + for (const b of input) byKey.set(`${b.instrumentId}|${b.resolution}|${b.ts}`, b);
65 + const bars = [...byKey.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId) || a.resolution.localeCompare(b.resolution) || a.ts - b.ts);
66 + await barsMutex.run(() => upsertSorted(bars));
67 +}
68 +
69 +async function upsertSorted(bars: Bar[]): Promise<void> {
61 70 const cols = 11;
62 71 for (let i = 0; i < bars.length; i += 1000) {
63 72 const chunk = bars.slice(i, i + 1000);
@@ -85,6 +94,10 @@ export async function upsertBars(bars: Bar[]): Promise<void> {
85 94
86 95 /** Roll 1m bars into 5m/15m/1h and 1m→1d (UTC days) for the last `hours` hours. */
87 96 export async function rollupBars(hours = 3): Promise<void> {
97 + await barsMutex.run(() => rollupInner(hours));
98 +}
99 +
100 +async function rollupInner(hours: number): Promise<void> {
88 101 const targets: Array<[BarResolution, string]> = [
89 102 ["5m", "5 minutes"],
90 103 ["15m", "15 minutes"],
@@ -97,6 +110,7 @@ export async function rollupBars(hours = 3): Promise<void> {
97 110 select instrument_id, $1, bucket, (array_agg(open order by ts))[1], max(high), min(low), (array_agg(close order by ts desc))[1], sum(volume), max(source_count), 'consensus', $3
98 111 from (select *, date_bin($2::interval, ts, timestamptz '2000-01-01') as bucket from bars where resolution = '1m' and producer = 'consensus' and ts > now() - ($4 || ' hours')::interval) b
99 112 group by instrument_id, bucket
113 + order by instrument_id, bucket
100 114 on conflict (instrument_id, resolution, ts) do update set open = excluded.open, high = excluded.high, low = excluded.low, close = excluded.close, volume = excluded.volume, source_count = excluded.source_count
101 115 where bars.producer = 'consensus'`,
102 116 [res, interval, CANONICAL_VERSIONS.consensus, String(hours)],
modified apps/api/src/core/events.ts +4 −4
@@ -75,10 +75,10 @@ export class EventEngine {
75 75 }
76 76 // Price change events on a per-instrument adaptive threshold (≥ 0.5% and ≥ 8× typical tick move).
77 77 const ref = s.lastEventPrice ?? s.lastPrice;
78 − if (ref != null && ref !== 0 && s.ticks > 5) {
78 + if (ref != null && ref !== 0 && s.ticks > 60) {
79 79 const move = (p - ref) / ref;
80 − const threshold = Math.max(0.005, 8 * s.ewmaAbsRet);
81 − if (Math.abs(move) >= threshold && this.cooldown(s, "PRICE_CHANGE", now, 60_000)) {
80 + const threshold = Math.max(0.01, 12 * s.ewmaAbsRet);
81 + if (Math.abs(move) >= threshold && this.cooldown(s, "PRICE_CHANGE", now, 5 * 60_000)) {
82 82 this.emitDerived("PRICE_CHANGE", q, name, now, Math.abs(move) >= 0.03 ? "WARNING" : "INFO", `${q.symbol} ${move >= 0 ? "+" : ""}${(move * 100).toFixed(2)}%`, {
83 83 from: ref,
84 84 to: p,
@@ -100,7 +100,7 @@ export class EventEngine {
100 100 if (p > s.sessionHigh) s.sessionHigh = p;
101 101 if (p < s.sessionLow) s.sessionLow = p;
102 102 // Source divergence: included sources disagree by more than 50 bps (crypto venues legitimately differ; flag ≥ 50).
103 − if (q.dispersionBps != null && q.dispersionBps > 50 && q.sourceCount >= 2 && now - s.lastSourceDivergenceAt > 10 * 60_000) {
103 + if (q.dispersionBps != null && q.dispersionBps > 100 && q.sourceCount >= 2 && now - s.lastSourceDivergenceAt > 60 * 60_000) {
104 104 s.lastSourceDivergenceAt = now;
105 105 this.emitDerived("SOURCE_DIVERGENCE", q, name, now, "NOTICE", `${q.symbol}: sources diverge by ${q.dispersionBps.toFixed(0)} bps`, {
106 106 dispersionBps: q.dispersionBps,
modified apps/api/src/core/instruments.ts +11 −8
@@ -3,6 +3,7 @@ import { canonicalSymbol, makeCompanyId, makeInstrumentId } from "@market-atlas/
3 3 import { pool, query, type Queryable } from "../db/pool.js";
4 4 import { logger } from "../logger.js";
5 5 import { telemetry } from "./telemetry.js";
6 +import { masterMutex } from "./mutex.js";
6 7
7 8 interface Row {
8 9 id: string;
@@ -155,7 +156,7 @@ export class InstrumentStore {
155 156 }
156 157
157 158 async upsert(i: Instrument, client: Queryable = pool): Promise<void> {
158 − await query(
159 + await masterMutex.run(() => query(
159 160 `insert into instruments (id, symbol, name, asset_class, exchange_id, mic, currency, country, company_id, isin, security_type, base, quote, is_active, metadata)
160 161 values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
161 162 on conflict (id) do update set
@@ -172,7 +173,7 @@ export class InstrumentStore {
172 173 updated_at = now()`,
173 174 [i.id, i.symbol, i.name, i.assetClass, i.exchangeId, i.mic, i.currency, i.country, i.companyId, i.isin, i.securityType, i.base, i.quote, i.isActive, JSON.stringify(i.metadata ?? {})],
174 175 client,
175 − );
176 + ));
176 177 const merged = { ...(this.byId.get(i.id) ?? i), ...stripNulls(i) } as Instrument;
177 178 this.index(merged);
178 179 telemetry.gauge("instruments_total", this.byId.size);
@@ -182,10 +183,12 @@ export class InstrumentStore {
182 183 const key = `${sourceId}|${canonicalSymbol(alias)}`;
183 184 if (this.byAlias.get(key) === instrumentId) return;
184 185 this.byAlias.set(key, instrumentId);
185 − await query(
186 − `insert into symbol_aliases (alias, source_id, instrument_id) values ($1,$2,$3)
187 − on conflict (alias, source_id) do update set instrument_id = excluded.instrument_id`,
188 − [alias, sourceId, instrumentId],
186 + await masterMutex.run(() =>
187 + query(
188 + `insert into symbol_aliases (alias, source_id, instrument_id) values ($1,$2,$3)
189 + on conflict (alias, source_id) do update set instrument_id = excluded.instrument_id`,
190 + [alias, sourceId, instrumentId],
191 + ),
189 192 );
190 193 }
191 194
@@ -201,12 +204,12 @@ export class InstrumentStore {
201 204
202 205 async upsertCompany(c: { name: string; cik: string | null; country: string | null; sector?: string | null; website?: string | null }): Promise<string> {
203 206 const id = makeCompanyId(c.name);
204 − await query(
207 + await masterMutex.run(() => query(
205 208 `insert into companies (id, name, cik, country, sector, website) values ($1,$2,$3,$4,$5,$6)
206 209 on conflict (id) do update set cik = coalesce(excluded.cik, companies.cik), country = coalesce(excluded.country, companies.country),
207 210 sector = coalesce(excluded.sector, companies.sector), website = coalesce(excluded.website, companies.website), updated_at = now()`,
208 211 [id, c.name, c.cik, c.country, c.sector ?? null, c.website ?? null],
209 − );
212 + ));
210 213 return id;
211 214 }
212 215
added apps/api/src/core/mutex.ts +18 −0
@@ -0,0 +1,18 @@
1 +/** Minimal async mutex: serializes critical sections (multi-row upserts) so Postgres never sees two overlapping lock orders. */
2 +export class Mutex {
3 + private tail: Promise<void> = Promise.resolve();
4 +
5 + run<T>(fn: () => Promise<T>): Promise<T> {
6 + const next = this.tail.then(fn, fn);
7 + this.tail = next.then(
8 + () => undefined,
9 + () => undefined,
10 + );
11 + return next;
12 + }
13 +}
14 +
15 +/** One writer at a time for each hot table family. */
16 +export const barsMutex = new Mutex();
17 +export const quotesMutex = new Mutex();
18 +export const masterMutex = new Mutex(); // instruments, companies, symbol_aliases
modified apps/api/src/core/quotes.ts +9 −1
@@ -12,6 +12,7 @@ export class QuoteStore {
12 12 private quotes = new Map<string, CanonicalQuote>();
13 13 private dirty = new Map<string, CanonicalQuote>();
14 14 private timer: NodeJS.Timeout | null = null;
15 + private flushing = false;
15 16
16 17 async load(): Promise<void> {
17 18 const r = await pool.query("select * from canonical_quotes");
@@ -41,8 +42,13 @@ export class QuoteStore {
41 42
42 43 async flush(): Promise<void> {
43 44 this.timer = null;
45 + if (this.flushing) {
46 + this.timer = setTimeout(() => void this.flush(), 500);
47 + return;
48 + }
44 49 if (!this.dirty.size) return;
45 − const batch = [...this.dirty.values()];
50 + this.flushing = true;
51 + const batch = [...this.dirty.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId));
46 52 this.dirty.clear();
47 53 const cols = 25;
48 54 const tuples: string[] = [];
@@ -73,6 +79,8 @@ export class QuoteStore {
73 79 logger.error({ err, size: batch.length }, "canonical quote upsert failed");
74 80 for (const q of batch) if (!this.dirty.has(q.instrumentId)) this.dirty.set(q.instrumentId, q);
75 81 this.timer = setTimeout(() => void this.flush(), 5000);
82 + } finally {
83 + this.flushing = false;
76 84 }
77 85 }
78 86 }
modified apps/api/src/db/pool.ts +2 −1
@@ -9,9 +9,10 @@ types.setTypeParser(1700, (v) => Number(v));
9 9
10 10 export const pool = new Pool({
11 11 connectionString: config.databaseUrl,
12 − max: config.role === "api" ? 8 : 12,
12 + max: config.role === "api" ? 8 : 16,
13 13 idleTimeoutMillis: 30_000,
14 14 statement_timeout: 60_000,
15 + options: "-c lock_timeout=15000", // never let a lock wait starve the pool (health checks must keep answering)
15 16 application_name: `market-atlas-${config.role}`,
16 17 });
17 18
modified apps/api/src/main.ts +1 −1
@@ -53,7 +53,7 @@ async function main() {
53 53 if (shuttingDown) return;
54 54 shuttingDown = true;
55 55 logger.info({ signal }, "shutting down");
56 − const force = setTimeout(() => process.exit(1), 20_000);
56 + const force = setTimeout(() => process.exit(1), 10_000);
57 57 try {
58 58 scheduler.stop();
59 59 await connectorManager.stopAll();
modified deploy/market-atlas.mld.json +1 −0
@@ -65,6 +65,7 @@
65 65 "MA_LOG_JSON": "1",
66 66 "MA_RAW_SAMPLE_RATE": "0.02",
67 67 "MA_OBSERVATION_RETENTION_DAYS": "45",
68 + "MA_TICK_PERSIST_INTERVAL_MS": "5000",
68 69 "PG_DUMP": "/opt/homebrew/opt/postgresql@17/bin/pg_dump",
69 70 "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin"
70 71 },
modified docs/ARCHITECTURE.md +1 −1
@@ -58,7 +58,7 @@ consensus freshness, MARKET_OPEN/CLOSE events, session resets and the world map.
58 58
59 59 Derived from canonical quotes with per-instrument adaptive baselines (EWMA of tick moves): PRICE_CHANGE (≥ max(0.5 %, 8× typical
60 60 tick)), SESSION_HIGH/LOW (after history, ≥ 0.1 % improvement, cooldown), VOLATILITY_SPIKE (30-tick realized vol ≥ 4× baseline),
61 −SOURCE_DIVERGENCE (> 50 bps between included sources). Connector-proposed events (halts, filings, document changes) and system events
61 +SOURCE_DIVERGENCE (> 100 bps between included sources, hourly per instrument). Connector-proposed events (halts, filings, document changes) and system events
62 62 (SOURCE_FAILURE, SCHEMA_DRIFT, MARKET_OPEN/CLOSE) go through the same dedupe (fingerprint, 24 h) and confirmation counting.
63 63
64 64 ## Resilience
65 65