feat(admin): preserve work and track accepted jobs (stage A)

.github/workflows/ci.yml

@@ -341,9 +341,10 @@ jobs:
   database:
     name: test (database, ${{ matrix.group }})
     runs-on: ubuntu-latest
-    # Native recovery suites run independently with real ownership/staging deadlines.
-    # Retain ingestion's existing budget while gathering its reduced runtime.
-    timeout-minutes: ${{ contains(fromJSON('["ingestion", "configuration-recovery", "coordinator-recovery"]'), matrix.group) && 90 || 30 }}
+    # API/privacy includes the full native platform-helper recovery matrix.
+    # Give native recovery groups headroom for setup and all serial suites;
+    # individual test and execution deadlines remain independently enforced.
+    timeout-minutes: ${{ contains(fromJSON('["api-privacy", "ingestion", "configuration-recovery", "coordinator-recovery"]'), matrix.group) && 90 || 30 }}
     strategy:
       # Each group has a separate runner and service; keep the other groups
       # running after a failure so all applicable evidence is collected.

apps/ops-agent/src/artifact-transfer/routes.test.ts

@@ -4,13 +4,14 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { issueArtifactStreamCapability } from "@openmapx/core/ops";
 import Fastify from "fastify";
-import { afterEach, expect, it } from "vitest";
+import { afterEach, expect, it, vi } from "vitest";
 import { registerArtifactTransferRoutes } from "./routes";
 import { ArtifactTransferStore } from "./store";
 
 const cleanups: Array<() => Promise<void>> = [];
 afterEach(async () => {
   for (const close of cleanups.splice(0).reverse()) await close();
+  vi.restoreAllMocks();
 });
 it("authenticates before streaming, preserves binary bytes, refuses purpose and attempt replay", async () => {
   const root = await mkdtemp(join(tmpdir(), "artifact-http-"));
@@ -37,6 +38,12 @@ it("authenticates before streaming, preserves binary bytes, refuses purpose and
     declaredBytes: 4,
     operationKey: `opk1_${"a".repeat(20)}`,
   });
+  const now = Date.now();
+  vi.spyOn(Date, "now")
+    .mockImplementationOnce(() => now)
+    .mockImplementationOnce(() => now + 1)
+    .mockImplementationOnce(() => now + 2);
+  const issuedAt = Date.now();
   const input = {
     version: 1 as const,
     stageId: stage.stageId,
@@ -45,9 +52,9 @@ it("authenticates before streaming, preserves binary bytes, refuses purpose and
     direction: "upload" as const,
     byteCap: 4,
     attemptId: randomUUID(),
-    issuedAt: Date.now(),
-    expiresAt: Date.now() + 60000,
-    deadlineAt: Date.now() + 10000,
+    issuedAt,
+    expiresAt: issuedAt + 60000,
+    deadlineAt: issuedAt + 10000,
   };
   const cap = issueArtifactStreamCapability(token, input);
   const path = `/v1/artifact-transfers/${stage.stageId}/content`;

apps/ops-agent/src/configuration-handoff-control.test.ts

@@ -0,0 +1,185 @@
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
+
+const mock = vi.hoisted(() => ({
+  openFixedChildInstallation: vi.fn(),
+  readConfigurationHandoffSpawnIntent: vi.fn(),
+  verifyFixedCliControlPeer: vi.fn(),
+  verifiedParentForArtifactDelegation: vi.fn(),
+  adoptArtifactTargetLease: vi.fn(),
+  retainAdmittedControlExecution: vi.fn(),
+  bindBorrowedControlExecution: vi.fn(),
+  assertArtifactTargetLeaseExecution: vi.fn(),
+  beginFixedCliExecution: vi.fn(),
+  closeArtifactTargetLeaseHandle: vi.fn(),
+  closeControlInstallation: vi.fn(),
+  executionAuthorityState: vi.fn(),
+}));
+vi.mock("@openmapx/core/build-artifacts", async (original) => ({
+  ...(await original<typeof import("@openmapx/core/build-artifacts")>()),
+  ...mock,
+}));
+vi.mock("node:timers/promises", async (original) => ({
+  ...(await original<typeof import("node:timers/promises")>()),
+  setTimeout: (milliseconds: number) =>
+    new Promise<void>((resolve) => setTimeout(resolve, milliseconds)),
+}));
+
+import {
+  ArtifactTargetAdoptionObservationBusyError,
+  ArtifactTargetJournalBusyError,
+} from "@openmapx/core/build-artifacts";
+import { createControlRootOperationKey } from "@openmapx/core/ops";
+import {
+  closeConfigurationHandoffChild,
+  openConfigurationHandoffChild,
+} from "./configuration-handoff-control";
+
+// These mocks isolate retry/cleanup control flow. The target-lease filesystem
+// tests and native platform suite verify the actual authority and adoption.
+function fixture() {
+  const deadlineAt = Date.now() + 60000,
+    execution = {
+      executor: "host-cli" as const,
+      principal: "host-cli" as const,
+      executionId: "11111111-1111-4111-8111-111111111111",
+      operationKey: createControlRootOperationKey({
+        notAfter: deadlineAt,
+        nonce: Buffer.alloc(16),
+      }),
+      operationFingerprint: "a".repeat(64),
+      fingerprint: "b".repeat(64),
+      deadlineAt,
+    },
+    input = {
+      rootDir: "/synthetic/owned-deployment",
+      handoffId: "22222222-2222-4222-8222-222222222222",
+      assertion: "original-fixed-child-assertion",
+      execution,
+    },
+    installation = { installationId: "fixture-installation", publicKey: "original-child-key" },
+    preparer = { operationKey: "original-preparer", fingerprint: "c".repeat(64) },
+    peer = Object.freeze({ fixture: "original-peer" }),
+    lease = Object.freeze({ fixture: "original-lease" });
+  mock.openFixedChildInstallation.mockReturnValue(installation);
+  mock.readConfigurationHandoffSpawnIntent.mockReturnValue({
+    installationId: installation.installationId,
+    operation: { handoffId: input.handoffId, serviceIds: ["motis"] },
+    assertion: input.assertion,
+    helperPublicKey: installation.publicKey,
+    helper: execution,
+    preparer,
+    admittedAt: Date.now(),
+    requestedDeadlineAt: deadlineAt,
+  });
+  mock.verifyFixedCliControlPeer.mockReturnValue(peer);
+  mock.verifiedParentForArtifactDelegation.mockReturnValue(preparer);
+  mock.adoptArtifactTargetLease.mockResolvedValue(lease);
+  mock.retainAdmittedControlExecution.mockReturnValue(Object.freeze({ fixture: "authority" }));
+  return { input, installation, peer, lease };
+}
+beforeEach(() => {
+  vi.resetAllMocks();
+  vi.useFakeTimers();
+});
+afterEach(() => vi.useRealTimers());
+
+it.each([
+  () => new ArtifactTargetAdoptionObservationBusyError(Error("held observation guard")),
+  () => new ArtifactTargetJournalBusyError(Error("held journal guard")),
+])("retries only a nominal pre-adoption refusal with the original identity", async (busy) => {
+  const f = fixture();
+  mock.adoptArtifactTargetLease.mockRejectedValueOnce(busy());
+  const pending = openConfigurationHandoffChild(f.input);
+  void pending.catch(() => {});
+  await vi.advanceTimersByTimeAsync(50);
+  const child = await pending;
+  try {
+    expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(2);
+    const [first, second] = mock.adoptArtifactTargetLease.mock.calls;
+    expect(second).toEqual(first);
+    expect(second?.[0].peer).toBe(first?.[0].peer);
+    expect(first?.[0]).toMatchObject({
+      assertion: f.input.assertion,
+      execution: f.input.execution,
+      peer: f.peer,
+      serviceIds: ["motis"],
+    });
+    expect(mock.retainAdmittedControlExecution).toHaveBeenCalledTimes(1);
+    expect(mock.beginFixedCliExecution).toHaveBeenCalledTimes(1);
+  } finally {
+    closeConfigurationHandoffChild(child);
+  }
+});
+
+it("stops at the five-second contention budget without adopting or retaining authority", async () => {
+  const f = fixture(),
+    error = new ArtifactTargetAdoptionObservationBusyError(Error("persistent guard"));
+  mock.adoptArtifactTargetLease.mockRejectedValue(error);
+  const failed = expect(openConfigurationHandoffChild(f.input)).rejects.toBe(error);
+  await vi.advanceTimersByTimeAsync(5050);
+  await failed;
+  expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(101);
+  expect(mock.retainAdmittedControlExecution).not.toHaveBeenCalled();
+  expect(mock.beginFixedCliExecution).not.toHaveBeenCalled();
+  expect(mock.closeControlInstallation).toHaveBeenCalledExactlyOnceWith(f.installation);
+});
+
+it("stops at the original deadline before the five-second wait budget", async () => {
+  const f = fixture(),
+    error = new ArtifactTargetAdoptionObservationBusyError(Error("held guard"));
+  f.input.execution.deadlineAt = Date.now() + 20;
+  mock.adoptArtifactTargetLease.mockRejectedValue(error);
+  const failed = expect(openConfigurationHandoffChild(f.input)).rejects.toBe(error);
+  await vi.advanceTimersByTimeAsync(100);
+  await failed;
+  expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(2);
+  for (const [input] of mock.adoptArtifactTargetLease.mock.calls)
+    expect(input.execution.deadlineAt).toBe(f.input.execution.deadlineAt);
+  expect(mock.retainAdmittedControlExecution).not.toHaveBeenCalled();
+  expect(mock.closeControlInstallation).toHaveBeenCalledExactlyOnceWith(f.installation);
+});
+
+it("propagates raw family contention after exactly one adoption attempt", async () => {
+  const f = fixture(),
+    error = Object.assign(Error("family writer busy after adoption started"), {
+      code: "EEXIST",
+      syscall: "mkdir",
+      path: "/synthetic/.family-writer",
+    });
+  mock.adoptArtifactTargetLease.mockRejectedValueOnce(error);
+  await expect(openConfigurationHandoffChild(f.input)).rejects.toBe(error);
+  expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(1);
+  expect(mock.retainAdmittedControlExecution).not.toHaveBeenCalled();
+  expect(mock.closeControlInstallation).toHaveBeenCalledExactlyOnceWith(f.installation);
+});
+
+it("does not retry any failure after the one-use execution marker", async () => {
+  const f = fixture(),
+    error = Error("one-use marker publication refused");
+  mock.beginFixedCliExecution.mockImplementation(() => {
+    throw error;
+  });
+  await expect(openConfigurationHandoffChild(f.input)).rejects.toBe(error);
+  expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(1);
+  expect(mock.beginFixedCliExecution).toHaveBeenCalledTimes(1);
+  expect(mock.closeArtifactTargetLeaseHandle).toHaveBeenCalledExactlyOnceWith(f.lease);
+  expect(mock.closeControlInstallation).toHaveBeenCalledExactlyOnceWith(f.installation);
+});
+
+it("keeps the existing raw family-writer wait limited to source reads", async () => {
+  const f = fixture(),
+    error = Object.assign(Error("source writer busy"), {
+      code: "EEXIST",
+      syscall: "mkdir",
+      path: "/synthetic/.family-writer",
+    });
+  mock.readConfigurationHandoffSpawnIntent.mockImplementationOnce(() => {
+    throw error;
+  });
+  const pending = openConfigurationHandoffChild(f.input);
+  void pending.catch(() => {});
+  await vi.advanceTimersByTimeAsync(50);
+  closeConfigurationHandoffChild(await pending);
+  expect(mock.readConfigurationHandoffSpawnIntent).toHaveBeenCalledTimes(2);
+  expect(mock.adoptArtifactTargetLease).toHaveBeenCalledTimes(1);
+});

apps/ops-agent/src/configuration-handoff-control.ts

@@ -4,6 +4,7 @@ import { existsSync } from "node:fs";
 import { basename, join } from "node:path";
 import { setTimeout as delay } from "node:timers/promises";
 import {
+  ArtifactTargetAdoptionObservationBusyError,
   ArtifactTargetJournalBusyError,
   adoptArtifactTargetLease,
   assertArtifactTargetLeaseExecution,
@@ -50,23 +51,30 @@ const children = new WeakMap<
     intent: ConfigurationHandoffIntentRecord;
   }
 >();
-/** Parent terminal/copy publication can briefly own the authenticated source
- * index. Wait only around source reads, before adoption/forward writes. The
- * original deadline remains authoritative; no lock is removed or stolen. */
-async function waitForHandoffSource<T>(execution: ExecutionRef, read: () => T): Promise<T> {
+function sourceReadBusy(error: unknown): boolean {
+  const io = error as NodeJS.ErrnoException | undefined;
+  return (
+    error instanceof ArtifactTargetJournalBusyError ||
+    (io?.code === "EEXIST" &&
+      io.syscall === "mkdir" &&
+      typeof io.path === "string" &&
+      basename(io.path) === ".family-writer")
+  );
+}
+/** Retry only the caller's certified pre-write contention. The original
+ * deadline remains authoritative; no lock is removed or stolen. */
+async function waitForHandoffContention<T>(
+  execution: ExecutionRef,
+  attempt: () => T | Promise<T>,
+  canRetry: (error: unknown) => boolean,
+): Promise<T> {
   const started = performance.now();
   for (;;) {
     try {
-      return read();
+      return await attempt();
     } catch (error) {
-      const io = error as NodeJS.ErrnoException;
-      const familyBusy =
-        io.code === "EEXIST" &&
-        io.syscall === "mkdir" &&
-        !!io.path &&
-        basename(io.path) === ".family-writer";
       if (
-        (!familyBusy && !(error instanceof ArtifactTargetJournalBusyError)) ||
+        !canRetry(error) ||
         Date.now() >= execution.deadlineAt ||
         performance.now() - started >= 5000
       )
@@ -99,8 +107,10 @@ export async function openConfigurationHandoffChild(input: {
   let lease: BorrowedArtifactTargetLease | undefined;
   try {
     const intent = immutable(
-      await waitForHandoffSource(execution, () =>
-        readConfigurationHandoffSpawnIntent(installation, execution, input.handoffId),
+      await waitForHandoffContention(
+        execution,
+        () => readConfigurationHandoffSpawnIntent(installation, execution, input.handoffId),
+        sourceReadBusy,
       ),
     );
     if (
@@ -116,27 +126,39 @@ export async function openConfigurationHandoffChild(input: {
         handoffId: input.handoffId,
         intentDigest: controlDigest(intent.operation),
       },
-      { peer, parent } = await waitForHandoffSource(execution, () => ({
-        peer: verifyFixedCliControlPeer(installation, {
-          assertion: input.assertion,
-          execution,
-          kind: body.kind,
-          recipient: "configuration-helper",
+      { peer, parent } = await waitForHandoffContention(
+        execution,
+        () => ({
+          peer: verifyFixedCliControlPeer(installation, {
+            assertion: input.assertion,
+            execution,
+            kind: body.kind,
+            recipient: "configuration-helper",
+          }),
+          parent: verifiedParentForArtifactDelegation(installation, input.assertion, execution),
         }),
-        parent: verifiedParentForArtifactDelegation(installation, input.assertion, execution),
-      }));
+        sourceReadBusy,
+      );
     if (
       canonicalJson(parent.operationKey) !== canonicalJson(intent.preparer.operationKey) ||
       parent.fingerprint !== intent.preparer.fingerprint
     )
       throw Error("Original helper preparer required");
-    lease = await adoptArtifactTargetLease({
-      dataRoot: join(infraDir, "data"),
-      assertion: input.assertion,
-      peer,
+    lease = await waitForHandoffContention(
       execution,
-      serviceIds: intent.operation.serviceIds,
-    });
+      () =>
+        adoptArtifactTargetLease({
+          dataRoot: join(infraDir, "data"),
+          assertion: input.assertion,
+          peer,
+          execution,
+          serviceIds: intent.operation.serviceIds,
+        }),
+      // A raw family-writer error may follow partial adoption and is never retried.
+      (error) =>
+        error instanceof ArtifactTargetAdoptionObservationBusyError ||
+        error instanceof ArtifactTargetJournalBusyError,
+    );
     const authority = retainAdmittedControlExecution(installation, {
       execution,
       kind: body.kind,
@@ -173,8 +195,10 @@ export async function reconcileCompletedConfigurationHandoff(input: {
       container: existsSync("/run/secrets/openmapx-control-verifiers"),
     });
   try {
-    const intent = await waitForHandoffSource(execution, () =>
-        readConfigurationHandoffSpawnIntent(installation, execution, input.handoffId),
+    const intent = await waitForHandoffContention(
+        execution,
+        () => readConfigurationHandoffSpawnIntent(installation, execution, input.handoffId),
+        sourceReadBusy,
       ),
       state = installationState(installation),
       file = configurationHandoffRecordName(input.handoffId, "completion");
@@ -192,13 +216,16 @@ export async function reconcileCompletedConfigurationHandoff(input: {
       raw,
       configurationHandoffWorkerRecordSchema,
     );
-    await waitForHandoffSource(execution, () =>
-      verifyFamilyOwnedRecord(
-        state,
-        intent.origin,
-        { scope: "private", namespace: "observations", container: "", name: file },
-        canonicalJson(raw),
-      ),
+    await waitForHandoffContention(
+      execution,
+      () =>
+        verifyFamilyOwnedRecord(
+          state,
+          intent.origin,
+          { scope: "private", namespace: "observations", container: "", name: file },
+          canonicalJson(raw),
+        ),
+      sourceReadBusy,
     );
     if (worker.phase !== "applied" || canonicalJson(worker.execution) !== canonicalJson(execution))
       throw Error("Unfinished original helper remains fenced");