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); + }); + }); +});