diff --git a/docs/solutions/test-failures/postgres-loaded-lane-unrelated-failure-population.md b/docs/solutions/test-failures/postgres-loaded-lane-unrelated-failure-population.md index a02fec526f..346522f648 100644 --- a/docs/solutions/test-failures/postgres-loaded-lane-unrelated-failure-population.md +++ b/docs/solutions/test-failures/postgres-loaded-lane-unrelated-failure-population.md @@ -15,7 +15,7 @@ tags: # PostgreSQL loaded-lane unrelated failure population -## Verdict: reproduced but unattributed +## Verdict: reproduced, but attribution remains coverage-limited FN-9148 reproduced the unrelated population in three of five pre-registered, diagnostics-enabled 27-worker directory runs. A03/A04/A05 reported 45, 35, and @@ -52,15 +52,27 @@ watchdog snapshots included checkpoint, ProcSignalBarrier, and object-lock waits. The checkpointed task document `evidence` is the detailed durable record. +## FN-9149 timeout-boundary observation + +FN-9149 added a default-off setup/body/teardown observer and ran its enabled-wiring gate before the campaign. The gate emitted full, non-suppressed watchdog payloads at all three boundaries, including `shared.body`; it used only observer environment settings (1ms per-boundary watchdogs, 1500ms probe timeout, 1200ms statement timeout, 2000ms drain, max probes 20, concurrency 3). The raised cap and concurrency were gate-only; campaign runs retained concurrency 1 and queue timeout 0. + +The original I01–I03 samples (14, 31, and 52 failures; peaks 66, 68, and 70) and unset control (27 failures; peak 70) remain retained as superseded: their 12s watchdog was coverage-limited and the original 25% population rule flagged perturbation. + +Before any J-series invocation, FN-9149 prospectively amended the registration. FN-9148’s fixed baseline dispersion was 13/24/45/35/32 (range 32, or 71.1% of the 45-file maximum), so the J-only perturbation threshold is conservatively rounded to 72% of the larger population. The fixed amended observer stack used 14s setup/body/teardown watchdogs (strictly below the 15s inherited budgets and above the measured healthy gate maximum), threshold 2s, probe/statement/drain 1500/1400/3000ms, cap 4, concurrency 1, and queue 0. The forced gate’s raised cap/concurrency remained gate-only. + +The J-series completed three instrumented runs and an unset-observer control: J01=47 failures, peak 58; J02=26, peak 79; J03=54, peak 67; control=40, peak 70. Their enabled/control differences (7/14/14) are below the prospectively fixed 72% limits (34/29/39), so the J instrumented output is not perturbation-flagged. All peaks remain below the 97 ordinary-slot ceiling. J01/J02/J03 joined 9/2/1 cluster-implicated failures and left 38/24/53 unjoined; no watchdog probe was suppressed and one record per run settled during its probe. This remains insufficient body and boundary coverage for M2–M4 attribution, not evidence for a remedy. Full JSONL/census and the retained prior sample are checkpointed in FN-9149’s `evidence` document. + +The superseded 12s clean/dirty arm remains retained but cannot decide M5. The required amended-stack arm then ran interleaved `clean → dirty → clean → dirty` as J04–J07. Clean J04/J06 began with zero matching leftovers and reported 29/36 failures (peaks 79/68; cluster/unjoined=0/29 and 3/33). Dirty J05/J07 deliberately retained one `fusion_test_%`, two `fusion_schema_template%` databases (including a live-owner golden template), and one `fusion_pool_%` database; they reported 43/50 failures (peaks 88/65; cluster/unjoined=6/37 and 2/48). All four used the fixed J observer stack and were retained; J05's 13 `cap` suppressions limit boundary attribution but do not erase the independently repeated clean/dirty failure-count covariation. The monotonic watchdog-drift host sample was nonzero on 4/4, 114/130, 19/20, and 26/28 watchdog records respectively (max 3/18/10/4ms), so this arm does not support host-starvation attribution. Final cluster hygiene was restored to zero matching test, pool, and schema-template databases. + ## Discrimination table | mechanism | verdict | evidence / missing discriminator | |---|---|---| -| M1 ordinary backend exhaustion | undecided; generic version contradicted | Peaks stay 22–36 below 97, but watchdog-only snapshots cannot rule out a per-user/per-database limit or a missed transient peak. | -| M2 DDL serialization | undecided | Hook concentration and checkpoint/catalog/object waits are observations, not per-failed-hook DDL correlation. | -| M3 golden-template/advisory convoy | undecided | No template-owner/lock-wait timeline was captured. | -| M4 host CPU/event-loop starvation | undecided | Host load was material, but teardown-only probes cannot show PostgreSQL idle versus in-flight at setup/body timeout time. | -| M5 dirty-cluster carryover | undecided | Clean resets still reproduced, but no controlled clean/dirty covariation measurement was run. | +| M1 ordinary backend exhaustion | eliminated (generic ordinary-slot form) | A01–A05 and J01–J03 peaks were 58–79, all below the 97 ordinary ceiling; no per-user/database limit was observed. A successor must separately measure any scoped limit before claiming that variant. | +| M2 DDL serialization | still undecided (missing evidence: join coverage) | Twelve J watchdog joins cannot correlate DDL/locks to 127 failures. | +| M3 golden-template/advisory convoy | still undecided (missing evidence: joined golden-lock waiters) | The probe now records granted holders and non-granted golden advisory waiters; no sufficient joined timeout population exists. | +| M4 host CPU/event-loop starvation | still undecided (missing evidence: join coverage) | The observer now records watchdog scheduling drift rather than a fixed zero, but the campaign cannot distinguish idle-cluster host starvation from blocked SQL at required body coverage. | +| M5 dirty-cluster carryover | affirmed | The amended interleaved J04–J07 arm covaried at 0 leftovers → 29/36 failures and 1 test + 2 schema-template (one golden) + 1 pool leftover → 43/50 failures. FN-9151 owns identification and a regression-proven structural remedy; no remedy is implemented here. | ## Remedies disqualified by this evidence @@ -72,9 +84,4 @@ not a resolution. ## Successor measurement seam -The successor must design a separately reviewed, default-off observer that can -join a setup/body or teardown timeout to host pressure, the active SQL/lock -state, and template ownership without changing harness execution. It must run -three instrumented reproductions and an unset-environment control before -attribution. A controlled dirty-cluster arm is also required for M5. Reuse the -census tool and retain all logs/JSONL; do not return to impression-based claims. +FN-9151 owns the affirmed dirty-carryover structural seam: identify why retained `fusion_test_%`, `fusion_schema_template%` (including golden), and `fusion_pool_%` state covaries with the 27-worker failure population, then prove a remedy with a regression without changing timeouts, retries, skips, worker caps, quarantine, DDL, connection-budget, or admission behavior. A separate successor may improve timeout-boundary join coverage for M2–M4; it must preserve the default-off observer and repeat the perturbation control. Generic ordinary-slot exhaustion remains disqualified. diff --git a/docs/testing.md b/docs/testing.md index 0655c07c5c..8f3d24eaf7 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -545,6 +545,33 @@ Use `scripts/pg-loaded-failure-census.mjs` to inspect an already-retained Vitest A run is `insufficient-data` when its runner log lacks a complete `Test Files` summary or when the reported failed-file count cannot be reconciled to parsed failure blocks. A complete passing summary instead produces `status: "measured"` with `failingFileCount: 0`; never treat that zero as missing evidence or coerce incomplete input to a healthy result. + + +### PostgreSQL timeout-boundary observer + +`FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER=1` enables a diagnostic-only JSONL channel; when unset it creates no timer, probe connection, sink write, listener, or environment mutation. `observeBoundary()` wraps harness-owned setup and teardown work. `openBoundary()`/`closeBoundary()` span the shared harness's separate `beforeEach` and `afterEach` hooks, so `shared.body` measures test bodies without editing consumers. Closing is idempotent; an abandoned window is superseded by a later open or disposed without inventing a completion. + +The observer arms an unref'd watchdog at boundary start. Only a pending watchdog dispatches a maintenance-connection probe; completion records are host-only and emit when elapsed time meets `FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_THRESHOLD_MS` (default 2000; `0` measures every completion). A dispatched probe survives boundary settlement and appends when it resolves, with `settledDuringProbe`, `probeLatencyMs`, and optionally `probeStartDelayMs`; records can therefore be out of order and consumers must join by pid, worker, file, boundary, and timestamp rather than JSONL order. `flush()`/`dispose()` drain bounded pending and queued probes; a missed drain becomes `probeSuppressed:"drain-timeout"`. The enabled-only `beforeExit` hook performs the same best-effort drain. + +The independent, tighten-only bounds are `statement timeout < probe timeout < inherited budget`, `queue timeout < probe timeout`, and `threshold <= per-boundary watchdog < inherited budget`. A probe timeout is deliberately **not** tied to a short watchdog: forced wiring runs need a fast watchdog and a real cluster round trip. Floors are `MIN_WATCHDOG_MS`, `MIN_PROBE_TIMEOUT_MS`, and `MIN_STATEMENT_TIMEOUT_MS`; an impossible floor yields `bounds-floor`, never a wider budget. + +| variable | default | purpose | +|---|---:|---| +| `..._LOG` | unset | JSONL destination | +| `..._THRESHOLD_MS` | 2000 | completion-record threshold; 0 measures all | +| `..._WATCHDOG_MS`, `..._WATCHDOG_SETUP_MS`, `..._WATCHDOG_BODY_MS`, `..._WATCHDOG_TEARDOWN_MS` | 12000 | global and per-boundary watchdogs | +| `..._PROBE_TIMEOUT_MS` / `..._STATEMENT_TIMEOUT_MS` | 1500 / derived below probe | bounded client/server probe stack | +| `..._PROBE_DRAIN_TIMEOUT_MS` | 3000 | bounded `flush()`/`dispose()` drain | +| `..._MAX_CONCURRENT_PROBES` | 1 | counting limiter, capped by `MAX_CONCURRENT_PROBES_CEILING` (8) | +| `..._PROBE_QUEUE_TIMEOUT_MS` | 0 | bounded slot wait; a queued probe records delay | +| `..._MAX_PROBES` | 4 | per-process diagnostic cost cap | + +At concurrency 1 the cost profile matches the older single-flight observer. `probeSuppressed:"concurrency"` supersedes legacy `single-flight`; neither is a valid forced-wiring payload. The cap is configurable because once probes survive settle, a strict single flight would suppress simultaneous setup/body/teardown breaches. Probe payloads contain host load/CPU and watchdog scheduling drift (`eventLoopLagMs`, measured as the monotonic callback delay past its deadline at the snapshot instant), active SQL and lock/blocking rows, backend count, and golden-template owner/readiness plus separately reported granted advisory holders and non-granted advisory waiters. Only a non-owner waiter supports a template-convoy attribution; a holder alone does not. + +For an enabled-wiring check, first run disabled and normal enabled bounds to compare duration, then threshold 0 to verify setup, `shared.body`, and teardown completion records, then force per-boundary watchdogs below measured elapsed times while independently retaining a probe/drain timeout above cluster round-trip and raising probe cap/concurrency to at least 3. Every forced boundary must have a non-suppressed payload; `cap`, `concurrency`, `bounds-floor`, `drain-timeout`, and `error` require diagnosis, not acceptance. Files using the shared harness are body-observable; direct `createTaskStoreForTest` files deliberately have no harness-owned body bracket and must be reported as `body-unobservable`. + +Pass `--boundary-observer `, `--body-unobservable-files `, and `--fully-unobservable-files ` to `scripts/pg-loaded-failure-census.mjs`. It tolerates malformed lines and classifies each failing boundary as `cluster-implicated`, `host-implicated`, `template-convoy`, `body-unobservable`, or `unjoined`; failures from fully-unobservable files remain `unjoined` but are counted and listed separately so the known no-harness limitation is never mistaken for a missing observer join. Missing/empty input stays explicitly absent rather than a measured zero. + ### PostgreSQL DDL loaded-lane acceptance metric Use `scripts/pg-ddl-lane-metric.mjs` before judging a PostgreSQL DDL structural candidate. Run at least seven **interleaved** control/candidate pairs at `VITEST_MAX_WORKERS=12`; preserve one diagnostics JSONL sink and complete runner log per invocation. The exact lane is: diff --git a/packages/core/src/__test-utils__/pg-test-harness.ts b/packages/core/src/__test-utils__/pg-test-harness.ts index c918b1be88..fea7f72dd8 100644 --- a/packages/core/src/__test-utils__/pg-test-harness.ts +++ b/packages/core/src/__test-utils__/pg-test-harness.ts @@ -45,6 +45,12 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { basename, join } from "node:path"; import { createPostgresDdlAdmissionGate } from "./pg-ddl-admission.js"; import { tmpdir } from "node:os"; +import { + createPgTimeoutBoundaryObserver, + type PgTimeoutBoundaryObserver, + type PgTimeoutBoundaryProbePayload, + type PgTimeoutBoundaryProbeBounds, +} from "./pg-timeout-boundary-observer.js"; import { createPgTeardownDiagnostics, getPgTeardownDiagnosticsProbeTimeoutMs, @@ -237,6 +243,8 @@ export interface PgTestHarness { readonly rootDir: string; /** The unique test database name (for diagnostics). */ readonly dbName: string; + /** The default-off boundary observer retained for this harness lifecycle. */ + readonly timeoutObserver: PgTimeoutBoundaryObserver; /** The full test connection URL. */ readonly testUrl: string; /** Drop the test database, close connections, and remove the temp dir. */ @@ -307,6 +315,67 @@ function createPgStatActivityProbe( }; } +/** + * The timeout-boundary observer owns this separate maintenance connection. It + * never reuses a harness pool, so a snapshot cannot wait behind the operation + * it is diagnosing. + */ +/* +FNXC:PgTimeoutBoundaryObserver 2026-08-19-14:43: +FN-9149 requires probe records to describe the same safety bounds the maintenance +connection actually enforces. The observer resolves and tightens untrusted env +values once, then passes those resolved limits here rather than allowing this +production probe to reread a larger raw timeout. +*/ +function createPgTimeoutBoundaryProbe(): (signal: AbortSignal, bounds: PgTimeoutBoundaryProbeBounds) => Promise { + return async (signal, bounds) => { + const maintUrl = new URL(PG_TEST_URL_BASE); + maintUrl.pathname = "/postgres"; + const { probeTimeoutMs: probeTimeout, statementTimeoutMs: statementTimeout } = bounds; + const client = postgres(maintUrl.toString(), { + max: 1, + prepare: false, + // postgres accepts whole seconds here; the observer AbortSignal remains + // the precise client-side deadline when the resolved bound is subsecond. + connect_timeout: Math.max(1, Math.ceil(probeTimeout / 1_000)), + onnotice: () => {}, + }); + const abort = () => { void client.end({ timeout: 0 }).catch(() => {}); }; + signal.addEventListener("abort", abort, { once: true }); + const goldenName = goldenTemplateName(); + try { + await client.unsafe(`SET statement_timeout = ${Math.trunc(statementTimeout)}`); + const [activity, locks, marker] = await Promise.all([ + client.unsafe>(` + SELECT pid, datname, usename, state, wait_event_type, wait_event, backend_type, + now() - query_start AS query_age, left(query, 200) AS query, + count(*) OVER ()::int AS total_backends, pg_blocking_pids(pid) AS "blockingPids" + FROM pg_stat_activity ORDER BY datname NULLS LAST, pid + `), + client.unsafe>(` + SELECT l.pid, l.locktype, l.granted, pg_blocking_pids(l.pid) AS "blockingPids" + FROM pg_locks l WHERE l.pid IS NOT NULL + `), + client.unsafe>( + `SELECT EXISTS(SELECT 1 FROM ${GOLDEN_MARKER_QUALIFIED} WHERE name = $1) AS "markerPresent", + NULLIF(regexp_replace($1, '^fusion_schema_template_([0-9]+).*$', '\\1'), $1)::int AS "ownerPid", + ARRAY(SELECT pid FROM pg_locks WHERE locktype = 'advisory' AND objid = hashtext($1) AND granted) AS "advisoryHolders", + ARRAY(SELECT pid FROM pg_locks WHERE locktype = 'advisory' AND objid = hashtext($1) AND NOT granted) AS "advisoryWaiters"`, + [goldenName], + ).catch(() => [{ markerPresent: false, ownerPid: null, advisoryHolders: [], advisoryWaiters: [] }]), + ]); + const template = marker[0] ?? { markerPresent: false, ownerPid: null, advisoryHolders: [], advisoryWaiters: [] }; + return { + cluster: { activity, locks, totalBackends: activity[0]?.total_backends ?? 0 }, + template: { goldenTemplateName: goldenName, ...template, isOwner: template.ownerPid === process.pid }, + }; + } finally { + signal.removeEventListener("abort", abort); + await client.end({ timeout: 5 }).catch(() => {}); + } + }; +} + /** * FNXC:PgTestDdlAdmission 2026-08-16-21:29: * FN-9130 measured both uniform and drop-only advisory admission as worse than @@ -871,6 +940,11 @@ export async function createTaskStoreForTest(options?: { const projectId = options?.projectId; const dbName = uniqueDbName(prefix); + const testFile = vitestExpect.getState().testPath; + const timeoutObserver = createPgTimeoutBoundaryObserver({ + probe: createPgTimeoutBoundaryProbe(), + ...(testFile ? { testFile } : {}), + }); // FNXC:PgTestTemplateDb 2026-07-19-17:20: // Create the test database as a fast server-side copy of a pre-baked template. @@ -880,10 +954,12 @@ export async function createTaskStoreForTest(options?: { // Concurrent CREATE DATABASE ... TEMPLATE copies from one connection-free // source are safe; only an active session on the source triggers "source // database is being accessed". - const template = options?.copyFromGolden - ? await ensureGoldenTemplate() - : await ensureSchemaTemplate(); - await cloneDatabaseFromTemplate(dbName, template); + const template = await timeoutObserver.observeBoundary( + "setup", + options?.copyFromGolden ? "template.ensure-golden" : "template.ensure-schema", + () => options?.copyFromGolden ? ensureGoldenTemplate() : ensureSchemaTemplate(), + ); + await timeoutObserver.observeBoundary("setup", "database.clone", () => cloneDatabaseFromTemplate(dbName, template)); const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; // The database already carries the full schema (copied from the template), @@ -902,11 +978,11 @@ export async function createTaskStoreForTest(options?: { directSessionUrl: testUrl, directSessionProvenance: "migration-override", }; - const connections = await createConnectionSetFromUrl(schemaBackend, { + const connections = await timeoutObserver.observeBoundary("setup", "connections.create", () => createConnectionSetFromUrl(schemaBackend, { poolMax, connectTimeoutSeconds: 5, projectId, - }); + })); const layer = createAsyncDataLayer(connections, projectId ? { projectId } : undefined); // Admin connection for direct row inspection/seeding in tests. @@ -922,7 +998,7 @@ export async function createTaskStoreForTest(options?: { // Construct the TaskStore in backend mode. const store = new TaskStore(rootDir, undefined, { asyncLayer: layer }); - await store.init(); + await timeoutObserver.observeBoundary("setup", "store.init", () => store.init()); let tornDown = false; const teardown = async (): Promise => { @@ -948,34 +1024,36 @@ export async function createTaskStoreForTest(options?: { // best-effort } try { - await diagnostics.runPhase("store.close", () => store.close()); + await timeoutObserver.observeBoundary("teardown", "store.close", () => diagnostics.runPhase("store.close", () => store.close())); } catch { // best-effort } try { - await diagnostics.runPhase("layer.close", () => layer.close()); + await timeoutObserver.observeBoundary("teardown", "layer.close", () => diagnostics.runPhase("layer.close", () => layer.close())); } catch { // best-effort } try { - await diagnostics.runPhase("adminSql.end", () => adminSql.end({ timeout: 5 })); + await timeoutObserver.observeBoundary("teardown", "adminSql.end", () => diagnostics.runPhase("adminSql.end", () => adminSql.end({ timeout: 5 }))); } catch { // best-effort } try { // FNXC:PgTestHarness 2026-07-18-17:27: FORCE so open pool sockets cannot block drop after close races. - await diagnostics.runPhase("dropDatabase", () => adminExecAsync(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`)); + await timeoutObserver.observeBoundary("teardown", "dropDatabase", () => diagnostics.runPhase("dropDatabase", () => adminExecAsync(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`))); } catch { // best-effort } try { - await diagnostics.runPhase("rmRootDir", () => rm(rootDir, { recursive: true, force: true })); + await timeoutObserver.observeBoundary("teardown", "rmRootDir", () => diagnostics.runPhase("rmRootDir", () => rm(rootDir, { recursive: true, force: true }))); } catch { // best-effort } } finally { diagnostics.completeTeardown(); diagnostics.dispose(); + await timeoutObserver.flush().catch(() => {}); + await timeoutObserver.dispose().catch(() => {}); } }; @@ -986,6 +1064,7 @@ export async function createTaskStoreForTest(options?: { adminSql, rootDir, dbName, + timeoutObserver, testUrl, teardown, }; @@ -1142,6 +1221,7 @@ export function createSharedPgTaskStoreTestHarness(options?: { const boundProjectId = options?.projectId ?? ""; let harness: PgTestHarness | null = null; let store: TaskStore | null = null; + let bodyHandle: import("./pg-timeout-boundary-observer.js").PgTimeoutBoundaryHandle | null = null; // Lazily import DEFAULT_PROJECT_SETTINGS to avoid pulling the full types // graph at module load in environments that only use createTaskStoreForTest. let defaultSettingsCache: Record | null = null; @@ -1269,8 +1349,15 @@ export function createSharedPgTaskStoreTestHarness(options?: { } catch { // best-effort: reconciliation is idempotent and fail-soft } + // beforeEach and afterEach are separate hooks, so only this paired API + // can observe the test body without charging reset/setup to it. + const testFile = vitestExpect.getState().testPath ?? "unknown-test-file"; + bodyHandle = harness.timeoutObserver.openBoundary("body", "shared.body", `${process.pid}:${process.env.VITEST_WORKER_ID ?? "main"}:${testFile}`); }, afterEach: async () => { + // Close before watcher cleanup so teardown work is never body cost. + if (bodyHandle && harness) harness.timeoutObserver.closeBoundary(bodyHandle); + bodyHandle = null; // No per-test connection teardown — the shared DB lives until afterAll. // Just quiesce any watchers/timers the test may have armed. if (store) { @@ -1284,6 +1371,7 @@ export function createSharedPgTaskStoreTestHarness(options?: { afterAll: async () => { if (harness) { await harness.teardown(); + bodyHandle = null; harness = null; store = null; } @@ -1312,6 +1400,7 @@ export function createSharedPgTaskStoreTestHarness(options?: { teardown: async () => { if (harness) { await harness.teardown(); + bodyHandle = null; harness = null; store = null; } diff --git a/packages/core/src/__test-utils__/pg-timeout-boundary-observer.ts b/packages/core/src/__test-utils__/pg-timeout-boundary-observer.ts new file mode 100644 index 0000000000..db60f69c29 --- /dev/null +++ b/packages/core/src/__test-utils__/pg-timeout-boundary-observer.ts @@ -0,0 +1,451 @@ +import { appendFileSync } from "node:fs"; +import { cpus, loadavg } from "node:os"; + +export const MIN_WATCHDOG_MS = 1; +export const MIN_PROBE_TIMEOUT_MS = 10; +export const MIN_STATEMENT_TIMEOUT_MS = 1; +export const MAX_CONCURRENT_PROBES_CEILING = 8; + +export type PgTimeoutBoundary = "setup" | "body" | "teardown"; +export type PgTimeoutBoundaryTrigger = "boundary-complete" | "boundary-watchdog"; +export type PgTimeoutBoundarySuppression = "cap" | "concurrency" | "bounds-floor" | "drain-timeout" | "error"; + +type TimerHandle = ReturnType; +type TimerFactory = (callback: () => void, ms: number) => TimerHandle; + +export interface PgTimeoutBoundaryHostSample { + readonly loadavg1: number; + readonly loadavg5: number; + readonly loadavg15: number; + readonly cpuCount: number; + readonly resolvedWorkers?: number; + readonly eventLoopLagMs: number; +} + +export interface PgTimeoutBoundaryProbePayload { + readonly cluster: Record; + readonly template: Record; +} + +/** Resolved observer bounds passed to the production maintenance probe. */ +export interface PgTimeoutBoundaryProbeBounds { + readonly probeTimeoutMs: number; + readonly statementTimeoutMs: number; +} + +export interface PgTimeoutBoundaryRecord { + readonly timestamp: string; + readonly pid: number; + readonly workerId?: string; + readonly testFile?: string; + readonly testName?: string; + readonly boundary: PgTimeoutBoundary; + readonly phase: string; + readonly trigger: PgTimeoutBoundaryTrigger; + readonly elapsedMs: number; + readonly boundaryIncomplete: boolean; + readonly settledDuringProbe?: boolean; + readonly probeLatencyMs?: number; + readonly probeStartDelayMs?: number; + readonly supersededOpenWindow?: boolean; + readonly outcome?: "resolved" | "rejected"; + readonly thresholdMs: number; + readonly watchdogMs: number; + readonly probeTimeoutMs: number; + readonly statementTimeoutMs: number; + readonly probeDrainTimeoutMs: number; + readonly maxConcurrentProbes: number; + readonly probeQueueTimeoutMs: number; + readonly boundsClamped: boolean; + readonly probeSuppressed?: PgTimeoutBoundarySuppression; + readonly host: PgTimeoutBoundaryHostSample; + readonly cluster?: Record; + readonly template?: Record; +} + +export interface PgTimeoutBoundaryObserverOptions { + readonly env?: NodeJS.ProcessEnv; + /** Monotonic clock used only for elapsed durations. */ + readonly now?: () => number; + /** Wall clock is retained solely as a cross-process join key. */ + readonly wallNow?: () => number; + readonly setTimer?: TimerFactory; + readonly clearTimer?: (timer: TimerHandle) => void; + readonly append?: (path: string, line: string) => void; + readonly writeError?: (line: string) => void; + readonly probe?: (signal: AbortSignal, bounds: PgTimeoutBoundaryProbeBounds) => Promise; + readonly hostSample?: (eventLoopLagMs?: number) => PgTimeoutBoundaryHostSample; + readonly testFile?: string; + readonly testName?: () => string | undefined; + readonly inheritedHookBudgetMs?: number; + readonly inheritedBodyBudgetMs?: number; +} + +export interface PgTimeoutBoundaryHandle { + readonly id: number; +} + +export interface PgTimeoutBoundaryObserver { + readonly enabled: boolean; + observeBoundary(boundary: PgTimeoutBoundary, phase: string, action: () => Promise): Promise; + openBoundary(boundary: PgTimeoutBoundary, phase: string, key: string): PgTimeoutBoundaryHandle; + closeBoundary(handle: PgTimeoutBoundaryHandle, outcome?: "resolved" | "rejected"): void; + flush(timeoutMs?: number): Promise; + dispose(): Promise; +} + +let processProbeCount = 0; + +/** Test-only reset for deterministic process-cap coverage. */ +export function __resetPgTimeoutBoundaryObserverProbeCountForTest(): void { + processProbeCount = 0; +} + +function envNumber(env: NodeJS.ProcessEnv, key: string, fallback: number, minimum = 0): number { + const value = Number(env[key]); + return Number.isFinite(value) && value >= minimum ? Math.trunc(value) : fallback; +} + +function defaultHostSample(eventLoopLagMs = 0): PgTimeoutBoundaryHostSample { + const [loadavg1, loadavg5, loadavg15] = loadavg(); + const workers = Number(process.env.VITEST_MAX_WORKERS); + return { + loadavg1, + loadavg5, + loadavg15, + cpuCount: cpus().length, + ...(Number.isFinite(workers) ? { resolvedWorkers: workers } : {}), + // Watchdog scheduling drift is a bounded, allocation-free event-loop lag sample. + eventLoopLagMs: Math.max(0, eventLoopLagMs), + }; +} + +interface Bounds { + readonly thresholdMs: number; + readonly watchdog: Record; + readonly probeTimeoutMs: number; + readonly statementTimeoutMs: number; + readonly probeDrainTimeoutMs: number; + readonly maxConcurrentProbes: number; + readonly probeQueueTimeoutMs: number; + readonly maxProbes: number; + readonly probeAllowed: boolean; + readonly clamped: boolean; +} + +function resolveBounds(env: NodeJS.ProcessEnv, hookBudget: number, bodyBudget: number): Bounds { + let clamped = false; + const thresholdMs = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_THRESHOLD_MS", 2_000); + const requestedProbe = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_TIMEOUT_MS", 1_500, 1); + const requestedStatement = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_STATEMENT_TIMEOUT_MS", Math.max(MIN_STATEMENT_TIMEOUT_MS, requestedProbe - 100), 1); + const requestedDrain = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_DRAIN_TIMEOUT_MS", 3_000, 1); + const capBelowHook = Math.max(0, hookBudget - 1); + // Floors make malformed tiny values observable, but never widen an inherited + // budget: a budget below the floor disables probing for that boundary. + const probeTimeoutMs = Math.min(Math.max(requestedProbe, MIN_PROBE_TIMEOUT_MS), capBelowHook); + if (probeTimeoutMs !== requestedProbe) clamped = true; + const probeAllowed = probeTimeoutMs >= MIN_PROBE_TIMEOUT_MS; + if (!probeAllowed) clamped = true; + let statementTimeoutMs = Math.min(requestedStatement, Math.max(0, probeTimeoutMs - 1)); + if (statementTimeoutMs !== requestedStatement) clamped = true; + if (statementTimeoutMs < MIN_STATEMENT_TIMEOUT_MS) clamped = true; + statementTimeoutMs = Math.max(MIN_STATEMENT_TIMEOUT_MS, statementTimeoutMs); + let probeDrainTimeoutMs = Math.min(requestedDrain, capBelowHook); + if (probeDrainTimeoutMs !== requestedDrain) clamped = true; + probeDrainTimeoutMs = Math.max(1, probeDrainTimeoutMs); + let maxConcurrentProbes = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_CONCURRENT_PROBES", 1, 1); + const requestedConcurrency = maxConcurrentProbes; + maxConcurrentProbes = Math.min(MAX_CONCURRENT_PROBES_CEILING, Math.max(1, maxConcurrentProbes)); + if (maxConcurrentProbes !== requestedConcurrency) clamped = true; + let probeQueueTimeoutMs = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_QUEUE_TIMEOUT_MS", 0); + const requestedQueue = probeQueueTimeoutMs; + probeQueueTimeoutMs = Math.min(probeQueueTimeoutMs, Math.max(0, probeTimeoutMs - 1)); + if (probeQueueTimeoutMs !== requestedQueue) clamped = true; + const globalWatchdog = envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_WATCHDOG_MS", 12_000, 1); + const watchdog = {} as Record; + for (const boundary of ["setup", "body", "teardown"] as const) { + const requested = envNumber(env, `FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_WATCHDOG_${boundary.toUpperCase()}_MS`, globalWatchdog, 1); + const budget = boundary === "body" ? bodyBudget : hookBudget; + const upper = Math.max(0, budget - 1); + const resolved = Math.min(requested, upper); + watchdog[boundary] = resolved; + if (resolved !== requested || resolved < MIN_WATCHDOG_MS) clamped = true; + } + return { + thresholdMs, + watchdog, + probeTimeoutMs, + statementTimeoutMs, + probeDrainTimeoutMs, + maxConcurrentProbes, + probeQueueTimeoutMs, + maxProbes: envNumber(env, "FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_PROBES", 4, 1), + probeAllowed, + clamped, + }; +} + +interface BoundaryState { + readonly handle: PgTimeoutBoundaryHandle; + readonly boundary: PgTimeoutBoundary; + readonly phase: string; + readonly key?: string; + readonly startedAt: number; + readonly timestamp: string; + readonly host: PgTimeoutBoundaryHostSample; + timer?: TimerHandle; + settled: boolean; + watchdogFired: boolean; + outcome?: "resolved" | "rejected"; + supersededOpenWindow?: boolean; +} + +interface PendingProbe { + readonly state: BoundaryState; + readonly watchdogAt: number; + readonly host: PgTimeoutBoundaryHostSample; + readonly controller: AbortController; + queuedTimer?: TimerHandle; + timeoutTimer?: TimerHandle; + started: boolean; + /** Monotonic delay from watchdog firing until the probe received a slot. */ + probeStartDelayMs?: number; + /** Monotonic instant at which the probe actually started. */ + probeStartedAt?: number; + finalized: boolean; +} + +/* +FNXC:PgTimeoutBoundaryObserver 2026-08-19-13:51: +FN-9148 reproduced unrelated 27-worker PostgreSQL timeouts but left M1–M5 +undecided because teardown-only snapshots cannot attribute setup or body waits. +This default-off observer records whether PostgreSQL is in-flight/blocked, the +host is starved, or golden-template ownership is convoying at a timeout. + +Watchdog scheduling drift is sampled as bounded event-loop lag at the watchdog +instant. A fixed zero hid the M4 discriminator; the sample adds no timer, I/O, +or work to the observed boundary. + +The paired API exists because shared-harness beforeEach and afterEach are +separate Vitest hooks: a promise wrapper cannot span a body timeout. Watchdogs +are per boundary and bounds only tighten below inherited budgets so observation +cannot extend work. A short watchdog intentionally remains independent from a +long probe timeout; probes survive settle and are drained, otherwise fast body +windows lose their only cluster payload. + +Once probes survive their boundary, strict single-flight would suppress body and +teardown behind setup. The configurable limiter defaults to one (the inherited +cost profile) but is raisable for the forced wiring gate; its ceiling prevents a +bad environment from flooding PostgreSQL. +*/ +export function createPgTimeoutBoundaryObserver(options: PgTimeoutBoundaryObserverOptions = {}): PgTimeoutBoundaryObserver { + const env = options.env ?? process.env; + if (env.FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER !== "1") { + return { + enabled: false, + async observeBoundary(_boundary: PgTimeoutBoundary, _phase: string, action: () => Promise): Promise { return action(); }, + openBoundary: () => ({ id: 0 }), + closeBoundary() {}, + async flush() {}, + async dispose() {}, + }; + } + + const now = options.now ?? Date.now; + const wallNow = options.wallNow ?? Date.now; + const setTimer = options.setTimer ?? setTimeout; + const clearTimer = options.clearTimer ?? clearTimeout; + const append = options.append ?? ((path, line) => appendFileSync(path, line)); + const writeError = options.writeError ?? ((line) => console.error(line)); + const hostSample = options.hostSample ?? defaultHostSample; + const bounds = resolveBounds(env, options.inheritedHookBudgetMs ?? 15_000, options.inheritedBodyBudgetMs ?? 15_000); + const sink = env.FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_LOG; + let disposed = false; + let nextHandle = 1; + let activeSlots = 0; + const states = new Map(); + const windows = new Map(); + const pending = new Set(); + + const unref = (timer: TimerHandle | undefined): void => { + (timer as unknown as { unref?: () => void } | undefined)?.unref?.(); + }; + const emit = (state: BoundaryState, trigger: PgTimeoutBoundaryTrigger, fields: Partial): void => { + const record: PgTimeoutBoundaryRecord = { + timestamp: state.timestamp, + pid: process.pid, + ...(env.VITEST_WORKER_ID ? { workerId: env.VITEST_WORKER_ID } : {}), + ...(options.testFile ? { testFile: options.testFile } : {}), + ...(options.testName?.() ? { testName: options.testName() } : {}), + boundary: state.boundary, + phase: state.phase, + trigger, + elapsedMs: Math.max(0, now() - state.startedAt), + boundaryIncomplete: !state.settled, + thresholdMs: bounds.thresholdMs, + watchdogMs: bounds.watchdog[state.boundary], + probeTimeoutMs: bounds.probeTimeoutMs, + statementTimeoutMs: bounds.statementTimeoutMs, + probeDrainTimeoutMs: bounds.probeDrainTimeoutMs, + maxConcurrentProbes: bounds.maxConcurrentProbes, + probeQueueTimeoutMs: bounds.probeQueueTimeoutMs, + boundsClamped: bounds.clamped, + host: state.host, + ...fields, + }; + if (sink) { + try { append(sink, `${JSON.stringify(record)}\n`); } catch { /* sink failures must not affect a test */ } + } + try { writeError(`[pg-timeout-boundary-observer] ${trigger} ${state.boundary}/${state.phase}`); } catch { /* diagnostic stderr is best effort */ } + }; + const finalize = (pendingProbe: PendingProbe, suppression?: PgTimeoutBoundarySuppression, payload?: PgTimeoutBoundaryProbePayload): void => { + if (pendingProbe.finalized) return; + pendingProbe.finalized = true; + pending.delete(pendingProbe); + if (pendingProbe.queuedTimer) clearTimer(pendingProbe.queuedTimer); + if (pendingProbe.timeoutTimer) clearTimer(pendingProbe.timeoutTimer); + if (pendingProbe.started) { + activeSlots = Math.max(0, activeSlots - 1); + admitQueued(); + } + const state = pendingProbe.state; + emit(state, "boundary-watchdog", { + elapsedMs: bounds.watchdog[state.boundary], + boundaryIncomplete: true, + settledDuringProbe: state.settled, + ...(pendingProbe.probeStartedAt !== undefined ? { probeLatencyMs: Math.max(0, now() - pendingProbe.probeStartedAt) } : {}), + ...(pendingProbe.probeStartDelayMs !== undefined ? { probeStartDelayMs: pendingProbe.probeStartDelayMs } : {}), + ...(state.supersededOpenWindow ? { supersededOpenWindow: true } : {}), + ...(state.outcome ? { outcome: state.outcome } : {}), + ...(suppression ? { probeSuppressed: suppression } : {}), + ...(payload ? { cluster: payload.cluster, template: payload.template } : {}), + host: pendingProbe.host, + }); + }; + const startProbe = (pendingProbe: PendingProbe): void => { + if (pendingProbe.finalized) return; + if (processProbeCount >= bounds.maxProbes) { + finalize(pendingProbe, "cap"); + return; + } + processProbeCount += 1; + pendingProbe.started = true; + pendingProbe.probeStartDelayMs = Math.max(0, now() - pendingProbe.watchdogAt); + pendingProbe.probeStartedAt = now(); + activeSlots += 1; + pendingProbe.timeoutTimer = setTimer(() => { + pendingProbe.controller.abort(); + finalize(pendingProbe, "error"); + }, bounds.probeTimeoutMs); + unref(pendingProbe.timeoutTimer); + void Promise.resolve().then(() => options.probe?.(pendingProbe.controller.signal, { + probeTimeoutMs: bounds.probeTimeoutMs, + statementTimeoutMs: bounds.statementTimeoutMs, + })).then( + (payload) => finalize(pendingProbe, undefined, payload), + () => finalize(pendingProbe, "error"), + ); + }; + const admitQueued = (): void => { + for (const candidate of pending) { + if (activeSlots >= bounds.maxConcurrentProbes) return; + if (!candidate.started && !candidate.finalized) startProbe(candidate); + } + }; + const requestProbe = (state: BoundaryState, eventLoopLagMs: number): void => { + const watchdogAt = now(); + const pendingProbe: PendingProbe = { state, watchdogAt, host: hostSample(eventLoopLagMs), controller: new AbortController(), started: false, finalized: false }; + if (!options.probe) return finalize(pendingProbe, "error"); + if (!bounds.probeAllowed) return finalize(pendingProbe, "bounds-floor"); + pending.add(pendingProbe); + if (activeSlots < bounds.maxConcurrentProbes) { + startProbe(pendingProbe); + return; + } + if (bounds.probeQueueTimeoutMs === 0) return finalize(pendingProbe, "concurrency"); + pendingProbe.queuedTimer = setTimer(() => finalize(pendingProbe, "concurrency"), bounds.probeQueueTimeoutMs); + unref(pendingProbe.queuedTimer); + }; + const fireWatchdog = (state: BoundaryState): void => { + if (disposed || state.settled || state.watchdogFired) return; + state.watchdogFired = true; + // The overdue watchdog callback measures scheduling delay at the same instant + // as the host/cluster snapshot, without adding another timer or I/O path. + requestProbe(state, Math.max(0, now() - (state.startedAt + bounds.watchdog[state.boundary]))); + }; + const arm = (state: BoundaryState): void => { + state.timer = setTimer(() => fireWatchdog(state), bounds.watchdog[state.boundary]); + unref(state.timer); + }; + const complete = (state: BoundaryState, outcome: "resolved" | "rejected"): void => { + if (state.settled) return; + state.settled = true; + state.outcome = outcome; + if (state.timer) clearTimer(state.timer); + states.delete(state.handle.id); + if (state.key) windows.delete(state.key); + if (!state.watchdogFired && Math.max(0, now() - state.startedAt) >= bounds.thresholdMs) { + emit(state, "boundary-complete", { outcome, boundaryIncomplete: false }); + } + }; + const beforeExit = (): void => { void flush(); }; + process.on("beforeExit", beforeExit); + + const flush = async (timeoutMs = bounds.probeDrainTimeoutMs): Promise => { + if (pending.size === 0) return; + await new Promise((resolve) => { + const timer = setTimer(() => { + for (const item of [...pending]) finalize(item, "drain-timeout"); + resolve(); + }, Math.max(1, timeoutMs)); + unref(timer); + const check = (): void => { + if (pending.size === 0) { + clearTimer(timer); + resolve(); + } else setTimer(check, 1); + }; + check(); + }); + }; + + return { + enabled: true, + observeBoundary(boundary: PgTimeoutBoundary, phase: string, action: () => Promise): Promise { + const state: BoundaryState = { handle: { id: nextHandle++ }, boundary, phase, startedAt: now(), timestamp: new Date(wallNow()).toISOString(), host: hostSample(), settled: false, watchdogFired: false }; + states.set(state.handle.id, state); + arm(state); + let result: Promise; + try { result = action(); } catch (error) { complete(state, "rejected"); throw error; } + return result.then((value) => { complete(state, "resolved"); return value; }, (error: unknown) => { complete(state, "rejected"); throw error; }); + }, + openBoundary(boundary: PgTimeoutBoundary, phase: string, key: string): PgTimeoutBoundaryHandle { + const prior = windows.get(key); + const superseded = Boolean(prior && !prior.settled); + if (prior && !prior.settled) { + if (prior.timer) clearTimer(prior.timer); + prior.settled = true; + states.delete(prior.handle.id); + } + const state: BoundaryState = { handle: { id: nextHandle++ }, boundary, phase, key, startedAt: now(), timestamp: new Date(wallNow()).toISOString(), host: hostSample(), settled: false, watchdogFired: false, ...(superseded ? { supersededOpenWindow: true } : {}) }; + windows.set(key, state); + states.set(state.handle.id, state); + arm(state); + return state.handle; + }, + closeBoundary(handle: PgTimeoutBoundaryHandle, outcome = "resolved"): void { + const state = states.get(handle.id); + if (state) complete(state, outcome); + }, + flush, + async dispose(): Promise { + if (disposed) return; + disposed = true; + for (const state of states.values()) if (state.timer) clearTimer(state.timer); + states.clear(); + windows.clear(); + await flush(); + process.removeListener("beforeExit", beforeExit); + }, + }; +} diff --git a/packages/core/src/__tests__/pg-test-harness-observer-inertness.test.ts b/packages/core/src/__tests__/pg-test-harness-observer-inertness.test.ts new file mode 100644 index 0000000000..27a3f1d584 --- /dev/null +++ b/packages/core/src/__tests__/pg-test-harness-observer-inertness.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { createPgTimeoutBoundaryObserver } from "../__test-utils__/pg-timeout-boundary-observer.js"; + +/* +FNXC:PgTimeoutBoundaryObserver 2026-08-19-13:51: +The harness must remain byte-for-byte inert unless the explicit observer switch +is set. This cluster-free guard exercises the factory contract used by harness +setup, paired body hooks, teardown wrapping, and its drain. +*/ +describe("PG harness timeout-boundary observer inertness", () => { + it("does not create observable work without its opt-in environment switch", async () => { + let timers = 0; + let probes = 0; + let appends = 0; + const observer = createPgTimeoutBoundaryObserver({ + env: {}, + setTimer: ((callback, timeout) => { timers += 1; return setTimeout(callback, timeout); }) as typeof setTimeout, + probe: async () => { probes += 1; return { cluster: {}, template: {} }; }, + append: () => { appends += 1; }, + }); + const handle = observer.openBoundary("body", "shared.body", "inert"); + observer.closeBoundary(handle); + await expect(observer.observeBoundary("setup", "store.init", async () => "transparent")).resolves.toBe("transparent"); + await observer.flush(); + await observer.dispose(); + expect(observer.enabled).toBe(false); + expect(timers).toBe(0); + expect(probes).toBe(0); + expect(appends).toBe(0); + }); +}); diff --git a/packages/core/src/__tests__/pg-timeout-boundary-observer.test.ts b/packages/core/src/__tests__/pg-timeout-boundary-observer.test.ts new file mode 100644 index 0000000000..7cea050412 --- /dev/null +++ b/packages/core/src/__tests__/pg-timeout-boundary-observer.test.ts @@ -0,0 +1,177 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + __resetPgTimeoutBoundaryObserverProbeCountForTest, + createPgTimeoutBoundaryObserver, + MAX_CONCURRENT_PROBES_CEILING, + type PgTimeoutBoundaryProbePayload, +} from "../__test-utils__/pg-timeout-boundary-observer.js"; + +const payload: PgTimeoutBoundaryProbePayload = { + cluster: { activity: [{ pid: 1, state: "active" }], totalBackends: 4 }, + template: { goldenTemplateName: "fusion_schema_template_test", markerPresent: true }, +}; + +const env = { + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER: "1", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_LOG: "memory", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_THRESHOLD_MS: "0", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_WATCHDOG_MS: "10", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_TIMEOUT_MS: "100", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_STATEMENT_TIMEOUT_MS: "50", + FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_DRAIN_TIMEOUT_MS: "100", +}; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); + return { promise, resolve, reject }; +} + +function records(lines: string[]) { + return lines.map((line) => JSON.parse(line) as Record); +} + +afterEach(() => { + vi.useRealTimers(); + __resetPgTimeoutBoundaryObserverProbeCountForTest(); +}); + +describe("PG timeout boundary observer", () => { + it("is completely inert while disabled for wrapping and paired APIs", async () => { + const setTimer = vi.fn(setTimeout); + const append = vi.fn(); + const probe = vi.fn(); + const observer = createPgTimeoutBoundaryObserver({ env: {}, setTimer, append, probe }); + await expect(observer.observeBoundary("setup", "store.init", async () => "value")).resolves.toBe("value"); + await expect(observer.observeBoundary("setup", "store.init", async () => { throw new Error("unchanged"); })).rejects.toThrow("unchanged"); + observer.closeBoundary(observer.openBoundary("body", "shared.body", "body-key")); + await observer.flush(); + await observer.dispose(); + expect(observer.enabled).toBe(false); + expect(setTimer).not.toHaveBeenCalled(); + expect(append).not.toHaveBeenCalled(); + expect(probe).not.toHaveBeenCalled(); + }); + + it("arms before the action and retains a watchdog payload after a boundary settles", async () => { + vi.useFakeTimers(); + const lines: string[] = []; + const pending = deferred(); + const probe = vi.fn(() => pending.promise); + const observer = createPgTimeoutBoundaryObserver({ env, probe, append: (_path, line) => lines.push(line), writeError: () => {} }); + let release!: () => void; + const action = new Promise((resolve) => { release = resolve; }); + const observed = observer.observeBoundary("body", "shared.body", () => action); + await vi.advanceTimersByTimeAsync(10); + expect(probe).toHaveBeenCalledTimes(1); + release(); + await observed; + pending.resolve(payload); + await vi.advanceTimersByTimeAsync(0); + expect(records(lines)).toEqual([expect.objectContaining({ + trigger: "boundary-watchdog", boundary: "body", phase: "shared.body", boundaryIncomplete: true, + settledDuringProbe: true, cluster: payload.cluster, template: payload.template, + })]); + await observer.dispose(); + }); + + it("records watchdog scheduling drift as event-loop lag for host attribution", async () => { + const lines: string[] = []; + let now = 0; + let watchdog: (() => void) | undefined; + const observer = createPgTimeoutBoundaryObserver({ + env, + now: () => now, + setTimer: (callback) => { watchdog = callback; return 0 as unknown as ReturnType; }, + clearTimer: () => {}, + hostSample: (eventLoopLagMs = 0) => ({ loadavg1: 0, loadavg5: 0, loadavg15: 0, cpuCount: 8, eventLoopLagMs }), + probe: async () => payload, + append: (_path, line) => lines.push(line), + writeError: () => {}, + }); + const handle = observer.openBoundary("body", "shared.body", "lagged-body"); + now = 125; + watchdog?.(); + await new Promise((resolve) => setImmediate(resolve)); + observer.closeBoundary(handle); + expect(records(lines)).toContainEqual(expect.objectContaining({ + trigger: "boundary-watchdog", + boundary: "body", + host: expect.objectContaining({ eventLoopLagMs: 115 }), + })); + await observer.dispose(); + }); + + it("records all three concurrent boundaries at a raised concurrency limit", async () => { + vi.useFakeTimers(); + const lines: string[] = []; + const probes = [deferred(), deferred(), deferred()]; + let index = 0; + const observer = createPgTimeoutBoundaryObserver({ + env: { ...env, FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_CONCURRENT_PROBES: "3", FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_PROBES: "3" }, + probe: () => probes[index++]!.promise, + append: (_path, line) => lines.push(line), writeError: () => {}, + }); + const setup = observer.openBoundary("setup", "template.ensure", "setup"); + const body = observer.openBoundary("body", "shared.body", "body"); + const teardown = observer.openBoundary("teardown", "dropDatabase", "teardown"); + await vi.advanceTimersByTimeAsync(10); + observer.closeBoundary(setup); observer.closeBoundary(body); observer.closeBoundary(teardown); + probes.forEach((entry) => entry.resolve(payload)); + await vi.advanceTimersByTimeAsync(0); + expect(records(lines).filter((record) => record.trigger === "boundary-watchdog" && record.cluster)).toHaveLength(3); + await observer.dispose(); + }); + + it("suppresses a concurrent watchdog at the default limit and releases its slot for a later probe", async () => { + vi.useFakeTimers(); + const lines: string[] = []; + const first = deferred(); + const second = deferred(); + const probe = vi.fn().mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise); + const observer = createPgTimeoutBoundaryObserver({ env: { ...env, FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_PROBES: "3" }, probe, append: (_path, line) => lines.push(line), writeError: () => {} }); + observer.openBoundary("setup", "template.ensure", "first"); + observer.openBoundary("body", "shared.body", "second"); + await vi.advanceTimersByTimeAsync(10); + expect(records(lines)).toContainEqual(expect.objectContaining({ probeSuppressed: "concurrency", boundary: "body" })); + first.resolve(payload); + await vi.advanceTimersByTimeAsync(0); + observer.openBoundary("teardown", "dropDatabase", "third"); + await vi.advanceTimersByTimeAsync(10); + expect(probe).toHaveBeenCalledTimes(2); + second.resolve(payload); + await vi.advanceTimersByTimeAsync(0); + await observer.dispose(); + }); + + it("records threshold-zero completions and accepts a short watchdog with a long independent probe timeout", async () => { + vi.useFakeTimers(); + const lines: string[] = []; + const observer = createPgTimeoutBoundaryObserver({ + env: { ...env, FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_WATCHDOG_BODY_MS: "5", FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_PROBE_TIMEOUT_MS: "100" }, + append: (_path, line) => lines.push(line), writeError: () => {}, + }); + const handle = observer.openBoundary("body", "shared.body", "body"); + observer.closeBoundary(handle); + expect(records(lines)).toContainEqual(expect.objectContaining({ trigger: "boundary-complete", watchdogMs: 5, probeTimeoutMs: 100 })); + await observer.dispose(); + }); + + it("clamps concurrency to its ceiling and never fabricates paired completion on a superseded window", async () => { + vi.useFakeTimers(); + const lines: string[] = []; + const observer = createPgTimeoutBoundaryObserver({ + env: { ...env, FUSION_PG_TEST_TIMEOUT_BOUNDARY_OBSERVER_MAX_CONCURRENT_PROBES: "999" }, + append: (_path, line) => lines.push(line), writeError: () => {}, + }); + const stale = observer.openBoundary("body", "shared.body", "same"); + const fresh = observer.openBoundary("body", "shared.body", "same"); + observer.closeBoundary(stale); + observer.closeBoundary(fresh); + const complete = records(lines).find((record) => record.trigger === "boundary-complete"); + expect(complete).toMatchObject({ maxConcurrentProbes: MAX_CONCURRENT_PROBES_CEILING, boundsClamped: true }); + expect(records(lines).filter((record) => record.trigger === "boundary-complete")).toHaveLength(1); + await observer.dispose(); + }); +}); diff --git a/scripts/__tests__/pg-loaded-failure-census.test.mjs b/scripts/__tests__/pg-loaded-failure-census.test.mjs index 5535142730..f4c40919a8 100644 --- a/scripts/__tests__/pg-loaded-failure-census.test.mjs +++ b/scripts/__tests__/pg-loaded-failure-census.test.mjs @@ -8,6 +8,8 @@ import { classifyLifecyclePosition, extractFailingFiles, parseDiagnosticsJsonl, + parseBoundaryObserverJsonl, + classifyBoundaryAttribution, stripAnsi, } from "../pg-loaded-failure-census.mjs"; @@ -49,6 +51,35 @@ test("censuses every high-failure file and joins snapshot diagnostics", () => { assert.equal(extractFailingFiles(fixture("high-run.txt")).length, 25); }); +test("joins out-of-order watchdog payloads by file and boundary, not line order", () => { + const failure = { file: "src/__tests__/postgres/body-case.test.ts", lifecyclePosition: "test body" }; + const parsed = parseBoundaryObserverJsonl(`${JSON.stringify({ testFile: failure.file, boundary: "body", trigger: "boundary-complete", timestamp: "2026-01-01T00:00:02Z", host: { loadavg1: 0, cpuCount: 8, eventLoopLagMs: 0 } })}\n${JSON.stringify({ testFile: failure.file, boundary: "body", trigger: "boundary-watchdog", timestamp: "2026-01-01T00:00:01Z", settledDuringProbe: true, host: { loadavg1: 0, cpuCount: 8, eventLoopLagMs: 0 }, cluster: { activity: [{ state: "active", blockingPids: [44] }], locks: [] }, template: { markerPresent: true } })}\nmalformed`); + assert.equal(parsed.malformedLines, 1); + assert.equal(classifyBoundaryAttribution(failure, parsed.rows).classification, "cluster-implicated"); +}); + +test("keeps explicit unobservable sets and suppressed watchdog failures distinct from joined attribution", () => { + const body = { file: "src/__tests__/postgres/direct.test.ts", lifecyclePosition: "test body" }; + assert.equal(classifyBoundaryAttribution(body, [], [body.file]).classification, "body-unobservable"); + const suppressed = [{ testFile: body.file, boundary: "body", trigger: "boundary-watchdog", probeSuppressed: "single-flight", host: { loadavg1: 0, cpuCount: 8, eventLoopLagMs: 0 } }]; + assert.equal(classifyBoundaryAttribution(body, suppressed).classification, "unjoined"); + const fully = { file: "src/__tests__/postgres/no-harness.test.ts", lifecyclePosition: "afterEach" }; + const census = buildCensus({ + log: ` FAIL ${fully.file} > leaves no harness boundary\nError: afterEach hook timed out in 15000ms.\n\n Test Files 1 failed (1)\n`, + fullyUnobservableFiles: [fully.file], + }); + assert.equal(census.fullyUnobservableFailingFileCount, 1); + assert.deepEqual(census.fullyUnobservableFailingFiles, [fully.file]); + assert.equal(census.attributions[0].boundaryAttribution.classification, "unjoined"); +}); + +test("requires a golden advisory waiter, not a holder, for template convoy attribution", () => { + const failure = { file: "src/__tests__/postgres/template.test.ts", lifecyclePosition: "beforeAll hook" }; + const base = { testFile: failure.file, boundary: "setup", trigger: "boundary-watchdog", host: { loadavg1: 0, cpuCount: 8, eventLoopLagMs: 0 }, cluster: { activity: [], locks: [] } }; + assert.notEqual(classifyBoundaryAttribution(failure, [{ ...base, template: { advisoryHolders: [10], advisoryWaiters: [], isOwner: false } }]).classification, "template-convoy"); + assert.equal(classifyBoundaryAttribution(failure, [{ ...base, template: { advisoryHolders: [10], advisoryWaiters: [11], isOwner: false } }]).classification, "template-convoy"); +}); + test("reports a complete healthy run as measured zero rather than insufficient data", () => { const census = buildCensus({ log: fixture("low-run.txt"), diagnostics: [], ordinarySlotCeiling: 97 }); assert.equal(census.status, "measured"); diff --git a/scripts/pg-loaded-failure-census.mjs b/scripts/pg-loaded-failure-census.mjs index ee9bdbb370..3d05ac12f7 100644 --- a/scripts/pg-loaded-failure-census.mjs +++ b/scripts/pg-loaded-failure-census.mjs @@ -34,7 +34,12 @@ export function parseDiagnosticsJsonl(text) { } function normalizeFile(value) { - const match = String(value).replaceAll("\\", "/").match(/(?:[\w@.-]+\/)*[\w@.-]+(?:\.pg)?\.test\.[cm]?[jt]sx?/i); + const normalized = String(value).replaceAll("\\", "/"); + // Vitest diagnostics use absolute paths while runner failures use repo paths. + // Canonicalize at the test-root segment before any key-based observer join. + const testRoot = normalized.indexOf("src/__tests__/"); + if (testRoot >= 0) return normalized.slice(testRoot); + const match = normalized.match(/(?:[\w@.-]+\/)*[\w@.-]+(?:\.pg)?\.test\.[cm]?[jt]sx?/i); return match?.[0] ?? null; } @@ -93,6 +98,86 @@ export function parseFileSummary(log) { return { complete: true, totalFiles: failed + passed + skipped, reportedFailedFiles: failed }; } +export function parseBoundaryObserverJsonl(text) { + return parseDiagnosticsJsonl(text); +} + +function boundaryForLifecycle(lifecyclePosition) { + if (lifecyclePosition === "beforeAll hook" || lifecyclePosition === "in-test setup") return "setup"; + if (lifecyclePosition === "afterEach" || lifecyclePosition === "afterAll hook" || lifecyclePosition === "global setup-teardown") return "teardown"; + return "body"; +} + +function observerFile(record) { + return normalizeFile(record?.testFile); +} + +/** + * FNXC:PgTimeoutBoundaryObserver 2026-08-19-13:51: + * Watchdog payloads win because they are the only records carrying a cluster + * snapshot. Completion-only and suppressed records may support host evidence, + * but must never be promoted to cluster causation by inference. + */ +export function classifyBoundaryAttribution(failure, observerRecords, bodyUnobservableFiles = [], fullyUnobservableFiles = []) { + const boundary = boundaryForLifecycle(failure.lifecyclePosition); + const sameFile = observerRecords.filter((record) => observerFile(record) === failure.file && record.boundary === boundary); + if (fullyUnobservableFiles.includes(failure.file)) { + // These files never enter a harness-owned boundary, so their missing join + // is an explicit coverage limit rather than a failed observer correlation. + return { classification: "unjoined", boundary, record: null, hostOnly: false, fullyUnobservable: true }; + } + if (boundary === "body" && bodyUnobservableFiles.includes(failure.file)) { + return { classification: "body-unobservable", boundary, record: null, hostOnly: false, fullyUnobservable: false }; + } + const watchdog = sameFile.filter((record) => record.trigger === "boundary-watchdog"); + const record = watchdog.find((candidate) => !candidate.probeSuppressed && candidate.cluster && candidate.template) + ?? watchdog[0] + ?? sameFile[0] + ?? null; + if (!record) return { classification: "unjoined", boundary, record: null, hostOnly: false }; + const hostOnly = record.trigger !== "boundary-watchdog" || Boolean(record.probeSuppressed) || !record.cluster; + if (hostOnly) { + const load = Number(record?.host?.loadavg1); + const cpus = Number(record?.host?.cpuCount); + const lag = Number(record?.host?.eventLoopLagMs); + return { classification: (Number.isFinite(load) && Number.isFinite(cpus) && load >= cpus) || lag >= 100 ? "host-implicated" : "unjoined", boundary, record, hostOnly: true }; + } + const template = record.template ?? {}; + // A holder alone is not a convoy: only a non-owner waiter proves the + // timed-out boundary was queued behind the golden template advisory lock. + if (Array.isArray(template.advisoryWaiters) && template.advisoryWaiters.length > 0 && template.isOwner === false) { + return { classification: "template-convoy", boundary, record, hostOnly: false, fullyUnobservable: false }; + } + const cluster = record.cluster ?? {}; + const active = Array.isArray(cluster.activity) && cluster.activity.some((row) => row?.state === "active" || row?.wait_event || row?.blockingPids?.length); + const blocked = Array.isArray(cluster.locks) && cluster.locks.some((lock) => lock?.granted === false || lock?.blockingPids?.length); + if (active || blocked) return { classification: "cluster-implicated", boundary, record, hostOnly: false }; + const load = Number(record?.host?.loadavg1); + const cpus = Number(record?.host?.cpuCount); + const lag = Number(record?.host?.eventLoopLagMs); + if ((Number.isFinite(load) && Number.isFinite(cpus) && load >= cpus) || lag >= 100) return { classification: "host-implicated", boundary, record, hostOnly: false }; + return { classification: "unjoined", boundary, record, hostOnly: false }; +} + +export function summarizeBoundaryObserver(records, failures, bodyUnobservableFiles = [], fullyUnobservableFiles = []) { + const rows = Array.isArray(records) ? records : []; + const attributions = failures.map((failure) => ({ ...failure, boundaryAttribution: classifyBoundaryAttribution(failure, rows, bodyUnobservableFiles, fullyUnobservableFiles) })); + const suppression = {}; + for (const row of rows) { + const reason = row?.probeSuppressed === "single-flight" ? "concurrency" : row?.probeSuppressed; + if (reason) suppression[reason] = (suppression[reason] ?? 0) + 1; + } + return { + boundaryObserver: rows.length ? "present" : "absent", + boundaryAttributionHistogram: Object.fromEntries(Object.entries(Object.groupBy(attributions, (row) => row.boundaryAttribution.classification)).map(([key, values]) => [key, values.length])), + observerProbeSuppression: suppression, + settledDuringProbeCount: rows.filter((row) => row?.settledDuringProbe === true).length, + fullyUnobservableFailingFiles: attributions.filter((row) => row.boundaryAttribution.fullyUnobservable).map((row) => row.file), + fullyUnobservableFailingFileCount: attributions.filter((row) => row.boundaryAttribution.fullyUnobservable).length, + attributions, + }; +} + export function summarizeDiagnostics(diagnostics) { const input = Array.isArray(diagnostics) ? diagnostics : []; const waits = new Map(); @@ -126,7 +211,7 @@ export function summarizeDiagnostics(diagnostics) { }; } -export function buildCensus({ log, diagnostics = [], ordinarySlotCeiling = null, subjects = [] }) { +export function buildCensus({ log, diagnostics = [], boundaryObserver = [], bodyUnobservableFiles = [], fullyUnobservableFiles = [], ordinarySlotCeiling = null, subjects = [] }) { const summary = parseFileSummary(log); if (!summary.complete) { return { status: "insufficient-data", reason: "missing Test Files summary", totalFiles: null, failingFiles: [], failingFileCount: null }; @@ -136,6 +221,7 @@ export function buildCensus({ log, diagnostics = [], ordinarySlotCeiling = null, return { status: "insufficient-data", reason: `summary reports ${summary.reportedFailedFiles} failed files but ${failingFiles.length} failure blocks were parsed`, totalFiles: summary.totalFiles, failingFiles, failingFileCount: null }; } const diagnosticSummary = summarizeDiagnostics(diagnostics); + const observerSummary = summarizeBoundaryObserver(boundaryObserver, failingFiles, bodyUnobservableFiles, fullyUnobservableFiles); const ceiling = Number.isFinite(ordinarySlotCeiling) && ordinarySlotCeiling >= 0 ? ordinarySlotCeiling : null; return { status: "measured", @@ -148,15 +234,19 @@ export function buildCensus({ log, diagnostics = [], ordinarySlotCeiling = null, ordinarySlotCeiling: ceiling, backendHeadroom: ceiling != null && diagnosticSummary.peakBackends != null ? ceiling - diagnosticSummary.peakBackends : null, ...diagnosticSummary, + ...observerSummary, }; } function parseArgs(args) { - const result = { log: undefined, diagnostics: undefined, ordinarySlotCeiling: null, subjects: [] }; + const result = { log: undefined, diagnostics: undefined, boundaryObserver: undefined, bodyUnobservableFiles: undefined, fullyUnobservableFiles: undefined, ordinarySlotCeiling: null, subjects: [] }; for (let index = 0; index < args.length; index += 1) { const argument = args[index]; if (argument === "--log") result.log = args[++index]; else if (argument === "--diagnostics") result.diagnostics = args[++index]; + else if (argument === "--boundary-observer") result.boundaryObserver = args[++index]; + else if (argument === "--body-unobservable-files") result.bodyUnobservableFiles = args[++index]; + else if (argument === "--fully-unobservable-files") result.fullyUnobservableFiles = args[++index]; else if (argument === "--ordinary-slot-ceiling") result.ordinarySlotCeiling = Number(args[++index]); else if (argument === "--subject") result.subjects.push(args[++index]); else throw new Error(`Unknown argument: ${argument}`); @@ -168,5 +258,8 @@ function parseArgs(args) { if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { const args = parseArgs(process.argv.slice(2)); const parsed = args.diagnostics ? parseDiagnosticsJsonl(readFileSync(args.diagnostics, "utf8")) : { rows: [], malformedLines: 0 }; - console.log(JSON.stringify({ ...buildCensus({ log: readFileSync(args.log, "utf8"), diagnostics: parsed.rows, ordinarySlotCeiling: args.ordinarySlotCeiling, subjects: args.subjects }), malformedDiagnosticLines: parsed.malformedLines }, null, 2)); + const observer = args.boundaryObserver ? parseBoundaryObserverJsonl(readFileSync(args.boundaryObserver, "utf8")) : { rows: [], malformedLines: 0 }; + const bodyUnobservableFiles = args.bodyUnobservableFiles ? readFileSync(args.bodyUnobservableFiles, "utf8").split(/\\r?\\n/).map(normalizeFile).filter(Boolean) : []; + const fullyUnobservableFiles = args.fullyUnobservableFiles ? readFileSync(args.fullyUnobservableFiles, "utf8").split(/\\r?\\n/).map(normalizeFile).filter(Boolean) : []; + console.log(JSON.stringify({ ...buildCensus({ log: readFileSync(args.log, "utf8"), diagnostics: parsed.rows, boundaryObserver: observer.rows, bodyUnobservableFiles, fullyUnobservableFiles, ordinarySlotCeiling: args.ordinarySlotCeiling, subjects: args.subjects }), malformedDiagnosticLines: parsed.malformedLines, malformedBoundaryObserverLines: observer.malformedLines }, null, 2)); }