fix: converge concurrent organization handoff publishers under load

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");
+  });
+});

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<string, unknown>;
+  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<void>;
+}
+
+/**
+ * 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<void>;
+  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<void> {
+    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);
+  }
+}