From 425d335351694cf65620daa04262d40050b33387 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 6 Nov 2026 09:30:05 -0800 Subject: [PATCH] refactor(hub): drop provisioner-side sidecar fences Interchange mints a new sidecar identity for every ensure and never re-ensures a generation after destroying it. Both provisioners run inside the hub and already order ensure before destroy, so the fences guarded against requests that cannot arrive. --- apps/hub/src/provisioning/docker.ts | 32 +----- apps/hub/src/sidecar/local-provisioner.ts | 118 +++------------------- apps/hub/src/sidecar/manifest-store.ts | 43 -------- apps/hub/src/sidecar/restore.ts | 29 +----- 4 files changed, 16 insertions(+), 206 deletions(-) diff --git a/apps/hub/src/provisioning/docker.ts b/apps/hub/src/provisioning/docker.ts index ded45750..aa450aee 100644 --- a/apps/hub/src/provisioning/docker.ts +++ b/apps/hub/src/provisioning/docker.ts @@ -66,11 +66,6 @@ type CreateDockerSidecarProvisionerOpts = { readonly hubWebSocketUrl?: string; }; -type Fence = { - readonly generation: number; - readonly destroyedSidecarId?: string; -}; - function isDestroyedBy( container: Container, request: DestroySidecarRequest, @@ -142,7 +137,6 @@ export function createDockerSidecarProvisioner({ throw new Error("Docker limits must be positive safe integers"); if (!/^[a-zA-Z0-9][a-zA-Z0-9_.-]*$/.test(hubId)) throw new Error("Docker hub identity is invalid"); - const fences = new Map(); const queues = new Map>(); /** Containers being created, which a listing may not show yet. */ let starting = 0; @@ -246,7 +240,7 @@ export function createDockerSidecarProvisioner({ } // A late ensure the hub stopped waiting for still finishes before the - // destroy that fences it, so the destroy removes what it created. + // destroy that follows it, so the destroy removes what it created. function serialize( allocationId: string, run: () => Promise, @@ -261,10 +255,9 @@ export function createDockerSidecarProvisioner({ request: EnsureSidecarRequest, ): Promise { request.signal?.throwIfAborted(); - const fence = fences.get(request.allocationId); const containers = await list(request.allocationId, request.signal); const observed = Math.max( - fence?.generation ?? 0, + 0, ...containers.map((container) => container.Labels[LABEL.generation]), ); if (request.generation < observed) { @@ -275,17 +268,6 @@ export function createDockerSidecarProvisioner({ retryable: false, }; } - if ( - request.generation === fence?.generation && - request.sidecarId === fence.destroyedSidecarId - ) { - return { - kind: "rejected", - code: "sidecar_destroyed", - message: `Sidecar ${request.sidecarId} was destroyed`, - retryable: false, - }; - } const existing = containers.find( (container) => container.Labels[LABEL.generation] === request.generation && @@ -307,9 +289,6 @@ export function createDockerSidecarProvisioner({ } starting += 1; try { - if (fence === undefined || request.generation > fence.generation) { - fences.set(request.allocationId, { generation: request.generation }); - } return { kind: "accepted", externalRef: await create(request) }; } finally { starting -= 1; @@ -319,13 +298,6 @@ export function createDockerSidecarProvisioner({ async function destroySidecar( request: DestroySidecarRequest, ): Promise { - const fence = fences.get(request.allocationId); - if (fence === undefined || request.generation >= fence.generation) { - fences.set(request.allocationId, { - generation: request.generation, - destroyedSidecarId: request.sidecarId, - }); - } for (const container of ( await list(request.allocationId, request.signal) ).filter((container) => isDestroyedBy(container, request))) { diff --git a/apps/hub/src/sidecar/local-provisioner.ts b/apps/hub/src/sidecar/local-provisioner.ts index 9f14ca64..380228a0 100644 --- a/apps/hub/src/sidecar/local-provisioner.ts +++ b/apps/hub/src/sidecar/local-provisioner.ts @@ -20,7 +20,6 @@ import { type HostCapacity, } from "./host-capacity.ts"; import type { - LocalSidecarFence, LocalSidecarHold, LocalSidecarManifest, LocalSidecarManifestStore, @@ -73,8 +72,6 @@ type LocalSidecarRecords = { ): Promise; /** Whether the hub still has work on the sidecar, including copies it waits for the sidecar to remove. */ hasWork(sidecarId: string): Promise; - /** Those of these allocations and probes stock may still ensure: one it ended is never ensured again. */ - stillOpen(allocationIds: readonly string[]): Promise>; /** Whether the sidecar was minted with the hub's current adapter manifest and may take work again. */ reusable(sidecarId: string): Promise; }; @@ -127,12 +124,10 @@ export function createLocalProcessSidecarProvisioner({ const dataDir = localSidecarDataDir(dataRoot); const holds = new Map(); - const fences = new Map>(); let manifest: LocalSidecarManifest | null = null; let running: ManagedProcess | null = null; const queue = createSerialQueue(); let shutdownPromise: Promise | null = null; - let stillOpen: LocalSidecarRecords["stillOpen"] | undefined; let host: SidecarProvisionerHost | undefined; // Ensure and destroy answer from the sidecar and holds restore brings back, so they wait for it; a failed restore // fails them rather than letting an ensure start a second sidecar. @@ -154,26 +149,6 @@ export function createLocalProcessSidecarProvisioner({ await manifests.writeHolds([...holds.values()]); } - function openRecords(): LocalSidecarRecords["stillOpen"] { - if (stillOpen === undefined) - throw new Error("Local sidecar provisioner has not been restored"); - return stillOpen; - } - - // The hub never repeats a destroy, so a fence only memory held would let a late ensure through after a restart. - async function persistFences(): Promise { - await manifests.writeFences( - [...fences].map(([allocationId, fence]) => ({ allocationId, ...fence })), - ); - } - - // An allocation stock ended is never ensured again, and an ensure for one is refused anyway, so its fence goes. - async function pruneFences(): Promise { - const open = await openRecords()([...fences.keys()]); - for (const allocationId of fences.keys()) - if (!open.has(allocationId)) fences.delete(allocationId); - } - // One exit takes down every deployment the sidecar hosts, so each is logged. async function trackExit( managed: ManagedProcess, @@ -273,27 +248,11 @@ export function createLocalProcessSidecarProvisioner({ ): Promise { request.signal?.throwIfAborted(); const live = holds.get(request.allocationId); - const fence = fences.get(request.allocationId); - const newest = live?.generation ?? fence?.generation; - if (newest !== undefined && newest > request.generation) { + if (live !== undefined && live.generation > request.generation) { return { kind: "rejected", code: "stale_generation", - message: `Generation ${String(request.generation)} is older than ${String(newest)}`, - retryable: false, - }; - } - // Replacement destroys the old copy at the new generation before ensuring a fresh identity at that same - // generation. Fence only the identities that destroy named. - if ( - fence !== undefined && - fence.generation === request.generation && - fence.sidecarIds.includes(request.sidecarId) - ) { - return { - kind: "rejected", - code: "generation_destroyed", - message: `Generation ${String(request.generation)} was already destroyed`, + message: `Generation ${String(request.generation)} is older than ${String(live.generation)}`, retryable: false, }; } @@ -309,18 +268,6 @@ export function createLocalProcessSidecarProvisioner({ retryable: false, }; } - // A fence lasts only while stock may ensure its allocation, so one stock ended is refused from its record. - if ( - live === undefined && - !(await openRecords()([request.allocationId])).has(request.allocationId) - ) { - return { - kind: "rejected", - code: "allocation_ended", - message: `Allocation ${request.allocationId} has ended`, - retryable: false, - }; - } const sidecar = await sidecarFor(request); if (sidecar.kind === "rejected") return sidecar; try { @@ -329,12 +276,10 @@ export function createLocalProcessSidecarProvisioner({ generation: request.generation, offeredSidecarId: request.sidecarId, }); - fences.delete(request.allocationId); await persistHolds(); - if (fence !== undefined) await persistFences(); request.signal?.throwIfAborted(); } catch (error) { - await abandonEnsure(request.allocationId, { live, fence }, sidecar); + await abandonEnsure(request.allocationId, live, sidecar); throw error; } return { @@ -350,15 +295,11 @@ export function createLocalProcessSidecarProvisioner({ // was and a sidecar minted under that identity is stopped and forgotten; the next ensure mints another. async function abandonEnsure( allocationId: string, - before: { - readonly live: LocalSidecarHold | undefined; - readonly fence: Omit | undefined; - }, + before: LocalSidecarHold | undefined, sidecar: { readonly process: ManagedProcess; readonly minted: boolean }, ): Promise { - if (before.live === undefined) holds.delete(allocationId); - else holds.set(allocationId, before.live); - if (before.fence !== undefined) fences.set(allocationId, before.fence); + if (before === undefined) holds.delete(allocationId); + else holds.set(allocationId, before); if (sidecar.minted) { await stopProcess(sidecar.process); running = null; @@ -367,7 +308,6 @@ export function createLocalProcessSidecarProvisioner({ } try { await persistHolds(); - await persistFences(); } catch (error) { log({ level: "error", @@ -386,35 +326,14 @@ export function createLocalProcessSidecarProvisioner({ request: DestroySidecarRequest, ): Promise { const live = holds.get(request.allocationId); - const fence = fences.get(request.allocationId); - const newest = live?.generation ?? fence?.generation; - // A destroy older than the allocation's hold or fence is for a copy a newer generation already replaced. - const stale = newest !== undefined && newest > request.generation; - const hosted = - manifest !== null && - !stale && - (live !== undefined || - (fence?.hosted === true && fence.generation === request.generation)); - if (!stale) { - const earlier = - fence?.generation === request.generation ? fence.sidecarIds : []; - const sidecarIds = [ - ...new Set([ - ...earlier, - request.sidecarId, - ...(live === undefined ? [] : [live.offeredSidecarId]), - ]), - ]; - fences.set(request.allocationId, { - generation: request.generation, - sidecarIds, - ...(hosted && { hosted }), - }); + // A destroy older than the allocation's hold is for a copy a newer generation already replaced. + const released = + live !== undefined && live.generation <= request.generation; + if (released) { holds.delete(request.allocationId); - await pruneFences(); - await persistFences(); - if (live !== undefined) await persistHolds(); + await persistHolds(); } + const hosted = manifest !== null && released; return { kind: "destroyed", cleanup: @@ -424,17 +343,6 @@ export function createLocalProcessSidecarProvisioner({ }; } - // A fences file that could not be read loses the fences of allocations still open; an ended one's late ensure is - // still refused from its record. - async function restoreFences(): Promise { - const stored = await manifests.readFences(); - if (stored === null || stored.length === 0) return; - for (const { allocationId, ...fence } of stored) - fences.set(allocationId, fence); - await pruneFences(); - await persistFences(); - } - async function forgetIdentity(): Promise { await manifests.remove(); await manifests.removePid(); @@ -458,9 +366,7 @@ export function createLocalProcessSidecarProvisioner({ async function restoreSidecar(records: LocalSidecarRecords): Promise { const { isLive, hasWork, reusable } = records; - stillOpen = records.stillOpen; await stopStale(); - await restoreFences(); const stored = await manifests.read(); const recorded = await manifests.readHolds(); if (stored === null) { diff --git a/apps/hub/src/sidecar/manifest-store.ts b/apps/hub/src/sidecar/manifest-store.ts index 1e82de18..2a4db614 100644 --- a/apps/hub/src/sidecar/manifest-store.ts +++ b/apps/hub/src/sidecar/manifest-store.ts @@ -23,26 +23,12 @@ const LocalSidecarHold = type({ export type LocalSidecarHold = typeof LocalSidecarHold.infer; const LocalSidecarHolds = LocalSidecarHold.array(); -/** A destroyed generation of an allocation, and the identities a late ensure at it may still carry. */ -const LocalSidecarFence = type({ - allocationId: "string > 0", - generation: "number.integer >= 0", - sidecarIds: "string[]", - /** The sidecar still hosted the allocation's copy when it was destroyed. */ - "hosted?": "true", -}); -export type LocalSidecarFence = typeof LocalSidecarFence.infer; -const LocalSidecarFences = LocalSidecarFence.array(); - export interface LocalSidecarManifestStore { read(): Promise; write(manifest: LocalSidecarManifest): Promise; remove(): Promise; readHolds(): Promise; writeHolds(holds: readonly LocalSidecarHold[]): Promise; - /** The stored fences, or null when the file could not be read and was moved aside. */ - readFences(): Promise; - writeFences(fences: readonly LocalSidecarFence[]): Promise; /** The pid of the sidecar process last started on this data root, or null when none is recorded. */ readPid(): Promise; writePid(pid: number): Promise; @@ -88,7 +74,6 @@ export function createLocalSidecarManifestStore({ }: CreateLocalSidecarManifestStoreOpts): LocalSidecarManifestStore { const manifestFile = path.join(dataRoot, "sidecar.json"); const holdsFile = path.join(dataRoot, "holds.json"); - const fencesFile = path.join(dataRoot, "fences.json"); const pidFile = path.join(dataRoot, "sidecar.pid"); async function write(manifest: LocalSidecarManifest): Promise { @@ -153,32 +138,6 @@ export function createLocalSidecarManifestStore({ await writeAtomically(holdsFile, JSON.stringify(holds)); } - // An unreadable fences file is kept aside for the operator; null tells restore its fences are lost. - async function readFences(): Promise { - const text = await readOptional(fencesFile); - if (text === null) return []; - try { - return LocalSidecarFences.assert(JSON.parse(text)); - } catch (error) { - const aside = `${fencesFile}.corrupt-${String(Date.now())}`; - await fs.rename(fencesFile, aside); - log({ - level: "warn", - msg: "local_sidecar_fences_unreadable_moved", - file: fencesFile, - aside, - error: errorText(error), - }); - return null; - } - } - - async function writeFences( - fences: readonly LocalSidecarFence[], - ): Promise { - await writeAtomically(fencesFile, JSON.stringify(fences)); - } - async function readPid(): Promise { const pid = Number((await readOptional(pidFile))?.trim()); return Number.isSafeInteger(pid) && pid > 0 ? pid : null; @@ -198,8 +157,6 @@ export function createLocalSidecarManifestStore({ remove, readHolds, writeHolds, - readFences, - writeFences, readPid, writePid, removePid, diff --git a/apps/hub/src/sidecar/restore.ts b/apps/hub/src/sidecar/restore.ts index 1ef31096..33abb35a 100644 --- a/apps/hub/src/sidecar/restore.ts +++ b/apps/hub/src/sidecar/restore.ts @@ -4,7 +4,7 @@ import { createSidecarCredentialResolver, } from "@intx/hub-sessions"; import type { AdapterManifest } from "@intx/types"; -import { and, eq, inArray, isNull, notInArray } from "drizzle-orm"; +import { and, eq, isNull } from "drizzle-orm"; import type { LocalProcessSidecarProvisioner } from "./local-provisioner.ts"; import type { LocalSidecarHold } from "./manifest-store.ts"; @@ -42,30 +42,5 @@ export async function restoreLocalSidecars( async function hasWork(sidecarId: string): Promise { return (await credentials.resolveBindings(sidecarId)).length > 0; } - async function stillOpen( - allocationIds: readonly string[], - ): Promise> { - if (allocationIds.length === 0) return new Set(); - const ids = [...allocationIds]; - const allocations = await db - .select({ id: schema.sidecarAllocation.id }) - .from(schema.sidecarAllocation) - .where( - and( - inArray(schema.sidecarAllocation.id, ids), - notInArray(schema.sidecarAllocation.status, ["released", "failed"]), - ), - ); - const probes = await db - .select({ id: schema.workflowProbe.id }) - .from(schema.workflowProbe) - .where( - and( - inArray(schema.workflowProbe.id, ids), - notInArray(schema.workflowProbe.status, ["succeeded", "failed"]), - ), - ); - return new Set([...allocations, ...probes].map((row) => row.id)); - } - await local.restore({ isLive, hasWork, stillOpen, reusable }); + await local.restore({ isLive, hasWork, reusable }); }