FN-9149: Add PostgreSQL timeout-boundary diagnostics
Instrument the opt-in PostgreSQL test harness to attribute loaded-lane timeout failures without changing default behavior. - Add bounded setup, body, and teardown watchdog probes with host, cluster, and template evidence. - Wire observer records into the harness and loaded-failure census with explicit suppression and attribution handling. - Cover observer inertness, boundary behavior, and census joins while documenting the 27-worker campaign findings. Files changed: ...res-loaded-lane-unrelated-failure-population.md | 31 +- docs/testing.md | 27 ++ .../core/src/__test-utils__/pg-test-harness.ts | 113 +++++- .../__test-utils__/pg-timeout-boundary-observer.ts | 451 +++++++++++++++++++++ .../pg-test-harness-observer-inertness.test.ts | 31 ++ .../__tests__/pg-timeout-boundary-observer.test.ts | 177 ++++++++ .../__tests__/pg-loaded-failure-census.test.mjs | 31 ++ scripts/pg-loaded-failure-census.mjs | 101 ++++- 8 files changed, 934 insertions(+), 28 deletions(-) Fusion-Task-Id: FN-9149 Fusion-Task-Lineage: 4df4ee28-5369-41ae-bb0c-e7e9ae78d873 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
<!-- FNXC:PgTimeoutBoundaryObserver 2026-08-19-14:33: FN-9149 requires a default-off observer that attributes setup, body, and teardown timeout boundaries without widening the budget it measures. The paired body window is necessary because Vitest invokes shared-harness beforeEach and afterEach as distinct hooks. -->
|
||||
|
||||
### 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 <jsonl>`, `--body-unobservable-files <list>`, and `--fully-unobservable-files <list>` 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:
|
||||
|
||||
@@ -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<PgTimeoutBoundaryProbePayload> {
|
||||
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<Array<PgTeardownActivityRow & { blockingPids?: number[] }>>(`
|
||||
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<Array<{ pid: number; locktype: string; granted: boolean; blockingPids: number[] }>>(`
|
||||
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<Array<{ markerPresent: boolean; ownerPid: number | null; advisoryHolders: number[]; advisoryWaiters: number[] }>>(
|
||||
`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<void> => {
|
||||
@@ -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<string, unknown> | 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;
|
||||
}
|
||||
|
||||
451
packages/core/src/__test-utils__/pg-timeout-boundary-observer.ts
Normal file
451
packages/core/src/__test-utils__/pg-timeout-boundary-observer.ts
Normal file
@@ -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<typeof setTimeout>;
|
||||
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<string, unknown>;
|
||||
readonly template: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** 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<string, unknown>;
|
||||
readonly template?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
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<PgTimeoutBoundaryProbePayload>;
|
||||
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<T>(boundary: PgTimeoutBoundary, phase: string, action: () => Promise<T>): Promise<T>;
|
||||
openBoundary(boundary: PgTimeoutBoundary, phase: string, key: string): PgTimeoutBoundaryHandle;
|
||||
closeBoundary(handle: PgTimeoutBoundaryHandle, outcome?: "resolved" | "rejected"): void;
|
||||
flush(timeoutMs?: number): Promise<void>;
|
||||
dispose(): Promise<void>;
|
||||
}
|
||||
|
||||
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<PgTimeoutBoundary, number>;
|
||||
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<PgTimeoutBoundary, number>;
|
||||
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<T>(_boundary: PgTimeoutBoundary, _phase: string, action: () => Promise<T>): Promise<T> { 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<number, BoundaryState>();
|
||||
const windows = new Map<string, BoundaryState>();
|
||||
const pending = new Set<PendingProbe>();
|
||||
|
||||
const unref = (timer: TimerHandle | undefined): void => {
|
||||
(timer as unknown as { unref?: () => void } | undefined)?.unref?.();
|
||||
};
|
||||
const emit = (state: BoundaryState, trigger: PgTimeoutBoundaryTrigger, fields: Partial<PgTimeoutBoundaryRecord>): 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<void> => {
|
||||
if (pending.size === 0) return;
|
||||
await new Promise<void>((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<T>(boundary: PgTimeoutBoundary, phase: string, action: () => Promise<T>): Promise<T> {
|
||||
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<T>;
|
||||
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<void> {
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
177
packages/core/src/__tests__/pg-timeout-boundary-observer.test.ts
Normal file
177
packages/core/src/__tests__/pg-timeout-boundary-observer.test.ts
Normal file
@@ -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<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
let reject!: (error: Error) => void;
|
||||
const promise = new Promise<T>((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; });
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
function records(lines: string[]) {
|
||||
return lines.map((line) => JSON.parse(line) as Record<string, unknown>);
|
||||
}
|
||||
|
||||
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<PgTimeoutBoundaryProbePayload>();
|
||||
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<void>((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<typeof setTimeout>; },
|
||||
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<void>((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<PgTimeoutBoundaryProbePayload>(), deferred<PgTimeoutBoundaryProbePayload>(), deferred<PgTimeoutBoundaryProbePayload>()];
|
||||
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<PgTimeoutBoundaryProbePayload>();
|
||||
const second = deferred<PgTimeoutBoundaryProbePayload>();
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user