fix(ops-agent): release a traffic maintenance fence its own restart left behind

apps/ops-agent/src/anchored-directory.test.ts

@@ -142,3 +142,36 @@ describe.runIf(process.platform === "linux")("anchored directory descriptors", (
     ).toBe("{}\n");
   });
 });
+
+describe.each(modes)("anchored directory reads (%s)", (_mode, descriptorAnchorRoot) => {
+  it("reads a bounded regular file and reports an absent one", () => {
+    const { data, shared } = fixture();
+    writeFileSync(join(shared, ".traffic-maintenance.json"), "{}\n");
+    writeFileSync(join(shared, "large.json"), "x".repeat(65));
+    const directory = openAnchoredDirectory(data, ["valhalla", "osm-pbf"], {
+      descriptorAnchorRoot,
+    });
+    try {
+      expect(directory.readFile(".traffic-maintenance.json", 64)).toBe("{}\n");
+      expect(directory.readFile("absent.json", 64)).toBeNull();
+      expect(() => directory.readFile("large.json", 64)).toThrow("size limit");
+    } finally {
+      directory.close();
+    }
+  });
+
+  it("refuses a symlink or directory instead of reading through it", () => {
+    const { data, shared, sentinel } = fixture();
+    symlinkSync(sentinel, join(shared, ".traffic-maintenance.json"));
+    mkdirSync(join(shared, "nested"));
+    const directory = openAnchoredDirectory(data, ["valhalla", "osm-pbf"], {
+      descriptorAnchorRoot,
+    });
+    try {
+      expect(() => directory.readFile(".traffic-maintenance.json", 64)).toThrow();
+      expect(() => directory.readFile("nested", 64)).toThrow();
+    } finally {
+      directory.close();
+    }
+  });
+});

apps/ops-agent/src/anchored-directory.ts

@@ -8,6 +8,7 @@
   lstatSync,
   mkdirSync,
   openSync,
+  readSync,
   renameSync,
   unlinkSync,
   writeFileSync,
@@ -33,6 +34,11 @@
   createFile(name: string, contents: string, mode: number): void;
   /** Remove `name` if present; a symlink there is removed, not followed. */
   removeFile(name: string): void;
+  /**
+   * Read `name` if it is a regular file of at most `maxBytes`; null when absent.
+   * A symlink or any other kind of entry there is refused, not followed.
+   */
+  readFile(name: string, maxBytes: number): string | null;
   close(): void;
 }
 
@@ -143,6 +149,34 @@
           if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
         }
       },
+      readFile(name, maxBytes) {
+        let descriptor: number;
+        try {
+          descriptor = openSync(
+            entry(name),
+            constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK,
+          );
+        } catch (error) {
+          if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+          throw error;
+        }
+        try {
+          const metadata = fstatSync(descriptor);
+          if (!metadata.isFile()) throw new Error("Unsafe anchored file");
+          // Bounded by the read itself: another writer can grow the file after fstat.
+          const buffer = Buffer.alloc(maxBytes + 1);
+          let length = 0;
+          while (length < buffer.length) {
+            const read = readSync(descriptor, buffer, length, buffer.length - length, null);
+            if (read === 0) break;
+            length += read;
+          }
+          if (length > maxBytes) throw new Error("Anchored file exceeds its size limit");
+          return buffer.toString("utf8", 0, length);
+        } finally {
+          closeSync(descriptor);
+        }
+      },
       close() {
         for (const descriptor of descriptors.splice(0).reverse()) closeSync(descriptor);
       },

apps/ops-agent/src/docker-runtime.test.ts

@@ -14,7 +14,12 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { OpsOperation } from "@openmapx/core/ops";
 import { describe, expect, it, vi } from "vitest";
-import { createDockerRuntime, runContainedLogProcess, runContainedProcess } from "./docker-runtime";
+import {
+  createDockerRuntime,
+  recoverInterruptedTrafficMaintenance,
+  runContainedLogProcess,
+  runContainedProcess,
+} from "./docker-runtime";
 import { dispatchOpsOperation } from "./runtime";
 
 const context = () => ({
@@ -2028,3 +2033,209 @@ while :; do printf "WRITE %s\\n" "$name" >&3; sleep 0.1; done
     12_000,
   );
 });
