From 3a817ab8de1276b16ba3d5bd396dcab79209f346 Mon Sep 17 00:00:00 2001 From: MAC Date: Fri, 26 Jun 2026 01:47:20 +0100 Subject: [PATCH] fix(api): reconnect OraclePriceBroadcaster on channel drop + surface its health [BUG-103] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OraclePriceBroadcaster's Supabase Realtime .subscribe() status callback logged CHANNEL_ERROR/TIMED_OUT/CLOSED but never re-subscribed — Realtime does not retry on its own once it reports one of these terminal statuses. Once the channel dropped for any of these reasons, live WS price broadcasts went silently and permanently dead until a process restart, with zero visibility: the instance was a local const in index.ts, never passed anywhere, and /health had no reference to it at all. Added reconnect-with-backoff (1s base, doubling, capped at 30s, reset on a successful SUBSCRIBED) directly in the status callback, and getStatus()/ isHealthy() so the current connection state is queryable. Converted the class to a singleton-via-getter (getOraclePriceBroadcaster()) so health.ts can reach the exact instance index.ts starts — index.ts currently registers routes (including healthRoutes()) before it creates the broadcaster, so a direct reference couldn't have been threaded through without reordering index.ts's whole startup sequence. Wired a new `priceBroadcaster` check into /health's checks object, alongside the existing rpc/db/ws checks — a dead channel now makes /health report "degraded" instead of silently "ok" while no live prices flow. Added tests covering: backoff scheduling and reset, stop() cancelling any pending reconnect, the singleton returning the same instance, and the new /health check (including a thrown error from the broadcaster not crashing the health endpoint). Verified all new assertions fail against the pre-fix code (missing getStatus/isHealthy/getOraclePriceBroadcaster entirely, and the health check defaulting to a different shape) and pass against the fix. Co-authored-by: Claude Sonnet 4.6 --- src/index.ts | 5 +- src/routes/health.ts | 14 +- src/services/OraclePriceBroadcaster.ts | 80 +++++++- tests/routes/health.test.ts | 44 +++++ .../services/oracle-price-broadcaster.test.ts | 171 ++++++++++++++++++ 5 files changed, 309 insertions(+), 5 deletions(-) create mode 100644 tests/services/oracle-price-broadcaster.test.ts diff --git a/src/index.ts b/src/index.ts index bc78b89..9060938 100644 --- a/src/index.ts +++ b/src/index.ts @@ -26,7 +26,7 @@ import { candleRoutes } from "./routes/candles.js"; import { docsRoutes } from "./routes/docs.js"; import { adlRoutes } from "./routes/adl.js"; import { setupWebSocket, cleanupPriceUpdateTimers, cleanupEventBusListeners } from "./routes/ws.js"; -import { OraclePriceBroadcaster } from "./services/OraclePriceBroadcaster.js"; +import { getOraclePriceBroadcaster } from "./services/OraclePriceBroadcaster.js"; import { readRateLimit, writeRateLimit } from "./middleware/rate-limit.js"; import { ipBlocklist } from "./middleware/ip-blocklist.js"; import { cacheMiddleware } from "./middleware/cache.js"; @@ -329,8 +329,7 @@ const wss = setupWebSocket(server as unknown as import("node:http").Server); // Bridge oracle_prices INSERTs → local eventBus → WS clients. Without this // the cross-process price.updated events from the indexer never reach WS // subscribers, and the frontend only sees new prices on page refresh. -const oraclePriceBroadcaster = new OraclePriceBroadcaster(); -oraclePriceBroadcaster.start().catch((err) => { +getOraclePriceBroadcaster().start().catch((err) => { logger.error("OraclePriceBroadcaster start failed", { error: err instanceof Error ? err.message : String(err), }); diff --git a/src/routes/health.ts b/src/routes/health.ts index 92050f3..228e55d 100644 --- a/src/routes/health.ts +++ b/src/routes/health.ts @@ -5,6 +5,7 @@ import { withRpcFallback } from "../utils/rpc-fallback.js"; import { HEALTH_RPC_TIMEOUT_MS } from "../utils/rpc-timeout.js"; import { getWebSocketMetrics } from "./ws.js"; import { requireApiKey } from "../middleware/auth.js"; +import { getOraclePriceBroadcaster } from "../services/OraclePriceBroadcaster.js"; const logger = createLogger("api:health"); const startTime = Date.now(); @@ -24,7 +25,8 @@ export function healthRoutes(): Hono { if (cachedHealth && Date.now() - cachedHealth.checkedAt < HEALTH_CACHE_TTL_MS) { return c.json(cachedHealth.body, cachedHealth.statusCode as 200 | 503); } - const checks: { db: boolean; rpc: boolean; ws: boolean } = { db: false, rpc: false, ws: false }; + const checks: { db: boolean; rpc: boolean; ws: boolean; priceBroadcaster: boolean } = + { db: false, rpc: false, ws: false, priceBroadcaster: false }; let status: "ok" | "degraded" | "down" = "ok"; // Check RPC connectivity @@ -63,6 +65,16 @@ export function healthRoutes(): Hono { checks.ws = false; } + // Check the oracle-price Realtime bridge. Without this, the channel can + // drop (CHANNEL_ERROR/TIMED_OUT/CLOSED) and live WS price broadcasts go + // silently dead — this check is what makes that visible instead of + // reporting "ok" while no live prices are flowing. + try { + checks.priceBroadcaster = getOraclePriceBroadcaster().isHealthy(); + } catch { + checks.priceBroadcaster = false; + } + // Determine overall status const failedChecks = Object.values(checks).filter(v => !v).length; if (failedChecks === 0) { diff --git a/src/services/OraclePriceBroadcaster.ts b/src/services/OraclePriceBroadcaster.ts index 54c138b..9915239 100644 --- a/src/services/OraclePriceBroadcaster.ts +++ b/src/services/OraclePriceBroadcaster.ts @@ -37,12 +37,65 @@ interface OraclePriceRow { network: string; } +/** Statuses Supabase Realtime's .subscribe() callback can report, plus the + * two states this class itself tracks before a subscription has ever been + * attempted or after stop() has been called. */ +export type BroadcasterStatus = + | "not_started" + | "JOINING" + | "SUBSCRIBED" + | "CHANNEL_ERROR" + | "TIMED_OUT" + | "CLOSED" + | "stopped"; + +const RECONNECT_BASE_DELAY_MS = 1_000; +const RECONNECT_MAX_DELAY_MS = 30_000; + export class OraclePriceBroadcaster { private channel: RealtimeChannel | null = null; private started = false; + private stopped = false; + private lastStatus: BroadcasterStatus = "not_started"; + private reconnectTimer: ReturnType | null = null; + private reconnectAttempt = 0; + + /** Current Realtime subscription status — used by /health (see health.ts). */ + getStatus(): BroadcasterStatus { + return this.lastStatus; + } + + /** True only while genuinely subscribed and receiving live updates. */ + isHealthy(): boolean { + return this.lastStatus === "SUBSCRIBED"; + } + + private scheduleReconnect(network: string): void { + if (this.stopped || this.reconnectTimer) return; + const delayMs = Math.min( + RECONNECT_MAX_DELAY_MS, + RECONNECT_BASE_DELAY_MS * 2 ** this.reconnectAttempt, + ); + this.reconnectAttempt++; + logger.warn("oracle-price broadcaster scheduling reconnect", { + network, + delayMs, + attempt: this.reconnectAttempt, + }); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.started = false; // allow start() to actually run again + this.start().catch((err) => { + logger.error("oracle-price broadcaster reconnect attempt failed", { + error: err instanceof Error ? err.message : String(err), + }); + }); + }, delayMs); + this.reconnectTimer.unref?.(); + } async start(): Promise { - if (this.started) return; + if (this.started || this.stopped) return; this.started = true; const network = getNetwork(); @@ -90,12 +143,19 @@ export class OraclePriceBroadcaster { // Log every status transition so we can see where we are if a // SUBSCRIBED never lands. Supabase Realtime emits: CHANNEL_ERROR, // TIMED_OUT, CLOSED, SUBSCRIBED — plus occasional JOINING. + this.lastStatus = status as BroadcasterStatus; const fields: Record = { status, network }; if (err) fields.error = err instanceof Error ? err.message : String(err); if (status === "SUBSCRIBED") { logger.info("oracle-price broadcaster subscribed", fields); + this.reconnectAttempt = 0; // backoff resets once a connection actually succeeds } else if (status === "CHANNEL_ERROR" || status === "TIMED_OUT" || status === "CLOSED") { logger.error("oracle-price broadcaster channel problem", fields); + // Without this, the channel stays dead until process restart — + // Realtime does not retry on its own once it reports one of + // these terminal statuses. + this.started = false; + this.scheduleReconnect(network); } else { logger.info("oracle-price broadcaster status", fields); } @@ -105,10 +165,16 @@ export class OraclePriceBroadcaster { error: err instanceof Error ? err.message : String(err), }); this.started = false; + this.scheduleReconnect(network); } } async stop(): Promise { + this.stopped = true; + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } if (this.channel) { try { await getSupabase().removeChannel(this.channel); @@ -118,5 +184,17 @@ export class OraclePriceBroadcaster { this.channel = null; } this.started = false; + this.lastStatus = "stopped"; } } + +// Singleton accessor so health.ts can read the same instance index.ts +// starts, regardless of module import/call order — index.ts currently +// registers routes (including healthRoutes(), which reads this) before it +// creates and starts the broadcaster. +let _instance: OraclePriceBroadcaster | null = null; + +export function getOraclePriceBroadcaster(): OraclePriceBroadcaster { + if (!_instance) _instance = new OraclePriceBroadcaster(); + return _instance; +} diff --git a/tests/routes/health.test.ts b/tests/routes/health.test.ts index 6e08651..f62a89b 100644 --- a/tests/routes/health.test.ts +++ b/tests/routes/health.test.ts @@ -9,6 +9,14 @@ vi.mock("../../src/routes/ws.js", () => ({ })), })); +// Mock the oracle-price broadcaster singleton — defaults to healthy so +// existing tests' "ok"/"degraded" expectations (written before this check +// existed) don't need touching beyond the "down" test below, which must +// fail every check including this one to keep asserting 503. +vi.mock("../../src/services/OraclePriceBroadcaster.js", () => ({ + getOraclePriceBroadcaster: vi.fn(() => ({ isHealthy: vi.fn(() => true) })), +})); + // Mock @percolator/shared vi.mock("@percolator/shared", () => ({ getSupabase: vi.fn(), @@ -32,6 +40,7 @@ vi.mock("@percolator/shared", () => ({ const { getConnection, getSupabase } = await import("@percolator/shared"); const { getWebSocketMetrics } = await import("../../src/routes/ws.js"); +const { getOraclePriceBroadcaster } = await import("../../src/services/OraclePriceBroadcaster.js"); describe("health routes", () => { let mockConnection: any; @@ -102,6 +111,7 @@ describe("health routes", () => { mockConnection.getSlot.mockRejectedValue(new Error("RPC error")); mockSupabase.select.mockRejectedValue(new Error("DB error")); vi.mocked(getWebSocketMetrics).mockImplementation(() => { throw new Error("WS unavailable"); }); + vi.mocked(getOraclePriceBroadcaster).mockReturnValue({ isHealthy: () => false } as any); const app = healthRoutes(); const res = await app.request("/health"); @@ -112,6 +122,40 @@ describe("health routes", () => { expect(data.checks.rpc).toBe(false); expect(data.checks.db).toBe(false); expect(data.checks.ws).toBe(false); + expect(data.checks.priceBroadcaster).toBe(false); + }); + + describe("priceBroadcaster check (BUG-103)", () => { + it("reports degraded (not down) when only the oracle-price broadcaster is unhealthy", async () => { + mockConnection.getSlot.mockResolvedValue(123456789); + mockSupabase.select.mockResolvedValue({ count: 5, error: null }); + vi.mocked(getOraclePriceBroadcaster).mockReturnValue({ isHealthy: () => false } as any); + + const app = healthRoutes(); + const res = await app.request("/health"); + + expect(res.status).toBe(200); + const data = await res.json(); + expect(data.status).toBe("degraded"); + expect(data.checks.priceBroadcaster).toBe(false); + expect(data.checks.rpc).toBe(true); + expect(data.checks.db).toBe(true); + }); + + it("treats a thrown error from the broadcaster as unhealthy rather than crashing the health check", async () => { + mockConnection.getSlot.mockResolvedValue(123456789); + mockSupabase.select.mockResolvedValue({ count: 5, error: null }); + vi.mocked(getOraclePriceBroadcaster).mockImplementation(() => { + throw new Error("broadcaster unavailable"); + }); + + const app = healthRoutes(); + const res = await app.request("/health"); + + expect(res.status).toBe(200); + const data = await res.json(); + expect(data.checks.priceBroadcaster).toBe(false); + }); }); it("should include uptime in response", async () => { diff --git a/tests/services/oracle-price-broadcaster.test.ts b/tests/services/oracle-price-broadcaster.test.ts new file mode 100644 index 0000000..505806b --- /dev/null +++ b/tests/services/oracle-price-broadcaster.test.ts @@ -0,0 +1,171 @@ +/** + * Tests for OraclePriceBroadcaster, including BUG-103: the broadcaster must + * reconnect with backoff after Supabase Realtime reports CHANNEL_ERROR/ + * TIMED_OUT/CLOSED (Realtime does not retry on its own), and must expose its + * status so /health can detect a dead channel instead of reporting "ok" + * while no live prices are flowing. + */ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +const { publishSpy } = vi.hoisted(() => ({ publishSpy: vi.fn() })); + +vi.mock("@percolator/shared", () => ({ + eventBus: { publish: publishSpy }, + getSupabase: vi.fn(), + getNetwork: vi.fn(() => "devnet"), + createLogger: vi.fn(() => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + })), +})); + +import { getSupabase } from "@percolator/shared"; +import { OraclePriceBroadcaster, getOraclePriceBroadcaster } from "../../src/services/OraclePriceBroadcaster.js"; + +type StatusCallback = (status: string, err?: unknown) => void; + +describe("OraclePriceBroadcaster", () => { + let insertHandler: (payload: { new: unknown }) => void; + let statusCallback: StatusCallback | undefined; + let subscribeCallCount: number; + let mockChannel: any; + let mockSupabase: any; + + beforeEach(() => { + publishSpy.mockClear(); + subscribeCallCount = 0; + mockChannel = { + on: vi.fn((_event: string, _filter: unknown, handler: typeof insertHandler) => { + insertHandler = handler; + return mockChannel; + }), + subscribe: vi.fn((cb?: StatusCallback) => { + subscribeCallCount++; + statusCallback = cb; + return mockChannel; + }), + }; + mockSupabase = { + channel: vi.fn(() => mockChannel), + removeChannel: vi.fn(), + }; + vi.mocked(getSupabase).mockReturnValue(mockSupabase); + }); + + describe("price publishing", () => { + it("publishes price.updated with the slab and priceE6 from the row", async () => { + const broadcaster = new OraclePriceBroadcaster(); + await broadcaster.start(); + + insertHandler({ + new: { + slab_address: "SLAB1", + price_e6: "1500000", + timestamp: Date.now(), + tx_signature: "sig123", + network: "devnet", + }, + }); + + expect(publishSpy).toHaveBeenCalledTimes(1); + const [event, slab, data] = publishSpy.mock.calls[0]; + expect(event).toBe("price.updated"); + expect(slab).toBe("SLAB1"); + expect(data.priceE6).toBe(1500000); + }); + + it("ignores a row with a non-positive or non-finite price", async () => { + const broadcaster = new OraclePriceBroadcaster(); + await broadcaster.start(); + + insertHandler({ new: { slab_address: "SLAB1", price_e6: "0", timestamp: Date.now(), tx_signature: null, network: "devnet" } }); + insertHandler({ new: { slab_address: "SLAB1", price_e6: "not-a-number", timestamp: Date.now(), tx_signature: null, network: "devnet" } }); + + expect(publishSpy).not.toHaveBeenCalled(); + }); + }); + + describe("status tracking and reconnect (BUG-103)", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("starts as not_started/unhealthy, becomes healthy once SUBSCRIBED", async () => { + const broadcaster = new OraclePriceBroadcaster(); + expect(broadcaster.getStatus()).toBe("not_started"); + expect(broadcaster.isHealthy()).toBe(false); + + await broadcaster.start(); + statusCallback!("SUBSCRIBED"); + + expect(broadcaster.getStatus()).toBe("SUBSCRIBED"); + expect(broadcaster.isHealthy()).toBe(true); + }); + + it("reconnects after a channel error instead of staying dead until restart", async () => { + const broadcaster = new OraclePriceBroadcaster(); + await broadcaster.start(); + expect(subscribeCallCount).toBe(1); + + statusCallback!("CHANNEL_ERROR"); + expect(broadcaster.isHealthy()).toBe(false); + + // First backoff attempt: 1000ms. Without the fix, nothing would ever + // re-subscribe here — the channel would stay dead until process restart. + await vi.advanceTimersByTimeAsync(999); + expect(subscribeCallCount).toBe(1); + await vi.advanceTimersByTimeAsync(1); + expect(subscribeCallCount).toBe(2); + + statusCallback!("SUBSCRIBED"); + expect(broadcaster.isHealthy()).toBe(true); + }); + + it("increases backoff delay on repeated failures and resets it after a successful reconnect", async () => { + const broadcaster = new OraclePriceBroadcaster(); + await broadcaster.start(); + + statusCallback!("CHANNEL_ERROR"); // attempt 1 -> 1000ms + await vi.advanceTimersByTimeAsync(1000); + expect(subscribeCallCount).toBe(2); + + statusCallback!("TIMED_OUT"); // attempt 2 -> 2000ms + await vi.advanceTimersByTimeAsync(1999); + expect(subscribeCallCount).toBe(2); + await vi.advanceTimersByTimeAsync(1); + expect(subscribeCallCount).toBe(3); + + statusCallback!("SUBSCRIBED"); // success resets the backoff counter + statusCallback!("CLOSED"); // attempt 1 again -> 1000ms, not 4000ms + await vi.advanceTimersByTimeAsync(1000); + expect(subscribeCallCount).toBe(4); + }); + + it("stop() prevents any pending or future reconnect from firing", async () => { + const broadcaster = new OraclePriceBroadcaster(); + await broadcaster.start(); + statusCallback!("CHANNEL_ERROR"); + + await broadcaster.stop(); + await vi.advanceTimersByTimeAsync(60_000); + + expect(subscribeCallCount).toBe(1); + expect(broadcaster.getStatus()).toBe("stopped"); + expect(broadcaster.isHealthy()).toBe(false); + }); + }); + + describe("getOraclePriceBroadcaster singleton", () => { + it("returns the same instance across calls", () => { + const a = getOraclePriceBroadcaster(); + const b = getOraclePriceBroadcaster(); + expect(a).toBe(b); + }); + }); +});