FN-9131: Add PostgreSQL harness budget characterization
Characterize cluster-shared PostgreSQL connection admission while keeping regressive harness wiring disabled. - add advisory-lock slot budgeting, bootstrap gating, and local allocation accounting - cover budget arithmetic, queueing, degradation, and PostgreSQL lock behavior - document loaded-lane failures and the terminal-negative lifecycle boundary - clarify that the active harness neither admits nor clamps against the experimental budget Files changed: .../test-failures/pg-harness-connection-budget.md | 21 + .../suite-only-flakes-observed-register.md | 2 + docs/testing.md | 6 + .../__tests__/pg-connection-budget.test.ts | 146 ++++++ .../src/__test-utils__/pg-connection-budget.ts | 492 +++++++++++++++++++++ .../core/src/__test-utils__/pg-test-harness.ts | 8 + .../postgres/pg-connection-budget.pg.test.ts | 36 ++ 7 files changed, 711 insertions(+) Fusion-Task-Id: FN-9131 Fusion-Task-Lineage: c83611ad-95f9-44ef-a0fd-1f182a26725d Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
21
docs/solutions/test-failures/pg-harness-connection-budget.md
Normal file
21
docs/solutions/test-failures/pg-harness-connection-budget.md
Normal file
@@ -0,0 +1,21 @@
|
||||
---
|
||||
category: test-failures
|
||||
module: testing
|
||||
problem_type: loaded-postgresql-capacity
|
||||
applies_when: PostgreSQL harnesses time out or lose connections under high Vitest fork fan-out.
|
||||
tags: [postgresql, vitest, harness, connection-budget, fn-9131]
|
||||
---
|
||||
|
||||
# PostgreSQL harness connection-budget terminal negative
|
||||
|
||||
FN-9131 reproduced the project-identity loaded-lane symptom with 27 workers on a local PostgreSQL 15.15 cluster (`max_connections=100`, `superuser_reserved_connections=3`). The original subject passed, but the PostgreSQL directory run failed broadly: the first budget wiring failed 135 files in 174.1 seconds, and the queueing/lease-retention revision failed 144 files in 223.3 seconds. Neither result is acceptable evidence for shipping harness admission.
|
||||
|
||||
The experimental primitive remains available but is deliberately **unwired** from `pg-test-harness.ts`. It derives a closed server-shared advisory-lock space: a backend is one slot; the default minimum harness cost is runtime pool + dedicated migration pool + admin pool (`1 + 1 + 1 = 3`); template construction requires three funded slots; a participant's reserve is six work slots plus a lease slot. On the measured cluster that yields 85 usable slots, a floor cost of seven, and 12 participants. At P=27, oversubscription is normal and must queue rather than throw.
|
||||
|
||||
The primitive uses a fixed advisory-lock class ID, separate lease/work bands, conservative degraded derivation, and a fixed-name atomic-directory bootstrap token gate. Node v26.3.0 has no `fs.flock`, `fs.flockSync`, or `O_EXLOCK`; the token is therefore payload-free, reclaimed only after `TOKEN_STALE_MS`, and is not correctness-critical. A failed bootstrap connection closes before retry. The degraded floor must never be raised to a typical cluster value. FN-9130's DDL admission key remains separate and unwired.
|
||||
|
||||
The initial wiring exposed two implementation defects: bounded registry retries threw `PgConnectionBudgetConcurrencyError` into tests, and failed reservations released their lease, creating bootstrap/lease thrash. FN-9131 changed the primitive so registry contention queues and retains only a lease during a reserve retry; local ledger failures remain the only `PgConnectionBudgetConcurrencyError` case. That repaired the local primitive but not the loaded harness outcome.
|
||||
|
||||
A setup-module top-level admission was not shipped: the shared setup module is loaded by core unit and engine workers with no PostgreSQL harness participants, so it has no per-worker PostgreSQL-suite signal and would consume cluster slots in unrelated lanes. This violates the required inertness condition. FN-9139 owns finding a lifecycle boundary that admits only actual PostgreSQL harness participants off the individual 15-second test budget before the registry is reconsidered.
|
||||
|
||||
No timeout, retry, skip, assertion weakening, quarantine, or worker-cap change was made. The dedicated primitive tests remain as characterization; loaded harness connection admission must remain unwired until a successor demonstrates the 27-worker and concurrent gate shapes without a wall-time regression.
|
||||
@@ -333,3 +333,5 @@ AssertionError: expected [ 'approved', 'created' ] to deeply equal [ 'created',
|
||||
**Resolved 2026-08-16 (FN-9132):** This was a product ordering defect in `getApprovalAuditHistory`, not PostgreSQL DDL contention, harness identity reuse, or test timing. `appendAuditEvent` creates deterministic IDs containing the event type, while the read ordered tied timestamps by `id ASC`; that lexically placed `approved` before `created`. The read now applies a lifecycle rank derived from `APPROVAL_REQUEST_AUDIT_EVENT_TYPES`, followed by ID only as a final total-order tiebreak. Regression coverage freezes `Date` around real create/decide/complete writes and proves tied approved, denied, and completed states, distinct timestamps, mixed ties, project isolation, and the public store delegate. No timeout, retry, worker-count, skip, assertion weakening, or quarantine change was made; `quarantinedCoreTests` remains empty.
|
||||
|
||||
This resolves the previously unclassified “unrelated satellite-store ordering failure” mentions in entry 1's 12-worker verification table, entry 2's 12-worker verification table, and entry 11's FN-9129 4-worker run table. Those sightings are now classified separately from their entries' identity and DDL investigations.
|
||||
|
||||
**Terminal negative 2026-08-17 (FN-9131):** The reproduced 27-worker PostgreSQL-directory symptom was investigated with a cluster-shared connection-budget primitive. The first harness wiring and a follow-up that queued registry over-subscription while retaining leases both made the loaded run worse (135 failed files in 174.1s, then 144 failed files in 223.3s); the subject itself was not the only failure. The harness wiring was reverted, the primitive remains characterized independently, and FN-9139 owns a setup-safe admission boundary. No quarantine, timeout change, test retry, skip, worker cap, or assertion change was made.
|
||||
|
||||
@@ -25,6 +25,12 @@ Gate membership is the explicit allow-list in `packages/engine/vitest.config.ts`
|
||||
<!-- FNXC:PgTestBootstrap 2026-08-16-18:59: PostgreSQL integration fixtures must use pg-test-harness bootstrap primitives so reachability, bounded maintenance DDL, and forced cleanup do not drift between files under forked loaded runs. Select createEmptyPgTestDatabase when the test proves first application/upgrades; use a baselined clone only when the schema-present state itself is the contract. -->
|
||||
**PostgreSQL fixture bootstrap:** Do not hand-roll per-file `CREATE DATABASE` or drop helpers. Use `createEmptyPgTestDatabase` for migration-application and upgrade contracts, and `createBaselinedPgTestDatabase` only when an already-applied schema is the intended fixture state. Both keep database lifecycle and cleanup behavior aligned with the shared harness. Engine reliability fixtures follow the same bounded maintenance-connection and forced-cleanup contract; see [the reliability helper DDL audit](solutions/test-failures/postgres-reliability-helper-ddl-audit.md).
|
||||
|
||||
### PostgreSQL connection-budget experiment
|
||||
|
||||
FN-9131 retains `pg-connection-budget.ts` as a tested cluster-shared advisory-lock primitive, but it is deliberately **not wired into the shared harness**. The experiment charges a backend slot at the pool ceiling (runtime + dedicated migration + admin), partitions fixed lease/work bands, derives a conservative degraded floor, and records bootstrap/admission counters (`bootstrapRetries`, `bootstrapTokenWaits`, `bootstrapTokenReclaims`, `deburstDelayMs`, `floorAdmissionWaits`, `maxFloorWaitMs`, `lazyAdmissions`, `concurrencyRejections`, and `degradedCount`). A non-zero degradation count invalidates a capacity measurement.
|
||||
|
||||
The initial harness wiring made a 27-worker PostgreSQL directory run worse, even after registry contention was changed from bounded rejection to queueing. Do not restore the wiring, alter test timeouts, add retries, or cap workers as a workaround. FN-9139 owns selecting a per-worker lifecycle point that is inert for non-PostgreSQL lanes and can charge admission wait outside individual test budgets. See [the terminal-negative record](solutions/test-failures/pg-harness-connection-budget.md).
|
||||
|
||||
<!-- FNXC:EngineTests 2026-07-08-03:00: FN-7667 decouples the engine-core gate's module graph from full-barrel growth so new feature modules don't silently inflate every gate fork's transform/import cost. -->
|
||||
**Gate-safe `@fusion/core` barrel:** the `engine-core` project resolves `@fusion/core` to `packages/core/src/index.gate.ts` (a project-scoped `resolve.alias`, not the root map), not the full `packages/core/src/index.ts` barrel. `index.gate.ts` is a byte-for-byte copy of the full barrel minus the `export ... from` statements for modules added to the barrel after the last re-audit baseline — i.e. it re-exports everything the full barrel does except genuinely new, gate-irrelevant feature modules (diffed against the prior baseline commit's barrel, not hand-picked from what gate *test* files import — production modules under test pull in far more of the barrel transitively than their own imports suggest). `engine-default`/`engine-reliability`/`engine-slow` are unaffected and keep resolving the full barrel. `@fusion/engine` is untouched (no gate file imports it). When adding a new barrel module that no gate test needs, mirror the exclusion in `index.gate.ts` rather than letting gate wall-time grow — see the FNXC comment at the top of `index.gate.ts` and `packages/engine/vitest.config.ts`'s `engine-core` project for the audit procedure.
|
||||
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
const postgresMockState = vi.hoisted(() => ({
|
||||
calls: 0,
|
||||
heldLocks: new Set<string>(),
|
||||
}));
|
||||
|
||||
vi.mock("postgres", () => ({
|
||||
default: vi.fn(() => {
|
||||
postgresMockState.calls += 1;
|
||||
const clientLocks = new Set<string>();
|
||||
return {
|
||||
unsafe: async (query: string) => {
|
||||
if (query.includes("current_setting")) {
|
||||
return [{ max_connections: "100", superuser_reserved_connections: "3" }];
|
||||
}
|
||||
const match = query.match(/\((\d+), (\d+)\)/);
|
||||
const lock = match ? `${match[1]}:${match[2]}` : undefined;
|
||||
if (query.includes("pg_try_advisory_lock")) {
|
||||
if (!lock || postgresMockState.heldLocks.has(lock)) return [{ acquired: false }];
|
||||
postgresMockState.heldLocks.add(lock);
|
||||
clientLocks.add(lock);
|
||||
return [{ acquired: true }];
|
||||
}
|
||||
if (query.includes("pg_advisory_unlock") && lock) {
|
||||
postgresMockState.heldLocks.delete(lock);
|
||||
clientLocks.delete(lock);
|
||||
}
|
||||
return [];
|
||||
},
|
||||
end: async () => {
|
||||
for (const lock of clientLocks) postgresMockState.heldLocks.delete(lock);
|
||||
clientLocks.clear();
|
||||
},
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
import {
|
||||
__resetPgConnectionBudgetForTest,
|
||||
__tryAcquirePgConnectionBudgetTokenForTest,
|
||||
BOOTSTRAP_RESERVE,
|
||||
DEGRADED_MAX_SUPERUSER_RESERVED,
|
||||
DEGRADED_MIN_MAX_CONNECTIONS,
|
||||
MAX_LIVE_HARNESSES_PER_PARTICIPANT,
|
||||
MIGRATION_SLOTS_PER_HARNESS,
|
||||
PG_FORK_WORK_RESERVE,
|
||||
TEMPLATE_BUILD_SLOTS,
|
||||
deriveDegradedPgSlotSpace,
|
||||
derivePgConnectionBudget,
|
||||
derivePgForkWorkReserve,
|
||||
derivePgSlotSpace,
|
||||
expectedPgAdmissionWaitMs,
|
||||
PgConnectionBudgetConcurrencyError,
|
||||
resolvePgConnectionBudget,
|
||||
} from "../pg-connection-budget.js";
|
||||
|
||||
describe("PostgreSQL test connection budget arithmetic", () => {
|
||||
it("charges a harness at its runtime, migration, and admin ceilings", () => {
|
||||
const reserve = derivePgForkWorkReserve({
|
||||
maxLiveHarnesses: MAX_LIVE_HARNESSES_PER_PARTICIPANT,
|
||||
flooredPoolMax: 1,
|
||||
migrationSlots: MIGRATION_SLOTS_PER_HARNESS,
|
||||
flooredAdminMax: 1,
|
||||
templateBuildSlots: TEMPLATE_BUILD_SLOTS,
|
||||
});
|
||||
expect(reserve).toEqual({ minHarnessSlotCost: 3, forkWorkReserve: 6, floorSlotCost: 7 });
|
||||
expect(PG_FORK_WORK_RESERVE).toEqual(reserve);
|
||||
|
||||
const minimum = derivePgConnectionBudget({ lentWorkSlots: 3, liveHarnesses: 1 });
|
||||
const funded = derivePgConnectionBudget({ lentWorkSlots: 6, liveHarnesses: 1 });
|
||||
expect(minimum).toMatchObject({ poolMax: 1, migrationSlots: 1, adminMax: 1, totalSlots: 3, floored: true });
|
||||
expect(funded.totalSlots).toBe(funded.poolMax + funded.migrationSlots + funded.adminMax);
|
||||
expect(funded.poolMax).toBeGreaterThan(minimum.poolMax);
|
||||
});
|
||||
|
||||
it("partitions a cluster-derived closed space into lease and work bands", () => {
|
||||
const space = derivePgSlotSpace({ maxConnections: 100, superuserReserved: 3, foreignReserve: 8, bootstrapReserve: BOOTSTRAP_RESERVE, floorSlotCost: 7 });
|
||||
expect(space).toEqual({ slotCount: 85, maxParticipants: 12, leaseBand: [0, 12], workBand: [12, 85] });
|
||||
expect(space.maxParticipants * 7).toBeLessThanOrEqual(space.slotCount);
|
||||
});
|
||||
|
||||
it("keeps conservative degraded ranges inside healthy ranges", () => {
|
||||
const degraded = deriveDegradedPgSlotSpace();
|
||||
expect(DEGRADED_MIN_MAX_CONNECTIONS).toBe(20);
|
||||
expect(DEGRADED_MAX_SUPERUSER_RESERVED).toBe(5);
|
||||
for (const maxConnections of [20, 50, 100, 250]) {
|
||||
for (const superuserReserved of [0, 3, 5]) {
|
||||
const healthy = derivePgSlotSpace({ maxConnections, superuserReserved, foreignReserve: 8, bootstrapReserve: 4, floorSlotCost: 7 });
|
||||
expect(degraded.slotCount).toBeLessThanOrEqual(healthy.slotCount);
|
||||
expect(degraded.leaseBand[1]).toBeLessThanOrEqual(healthy.leaseBand[1]);
|
||||
expect(degraded.workBand[1]).toBeLessThanOrEqual(healthy.workBand[1]);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps measured queueing below one third of the unchanged test budget", () => {
|
||||
expect(expectedPgAdmissionWaitMs({ participants: 27, maxParticipants: 12, p95WindowMs: 1_000, lingerMs: 0, deburstMs: 2_000 })).toBe(5_000);
|
||||
});
|
||||
|
||||
it("reserves the concurrency error for unfundable local ledger requests", () => {
|
||||
expect(() => derivePgConnectionBudget({ lentWorkSlots: 2, liveHarnesses: 1 }))
|
||||
.toThrow(PgConnectionBudgetConcurrencyError);
|
||||
expect(() => derivePgConnectionBudget({ lentWorkSlots: 2, liveHarnesses: 1 }))
|
||||
.toThrow("FORK_WORK_RESERVE");
|
||||
});
|
||||
|
||||
it("does not derive capacity from a lane-local worker fairness hint", () => {
|
||||
const first = derivePgSlotSpace({ maxConnections: 100, superuserReserved: 3, foreignReserve: 8, bootstrapReserve: 4, floorSlotCost: 7 });
|
||||
const second = derivePgSlotSpace({ maxConnections: 100, superuserReserved: 3, foreignReserve: 8, bootstrapReserve: 4, floorSlotCost: 7 });
|
||||
expect(second).toEqual(first);
|
||||
});
|
||||
|
||||
it("serializes concurrent first-window callers into one lease session", async () => {
|
||||
postgresMockState.calls = 0;
|
||||
postgresMockState.heldLocks.clear();
|
||||
try {
|
||||
await Promise.all([
|
||||
resolvePgConnectionBudget({ available: true, urlBase: "postgres://localhost/fusion" }),
|
||||
resolvePgConnectionBudget({ available: true, urlBase: "postgres://localhost/fusion" }),
|
||||
]);
|
||||
expect(postgresMockState.calls).toBe(1);
|
||||
} finally {
|
||||
await __resetPgConnectionBudgetForTest();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps healthy token exhaustion outside the ungated bootstrap fallback", async () => {
|
||||
// FNXC:PgTestConnectionBudget 2026-08-17-01:55:
|
||||
// A full healthy host semaphore queues before a bootstrap connect; R11 applies
|
||||
// only when the token directory itself is unavailable, never at capacity.
|
||||
const attempts = await Promise.all(
|
||||
Array.from({ length: BOOTSTRAP_RESERVE * 2 }, (_, index) =>
|
||||
__tryAcquirePgConnectionBudgetTokenForTest(index % BOOTSTRAP_RESERVE),
|
||||
),
|
||||
);
|
||||
const acquired = attempts.filter((attempt) => attempt.kind === "acquired");
|
||||
try {
|
||||
expect(acquired).toHaveLength(BOOTSTRAP_RESERVE);
|
||||
expect(attempts.filter((attempt) => attempt.kind === "exhausted")).toHaveLength(BOOTSTRAP_RESERVE);
|
||||
expect(attempts.some((attempt) => attempt.kind === "unavailable")).toBe(false);
|
||||
} finally {
|
||||
await Promise.all(acquired.map((attempt) => attempt.token.release()));
|
||||
}
|
||||
});
|
||||
});
|
||||
492
packages/core/src/__test-utils__/pg-connection-budget.ts
Normal file
492
packages/core/src/__test-utils__/pg-connection-budget.ts
Normal file
@@ -0,0 +1,492 @@
|
||||
import { mkdir, rm, stat } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import postgres, { type Sql } from "postgres";
|
||||
|
||||
/** A server backend is the unit used by the PostgreSQL test-harness budget. */
|
||||
export interface PgSlotSpace {
|
||||
slotCount: number;
|
||||
maxParticipants: number;
|
||||
leaseBand: readonly [number, number];
|
||||
workBand: readonly [number, number];
|
||||
}
|
||||
|
||||
export interface PgForkWorkReserve {
|
||||
minHarnessSlotCost: number;
|
||||
forkWorkReserve: number;
|
||||
floorSlotCost: number;
|
||||
}
|
||||
|
||||
export interface PgHarnessConnectionBudget {
|
||||
poolMax: number;
|
||||
adminMax: number;
|
||||
migrationSlots: number;
|
||||
totalSlots: number;
|
||||
floored: boolean;
|
||||
}
|
||||
|
||||
export type PgBudgetDegradation =
|
||||
| "capacity-unreadable"
|
||||
| "bootstrap-gate-unavailable"
|
||||
| "bootstrap-connect-failed";
|
||||
|
||||
export class PgConnectionBudgetConcurrencyError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "PgConnectionBudgetConcurrencyError";
|
||||
}
|
||||
}
|
||||
|
||||
export const MAX_LIVE_HARNESSES_PER_PARTICIPANT = 2;
|
||||
export const MIGRATION_SLOTS_PER_HARNESS = 1;
|
||||
export const TEMPLATE_BUILD_SLOTS = 3;
|
||||
export const FOREIGN_RESERVE = 8;
|
||||
export const BOOTSTRAP_RESERVE = 4;
|
||||
export const TOKEN_STALE_MS = 30_000;
|
||||
export const IDLE_LINGER_MS = 250;
|
||||
export const BOOTSTRAP_DEBURST_WINDOW_MS = 2_000;
|
||||
/** This is intentionally a minimum supported cluster capacity, never a typical value. */
|
||||
export const DEGRADED_MIN_MAX_CONNECTIONS = 20;
|
||||
/** This deliberately over-estimates reserved superuser capacity for conservative fallback. */
|
||||
export const DEGRADED_MAX_SUPERUSER_RESERVED = 5;
|
||||
|
||||
const LEASE_CLASS_ID = 1_913_101;
|
||||
const TOKEN_DIRECTORY = join(tmpdir(), "fusion-pg-connection-budget");
|
||||
let moduleInstanceSequence = 0;
|
||||
|
||||
/*
|
||||
FNXC:PgTestConnectionBudget 2026-08-17-00:41:
|
||||
FN-9131 fixes the 27-worker / 100-backend PostgreSQL harness failure by making
|
||||
admission cluster-shared instead of dividing a lane-local worker count. A live
|
||||
harness consumes its pool ceiling plus its dedicated migration pool and admin
|
||||
pool; template construction needs three funded backends before it takes its
|
||||
blocking golden-template lock.
|
||||
|
||||
The bootstrap session is the one unavoidable pre-slot connection: it reads
|
||||
cluster settings then becomes the lease session. Fixed-name payload-free mkdir
|
||||
tokens shape that burst on one host; stale entries are reclaimed after
|
||||
TOKEN_STALE_MS and all correctness still comes from close-on-failure plus the
|
||||
server advisory-lock registry. If the token directory is unavailable, a
|
||||
participant uses an identity-derived delay and retries connects, never a local
|
||||
token presented as a cross-process bound. Capacity-read failure uses a low
|
||||
floor and high reserved value, so degraded lock ranges are subsets of healthy
|
||||
ranges and mixed participants can under-issue but cannot over-issue.
|
||||
|
||||
A participant is a loaded module instance, not a PID. It claims lease and all
|
||||
work reserve once per active window, then lends slots locally without waiting.
|
||||
Template slots are acquired before a golden lock; acquiring after that lock
|
||||
could deadlock a holder behind its own live harness. This namespace is fixed
|
||||
and distinct from the harness hashtext locks and FN-9130 DDL admission.
|
||||
|
||||
FNXC:PgTestConnectionBudget 2026-08-17-01:36:
|
||||
At P=27 against 12 participants, registry over-subscription is normal rather
|
||||
than an exceptional test failure. A participant now retains its lease while it
|
||||
waits for an all-or-nothing work reserve; this avoids re-bootstrap and lease
|
||||
band churn, while a participant without a lease holds no backend or token.
|
||||
Only a local request exceeding MAX_LIVE_HARNESSES_PER_PARTICIPANT or the held
|
||||
reserve may raise PgConnectionBudgetConcurrencyError.
|
||||
|
||||
FNXC:PgTestConnectionBudget 2026-08-17-01:55:
|
||||
A healthy exhausted bootstrap-token band is ordinary queue pressure, not the
|
||||
R11 gate-unavailable condition: it must wait without opening a bootstrap
|
||||
backend. The first-window promise is shared by concurrent callers in one
|
||||
loaded module instance, so one participant cannot create or leak multiple
|
||||
lease sessions while its reserve is being acquired.
|
||||
*/
|
||||
|
||||
export function derivePgForkWorkReserve(input: {
|
||||
maxLiveHarnesses: number;
|
||||
flooredPoolMax: number;
|
||||
migrationSlots: number;
|
||||
flooredAdminMax: number;
|
||||
templateBuildSlots: number;
|
||||
}): PgForkWorkReserve {
|
||||
const minHarnessSlotCost = input.flooredPoolMax + input.migrationSlots + input.flooredAdminMax;
|
||||
const forkWorkReserve = Math.max(
|
||||
input.maxLiveHarnesses * minHarnessSlotCost,
|
||||
input.templateBuildSlots + minHarnessSlotCost,
|
||||
);
|
||||
return { minHarnessSlotCost, forkWorkReserve, floorSlotCost: 1 + forkWorkReserve };
|
||||
}
|
||||
|
||||
export const PG_FORK_WORK_RESERVE = derivePgForkWorkReserve({
|
||||
maxLiveHarnesses: MAX_LIVE_HARNESSES_PER_PARTICIPANT,
|
||||
flooredPoolMax: 1,
|
||||
migrationSlots: MIGRATION_SLOTS_PER_HARNESS,
|
||||
flooredAdminMax: 1,
|
||||
templateBuildSlots: TEMPLATE_BUILD_SLOTS,
|
||||
});
|
||||
|
||||
export function derivePgSlotSpace(input: {
|
||||
maxConnections: number;
|
||||
superuserReserved: number;
|
||||
foreignReserve: number;
|
||||
bootstrapReserve: number;
|
||||
floorSlotCost: number;
|
||||
}): PgSlotSpace {
|
||||
const slotCount = Math.max(
|
||||
input.floorSlotCost,
|
||||
Math.floor(input.maxConnections) - Math.floor(input.superuserReserved) - input.foreignReserve - input.bootstrapReserve,
|
||||
);
|
||||
const maxParticipants = Math.floor(slotCount / input.floorSlotCost);
|
||||
return {
|
||||
slotCount,
|
||||
maxParticipants,
|
||||
leaseBand: [0, maxParticipants],
|
||||
workBand: [maxParticipants, slotCount],
|
||||
};
|
||||
}
|
||||
|
||||
export function deriveDegradedPgSlotSpace(): PgSlotSpace {
|
||||
return derivePgSlotSpace({
|
||||
maxConnections: DEGRADED_MIN_MAX_CONNECTIONS,
|
||||
superuserReserved: DEGRADED_MAX_SUPERUSER_RESERVED,
|
||||
foreignReserve: FOREIGN_RESERVE,
|
||||
bootstrapReserve: BOOTSTRAP_RESERVE,
|
||||
floorSlotCost: PG_FORK_WORK_RESERVE.floorSlotCost,
|
||||
});
|
||||
}
|
||||
|
||||
export function derivePgConnectionBudget(input: {
|
||||
lentWorkSlots: number;
|
||||
liveHarnesses: number;
|
||||
}): PgHarnessConnectionBudget {
|
||||
if (input.liveHarnesses < 1 || input.lentWorkSlots < 3) {
|
||||
throw new PgConnectionBudgetConcurrencyError("FORK_WORK_RESERVE cannot fund the minimum three-slot harness");
|
||||
}
|
||||
// Reserve one migration and one admin backend; any remaining funded capacity is runtime pool.
|
||||
const poolMax = Math.max(1, Math.min(5, input.lentWorkSlots - 2));
|
||||
const adminMax = 1;
|
||||
const migrationSlots = 1;
|
||||
const totalSlots = poolMax + adminMax + migrationSlots;
|
||||
return { poolMax, adminMax, migrationSlots, totalSlots, floored: totalSlots === 3 };
|
||||
}
|
||||
|
||||
export function expectedPgAdmissionWaitMs(input: {
|
||||
participants: number;
|
||||
maxParticipants: number;
|
||||
p95WindowMs: number;
|
||||
lingerMs: number;
|
||||
deburstMs: number;
|
||||
}): number {
|
||||
if (input.maxParticipants < 1) return Number.POSITIVE_INFINITY;
|
||||
return Math.ceil(input.participants / input.maxParticipants) * (input.p95WindowMs + input.lingerMs) + input.deburstMs;
|
||||
}
|
||||
|
||||
export interface PgConnectionBudgetObservation {
|
||||
slotCount?: number;
|
||||
maxParticipants?: number;
|
||||
forkWorkReserve: number;
|
||||
minHarnessSlotCost: number;
|
||||
derivationMode?: "healthy" | "degraded-floor";
|
||||
leaseHeld: boolean;
|
||||
heldWorkSlots: number;
|
||||
lentWorkSlots: number;
|
||||
templateSlotsHeld: number;
|
||||
liveHarnesses: number;
|
||||
degradedCount: Readonly<Record<PgBudgetDegradation, number>>;
|
||||
bootstrapAttempts: number;
|
||||
bootstrapRetries: number;
|
||||
bootstrapTokenWaits: number;
|
||||
bootstrapTokenReclaims: number;
|
||||
deburstDelayMs: number;
|
||||
maxBootstrapWaitMs: number;
|
||||
floorAdmissionWaits: number;
|
||||
maxFloorWaitMs: number;
|
||||
concurrencyRejections: number;
|
||||
}
|
||||
|
||||
interface Token { release(): Promise<void>; }
|
||||
|
||||
type TokenAttempt =
|
||||
| { kind: "acquired"; token: Token }
|
||||
| { kind: "exhausted" }
|
||||
| { kind: "unavailable" };
|
||||
|
||||
async function tryAcquireToken(index: number): Promise<TokenAttempt> {
|
||||
const path = join(TOKEN_DIRECTORY, `token-${index}`);
|
||||
try {
|
||||
await mkdir(TOKEN_DIRECTORY, { recursive: true });
|
||||
} catch {
|
||||
return { kind: "unavailable" };
|
||||
}
|
||||
|
||||
try {
|
||||
await mkdir(path);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "EEXIST") return { kind: "unavailable" };
|
||||
try {
|
||||
if (Date.now() - (await stat(path)).mtimeMs > TOKEN_STALE_MS) {
|
||||
await rm(path, { recursive: true, force: true });
|
||||
await mkdir(path);
|
||||
tokenReclaims += 1;
|
||||
} else {
|
||||
// A healthy semaphore at capacity is queueing, not R11 degradation.
|
||||
return { kind: "exhausted" };
|
||||
}
|
||||
} catch (retryError) {
|
||||
if ((retryError as NodeJS.ErrnoException).code === "EEXIST") return { kind: "exhausted" };
|
||||
return { kind: "unavailable" };
|
||||
}
|
||||
}
|
||||
let released = false;
|
||||
return {
|
||||
kind: "acquired",
|
||||
token: { async release() { if (!released) { released = true; await rm(path, { recursive: true, force: true }); } } },
|
||||
};
|
||||
}
|
||||
|
||||
/** Test-only exact-name access; production admission only scans the fixed token band. */
|
||||
export async function __tryAcquirePgConnectionBudgetTokenForTest(index: number): Promise<TokenAttempt> {
|
||||
return tryAcquireToken(index);
|
||||
}
|
||||
|
||||
let tokenReclaims = 0;
|
||||
|
||||
class ConnectionBudget {
|
||||
readonly instance = ++moduleInstanceSequence;
|
||||
readonly degradedCount: Record<PgBudgetDegradation, number> = {
|
||||
"capacity-unreadable": 0,
|
||||
"bootstrap-gate-unavailable": 0,
|
||||
"bootstrap-connect-failed": 0,
|
||||
};
|
||||
readonly heldWork = new Set<number>();
|
||||
private lease?: Sql;
|
||||
private space?: PgSlotSpace;
|
||||
private derivationMode?: "healthy" | "degraded-floor";
|
||||
private reserveHeld = false;
|
||||
/** One module instance may own only one bootstrap/lease acquisition at a time. */
|
||||
private windowAcquisition?: Promise<void>;
|
||||
private liveHarnesses = 0;
|
||||
private lentWorkSlots = 0;
|
||||
private templateSlotsHeld = 0;
|
||||
private releaseTimer?: ReturnType<typeof setTimeout>;
|
||||
private contention = false;
|
||||
bootstrapAttempts = 0;
|
||||
bootstrapRetries = 0;
|
||||
bootstrapTokenWaits = 0;
|
||||
deburstDelayMs = 0;
|
||||
maxBootstrapWaitMs = 0;
|
||||
floorAdmissionWaits = 0;
|
||||
maxFloorWaitMs = 0;
|
||||
concurrencyRejections = 0;
|
||||
|
||||
private degrade(reason: PgBudgetDegradation): void {
|
||||
this.degradedCount[reason] += 1;
|
||||
if (this.degradedCount[reason] === 1) console.warn(`[pg-connection-budget] degraded=${reason}`);
|
||||
}
|
||||
|
||||
private async acquireLease(urlBase: string): Promise<"acquired" | "token-exhausted"> {
|
||||
const started = Date.now();
|
||||
let token: Token | undefined;
|
||||
let client: Sql | undefined;
|
||||
let retainedLease = false;
|
||||
try {
|
||||
let gateUnavailable = false;
|
||||
for (let index = 0; index < BOOTSTRAP_RESERVE; index += 1) {
|
||||
const attempt = await tryAcquireToken((this.instance + index) % BOOTSTRAP_RESERVE);
|
||||
if (attempt.kind === "acquired") { token = attempt.token; break; }
|
||||
if (attempt.kind === "unavailable") { gateUnavailable = true; break; }
|
||||
}
|
||||
if (!token && !gateUnavailable) {
|
||||
// The healthy host semaphore is saturated: do not turn this into an ungated connect.
|
||||
this.bootstrapTokenWaits += 1;
|
||||
return "token-exhausted";
|
||||
}
|
||||
if (!token) {
|
||||
this.degrade("bootstrap-gate-unavailable");
|
||||
this.deburstDelayMs = ((process.pid * 31 + this.instance * 17) >>> 0) % BOOTSTRAP_DEBURST_WINDOW_MS;
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, this.deburstDelayMs));
|
||||
}
|
||||
const maintenanceUrl = new URL(urlBase);
|
||||
maintenanceUrl.pathname = "/postgres";
|
||||
this.bootstrapAttempts += 1;
|
||||
client = postgres(maintenanceUrl.toString(), { max: 1, prepare: false, onnotice: () => {} });
|
||||
let maxConnections = DEGRADED_MIN_MAX_CONNECTIONS;
|
||||
let superuserReserved = DEGRADED_MAX_SUPERUSER_RESERVED;
|
||||
try {
|
||||
const rows = await client.unsafe<{ max_connections: string; superuser_reserved_connections: string }>(
|
||||
"SELECT current_setting('max_connections') AS max_connections, current_setting('superuser_reserved_connections') AS superuser_reserved_connections",
|
||||
);
|
||||
maxConnections = Number(rows[0]?.max_connections);
|
||||
superuserReserved = Number(rows[0]?.superuser_reserved_connections);
|
||||
if (!Number.isFinite(maxConnections) || !Number.isFinite(superuserReserved)) throw new Error("invalid capacity settings");
|
||||
this.derivationMode = "healthy";
|
||||
} catch {
|
||||
this.degrade("capacity-unreadable");
|
||||
this.derivationMode = "degraded-floor";
|
||||
}
|
||||
this.space = derivePgSlotSpace({ maxConnections, superuserReserved, foreignReserve: FOREIGN_RESERVE, bootstrapReserve: BOOTSTRAP_RESERVE, floorSlotCost: PG_FORK_WORK_RESERVE.floorSlotCost });
|
||||
for (let index = this.space.leaseBand[0]; index < this.space.leaseBand[1]; index += 1) {
|
||||
const rows = await client.unsafe<{ acquired: boolean }>(`SELECT pg_try_advisory_lock(${LEASE_CLASS_ID}, ${index}) AS acquired`);
|
||||
if (rows[0]?.acquired) {
|
||||
this.lease = client;
|
||||
retainedLease = true;
|
||||
return "acquired";
|
||||
}
|
||||
}
|
||||
return "token-exhausted";
|
||||
} catch (error) {
|
||||
// Connect and registry-query failures are real failures, not degradation aliases.
|
||||
this.degrade("bootstrap-connect-failed");
|
||||
throw error;
|
||||
} finally {
|
||||
if (!retainedLease) await client?.end({ timeout: 0 }).catch(() => {});
|
||||
await token?.release().catch(() => {});
|
||||
this.maxBootstrapWaitMs = Math.max(this.maxBootstrapWaitMs, Date.now() - started);
|
||||
}
|
||||
}
|
||||
|
||||
private async claimReserve(): Promise<boolean> {
|
||||
if (!this.lease || !this.space) return false;
|
||||
const started = Date.now();
|
||||
const claimed: number[] = [];
|
||||
for (let index = this.space.workBand[0]; index < this.space.workBand[1] && claimed.length < PG_FORK_WORK_RESERVE.forkWorkReserve; index += 1) {
|
||||
const rows = await this.lease.unsafe<{ acquired: boolean }>(`SELECT pg_try_advisory_lock(${LEASE_CLASS_ID}, ${index}) AS acquired`);
|
||||
if (rows[0]?.acquired) claimed.push(index);
|
||||
}
|
||||
if (claimed.length !== PG_FORK_WORK_RESERVE.forkWorkReserve) {
|
||||
for (const index of claimed) await this.lease.unsafe(`SELECT pg_advisory_unlock(${LEASE_CLASS_ID}, ${index})`);
|
||||
this.floorAdmissionWaits += 1;
|
||||
this.contention = true;
|
||||
this.maxFloorWaitMs = Math.max(this.maxFloorWaitMs, Date.now() - started);
|
||||
return false;
|
||||
}
|
||||
claimed.forEach((index) => this.heldWork.add(index));
|
||||
this.reserveHeld = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
async acquireWindow(urlBase: string): Promise<void> {
|
||||
if (this.releaseTimer) { clearTimeout(this.releaseTimer); this.releaseTimer = undefined; }
|
||||
if (this.reserveHeld) return;
|
||||
if (this.windowAcquisition) return this.windowAcquisition;
|
||||
const acquisition = this.acquireWindowInner(urlBase);
|
||||
this.windowAcquisition = acquisition;
|
||||
try {
|
||||
await acquisition;
|
||||
} finally {
|
||||
if (this.windowAcquisition === acquisition) this.windowAcquisition = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
private async acquireWindowInner(urlBase: string): Promise<void> {
|
||||
/*
|
||||
FNXC:PgTestConnectionBudget 2026-08-17-01:36:
|
||||
Registry contention queues here instead of consuming a harness test or hook
|
||||
with a bounded-attempt failure. Failed bootstrap attempts have already
|
||||
closed their client and released their token. Once a lease is held, R4's
|
||||
separate work band makes it safe to retain only that lease across partial
|
||||
reserve rollbacks; releasing it would let late arrivals thrash the lease
|
||||
band and multiply bootstrap connections.
|
||||
*/
|
||||
let attempt = 0;
|
||||
while (!this.reserveHeld) {
|
||||
if (!this.lease) {
|
||||
const leaseAttempt = await this.acquireLease(urlBase);
|
||||
if (leaseAttempt === "token-exhausted") {
|
||||
this.bootstrapRetries += 1;
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, Math.min(100, 5 * 2 ** Math.min(attempt, 4))));
|
||||
attempt += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (await this.claimReserve()) return;
|
||||
this.bootstrapRetries += 1;
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, Math.min(100, 5 * 2 ** Math.min(attempt, 4))));
|
||||
attempt += 1;
|
||||
}
|
||||
}
|
||||
|
||||
acquireHarness(): PgHarnessConnectionBudget {
|
||||
if (!this.reserveHeld || this.liveHarnesses >= MAX_LIVE_HARNESSES_PER_PARTICIPANT) {
|
||||
this.concurrencyRejections += 1;
|
||||
throw new PgConnectionBudgetConcurrencyError("MAX_LIVE_HARNESSES_PER_PARTICIPANT reserve cannot fund another harness");
|
||||
}
|
||||
const available = this.heldWork.size - this.lentWorkSlots - this.templateSlotsHeld;
|
||||
// Keep the minimum allotment for every allowed sibling harness. This makes
|
||||
// Promise.all creation fail fast only above the declared local limit.
|
||||
const siblingsStillAllowed = MAX_LIVE_HARNESSES_PER_PARTICIPANT - this.liveHarnesses - 1;
|
||||
const fundedForThisHarness = available - siblingsStillAllowed * PG_FORK_WORK_RESERVE.minHarnessSlotCost;
|
||||
const budget = derivePgConnectionBudget({ lentWorkSlots: fundedForThisHarness, liveHarnesses: this.liveHarnesses + 1 });
|
||||
this.lentWorkSlots += budget.totalSlots;
|
||||
this.liveHarnesses += 1;
|
||||
return budget;
|
||||
}
|
||||
|
||||
releaseHarness(budget: PgHarnessConnectionBudget): void {
|
||||
this.lentWorkSlots = Math.max(0, this.lentWorkSlots - budget.totalSlots);
|
||||
this.liveHarnesses = Math.max(0, this.liveHarnesses - 1);
|
||||
if (this.liveHarnesses === 0) {
|
||||
const linger = this.contention ? 0 : IDLE_LINGER_MS;
|
||||
this.releaseTimer = setTimeout(() => { void this.closeWindow(); }, linger);
|
||||
}
|
||||
}
|
||||
|
||||
acquireTemplate(): () => void { return this.acquireLocal(TEMPLATE_BUILD_SLOTS, "TEMPLATE_BUILD_SLOTS must be funded before the golden advisory lock"); }
|
||||
|
||||
acquireMaintenance(): () => void { return this.acquireLocal(1, "the funded reserve cannot open a maintenance client"); }
|
||||
|
||||
acquireCharge(cost: number): () => void {
|
||||
return this.acquireLocal(cost, "requested connection charge exceeds the held reserve");
|
||||
}
|
||||
|
||||
private acquireLocal(cost: number, message: string): () => void {
|
||||
if (!this.reserveHeld || this.heldWork.size - this.lentWorkSlots - this.templateSlotsHeld < cost) {
|
||||
this.concurrencyRejections += 1;
|
||||
throw new PgConnectionBudgetConcurrencyError(message);
|
||||
}
|
||||
this.templateSlotsHeld += cost;
|
||||
return () => { this.templateSlotsHeld = Math.max(0, this.templateSlotsHeld - cost); };
|
||||
}
|
||||
|
||||
async closeWindow(): Promise<void> {
|
||||
if (this.releaseTimer) { clearTimeout(this.releaseTimer); this.releaseTimer = undefined; }
|
||||
if (!this.lease) return;
|
||||
for (const slot of this.heldWork) await this.lease.unsafe(`SELECT pg_advisory_unlock(${LEASE_CLASS_ID}, ${slot})`).catch(() => {});
|
||||
this.heldWork.clear();
|
||||
await this.lease.end({ timeout: 0 }).catch(() => {});
|
||||
this.lease = undefined;
|
||||
this.space = undefined;
|
||||
this.reserveHeld = false;
|
||||
this.lentWorkSlots = 0;
|
||||
}
|
||||
|
||||
observe(): PgConnectionBudgetObservation {
|
||||
return {
|
||||
slotCount: this.space?.slotCount, maxParticipants: this.space?.maxParticipants,
|
||||
forkWorkReserve: PG_FORK_WORK_RESERVE.forkWorkReserve, minHarnessSlotCost: PG_FORK_WORK_RESERVE.minHarnessSlotCost,
|
||||
derivationMode: this.derivationMode, leaseHeld: Boolean(this.lease), heldWorkSlots: this.heldWork.size,
|
||||
lentWorkSlots: this.lentWorkSlots, templateSlotsHeld: this.templateSlotsHeld, liveHarnesses: this.liveHarnesses,
|
||||
degradedCount: { ...this.degradedCount }, bootstrapAttempts: this.bootstrapAttempts, bootstrapRetries: this.bootstrapRetries,
|
||||
bootstrapTokenWaits: this.bootstrapTokenWaits, bootstrapTokenReclaims: tokenReclaims, deburstDelayMs: this.deburstDelayMs,
|
||||
maxBootstrapWaitMs: this.maxBootstrapWaitMs, floorAdmissionWaits: this.floorAdmissionWaits,
|
||||
maxFloorWaitMs: this.maxFloorWaitMs, concurrencyRejections: this.concurrencyRejections,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const budget = new ConnectionBudget();
|
||||
|
||||
export async function resolvePgConnectionBudget(input: { available: boolean; urlBase: string }): Promise<void> {
|
||||
if (!input.available || process.env.FUSION_PG_TEST_SKIP === "1") return;
|
||||
await budget.acquireWindow(input.urlBase);
|
||||
}
|
||||
|
||||
export async function acquirePgHarnessConnectionBudget(input: { available: boolean; urlBase: string }): Promise<PgHarnessConnectionBudget> {
|
||||
await resolvePgConnectionBudget(input);
|
||||
return budget.acquireHarness();
|
||||
}
|
||||
|
||||
export function releasePgHarnessConnectionBudget(allotment: PgHarnessConnectionBudget): void { budget.releaseHarness(allotment); }
|
||||
export function acquirePgTemplateBuildAllotment(): () => void { return budget.acquireTemplate(); }
|
||||
/** Maintenance callers may only borrow a funded reserve; they never wait or touch registry locks. */
|
||||
export function acquirePgMaintenanceAllotment(): () => void { return budget.acquireMaintenance(); }
|
||||
export function observePgConnectionBudget(): PgConnectionBudgetObservation { return budget.observe(); }
|
||||
export async function __resetPgConnectionBudgetForTest(): Promise<void> { await budget.closeWindow(); }
|
||||
export async function withConnectionCharge<T>(cost: number, fn: () => Promise<T>): Promise<T> {
|
||||
// Transient maintenance work is charged at its requested ceiling, not as a
|
||||
// fictitious second harness that would consume the local concurrency limit.
|
||||
const release = budget.acquireCharge(cost);
|
||||
try { return await fn(); } finally { release(); }
|
||||
}
|
||||
@@ -843,6 +843,14 @@ export async function createEmptyPgTestDatabase(prefix = "fusion_test"): Promise
|
||||
* high-core machines. Use it for shared-harness files that create a single
|
||||
* database and do not exercise the per-module template lifecycle hooks.
|
||||
*/
|
||||
/*
|
||||
FNXC:PgTestHarnessConnectionBudget 2026-08-17-02:22:
|
||||
FN-9131 leaves the experimental PostgreSQL connection budget deliberately
|
||||
unwired because loaded-lane trials regressed broadly. This harness neither
|
||||
admits a budget window nor clamps caller poolMax; a successor must prove a
|
||||
lifecycle boundary that covers only PostgreSQL participants before wiring the
|
||||
characterization primitive in pg-connection-budget.ts.
|
||||
*/
|
||||
export async function createTaskStoreForTest(options?: {
|
||||
readonly poolMax?: number;
|
||||
readonly prefix?: string;
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import {
|
||||
__resetPgConnectionBudgetForTest,
|
||||
observePgConnectionBudget,
|
||||
resolvePgConnectionBudget,
|
||||
} from "../../__test-utils__/pg-connection-budget.js";
|
||||
import {
|
||||
PG_AVAILABLE,
|
||||
PG_TEST_URL_BASE,
|
||||
pgDescribe,
|
||||
} from "../../__test-utils__/pg-test-harness.js";
|
||||
|
||||
/**
|
||||
* FNXC:PgTestConnectionBudget 2026-08-17-01:36:
|
||||
* The primitive is deliberately unwired after loaded measurements showed that
|
||||
* harness-side admission regressed broad PostgreSQL suites. Keep a focused
|
||||
* server-backed check of its advisory-lock allocation while successor work
|
||||
* finds a setup-safe place to apply aggregate admission.
|
||||
*/
|
||||
const describeWhenPg = PG_AVAILABLE ? pgDescribe : pgDescribe.skip;
|
||||
|
||||
describeWhenPg("PostgreSQL connection-budget primitive", () => {
|
||||
afterEach(async () => {
|
||||
await __resetPgConnectionBudgetForTest();
|
||||
});
|
||||
|
||||
it("allocates a closed advisory-lock reserve against the reachable cluster", async () => {
|
||||
await resolvePgConnectionBudget({ available: PG_AVAILABLE, urlBase: PG_TEST_URL_BASE });
|
||||
|
||||
const observation = observePgConnectionBudget();
|
||||
expect(observation.slotCount).toBeGreaterThanOrEqual(observation.forkWorkReserve + 1);
|
||||
expect(observation.heldWorkSlots).toBe(observation.forkWorkReserve);
|
||||
expect(observation.leaseHeld).toBe(true);
|
||||
expect(observation.degradedCount["capacity-unreadable"]).toBe(0);
|
||||
}, 15_000);
|
||||
});
|
||||
Reference in New Issue
Block a user