diff --git a/.env.example b/.env.example index ae06e25..a76d0f1 100644 --- a/.env.example +++ b/.env.example @@ -582,4 +582,16 @@ LOG_SHARD_RETENTION_DAYS=7 # Dedicated Sandbox Environment # --------------------------------------------------------------------------- IS_SANDBOX=false -SANDBOX_DATABASE_URL=postgresql://user:password@localhost:5432/mobile_money_sandbox?schema=public \ No newline at end of file +SANDBOX_DATABASE_URL=postgresql://user:password@localhost:5432/mobile_money_sandbox?schema=public + +# --------------------------------------------------------------------------- +# Stellar Ledger Close Time Monitor +# --------------------------------------------------------------------------- +# Alert threshold: fire Slack/PagerDuty when rolling average close time exceeds this (seconds) +LEDGER_CLOSE_TIME_ALERT_THRESHOLD_S=10 +# Minimum gap between repeated alerts for the same breach episode (ms, default: 5 minutes) +LEDGER_ALERT_COOLDOWN_MS=300000 +# SSE stream reconnect delay on error (ms) +LEDGER_MONITOR_RECONNECT_DELAY_MS=5000 +# Slack Incoming Webhook URL for ledger close time alerts (leave blank to disable) +SLACK_WEBHOOK_URL= diff --git a/src/index.ts b/src/index.ts index d95a5bf..7aaa2a2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -89,6 +89,7 @@ import { paymentLinkRoutes } from "./routes/paymentLinkRoutes.js"; import providerStatusRouter from "./routes/providerStatus"; import { startHeartbeatService, stopHeartbeatService } from "./services/heartbeatService"; import { startStellarExporter } from "./services/stellarExporter"; +import { startLedgerMonitor, stopLedgerMonitor } from "./services/ledgerMonitor"; // Sentry Middleware import { initSentry, sentryBreadcrumbMiddleware } from "./middleware/sentry"; @@ -505,6 +506,10 @@ async function gracefulShutdown(signal: NodeJS.Signals): Promise { stopHeartbeatService(); console.log("[Shutdown] Heartbeat service stopped"); + console.log("[Shutdown] Stopping ledger monitor"); + stopLedgerMonitor(); + console.log("[Shutdown] Ledger monitor stopped"); + console.log("[Shutdown] Closing PostgreSQL pool"); await pool.end(); console.log("[Shutdown] PostgreSQL pool closed"); @@ -543,6 +548,9 @@ async function initializeRuntime(): Promise { // Initialize Prometheus Horizon Scraper startStellarExporter(); + // Initialize Stellar Ledger Close Time Monitor + startLedgerMonitor(); + // Initialize System Heartbeat Metric startHeartbeatService(); diff --git a/src/services/ledgerMonitor.ts b/src/services/ledgerMonitor.ts new file mode 100644 index 0000000..8f9f20b --- /dev/null +++ b/src/services/ledgerMonitor.ts @@ -0,0 +1,303 @@ +import axios from "axios"; +import { getHorizonUrls } from "../config/stellar"; +import { + stellarLedgerCloseTimeSeconds, + stellarLedgerSequence, +} from "../utils/metrics"; + +// --------------------------------------------------------------------------- +// Configuration +// --------------------------------------------------------------------------- + +const ROLLING_WINDOW = 10; // number of ledgers to average over +const ALERT_THRESHOLD_SECONDS = parseFloat( + process.env.LEDGER_CLOSE_TIME_ALERT_THRESHOLD_S || "10", +); +// Minimum gap (ms) between repeated alerts for the same breach episode +const ALERT_COOLDOWN_MS = parseInt( + process.env.LEDGER_ALERT_COOLDOWN_MS || "300000", // 5 minutes + 10, +); +// SSE reconnect delay on stream error (ms) +const RECONNECT_DELAY_MS = parseInt( + process.env.LEDGER_MONITOR_RECONNECT_DELAY_MS || "5000", + 10, +); + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +interface HorizonLedgerRecord { + sequence: number; + closed_at: string; // ISO 8601 +} + +// --------------------------------------------------------------------------- +// State +// --------------------------------------------------------------------------- + +/** Ring buffer of the last ROLLING_WINDOW close-time deltas (seconds). */ +const closeTimes: number[] = []; + +let prevClosedAt: Date | null = null; +let lastAlertAt: number | null = null; +let activeStreamClose: (() => void) | null = null; +let isRunning = false; + +// --------------------------------------------------------------------------- +// Alert helpers +// --------------------------------------------------------------------------- + +async function sendSlackAlert(avgSeconds: number): Promise { + const webhookUrl = process.env.SLACK_WEBHOOK_URL; + if (!webhookUrl) return; + + const message = { + text: `⚠️ *Stellar Ledger Close Time Alert*\nRolling average close time is *${avgSeconds.toFixed(2)}s* (threshold: ${ALERT_THRESHOLD_SECONDS}s).\nPayment confirmation times may be degraded.`, + }; + + try { + await axios.post(webhookUrl, message, { timeout: 5000 }); + } catch (err) { + console.error("[LedgerMonitor] Failed to send Slack alert:", err); + } +} + +async function sendPagerDutyAlert(avgSeconds: number): Promise { + const integrationKey = process.env.PAGERDUTY_INTEGRATION_KEY; + if (!integrationKey) return; + + const dedupKey = `${process.env.PAGERDUTY_DEDUP_KEY || "proxypay"}-stellar-ledger-close-time`; + + const event = { + routing_key: integrationKey, + event_action: "trigger", + dedup_key: dedupKey, + payload: { + summary: `Stellar ledger close time ${avgSeconds.toFixed(2)}s exceeds ${ALERT_THRESHOLD_SECONDS}s threshold`, + timestamp: new Date().toISOString(), + severity: "warning", + source: "proxypay-ledger-monitor", + custom_details: { + rolling_average_seconds: avgSeconds.toFixed(2), + threshold_seconds: ALERT_THRESHOLD_SECONDS, + window_ledgers: ROLLING_WINDOW, + environment: process.env.NODE_ENV || "development", + }, + }, + }; + + try { + await axios.post("https://events.pagerduty.com/v2/enqueue", event, { + timeout: 5000, + }); + } catch (err) { + console.error("[LedgerMonitor] Failed to send PagerDuty alert:", err); + } +} + +async function maybeAlert(avgSeconds: number): Promise { + if (avgSeconds <= ALERT_THRESHOLD_SECONDS) return; + + const now = Date.now(); + if (lastAlertAt !== null && now - lastAlertAt < ALERT_COOLDOWN_MS) return; + + lastAlertAt = now; + + console.warn( + JSON.stringify({ + timestamp: new Date().toISOString(), + level: "WARN", + message: "Stellar ledger close time exceeded threshold", + rolling_average_seconds: avgSeconds.toFixed(2), + threshold_seconds: ALERT_THRESHOLD_SECONDS, + window_ledgers: ROLLING_WINDOW, + }), + ); + + await Promise.allSettled([sendSlackAlert(avgSeconds), sendPagerDutyAlert(avgSeconds)]); +} + +// --------------------------------------------------------------------------- +// Per-ledger processing +// --------------------------------------------------------------------------- + +function onLedger(record: HorizonLedgerRecord): void { + const closedAt = new Date(record.closed_at); + + stellarLedgerSequence.set(record.sequence); + + if (prevClosedAt !== null) { + const deltaSeconds = (closedAt.getTime() - prevClosedAt.getTime()) / 1000; + + // Only accept positive, sane values (guard against out-of-order events) + if (deltaSeconds > 0 && deltaSeconds < 120) { + closeTimes.push(deltaSeconds); + if (closeTimes.length > ROLLING_WINDOW) { + closeTimes.shift(); + } + + if (closeTimes.length >= 2) { + const avg = + closeTimes.reduce((a, b) => a + b, 0) / closeTimes.length; + stellarLedgerCloseTimeSeconds.set(avg); + + maybeAlert(avg).catch(() => { + // fire-and-forget; errors already logged inside maybeAlert + }); + } + } + } + + prevClosedAt = closedAt; +} + +// --------------------------------------------------------------------------- +// SSE stream subscription +// --------------------------------------------------------------------------- + +/** + * Opens an SSE connection to the Horizon /ledgers endpoint and pumps events + * into `onLedger`. Returns a `close()` function that tears down the stream. + * + * Uses the first available Horizon URL from the pool. If the stream errors, + * the caller (the retry loop) will reconnect. + */ +function subscribeToLedgerStream(horizonUrl: string): () => void { + const url = `${horizonUrl.replace(/\/$/, "")}/ledgers?order=asc&cursor=now`; + + // We use axios with responseType 'stream' to get a raw Node.js Readable. + // This avoids pulling in an extra EventSource polyfill while reusing the + // axios instance that already has a timeout/interceptor setup in the project. + let closed = false; + const controller = new AbortController(); + + (async () => { + try { + const response = await axios.get(url, { + responseType: "stream", + signal: controller.signal, + headers: { Accept: "text/event-stream" }, + timeout: 0, // long-lived stream — no timeout + }); + + let buffer = ""; + + response.data.on("data", (chunk: Buffer) => { + buffer += chunk.toString(); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; // last incomplete line stays in buffer + + let dataPayload = ""; + for (const line of lines) { + if (line.startsWith("data: ")) { + dataPayload = line.slice(6).trim(); + } else if (line === "" && dataPayload) { + // End of an SSE event block + try { + if (dataPayload !== '"hello"' && dataPayload !== '"byebye"') { + const record = JSON.parse(dataPayload) as HorizonLedgerRecord; + if (record.sequence && record.closed_at) { + onLedger(record); + } + } + } catch { + // Ignore malformed events + } + dataPayload = ""; + } + } + }); + + response.data.on("error", (err: Error) => { + if (!closed) { + console.error("[LedgerMonitor] Stream error:", err.message); + } + }); + + response.data.on("end", () => { + if (!closed) { + console.warn("[LedgerMonitor] Stream ended unexpectedly"); + } + }); + } catch (err: any) { + if (!closed && err?.code !== "ERR_CANCELED") { + console.error("[LedgerMonitor] Failed to open ledger stream:", err?.message); + } + } + })(); + + return () => { + closed = true; + controller.abort(); + }; +} + +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- + +/** + * Start the Stellar ledger close time monitoring service. + * + * Subscribes to the Horizon `/ledgers` SSE stream, maintains a rolling + * average over the last 10 ledgers, exposes it at the `/metrics` endpoint, + * and fires Slack/PagerDuty alerts when the average exceeds 10 seconds. + */ +export function startLedgerMonitor(): void { + if (isRunning) return; + isRunning = true; + + console.log("[LedgerMonitor] Starting Stellar ledger close time monitor..."); + + const urls = getHorizonUrls(); + const primaryUrl = urls[0]; + + // Retry loop: reconnect on stream termination/error + const connect = (): void => { + if (!isRunning) return; + + const closeStream = subscribeToLedgerStream(primaryUrl); + activeStreamClose = closeStream; + + // Schedule a reconnect check — if the stream drops it will be restarted + setTimeout(() => { + if (!isRunning) return; + // Close current stream and reconnect + try { + closeStream(); + } catch { + // ignore + } + connect(); + }, RECONNECT_DELAY_MS + 60_000); // reconnect every ~65s to stay fresh + }; + + connect(); +} + +/** + * Stop the ledger monitor (called during graceful shutdown). + */ +export function stopLedgerMonitor(): void { + isRunning = false; + if (activeStreamClose) { + try { + activeStreamClose(); + } catch { + // ignore + } + activeStreamClose = null; + } + console.log("[LedgerMonitor] Stopped."); +} + +/** + * Returns the current rolling average ledger close time in seconds, + * or null if not enough data has been collected yet. + */ +export function getLedgerCloseTimeAvg(): number | null { + if (closeTimes.length < 2) return null; + return closeTimes.reduce((a, b) => a + b, 0) / closeTimes.length; +} diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 5d5cf5f..289d72b 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -217,3 +217,16 @@ export const systemHeartbeat = new Gauge({ labelNames: ["service"], registers: [register], }); + +// Stellar Ledger Close Time Metrics +export const stellarLedgerCloseTimeSeconds = new Gauge({ + name: "stellar_ledger_close_time_seconds", + help: "Rolling average ledger close time in seconds over the last 10 ledgers", + registers: [register], +}); + +export const stellarLedgerSequence = new Gauge({ + name: "stellar_ledger_sequence", + help: "Latest Stellar ledger sequence number observed", + registers: [register], +});