diff --git a/.env.example b/.env.example index ad26f0d..77c4b3c 100644 --- a/.env.example +++ b/.env.example @@ -67,6 +67,16 @@ REDIS_URL=redis://localhost:6379 ALERT_TELEGRAM_CHAT_ID= DIGEST_HOUR_UTC=8 +# Dead-man's switch: ping URLs from an external monitoring service (any +# healthchecks.io-style "expect a request every N minutes" check). The worker +# pings only while every loop is fresh and Redis answers; the bot only while +# its Telegram long-poll succeeds. Unset = no pings. CLOUDFLARED_READY_URL +# lets the worker alert when the tunnel loses its edge connections; set it to +# http://cloudflared:2000/ready in the compose deployment. +HEARTBEAT_URL_WORKER= +HEARTBEAT_URL_BOT= +CLOUDFLARED_READY_URL= + # Worker only: max ms to wait for in-flight webhook batches on SIGTERM before # forcing close. Tune to ~2x observed p95 delivery time. GRACEFUL_SHUTDOWN_TIMEOUT_MS=10000 diff --git a/bot/index.js b/bot/index.js index 565f922..a331f64 100644 --- a/bot/index.js +++ b/bot/index.js @@ -9,6 +9,34 @@ const apiBase = `https://api.telegram.org/bot${botToken}`; const longPollSeconds = 30; let offset = 0; +// Dead-man's switch: ping HEARTBEAT_URL once a minute, but only while the +// long-poll keeps succeeding. A poll is healthy even when it returns no +// updates; what matters is that Telegram answered. No URL, no pings. +const heartbeatUrl = process.env.HEARTBEAT_URL || ""; +const HEARTBEAT_INTERVAL_MS = 60 * 1000; +const POLL_STALE_MS = 2 * (longPollSeconds + 5) * 1000; +let lastPollOkAt = 0; + +function pollIsFresh() { + return Date.now() - lastPollOkAt < POLL_STALE_MS; +} + +async function pingHeartbeat() { + if (!heartbeatUrl) return; + if (!pollIsFresh()) { + logEvent("warn", "heartbeat_withheld", { lastPollAgeMs: lastPollOkAt ? Date.now() - lastPollOkAt : null }); + return; + } + try { + const res = await fetch(heartbeatUrl, { signal: AbortSignal.timeout(5000) }); + if (!res.ok) logEvent("warn", "heartbeat_ping_rejected", { status: res.status }); + } catch (err) { + logEvent("warn", "heartbeat_ping_failed", { error: err }); + } +} + +if (heartbeatUrl) setInterval(pingHeartbeat, HEARTBEAT_INTERVAL_MS); + // Structured logging, inlined because the bot image ships only index.js and // cannot import the worker's logger. Same JSON-line shape as worker-log.js so // one log pipeline reads every process. @@ -77,7 +105,12 @@ async function startStatusMonitor() { }).catch(err => logEvent("warn", "pin_status_failed", { error: err })); setInterval(async () => { - const updatedText = `server status: UP\nStarted at: ${startTime}\nLast checked: ${new Date().toISOString()}`; + // The pinned message used to say UP unconditionally; now it reflects + // whether the long-poll is actually succeeding. + const state = pollIsFresh() + ? "UP" + : `DEGRADED (last successful poll ${lastPollOkAt ? Math.round((Date.now() - lastPollOkAt) / 1000) + "s ago" : "never"})`; + const updatedText = `server status: ${state}\nStarted at: ${startTime}\nLast checked: ${new Date().toISOString()}`; await editMessage(analyticsChatId, messageId, updatedText); }, 10 * 60 * 1000); } @@ -118,6 +151,7 @@ async function getUpdates() { await sleep(1000); return []; } + lastPollOkAt = Date.now(); return data.result; } catch (err) { logEvent("warn", "get_updates_error", { error: err }); diff --git a/docker-compose.yml b/docker-compose.yml index 8de69a4..4728f4f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -128,6 +128,7 @@ services: # accepts a decision only from this id, so an unset value silently # rejects every tap. Mandatory here rather than defaulted. BEARER_ADMIN_TELEGRAM_ID: ${BEARER_ADMIN_TELEGRAM_ID:?set BEARER_ADMIN_TELEGRAM_ID} + HEARTBEAT_URL: ${HEARTBEAT_URL_BOT:-} TELEGRAM_ANALYTICS_CHAT_ID: ${TELEGRAM_ANALYTICS_CHAT_ID:-} TELEGRAM_ANALYTICS_THREAD_ID: ${TELEGRAM_ANALYTICS_THREAD_ID:-} depends_on: @@ -189,6 +190,8 @@ services: BEARER_ADMIN_TELEGRAM_ID: ${BEARER_ADMIN_TELEGRAM_ID:-} GRACEFUL_SHUTDOWN_TIMEOUT_MS: ${GRACEFUL_SHUTDOWN_TIMEOUT_MS:-10000} DIGEST_HOUR_UTC: ${DIGEST_HOUR_UTC:-8} + HEARTBEAT_URL: ${HEARTBEAT_URL_WORKER:-} + CLOUDFLARED_READY_URL: ${CLOUDFLARED_READY_URL:-} NODE_ENV: production depends_on: redis: diff --git a/docs/deployment.md b/docs/deployment.md index 43832e9..7b4bc3b 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -139,8 +139,50 @@ absence is itself the alert. `docker compose exec -T worker node worker.js runbook's "is the channel alive" check. These alerts are sent from the host. When the host, the tunnel, or the -worker is down nothing can send them, so they do not replace an external -uptime monitor. +worker is down nothing can send them, so they do not replace the external +monitor described next. + +## Monitoring + +Two kinds of check, both run by a monitoring service outside the host (any +vendor that offers HTTP probes and "expect a ping every N minutes" checks, +delivering to the same Telegram chat if you like): + +HTTP probes, from outside, through the tunnel: + +- `GET https://auth.bneck.com/api/health/ready`, expect 200 and a body + containing `"ok":true`, every 1 to 2 minutes. This is the end-to-end check: + tunnel, app, Postgres and Redis. +- `GET https://auth.bneck.com/.well-known/openid-configuration`, expect 200 + and `"issuer":"https://auth.bneck.com"`, every 5 minutes. Catches a + misconfigured base URL that the readiness probe would not. +- Optional: `GET https://auth.bneck.com/oauth/jwks`, expect the active + `kid`, every 15 minutes, to catch a signing-key regression. + +Cloudflare's bot protection may block a vendor's probes; if the checks fail +with 403 while the site works, add a WAF skip rule for `/api/health/*` and +the discovery path scoped to the vendor's user agent or IP ranges. + +Heartbeats (dead-man's switches), pinged from inside: + +- `HEARTBEAT_URL_WORKER`: the worker pings once a minute, but only while + every background loop has completed without error inside its tolerance + and Redis answers. Configure the check for a 1 minute period and a 3 + minute grace. When it withholds the ping it also logs + `heartbeat_withheld` and sends a `worker_unhealthy` alert, so the two + signals corroborate each other. +- `HEARTBEAT_URL_BOT`: the bot pings once a minute while its Telegram + long-poll keeps succeeding. Same period and grace. The pinned status + message in the analytics chat also reads DEGRADED with the age of the last + good poll when it is stale. +- `CLOUDFLARED_READY_URL` (`http://cloudflared:2000/ready`): not a heartbeat, + but the worker checks it every minute and alerts `cloudflared_not_ready` + while the tunnel has no edge connection. The HTTP probes above are the + tunnel's dead-man's switch; this only gets the news out faster while the + host can still speak. + +The daily digest (above) is the slowest of the switches: its absence means +the worker, the database, or the alert channel has been dead for up to a day. ## Backup and restore diff --git a/tests/unit/worker-heartbeat.test.ts b/tests/unit/worker-heartbeat.test.ts new file mode 100644 index 0000000..0119cdf --- /dev/null +++ b/tests/unit/worker-heartbeat.test.ts @@ -0,0 +1,63 @@ +import { createRequire } from 'node:module'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const requireCjs = createRequire(import.meta.url); +const worker = requireCjs('../../worker.js') as { + LOOP_TOLERANCE_MS: Record; + markLoopOk: (loop: string, at?: number) => void; + staleLoops: (now?: number, since?: number) => string[]; + _resetLoopStateForTests: () => void; + pingHeartbeat: (url: string, fetchImpl?: (url: string, init: RequestInit) => Promise<{ ok: boolean; status?: number }>) => Promise; +}; + +const LOOPS = Object.keys(worker.LOOP_TOLERANCE_MS); +const T0 = Date.UTC(2026, 8, 6, 12, 0, 0); + +describe('worker heartbeat: loop freshness', () => { + beforeEach(() => { + for (const loop of LOOPS) worker.markLoopOk(loop, T0); + }); + + it('reports no stale loops right after every loop completed', () => { + expect(worker.staleLoops(T0 + 1000)).toEqual([]); + }); + + it('flags a loop once its tolerance has passed, and only that loop', () => { + worker.markLoopOk('webhook_delivery', T0 - worker.LOOP_TOLERANCE_MS.webhook_delivery - 1); + expect(worker.staleLoops(T0)).toEqual(['webhook_delivery']); + }); + + it('counts a loop that never completed from process start', () => { + worker._resetLoopStateForTests(); + const olderThanEveryTolerance = T0 - worker.LOOP_TOLERANCE_MS.hygiene - 1; + expect(worker.staleLoops(T0, olderThanEveryTolerance)).toEqual(LOOPS); + expect(worker.staleLoops(T0, T0)).toEqual([]); + }); + + it('tolerances are a little over two periods of each loop', () => { + expect(worker.LOOP_TOLERANCE_MS.webhook_delivery).toBeGreaterThan(2 * 1000); + expect(worker.LOOP_TOLERANCE_MS.activation_expiry).toBeGreaterThan(2 * 60_000); + expect(worker.LOOP_TOLERANCE_MS.hygiene).toBeGreaterThan(60 * 60_000); + }); +}); + +describe('worker heartbeat: ping', () => { + it('is a no-op without a URL', async () => { + const fetchImpl = vi.fn(); + expect(await worker.pingHeartbeat('', fetchImpl)).toBe(false); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it('GETs the URL with a timeout and reports success', async () => { + const fetchImpl = vi.fn(async () => ({ ok: true })); + expect(await worker.pingHeartbeat('https://ping.example/abc', fetchImpl)).toBe(true); + const [url, init] = fetchImpl.mock.calls[0] as unknown as [string, RequestInit]; + expect(url).toBe('https://ping.example/abc'); + expect(init.signal).toBeInstanceOf(AbortSignal); + }); + + it('never throws: a rejected or failed ping is reported as false', async () => { + expect(await worker.pingHeartbeat('https://ping.example/abc', async () => ({ ok: false, status: 503 }))).toBe(false); + expect(await worker.pingHeartbeat('https://ping.example/abc', async () => { throw new Error('ECONNRESET'); })).toBe(false); + }); +}); diff --git a/worker.js b/worker.js index f23cc57..e48e096 100644 --- a/worker.js +++ b/worker.js @@ -24,6 +24,103 @@ function alertLoopError(event, err) { return alerts.send(event, `${event}\n${message}`, { windowSeconds: LOOP_ALERT_WINDOW_SECONDS }); } +// Dead-man's switch. Every loop records its last error-free completion; once +// a minute the worker pings HEARTBEAT_URL only if every loop is fresh and +// Redis answers. A hung or erroring loop withholds the ping, and the external +// monitor expecting it is what pages the operator: the one alert that still +// works when nothing on this host can send one. Tolerances are a little over +// two periods so a single slow tick does not trip it. +const HEARTBEAT_INTERVAL_MS = Number(process.env.HEARTBEAT_INTERVAL_MS || 60_000); +const LOOP_TOLERANCE_MS = { + webhook_delivery: 30_000, + activation_expiry: 3 * 60_000, + hygiene: 2 * 60 * 60_000, + restriction: 2 * 60 * 60_000, + deletion: 2 * 60 * 60_000, + digest: 2 * 60 * 60_000, +}; +const lastLoopOkAt = new Map(); + +function markLoopOk(loop, at = Date.now()) { + if (loop) lastLoopOkAt.set(loop, at); +} + +// A loop that has never completed counts from process start, so a loop that +// fails from the very first tick is reported rather than ignored. +function _resetLoopStateForTests() { + lastLoopOkAt.clear(); +} + +function staleLoops(now = Date.now(), since = startedAt) { + return Object.entries(LOOP_TOLERANCE_MS) + .filter(([loop, tolerance]) => now - (lastLoopOkAt.get(loop) ?? since) > tolerance) + .map(([loop]) => loop); +} + +// GET the ping URL; best effort and never throws, because a monitoring +// vendor being down must not become a worker problem. +async function pingHeartbeat(url, fetchImpl = fetch) { + if (!url) return false; + try { + const res = await fetchImpl(url, { signal: AbortSignal.timeout(5000) }); + if (!res.ok) { + logger.warn("heartbeat_ping_rejected", { status: res.status }); + return false; + } + return true; + } catch (err) { + logger.warn("heartbeat_ping_failed", { error: err }); + return false; + } +} + +async function redisAnswers(redis, timeoutMs = 2000) { + if (!redis) return false; + try { + const pong = await Promise.race([ + redis.ping(), + new Promise((_, reject) => setTimeout(() => reject(new Error("redis ping timeout")), timeoutMs)), + ]); + return pong === "PONG"; + } catch { + return false; + } +} + +async function heartbeatTick() { + const stale = staleLoops(); + const redisOk = await redisAnswers(opsRedis); + if (stale.length > 0 || !redisOk) { + logger.warn("heartbeat_withheld", { stale, redisOk }); + await alerts.send( + "worker_unhealthy", + `Worker unhealthy\nstale loops: ${stale.join(", ") || "none"}\nredis: ${redisOk ? "ok" : "unreachable"}`, + { windowSeconds: LOOP_ALERT_WINDOW_SECONDS }, + ); + return; + } + await pingHeartbeat(process.env.HEARTBEAT_URL); +} + +// The tunnel is a separate failure domain with its own external probe, so it +// does not gate the worker's ping; it only raises an alert while the host can +// still send one. +async function checkTunnelReady(url = process.env.CLOUDFLARED_READY_URL, fetchImpl = fetch) { + if (!url) return; + let detail; + try { + const res = await fetchImpl(url, { signal: AbortSignal.timeout(3000) }); + if (res.ok) return; + detail = `status ${res.status}`; + } catch (err) { + detail = err instanceof Error ? err.message : String(err); + } + logger.warn("cloudflared_not_ready", { detail }); + await alerts.send("cloudflared_not_ready", `Cloudflare tunnel not ready\n${detail}`, { + windowSeconds: LOOP_ALERT_WINDOW_SECONDS, + }); +} + // Graceful-shutdown state. On SIGTERM/SIGINT we stop scheduling new work, // let the in-flight delivery batch drain (bounded), then close the pool and // redis so a deploy does not sever connections mid-write. @@ -369,6 +466,7 @@ async function processWebhookBatch() { } catch (err) { logger.error("webhook_batch_error", { error: err }); await alertLoopError("webhook_batch_error", err); + return false; } } @@ -488,6 +586,7 @@ async function sweepHygiene() { } catch (err) { logger.error("hygiene_sweep_error", { error: err }); await alertLoopError("hygiene_sweep_error", err); + return false; } } @@ -541,6 +640,8 @@ async function sweepExpiredActivations() { } } catch (err) { logger.error("activation_expiry_sweep_error", { error: err }); + await alertLoopError("activation_expiry_sweep_error", err); + return false; } } @@ -603,6 +704,7 @@ async function sweepRestrictedInactive() { } catch (err) { logger.error("restriction_sweep_error", { error: err }); await alertLoopError("restriction_sweep_error", err); + return false; } } @@ -626,7 +728,7 @@ async function sweepPendingDeletions() { } catch (err) { logger.error("deletion_sweep_error", { error: err }); await alertLoopError("deletion_sweep_error", err); - return; + return false; } let purged = 0; @@ -705,10 +807,15 @@ async function sweepPendingDeletions() { // by tests without opening a Redis connection or starting the timers. // Runs a periodic job while tracking it as in-flight so graceful shutdown can // wait for it to finish. Skips scheduling once shutdown has begun. -function runBatch(fn, errorEvent) { +function runBatch(fn, errorEvent, loop) { if (isShuttingDown) return; inFlightBatches += 1; fn() + .then(result => { + // Loop bodies swallow their own errors and resolve false; a throw + // lands in catch. Either way the loop is not marked fresh. + if (result !== false) markLoopOk(loop); + }) .catch(err => { logger.error(errorEvent, { error: err }); return alertLoopError(errorEvent, err); @@ -797,40 +904,52 @@ function startWorker() { // Poll roughly once a second. A crashed worker leaves pending rows in // the DB; they are picked up on next start. - intervalIds.push(setInterval(() => runBatch(processWebhookBatch, "webhook_loop_error"), 1000)); + intervalIds.push(setInterval(() => runBatch(processWebhookBatch, "webhook_loop_error", "webhook_delivery"), 1000)); logger.info("webhook_delivery_loop_started"); - intervalIds.push(setInterval(() => runBatch(sweepHygiene, "hygiene_loop_error"), 60 * 60 * 1000)); + intervalIds.push(setInterval(() => runBatch(sweepHygiene, "hygiene_loop_error", "hygiene"), 60 * 60 * 1000)); // Run once at startup so the first sweep doesn't wait an hour. - runBatch(sweepHygiene, "initial_hygiene_sweep_error"); + runBatch(sweepHygiene, "initial_hygiene_sweep_error", "hygiene"); logger.info("hygiene_sweep_started"); // Activations carry a short TTL (minutes), so sweep every minute to // fire activation.expired close to the actual lapse. - intervalIds.push(setInterval(() => runBatch(sweepExpiredActivations, "activation_sweep_loop_error"), 60 * 1000)); - runBatch(sweepExpiredActivations, "initial_activation_sweep_error"); + intervalIds.push(setInterval(() => runBatch(sweepExpiredActivations, "activation_sweep_loop_error", "activation_expiry"), 60 * 1000)); + runBatch(sweepExpiredActivations, "initial_activation_sweep_error", "activation_expiry"); logger.info("activation_expiry_sweep_started"); // Restricted accounts inactive for the threshold get auto-banned (the case is // closed). Hourly is plenty for a 60-day clock. - intervalIds.push(setInterval(() => runBatch(sweepRestrictedInactive, "restriction_sweep_loop_error"), 60 * 60 * 1000)); - runBatch(sweepRestrictedInactive, "initial_restriction_sweep_error"); + intervalIds.push(setInterval(() => runBatch(sweepRestrictedInactive, "restriction_sweep_loop_error", "restriction"), 60 * 60 * 1000)); + runBatch(sweepRestrictedInactive, "initial_restriction_sweep_error", "restriction"); logger.info("restriction_sweep_started"); // Soft-deleted accounts past their grace window get purged. Hourly is plenty // for a 30-day clock. - intervalIds.push(setInterval(() => runBatch(sweepPendingDeletions, "deletion_sweep_loop_error"), 60 * 60 * 1000)); - runBatch(sweepPendingDeletions, "initial_deletion_sweep_error"); + intervalIds.push(setInterval(() => runBatch(sweepPendingDeletions, "deletion_sweep_loop_error", "deletion"), 60 * 60 * 1000)); + runBatch(sweepPendingDeletions, "initial_deletion_sweep_error", "deletion"); logger.info("deletion_sweep_started"); // Hourly tick; the hour gate and the 36h NX window inside make it one // send per UTC day. Running once at start covers a restart during the // digest hour. - const digest = () => sendDailyDigest({ pool, alerts, redis: opsRedis, hourUtc: DIGEST_HOUR_UTC, startedAt }); - intervalIds.push(setInterval(() => runBatch(digest, "digest_loop_error"), 60 * 60 * 1000)); - runBatch(digest, "initial_digest_error"); + // sendDailyDigest resolves false outside the digest hour; that is a skip, + // not a failure, so the loop still counts as fresh. + const digest = async () => { + await sendDailyDigest({ pool, alerts, redis: opsRedis, hourUtc: DIGEST_HOUR_UTC, startedAt }); + }; + intervalIds.push(setInterval(() => runBatch(digest, "digest_loop_error", "digest"), 60 * 60 * 1000)); + runBatch(digest, "initial_digest_error", "digest"); logger.info("daily_digest_started", { hourUtc: DIGEST_HOUR_UTC }); + intervalIds.push(setInterval(() => runBatch(heartbeatTick, "heartbeat_error"), HEARTBEAT_INTERVAL_MS)); + intervalIds.push(setInterval(() => runBatch(checkTunnelReady, "tunnel_check_error"), HEARTBEAT_INTERVAL_MS)); + logger.info("heartbeat_started", { + intervalMs: HEARTBEAT_INTERVAL_MS, + heartbeatUrl: Boolean(process.env.HEARTBEAT_URL), + tunnelCheck: Boolean(process.env.CLOUDFLARED_READY_URL), + }); + process.on("SIGTERM", () => shutdownGracefully("SIGTERM")); process.on("SIGINT", () => shutdownGracefully("SIGINT")); @@ -895,4 +1014,10 @@ module.exports = { MAX_ATTEMPTS, AUTO_DISABLE_THRESHOLD, CLAIM_BATCH_SIZE, + LOOP_TOLERANCE_MS, + markLoopOk, + staleLoops, + _resetLoopStateForTests, + pingHeartbeat, + checkTunnelReady, };