fix(api): reconnect OraclePriceBroadcaster on channel drop + surface its health [BUG-103]

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

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) {

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<typeof setTimeout> | 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<void> {
-    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<string, unknown> = { 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<void> {
+    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;
+}

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 () => {