diff --git a/.changeset/feat-f8-event-loop-lag-observability.md b/.changeset/feat-f8-event-loop-lag-observability.md new file mode 100644 index 0000000..8396dbb --- /dev/null +++ b/.changeset/feat-f8-event-loop-lag-observability.md @@ -0,0 +1,26 @@ +--- +"sql-fs-api": minor +--- + +feat(observability): event-loop lag monitoring for the Redis leases (F8). + +The exec-lock writer lease, the RW-lock writer flag, and the RW-lock reader ZSET +scores are all kept alive by `setTimeout` heartbeats that silently assume timers +fire on schedule. A long event-loop stall (a V8 GC pause or a pathological +synchronous bash stretch) can fire a renewal past the lease, voiding it — Lock 3 +keeps Postgres consistent, so this was always an observability gap, not a +correctness bug, but nothing measured it. + +New `src/api/event-loop-monitor.ts` (purely observational, no behavior change): + +- A `perf_hooks.monitorEventLoopDelay` histogram started at boot, sampled every + `EVENT_LOOP_MONITOR_INTERVAL_MS` (default 10s) and logged as + `event:"event_loop_lag"` (`p50Ms`/`p99Ms`/`maxMs`/`meanMs`), then reset. +- Per-heartbeat gap measurement wired into all three lease sites: each heartbeat + reports actual-minus-expected fire time as `event:"heartbeat_gap"` at + `severity:"warn"` (gap > renewMs) or `"critical"` (gap > leaseMs), tagged with + the lock kind (`exec`/`rw-writer`/`rw-reader`) and key. + +Alert thresholds are documented in DEVELOPER.md ("Lock observability"). End-to-end +smoke tests reproduce a >lease stall on each lease and assert the critical +heartbeat_gap fires (with a no-stall control proving no false positives). diff --git a/CLAUDE.md b/CLAUDE.md index 5a6c693..4768462 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -219,6 +219,7 @@ const TABLE = Object.assign(Object.create(null) as Record, { | `REDIS_RWLOCK_READER_LEASE_MS` | No (default: 60000) | TTL (ms) for reader entries in the distributed RW lock ZSET. Bounds the time a writer must wait for a crashed reader to be reaped. | | `REDIS_PATH_SNAPSHOT_ENABLED` | No (default: false) | Set to `true` to enable the Redis path snapshot cache. When enabled, cold-start pathCache is loaded from Redis instead of a full Postgres `loadAllPaths` scan. Requires `REDIS_URL`. | | `REDIS_PATH_SNAPSHOT_TTL_MS` | No (default: 3600000) | TTL for path snapshot entries (ms, default 1h). | +| `EVENT_LOOP_MONITOR_INTERVAL_MS` | No (default: 10000) | Sampling interval (ms) for the F8 event-loop lag monitor. Each window logs `event:"event_loop_lag"` (`p50Ms`/`p99Ms`/`maxMs`/`meanMs`) then resets the histogram. Purely observational; pairs with per-heartbeat `event:"heartbeat_gap"` warn/critical events that flag a stall eating into a Redis lease (see DEVELOPER.md "Lock observability"). | | `JUST_BASH_DEFENSE_IN_DEPTH` | No (default: `false`) | Enables just-bash's defense-in-depth security layer (monkey-patches `setTimeout`, `eval`, `Function`, dynamic `import`, etc. for the duration of `bash.exec`). All Postgres I/O is wrapped in `DefenseInDepthBox.runTrustedAsync` to remain compatible. | | `JUST_BASH_DEFENSE_AUDIT_MODE` | No (default: `true`) | When `JUST_BASH_DEFENSE_IN_DEPTH=true`, controls whether violations throw (`false`) or are logged only (`true`). Recommended `true` for initial rollout, then flip to `false` once logs are clean. | | `BLOB_GC_MIN_AGE_MS` | No (default: 10800000) | Grace window (ms, default 3h) before an orphan blob becomes collectible by `pnpm db:gc`. Orphans whose `last_referenced_at` is newer than this are kept; the `ON CONFLICT DO UPDATE` row lock on blob writes is the actual dedup re-adoption race guard — this window is churn/margin control. Rows with NULL `last_referenced_at` (legacy, pre-migration-0006) are treated as ancient and always collectible. Override per-run with `--min-age-ms`. | diff --git a/DEVELOPER.md b/DEVELOPER.md index 65b0a65..347e77f 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -202,6 +202,19 @@ Bypassing the lock to write directly to `SqlFs` from elsewhere in the codebase w | `DELETE /sandboxes/:id` races an active exec | drains via exclusive acquire | blocks (routed through lock) | backstop | | Admin/test route bypasses `withSession` | — | — | still catches | +### Lock observability — event-loop lag (F8) + +Every lease above (the exec-lock writer lease, the RW-lock writer flag, and the RW-lock reader ZSET scores) is kept alive by a `setTimeout` heartbeat that renews well before expiry (`renewMs` 20s < `leaseMs` 60s). They all assume the timer fires roughly on schedule. The GC-pause failure mode in the table above is exactly a violation of that assumption: when the event loop stalls longer than the lease, the renewal fires too late, Redis has already expired the key / reaped the reader entry, and the next renew returns 0 → `LockLostError`. Lock 3 keeps the DB consistent, so this is an **observability** concern, not a correctness one — but until F8 nothing measured how close the process ran to the lease floor. + +`src/api/event-loop-monitor.ts` is purely observational and emits two structured log events: + +| Event | Emitted by | Fields | Meaning / alert threshold | +|---|---|---|---| +| `event_loop_lag` | boot sampler (`startEventLoopMonitor`, every `EVENT_LOOP_MONITOR_INTERVAL_MS`, default 10s) | `p50Ms`, `p99Ms`, `maxMs`, `meanMs`, `windowMs` | Process-wide event-loop delay from `perf_hooks.monitorEventLoopDelay`. **Page** when `p99Ms`/`maxMs` approaches `renewMs` (~20s) — the lease margin is being eaten. | +| `heartbeat_gap` | each heartbeat callback (`recordHeartbeatGap`) | `severity`, `lock` (`exec`\|`rw-writer`\|`rw-reader`), `key`, `gapMs`, `renewMs`, `leaseMs` | Actual-minus-expected fire time for one heartbeat. `severity:"warn"` at `gapMs > renewMs` (a full interval late); `severity:"critical"` at `gapMs > leaseMs` (the lease almost certainly lapsed). Silent below `renewMs`. | + +`warn`/`critical` go to `console.warn`/`console.error` respectively; the sampler logs at `console.log`. A `critical` `heartbeat_gap` for `exec`/`rw-writer` is the breadcrumb that explains a subsequent `LockLostError`; for `rw-reader` it flags the window during which a writer could have reaped the stale reader entry and entered mid-read. `eventLoopLagSnapshot()` exposes the live histogram for a future health endpoint without perturbing the windowed log. + --- ## ReadOnly Safety Model diff --git a/src/api/distributed-lock.ts b/src/api/distributed-lock.ts index e5b8a77..36e4e5f 100644 --- a/src/api/distributed-lock.ts +++ b/src/api/distributed-lock.ts @@ -15,6 +15,7 @@ import { DEFAULT_ACQUIRE_ERROR_BUDGET_MS, getRedisCircuitBreaker, } from "../redis/circuit-breaker.js"; +import { recordHeartbeatGap } from "./event-loop-monitor.js"; const RELEASE_SCRIPT = ` if redis.call("get", KEYS[1]) == ARGV[1] then @@ -192,8 +193,13 @@ export async function withDistributedLock( const renewRetryMs = Math.max(50, Math.floor(renewMs / 4)); const scheduleRenew = (delayMs: number): void => { if (stopped) return; + // F8: capture when this tick is *expected* to fire so the callback can + // measure event-loop lag (a GC pause / sync stall that delays renewal past + // the lease is otherwise silent until LockLostError surfaces post-hoc). + const expectedFireAt = Date.now() + delayMs; renewTimer = setTimeout(async () => { if (stopped) return; + recordHeartbeatGap({ lock: "exec", key, expectedFireAt, renewMs, leaseMs }); // Cap renewal command wait so a stalled Redis can't block release. let outcome: "renewed" | "ownership_lost" | "transient"; // L1: track and clear the race timeout so a won race doesn't leave a diff --git a/src/api/distributed-rw-lock.ts b/src/api/distributed-rw-lock.ts index 593043e..f6c3e73 100644 --- a/src/api/distributed-rw-lock.ts +++ b/src/api/distributed-rw-lock.ts @@ -19,6 +19,7 @@ import { } from "../redis/circuit-breaker.js"; export { LockAcquireTimeoutError, LockLostError } from "./distributed-lock.js"; import { LockAcquireTimeoutError, LockLostError, jitteredDelayMs } from "./distributed-lock.js"; +import { recordHeartbeatGap } from "./event-loop-monitor.js"; // ── Lua scripts ────────────────────────────────────────────────────────────── @@ -222,8 +223,12 @@ function startSharedHeartbeat( const schedule = (delayMs: number): void => { if (stopped) return; + // F8: a stall that delays this tick past the reader lease lets a writer + // enter mid-read; measure the gap (readerLeaseMs is the reader's "lease"). + const expectedFireAt = Date.now() + delayMs; timer = setTimeout(async () => { if (stopped) return; + recordHeartbeatGap({ lock: "rw-reader", key: keys.readers, expectedFireAt, renewMs, leaseMs: readerLeaseMs }); let outcome: "renewed" | "ownership_lost" | "transient"; let raceTimeout: ReturnType | undefined; // L1: clear on settle try { @@ -349,8 +354,12 @@ function startExclusiveHeartbeat( const schedule = (delayMs: number): void => { if (stopped) return; + // F8: a stall that delays this tick past the writer lease lets a peer + // replica acquire the flag while we believe we still hold it; measure it. + const expectedFireAt = Date.now() + delayMs; timer = setTimeout(async () => { if (stopped) return; + recordHeartbeatGap({ lock: "rw-writer", key: keys.writer, expectedFireAt, renewMs, leaseMs }); let outcome: "renewed" | "ownership_lost" | "transient"; let raceTimeout: ReturnType | undefined; // L1: clear on settle try { diff --git a/src/api/event-loop-monitor.ts b/src/api/event-loop-monitor.ts new file mode 100644 index 0000000..f8e3915 --- /dev/null +++ b/src/api/event-loop-monitor.ts @@ -0,0 +1,182 @@ +/** + * F8: Event-loop lag observability (purely observational — no behavior change). + * + * The three Redis leases — the exec-lock writer lease (`distributed-lock.ts`), + * the RW-lock writer flag (`distributed-rw-lock.ts` exclusive heartbeat), and the + * RW-lock reader ZSET scores (shared heartbeat) — each keep themselves alive with + * a `setTimeout`-driven heartbeat that renews well before expiry (`renewMs` < + * `leaseMs`). They silently assume the timer fires roughly on schedule. A long + * event-loop stall (a V8 GC pause, a pathological synchronous bash stretch, or — + * only when `REDIS_PATH_SNAPSHOT_ENABLED=true` — the in-lock msgpack encode) fires + * the renewal late; by then Redis has already expired the key / reaped the ZSET + * entry, another replica can step in, and the next renew returns 0 → + * `LockLostError`. Lock 3 (PG advisory + per-tx serialization, see DEVELOPER.md) + * prevents any DB corruption, so this is an *observability* gap, not a correctness + * bug: nothing measured how close the process ran to the lease floor. + * + * This module measures it, two ways: + * - {@link startEventLoopMonitor}/{@link stopEventLoopMonitor}: a + * `perf_hooks.monitorEventLoopDelay` histogram, sampled on an interval and + * logged as `event:"event_loop_lag"` (p50/p99/max ms), then reset each window. + * - {@link recordHeartbeatGap}: called from each heartbeat callback with the time + * the timer was *expected* to fire; emits `event:"heartbeat_gap"` at severity + * `"warn"` (gap > renewMs — a full renewal interval late) or `"critical"` + * (gap > leaseMs — the lease almost certainly lapsed). No-op below renewMs and + * safe to call when the monitor was never started (unit tests / no-Redis runs). + */ + +import { type IntervalHistogram, monitorEventLoopDelay } from "node:perf_hooks"; + +const NS_PER_MS = 1_000_000; + +/** Default interval (ms) between event-loop-lag samples. Override via `EVENT_LOOP_MONITOR_INTERVAL_MS`. */ +export const DEFAULT_SAMPLE_INTERVAL_MS = 10_000; +/** Default histogram resolution (ms) — how often perf_hooks records a delay reading. */ +export const DEFAULT_RESOLUTION_MS = 20; + +/** Which lease a heartbeat gap belongs to, for alert routing. */ +export type HeartbeatLockKind = "exec" | "rw-writer" | "rw-reader"; + +/** Severity of a single heartbeat gap relative to the renew/lease windows. */ +export type HeartbeatGapSeverity = "ok" | "warn" | "critical"; + +export interface EventLoopLagSnapshot { + readonly p50Ms: number; + readonly p99Ms: number; + readonly maxMs: number; + readonly meanMs: number; +} + +export interface EventLoopMonitorOptions { + /** Sampling/log interval (ms). Each window is logged, then the histogram is reset. */ + readonly sampleIntervalMs?: number; + /** perf_hooks histogram resolution (ms). */ + readonly resolutionMs?: number; + /** Injectable log sink (defaults to `console.log`). Tests pass a collector. */ + readonly log?: (line: string) => void; +} + +interface MonitorState { + readonly histogram: IntervalHistogram; + readonly timer: ReturnType; +} + +let state: MonitorState | undefined; + +function toMs(ns: number): number { + // perf_hooks reports nanoseconds; round to whole ms for log readability. + return Math.round(ns / NS_PER_MS); +} + +function readSnapshot(h: IntervalHistogram): EventLoopLagSnapshot { + return { + p50Ms: toMs(h.percentile(50)), + p99Ms: toMs(h.percentile(99)), + maxMs: toMs(h.max), + meanMs: toMs(h.mean), + }; +} + +/** + * Start the process-wide event-loop-delay monitor. Idempotent: a second call + * while already running is a no-op. The sampling timer is `unref()`'d so it never + * keeps the process alive past shutdown. + */ +export function startEventLoopMonitor(opts: EventLoopMonitorOptions = {}): void { + if (state !== undefined) return; + const sampleIntervalMs = opts.sampleIntervalMs ?? DEFAULT_SAMPLE_INTERVAL_MS; + const resolutionMs = opts.resolutionMs ?? DEFAULT_RESOLUTION_MS; + const log = opts.log ?? ((line: string): void => console.log(line)); + + const histogram = monitorEventLoopDelay({ resolution: resolutionMs }); + histogram.enable(); + + const timer = setInterval(() => { + const snapshot = readSnapshot(histogram); + histogram.reset(); + log(JSON.stringify({ event: "event_loop_lag", ...snapshot, windowMs: sampleIntervalMs })); + }, sampleIntervalMs); + if (typeof timer.unref === "function") timer.unref(); + + state = { histogram, timer }; +} + +/** Stop the monitor and disable the histogram. Idempotent. */ +export function stopEventLoopMonitor(): void { + if (state === undefined) return; + clearInterval(state.timer); + state.histogram.disable(); + state = undefined; +} + +/** + * Current event-loop-delay snapshot, or `undefined` when the monitor is not + * running. Reads the live histogram WITHOUT resetting it (the sampler owns the + * reset), so a health endpoint can poll it without perturbing the windowed log. + */ +export function eventLoopLagSnapshot(): EventLoopLagSnapshot | undefined { + if (state === undefined) return undefined; + return readSnapshot(state.histogram); +} + +/** + * Classify a heartbeat gap. Pure — exported for direct unit testing. + * - `gap > leaseMs` → "critical" (the lease has almost certainly lapsed) + * - `gap > renewMs` → "warn" (a full renewal interval late) + * - otherwise → "ok" (within normal jitter) + */ +export function classifyHeartbeatGap(gapMs: number, renewMs: number, leaseMs: number): HeartbeatGapSeverity { + if (gapMs > leaseMs) return "critical"; + if (gapMs > renewMs) return "warn"; + return "ok"; +} + +export interface HeartbeatGapInput { + readonly lock: HeartbeatLockKind; + readonly key: string; + /** `Date.now()` value at which this heartbeat timer was scheduled to fire. */ + readonly expectedFireAt: number; + readonly renewMs: number; + readonly leaseMs: number; + /** Observed fire time (ms). Defaults to `Date.now()`; injectable for tests. */ + readonly nowMs?: number; + /** Injectable log sink. Defaults to `console.warn` (warn) / `console.error` (critical). */ + readonly log?: (line: string) => void; +} + +/** + * Record one heartbeat's actual-vs-expected fire gap. Emits an `event: + * "heartbeat_gap"` structured line at "warn" or "critical" severity; below the + * renew window it is a silent no-op (the common case — every healthy tick). + * + * Returns the classified severity so callers/tests can branch without re-deriving + * it. This function never throws and never depends on {@link startEventLoopMonitor}. + */ +export function recordHeartbeatGap(input: HeartbeatGapInput): HeartbeatGapSeverity { + const now = input.nowMs ?? Date.now(); + const gapMs = now - input.expectedFireAt; + const severity = classifyHeartbeatGap(gapMs, input.renewMs, input.leaseMs); + if (severity === "ok") return "ok"; + const line = JSON.stringify({ + event: "heartbeat_gap", + severity, + lock: input.lock, + key: input.key, + gapMs, + renewMs: input.renewMs, + leaseMs: input.leaseMs, + }); + if (input.log !== undefined) { + input.log(line); + } else if (severity === "critical") { + console.error(line); + } else { + console.warn(line); + } + return severity; +} + +/** Test hook: force the monitor back to the not-started state. */ +export function resetEventLoopMonitorForTest(): void { + stopEventLoopMonitor(); +} diff --git a/src/api/server.ts b/src/api/server.ts index df60984..b2898c0 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -17,6 +17,7 @@ import { RedisPathSnapshot } from "../sql-fs/redis-path-snapshot.js"; import type { SandboxListEntry, SandboxMeta } from "../sql-fs/types.js"; import { type AuthVariables, createAuthMiddleware, loadStaticMcpAuthConfig } from "./auth.js"; import { clientSafeErrorMessage, mapFsErrorToStatus } from "./errors.js"; +import { DEFAULT_SAMPLE_INTERVAL_MS, startEventLoopMonitor, stopEventLoopMonitor } from "./event-loop-monitor.js"; import { mcpOptionsResponse, withMcpCors } from "./mcp-cors.js"; import { handleMcpRequest, shutdownMcp, startMcpSessionSweeper } from "./mcp/server.js"; import { runMigrations } from "./migrations.js"; @@ -302,11 +303,19 @@ if (isMain) { sessionManager.startReaper(); startMcpSessionSweeper(); + // F8: process-wide event-loop-lag monitor. Purely observational — surfaces + // the GC-pause / sync-stall class that can silently void a Redis lease + // (see event-loop-monitor.ts). The sampling timer is unref()'d internally. + startEventLoopMonitor({ + sampleIntervalMs: parsePositiveInt("EVENT_LOOP_MONITOR_INTERVAL_MS", DEFAULT_SAMPLE_INTERVAL_MS), + }); + let shuttingDown = false; const shutdown = (): void => { if (shuttingDown) return; shuttingDown = true; console.log(JSON.stringify({ event: "shutdown_begin" })); + stopEventLoopMonitor(); // Force-exit guard so a hung Postgres or Redis cleanup cannot keep // the process alive past the orchestrator's grace period. const forceExit = setTimeout(() => { diff --git a/src/api/tests/integration/heartbeat-gap.redis.integration.test.ts b/src/api/tests/integration/heartbeat-gap.redis.integration.test.ts new file mode 100644 index 0000000..6551c52 --- /dev/null +++ b/src/api/tests/integration/heartbeat-gap.redis.integration.test.ts @@ -0,0 +1,189 @@ +/** + * F8 Layer-2 smoke test against a REAL Redis (skipped unless REDIS_URL is set). + * + * The unit smoke test (`heartbeat-gap.smoke.test.ts`) uses a FakeRedis; this one + * proves the same behaviour end-to-end against a live Redis server, exercising + * real `SET NX PX` expiry and real Lua renew scripts: + * + * 1. A synchronous event-loop stall longer than the lease causes a genuine + * `LockLostError` (real Redis PX-expired the key) AND emits a `critical` + * `heartbeat_gap` — the previously-silent failure is now observable. + * 2. A second connection (a "peer replica") can acquire the lease that the + * stalled holder still believes it owns — the real split-brain window. + * 3. The RW-lock writer flag behaves identically. + * + * Run with: REDIS_URL=redis://localhost:6380 pnpm test:integration + */ + +import { Redis } from "ioredis"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { LockLostError, execLockKey, withDistributedLock } from "../../distributed-lock.js"; +import { type RWLockKeys, rwLockKeys, withDistributedRWLock } from "../../distributed-rw-lock.js"; + +const SKIP = !process.env.REDIS_URL; +const TENANT = "default"; + +// Lease floor crossed by a ~1.5s busy-loop; renew well inside it. +const LEASE_MS = 800; +const RENEW_MS = 250; +const STALL_MS = 1_500; +const LOCK_OPTS = { leaseMs: LEASE_MS, renewMs: RENEW_MS, acquireTimeoutMs: 8_000, acquireRetryMs: 50 }; + +function busyLoopMs(ms: number): void { + const start = Date.now(); + while (Date.now() - start < ms) { + /* block the event loop — the real GC-pause / sync-bash-stretch model */ + } +} +const sleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + +interface HeartbeatGapLine { + event: string; + severity: "warn" | "critical"; + lock: string; + key: string; + gapMs: number; +} + +/** Spy console.warn/error and surface only the parsed `heartbeat_gap` lines. */ +function captureHeartbeatGaps(): { gaps: () => HeartbeatGapLine[]; restore: () => void } { + const lines: string[] = []; + const push = (l: unknown): void => { + lines.push(String(l)); + }; + const warnSpy = vi.spyOn(console, "warn").mockImplementation(push); + const errorSpy = vi.spyOn(console, "error").mockImplementation(push); + return { + gaps: () => + lines + .map((l) => { + try { + return JSON.parse(l) as { event?: string }; + } catch { + return {}; + } + }) + .filter((o): o is HeartbeatGapLine => (o as { event?: string }).event === "heartbeat_gap"), + restore: () => { + warnSpy.mockRestore(); + errorSpy.mockRestore(); + }, + }; +} + +describe.skipIf(SKIP)("F8 Layer-2 — heartbeat gap against real Redis", () => { + let redis: Redis; + const keysToClean: string[] = []; + + beforeAll(async () => { + redis = new Redis(process.env.REDIS_URL!, { lazyConnect: true, maxRetriesPerRequest: 1 }); + await redis.connect(); + }); + + afterEach(async () => { + if (keysToClean.length > 0) await redis.del(...keysToClean); + keysToClean.length = 0; + }); + + afterAll(async () => { + await redis.quit(); + }); + + function newExecKey(suffix: string): string { + const key = execLockKey(TENANT, `f8-${suffix}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`); + keysToClean.push(key); + return key; + } + + it("exec lock: a >lease sync stall loses the lease and emits a critical heartbeat_gap", async () => { + const key = newExecKey("exec-stall"); + const cap = captureHeartbeatGaps(); + try { + await expect( + withDistributedLock( + redis, + key, + async () => { + busyLoopMs(STALL_MS); + await sleep(50); // yield so the overdue heartbeat fires in-section + }, + LOCK_OPTS, + ), + ).rejects.toBeInstanceOf(LockLostError); + } finally { + cap.restore(); + } + const critical = cap.gaps().filter((g) => g.severity === "critical" && g.lock === "exec"); + expect(critical.length).toBeGreaterThanOrEqual(1); + expect(critical[0]?.key).toBe(key); + expect(critical[0]?.gapMs).toBeGreaterThanOrEqual(LEASE_MS); + // The lease genuinely lapsed in real Redis — the key no longer holds our token. + expect(await redis.get(key)).toBeNull(); + }); + + it("exec lock: a peer connection acquires the lease the stalled holder still believes it owns", async () => { + const key = newExecKey("exec-steal"); + const peer = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: 1 }); + const cap = captureHeartbeatGaps(); + let peerAcquired = false; + try { + // Holder A stalls past its lease; peer B races to acquire the same key. + // (Single process: B's poll is blocked during A's synchronous stall, then + // finds the key PX-expired the moment the loop frees — the real handoff a + // separate replica process would win mid-stall. Lock 3 is the DB backstop.) + const holder = withDistributedLock( + redis, + key, + async () => { + busyLoopMs(STALL_MS); + await sleep(50); + }, + LOCK_OPTS, + ); + const peerLock = withDistributedLock( + peer, + key, + async () => { + peerAcquired = true; + }, + { leaseMs: 2_000, renewMs: 500, acquireTimeoutMs: 8_000, acquireRetryMs: 50 }, + ); + + await expect(holder).rejects.toBeInstanceOf(LockLostError); + await peerLock; + } finally { + cap.restore(); + await peer.quit(); + } + expect(peerAcquired).toBe(true); + const critical = cap.gaps().filter((g) => g.severity === "critical" && g.lock === "exec"); + expect(critical.length).toBeGreaterThanOrEqual(1); + }); + + it("rw-lock writer: a >lease sync stall loses the flag and emits a critical heartbeat_gap", async () => { + const keys: RWLockKeys = rwLockKeys(TENANT, `f8-rw-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`); + keysToClean.push(keys.writer, keys.readers); + const cap = captureHeartbeatGaps(); + try { + await expect( + withDistributedRWLock( + redis, + keys, + "exclusive", + async () => { + busyLoopMs(STALL_MS); + await sleep(50); + }, + { ...LOCK_OPTS, readerLeaseMs: 5_000 }, + ), + ).rejects.toBeInstanceOf(LockLostError); + } finally { + cap.restore(); + } + const critical = cap.gaps().filter((g) => g.severity === "critical" && g.lock === "rw-writer"); + expect(critical.length).toBeGreaterThanOrEqual(1); + expect(critical[0]?.key).toBe(keys.writer); + expect(critical[0]?.gapMs).toBeGreaterThanOrEqual(LEASE_MS); + expect(await redis.get(keys.writer)).toBeNull(); + }); +}); diff --git a/src/api/tests/unit/event-loop-monitor.test.ts b/src/api/tests/unit/event-loop-monitor.test.ts new file mode 100644 index 0000000..1cfe6ba --- /dev/null +++ b/src/api/tests/unit/event-loop-monitor.test.ts @@ -0,0 +1,162 @@ +/** + * Unit tests for the F8 event-loop lag monitor (purely observational). + * Covers the pure gap classifier, the structured `heartbeat_gap` emitter, and + * the `monitorEventLoopDelay`-backed boot histogram (real busy-loop, no timers + * faked — perf_hooks measures wall-clock). + */ + +import { afterEach, describe, expect, it } from "vitest"; +import { + classifyHeartbeatGap, + eventLoopLagSnapshot, + recordHeartbeatGap, + resetEventLoopMonitorForTest, + startEventLoopMonitor, + stopEventLoopMonitor, +} from "../../event-loop-monitor.js"; + +function busyLoopMs(ms: number): void { + const start = Date.now(); + while (Date.now() - start < ms) { + // Block the event loop synchronously — the exact "sync stall starves + // timers" model that silently voids a Redis lease. + } +} + +describe("classifyHeartbeatGap", () => { + it("returns ok when the gap is within the renew window", () => { + expect(classifyHeartbeatGap(10, 80, 200)).toBe("ok"); + }); + it("returns warn when the gap exceeds renewMs but not leaseMs", () => { + expect(classifyHeartbeatGap(120, 80, 200)).toBe("warn"); + }); + it("returns critical when the gap exceeds leaseMs", () => { + expect(classifyHeartbeatGap(260, 80, 200)).toBe("critical"); + }); + it("treats exactly renewMs as ok (strictly greater triggers warn)", () => { + expect(classifyHeartbeatGap(80, 80, 200)).toBe("ok"); + }); + it("treats exactly leaseMs as warn (strictly greater triggers critical)", () => { + expect(classifyHeartbeatGap(200, 80, 200)).toBe("warn"); + }); +}); + +describe("recordHeartbeatGap", () => { + it("is a silent no-op when the gap is within the renew window", () => { + const lines: string[] = []; + const severity = recordHeartbeatGap({ + lock: "exec", + key: "vfs:default:lock:sbx", + expectedFireAt: 1_000, + nowMs: 1_050, + renewMs: 80, + leaseMs: 200, + log: (l) => lines.push(l), + }); + expect(severity).toBe("ok"); + expect(lines).toEqual([]); + }); + + it("emits a warn line with the full structured payload", () => { + const lines: string[] = []; + const severity = recordHeartbeatGap({ + lock: "rw-writer", + key: "vfs:default:rwlock:{sbx}:writer", + expectedFireAt: 1_000, + nowMs: 1_150, + renewMs: 80, + leaseMs: 200, + log: (l) => lines.push(l), + }); + expect(severity).toBe("warn"); + expect(lines).toHaveLength(1); + expect(JSON.parse(lines[0]!)).toEqual({ + event: "heartbeat_gap", + severity: "warn", + lock: "rw-writer", + key: "vfs:default:rwlock:{sbx}:writer", + gapMs: 150, + renewMs: 80, + leaseMs: 200, + }); + }); + + it("emits a critical line when the gap exceeds the lease", () => { + const lines: string[] = []; + const severity = recordHeartbeatGap({ + lock: "rw-reader", + key: "vfs:default:rwlock:{sbx}:readers", + expectedFireAt: 1_000, + nowMs: 1_300, + renewMs: 80, + leaseMs: 200, + log: (l) => lines.push(l), + }); + expect(severity).toBe("critical"); + const parsed = JSON.parse(lines[0]!); + expect(parsed.severity).toBe("critical"); + expect(parsed.gapMs).toBe(300); + expect(parsed.lock).toBe("rw-reader"); + }); +}); + +describe("event-loop monitor lifecycle", () => { + afterEach(() => resetEventLoopMonitorForTest()); + + it("snapshot is undefined before start and defined after", () => { + expect(eventLoopLagSnapshot()).toBeUndefined(); + startEventLoopMonitor({ sampleIntervalMs: 100_000, log: () => {} }); + const snap = eventLoopLagSnapshot(); + expect(snap).toBeDefined(); + expect(typeof snap?.maxMs).toBe("number"); + expect(typeof snap?.p99Ms).toBe("number"); + }); + + it("double start and double stop are safe (idempotent)", () => { + startEventLoopMonitor({ sampleIntervalMs: 100_000, log: () => {} }); + expect(() => startEventLoopMonitor({ log: () => {} })).not.toThrow(); + stopEventLoopMonitor(); + expect(() => stopEventLoopMonitor()).not.toThrow(); + expect(eventLoopLagSnapshot()).toBeUndefined(); + }); + + it("sampler periodically logs event_loop_lag with p50/p99/max/mean fields", async () => { + const lines: string[] = []; + startEventLoopMonitor({ sampleIntervalMs: 40, resolutionMs: 10, log: (l) => lines.push(l) }); + try { + await new Promise((res) => setTimeout(res, 110)); + } finally { + stopEventLoopMonitor(); + } + const samples = lines.map((l) => JSON.parse(l)).filter((o) => o.event === "event_loop_lag"); + expect(samples.length).toBeGreaterThan(0); + const s = samples[0]!; + expect(s).toMatchObject({ + event: "event_loop_lag", + p50Ms: expect.any(Number), + p99Ms: expect.any(Number), + maxMs: expect.any(Number), + meanMs: expect.any(Number), + windowMs: 40, + }); + }); + + it("snapshot reflects an elevated max after a synchronous stall", async () => { + startEventLoopMonitor({ sampleIntervalMs: 100_000, resolutionMs: 10, log: () => {} }); + try { + // Warm up: monitorEventLoopDelay only attributes a stall once its internal + // timer baseline is established (one loop tick after enable()). + await new Promise((res) => setTimeout(res, 40)); + busyLoopMs(250); + // Let the overdue internal tick fire and record the delay. + await new Promise((res) => setTimeout(res, 40)); + const snap = eventLoopLagSnapshot(); + expect(snap).toBeDefined(); + // The 250 ms busy-loop must surface as ≥100 ms of measured delay (slop for + // scheduler granularity / CI noise). + expect(snap?.maxMs).toBeGreaterThanOrEqual(100); + } finally { + stopEventLoopMonitor(); + } + }); +}); diff --git a/src/api/tests/unit/heartbeat-gap.smoke.test.ts b/src/api/tests/unit/heartbeat-gap.smoke.test.ts new file mode 100644 index 0000000..7675a5e --- /dev/null +++ b/src/api/tests/unit/heartbeat-gap.smoke.test.ts @@ -0,0 +1,317 @@ +/** + * F8 end-to-end smoke test: reproduce a >lease event-loop stall on each of the + * three Redis leases and prove the new observability fires. + * + * Each test acquires a lock against an in-process FakeRedis that honours PX + * expiry / ZSET TTL reaping, then blocks the event loop synchronously for longer + * than the lease. This is the genuine failure mode (a GC pause / pathological + * sync bash stretch): the renewal timer cannot fire, Redis expires the + * key/reaps the reader entry, and the late renew returns 0 → a real + * `LockLostError`. The assertions then REQUIRE the new `heartbeat_gap` critical + * event (with the correct `lock` tag + gap ≥ lease). Delete the emit calls in + * the heartbeats and these tests go red. + * + * A no-stall control proves there are no false positives on healthy ticks. + */ + +import type { Redis } from "ioredis"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { resetRedisCircuitBreakerForTest } from "../../../redis/circuit-breaker.js"; +import { LockLostError, execLockKey, withDistributedLock } from "../../distributed-lock.js"; +import { type RWLockKeys, rwLockKeys, withDistributedRWLock } from "../../distributed-rw-lock.js"; + +// Lease floor small enough to cross with a short busy-loop, large enough that a +// healthy tick (renewMs=80) lands well inside it. +const LEASE_MS = 200; +const RENEW_MS = 80; +const STALL_MS = 350; // > LEASE_MS, so the lease genuinely expires during the stall + +function busyLoopMs(ms: number): void { + const start = Date.now(); + while (Date.now() - start < ms) { + /* block the event loop */ + } +} + +interface HeartbeatGapLine { + event: string; + severity: "warn" | "critical"; + lock: string; + key: string; + gapMs: number; + renewMs: number; + leaseMs: number; +} + +/** Spy console.warn/error and surface only the parsed `heartbeat_gap` lines. */ +function captureHeartbeatGaps(): { gaps: () => HeartbeatGapLine[]; restore: () => void } { + const lines: string[] = []; + const push = (l: unknown): void => { + lines.push(String(l)); + }; + const warnSpy = vi.spyOn(console, "warn").mockImplementation(push); + const errorSpy = vi.spyOn(console, "error").mockImplementation(push); + return { + gaps: () => + lines + .map((l) => { + try { + return JSON.parse(l) as { event?: string }; + } catch { + return {}; + } + }) + .filter((o): o is HeartbeatGapLine => (o as { event?: string }).event === "heartbeat_gap"), + restore: () => { + warnSpy.mockRestore(); + errorSpy.mockRestore(); + }, + }; +} + +// ── Exec-lock FakeRedis (SET NX PX + Lua renew/release) ───────────────────────── + +class SimpleFakeRedis { + store = new Map(); + private gc(): void { + const now = Date.now(); + for (const [k, e] of this.store) if (e.expiresAt <= now) this.store.delete(k); + } + async set(key: string, value: string, _px: "PX", ms: number, _nx: "NX"): Promise<"OK" | null> { + this.gc(); + if (this.store.has(key)) return null; + this.store.set(key, { value, expiresAt: Date.now() + ms }); + return "OK"; + } + async eval(script: string, _n: number, key: string, token: string, ms?: string): Promise { + this.gc(); + if (script.includes("del")) { + const e = this.store.get(key); + if (e?.value === token) { + this.store.delete(key); + return 1; + } + return 0; + } + const e = this.store.get(key); + if (e?.value === token && ms !== undefined) { + e.expiresAt = Date.now() + Number(ms); + return 1; + } + return 0; + } +} + +// ── RW-lock FakeRedis (writer flag string + reader ZSET with TTL scores) ───────── + +class RwFakeRedis { + strings = new Map(); + zsets = new Map>(); + private gcStrings(): void { + const now = Date.now(); + for (const [k, e] of this.strings) if (e.expiresAt <= now) this.strings.delete(k); + } + private getZset(key: string): Map { + let z = this.zsets.get(key); + if (!z) { + z = new Map(); + this.zsets.set(key, z); + } + return z; + } + private reapZset(key: string, nowMs: number): void { + const z = this.zsets.get(key); + if (!z) return; + for (const [m, score] of z) if (score <= nowMs) z.delete(m); + } + async eval(script: string, numKeys: number, ...args: string[]): Promise { + this.gcStrings(); + const keys = args.slice(0, numKeys); + const argv = args.slice(numKeys); + // ACQUIRE_SHARED + if (script.includes("ZREMRANGEBYSCORE") && script.includes("EXISTS")) { + const [writerKey, readersKey] = keys as [string, string]; + const [token, nowStr, expireAtStr] = argv as [string, string, string]; + this.reapZset(readersKey, Number(nowStr)); + if (this.strings.has(writerKey)) return 0; + this.getZset(readersKey).set(token, Number(expireAtStr)); + return 1; + } + // RELEASE_SHARED + if (script.includes("ZREM") && !script.includes("ZREMRANGEBYSCORE")) { + const [readersKey] = keys as [string]; + const [token] = argv as [string]; + return this.zsets.get(readersKey)?.delete(token) ? 1 : 0; + } + // RENEW_SHARED + if (script.includes("ZSCORE")) { + const [readersKey] = keys as [string]; + const [token, expireAtStr] = argv as [string, string]; + const z = this.zsets.get(readersKey); + if (z?.has(token)) { + z.set(token, Number(expireAtStr)); + return 1; + } + return 0; + } + // ACQUIRE_EXCLUSIVE_FLAG + if (script.includes("SET") && script.includes("NX")) { + const [writerKey] = keys as [string]; + const [token, leaseMsStr] = argv as [string, string]; + if (this.strings.has(writerKey)) return null; + this.strings.set(writerKey, { value: token, expiresAt: Date.now() + Number(leaseMsStr) }); + return "OK"; + } + // CHECK_READERS_DRAINED + if (script.includes("ZCARD")) { + const [writerKey, readersKey] = keys as [string, string]; + const [token, nowStr] = argv as [string, string]; + if (this.strings.get(writerKey)?.value !== token) return -1; + this.reapZset(readersKey, Number(nowStr)); + return this.zsets.get(readersKey)?.size ?? 0; + } + // RENEW_EXCLUSIVE + if (script.includes("PEXPIRE")) { + const [writerKey] = keys as [string]; + const [token, leaseMsStr] = argv as [string, string]; + const e = this.strings.get(writerKey); + if (e?.value === token) { + e.expiresAt = Date.now() + Number(leaseMsStr); + return 1; + } + return 0; + } + // RELEASE_EXCLUSIVE + if (script.includes("DEL")) { + const [writerKey] = keys as [string]; + const [token] = argv as [string]; + if (this.strings.get(writerKey)?.value !== token) return 0; + this.strings.delete(writerKey); + return 1; + } + throw new Error(`RwFakeRedis: unrecognised script: ${script.slice(0, 40)}`); + } +} + +const asRedis = (f: unknown): Redis => f as Redis; + +describe("F8 heartbeat-gap smoke: a >lease stall is reproduced and observed", () => { + beforeEach(() => { + vi.useRealTimers(); + resetRedisCircuitBreakerForTest(); + }); + afterEach(() => { + vi.useRealTimers(); + resetRedisCircuitBreakerForTest(); + }); + + it("exec lock: sync stall loses the lease AND emits a critical heartbeat_gap", async () => { + const r = new SimpleFakeRedis(); + const key = execLockKey("default", "sbx-exec-stall"); + const cap = captureHeartbeatGaps(); + try { + await expect( + withDistributedLock( + asRedis(r), + key, + async () => { + busyLoopMs(STALL_MS); + // Yield so the overdue heartbeat fires inside the critical section. + await new Promise((res) => setTimeout(res, 30)); + }, + { leaseMs: LEASE_MS, renewMs: RENEW_MS, acquireTimeoutMs: 5_000, acquireRetryMs: 10 }, + ), + ).rejects.toBeInstanceOf(LockLostError); + } finally { + cap.restore(); + } + const critical = cap.gaps().filter((g) => g.severity === "critical"); + expect(critical.length).toBeGreaterThanOrEqual(1); + expect(critical[0]?.lock).toBe("exec"); + expect(critical[0]?.key).toBe(key); + expect(critical[0]?.gapMs).toBeGreaterThanOrEqual(LEASE_MS); + }); + + it("rw-lock writer: sync stall loses the flag AND emits a critical heartbeat_gap", async () => { + const r = new RwFakeRedis(); + const keys: RWLockKeys = rwLockKeys("default", "sbx-writer-stall"); + const cap = captureHeartbeatGaps(); + try { + await expect( + withDistributedRWLock( + asRedis(r), + keys, + "exclusive", + async () => { + busyLoopMs(STALL_MS); + await new Promise((res) => setTimeout(res, 30)); + }, + { leaseMs: LEASE_MS, renewMs: RENEW_MS, acquireTimeoutMs: 5_000, acquireRetryMs: 10, readerLeaseMs: 5_000 }, + ), + ).rejects.toBeInstanceOf(LockLostError); + } finally { + cap.restore(); + } + const critical = cap.gaps().filter((g) => g.severity === "critical"); + expect(critical.length).toBeGreaterThanOrEqual(1); + expect(critical[0]?.lock).toBe("rw-writer"); + expect(critical[0]?.key).toBe(keys.writer); + expect(critical[0]?.gapMs).toBeGreaterThanOrEqual(LEASE_MS); + }); + + it("rw-lock reader: sync stall past the reader lease emits a critical heartbeat_gap", async () => { + // Reader semantics differ from the writer/exec leases: a ZSET entry does not + // auto-expire by score — it is only reaped by a *competing writer's* + // acquire/check (ZREMRANGEBYSCORE). So a lone reader whose heartbeat fires + // late simply re-adds itself and does NOT self-detect a loss; the danger is + // that during the gap a writer could have reaped its stale entry and entered. + // The gap metric is exactly what surfaces that otherwise-silent window. + const r = new RwFakeRedis(); + const keys: RWLockKeys = rwLockKeys("default", "sbx-reader-stall"); + const cap = captureHeartbeatGaps(); + try { + await withDistributedRWLock( + asRedis(r), + keys, + "shared", + async () => { + busyLoopMs(STALL_MS); + await new Promise((res) => setTimeout(res, 30)); + }, + { leaseMs: 5_000, renewMs: RENEW_MS, acquireTimeoutMs: 5_000, acquireRetryMs: 10, readerLeaseMs: LEASE_MS }, + ); + } finally { + cap.restore(); + } + const critical = cap.gaps().filter((g) => g.severity === "critical"); + expect(critical.length).toBeGreaterThanOrEqual(1); + expect(critical[0]?.lock).toBe("rw-reader"); + expect(critical[0]?.key).toBe(keys.readers); + // gap ≥ readerLeaseMs proves the window during which a writer could have + // reaped this reader and entered while the read was still in flight. + expect(critical[0]?.gapMs).toBeGreaterThanOrEqual(LEASE_MS); + }); + + it("control: a healthy run (no stall) emits no heartbeat_gap events", async () => { + const r = new SimpleFakeRedis(); + const key = execLockKey("default", "sbx-healthy"); + const cap = captureHeartbeatGaps(); + try { + const result = await withDistributedLock( + asRedis(r), + key, + async () => { + // Run long enough for one on-time heartbeat (renewMs=80) to fire. + await new Promise((res) => setTimeout(res, 130)); + return "ok"; + }, + { leaseMs: LEASE_MS, renewMs: RENEW_MS, acquireTimeoutMs: 5_000, acquireRetryMs: 10 }, + ); + expect(result).toBe("ok"); + } finally { + cap.restore(); + } + expect(cap.gaps()).toEqual([]); + expect(r.store.has(key)).toBe(false); // released cleanly + }); +});