+
+describe("recoverInterruptedTrafficMaintenance", () => {
+  const fingerprint = "b".repeat(64);
+  const fence = `${JSON.stringify({
+    schemaVersion: 1,
+    operationId: fingerprint,
+    plan: "rebuild-extract",
+    startedAt: "2026-10-30T15:44:08.055Z",
+  })}\n`;
+  const job = (
+    state: "failed" | "succeeded",
+    errorClass: "recovery_required" | "runtime" | undefined,
+    submittedAt = "2026-10-30T15:44:08.048Z",
+  ) => ({
+    operation: { kind: "valhalla.traffic.maintain", plan: "rebuild-extract" } as OpsOperation,
+    fingerprint,
+    state,
+    ...(errorClass ? { errorClass } : {}),
+    submittedAt,
+  });
+  const withDataRoot = async (
+    files: Record<string, string>,
+    body: (dataRoot: string, shared: string) => Promise<void>,
+  ) => {
+    const dataRoot = mkdtempSync(join(tmpdir(), "valhalla-fence-recovery-"));
+    const shared = join(dataRoot, "valhalla", "osm-pbf");
+    mkdirSync(shared, { recursive: true });
+    for (const [name, contents] of Object.entries(files)) {
+      writeFileSync(join(shared, name), contents);
+    }
+    try {
+      await body(dataRoot, shared);
+    } finally {
+      rmSync(dataRoot, { recursive: true, force: true });
+    }
+  };
+  const docker =
+    (calls: string[][], answers: { remaining?: string; running?: string } = {}) =>
+    async (_file: string, args: readonly string[]) => {
+      calls.push([...args]);
+      if (args[0] === "ps") return { stdout: answers.remaining ?? "", stderr: "" };
+      if (args[0] === "inspect") return { stdout: answers.running ?? "false\n", stderr: "" };
+      return { stdout: "", stderr: "" };
+    };
+
+  it("releases the fence of a job the agent was stopped in", async () => {
+    await withDataRoot(
+      {
+        ".traffic-maintenance.json": fence,
+        "traffic-generations.json": "{}\n",
+        "traffic-generations.unverified.json": "{}\n",
+      },
+      async (dataRoot, shared) => {
+        const calls: string[][] = [];
+        await expect(
+          recoverInterruptedTrafficMaintenance({
+            trafficDataRoot: dataRoot,
+            journalRecords: [
+              job("succeeded", undefined, "2026-10-29T05:00:00.000Z"),
+              job("failed", "recovery_required"),
+            ],
+            execFile: docker(calls),
+            descriptorAnchorRoot: null,
+          }),
+        ).resolves.toEqual({
+          state: "released",
+          operationId: fingerprint,
+          plan: "rebuild-extract",
+        });
+
+        expect(calls).toEqual([
+          ["rm", "-f", `openmapx-valhalla-maint-${"b".repeat(16)}`],
+          [
+            "ps",
+            "-a",
+            "--filter",
+            `name=^/openmapx-valhalla-maint-${"b".repeat(16)}$`,
+            "--format",
+            "{{.Names}}",
+          ],
+          ["stop", "--time", "1", "docker-valhalla-1"],
+          ["inspect", "--format", "{{.State.Running}}", "docker-valhalla-1"],
+        ]);
+        expect(calls.some((call) => call[0] === "start")).toBe(false);
+        expect(readdirSync(shared).sort()).toEqual([
+          `.traffic-maintenance.interrupted-${"b".repeat(16)}.json`,
+        ]);
+        expect(
+          readFileSync(
+            join(shared, `.traffic-maintenance.interrupted-${"b".repeat(16)}.json`),
+            "utf8",
+          ),
+        ).toBe(fence);
+      },
+    );
+  });
+
+  it("keeps the fence of a job that failed on its own", async () => {
+    await withDataRoot(
+      { ".traffic-maintenance.json": fence, "traffic-generations.json": "{}\n" },
+      async (dataRoot, shared) => {
+        const calls: string[][] = [];
+        await expect(
+          recoverInterruptedTrafficMaintenance({
+            trafficDataRoot: dataRoot,
+            journalRecords: [
+              job("failed", "recovery_required", "2026-10-28T05:00:00.000Z"),
+              job("failed", "runtime"),
+            ],
+            execFile: docker(calls),
+            descriptorAnchorRoot: null,
+          }),
+        ).resolves.toEqual({ state: "kept", reason: "its job failed on its own" });
+        expect(calls).toEqual([]);
+        expect(existsSync(join(shared, ".traffic-maintenance.json"))).toBe(true);
+        expect(existsSync(join(shared, "traffic-generations.json"))).toBe(true);
+      },
+    );
+  });
+
+  it("keeps a fence written long after the interrupted job it shares a fingerprint with", async () => {
+    await withDataRoot({ ".traffic-maintenance.json": fence }, async (dataRoot, shared) => {
+      const calls: string[][] = [];
+      await expect(
+        recoverInterruptedTrafficMaintenance({
+          trafficDataRoot: dataRoot,
+          // The fence's own (failed, since pruned) job is gone; an older
+          // interrupted rebuild with the same fingerprint is retained.
+          journalRecords: [job("failed", "recovery_required", "2026-10-20T05:00:00.000Z")],
+          execFile: docker(calls),
+          descriptorAnchorRoot: null,
+        }),
+      ).resolves.toEqual({ state: "kept", reason: "the journal holds no job for the fence" });
+      expect(calls).toEqual([]);
+      expect(existsSync(join(shared, ".traffic-maintenance.json"))).toBe(true);
+    });
+  });
+
+  it("keeps the fence when the journal holds no job for it", async () => {
+    await withDataRoot({ ".traffic-maintenance.json": fence }, async (dataRoot, shared) => {
+      await expect(
+        recoverInterruptedTrafficMaintenance({
+          trafficDataRoot: dataRoot,
+          journalRecords: [],
+          execFile: docker([]),
+          descriptorAnchorRoot: null,
+        }),
+      ).resolves.toEqual({ state: "kept", reason: "the journal holds no job for the fence" });
+      expect(existsSync(join(shared, ".traffic-maintenance.json"))).toBe(true);
+    });
+  });
+
+  it.each([
+    ["its maintenance container survives removal", { remaining: "openmapx-valhalla-maint-x\n" }],
+    ["Valhalla still runs", { running: "true\n" }],
+  ])("keeps the fence when %s", async (_case, answers) => {
+    await withDataRoot(
+      { ".traffic-maintenance.json": fence, "traffic-generations.json": "{}\n" },
+      async (dataRoot, shared) => {
+        await expect(
+          recoverInterruptedTrafficMaintenance({
+            trafficDataRoot: dataRoot,
+            journalRecords: [job("failed", "recovery_required")],
+            execFile: docker([], answers),
+            descriptorAnchorRoot: null,
+          }),
+        ).resolves.toMatchObject({ state: "kept" });
+        expect(existsSync(join(shared, ".traffic-maintenance.json"))).toBe(true);
+        expect(existsSync(join(shared, "traffic-generations.json"))).toBe(true);
+      },
+    );
+  });
+
+  it("does nothing without a fence", async () => {
+    await withDataRoot({ "traffic-generations.json": "{}\n" }, async (dataRoot, shared) => {
+      const calls: string[][] = [];
+      await expect(
+        recoverInterruptedTrafficMaintenance({
+          trafficDataRoot: dataRoot,
+          journalRecords: [job("failed", "recovery_required")],
+          execFile: docker(calls),
+          descriptorAnchorRoot: null,
+        }),
+      ).resolves.toEqual({ state: "none" });
+      expect(calls).toEqual([]);
+      expect(existsSync(join(shared, "traffic-generations.json"))).toBe(true);
+    });
+  });
+
+  it("refuses a fence replaced by a symlink", async () => {
+    await withDataRoot({}, async (dataRoot, shared) => {
+      const outside = join(dataRoot, "outside.json");
+      writeFileSync(outside, fence);
+      symlinkSync(outside, join(shared, ".traffic-maintenance.json"));
+      await expect(
+        recoverInterruptedTrafficMaintenance({
+          trafficDataRoot: dataRoot,
+          journalRecords: [job("failed", "recovery_required")],
+          execFile: docker([]),
+          descriptorAnchorRoot: null,
+        }),
+      ).rejects.toThrow();
+      expect(lstatSync(join(shared, ".traffic-maintenance.json")).isSymbolicLink()).toBe(true);
+    });
+  });
+});