From ead7b3d0d6bcb8cb727455d1732987a5b9012e26 Mon Sep 17 00:00:00 2001 From: Florian Date: Sat, 31 Oct 2026 02:54:25 +0100 Subject: [PATCH] fix(ops-agent): release a traffic maintenance fence its own restart left behind When the operations agent was stopped during `valhalla.traffic.maintain` (an update recreating it, a crash, a reboot), the journal failed the job as `recovery_required`, but its fence stayed. Valhalla stayed stopped and data-manager's supervisor kept its worker paused without a word, and that worker is the only one that asks for the next maintenance. At start the agent now matches the fence to its journal job. When that job was interrupted, it removes the job's tool container, confirms Valhalla is stopped, withdraws the generation manifests and keeps the fence as `.traffic-maintenance.interrupted-.json`; the worker then requests a fresh, coordinated maintenance. A job that failed on its own keeps its fence for an operator. The supervisor now logs why its worker is paused, warns while a pause outlasts any maintenance, and logs the resume. The monitoring guide describes these pauses and the manual recovery after a failed maintenance. --- apps/ops-agent/src/anchored-directory.test.ts | 33 +++ apps/ops-agent/src/anchored-directory.ts | 34 +++ apps/ops-agent/src/docker-runtime.test.ts | 213 +++++++++++++++++- apps/ops-agent/src/docker-runtime.ts | 139 ++++++++++++ apps/ops-agent/src/index.ts | 25 +- docs/docs/administration/monitoring.md | 41 ++++ .../road-condition-engine-verification.md | 2 +- .../src/__tests__/traffic-supervision.test.ts | 71 +++++- .../src/jobs/traffic/supervision.ts | 64 ++++++ services/data-manager/src/supervisor.ts | 21 +- 10 files changed, 636 insertions(+), 7 deletions(-) diff --git a/apps/ops-agent/src/anchored-directory.test.ts b/apps/ops-agent/src/anchored-directory.test.ts index 7608805c5..46516d4cd 100644 --- a/apps/ops-agent/src/anchored-directory.test.ts +++ b/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(); + } + }); +}); diff --git a/apps/ops-agent/src/anchored-directory.ts b/apps/ops-agent/src/anchored-directory.ts index 277ff41df..5793755dd 100644 --- a/apps/ops-agent/src/anchored-directory.ts +++ b/apps/ops-agent/src/anchored-directory.ts @@ -8,6 +8,7 @@ import { lstatSync, mkdirSync, openSync, + readSync, renameSync, unlinkSync, writeFileSync, @@ -33,6 +34,11 @@ export interface AnchoredDirectory { 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 @@ export function openAnchoredDirectory( 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); }, diff --git a/apps/ops-agent/src/docker-runtime.test.ts b/apps/ops-agent/src/docker-runtime.test.ts index 65bbd6b6b..ecd0a8366 100644 --- a/apps/ops-agent/src/docker-runtime.test.ts +++ b/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, + body: (dataRoot: string, shared: string) => Promise, + ) => { + 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); + }); + }); +}); diff --git a/apps/ops-agent/src/docker-runtime.ts b/apps/ops-agent/src/docker-runtime.ts index e159e4219..28f509b25 100644 --- a/apps/ops-agent/src/docker-runtime.ts +++ b/apps/ops-agent/src/docker-runtime.ts @@ -4,6 +4,7 @@ import { readFile } from "node:fs/promises"; import { join } from "node:path"; import { StringDecoder } from "node:string_decoder"; import { + OPS_KIND_POLICIES, OPS_MAX_EVENT_MESSAGE_BYTES, OPS_MAX_FOLLOW_LOG_EVENTS, type OpsResultFor, @@ -14,6 +15,7 @@ import { services, } from "@openmapx/core/server"; import { openAnchoredDirectory } from "./anchored-directory"; +import type { PersistedOpsJob } from "./journal"; import { createUnavailableRuntime, type OpsRuntime } from "./runtime"; const TRAFFIC_MAINTENANCE_FENCE = ".traffic-maintenance.json"; @@ -326,6 +328,143 @@ function parseStackStatus(stdout: string): OpsResultFor<"stack.status"> { }; } +const TRAFFIC_MAINTENANCE_FENCE_MAX_BYTES = 4096; +// A fence written later than its job's deadline after the job's submission is +// not that job's. +const TRAFFIC_MAINTENANCE_JOB_WINDOW_MS = OPS_KIND_POLICIES["valhalla.traffic.maintain"].timeoutMs; + +export interface TrafficMaintenanceRecoveryOptions { + /** Host-side data root shared with data-manager and Valhalla. */ + trafficDataRoot: string; + /** The job journal as opened, which has already failed every interrupted job. */ + journalRecords: readonly Pick< + PersistedOpsJob, + "operation" | "fingerprint" | "state" | "errorClass" | "submittedAt" + >[]; + execFile?: ExecFile; + descriptorAnchorRoot?: string | null; +} + +export type TrafficMaintenanceRecovery = + | { state: "none" } + | { state: "released"; operationId: string; plan: string } + | { state: "kept"; reason: string }; + +/** + * Releases the durable maintenance fence a `valhalla.traffic.maintain` job + * left behind when the agent itself was stopped mid-job (an update recreating + * it, a crash, a reboot). Nothing else would: the fence keeps data-manager's + * worker quiesced, and only that worker asks for the next maintenance. + * + * Only an interrupted job is released. A job that failed on its own keeps its + * fence for an operator, since retrying a broken graph would only fail again. + * The fence names its job by fingerprint; the latest maintenance job with that + * fingerprint must have failed as `recovery_required`. Before releasing, the + * job's tool container must be gone and Valhalla stopped. The generation + * manifests are withdrawn, so the next inspection reports `not_ready` and the + * worker requests a fresh, coordinated maintenance. The fence is kept beside + * them as `.traffic-maintenance.interrupted-.json`. + */ +export async function recoverInterruptedTrafficMaintenance( + options: TrafficMaintenanceRecoveryOptions, +): Promise { + const execFile: ExecFile = + options.execFile ?? + ((file, args, execOptions) => + runContainedProcess(file, args, { ...execOptions, killGraceMs: KILL_GRACE_MS })); + const run = (args: readonly string[], timeout: number) => + execFile("docker", args, { + signal: AbortSignal.timeout(timeout), + timeout, + maxBuffer: MAX_OUTPUT_BYTES, + }); + + if (!existsSync(join(options.trafficDataRoot, "valhalla", "osm-pbf"))) return { state: "none" }; + const shared = openAnchoredDirectory(options.trafficDataRoot, ["valhalla", "osm-pbf"], { + ...(options.descriptorAnchorRoot !== undefined + ? { descriptorAnchorRoot: options.descriptorAnchorRoot } + : {}), + }); + try { + const raw = shared.readFile(TRAFFIC_MAINTENANCE_FENCE, TRAFFIC_MAINTENANCE_FENCE_MAX_BYTES); + if (raw === null) return { state: "none" }; + let fence: { operationId?: unknown; plan?: unknown; startedAt?: unknown }; + try { + fence = JSON.parse(raw) as { operationId?: unknown; plan?: unknown; startedAt?: unknown }; + } catch { + return { state: "kept", reason: "the fence is not readable" }; + } + const operationId = fence.operationId; + if (typeof operationId !== "string" || !/^[0-9a-f]{64}$/.test(operationId)) { + return { state: "kept", reason: "the fence names no maintenance job" }; + } + const job = options.journalRecords + .filter( + (record) => + record.operation.kind === "valhalla.traffic.maintain" && + record.fingerprint === operationId, + ) + .sort((a, b) => Date.parse(b.submittedAt) - Date.parse(a.submittedAt))[0]; + if (!job) return { state: "kept", reason: "the journal holds no job for the fence" }; + // Every rebuild shares one fingerprint, and only interrupted jobs outlive + // the journal's retention: the fence must also have been written while + // this job could run, or it belongs to a later job already pruned. + const fenceStartedAt = typeof fence.startedAt === "string" ? Date.parse(fence.startedAt) : NaN; + const submittedAt = Date.parse(job.submittedAt); + if ( + !Number.isFinite(fenceStartedAt) || + fenceStartedAt < submittedAt || + fenceStartedAt > submittedAt + TRAFFIC_MAINTENANCE_JOB_WINDOW_MS + ) { + return { state: "kept", reason: "the journal holds no job for the fence" }; + } + if (job.state !== "failed" || job.errorClass !== "recovery_required") { + return { + state: "kept", + reason: `its job ${job.state === "failed" ? "failed on its own" : `is ${job.state}`}`, + }; + } + + const maintenanceContainer = `openmapx-valhalla-maint-${operationId.slice(0, 16)}`; + await run(["rm", "-f", maintenanceContainer], 30_000).catch(() => undefined); + const remaining = await run( + ["ps", "-a", "--filter", `name=^/${maintenanceContainer}$`, "--format", "{{.Names}}"], + 10_000, + ).catch(() => null); + if (remaining?.stdout.trim() !== "") { + return { state: "kept", reason: "its maintenance container could not be removed" }; + } + await run(["stop", "--time", "1", VALHALLA_CONTAINER], 30_000).catch(() => undefined); + const serving = await run( + ["inspect", "--format", "{{.State.Running}}", VALHALLA_CONTAINER], + 10_000, + ).catch(() => null); + if (serving?.stdout.trim() !== "false") { + return { state: "kept", reason: "Valhalla could not be confirmed stopped" }; + } + + shared.replaceFile( + `.traffic-maintenance.interrupted-${operationId.slice(0, 16)}.json`, + raw, + 0o644, + ); + shared.removeFile(TRAFFIC_GENERATIONS_MANIFEST); + shared.removeFile(TRAFFIC_GENERATIONS_UNVERIFIED); + shared.removeFile(TRAFFIC_MAINTENANCE_FENCE); + return { + state: "released", + operationId, + // Written where data-manager can write too: only a plan-shaped value is echoed. + plan: + typeof fence.plan === "string" && /^[a-z][a-z-]{0,63}$/.test(fence.plan) + ? fence.plan + : "unknown", + }; + } finally { + shared.close(); + } +} + export function createDockerRuntime(options: DockerRuntimeOptions): OpsRuntime { const execFile: ExecFile = options.execFile ?? diff --git a/apps/ops-agent/src/index.ts b/apps/ops-agent/src/index.ts index c7b77f62c..ed5efb5c8 100644 --- a/apps/ops-agent/src/index.ts +++ b/apps/ops-agent/src/index.ts @@ -25,7 +25,11 @@ import { createDockerDawarichRuntimeInspector, } from "./dawarich-subject-export"; import { describeDeploymentUserAtStart } from "./deployment-user-at-start"; -import { createDockerRuntime, runContainedProcess } from "./docker-runtime"; +import { + createDockerRuntime, + recoverInterruptedTrafficMaintenance, + runContainedProcess, +} from "./docker-runtime"; import { IntegrationStore, installIntegrationRuntime } from "./integration-store"; import { openOpsJobJournal } from "./journal"; import { importLegacyAgentState, retireLegacyAgentState } from "./legacy-state"; @@ -284,6 +288,25 @@ async function main(): Promise { retry.unref(); } const journal = await openOpsJobJournal(config.journalFile); + // A maintenance job this agent was stopped in left its fence behind, and + // with it a quiesced data-manager worker that would never ask again. + try { + const recovery = await recoverInterruptedTrafficMaintenance({ + trafficDataRoot: join(paths.infraDir, "data"), + journalRecords: journal.records(), + }); + if (recovery.state === "released") { + process.stderr.write( + `Released the Valhalla traffic maintenance fence of interrupted job ${recovery.operationId.slice(0, 16)} (${recovery.plan}); data-manager requests a fresh maintenance\n`, + ); + } else if (recovery.state === "kept") { + process.stderr.write( + `Kept the Valhalla traffic maintenance fence: ${recovery.reason}. Valhalla stays stopped and data-manager's traffic worker paused; recover as described under "Graph maintenance pauses" in the monitoring guide\n`, + ); + } + } catch { + process.stderr.write("Could not check the Valhalla traffic maintenance fence\n"); + } // After a self-update this agent starts before anything has rendered with // its code, so the state directories it opens may not exist yet. await services.withConfigurationGenerationWriter(paths.infraDir, async () => diff --git a/docs/docs/administration/monitoring.md b/docs/docs/administration/monitoring.md index 7db69b996..10721a1f1 100644 --- a/docs/docs/administration/monitoring.md +++ b/docs/docs/administration/monitoring.md @@ -332,6 +332,8 @@ Open Data Manager's **Logs** drawer or run | `conditions classification failed, writing no overrides` | Classification failure; do not infer closures from live-speed success | | `span cache save failed` | Persistence permissions; a restart will lose unsaved cache state | | `skipped out-of-range edges (traffic.tar/waysToEdges mismatch)` | Rebuild the graph-associated traffic/map artifacts through the normal maintenance workflow | +| `Traffic worker paused for Valhalla maintenance` | Graph maintenance is running; Valhalla and the traffic worker resume when it ends | +| `Traffic worker still paused for Valhalla maintenance job` | The maintenance is no longer running: see [Graph maintenance pauses](#graph-maintenance-pauses) | ### Why a closure is not applied @@ -369,6 +371,45 @@ start: restored`, or `changed`, `invalid` or `unverifiable` before a rebuild), `endingBeforeRenewal` counts conditions left out because they end before the next lease renewal. +### Graph maintenance pauses + +Rebuilding the traffic extract or applying predicted traffic is graph +maintenance. The operations agent stops Valhalla, writes +`infra/docker/data/valhalla/osm-pbf/.traffic-maintenance.json`, runs the +Valhalla tools and starts Valhalla again. While that fence file exists, Data +Manager keeps its service stopped (its health check fails) and logs +`Traffic worker paused for Valhalla maintenance`. A rebuild takes a few +minutes. + +If the agent itself is stopped during maintenance (an update recreating it, a +crash or a reboot), its next start releases the fence and logs +`Released the Valhalla traffic maintenance fence of interrupted job`. Data +Manager then requests a fresh maintenance, which starts Valhalla again. + +Maintenance that failed on its own keeps the fence, and Valhalla stays stopped. +The agent's next start logs `Kept the Valhalla traffic maintenance fence` and +Data Manager repeats `Traffic worker still paused` every 30 minutes. Car, +walking and cycling routes served by Valhalla fail or fall back to another +router meanwhile. To recover: + +1. Find the failure in `pnpm openmapx services logs ops-agent` (the + `valhalla.traffic.maintain` entry) and repair its cause, such as a full disk + or a damaged graph. +2. Move the fence and the generation manifests aside, so Data Manager requests + a full rebuild: + + ```bash + cd infra/docker/data/valhalla/osm-pbf + mv .traffic-maintenance.json .traffic-maintenance.failed.json + for f in traffic-generations.json traffic-generations.unverified.json; do + [ -e "$f" ] && mv "$f" "$f.withdrawn" + done + ``` + +3. Within seconds Data Manager logs `Traffic worker resumed` and + `traffic-extract: building`. Valhalla is healthy again once the agent logs + the `valhalla.traffic.maintain` result `success`. + ### Closures on a planned trip A car trip with a departure or arrival time cannot use the live-traffic diff --git a/services/data-manager/docs/road-condition-engine-verification.md b/services/data-manager/docs/road-condition-engine-verification.md index bc4ba8241..02d27c46b 100644 --- a/services/data-manager/docs/road-condition-engine-verification.md +++ b/services/data-manager/docs/road-condition-engine-verification.md @@ -69,7 +69,7 @@ docker rm -f openmapx-watchdog-probe python3 -B -m unittest discover -s services/valhalla/tests ``` -Graph activation is separate from event freshness. `traffic-generations.json` must identify the same graph, extract and raw way-map generation; `traffic-engine.json` supplies the actual serving boot epoch. `.traffic-maintenance.json` quiesces the DM worker. A traced plan is rejected if its epoch changed before the locked write. An uncoordinated scripted-image restart (a recreate, crash or host reboot) withdraws the graph manifest as `traffic-generations.unverified.json` before announcing the new boot, since the image may rebuild base tiles; once the engine serves, the watchdog hashes the tile directory exactly as maintenance does and restores the manifest only if it still names that generation. Otherwise the manifest is dropped and the extract guard re-attests the graph through coordinated maintenance; while a withdrawn manifest awaits its check, `valhalla.traffic.inspect` reports `unknown` so no rebuild is requested. Coordinated startup prevents the image from rebuilding base tiles after the authority's hash. Missing/mismatched generations withhold live effects and remain actionable operator evidence. +Graph activation is separate from event freshness. `traffic-generations.json` must identify the same graph, extract and raw way-map generation; `traffic-engine.json` supplies the actual serving boot epoch. `.traffic-maintenance.json` quiesces the DM worker. A maintenance job the agent itself was stopped in (the journal fails it as `recovery_required`) has its fence released at the agent's next start, once the named maintenance container is gone and Valhalla is confirmed stopped; the manifests are withdrawn with it, so the worker requests a fresh coordinated maintenance. A job that failed on its own keeps its fence for an operator. A traced plan is rejected if its epoch changed before the locked write. An uncoordinated scripted-image restart (a recreate, crash or host reboot) withdraws the graph manifest as `traffic-generations.unverified.json` before announcing the new boot, since the image may rebuild base tiles; once the engine serves, the watchdog hashes the tile directory exactly as maintenance does and restores the manifest only if it still names that generation. Otherwise the manifest is dropped and the extract guard re-attests the graph through coordinated maintenance; while a withdrawn manifest awaits its check, `valhalla.traffic.inspect` reports `unknown` so no rebuild is requested. Coordinated startup prevents the image from rebuilding base tiles after the authority's hash. Missing/mismatched generations withhold live effects and remain actionable operator evidence. ## Coordinated maintenance probe diff --git a/services/data-manager/src/__tests__/traffic-supervision.test.ts b/services/data-manager/src/__tests__/traffic-supervision.test.ts index d0cb819bb..654933df4 100644 --- a/services/data-manager/src/__tests__/traffic-supervision.test.ts +++ b/services/data-manager/src/__tests__/traffic-supervision.test.ts @@ -2,7 +2,10 @@ import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { recoverTrafficWriterLock } from "../jobs/traffic/supervision.js"; +import { + createMaintenancePauseReporter, + recoverTrafficWriterLock, +} from "../jobs/traffic/supervision.js"; describe("traffic writer supervision", () => { it("fences its hung worker before releasing the lock", async () => { @@ -87,3 +90,69 @@ describe("traffic writer supervision", () => { } }); }); + +describe("maintenance pause reporting", () => { + const fence = JSON.stringify({ + schemaVersion: 1, + operationId: "b".repeat(64), + plan: "rebuild-extract", + startedAt: "2026-10-30T15:44:08.055Z", + }); + const started = Date.parse("2026-10-30T15:44:08.055Z"); + + it("reports a pause once, again when it outlasts any maintenance, and the resume", () => { + const messages: string[] = []; + let clock = started + 2_000; + let reads = 0; + const reporter = createMaintenancePauseReporter({ + report: (message) => messages.push(message), + now: () => clock, + }); + const read = () => { + reads += 1; + return fence; + }; + + reporter.paused(read); + clock += 10 * 60_000; + reporter.paused(read); + expect(messages).toEqual([ + "Traffic worker paused for Valhalla maintenance rebuild-extract (job bbbbbbbbbbbbbbbb, started 2026-10-30T15:44:08.055Z); it resumes when the maintenance fence is removed", + ]); + + clock = started + 36 * 60_000; + reporter.paused(read); + clock += 60_000; + reporter.paused(read); + clock += 30 * 60_000; + reporter.paused(read); + expect(messages.filter((message) => message.includes("still paused"))).toHaveLength(2); + expect(messages[1]).toContain("longer than any maintenance runs"); + expect(reads).toBe(1); + + reporter.resumed(); + reporter.resumed(); + expect(messages.at(-1)).toBe( + "Traffic worker resumed after Valhalla maintenance job bbbbbbbbbbbbbbbb", + ); + expect(messages).toHaveLength(4); + }); + + it("reports an unreadable fence as an unknown job timed from the pause", () => { + const messages: string[] = []; + const reporter = createMaintenancePauseReporter({ + report: (message) => messages.push(message), + now: () => started, + }); + reporter.paused(() => "not json"); + expect(messages).toEqual([ + "Traffic worker paused for Valhalla maintenance (job unknown, started 2026-10-30T15:44:08.055Z); it resumes when the maintenance fence is removed", + ]); + }); + + it("stays quiet when the worker was never paused", () => { + const messages: string[] = []; + createMaintenancePauseReporter({ report: (message) => messages.push(message) }).resumed(); + expect(messages).toEqual([]); + }); +}); diff --git a/services/data-manager/src/jobs/traffic/supervision.ts b/services/data-manager/src/jobs/traffic/supervision.ts index 3d8351e27..326e9bfa1 100644 --- a/services/data-manager/src/jobs/traffic/supervision.ts +++ b/services/data-manager/src/jobs/traffic/supervision.ts @@ -60,3 +60,67 @@ export async function recoverTrafficWriterLock(options: { } await rm(lock, { recursive: true, force: true }); } + +// The agent bounds a maintenance job at 30 minutes; a pause past that (with +// slack for stopping and starting Valhalla) means the job is no longer running. +const MAINTENANCE_OVERDUE_MS = 35 * 60_000; +const MAINTENANCE_OVERDUE_REPEAT_MS = 30 * 60_000; + +/** + * Says why the supervisor keeps its worker stopped while the maintenance + * fence is present: once on pausing, again whenever the pause outlasts any + * running job, and once on resuming. Without it a fence left behind leaves the + * service silently down. + */ +export function createMaintenancePauseReporter(options: { + report: (message: string) => void; + now?: () => number; +}) { + const now = options.now ?? Date.now; + let paused: { job: string; startedAt: number; startedLabel: string } | undefined; + let lastOverdueAt: number | undefined; + return { + /** Called on every pass while the fence is present; its contents are read once per pause. */ + paused(readFence: () => string | null) { + if (!paused) { + let parsed: { operationId?: unknown; plan?: unknown; startedAt?: unknown } = {}; + try { + parsed = JSON.parse(readFence() ?? "") ?? {}; + } catch { + // Reported as an unknown job below. + } + const job = + typeof parsed.operationId === "string" && /^[0-9a-f]{64}$/.test(parsed.operationId) + ? parsed.operationId.slice(0, 16) + : "unknown"; + const plan = + typeof parsed.plan === "string" && /^[a-z][a-z-]{0,63}$/.test(parsed.plan) + ? ` ${parsed.plan}` + : ""; + const fenceStarted = + typeof parsed.startedAt === "string" ? Date.parse(parsed.startedAt) : Number.NaN; + const startedAt = Number.isFinite(fenceStarted) ? fenceStarted : now(); + paused = { job, startedAt, startedLabel: new Date(startedAt).toISOString() }; + lastOverdueAt = undefined; + options.report( + `Traffic worker paused for Valhalla maintenance${plan} (job ${job}, started ${paused.startedLabel}); it resumes when the maintenance fence is removed`, + ); + } + const current = now(); + if ( + current - paused.startedAt >= MAINTENANCE_OVERDUE_MS && + (lastOverdueAt === undefined || current - lastOverdueAt >= MAINTENANCE_OVERDUE_REPEAT_MS) + ) { + lastOverdueAt = current; + options.report( + `Traffic worker still paused for Valhalla maintenance job ${paused.job}, started ${paused.startedLabel}: longer than any maintenance runs. The operations agent releases an interrupted job when it restarts; a failed one needs an operator: see "Graph maintenance pauses" in the monitoring guide`, + ); + } + }, + resumed() { + if (!paused) return; + options.report(`Traffic worker resumed after Valhalla maintenance job ${paused.job}`); + paused = undefined; + }, + }; +} diff --git a/services/data-manager/src/supervisor.ts b/services/data-manager/src/supervisor.ts index 84dc5bf8e..04963c57d 100644 --- a/services/data-manager/src/supervisor.ts +++ b/services/data-manager/src/supervisor.ts @@ -1,10 +1,13 @@ /** Independent process: a wedged or dead poller cannot prevent traffic expiry. */ import { type ChildProcess, spawn } from "node:child_process"; import { once } from "node:events"; -import { existsSync } from "node:fs"; +import { existsSync, readFileSync } from "node:fs"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; -import { recoverTrafficWriterLock } from "./jobs/traffic/supervision.js"; +import { + createMaintenancePauseReporter, + recoverTrafficWriterLock, +} from "./jobs/traffic/supervision.js"; import { expireLiveTraffic } from "./jobs/traffic/write-live.js"; import { runOpsOperation } from "./ops-client.js"; import { entryExecArgv } from "./utils/entry-module.js"; @@ -13,6 +16,10 @@ const dataDir = process.env.DATA_DIR ?? "/data"; const statePath = join(dataDir, "traffic", "live-state.json"); const tarPath = process.env.TRAFFIC_TAR_PATH?.trim() || join(dataDir, "valhalla", "osm-pbf", "traffic.tar"); +const maintenanceFence = join(dirname(tarPath), ".traffic-maintenance.json"); +const maintenancePause = createMaintenancePauseReporter({ + report: (message) => console.error(message), +}); let worker: ChildProcess | undefined; let stopping = false; let fenced = false; @@ -62,7 +69,14 @@ process.on("SIGINT", () => { while (!stopping) { try { - if (existsSync(join(dirname(tarPath), ".traffic-maintenance.json"))) { + if (existsSync(maintenanceFence)) { + maintenancePause.paused(() => { + try { + return readFileSync(maintenanceFence, "utf8").slice(0, 4096); + } catch { + return null; // Gone again, or unreadable: reported as an unknown job. + } + }); const workerPid = worker?.pid; await stopWorker(); await recoverTrafficWriterLock({ @@ -74,6 +88,7 @@ while (!stopping) { await new Promise((resolve) => setTimeout(resolve, 2000)); continue; } + maintenancePause.resumed(); await recoverTrafficWriterLock({ statePath, workerPid: worker?.pid, stopWorker }); if (existsSync(tarPath)) await expireLiveTraffic({ tarPath, statePath }); if (!worker && !fenced) {