Skip to content

refactor(hub): drop provisioner-side sidecar fences #217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 1 commit into from
Nov 6, 2026
Merged

refactor(hub): drop provisioner-side sidecar fences #217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 2 additions & 30 deletions apps/hub/src/provisioning/docker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,11 +66,6 @@ type CreateDockerSidecarProvisionerOpts = {
readonly hubWebSocketUrl?: string;
};

type Fence = {
readonly generation: number;
readonly destroyedSidecarId?: string;
};

function isDestroyedBy(
container: Container,
request: DestroySidecarRequest,
Expand Down Expand Up @@ -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<string, Fence>();
const queues = new Map<string, Promise<unknown>>();
/** Containers being created, which a listing may not show yet. */
let starting = 0;
Expand Down Expand Up @@ -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<T>(
allocationId: string,
run: () => Promise<T>,
Expand All @@ -261,10 +255,9 @@ export function createDockerSidecarProvisioner({
request: EnsureSidecarRequest,
): Promise<EnsureSidecarResult> {
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) {
Expand All @@ -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 &&
Expand All @@ -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;
Expand All @@ -319,13 +298,6 @@ export function createDockerSidecarProvisioner({
async function destroySidecar(
request: DestroySidecarRequest,
): Promise<DestroySidecarResult> {
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))) {
Expand Down
118 changes: 12 additions & 106 deletions apps/hub/src/sidecar/local-provisioner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import {
type HostCapacity,
} from "./host-capacity.ts";
import type {
LocalSidecarFence,
LocalSidecarHold,
LocalSidecarManifest,
LocalSidecarManifestStore,
Expand Down Expand Up @@ -73,8 +72,6 @@ type LocalSidecarRecords = {
): Promise<boolean>;
/** Whether the hub still has work on the sidecar, including copies it waits for the sidecar to remove. */
hasWork(sidecarId: string): Promise<boolean>;
/** Those of these allocations and probes stock may still ensure: one it ended is never ensured again. */
stillOpen(allocationIds: readonly string[]): Promise<ReadonlySet<string>>;
/** Whether the sidecar was minted with the hub's current adapter manifest and may take work again. */
reusable(sidecarId: string): Promise<boolean>;
};
Expand Down Expand Up @@ -127,12 +124,10 @@ export function createLocalProcessSidecarProvisioner({

const dataDir = localSidecarDataDir(dataRoot);
const holds = new Map<string, LocalSidecarHold>();
const fences = new Map<string, Omit<LocalSidecarFence, "allocationId">>();
let manifest: LocalSidecarManifest | null = null;
let running: ManagedProcess | null = null;
const queue = createSerialQueue();
let shutdownPromise: Promise<void> | 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.
Expand All @@ -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<void> {
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<void> {
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,
Expand Down Expand Up @@ -273,27 +248,11 @@ export function createLocalProcessSidecarProvisioner({
): Promise<EnsureSidecarResult> {
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,
};
}
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -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<LocalSidecarFence, "allocationId"> | undefined;
},
before: LocalSidecarHold | undefined,
sidecar: { readonly process: ManagedProcess; readonly minted: boolean },
): Promise<void> {
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;
Expand All @@ -367,7 +308,6 @@ export function createLocalProcessSidecarProvisioner({
}
try {
await persistHolds();
await persistFences();
} catch (error) {
log({
level: "error",
Expand All @@ -386,35 +326,14 @@ export function createLocalProcessSidecarProvisioner({
request: DestroySidecarRequest,
): Promise<DestroySidecarResult> {
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:
Expand All @@ -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<void> {
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<void> {
await manifests.remove();
await manifests.removePid();
Expand All @@ -458,9 +366,7 @@ export function createLocalProcessSidecarProvisioner({

async function restoreSidecar(records: LocalSidecarRecords): Promise<void> {
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) {
Expand Down
43 changes: 0 additions & 43 deletions apps/hub/src/sidecar/manifest-store.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<LocalSidecarManifest | null>;
write(manifest: LocalSidecarManifest): Promise<void>;
remove(): Promise<void>;
readHolds(): Promise<LocalSidecarHold[]>;
writeHolds(holds: readonly LocalSidecarHold[]): Promise<void>;
/** The stored fences, or null when the file could not be read and was moved aside. */
readFences(): Promise<LocalSidecarFence[] | null>;
writeFences(fences: readonly LocalSidecarFence[]): Promise<void>;
/** The pid of the sidecar process last started on this data root, or null when none is recorded. */
readPid(): Promise<number | null>;
writePid(pid: number): Promise<void>;
Expand Down Expand Up @@ -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<void> {
Expand Down Expand Up @@ -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<LocalSidecarFence[] | null> {
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<void> {
await writeAtomically(fencesFile, JSON.stringify(fences));
}

async function readPid(): Promise<number | null> {
const pid = Number((await readOptional(pidFile))?.trim());
return Number.isSafeInteger(pid) && pid > 0 ? pid : null;
Expand All @@ -198,8 +157,6 @@ export function createLocalSidecarManifestStore({
remove,
readHolds,
writeHolds,
readFences,
writeFences,
readPid,
writePid,
removePid,
Expand Down
Loading