diff --git a/apps/server/src/libs/errors/utils.ts b/apps/server/src/libs/errors/utils.ts index 57e2767d3..51b114d40 100644 --- a/apps/server/src/libs/errors/utils.ts +++ b/apps/server/src/libs/errors/utils.ts @@ -12,6 +12,8 @@ import type { Context } from "hono"; import { HTTPException } from "hono/http-exception"; import { ZodError } from "zod"; +import { redactHeartbeatToken } from "@/libs/heartbeat/capture"; + const logger = getLogger("api-server"); export class OpenStatusApiError extends HTTPException { public readonly code: ErrorCode; @@ -89,7 +91,7 @@ export function handleError(err: Error, c: Context): Response { stack: err.stack, }, method: c.req.method, - url: c.req.url, + url: redactHeartbeatToken(c.req.url), }); c.get("sentry").captureException(err); diff --git a/apps/server/src/libs/heartbeat/batcher.test.ts b/apps/server/src/libs/heartbeat/batcher.test.ts index 973f56255..cb1928b9c 100644 --- a/apps/server/src/libs/heartbeat/batcher.test.ts +++ b/apps/server/src/libs/heartbeat/batcher.test.ts @@ -1,5 +1,5 @@ import { Duration, Effect, Schedule } from "effect"; -import { describe, expect, test } from "vitest"; +import { afterEach, describe, expect, test, vi } from "vitest"; import { Batcher, retrySchedule } from "./batcher"; @@ -13,6 +13,10 @@ const options = { const instant = Schedule.recurs(2); describe("Batcher", () => { + afterEach(() => { + vi.useRealTimers(); + }); + test("sends full batches immediately and the rest on flush", async () => { const sent: number[][] = []; const batcher = new Batcher( @@ -24,11 +28,32 @@ describe("Batcher", () => { ); for (const n of [1, 2, 3, 4]) batcher.add(n); - await batcher.flush(); + expect(sent).toEqual([[1, 2, 3]]); + await batcher.flush(); expect(sent).toEqual([[1, 2, 3], [4]]); }); + test("sends a partial batch once the interval elapses", async () => { + vi.useFakeTimers(); + const sent: number[][] = []; + const batcher = new Batcher( + async (events) => { + sent.push(events); + }, + options, + instant, + ); + + batcher.add(1); + batcher.add(2); + await vi.advanceTimersByTimeAsync(options.flushIntervalMs - 1); + expect(sent).toEqual([]); + + await vi.advanceTimersByTimeAsync(1); + expect(sent).toEqual([[1, 2]]); + }); + test("retries a failing batch until it goes through", async () => { let calls = 0; const batcher = new Batcher( @@ -83,6 +108,37 @@ describe("Batcher", () => { }); }); + test("counts batches still retrying against the buffer cap", async () => { + let calls = 0; + let release = () => {}; + const sent: number[] = []; + const batcher = new Batcher( + async (events) => { + if (calls++ === 0) { + await new Promise((resolve) => { + release = resolve; + }); + throw new Error("429"); + } + sent.push(...events); + }, + options, + instant, + ); + + // 3 + 2 fill the cap of 5 while the first batch hangs in its first attempt + for (let n = 0; n < 6; n++) batcher.add(n); + release(); + await batcher.flush(); + + expect(sent.sort()).toEqual([0, 1, 2, 3, 4]); + + // once sent, the cap frees up again + batcher.add(6); + await batcher.flush(); + expect(sent).toContain(6); + }); + test("drops events past the buffer cap", async () => { const sent: number[] = []; const batcher = new Batcher( diff --git a/apps/server/src/libs/heartbeat/batcher.ts b/apps/server/src/libs/heartbeat/batcher.ts index 3756d0402..c504c8e3f 100644 --- a/apps/server/src/libs/heartbeat/batcher.ts +++ b/apps/server/src/libs/heartbeat/batcher.ts @@ -8,6 +8,7 @@ export type BatcherOptions = { flushIntervalMs: number; /** How long one batch may keep retrying (send time included) before it is dropped. */ retryBudgetMs: number; + /** Cap on events held at once, buffered or in a batch still retrying. */ maxBufferSize: number; }; @@ -34,6 +35,8 @@ export class Batcher { private buffer: T[] = []; private timer: ReturnType | null = null; private inFlight = new Set>(); + /** Buffered events plus those in batches not yet sent or dropped. */ + private pending = 0; constructor( private readonly send: (events: T[]) => Promise, @@ -44,12 +47,13 @@ export class Batcher { ) {} add(event: T): void { - if (this.buffer.length >= this.options.maxBufferSize) { + if (this.pending >= this.options.maxBufferSize) { logger.warn("heartbeat batcher buffer full, dropping event", { - buffer_size: this.buffer.length, + pending: this.pending, }); return; } + this.pending++; this.buffer.push(event); if (this.buffer.length >= this.options.maxBatchSize) { this.flushNow(); @@ -78,6 +82,7 @@ export class Batcher { while (this.buffer.length > 0) { const batch = this.buffer.splice(0, this.options.maxBatchSize); const task = this.sendWithRetry(batch).finally(() => { + this.pending -= batch.length; this.inFlight.delete(task); }); this.inFlight.add(task); diff --git a/apps/server/src/libs/heartbeat/capture.test.ts b/apps/server/src/libs/heartbeat/capture.test.ts index 3424dad58..d3a69bfb6 100644 --- a/apps/server/src/libs/heartbeat/capture.test.ts +++ b/apps/server/src/libs/heartbeat/capture.test.ts @@ -37,11 +37,24 @@ describe("captureHeaders", () => { expect(captureHeaders(new Headers({ host: "x" }))).toBeNull(); }); - test("cuts the serialised headers at 4 KB", () => { - const headers = new Headers({ "x-big": "a".repeat(10_000) }); - expect( - new TextEncoder().encode(captureHeaders(headers) ?? "").byteLength, - ).toBeLessThanOrEqual(4 * 1024); + test("leaves out whole entries past 4 KB and stays valid JSON", () => { + const headers = new Headers({ + "x-big": "a".repeat(10_000), + "x-medium-1": "b".repeat(2_500), + "x-medium-2": "c".repeat(2_500), + "content-type": "text/plain", + }); + + const captured = captureHeaders(headers) ?? ""; + + expect(new TextEncoder().encode(captured).byteLength).toBeLessThanOrEqual( + 4 * 1024, + ); + const parsed = JSON.parse(captured); + expect(parsed["content-type"]).toBe("text/plain"); + expect(parsed["x-big"]).toBeUndefined(); + expect(parsed["x-medium-1"]).toHaveLength(2_500); + expect(parsed["x-medium-2"]).toBeUndefined(); }); }); @@ -79,11 +92,29 @@ describe("readCappedBody", () => { }); test("does not leave a broken multi-byte character at the cut", async () => { + // the ASCII prefix puts the 10 KB cut in the middle of an `é` const result = await readCappedBody( - new Request("http://x", { method: "POST", body: "é".repeat(6_000) }), + new Request("http://x", { + method: "POST", + body: `a${"é".repeat(6_000)}`, + }), ); expect(result.truncated).toBe(true); - expect(result.body?.endsWith("é")).toBe(true); + expect(result.body).toBe(`a${"é".repeat((MAX_BODY_BYTES - 2) / 2)}`); + }); + + test("scrubs secrets from the body", async () => { + const result = await readCappedBody( + new Request("http://x", { + method: "POST", + body: JSON.stringify({ status: "ok", api_key: "sk_live_123" }), + }), + ); + expect(result.body).not.toContain("sk_live_123"); + expect(JSON.parse(result.body ?? "{}")).toEqual({ + status: "ok", + api_key: "[redacted]", + }); }); test("returns null without a body", async () => { diff --git a/apps/server/src/libs/heartbeat/capture.ts b/apps/server/src/libs/heartbeat/capture.ts index cf17341a8..8ee9bd9a1 100644 --- a/apps/server/src/libs/heartbeat/capture.ts +++ b/apps/server/src/libs/heartbeat/capture.ts @@ -1,4 +1,7 @@ -import { redactSensitiveHeaders } from "@openstatus/services/monitor"; +import { + redactSensitiveBody, + redactSensitiveHeaders, +} from "@openstatus/services/monitor"; export const MAX_BODY_BYTES = 10 * 1024; export const MAX_HEADERS_BYTES = 4 * 1024; @@ -15,7 +18,10 @@ const DROPPED_HEADERS = new Set([ const DROPPED_PREFIXES = ["fly-", "x-forwarded-"]; const SENSITIVE_NAME = /token|secret|key|password|signature/i; -/** Redacted, infrastructure noise removed, serialised and cut at 4 KB. */ +/** + * Redacted, infrastructure noise removed and serialised. Entries that would + * push it past 4 KB are left out whole, so the stored value stays valid JSON. + */ export function captureHeaders(headers: Headers): string | null { const kept: Record = {}; headers.forEach((value, rawName) => { @@ -24,9 +30,19 @@ export function captureHeaders(headers: Headers): string | null { if (DROPPED_PREFIXES.some((p) => name.startsWith(p))) return; kept[name] = SENSITIVE_NAME.test(name) ? REDACTED : value; }); - const redacted = redactSensitiveHeaders(kept); - if (Object.keys(redacted).length === 0) return null; - return truncateUtf8(JSON.stringify(redacted), MAX_HEADERS_BYTES).text; + const fitted: Record = {}; + let size = 2; // {} + for (const [name, value] of Object.entries(redactSensitiveHeaders(kept))) { + // `"name":"value"` plus the comma before it + const entry = byteLength( + `${JSON.stringify(name)}:${JSON.stringify(value)}`, + ); + if (size + entry + 1 > MAX_HEADERS_BYTES) continue; + fitted[name] = value; + size += entry + 1; + } + if (Object.keys(fitted).length === 0) return null; + return JSON.stringify(fitted); } export function captureUserAgent(headers: Headers): string | null { @@ -42,7 +58,10 @@ export function captureIp(headers: Headers): string | null { return forwarded?.split(",").at(-1)?.trim() || null; } -/** Reads at most `limit` bytes and cancels the rest, so a huge POST is never buffered. */ +/** + * Reads at most `limit` bytes and cancels the rest, so a huge POST is never + * buffered. Secrets are scrubbed after the cut, like stored response bodies. + */ export async function readCappedBody( request: Request, limit = MAX_BODY_BYTES, @@ -76,21 +95,14 @@ export async function readCappedBody( offset += chunk.byteLength; } // a cut may land inside a multi-byte character; drop the partial one - return { - body: new TextDecoder("utf-8", { fatal: false }) - .decode(bytes) - .replace(/�$/, ""), - truncated, - }; + const text = new TextDecoder("utf-8", { fatal: false }) + .decode(bytes) + .replace(/\uFFFD$/, ""); + return { body: redactSensitiveBody(text), truncated }; } -function truncateUtf8(text: string, limit: number) { - const bytes = new TextEncoder().encode(text); - if (bytes.byteLength <= limit) return { text, truncated: false }; - return { - text: new TextDecoder().decode(bytes.subarray(0, limit)).replace(/�$/, ""), - truncated: true, - }; +function byteLength(text: string): number { + return new TextEncoder().encode(text).byteLength; } /** Ping URLs carry the token; keep it out of request logs. */ diff --git a/apps/server/src/libs/middlewares/concurrency.test.ts b/apps/server/src/libs/middlewares/concurrency.test.ts index 4db594f9d..829ef8eb9 100644 --- a/apps/server/src/libs/middlewares/concurrency.test.ts +++ b/apps/server/src/libs/middlewares/concurrency.test.ts @@ -32,6 +32,7 @@ function build(maxInFlight: number) { throw new Error("boom"); }); app.get("/ping", (c) => c.text("pong")); + app.get("/heartbeat/:token", (c) => c.text("OK")); app.post("/rpc/*", (c) => c.text("ok")); app.onError((_, c) => c.text("error", 500)); @@ -91,6 +92,16 @@ describe("concurrency guard", () => { await held; }); + test("heartbeat pings are never shed", async () => { + const { app, guard, release } = build(1); + const held = app.request("/slow"); + await tick(); + expect(guard.inFlight()).toBe(1); + expect((await app.request("/heartbeat/some-token")).status).toBe(200); + release(); + await held; + }); + test("shed response uses the standard error envelope", async () => { const { app, release } = build(1); const held = app.request("/slow"); diff --git a/apps/server/src/libs/middlewares/concurrency.ts b/apps/server/src/libs/middlewares/concurrency.ts index 72059cced..92380f49a 100644 --- a/apps/server/src/libs/middlewares/concurrency.ts +++ b/apps/server/src/libs/middlewares/concurrency.ts @@ -5,13 +5,20 @@ import { shedResponse, wideEvent } from "./shed"; /** Health checks must never be shed or Fly restarts a machine that is merely busy. */ const EXEMPT_PATHS = new Set(["/ping"]); +/** A shed heartbeat ping reads as a missed one and alerts; the per-token limit bounds them. */ +const EXEMPT_PREFIXES = ["/heartbeat/"]; /** `opts.maxInFlight` is read per request so the shared `limits` object stays live. */ export function createConcurrencyGuard(opts: { maxInFlight: number }) { let inFlight = 0; const middleware: MiddlewareHandler = async (c, next) => { - if (EXEMPT_PATHS.has(c.req.path)) return next(); + if ( + EXEMPT_PATHS.has(c.req.path) || + EXEMPT_PREFIXES.some((p) => c.req.path.startsWith(p)) + ) { + return next(); + } const event = wideEvent(c); if (inFlight >= opts.maxInFlight) { diff --git a/apps/server/src/libs/middlewares/limits.ts b/apps/server/src/libs/middlewares/limits.ts index 34b921764..13ddc57fc 100644 --- a/apps/server/src/libs/middlewares/limits.ts +++ b/apps/server/src/libs/middlewares/limits.ts @@ -1,6 +1,11 @@ import { env } from "@/env"; -/** `skipValidation` hands back raw strings, so coerce; missing, non-numeric or zero values fall back. */ +/** `skipValidation` hands back raw strings, so coerce; anything but a positive integer falls back. */ +function positiveInt(value: unknown, fallback: number): number { + const n = Number(value); + return Number.isSafeInteger(n) && n > 0 ? n : fallback; +} + export const limits = { /** * Conservative default for self-hosted single-container deploys. Fly @@ -8,11 +13,13 @@ export const limits = { * above `soft_limit` and at most `hard_limit` so Fly scales out before the * app starts shedding. */ - maxInFlight: Number(env.API_MAX_IN_FLIGHT) || 128, - perMinute: Number(env.API_RATE_LIMIT_PER_MINUTE) || 600, - burstPer10s: Number(env.API_RATE_LIMIT_BURST_PER_10S) || 100, - writesPerMinute: Number(env.API_RATE_LIMIT_WRITES_PER_MINUTE) || 60, - publicPerMinute: Number(env.API_RATE_LIMIT_PUBLIC_PER_MINUTE) || 120, - heartbeatPerTokenPerMinute: - Number(env.API_RATE_LIMIT_HEARTBEAT_PER_MINUTE) || 6, + maxInFlight: positiveInt(env.API_MAX_IN_FLIGHT, 128), + perMinute: positiveInt(env.API_RATE_LIMIT_PER_MINUTE, 600), + burstPer10s: positiveInt(env.API_RATE_LIMIT_BURST_PER_10S, 100), + writesPerMinute: positiveInt(env.API_RATE_LIMIT_WRITES_PER_MINUTE, 60), + publicPerMinute: positiveInt(env.API_RATE_LIMIT_PUBLIC_PER_MINUTE, 120), + heartbeatPerTokenPerMinute: positiveInt( + env.API_RATE_LIMIT_HEARTBEAT_PER_MINUTE, + 6, + ), }; diff --git a/apps/server/src/serve.ts b/apps/server/src/serve.ts index ed38f922b..1258b5dc8 100644 --- a/apps/server/src/serve.ts +++ b/apps/server/src/serve.ts @@ -6,6 +6,7 @@ import { dispose, getLogger } from "@logtape/logtape"; import { env } from "./env"; import { app } from "./index"; import { settleBackgroundTasks } from "./libs/background"; +import { heartbeatBatcher } from "./routes/heartbeat"; const port = 3000; @@ -31,7 +32,8 @@ const server = serve({ fetch: app.fetch, port }); // Fly sends kill_signal (SIGTERM, see fly.toml) on deploy and stop, then // SIGKILLs after kill_timeout. Stop accepting connections, let in-flight // requests and the background work they started (Slack acks, see -// runInBackground) finish, then flush buffered OTLP logs before exiting. +// runInBackground) finish, send buffered heartbeat pings to Tinybird, then +// flush buffered OTLP logs before exiting. process.once("SIGTERM", () => { void (async () => { getLogger("api-server").info("SIGTERM received, draining connections"); @@ -39,6 +41,9 @@ process.once("SIGTERM", () => { await new Promise((resolve) => server.close(() => resolve())); // After the HTTP drain: a request finishing late can still spawn a task. await settleBackgroundTasks(); + // Last: a late ping still lands in the buffer. Bounded by the race below, + // which cuts a batch still retrying against a Tinybird outage. + await heartbeatBatcher.flush(); })(); await Promise.race([drained, sleep(SHUTDOWN_GRACE_MS)]); await dispose(); diff --git a/packages/services/src/monitor/__tests__/heartbeat.test.ts b/packages/services/src/monitor/__tests__/heartbeat.test.ts index 1ea14faa5..132652401 100644 --- a/packages/services/src/monitor/__tests__/heartbeat.test.ts +++ b/packages/services/src/monitor/__tests__/heartbeat.test.ts @@ -25,6 +25,7 @@ import { cloneMonitor } from "../clone"; import { createMonitor } from "../create"; import { getHeartbeatState, + recordHeartbeatPing, regenerateHeartbeatToken, updateMonitorHeartbeat, } from "../heartbeat"; @@ -454,6 +455,26 @@ describe("getHeartbeatState", () => { }); }); +describe("recordHeartbeatPing", () => { + test("an older ping landing late does not move the deadline back", async () => { + await withTestTransaction(async (tx) => { + const row = await createHeartbeat(tx); + const token = row.heartbeatToken ?? ""; + const newer = new Date("2026-01-01T00:01:00.000Z"); + const older = new Date("2026-01-01T00:00:00.000Z"); + + await recordHeartbeatPing({ token, at: newer, db: tx }); + await recordHeartbeatPing({ token, at: older, db: tx }); + + const state = await readState(tx, row.id); + expect(state?.lastPingAt?.getTime()).toBe(newer.getTime()); + expect(state?.nextDueAt?.getTime()).toBe( + newer.getTime() + (HOUR + FIVE_MIN) * 1000, + ); + }); + }); +}); + describe("triggerMonitorRun (heartbeat)", () => { test("is rejected", async () => { await withTestTransaction(async (tx) => { diff --git a/packages/services/src/monitor/heartbeat.ts b/packages/services/src/monitor/heartbeat.ts index c405e4fcb..a173d6675 100644 --- a/packages/services/src/monitor/heartbeat.ts +++ b/packages/services/src/monitor/heartbeat.ts @@ -1,4 +1,4 @@ -import { and, db as defaultDb, eq, isNull } from "@openstatus/db"; +import { and, db as defaultDb, eq, isNull, lt, or } from "@openstatus/db"; import { type HeartbeatPingKind, monitor, @@ -245,6 +245,11 @@ export async function recordHeartbeatPing(args: { .onConflictDoUpdate({ target: monitorHeartbeatState.monitorId, set: { lastPingAt: args.at, nextDueAt }, + // a slower, older ping must not pull the deadline back + setWhere: or( + isNull(monitorHeartbeatState.lastPingAt), + lt(monitorHeartbeatState.lastPingAt, args.at), + ), }); return { kind: "recorded", ...ids }; } diff --git a/packages/services/src/monitor/index.ts b/packages/services/src/monitor/index.ts index 1b0ea1189..c62969bc5 100644 --- a/packages/services/src/monitor/index.ts +++ b/packages/services/src/monitor/index.ts @@ -45,7 +45,10 @@ export { updateMonitorSchedulingRegions, updateMonitorTags, } from "./relations"; -export { redactSensitiveHeaders } from "./response-logs-internal"; +export { + redactSensitiveBody, + redactSensitiveHeaders, +} from "./response-logs-internal"; export { type CheckResult, type CheckResultError,