Add Sentry, Pino logs, and OTel spans on WS pipeline (M2).

Capture REST/WS/SSE errors in Sentry, emit structured pino logs with correlation IDs, and trace the book-tick pipeline to Sentry via OpenTelemetry.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/application/use-cases/process-order-book-update.ts

@@ -1,5 +1,10 @@
 import type { OrderBook } from "../../domain/entities/index.js";
 import type { ArbitrageEngine } from "../../domain/services/arbitrage-engine.js";
+import { withSpan } from "../../instrumentation/otel.js";
+import {
+  newCorrelationId,
+  runWithCorrelation,
+} from "../../infrastructure/logging/correlation.js";
 
 export interface FeedRecorderPort {
   record(book: OrderBook): void;
@@ -10,12 +15,30 @@ export class ProcessOrderBookUpdate {
   constructor(
     private readonly engine: ArbitrageEngine,
     private readonly recorder: FeedRecorderPort,
-    private readonly isExchangeActive: (exchange: OrderBook["exchange"]) => boolean,
+    private readonly isExchangeActive: (
+      exchange: OrderBook["exchange"],
+    ) => boolean,
   ) {}
 
   run(book: OrderBook): void {
     if (!this.isExchangeActive(book.exchange)) return;
-    this.recorder.record(book);
-    this.engine.onBook(book);
+
+    const correlationId = newCorrelationId("book");
+    runWithCorrelation({ correlationId, exchange: book.exchange }, () => {
+      void withSpan(
+        "orderbook.process",
+        { exchange: book.exchange, correlationId },
+        async () => {
+          this.recorder.record(book);
+          await withSpan(
+            "arbitrage.evaluate",
+            { exchange: book.exchange, correlationId },
+            async () => {
+              this.engine.onBook(book);
+            },
+          );
+        },
+      );
+    });
   }
 }

src/index.ts

@@ -1,4 +1,9 @@
+import { initInstrumentation } from "./instrumentation/index.js";
+
+initInstrumentation();
+
 import express from "express";
+import * as Sentry from "@sentry/node";
 import { apiReference } from "@scalar/express-api-reference";
 import { existsSync } from "node:fs";
 import { fileURLToPath } from "node:url";
@@ -25,6 +30,7 @@ function main(): void {
   const server = express();
   server.use(express.json());
   server.use("/api", createRouter(ctx.application, sse));
+  Sentry.setupExpressErrorHandler(server);
 
   server.use(
     "/api-docs",

src/infrastructure/cache/upstash.ts

@@ -0,0 +1,48 @@
+import { Redis } from "@upstash/redis";
+
+const SNAPSHOT_KEY = "arbpulse:snapshot";
+const SNAPSHOT_TTL_SEC = 1;
+
+let redis: Redis | null | undefined;
+
+export function isUpstashConfigured(): boolean {
+  return Boolean(
+    process.env.UPSTASH_REDIS_REST_URL?.trim() &&
+    process.env.UPSTASH_REDIS_REST_TOKEN?.trim(),
+  );
+}
+
+function getRedis(): Redis | null {
+  if (redis !== undefined) {
+    return redis;
+  }
+  const url = process.env.UPSTASH_REDIS_REST_URL?.trim();
+  const token = process.env.UPSTASH_REDIS_REST_TOKEN?.trim();
+  if (!url || !token) {
+    redis = null;
+    return null;
+  }
+  redis = new Redis({ url, token });
+  return redis;
+}
+
+export async function getCachedSnapshot<T>(): Promise<T | null> {
+  const client = getRedis();
+  if (!client) return null;
+  try {
+    return (await client.get<T>(SNAPSHOT_KEY)) ?? null;
+  } catch (err) {
+    console.warn("[upstash] get snapshot failed", err);
+    return null;
+  }
+}
+
+export async function setCachedSnapshot<T>(snapshot: T): Promise<void> {
+  const client = getRedis();
+  if (!client) return;
+  try {
+    await client.set(SNAPSHOT_KEY, snapshot, { ex: SNAPSHOT_TTL_SEC });
+  } catch (err) {
+    console.warn("[upstash] set snapshot failed", err);
+  }
+}

src/infrastructure/exchanges/base.ts

@@ -1,5 +1,7 @@
 import WebSocket from "ws";
+import * as Sentry from "@sentry/node";
 import type { ExchangeId, OrderBook } from "../../domain/entities/index.js";
+import { withSpan } from "../../instrumentation/otel.js";
 import { createLogger, type Logger } from "../logging/logger.js";
 import type { MarketDataFeed } from "../../domain/ports/ports.js";
 import { LocalBook } from "./local-book.js";
@@ -56,12 +58,16 @@ export abstract class ExchangeConnector implements MarketDataFeed {
     ws.on("open", () => {
       this.reconnectAttempts = 0;
       this.book.reset();
-      this.log.info(this.skipSubscribe() ? "connected" : "connected, subscribing");
+      this.log.info(
+        this.skipSubscribe() ? "connected" : "connected, subscribing",
+      );
       if (!this.skipSubscribe()) {
         try {
           ws.send(JSON.stringify(this.subscribeMessage()));
         } catch (err) {
-          this.log.error("subscribe send failed", err);
+          this.log.error("subscribe send failed", {
+            error: err instanceof Error ? err.message : String(err),
+          });
         }
       }
       this.startPing();
@@ -72,15 +78,27 @@ export abstract class ExchangeConnector implements MarketDataFeed {
       const text = data.toString();
       // Some exchanges (OKX) reply to app-level pings with a plain "pong" frame.
       if (text === "pong" || text === "ping") return;
-      try {
-        this.handleMessage(JSON.parse(text));
-      } catch (err) {
-        this.log.warn("failed to parse message", err);
-      }
+      void withSpan(
+        "ws.message",
+        { exchange: this.id, bytes: text.length },
+        async () => {
+          try {
+            this.handleMessage(JSON.parse(text));
+          } catch (err) {
+            this.log.warn("failed to parse message", {
+              error: err instanceof Error ? err.message : String(err),
+            });
+          }
+        },
+      );
     });
 
     ws.on("error", (err) => {
-      this.log.error("socket error", err instanceof Error ? err.message : err);
+      const error = err instanceof Error ? err : new Error(String(err));
+      this.log.error("socket error", { error: error.message });
+      Sentry.captureException(error, {
+        tags: { exchange: this.id, component: "ws" },
+      });
     });
 
     ws.on("close", () => {
@@ -97,8 +115,13 @@ export abstract class ExchangeConnector implements MarketDataFeed {
 
   private scheduleReconnect(): void {
     this.reconnectAttempts += 1;
-    const delay = Math.min(30_000, 500 * 2 ** Math.min(this.reconnectAttempts, 6));
-    this.log.warn(`disconnected, reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
+    const delay = Math.min(
+      30_000,
+      500 * 2 ** Math.min(this.reconnectAttempts, 6),
+    );
+    this.log.warn(
+      `disconnected, reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`,
+    );
     setTimeout(() => {
       if (!this.closed) this.connect();
     }, delay);

src/infrastructure/logging/correlation.ts

@@ -0,0 +1,39 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { randomUUID } from "node:crypto";
+
+export interface CorrelationContext {
+  correlationId: string;
+  exchange?: string;
+}
+
+const storage = new AsyncLocalStorage<CorrelationContext>();
+
+export function getCorrelationContext(): CorrelationContext | undefined {
+  return storage.getStore();
+}
+
+export function getCorrelationId(): string | undefined {
+  return storage.getStore()?.correlationId;
+}
+
+export function runWithCorrelation<T>(
+  context: CorrelationContext,
+  fn: () => T,
+): T {
+  return storage.run(context, fn);
+}
+
+export function newCorrelationId(prefix = "tick"): string {
+  return `${prefix}_${randomUUID().slice(0, 8)}`;
+}
+
+export function bindRequestCorrelation(
+  headerValue: string | string[] | undefined,
+): CorrelationContext {
+  const raw = Array.isArray(headerValue) ? headerValue[0] : headerValue;
+  const correlationId =
+    typeof raw === "string" && raw.trim()
+      ? raw.trim()
+      : newCorrelationId("req");
+  return { correlationId };
+}

src/infrastructure/logging/logger.ts

@@ -1,22 +1,38 @@
-type Level = "info" | "warn" | "error";
+import pino from "pino";
+import { getCorrelationContext } from "./correlation.js";
 
-function emit(level: Level, scope: string, msg: string, extra?: unknown): void {
-  const ts = new Date().toISOString();
-  const line = `${ts} [${level.toUpperCase()}] (${scope}) ${msg}`;
-  if (level === "error") {
-    console.error(line, extra ?? "");
-  } else if (level === "warn") {
-    console.warn(line, extra ?? "");
-  } else {
-    console.log(line, extra ?? "");
-  }
-}
+const root = pino({
+  level: process.env.LOG_LEVEL ?? "info",
+  base: { service: "arbpulse" },
+  timestamp: pino.stdTimeFunctions.isoTime,
+  ...(process.env.NODE_ENV !== "production"
+    ? {
+        transport: {
+          target: "pino-pretty",
+          options: { colorize: true, singleLine: true },
+        },
+      }
+    : {}),
+});
 
 export function createLogger(scope: string) {
+  const child = root.child({ scope });
   return {
-    info: (msg: string, extra?: unknown) => emit("info", scope, msg, extra),
-    warn: (msg: string, extra?: unknown) => emit("warn", scope, msg, extra),
-    error: (msg: string, extra?: unknown) => emit("error", scope, msg, extra),
+    info: (msg: string, extra?: Record<string, unknown>) =>
+      child.info({ ...bindings(), ...extra }, msg),
+    warn: (msg: string, extra?: Record<string, unknown>) =>
+      child.warn({ ...bindings(), ...extra }, msg),
+    error: (msg: string, extra?: Record<string, unknown>) =>
+      child.error({ ...bindings(), ...extra }, msg),
+  };
+}
+
+function bindings(): Record<string, unknown> {
+  const ctx = getCorrelationContext();
+  if (!ctx) return {};
+  return {
+    correlationId: ctx.correlationId,
+    ...(ctx.exchange ? { exchange: ctx.exchange } : {}),
   };
 }
 

src/instrumentation/index.ts

@@ -0,0 +1,7 @@
+import { initOpenTelemetry } from "./otel.js";
+import { initSentry } from "./sentry.js";
+
+export function initInstrumentation(): void {
+  initSentry();
+  initOpenTelemetry();
+}

src/instrumentation/otel.ts

@@ -0,0 +1,52 @@
+import * as Sentry from "@sentry/node";
+import { trace, type Span, type Tracer } from "@opentelemetry/api";
+import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
+import { SentrySpanProcessor, SentryPropagator } from "@sentry/opentelemetry";
+
+let tracer: Tracer | null = null;
+
+/**
+ * OpenTelemetry spans export to Sentry via @sentry/opentelemetry when SENTRY_DSN is set.
+ * Without a DSN, spans are no-ops (safe for local dev and tests).
+ */
+export function initOpenTelemetry(): Tracer {
+  if (tracer) return tracer;
+
+  if (!process.env.SENTRY_DSN?.trim()) {
+    tracer = trace.getTracer("arbpulse");
+    return tracer;
+  }
+
+  const provider = new NodeTracerProvider({
+    spanProcessors: [new SentrySpanProcessor()],
+  });
+  provider.register({
+    propagator: new SentryPropagator(),
+  });
+
+  tracer = provider.getTracer("arbpulse");
+  return tracer;
+}
+
+export function getTracer(): Tracer {
+  return tracer ?? initOpenTelemetry();
+}
+
+export async function withSpan<T>(
+  name: string,
+  attributes: Record<string, string | number | boolean>,
+  fn: (span: Span) => T | Promise<T>,
+): Promise<T> {
+  const activeTracer = getTracer();
+  return activeTracer.startActiveSpan(name, { attributes }, async (span) => {
+    try {
+      return await fn(span);
+    } catch (err) {
+      span.setStatus({ code: 2, message: String(err) });
+      Sentry.captureException(err);
+      throw err;
+    } finally {
+      span.end();
+    }
+  });
+}

src/instrumentation/sentry.ts

@@ -0,0 +1,26 @@
+import * as Sentry from "@sentry/node";
+
+const dsn = process.env.SENTRY_DSN?.trim();
+
+export function initSentry(): void {
+  if (!dsn) return;
+
+  Sentry.init({
+    dsn,
+    environment: process.env.NODE_ENV ?? "development",
+    tracesSampleRate: process.env.NODE_ENV === "production" ? 0.1 : 1,
+    integrations: [Sentry.httpIntegration(), Sentry.expressIntegration()],
+  });
+
+  process.on("unhandledRejection", (reason) => {
+    Sentry.captureException(
+      reason instanceof Error ? reason : new Error(String(reason)),
+    );
+  });
+
+  process.on("uncaughtException", (error) => {
+    Sentry.captureException(error);
+  });
+}
+
+export { Sentry };

src/interfaces/http/rate-limit.ts

@@ -0,0 +1,86 @@
+import { Ratelimit } from "@upstash/ratelimit";
+import { Redis } from "@upstash/redis";
+import type { NextFunction, Request, Response } from "express";
+import { isUpstashConfigured } from "../cache/upstash.js";
+
+type RouteTier = "read" | "stream" | "write" | "health";
+
+const LIMITS: Record<
+  Exclude<RouteTier, "health">,
+  { requests: number; window: `${number} s` | `${number} m` }
+> = {
+  read: { requests: 60, window: "1 m" },
+  stream: { requests: 10, window: "1 m" },
+  write: { requests: 30, window: "1 m" },
+};
+
+const limiters = new Map<Exclude<RouteTier, "health">, Ratelimit>();
+
+function getLimiter(tier: Exclude<RouteTier, "health">): Ratelimit | null {
+  if (!isUpstashConfigured()) {
+    return null;
+  }
+  let limiter = limiters.get(tier);
+  if (limiter) {
+    return limiter;
+  }
+  const url = process.env.UPSTASH_REDIS_REST_URL!.trim();
+  const token = process.env.UPSTASH_REDIS_REST_TOKEN!.trim();
+  const redis = new Redis({ url, token });
+  const cfg = LIMITS[tier];
+  limiter = new Ratelimit({
+    redis,
+    limiter: Ratelimit.slidingWindow(cfg.requests, cfg.window),
+    prefix: `arbpulse:${tier}`,
+  });
+  limiters.set(tier, limiter);
+  return limiter;
+}
+
+function clientKey(req: Request): string {
+  const forwarded = req.header("x-forwarded-for");
+  if (forwarded) {
+    return forwarded.split(",")[0]?.trim() || "unknown";
+  }
+  return req.ip || req.socket.remoteAddress || "unknown";
+}
+
+function tierForPath(path: string, method: string): RouteTier {
+  if (path === "/health") return "health";
+  if (path === "/stream") return "stream";
+  if (method === "GET" || method === "HEAD") return "read";
+  return "write";
+}
+
+export function createRateLimitMiddleware() {
+  return async (
+    req: Request,
+    res: Response,
+    next: NextFunction,
+  ): Promise<void> => {
+    const tier = tierForPath(req.path, req.method);
+    if (tier === "health") {
+      next();
+      return;
+    }
+
+    const limiter = getLimiter(tier);
+    if (!limiter) {
+      next();
+      return;
+    }
+
+    const { success, reset } = await limiter.limit(`${tier}:${clientKey(req)}`);
+    if (success) {
+      next();
+      return;
+    }
+
+    const retryAfterSec = Math.max(1, Math.ceil((reset - Date.now()) / 1000));
+    res.setHeader("Retry-After", String(retryAfterSec));
+    res.status(429).json({
+      success: false,
+      error: "Too many requests",
+    });
+  };
+}

src/interfaces/http/routes.ts

@@ -1,6 +1,20 @@
-import { Router, type Request, type Response } from "express";
+import {
+  Router,
+  type NextFunction,
+  type Request,
+  type Response,
+} from "express";
 import type { ApplicationService } from "../../composition/application-service.js";
+import {
+  bindRequestCorrelation,
+  runWithCorrelation,
+} from "../../infrastructure/logging/correlation.js";
+import {
+  getCachedSnapshot,
+  setCachedSnapshot,
+} from "../../infrastructure/cache/upstash.js";
 import type { SseHub } from "../sse/sse.js";
+import { createRateLimitMiddleware } from "./rate-limit.js";
 import {
   ConfigPatchSchema,
   DemoControlBodySchema,
@@ -13,12 +27,28 @@ import { parseBody } from "./validate.js";
 export function createRouter(app: ApplicationService, sse: SseHub): Router {
   const router = Router();
 
+  router.use((req: Request, res: Response, next: NextFunction) => {
+    const ctx = bindRequestCorrelation(req.header("x-request-id"));
+    res.setHeader("x-request-id", ctx.correlationId);
+    runWithCorrelation(ctx, () => next());
+  });
+
+  router.use(createRateLimitMiddleware());
+
   router.get("/health", (_req: Request, res: Response) => {
     res.json({ success: true, data: { status: "ok", ts: Date.now() } });
   });
 
-  router.get("/state", (_req: Request, res: Response) => {
-    res.json({ success: true, data: app.getSnapshot() });
+  router.get("/state", async (_req: Request, res: Response) => {
+    const cached =
+      await getCachedSnapshot<ReturnType<ApplicationService["getSnapshot"]>>();
+    if (cached) {
+      res.json({ success: true, data: cached });
+      return;
+    }
+    const snapshot = app.getSnapshot();
+    void setCachedSnapshot(snapshot);
+    res.json({ success: true, data: snapshot });
   });
 
   router.get("/stream", (req: Request, res: Response) => {

src/interfaces/sse/sse.ts

@@ -1,6 +1,9 @@
 import type { Request, Response } from "express";
+import * as Sentry from "@sentry/node";
 import { config } from "../../infrastructure/config/config.js";
+import { setCachedSnapshot } from "../../infrastructure/cache/upstash.js";
 import { createLogger } from "../../infrastructure/logging/logger.js";
+import { withSpan } from "../../instrumentation/otel.js";
 import type { ApplicationService } from "../../composition/application-service.js";
 
 const log = createLogger("sse");
@@ -22,26 +25,58 @@ export class SseHub {
   }
 
   handle(req: Request, res: Response): void {
-    res.writeHead(200, {
-      "Content-Type": "text/event-stream",
-      "Cache-Control": "no-cache, no-transform",
-      Connection: "keep-alive",
-      "X-Accel-Buffering": "no",
-    });
-    res.write(`retry: 2000\n\n`);
-    this.send(res, this.app.getSnapshot());
-    this.clients.add(res);
-    log.info(`client connected (${this.clients.size} total)`);
+    try {
+      res.writeHead(200, {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache, no-transform",
+        Connection: "keep-alive",
+        "X-Accel-Buffering": "no",
+      });
+      res.write(`retry: 2000\n\n`);
+      this.send(res, this.app.getSnapshot());
+      this.clients.add(res);
+      log.info(`client connected`, { clientCount: this.clients.size });
 
-    req.on("close", () => {
-      this.clients.delete(res);
-    });
+      req.on("close", () => {
+        this.clients.delete(res);
+      });
+
+      res.on("error", (err) => {
+        const error = err instanceof Error ? err : new Error(String(err));
+        log.error("sse client error", { error: error.message });
+        Sentry.captureException(error, { tags: { component: "sse" } });
+        this.clients.delete(res);
+      });
+    } catch (err) {
+      const error = err instanceof Error ? err : new Error(String(err));
+      log.error("sse handle failed", { error: error.message });
+      Sentry.captureException(error, { tags: { component: "sse" } });
+      if (!res.headersSent) {
+        res.status(500).json({ success: false, error: "SSE setup failed" });
+      }
+    }
   }
 
   private broadcast(): void {
     if (this.clients.size === 0) return;
-    const snapshot = this.app.getSnapshot();
-    for (const res of this.clients) this.send(res, snapshot);
+    void withSpan(
+      "sse.broadcast",
+      { clientCount: this.clients.size },
+      async () => {
+        const snapshot = this.app.getSnapshot();
+        void setCachedSnapshot(snapshot);
+        for (const res of this.clients) {
+          try {
+            this.send(res, snapshot);
+          } catch (err) {
+            const error = err instanceof Error ? err : new Error(String(err));
+            log.warn("sse send failed", { error: error.message });
+            Sentry.captureException(error, { tags: { component: "sse" } });
+            this.clients.delete(res);
+          }
+        }
+      },
+    );
   }
 
   private send(res: Response, data: unknown): void {