From 7ff4bf1fde1e45a59281b9aac5940f3c67c55b5e Mon Sep 17 00:00:00 2001 From: Juan Cruz Fortunatti Date: Fri, 4 Sep 2026 05:04:35 +0200 Subject: [PATCH 1/4] fix: diagnose organization handoff authority failures instead of swallowing them --- ...ganizationHandoffAuthorityFsBudget.test.ts | 146 +++++++++++ .../organizationHandoffAuthorityFsBudget.ts | 204 +++++++++++++++ .../organizationHandoffAuthorityFsClient.ts | 100 ++++++-- .../organizationHandoffAuthorityFsWorker.ts | 236 ++++++++++-------- .../organizationHandoffAuthorityStore.ts | 19 +- .../organizationHandoffAuthorityTypes.ts | 3 +- 6 files changed, 584 insertions(+), 124 deletions(-) create mode 100644 src/deployment/organizationHandoffAuthorityFsBudget.test.ts create mode 100644 src/deployment/organizationHandoffAuthorityFsBudget.ts diff --git a/src/deployment/organizationHandoffAuthorityFsBudget.test.ts b/src/deployment/organizationHandoffAuthorityFsBudget.test.ts new file mode 100644 index 00000000..ad43621e --- /dev/null +++ b/src/deployment/organizationHandoffAuthorityFsBudget.test.ts @@ -0,0 +1,146 @@ +import { describe, expect, it } from "vitest"; + +import { + createOrganizationHandoffAuthorityError, + describeOrganizationHandoffAuthorityFailure, + OrganizationHandoffAuthorityBudget, OrganizationHandoffAuthorityFailure, + ORGANIZATION_HANDOFF_AUTHORITY_ERROR, + parseOrganizationHandoffAuthorityFailureDetail, + PUBLICATION_BUDGET_MS, PUBLICATION_MAX_ATTEMPTS, PUBLICATION_PATIENCE_FRACTION, + toOrganizationHandoffAuthorityFailureDetail +} from "./organizationHandoffAuthorityFsBudget.js"; + +/** A controllable clock so budget behaviour is asserted, never timed. */ +const clock = (): { advance(ms: number): void; now(): number } => { + let value = 0; + return { advance: (ms) => { value += ms; }, now: () => value }; +}; + +describe("organization handoff authority budget", () => { + it("bounds convergence by wall clock rather than attempt count", async () => { + const time = clock(); + // Each attempt costs a tenth of the budget in real time, whatever the + // nominal backoff was: this is the loaded-machine case. + const budget = new OrganizationHandoffAuthorityBudget({ + limitMs: 1_000, now: time.now, sleep: async () => { time.advance(100); } + }); + let attempts = 0; + while (!budget.exhausted()) { await budget.wait(); attempts += 1; } + expect(attempts).toBe(10); + expect(budget.elapsedMs).toBe(1_000); + }); + + it("spends the same deadline over far more attempts when attempts are cheap", async () => { + const time = clock(); + const budget = new OrganizationHandoffAuthorityBudget({ + limitMs: 1_000, now: time.now, sleep: async () => { time.advance(1); } + }); + let attempts = 0; + while (!budget.exhausted()) { await budget.wait(); attempts += 1; } + // The attempt-counted predecessor would have given up after a fixed 64 + // attempts in both cases; the deadline is what must stay constant. + expect(attempts).toBe(1_000); + expect(budget.elapsedMs).toBe(1_000); + }); + + it("keeps the attempt cap as a secondary bound against a stalled clock", async () => { + const time = clock(); + const budget = new OrganizationHandoffAuthorityBudget({ + limitMs: 1_000, maxAttempts: 5, now: time.now, sleep: async () => undefined + }); + let attempts = 0; + while (!budget.exhausted()) { await budget.wait(); attempts += 1; } + expect(attempts).toBe(5); + expect(budget.elapsedMs).toBe(0); + }); + + it("expires patience strictly before the budget so recovery can still run", () => { + const time = clock(); + const budget = new OrganizationHandoffAuthorityBudget({ limitMs: 1_000, now: time.now }); + expect(budget.patient()).toBe(true); + time.advance(1_000 * PUBLICATION_PATIENCE_FRACTION); + expect(budget.patient()).toBe(false); + expect(budget.exhausted()).toBe(false); + time.advance(1_000); + expect(budget.exhausted()).toBe(true); + expect(budget.patient()).toBe(false); + }); + + it("grows and jitters the backoff without ever waiting past the deadline", async () => { + const time = clock(); + const delays: number[] = []; + const budget = new OrganizationHandoffAuthorityBudget({ + limitMs: 100, now: time.now, sleep: async (ms) => { delays.push(ms); time.advance(ms); } + }); + for (let index = 0; index < 40 && !budget.exhausted(); index += 1) await budget.wait(); + expect(delays.length).toBeGreaterThan(1); + // Jittered, so never assert exact values: assert the envelope instead. + for (const delay of delays) expect(delay).toBeGreaterThanOrEqual(0); + expect(Math.max(...delays)).toBeLessThanOrEqual(25); + expect(delays.slice(-1)[0]).toBeGreaterThan(delays[0] as number); + expect(time.now()).toBeLessThanOrEqual(100); + }); + + it("ships defaults that fit inside one client request", () => { + expect(PUBLICATION_BUDGET_MS).toBeGreaterThan(0); + expect(PUBLICATION_MAX_ATTEMPTS).toBeGreaterThan(PUBLICATION_BUDGET_MS / 25); + expect(PUBLICATION_PATIENCE_FRACTION).toBeGreaterThan(0); + expect(PUBLICATION_PATIENCE_FRACTION).toBeLessThan(1); + }); +}); + +describe("organization handoff authority failure detail", () => { + it("keeps the uniform message and appends the diagnosis", () => { + const time = clock(); + const budget = new OrganizationHandoffAuthorityBudget({ limitMs: 500, now: time.now }); + time.advance(500); + const error = createOrganizationHandoffAuthorityError( + budget.snapshot("settle_budget_exhausted", "settle", "final:absent pending:nlink=1,size=10,mode=600")); + expect(error.message).toContain(ORGANIZATION_HANDOFF_AUTHORITY_ERROR); + expect(error.message).toContain("settle_budget_exhausted"); + expect(error.message).toContain("budget=settle"); + expect(error.message).toContain("attempts=0"); + expect(error.message).toContain("elapsed=500ms"); + expect(error.message).toContain("limit=500ms"); + expect(error.message).toContain("pending:nlink=1"); + }); + + it("produces the exact historical message when there is nothing to add", () => { + expect(createOrganizationHandoffAuthorityError().message).toBe(ORGANIZATION_HANDOFF_AUTHORITY_ERROR); + }); + + it("renders a bare code with no field list", () => { + expect(describeOrganizationHandoffAuthorityFailure({ code: "not_ready" })).toBe("not_ready"); + }); + + it("narrows and clamps an untrusted IPC payload", () => { + expect(parseOrganizationHandoffAuthorityFailureDetail(undefined)).toBeUndefined(); + expect(parseOrganizationHandoffAuthorityFailureDetail({ code: "" })).toBeUndefined(); + expect(parseOrganizationHandoffAuthorityFailureDetail(["code"])).toBeUndefined(); + const parsed = parseOrganizationHandoffAuthorityFailureDetail({ + attempts: 3, budget: "settle", code: "x".repeat(200), elapsedMs: 12.5, + extra: "dropped", limitMs: Number.NaN, state: "y".repeat(1_000) + }); + expect(parsed?.code).toHaveLength(64); + expect(parsed?.state).toHaveLength(240); + expect(parsed?.attempts).toBe(3); + expect(parsed?.elapsedMs).toBe(12.5); + expect(parsed?.limitMs).toBeUndefined(); + expect(parsed).not.toHaveProperty("extra"); + }); + + it("reduces an arbitrary thrown value to a bounded detail", () => { + const failure = new OrganizationHandoffAuthorityFailure({ code: "settle_budget_exhausted" }); + expect(toOrganizationHandoffAuthorityFailureDetail(failure)).toBe(failure.detail); + expect(toOrganizationHandoffAuthorityFailureDetail(new TypeError("boom"))) + .toEqual({ code: "unexpected_error", state: "TypeError" }); + expect(toOrganizationHandoffAuthorityFailureDetail("boom")) + .toEqual({ code: "unexpected_error", state: "string" }); + }); + + it("never carries record bytes into the diagnostic", () => { + const secret = "s".repeat(30_000); + const detail = toOrganizationHandoffAuthorityFailureDetail(new Error(secret)); + expect(describeOrganizationHandoffAuthorityFailure(detail)).not.toContain("ssss"); + }); +}); diff --git a/src/deployment/organizationHandoffAuthorityFsBudget.ts b/src/deployment/organizationHandoffAuthorityFsBudget.ts new file mode 100644 index 00000000..6c637e2f --- /dev/null +++ b/src/deployment/organizationHandoffAuthorityFsBudget.ts @@ -0,0 +1,204 @@ +/** + * Diagnostics and convergence budget for the organization handoff filesystem + * authority. + * + * This module is deliberately dependency-free: the authority worker is forked + * as a standalone process whose only other imports are `node:fs`, and that + * minimal surface is part of its threat model. + * + * ## Why a wall-clock budget + * + * Concurrent publishers converge by waiting for a peer to finish writing and + * linking a record. That is a wall-clock event — it completes when the peer is + * scheduled and its I/O lands. An attempt-counted budget answers a different + * question ("have I looked N times?") whose relationship to elapsed time is set + * by machine load, so the effective patience of the loop drifts with the very + * contention it exists to absorb. The budget below is therefore primarily a + * deadline; the attempt cap is only a secondary guard against a pathological + * hot loop if a timer or clock misbehaves. + * + * ## Why one budget per request + * + * The convergence loops nest (write -> settle -> read-staging -> read). Giving + * each loop its own budget bounds no level usefully: each individual loop is + * too impatient while their product is effectively unbounded. A single budget + * threaded through one request bounds the thing that actually has a deadline — + * the request — and is what the client's own request deadline must exceed. + */ + +/** Uniform public failure message. Detail is appended, never substituted. */ +export const ORGANIZATION_HANDOFF_AUTHORITY_ERROR = "Organization handoff authority failed"; + +/** Wall-clock convergence budget for one authority request. */ +export const PUBLICATION_BUDGET_MS = 2_000; +/** + * Secondary bound only. It is far above the attempt count any healthy + * convergence needs; it exists so a broken timer cannot spin forever. + */ +export const PUBLICATION_MAX_ATTEMPTS = 4_096; +/** + * Fraction of the budget spent waiting for a peer before a proven, incomplete + * staging prefix is retired instead. Patience must expire strictly before the + * budget does, or the recovery path could never run. + */ +export const PUBLICATION_PATIENCE_FRACTION = 0.5; + +const BASE_WAIT_MS = 1; +const MAX_WAIT_MS = 25; +const MAX_CODE_LENGTH = 64; +const MAX_STATE_LENGTH = 240; + +/** + * Structural diagnostics for a failed authority request. + * + * `state` describes the *shape* of the contending filesystem state — sidecar + * presence, link counts, sizes. It never carries record bytes, leaf names, or + * paths: this is the secret-publication path, and the diagnostic must not + * become a disclosure channel. + */ +export interface OrganizationHandoffAuthorityFailureDetail { + readonly attempts?: number; + readonly budget?: string; + readonly code: string; + readonly elapsedMs?: number; + readonly limitMs?: number; + readonly state?: string; +} + +const clamp = (value: string, limit: number): string => + value.length > limit ? `${value.slice(0, limit - 1)}~` : value; + +const isSafeNumber = (value: unknown): value is number => + typeof value === "number" && Number.isFinite(value); + +/** Render a detail as the suffix appended to the uniform failure message. */ +export const describeOrganizationHandoffAuthorityFailure = ( + detail: OrganizationHandoffAuthorityFailureDetail +): string => { + const fields: string[] = []; + if (detail.budget !== undefined) fields.push(`budget=${detail.budget}`); + if (detail.attempts !== undefined) fields.push(`attempts=${detail.attempts}`); + if (detail.elapsedMs !== undefined) fields.push(`elapsed=${Math.round(detail.elapsedMs)}ms`); + if (detail.limitMs !== undefined) fields.push(`limit=${Math.round(detail.limitMs)}ms`); + if (detail.state !== undefined) fields.push(`state=${detail.state}`); + return fields.length === 0 ? detail.code : `${detail.code} (${fields.join(" ")})`; +}; + +/** An authority failure that carries its structural diagnostics. */ +export class OrganizationHandoffAuthorityFailure extends Error { + public readonly detail: OrganizationHandoffAuthorityFailureDetail; + public constructor(detail: OrganizationHandoffAuthorityFailureDetail) { + super(`${ORGANIZATION_HANDOFF_AUTHORITY_ERROR}: ${describeOrganizationHandoffAuthorityFailure(detail)}`); + this.name = "OrganizationHandoffAuthorityFailure"; + this.detail = detail; + } +} + +/** + * Build an error for the uniform message plus optional detail. Callers that + * have no diagnostics still produce the exact historical message, so existing + * substring expectations and the store's opaque public boundary hold. + */ +export const createOrganizationHandoffAuthorityError = ( + detail?: OrganizationHandoffAuthorityFailureDetail +): Error => detail === undefined + ? new Error(ORGANIZATION_HANDOFF_AUTHORITY_ERROR) + : new OrganizationHandoffAuthorityFailure(detail); + +/** Narrow an unknown IPC payload to a bounded, well-formed detail. */ +export const parseOrganizationHandoffAuthorityFailureDetail = ( + raw: unknown +): OrganizationHandoffAuthorityFailureDetail | undefined => { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined; + const value = raw as Record; + if (typeof value.code !== "string" || value.code.length === 0) return undefined; + return { + ...(isSafeNumber(value.attempts) ? { attempts: value.attempts } : {}), + ...(typeof value.budget === "string" ? { budget: clamp(value.budget, MAX_CODE_LENGTH) } : {}), + code: clamp(value.code, MAX_CODE_LENGTH), + ...(isSafeNumber(value.elapsedMs) ? { elapsedMs: value.elapsedMs } : {}), + ...(isSafeNumber(value.limitMs) ? { limitMs: value.limitMs } : {}), + ...(typeof value.state === "string" ? { state: clamp(value.state, MAX_STATE_LENGTH) } : {}) + }; +}; + +/** Reduce an unknown thrown value to a bounded detail suitable for IPC. */ +export const toOrganizationHandoffAuthorityFailureDetail = ( + error: unknown +): OrganizationHandoffAuthorityFailureDetail => + error instanceof OrganizationHandoffAuthorityFailure + ? error.detail + : { code: "unexpected_error", state: clamp( + error instanceof Error ? `${error.name}` : typeof error, MAX_STATE_LENGTH) }; + +export interface OrganizationHandoffAuthorityBudgetOptions { + readonly limitMs?: number; + readonly maxAttempts?: number; + readonly now?: () => number; + readonly sleep?: (ms: number) => Promise; +} + +/** + * One request's convergence budget: a wall-clock deadline, a secondary attempt + * cap, and adaptive backoff with jitter. + * + * The jitter matters as much as the deadline. Symmetric publishers waking on a + * fixed 2ms timer retry in lockstep, so every peer re-observes the same + * unfinished state and burns the budget in phase. Randomised, growing waits + * break that convoy so peers observe each other's progress instead. + */ +export class OrganizationHandoffAuthorityBudget { + #attempts = 0; + readonly #limitMs: number; + readonly #maxAttempts: number; + readonly #now: () => number; + readonly #sleep: (ms: number) => Promise; + readonly #startedAt: number; + + public constructor(options: OrganizationHandoffAuthorityBudgetOptions = {}) { + this.#limitMs = options.limitMs ?? PUBLICATION_BUDGET_MS; + this.#maxAttempts = options.maxAttempts ?? PUBLICATION_MAX_ATTEMPTS; + this.#now = options.now ?? (() => performance.now()); + this.#sleep = options.sleep ?? (async (ms) => new Promise((resolve) => { setTimeout(resolve, ms); })); + this.#startedAt = this.#now(); + } + + public get attempts(): number { return this.#attempts; } + public get elapsedMs(): number { return this.#now() - this.#startedAt; } + public get limitMs(): number { return this.#limitMs; } + + /** True once no further waiting is permitted for this request. */ + public exhausted(): boolean { + return this.elapsedMs >= this.#limitMs || this.#attempts >= this.#maxAttempts; + } + + /** + * True while a peer still deserves the benefit of the doubt. Once false, a + * proven incomplete staging prefix may be retired — the caller has already + * linked, or is about to link, the immutable final record that prevents that + * prefix from winning a later election. + */ + public patient(): boolean { + return !this.exhausted() && this.elapsedMs < this.#limitMs * PUBLICATION_PATIENCE_FRACTION; + } + + /** Snapshot the budget for a failure detail. */ + public snapshot( + code: string, budget: string, state?: string + ): OrganizationHandoffAuthorityFailureDetail { + return { + attempts: this.#attempts, budget, code, + elapsedMs: this.elapsedMs, limitMs: this.#limitMs, + ...(state === undefined ? {} : { state: clamp(state, MAX_STATE_LENGTH) }) + }; + } + + /** Back off before the next observation, never past the deadline. */ + public async wait(): Promise { + this.#attempts += 1; + const growth = Math.min(MAX_WAIT_MS, BASE_WAIT_MS * 2 ** Math.min(this.#attempts - 1, 10)); + const remaining = Math.max(0, this.#limitMs - this.elapsedMs); + const delay = Math.max(0, Math.min(remaining, growth * (0.5 + Math.random() * 0.5))); + await this.#sleep(delay); + } +} diff --git a/src/deployment/organizationHandoffAuthorityFsClient.ts b/src/deployment/organizationHandoffAuthorityFsClient.ts index 4ec98d2d..974fbdce 100644 --- a/src/deployment/organizationHandoffAuthorityFsClient.ts +++ b/src/deployment/organizationHandoffAuthorityFsClient.ts @@ -2,11 +2,33 @@ import { fork, type ChildProcess } from "node:child_process"; import { createRequire } from "node:module"; import { fileURLToPath } from "node:url"; +import { + createOrganizationHandoffAuthorityError, parseOrganizationHandoffAuthorityFailureDetail, + PUBLICATION_BUDGET_MS, type OrganizationHandoffAuthorityFailureDetail +} from "./organizationHandoffAuthorityFsBudget.js"; + const VERSION = "spawnfile.organization-handoff-fs-worker.v1"; const MAX_BYTES = 32_768; -const fail = (): never => { throw new Error("Organization handoff authority failed"); }; +/** + * The client deadline must exceed the worker's own convergence budget, or the + * client times out first and replaces a diagnosable worker failure with an + * opaque one. The margin covers fork/IPC latency and event-loop lag. + */ +const REQUEST_DEADLINE_MS = PUBLICATION_BUDGET_MS + 3_000; +/** + * Startup deadline for the worker's ready handshake. This bounds a hung or + * broken helper; it is not a latency budget for process startup. Several + * helpers are forked concurrently and a loaded host schedules their + * interpreter startup slowly, so it is sized for a saturated machine. + */ +const READY_DEADLINE_MS = 20_000; +const DISPOSE_GRACE_MS = 1_000; + +const fail = (code: string, state?: string): never => { + throw createOrganizationHandoffAuthorityError({ code, ...(state === undefined ? {} : { state }) }); +}; const bytes = (value: string): number => Buffer.byteLength(value, "utf8"); -const name = (value: unknown): string => typeof value === "string" && /^(?:[a-f0-9]{128}|[a-f0-9]{142})\.json$/u.test(value) ? value : fail(); +const name = (value: unknown): string => typeof value === "string" && /^(?:[a-f0-9]{128}|[a-f0-9]{142})\.json$/u.test(value) ? value : fail("invalid_name"); export interface OrganizationHandoffAuthorityFsClientOptions { readonly cwd: string; @@ -26,18 +48,23 @@ export interface OrganizationHandoffAuthorityFsClient { write(name: string, content: string): Promise; } +type Pending = { + reject(detail?: OrganizationHandoffAuthorityFailureDetail): void; + resolve(value: { content?: string; created?: boolean }): void; +}; + class Client implements OrganizationHandoffAuthorityFsClient { readonly #child: ChildProcess; #closed = false; #next = 1; readonly #exited: Promise; - readonly #pending = new Map(); + readonly #pending = new Map(); public constructor(child: ChildProcess) { this.#child = child; let settleExit!: () => void; this.#exited = new Promise((resolve) => { settleExit = resolve; }); - const settle = (): void => { this.rejectPending(); settleExit(); }; + const settle = (): void => { this.rejectPending("worker_exited"); settleExit(); }; child.on("message", (raw: unknown) => this.message(raw)); - child.on("error", () => this.rejectPending()); - child.on("disconnect", () => this.rejectPending()); + child.on("error", () => this.rejectPending("worker_errored")); + child.on("disconnect", () => this.rejectPending("worker_disconnected")); child.once("exit", settle); if (child.exitCode !== null || child.signalCode !== null) settle(); // `fork(..., { silent: true })` creates pipes. Consume them so a failing @@ -45,7 +72,10 @@ class Client implements OrganizationHandoffAuthorityFsClient { child.stdout?.resume(); child.stderr?.resume(); } private terminal(): boolean { return this.#child.exitCode !== null || this.#child.signalCode !== null; } - private rejectPending(): void { for (const pending of this.#pending.values()) pending.reject(); this.#pending.clear(); } + private rejectPending(code: string): void { + for (const pending of this.#pending.values()) pending.reject({ code }); + this.#pending.clear(); + } private message(raw: unknown): void { if (!raw || typeof raw !== "object") return; const value = raw as Record; if (value.version !== VERSION) return; @@ -53,13 +83,39 @@ class Client implements OrganizationHandoffAuthorityFsClient { if (!Number.isSafeInteger(value.id) || typeof value.ok !== "boolean") return; const pending = this.#pending.get(value.id as number); if (!pending) return; this.#pending.delete(value.id as number); if (value.ok !== true || value.content !== undefined && typeof value.content !== "string" - || value.created !== undefined && typeof value.created !== "boolean") pending.reject(); else pending.resolve(value as { content?: string; created?: boolean }); + || value.created !== undefined && typeof value.created !== "boolean") { + // The worker reports which budget or invariant failed. Preserve it: a + // bare failure here is what made this path undiagnosable from CI logs. + pending.reject(parseOrganizationHandoffAuthorityFailureDetail(value.failure) ?? { code: "worker_rejected" }); + return; + } + pending.resolve(value as { content?: string; created?: boolean }); } private request(op: "create" | "read" | "write", file: string, content?: string): Promise<{ content?: string; created?: boolean }> { - if (this.#closed || content !== undefined && bytes(content) > MAX_BYTES) return Promise.reject(new Error("Organization handoff authority failed")); + if (this.#closed) return Promise.reject(createOrganizationHandoffAuthorityError({ code: "client_closed" })); + if (content !== undefined && bytes(content) > MAX_BYTES) return Promise.reject(createOrganizationHandoffAuthorityError({ code: "content_too_large", state: `bytes=${bytes(content)} limit=${MAX_BYTES}` })); const id = this.#next++; const packet = { version: VERSION, id, op, name: name(file), ...(content === undefined ? {} : { content }) }; - if (bytes(JSON.stringify(packet)) > MAX_BYTES) return Promise.reject(new Error("Organization handoff authority failed")); - return new Promise((resolve, reject) => { const timer = setTimeout(() => { this.#pending.delete(id); reject(new Error("Organization handoff authority failed")); }, 3_000); this.#pending.set(id, { resolve: (value) => { clearTimeout(timer); resolve(value); }, reject: () => { clearTimeout(timer); reject(new Error("Organization handoff authority failed")); } }); if (!this.#child.send(packet)) { clearTimeout(timer); this.#pending.delete(id); reject(new Error("Organization handoff authority failed")); } }); + const size = bytes(JSON.stringify(packet)); + if (size > MAX_BYTES) return Promise.reject(createOrganizationHandoffAuthorityError({ code: "packet_too_large", state: `bytes=${size} limit=${MAX_BYTES}` })); + return new Promise((resolve, reject) => { + const startedAt = performance.now(); + const settle = (detail?: OrganizationHandoffAuthorityFailureDetail): void => { + clearTimeout(timer); + reject(createOrganizationHandoffAuthorityError(detail ?? { code: "request_failed" })); + }; + const timer = setTimeout(() => { + this.#pending.delete(id); + settle({ budget: "client_request", code: "client_request_deadline", elapsedMs: performance.now() - startedAt, limitMs: REQUEST_DEADLINE_MS, state: `op=${op}` }); + }, REQUEST_DEADLINE_MS); + this.#pending.set(id, { + reject: (detail) => settle(detail === undefined ? detail : { ...detail, state: `${detail.state === undefined ? "" : `${detail.state} `}op=${op}` }), + resolve: (value) => { clearTimeout(timer); resolve(value); } + }); + if (!this.#child.send(packet)) { + clearTimeout(timer); this.#pending.delete(id); + reject(createOrganizationHandoffAuthorityError({ code: "worker_send_failed", state: `op=${op}` })); + } + }); } public async create(file: string, content: string): Promise { const result = await this.request("create", file, content); @@ -68,14 +124,14 @@ class Client implements OrganizationHandoffAuthorityFsClient { public async read(file: string): Promise { return (await this.request("read", file)).content ?? null; } public async write(file: string, content: string): Promise { await this.request("write", file, content); } public async dispose(): Promise { - if (this.#closed) return; this.#closed = true; this.rejectPending(); + if (this.#closed) return; this.#closed = true; this.rejectPending("client_disposed"); if (!this.terminal()) { if (this.#child.connected) { try { this.#child.disconnect(); } catch { /* already disconnected */ } } this.#child.kill(); } let clear: (() => void) | undefined; const graceful = await Promise.race([this.#exited.then(() => true), new Promise((resolve) => { - const timer = setTimeout(() => resolve(false), 1_000); clear = () => clearTimeout(timer); + const timer = setTimeout(() => resolve(false), DISPOSE_GRACE_MS); clear = () => clearTimeout(timer); })]); clear?.(); if (!graceful && !this.terminal()) this.#child.kill("SIGKILL"); @@ -95,17 +151,27 @@ export const initializeOrganizationHandoffAuthorityFsClient = async (options: Or const child = fork(workerPath, [], { cwd: options.cwd, env: { SPAWNFILE_AUTHORITY_FS_ANCHOR: JSON.stringify({ dev: options.dev, ino: options.ino, uid: options.uid, parent_pid: options.parentPid ?? process.pid }) }, silent: true, ...(tsxLoader === undefined ? {} : { execArgv: ["--import", tsxLoader] }) }); const client = new Client(child); options.testOnChildStarted?.(child); + const startedAt = performance.now(); + let detail: OrganizationHandoffAuthorityFailureDetail = { code: "worker_ready_failed" }; try { await new Promise((resolve, reject) => { let timer: ReturnType; const message = (raw: unknown): void => { if (typeof raw === "object" && raw !== null && (raw as { ready?: unknown }).ready === true) doneResolve(); }; - const exited = (): void => doneReject(); const errored = (): void => doneReject(); + const exited = (): void => doneReject("worker_exited_before_ready"); + const errored = (): void => doneReject("worker_errored_before_ready"); const clean = (): void => { clearTimeout(timer); child.off("message", message); child.off("exit", exited); child.off("error", errored); }; function doneResolve(): void { clean(); resolve(); } - function doneReject(): void { clean(); reject(new Error("Organization handoff authority failed")); } - timer = setTimeout(doneReject, 3_000); + function doneReject(code: string): void { + clean(); + detail = { budget: "worker_ready", code, elapsedMs: performance.now() - startedAt, limitMs: READY_DEADLINE_MS, state: `source=${String(tsWorker)}` }; + reject(createOrganizationHandoffAuthorityError(detail)); + } + timer = setTimeout(() => doneReject("worker_ready_deadline"), READY_DEADLINE_MS); child.on("message", message); child.once("exit", exited); child.once("error", errored); }); return client; - } catch { await client.dispose(); return fail(); } + } catch { + await client.dispose(); + throw createOrganizationHandoffAuthorityError(detail); + } }; diff --git a/src/deployment/organizationHandoffAuthorityFsWorker.ts b/src/deployment/organizationHandoffAuthorityFsWorker.ts index f2fbe0c3..5e868b88 100644 --- a/src/deployment/organizationHandoffAuthorityFsWorker.ts +++ b/src/deployment/organizationHandoffAuthorityFsWorker.ts @@ -1,28 +1,37 @@ import { constants } from "node:fs"; import { link, lstat, open, unlink } from "node:fs/promises"; +import { + OrganizationHandoffAuthorityBudget, OrganizationHandoffAuthorityFailure, + toOrganizationHandoffAuthorityFailureDetail +} from "./organizationHandoffAuthorityFsBudget.js"; + const VERSION = "spawnfile.organization-handoff-fs-worker.v1"; const MAX_BYTES = 32_768; -const PUBLICATION_READ_ATTEMPTS = 64; -const PUBLICATION_SETTLE_ATTEMPTS = 64; // Store keys encode either a 64-character pending key or a 71-character // `opaque_` handoff handle. No other leaf namespace is reachable. const NAME = /^(?:[a-f0-9]{128}|[a-f0-9]{142})\.json$/u; const owner = typeof process.getuid === "function" ? process.getuid() : undefined; type Request = { readonly version: typeof VERSION; readonly id: number; readonly op: "create" | "read" | "write"; readonly name: string; readonly content?: string }; type Anchor = { readonly dev: number; readonly ino: number; readonly uid?: number; readonly parent_pid: number; }; -const fail = (): never => { throw new Error("Organization handoff authority failed"); }; +type Budget = OrganizationHandoffAuthorityBudget; +const fail = (code: string, state?: string): never => { + throw new OrganizationHandoffAuthorityFailure({ code, ...(state === undefined ? {} : { state }) }); +}; +/** Fail with the budget's attempt/wall-clock accounting and the contending shape. */ +const failBudget = (budget: Budget, code: string, loop: string, state?: string): never => { + throw new OrganizationHandoffAuthorityFailure(budget.snapshot(code, loop, state)); +}; const bytes = (value: string): number => Buffer.byteLength(value, "utf8"); -const waitForPublisher = async (): Promise => new Promise((resolve) => setTimeout(resolve, 2)); -const validName = (value: unknown): string => typeof value === "string" && NAME.test(value) ? value : fail(); +const validName = (value: unknown): string => typeof value === "string" && NAME.test(value) ? value : fail("invalid_name"); const validRequest = (raw: unknown): Request => { - if (!raw || typeof raw !== "object" || Array.isArray(raw) || Object.getPrototypeOf(raw) !== Object.prototype) return fail(); + if (!raw || typeof raw !== "object" || Array.isArray(raw) || Object.getPrototypeOf(raw) !== Object.prototype) return fail("malformed_request"); const value = raw as Record; - if (value.version !== VERSION || !Number.isSafeInteger(value.id) || typeof value.op !== "string" || (value.op !== "read" && value.op !== "write" && value.op !== "create")) return fail(); + if (value.version !== VERSION || !Number.isSafeInteger(value.id) || typeof value.op !== "string" || (value.op !== "read" && value.op !== "write" && value.op !== "create")) return fail("unsupported_request"); const content = value.content; const keys = Object.keys(value).sort(); const expected = value.op === "read" ? ["id", "name", "op", "version"] : ["content", "id", "name", "op", "version"]; - if (keys.length !== expected.length || keys.some((key, index) => key !== expected[index])) return fail(); - if (content !== undefined && (typeof content !== "string" || bytes(content) > MAX_BYTES)) return fail(); + if (keys.length !== expected.length || keys.some((key, index) => key !== expected[index])) return fail("unexpected_request_keys"); + if (content !== undefined && (typeof content !== "string" || bytes(content) > MAX_BYTES)) return fail("oversized_content"); return { version: VERSION, id: value.id as number, op: value.op, name: validName(value.name), ...(content === undefined ? {} : { content }) }; }; const send = (value: unknown): void => { @@ -30,24 +39,41 @@ const send = (value: unknown): void => { process.send(JSON.parse(serialized)); }; const statFile = async (name: string) => { - const stat = await lstat(name).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail()); + const stat = await lstat(name).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("lstat_failed")); if (stat === null) return null; if (!stat.isFile() || stat.isSymbolicLink() || stat.nlink > 2 || stat.size > MAX_BYTES || (stat.mode & 0o077) !== 0 - || owner !== undefined && stat.uid !== owner) return fail(); + || owner !== undefined && stat.uid !== owner) return fail("leaf_not_ordinary"); return stat; }; +const publicationSidecars = (name: string): readonly string[] => [`${name}.pending`, `${name}.recovery`]; +/** + * Structural description of the contending state for a diagnostic. Reports + * only shape — presence, link count, size, mode. Never record bytes, leaf + * names, or paths: this is the secret-publication path. + */ +const describeContention = async (name: string): Promise => { + const roles: readonly (readonly [string, string])[] = [["final", name], ["pending", `${name}.pending`], ["recovery", `${name}.recovery`]]; + const parts = await Promise.all(roles.map(async ([role, leaf]) => { + const stat = await lstat(leaf).catch(() => null); + return stat === null + ? `${role}:absent` + : `${role}:nlink=${stat.nlink},size=${stat.size},mode=${(stat.mode & 0o777).toString(8)}`; + })); + return parts.join(" "); +}; +const aliasesOf = (name: string): readonly string[] => [ + `${name}.pending`, `${name}.recovery`, + ...(name.endsWith(".pending") ? [name.slice(0, -".pending".length)] : []), + ...(name.endsWith(".recovery") ? [name.slice(0, -".recovery".length)] : []) +]; const read = async (name: string): Promise => { let before = await statFile(name); if (before === null) return null; if (before.nlink === 2) { - const aliases = [ - `${name}.pending`, `${name}.recovery`, - ...(name.endsWith(".pending") ? [name.slice(0, -".pending".length)] : []), - ...(name.endsWith(".recovery") ? [name.slice(0, -".recovery".length)] : []) - ]; let match: string | undefined; - for (const alias of aliases) { - const stat = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail()); + let match: string | undefined; + for (const alias of aliasesOf(name)) { + const stat = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("alias_lstat_failed")); if (stat && stat.isFile() && !stat.isSymbolicLink() && stat.nlink === 2 && stat.dev === before.dev && stat.ino === before.ino) { - if (match) return fail(); match = alias; + if (match) return fail("ambiguous_link_alias"); match = alias; } } if (!match) { @@ -55,17 +81,17 @@ const read = async (name: string): Promise => { // helper that won this link election. Re-read only if that left this // same checked file with its ordinary single link. const current = await statFile(name); if (current === null) return null; - if (current.nlink !== 1) return fail(); before = current; + if (current.nlink !== 1) return fail("unresolved_second_link"); before = current; } // Reading the canonical record completes a crashed publisher. An // intermediate-name reader may be racing that publisher, so it merely // observes the linked content and lets its normal publish path converge. if (match && !name.endsWith(".pending") && !name.endsWith(".recovery")) { - await unlink(match).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail(); }); - await sync(); before = await statFile(name); if (before === null) return fail(); + await unlink(match).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail("alias_unlink_failed"); }); + await sync(); before = await statFile(name); if (before === null) return fail("record_vanished_after_join"); } } - const handle = await open(name, constants.O_RDONLY | constants.O_NOFOLLOW).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail()); + const handle = await open(name, constants.O_RDONLY | constants.O_NOFOLLOW).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("open_failed")); if (handle === null) return null; try { const stat = await handle.stat(); @@ -73,36 +99,35 @@ const read = async (name: string): Promise => { // this descriptor is open, reducing nlink from two to one. That is the // sole permitted link-count transition; inode, device, size, and all // increases remain fail-closed. - if (!stat.isFile() || stat.nlink < 1 || stat.nlink > before.nlink || stat.dev !== before.dev || stat.ino !== before.ino || stat.size !== before.size) return fail(); + if (!stat.isFile() || stat.nlink < 1 || stat.nlink > before.nlink || stat.dev !== before.dev || stat.ino !== before.ino || stat.size !== before.size) return fail("record_drifted_while_open"); return await handle.readFile({ encoding: "utf8" }); - } catch { return fail(); } finally { await handle.close().catch(() => undefined); } + } catch (error) { + if (error instanceof OrganizationHandoffAuthorityFailure) throw error; + return fail("read_failed"); + } finally { await handle.close().catch(() => undefined); } }; -const sync = async (): Promise => { const handle = await open(".", constants.O_RDONLY | constants.O_DIRECTORY).catch(fail); try { await handle.sync(); } finally { await handle.close().catch(() => undefined); } }; +const sync = async (): Promise => { const handle = await open(".", constants.O_RDONLY | constants.O_DIRECTORY).catch(() => fail("directory_open_failed")); try { await handle.sync(); } finally { await handle.close().catch(() => undefined); } }; const expectedElectionState = async (name: string): Promise => { let stat = await statFile(name); if (stat === null) return null; if (stat.nlink === 1) return true; - const aliases = [ - `${name}.pending`, `${name}.recovery`, - ...(name.endsWith(".pending") ? [name.slice(0, -".pending".length)] : []), - ...(name.endsWith(".recovery") ? [name.slice(0, -".recovery".length)] : []) - ]; - for (const alias of aliases) { - const counterpart = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail()); + for (const alias of aliasesOf(name)) { + const counterpart = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("election_lstat_failed")); if (counterpart?.isFile() && !counterpart.isSymbolicLink() && counterpart.nlink === 2 && counterpart.dev === stat.dev && counterpart.ino === stat.ino) return true; } // The counterpart may have disappeared just before this check. Accept only // the resulting ordinary single-link state, never an unknown hard link. stat = await statFile(name); return stat?.nlink === 1; }; -const readDuringPublication = async (name: string, attempt = 0): Promise => { - try { return await read(name); } catch { +const readDuringPublication = async (name: string, budget: Budget): Promise => { + try { return await read(name); } catch (error) { // Re-validate the leaf before retrying. This permits only a checked, // ordinary one/two-link publication race; symlinks, mode/owner drift, // extra links, and other malformed states still fail immediately. const election = await expectedElectionState(name); if (election === null) return null; - if (attempt >= PUBLICATION_READ_ATTEMPTS || election !== true) return fail(); - await waitForPublisher(); return readDuringPublication(name, attempt + 1); + if (election !== true) throw error; + if (budget.exhausted()) return failBudget(budget, "read_budget_exhausted", "read", await describeContention(name)); + await budget.wait(); return readDuringPublication(name, budget); } }; type StagingState = "absent" | "exact" | "expected-prefix"; @@ -112,149 +137,160 @@ type StagingState = "absent" | "exact" | "expected-prefix"; * so cleanup still relies on a stable checked sidecar rather than an inference. */ const readStaging = async ( - staging: string, final: string, content: string, attempt = 0 + staging: string, final: string, content: string, budget: Budget ): Promise => { try { const observed = await read(staging); if (observed === null) return "absent"; if (observed === content) return "exact"; - return content.startsWith(observed) ? "expected-prefix" : fail(); - } catch { - const published = await readDuringPublication(final); + return content.startsWith(observed) ? "expected-prefix" : fail("staging_content_unrelated"); + } catch (error) { + if (error instanceof OrganizationHandoffAuthorityFailure && error.detail.budget !== undefined) throw error; + const published = await readDuringPublication(final, budget); if (published !== null) { - if (published !== content || attempt >= PUBLICATION_READ_ATTEMPTS) return fail(); - await waitForPublisher(); return readStaging(staging, final, content, attempt + 1); + if (published !== content) return fail("published_content_mismatch"); + if (budget.exhausted()) return failBudget(budget, "staging_budget_exhausted", "staging", await describeContention(final)); + await budget.wait(); return readStaging(staging, final, content, budget); } const election = await expectedElectionState(staging); if (election === null) return "absent"; - if (election !== true || attempt >= PUBLICATION_READ_ATTEMPTS) return fail(); - await waitForPublisher(); return readStaging(staging, final, content, attempt + 1); + if (election !== true) throw error; + if (budget.exhausted()) return failBudget(budget, "staging_budget_exhausted", "staging", await describeContention(final)); + await budget.wait(); return readStaging(staging, final, content, budget); } }; -const publicationSidecars = (name: string): readonly string[] => [`${name}.pending`, `${name}.recovery`]; /** * A successful link election can race a peer which had already created the * other staging leaf. The final immutable record is authoritative, but the * stale leaf must not survive a completed join: it would otherwise be * mistaken for an in-progress publication after restart. Delete exact bytes * immediately. An incomplete expected prefix may still belong to a publisher, - * so wait for it first; once that bounded wait expires, removing that proven - * prefix is safe because the final record already prevents it from winning a - * later link election. + * so wait for it first; once the budget's patience expires, removing that + * proven prefix is safe because the final record already prevents it from + * winning a later link election. */ -const settlePublished = async (name: string, content: string, attempt = 0): Promise => { - if (attempt > PUBLICATION_SETTLE_ATTEMPTS || await readDuringPublication(name) !== content) return fail(); +const settlePublished = async (name: string, content: string, budget: Budget): Promise => { + if (budget.exhausted()) return failBudget(budget, "settle_budget_exhausted", "settle", await describeContention(name)); + if (await readDuringPublication(name, budget) !== content) return fail("settle_record_mismatch"); let incomplete = false; for (const sidecar of publicationSidecars(name)) { - const observed = await readStaging(sidecar, name, content); + const observed = await readStaging(sidecar, name, content, budget); if (observed === "absent") continue; - if (observed === "expected-prefix" && attempt < PUBLICATION_SETTLE_ATTEMPTS) { + if (observed === "expected-prefix" && budget.patient()) { incomplete = true; continue; } - await unlink(sidecar).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail(); }); + await unlink(sidecar).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail("sidecar_unlink_failed"); }); await sync(); } if (incomplete) { - await waitForPublisher(); return settlePublished(name, content, attempt + 1); + await budget.wait(); return settlePublished(name, content, budget); } - if (await readDuringPublication(name) !== content) return fail(); - const remaining = await Promise.all(publicationSidecars(name).map(async (sidecar) => readStaging(sidecar, name, content))); + if (await readDuringPublication(name, budget) !== content) return fail("settle_record_mismatch"); + const remaining = await Promise.all(publicationSidecars(name).map(async (sidecar) => readStaging(sidecar, name, content, budget))); if (remaining.every((sidecar) => sidecar === "absent")) return; - if (attempt >= PUBLICATION_SETTLE_ATTEMPTS) return fail(); - await waitForPublisher(); return settlePublished(name, content, attempt + 1); + if (budget.exhausted()) return failBudget(budget, "settle_budget_exhausted", "settle", await describeContention(name)); + await budget.wait(); return settlePublished(name, content, budget); }; -const readPublished = async (name: string): Promise => { - const content = await readDuringPublication(name); +const readPublished = async (name: string, budget: Budget): Promise => { + const content = await readDuringPublication(name, budget); if (content === null) return null; - await settlePublished(name, content); return content; + await settlePublished(name, content, budget); return content; }; -const write = async (name: string, content: string, attempt = 0): Promise => { - if (attempt > PUBLICATION_SETTLE_ATTEMPTS) return fail(); +const write = async (name: string, content: string, budget: Budget): Promise => { + if (budget.exhausted()) return failBudget(budget, "write_budget_exhausted", "write", await describeContention(name)); const joinOrRetry = async (): Promise => { - const published = await readDuringPublication(name); - if (published !== null) { if (published !== content) return fail(); await settlePublished(name, content); return false; } - await waitForPublisher(); return write(name, content, attempt + 1); + const published = await readDuringPublication(name, budget); + if (published !== null) { if (published !== content) return fail("join_content_mismatch"); await settlePublished(name, content, budget); return false; } + await budget.wait(); return write(name, content, budget); }; const reproveStaging = async (): Promise => { - const published = await readDuringPublication(name); - if (published === content) { await settlePublished(name, content); return true; } - if (published !== null) return fail(); - await waitForPublisher(); return false; + const published = await readDuringPublication(name, budget); + if (published === content) { await settlePublished(name, content, budget); return true; } + if (published !== null) return fail("reprove_content_mismatch"); + await budget.wait(); return false; }; - const nextAttempt = (): number => attempt + 1; - const existing = await readDuringPublication(name); if (existing !== null) { if (existing !== content) fail(); await settlePublished(name, content); return false; } - const pending = `${name}.pending`; const recovery = `${name}.recovery`; const incomplete = await readStaging(pending, name, content); + const existing = await readDuringPublication(name, budget); if (existing !== null) { if (existing !== content) fail("existing_content_mismatch"); await settlePublished(name, content, budget); return false; } + const pending = `${name}.pending`; const recovery = `${name}.recovery`; const incomplete = await readStaging(pending, name, content, budget); if (incomplete === "expected-prefix") { - let recovered = await readStaging(recovery, name, content); + let recovered = await readStaging(recovery, name, content, budget); if (recovered === "expected-prefix") { - if (attempt < PUBLICATION_SETTLE_ATTEMPTS) { - await waitForPublisher(); return write(name, content, attempt + 1); + if (budget.patient()) { + await budget.wait(); return write(name, content, budget); } // A crashed recovery publisher can leave the same bounded prefix. It // cannot win after this writer links the immutable final record, so - // retire it only after the full wait budget and reconstruct it below. - await unlink(recovery).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail(); }); + // retire it only after the budget's patience expires and reconstruct it + // below. + await unlink(recovery).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail("recovery_unlink_failed"); }); await sync(); recovered = "absent"; } if (recovered === "absent") { - const handle = await open(recovery, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY | constants.O_NOFOLLOW, 0o600).catch((error: NodeJS.ErrnoException) => error.code === "EEXIST" ? null : fail()); - if (handle === null) { await waitForPublisher(); return write(name, content, attempt + 1); } + const handle = await open(recovery, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY | constants.O_NOFOLLOW, 0o600).catch((error: NodeJS.ErrnoException) => error.code === "EEXIST" ? null : fail("recovery_create_failed")); + if (handle === null) { await budget.wait(); return write(name, content, budget); } try { await handle.writeFile(content, "utf8"); await handle.sync(); } finally { await handle.close().catch(() => undefined); } - const staged = await readStaging(recovery, name, content); + const staged = await readStaging(recovery, name, content, budget); if (staged !== "exact") { if (await reproveStaging()) return false; - return write(name, content, nextAttempt()); + return write(name, content, budget); } await sync(); } const recoveredLinked = await link(recovery, name).then(() => true).catch((error: NodeJS.ErrnoException) => { if (error.code === "EEXIST") return false; if (error.code === "ENOENT") return null; - return fail(); + return fail("recovery_link_failed"); }); if (recoveredLinked !== true) return joinOrRetry(); - await sync(); await unlink(recovery).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail(); }); await sync(); - if (await readDuringPublication(name) !== content) fail(); await settlePublished(name, content); return true; + await sync(); await unlink(recovery).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail("recovery_unlink_failed"); }); await sync(); + if (await readDuringPublication(name, budget) !== content) fail("recovery_publish_mismatch"); await settlePublished(name, content, budget); return true; } if (incomplete === "absent") { - const handle = await open(pending, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY | constants.O_NOFOLLOW, 0o600).catch((error: NodeJS.ErrnoException) => error.code === "EEXIST" ? null : fail()); - if (handle === null) { await waitForPublisher(); return write(name, content, attempt + 1); } + const handle = await open(pending, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY | constants.O_NOFOLLOW, 0o600).catch((error: NodeJS.ErrnoException) => error.code === "EEXIST" ? null : fail("pending_create_failed")); + if (handle === null) { await budget.wait(); return write(name, content, budget); } try { await handle.writeFile(content, "utf8"); await handle.sync(); } finally { await handle.close().catch(() => undefined); } - const staged = await readStaging(pending, name, content); + const staged = await readStaging(pending, name, content, budget); if (staged !== "exact") { if (await reproveStaging()) return false; - return write(name, content, nextAttempt()); + return write(name, content, budget); } await sync(); } const linked = await link(pending, name).then(() => true).catch((error: NodeJS.ErrnoException) => { if (error.code === "EEXIST") return false; if (error.code === "ENOENT") return null; - return fail(); + return fail("pending_link_failed"); }); if (linked !== true) return joinOrRetry(); - await sync(); await unlink(pending).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail(); }); await sync(); - if (await readDuringPublication(name) !== content) fail(); await settlePublished(name, content); return true; + await sync(); await unlink(pending).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") fail("pending_unlink_failed"); }); await sync(); + if (await readDuringPublication(name, budget) !== content) fail("pending_publish_mismatch"); await settlePublished(name, content, budget); return true; }; let anchor: Anchor | undefined; let queue = Promise.resolve(); const start = async (): Promise => { - const raw = process.env.SPAWNFILE_AUTHORITY_FS_ANCHOR; if (!raw) return fail(); + const raw = process.env.SPAWNFILE_AUTHORITY_FS_ANCHOR; if (!raw) return fail("missing_anchor"); const expected = JSON.parse(raw) as Anchor; const stat = await lstat("."); if (!stat.isDirectory() || stat.isSymbolicLink() || stat.dev !== expected.dev || stat.ino !== expected.ino || (stat.mode & 0o077) !== 0 || owner !== undefined && stat.uid !== expected.uid - || !Number.isSafeInteger(expected.parent_pid) || expected.parent_pid < 1 || process.ppid !== expected.parent_pid) return fail(); + || !Number.isSafeInteger(expected.parent_pid) || expected.parent_pid < 1 || process.ppid !== expected.parent_pid) return fail("anchor_mismatch"); anchor = expected; send({ version: VERSION, ready: true }); }; process.on("message", (raw: unknown) => { queue = queue.then(async () => { - const request = validRequest(raw); if (!anchor) return fail(); const stat = await lstat("."); - if (stat.dev !== anchor.dev || stat.ino !== anchor.ino || (stat.mode & 0o077) !== 0) return fail(); + const request = validRequest(raw); if (!anchor) return fail("not_ready"); const stat = await lstat("."); + if (stat.dev !== anchor.dev || stat.ino !== anchor.ino || (stat.mode & 0o077) !== 0) return fail("anchor_drifted"); + // One budget per request: the request is the thing with a deadline, and + // the client's own request deadline must exceed it. + const budget = new OrganizationHandoffAuthorityBudget(); const result = request.op === "read" - ? { content: await readPublished(request.name) } - : { created: request.content === undefined ? fail() : await write(request.name, request.content) }; + ? { content: await readPublished(request.name, budget) } + : { created: request.content === undefined ? fail("missing_content") : await write(request.name, request.content, budget) }; send({ version: VERSION, id: request.id, ok: true, ...(result.content === null ? {} : { content: result.content }), ...(request.op === "create" ? { created: result.created } : {}) }); - }).catch(() => send({ version: VERSION, id: typeof raw === "object" && raw !== null && Number.isSafeInteger((raw as { id?: unknown }).id) ? (raw as { id: number }).id : 0, ok: false })); + }).catch((error: unknown) => send({ + version: VERSION, + id: typeof raw === "object" && raw !== null && Number.isSafeInteger((raw as { id?: unknown }).id) ? (raw as { id: number }).id : 0, + ok: false, + failure: toOrganizationHandoffAuthorityFailureDetail(error) + })); }); process.on("disconnect", () => process.exit(0)); // IPC disconnect covers a cooperative parent shutdown. ppid surveillance also diff --git a/src/deployment/organizationHandoffAuthorityStore.ts b/src/deployment/organizationHandoffAuthorityStore.ts index f366fb8e..b89e3cdf 100644 --- a/src/deployment/organizationHandoffAuthorityStore.ts +++ b/src/deployment/organizationHandoffAuthorityStore.ts @@ -18,7 +18,14 @@ import { parseOrganizationHandoff, type OrganizationHandoff } from "./organizati import { initializeOrganizationHandoffAuthorityFsClient, type OrganizationHandoffAuthorityFsClient, type OrganizationHandoffAuthorityFsClientOptions } from "./organizationHandoffAuthorityFsClient.js"; const owner = typeof process.getuid === "function" ? process.getuid() : undefined; -const fail = (): never => { throw new Error(ORGANIZATION_HANDOFF_AUTHORITY_ERROR); }; +/** + * The store's public message stays uniform: it reaches CLI output and must not + * describe private resolution state. Diagnostics from the helper are preserved + * on `cause` so an operator log still shows which budget or invariant failed. + */ +const fail = (cause?: unknown): never => { + throw new Error(ORGANIZATION_HANDOFF_AUTHORITY_ERROR, ...(cause === undefined ? [] : [{ cause }] as const)); +}; const key = (value: string): string => Buffer.from(value, "utf8").toString("hex"); const same = (left: unknown, right: unknown): boolean => JSON.stringify(left) === JSON.stringify(right); const pendingKey = (value: unknown): string => typeof value === "string" && /^[a-f0-9]{64}$/u.test(value) ? value : fail(); @@ -38,8 +45,8 @@ const closeRequest = (value: unknown): { expectedHandoff: parseOrganizationHandoff(input.expectedHandoff), organizationHandoffHandle: parseOpaqueTargetHandle(input.organizationHandoffHandle) }); - } catch { - return fail(); + } catch (error) { + return fail(error); } }; @@ -236,7 +243,7 @@ class Store implements OrganizationHandoffAuthorityStore { handoff: final.handoff, network_attachment: Object.freeze({ container_id: final.container_id, deployment_labels: final.deployment_labels, network_attachment_handle: final.handoff.network_attachment_handle }), selected_target_binding: Object.freeze({ receipt: final.selected_target, receipt_digest: final.selected_target_receipt_digest }) }); - } catch { return fail(); } + } catch (error) { return fail(error); } } } @@ -262,10 +269,10 @@ export const initializeOrganizationHandoffAuthorityStore = async (options: Organ // Transfer each completed helper into rollback ownership immediately. clients.set(part, client); } - } catch { + } catch (error) { await Promise.all([...clients.values()].map(async (client) => client.dispose())); await Promise.all([...anchors.values()].map(async ({ handle }) => { await handle.close().catch(() => undefined); })); - return fail(); + return fail(error); } return new Store(root, anchors, clients, options.testHooks); }; diff --git a/src/deployment/organizationHandoffAuthorityTypes.ts b/src/deployment/organizationHandoffAuthorityTypes.ts index 18c8e1f6..ff812cdc 100644 --- a/src/deployment/organizationHandoffAuthorityTypes.ts +++ b/src/deployment/organizationHandoffAuthorityTypes.ts @@ -6,13 +6,14 @@ import { type OpaqueTargetHandle, type SelectedTargetReceipt } from "../target/index.js"; import { dockerDeploymentLabelKeys } from "./dockerLabels.js"; +import { ORGANIZATION_HANDOFF_AUTHORITY_ERROR } from "./organizationHandoffAuthorityFsBudget.js"; import { parseOrganizationHandoff, type OrganizationHandoff } from "./organizationHandoffTypes.js"; export const ORGANIZATION_HANDOFF_CAPABILITY_VERSION = "spawnfile.organization-handoff-capability.private.v1" as const; export const ORGANIZATION_HANDOFF_RECOVERY_VERSION = "spawnfile.organization-handoff-recovery.private.v1" as const; -export const ORGANIZATION_HANDOFF_AUTHORITY_ERROR = "Organization handoff authority failed"; +export { ORGANIZATION_HANDOFF_AUTHORITY_ERROR }; const DIGEST = /^sha256:[a-f0-9]{64}$/u; const CONTAINER = /^[a-f0-9]{64}$/u; From da9098ffa2b2cc0dccb624f75d040e7405f0fe9c Mon Sep 17 00:00:00 2001 From: Juan Cruz Fortunatti Date: Fri, 4 Sep 2026 05:14:54 +0200 Subject: [PATCH 2/4] fix: stop the handoff worker watchdog from reaping the helper during startup --- ...ganizationHandoffAuthorityFsClient.test.ts | 46 ++++++++++++++++++- .../organizationHandoffAuthorityFsClient.ts | 18 ++++++-- .../organizationHandoffAuthorityFsWorker.ts | 44 ++++++++++++++++-- ...onHandoffAuthorityFsWorkerStall.fixture.ts | 18 ++++++++ 4 files changed, 116 insertions(+), 10 deletions(-) create mode 100644 src/deployment/organizationHandoffAuthorityFsWorkerStall.fixture.ts diff --git a/src/deployment/organizationHandoffAuthorityFsClient.test.ts b/src/deployment/organizationHandoffAuthorityFsClient.test.ts index e1669a20..4bd6e9ed 100644 --- a/src/deployment/organizationHandoffAuthorityFsClient.test.ts +++ b/src/deployment/organizationHandoffAuthorityFsClient.test.ts @@ -1,7 +1,9 @@ -import type { ChildProcess } from "node:child_process"; +import { fork, type ChildProcess } from "node:child_process"; import { lstat, mkdtemp, readdir, rm } from "node:fs/promises"; +import { createRequire } from "node:module"; import os from "node:os"; import path from "node:path"; +import { fileURLToPath } from "node:url"; import { afterEach, expect, it } from "vitest"; @@ -22,6 +24,48 @@ it("disposes promptly and idempotently after a worker has already exited by sign await client.dispose(); }); +it("reaches readiness when startup is stalled past a liveness watchdog tick", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "spawnfile-handoff-fs-client-")); directories.push(directory); + const stat = await lstat(directory); + const require = createRequire(import.meta.url); + // Fork the worker directly: the stall has to be injected into the helper's + // own startup, which the client API deliberately does not expose. + const worker = fork(fileURLToPath(new URL("./organizationHandoffAuthorityFsWorker.ts", import.meta.url)), [], { + cwd: directory, + env: { + SPAWNFILE_AUTHORITY_FS_ANCHOR: JSON.stringify({ + dev: stat.dev, ino: stat.ino, + ...(typeof process.getuid === "function" ? { uid: process.getuid() } : {}), + parent_pid: process.pid + }), + // One threadpool slot, held by the stall preload, so the anchor lstat + // cannot complete before the watchdog ticks. + UV_THREADPOOL_SIZE: "1" + }, + execArgv: [ + "--import", require.resolve("tsx"), + "--import", fileURLToPath(new URL("./organizationHandoffAuthorityFsWorkerStall.fixture.ts", import.meta.url)) + ], + silent: true + }); + worker.stdout?.resume(); worker.stderr?.resume(); + try { + const outcome = await new Promise((resolve) => { + const timer = setTimeout(() => resolve("timeout"), 10_000); + worker.on("message", (raw: unknown) => { + if ((raw as { ready?: unknown } | null)?.ready === true) { clearTimeout(timer); resolve("ready"); } + }); + worker.once("exit", (code, signal) => { clearTimeout(timer); resolve(`exit:${String(code)}:${String(signal)}`); }); + }); + // A watchdog that treats "not yet ready" as "orphaned" reaps the helper + // here with a clean exit(0) before it can ever report readiness. + expect(outcome).toBe("ready"); + } finally { + worker.kill("SIGKILL"); + await new Promise((resolve) => { worker.exitCode !== null || worker.signalCode !== null ? resolve() : worker.once("exit", () => resolve()); }); + } +}); + it("converges full-size concurrent publishers without leaving staging sidecars", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "spawnfile-handoff-fs-client-")); directories.push(directory); const stat = await lstat(directory); diff --git a/src/deployment/organizationHandoffAuthorityFsClient.ts b/src/deployment/organizationHandoffAuthorityFsClient.ts index 974fbdce..e3d4fbfc 100644 --- a/src/deployment/organizationHandoffAuthorityFsClient.ts +++ b/src/deployment/organizationHandoffAuthorityFsClient.ts @@ -17,11 +17,16 @@ const MAX_BYTES = 32_768; const REQUEST_DEADLINE_MS = PUBLICATION_BUDGET_MS + 3_000; /** * Startup deadline for the worker's ready handshake. This bounds a hung or - * broken helper; it is not a latency budget for process startup. Several - * helpers are forked concurrently and a loaded host schedules their - * interpreter startup slowly, so it is sized for a saturated machine. + * broken helper; it is not a latency budget for process startup. + * + * Measured ready latency for eight concurrently forked source-mode helpers on + * a host at eleven times CPU oversubscription is 67-411ms, so this is margin + * rather than a fix — the observed startup failures were the helper's own + * liveness watchdog killing it, not this deadline expiring. It is sized above + * that measurement because CI forks far more helpers at once, and a caller + * initializes several clients in sequence, which bounds the worst case. */ -const READY_DEADLINE_MS = 20_000; +const READY_DEADLINE_MS = 5_000; const DISPOSE_GRACE_MS = 1_000; const fail = (code: string, state?: string): never => { @@ -163,7 +168,10 @@ export const initializeOrganizationHandoffAuthorityFsClient = async (options: Or function doneResolve(): void { clean(); resolve(); } function doneReject(code: string): void { clean(); - detail = { budget: "worker_ready", code, elapsedMs: performance.now() - startedAt, limitMs: READY_DEADLINE_MS, state: `source=${String(tsWorker)}` }; + detail = { + budget: "worker_ready", code, elapsedMs: performance.now() - startedAt, limitMs: READY_DEADLINE_MS, + state: `source=${String(tsWorker)} exit=${String(child.exitCode)} signal=${String(child.signalCode)}` + }; reject(createOrganizationHandoffAuthorityError(detail)); } timer = setTimeout(() => doneReject("worker_ready_deadline"), READY_DEADLINE_MS); diff --git a/src/deployment/organizationHandoffAuthorityFsWorker.ts b/src/deployment/organizationHandoffAuthorityFsWorker.ts index 5e868b88..5ff845be 100644 --- a/src/deployment/organizationHandoffAuthorityFsWorker.ts +++ b/src/deployment/organizationHandoffAuthorityFsWorker.ts @@ -22,6 +22,18 @@ const fail = (code: string, state?: string): never => { const failBudget = (budget: Budget, code: string, loop: string, state?: string): never => { throw new OrganizationHandoffAuthorityFailure(budget.snapshot(code, loop, state)); }; +/** + * Re-raise a race failure that the election check refused to retry, recording + * the verdict. Without it a fail-closed admission decision is indistinguishable + * from the underlying race in a log. + */ +const failElection = (error: unknown, election: boolean | null, loop: string): never => { + const detail = toOrganizationHandoffAuthorityFailureDetail(error); + throw new OrganizationHandoffAuthorityFailure({ + ...detail, + state: `${detail.state === undefined ? "" : `${detail.state} `}loop=${loop} election=${String(election)}` + }); +}; const bytes = (value: string): number => Buffer.byteLength(value, "utf8"); const validName = (value: unknown): string => typeof value === "string" && NAME.test(value) ? value : fail("invalid_name"); const validRequest = (raw: unknown): Request => { @@ -116,7 +128,14 @@ const expectedElectionState = async (name: string): Promise => { } // The counterpart may have disappeared just before this check. Accept only // the resulting ordinary single-link state, never an unknown hard link. - stat = await statFile(name); return stat?.nlink === 1; + // + // The leaf itself may also have been unlinked in that same window: the + // helper that won the election removes its staging sidecar immediately after + // linking the final record. That is an absence, not an unknown link, and + // reporting it as a rejected election is what turned this benign race into a + // hard failure for a concurrent publisher. + stat = await statFile(name); if (stat === null) return null; + return stat.nlink === 1; }; const readDuringPublication = async (name: string, budget: Budget): Promise => { try { return await read(name); } catch (error) { @@ -125,7 +144,7 @@ const readDuringPublication = async (name: string, budget: Budget): Promise { })); }); process.on("disconnect", () => process.exit(0)); +/** + * Expected parent pid read synchronously at module load, before any await. + * + * The watchdog below previously exited whenever `anchor` was still unset. That + * made a liveness net race the startup path: on a loaded host the interpreter + * and the anchor's `lstat` can take longer than one 100ms tick, and the helper + * killed itself before it could ever report ready. Deriving the pid from the + * environment removes the dependency on startup having finished, without + * weakening anything: `start` still validates the full anchor independently, + * and no request is served until it does. + */ +const watchedParentPid = ((): number | undefined => { + try { + const value = JSON.parse(process.env.SPAWNFILE_AUTHORITY_FS_ANCHOR ?? "{}") as { parent_pid?: unknown }; + return Number.isSafeInteger(value.parent_pid) ? value.parent_pid as number : undefined; + } catch { return undefined; } +})(); // IPC disconnect covers a cooperative parent shutdown. ppid surveillance also // covers abrupt parent death, where an orphan might otherwise retain its cwd. -setInterval(() => { if (!anchor || process.ppid !== anchor.parent_pid) process.exit(0); }, 100).unref(); +setInterval(() => { if (watchedParentPid === undefined || process.ppid !== watchedParentPid) process.exit(0); }, 100).unref(); void start().catch(() => process.exit(1)); diff --git a/src/deployment/organizationHandoffAuthorityFsWorkerStall.fixture.ts b/src/deployment/organizationHandoffAuthorityFsWorkerStall.fixture.ts new file mode 100644 index 00000000..2799e94a --- /dev/null +++ b/src/deployment/organizationHandoffAuthorityFsWorkerStall.fixture.ts @@ -0,0 +1,18 @@ +import { pbkdf2 } from "node:crypto"; + +/** + * Test-only preload that delays the authority worker's startup filesystem call. + * + * The worker validates its anchor with an `lstat`, which libuv runs on the + * threadpool. Forked with `UV_THREADPOOL_SIZE=1`, this preload occupies that + * single slot long enough that the anchor cannot be established until well + * after the worker's 100ms liveness watchdog has ticked several times — the + * deterministic equivalent of what CPU contention does to worker startup on a + * loaded host, and the condition under which the watchdog used to reap the + * helper before it could ever report readiness. + * + * The event loop itself stays free throughout, so the watchdog does tick. + */ +const STALL_ITERATIONS = 1_500_000; + +pbkdf2("spawnfile-stall", "spawnfile-stall", STALL_ITERATIONS, 64, "sha512", () => undefined); From 69b686a65a394060a4e15c6b338dd0ffbacd4b6e Mon Sep 17 00:00:00 2001 From: Juan Cruz Fortunatti Date: Fri, 4 Sep 2026 05:23:23 +0200 Subject: [PATCH 3/4] fix: report an unlinked handoff staging leaf as absent instead of a rejected election --- ...nizationHandoffAuthorityFsElection.test.ts | 107 ++++++++++++++++++ .../organizationHandoffAuthorityFsElection.ts | 83 ++++++++++++++ .../organizationHandoffAuthorityFsWorker.ts | 32 +----- 3 files changed, 192 insertions(+), 30 deletions(-) create mode 100644 src/deployment/organizationHandoffAuthorityFsElection.test.ts create mode 100644 src/deployment/organizationHandoffAuthorityFsElection.ts diff --git a/src/deployment/organizationHandoffAuthorityFsElection.test.ts b/src/deployment/organizationHandoffAuthorityFsElection.test.ts new file mode 100644 index 00000000..c281b373 --- /dev/null +++ b/src/deployment/organizationHandoffAuthorityFsElection.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it } from "vitest"; + +import { OrganizationHandoffAuthorityFailure } from "./organizationHandoffAuthorityFsBudget.js"; +import { + createAuthorityLeafInspector, type AuthorityLeafStat +} from "./organizationHandoffAuthorityFsElection.js"; + +const enoent = (): NodeJS.ErrnoException => + Object.assign(new Error("ENOENT"), { code: "ENOENT" }); + +const leaf = (overrides: Partial> & { + readonly file?: boolean; readonly symlink?: boolean; +} = {}): AuthorityLeafStat => ({ + dev: overrides.dev ?? 1, ino: overrides.ino ?? 10, mode: overrides.mode ?? 0o600, + nlink: overrides.nlink ?? 1, size: overrides.size ?? 30_000, uid: overrides.uid ?? 501, + isFile: () => overrides.file ?? true, + isSymbolicLink: () => overrides.symlink ?? false +} as unknown as AuthorityLeafStat); + +/** + * Drive the inspector through a scripted sequence of `lstat` outcomes, which is + * the only way to stage the mid-race interleavings this logic exists to judge. + */ +const scripted = (steps: readonly (AuthorityLeafStat | NodeJS.ErrnoException)[]) => { + const seen: string[] = []; + let index = 0; + const inspector = createAuthorityLeafInspector({ + lstat: async (name) => { + seen.push(name); + const step = steps[Math.min(index, steps.length - 1)]; + index += 1; + if (step instanceof Error) throw step; + return step; + }, + owner: 501 + }); + return { inspector, seen }; +}; + +describe("authority leaf election", () => { + it("admits a retry for an ordinary single-link leaf", async () => { + const { inspector } = scripted([leaf({ nlink: 1 })]); + expect(await inspector.expectedElectionState("record.json")).toBe(true); + }); + + it("reports an absent leaf rather than a verdict", async () => { + const { inspector } = scripted([enoent()]); + expect(await inspector.expectedElectionState("record.json")).toBeNull(); + }); + + it("admits a retry when an alias holds the other end of the publication", async () => { + // Leaf at two links, then the final record matching device and inode. + const { inspector } = scripted([ + leaf({ nlink: 2 }), enoent(), enoent(), leaf({ nlink: 2 }) + ]); + expect(await inspector.expectedElectionState("record.json.pending")).toBe(true); + }); + + it("reports absence when the winning publisher unlinks the sidecar mid-check", async () => { + // The exact losing-publisher interleaving: the sidecar is seen at two + // links; by the time its aliases are scanned the winner has already linked + // the final record and unlinked this sidecar, so every alias lstat and the + // rescan miss. Reporting `false` here fails a benign, converging race. + const { inspector } = scripted([ + leaf({ nlink: 2 }), enoent(), enoent(), leaf({ nlink: 1, ino: 99 }), enoent() + ]); + expect(await inspector.expectedElectionState("record.json.pending")).toBeNull(); + }); + + it("still fails closed on an unexplained second hard link", async () => { + const { inspector } = scripted([ + leaf({ nlink: 2 }), enoent(), enoent(), leaf({ nlink: 1, ino: 99 }), leaf({ nlink: 2 }) + ]); + expect(await inspector.expectedElectionState("record.json.pending")).toBe(false); + }); + + it("scans exactly the leaves that can hold the counterpart", () => { + const { inspector } = scripted([leaf()]); + expect(inspector.aliasesOf("record.json")).toEqual(["record.json.pending", "record.json.recovery"]); + expect(inspector.aliasesOf("record.json.pending")) + .toEqual(["record.json.pending.pending", "record.json.pending.recovery", "record.json"]); + expect(inspector.aliasesOf("record.json.recovery")) + .toEqual(["record.json.recovery.pending", "record.json.recovery.recovery", "record.json"]); + }); + + it("rejects leaves that are not ordinary owned records", async () => { + for (const bad of [ + leaf({ file: false }), leaf({ symlink: true }), leaf({ nlink: 3 }), + leaf({ size: 32_769 }), leaf({ mode: 0o640 }), leaf({ uid: 502 }) + ]) { + const { inspector } = scripted([bad]); + await expect(inspector.statFile("record.json")).rejects.toThrow("leaf_not_ordinary"); + } + }); + + it("surfaces a non-ENOENT stat error as a diagnosable failure", async () => { + const { inspector } = scripted([Object.assign(new Error("EIO"), { code: "EIO" })]); + await expect(inspector.statFile("record.json")).rejects.toBeInstanceOf(OrganizationHandoffAuthorityFailure); + const { inspector: other } = scripted([Object.assign(new Error("EIO"), { code: "EIO" })]); + await expect(other.statFile("record.json")).rejects.toThrow("lstat_failed"); + }); + + it("treats an absent leaf as absent regardless of ownership checks", async () => { + const inspector = createAuthorityLeafInspector({ lstat: async () => { throw enoent(); } }); + expect(await inspector.statFile("record.json")).toBeNull(); + }); +}); diff --git a/src/deployment/organizationHandoffAuthorityFsElection.ts b/src/deployment/organizationHandoffAuthorityFsElection.ts new file mode 100644 index 00000000..419f9ed9 --- /dev/null +++ b/src/deployment/organizationHandoffAuthorityFsElection.ts @@ -0,0 +1,83 @@ +import type { Stats } from "node:fs"; +import { lstat as nodeLstat } from "node:fs/promises"; + +import { OrganizationHandoffAuthorityFailure } from "./organizationHandoffAuthorityFsBudget.js"; + +/** + * Leaf inspection and link-election admission for the organization handoff + * filesystem authority. + * + * These three functions decide whether a failed read may be retried or must + * fail closed, which makes them the authority's most consequential branch and + * the reason they live behind an injectable `lstat`: the interesting states + * are mid-race interleavings that cannot be staged on a real filesystem. + */ + +const MAX_BYTES = 32_768; + +export type AuthorityLeafStat = Stats; +export type AuthorityLstat = (name: string) => Promise; + +export interface AuthorityLeafInspector { + /** Every leaf that could legitimately hold the other end of a two-link publication. */ + aliasesOf(name: string): readonly string[]; + /** + * Whether an observed publication race is one the caller may retry. + * + * `true` permits a retry, `false` fails closed, and `null` reports that the + * leaf is simply absent. + */ + expectedElectionState(name: string): Promise; + /** Stat a leaf, rejecting anything that is not an ordinary owned record. */ + statFile(name: string): Promise; +} + +export interface AuthorityLeafInspectorOptions { + readonly lstat?: AuthorityLstat; + readonly owner?: number; +} + +const fail = (code: string): never => { throw new OrganizationHandoffAuthorityFailure({ code }); }; + +export const createAuthorityLeafInspector = ( + options: AuthorityLeafInspectorOptions = {} +): AuthorityLeafInspector => { + const lstat = options.lstat ?? nodeLstat; + const owner = options.owner; + + const statFile = async (name: string): Promise => { + const stat = await lstat(name).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("lstat_failed")); + if (stat === null) return null; + if (!stat.isFile() || stat.isSymbolicLink() || stat.nlink > 2 || stat.size > MAX_BYTES || (stat.mode & 0o077) !== 0 + || owner !== undefined && stat.uid !== owner) return fail("leaf_not_ordinary"); + return stat; + }; + + const aliasesOf = (name: string): readonly string[] => [ + `${name}.pending`, `${name}.recovery`, + ...(name.endsWith(".pending") ? [name.slice(0, -".pending".length)] : []), + ...(name.endsWith(".recovery") ? [name.slice(0, -".recovery".length)] : []) + ]; + + const expectedElectionState = async (name: string): Promise => { + let stat = await statFile(name); if (stat === null) return null; + if (stat.nlink === 1) return true; + for (const alias of aliasesOf(name)) { + const counterpart = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("election_lstat_failed")); + if (counterpart?.isFile() && !counterpart.isSymbolicLink() && counterpart.nlink === 2 && counterpart.dev === stat.dev && counterpart.ino === stat.ino) return true; + } + // The counterpart may have disappeared just before this check. Accept only + // the resulting ordinary single-link state, never an unknown hard link. + // + // The leaf itself may also have gone: the helper that won the election + // unlinks its staging sidecar immediately after linking the final record, + // so a peer that observed the sidecar at two links can find it absent one + // syscall later. That is an absence, not an unknown link. Reporting it as + // a rejected election turned a benign, converging race into a hard failure + // for the losing publisher. + stat = await statFile(name); if (stat === null) return null; + return stat.nlink === 1; + }; + + return { aliasesOf, expectedElectionState, statFile }; +}; diff --git a/src/deployment/organizationHandoffAuthorityFsWorker.ts b/src/deployment/organizationHandoffAuthorityFsWorker.ts index 5ff845be..7321c56b 100644 --- a/src/deployment/organizationHandoffAuthorityFsWorker.ts +++ b/src/deployment/organizationHandoffAuthorityFsWorker.ts @@ -5,6 +5,7 @@ import { OrganizationHandoffAuthorityBudget, OrganizationHandoffAuthorityFailure, toOrganizationHandoffAuthorityFailureDetail } from "./organizationHandoffAuthorityFsBudget.js"; +import { createAuthorityLeafInspector } from "./organizationHandoffAuthorityFsElection.js"; const VERSION = "spawnfile.organization-handoff-fs-worker.v1"; const MAX_BYTES = 32_768; @@ -50,13 +51,7 @@ const send = (value: unknown): void => { const serialized = JSON.stringify(value); if (bytes(serialized) > MAX_BYTES || typeof process.send !== "function") process.exit(1); process.send(JSON.parse(serialized)); }; -const statFile = async (name: string) => { - const stat = await lstat(name).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("lstat_failed")); - if (stat === null) return null; - if (!stat.isFile() || stat.isSymbolicLink() || stat.nlink > 2 || stat.size > MAX_BYTES || (stat.mode & 0o077) !== 0 - || owner !== undefined && stat.uid !== owner) return fail("leaf_not_ordinary"); - return stat; -}; +const { aliasesOf, expectedElectionState, statFile } = createAuthorityLeafInspector(owner === undefined ? {} : { owner }); const publicationSidecars = (name: string): readonly string[] => [`${name}.pending`, `${name}.recovery`]; /** * Structural description of the contending state for a diagnostic. Reports @@ -73,11 +68,6 @@ const describeContention = async (name: string): Promise => { })); return parts.join(" "); }; -const aliasesOf = (name: string): readonly string[] => [ - `${name}.pending`, `${name}.recovery`, - ...(name.endsWith(".pending") ? [name.slice(0, -".pending".length)] : []), - ...(name.endsWith(".recovery") ? [name.slice(0, -".recovery".length)] : []) -]; const read = async (name: string): Promise => { let before = await statFile(name); if (before === null) return null; if (before.nlink === 2) { @@ -119,24 +109,6 @@ const read = async (name: string): Promise => { } finally { await handle.close().catch(() => undefined); } }; const sync = async (): Promise => { const handle = await open(".", constants.O_RDONLY | constants.O_DIRECTORY).catch(() => fail("directory_open_failed")); try { await handle.sync(); } finally { await handle.close().catch(() => undefined); } }; -const expectedElectionState = async (name: string): Promise => { - let stat = await statFile(name); if (stat === null) return null; - if (stat.nlink === 1) return true; - for (const alias of aliasesOf(name)) { - const counterpart = await lstat(alias).catch((error: NodeJS.ErrnoException) => error.code === "ENOENT" ? null : fail("election_lstat_failed")); - if (counterpart?.isFile() && !counterpart.isSymbolicLink() && counterpart.nlink === 2 && counterpart.dev === stat.dev && counterpart.ino === stat.ino) return true; - } - // The counterpart may have disappeared just before this check. Accept only - // the resulting ordinary single-link state, never an unknown hard link. - // - // The leaf itself may also have been unlinked in that same window: the - // helper that won the election removes its staging sidecar immediately after - // linking the final record. That is an absence, not an unknown link, and - // reporting it as a rejected election is what turned this benign race into a - // hard failure for a concurrent publisher. - stat = await statFile(name); if (stat === null) return null; - return stat.nlink === 1; -}; const readDuringPublication = async (name: string, budget: Budget): Promise => { try { return await read(name); } catch (error) { // Re-validate the leaf before retrying. This permits only a checked, From 9524a070ff72273e6194a32287705177086e1027 Mon Sep 17 00:00:00 2001 From: Juan Cruz Fortunatti Date: Fri, 4 Sep 2026 05:57:03 +0200 Subject: [PATCH 4/4] test: cover the handoff authority client's diagnostic failure contract --- ...ganizationHandoffAuthorityFsClient.test.ts | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/deployment/organizationHandoffAuthorityFsClient.test.ts b/src/deployment/organizationHandoffAuthorityFsClient.test.ts index 4bd6e9ed..ef2f91ff 100644 --- a/src/deployment/organizationHandoffAuthorityFsClient.test.ts +++ b/src/deployment/organizationHandoffAuthorityFsClient.test.ts @@ -86,3 +86,41 @@ it("converges full-size concurrent publishers without leaving staging sidecars", await Promise.all(clients.map(async (client) => client.dispose())); } }); + +it("reports which request the worker rejected and why", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "spawnfile-handoff-fs-client-")); directories.push(directory); + const stat = await lstat(directory); + const client = await initializeOrganizationHandoffAuthorityFsClient({ + cwd: directory, dev: stat.dev, ino: stat.ino, + ...(typeof process.getuid === "function" ? { uid: process.getuid() } : {}) + }); + const name = `${"a".repeat(128)}.json`; + try { + expect(await client.read(name)).toBeNull(); + expect(await client.create(name, "first")).toBe(true); + expect(await client.create(name, "first")).toBe(false); + // The record is immutable, so a conflicting publication must fail — and + // must say which invariant refused it rather than reporting a bare failure. + await expect(client.create(name, "second")).rejects.toThrow(/existing_content_mismatch.*op=create/u); + } finally { + await client.dispose(); + } +}); + +it("names the client-side limit it refused a request against", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "spawnfile-handoff-fs-client-")); directories.push(directory); + const stat = await lstat(directory); + const client = await initializeOrganizationHandoffAuthorityFsClient({ + cwd: directory, dev: stat.dev, ino: stat.ino, + ...(typeof process.getuid === "function" ? { uid: process.getuid() } : {}) + }); + const name = `${"b".repeat(128)}.json`; + try { + await expect(client.create("not-a-record-name", "x")).rejects.toThrow("invalid_name"); + await expect(client.create(name, "x".repeat(32_769))).rejects.toThrow(/content_too_large.*limit=32768/u); + await expect(client.create(name, "x".repeat(32_700))).rejects.toThrow(/packet_too_large/u); + } finally { + await client.dispose(); + } + await expect(client.read(name)).rejects.toThrow("client_closed"); +});