diff --git a/.env.example b/.env.example index bd561c128..f6d66d4f6 100644 --- a/.env.example +++ b/.env.example @@ -25,9 +25,12 @@ HUB_DATA_DIR=./tmp/hub-data # The origin sidecars reach the hub REST API at, for workflows that call it. Unset = the origin of BETTER_AUTH_BASE_URL # HUB_API_ORIGIN= -# Milliseconds a local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000. +# Milliseconds the local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000. # HUB_SIDECAR_STOP_TIMEOUT_MS= +# Deployments the hub's local sidecar runs at once: the listener, one manager per pull request and the judge. Default 16, at most 1792. +# TRIAGE_SIDECAR_MAX_DEPLOYMENTS= + # Docker daemon for workflows that require isolation:workload, registered when it answers # and holds the image from: docker build --target sidecar -t corbits-triage-sidecar . # Unset: the docker CLI's context (DOCKER_CONTEXT, else its current one), else @@ -48,7 +51,7 @@ HUB_DATA_DIR=./tmp/hub-data # Milliseconds between passes that re-queue open pull requests whose head is not triaged. Default 300000. # TRIAGE_RECONCILE_INTERVAL_MS= -# Pull request heads the reconciler lets a tenant have queued or running at once, each in a sidecar of its own. Default 5. +# Pull request heads the reconciler lets a tenant have queued or running at once. Default 5. # TRIAGE_MAX_IN_FLIGHT= # Milliseconds after a head is capped by runs that never started before it gets a fresh set of attempts. Default 21600000. # TRIAGE_CAPPED_RETRY_MS= diff --git a/apps/hub/src/env.ts b/apps/hub/src/env.ts index 62dc27891..60eb28922 100644 --- a/apps/hub/src/env.ts +++ b/apps/hub/src/env.ts @@ -3,6 +3,7 @@ import { homedir } from "node:os"; import { type, type ArkError } from "arktype"; import type { DockerTls, FindDockerHostOpts } from "@corbits/provisioner-docker"; import { LifecycleDuration, lifecycleDurationMs, type ResolvedWorkflowLifecyclePolicy } from "@intx/types"; +import { DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS, MAX_SIDECAR_INCARNATIONS } from "@intx/types/sidecar"; import { DEFAULT_GITHUB_API_ORIGIN } from "./github/github-app-credential-adapter.js"; import type { CoalesceWindow } from "./github/coalescer.js"; import { DEFAULT_RECONCILE_POLICY, type ReconcilePolicy } from "./github/reconcile-plan.js"; @@ -13,6 +14,10 @@ const secret = type("string").narrow((value, ctx) => const positiveInteger = type("string.integer").narrow((value, ctx) => Number(value) > 0 || ctx.mustBe("a positive integer")); +/** The sidecar holds its active and kept deployments within Interchange's combined limit; the hub leaves it the default kept ones. */ +const MAX_LOCAL_SIDECAR_DEPLOYMENTS = MAX_SIDECAR_INCARNATIONS - DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS; +const DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS = 16; + const lifecycleDuration = type("string.trim").pipe(LifecycleDuration); const positiveDuration = lifecycleDuration.narrow((value, ctx) => @@ -75,6 +80,8 @@ const HubEnvSchema = type({ "TRIAGE_MAX_IN_FLIGHT?": positiveInteger, "TRIAGE_CAPPED_RETRY_MS?": positiveInteger, "TRIAGE_COALESCE_QUIET_MS?": positiveInteger, + "TRIAGE_SIDECAR_MAX_DEPLOYMENTS?": positiveInteger.narrow((value, ctx) => + Number(value) <= MAX_LOCAL_SIDECAR_DEPLOYMENTS || ctx.mustBe(`at most ${String(MAX_LOCAL_SIDECAR_DEPLOYMENTS)}`)), "HUB_AGENT_GC_PACK_THRESHOLD?": positiveInteger, "HUB_AGENT_GC_LOOSE_THRESHOLD?": positiveInteger, "HUB_AGENT_GC_WARN_BYTES?": positiveInteger, @@ -267,7 +274,7 @@ export function triageReconcileIntervalMs(env: HubEnv): number { return env.TRIAGE_RECONCILE_INTERVAL_MS === undefined ? 5 * 60_000 : Number(env.TRIAGE_RECONCILE_INTERVAL_MS); } -/** Heads the reconciler lets a tenant have queued or running at once; each runs in a worker sidecar of its own. */ +/** Heads the reconciler lets a tenant have queued or running at once. */ export function triageReconcilePolicy(env: Pick): ReconcilePolicy { return { ...DEFAULT_RECONCILE_POLICY, @@ -276,6 +283,11 @@ export function triageReconcilePolicy(env: Pick): number { + return numberOr(env.TRIAGE_SIDECAR_MAX_DEPLOYMENTS, DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS); +} + const COALESCE_BURST_MS = 3_000; /** diff --git a/apps/hub/src/interchange-hub.ts b/apps/hub/src/interchange-hub.ts index c3b9dc9fb..7b5cf20c0 100644 --- a/apps/hub/src/interchange-hub.ts +++ b/apps/hub/src/interchange-hub.ts @@ -47,6 +47,7 @@ import { resolveSenderKey, resolveSenderPrincipal, withExecutableWorkflowRun, + type DB, } from "@intx/db"; import { createEnvKeyCredentialCipher } from "@intx/crypto"; import { @@ -112,12 +113,10 @@ export type CreateHubServerOpts = { readonly credentialCipher: CredentialCipher; /** Provisioners eligible to host frozen workflow deployments. */ readonly sidecarProvisioners?: readonly SidecarProvisioner[]; - /** Selects among matching deployment provisioners. Defaults to the first. */ - readonly sidecarProvisionerChooser?: SidecarProvisionerChooser; + /** Selects among matching deployment provisioners, reading the hub's database. Defaults to the first. */ + readonly sidecarProvisionerChooser?: (db: DB["db"]) => SidecarProvisionerChooser; /** Provisioners eligible to evaluate workflow source code. */ readonly probeSidecarProvisioners?: readonly SidecarProvisioner[]; - /** Selects among matching probe provisioners. Defaults to the first. */ - readonly probeSidecarProvisionerChooser?: SidecarProvisionerChooser; readonly probeSidecarCapabilityRules?: readonly SidecarCapabilityRule[]; /** Maximum simultaneous allocation reconciliations. Defaults to eight. */ readonly sidecarAllocationConcurrency?: number; @@ -138,7 +137,6 @@ export async function createInterchangeHub({ sidecarProvisioners = [], sidecarProvisionerChooser, probeSidecarProvisioners = [], - probeSidecarProvisionerChooser, probeSidecarCapabilityRules = [], sidecarAllocationConcurrency = DEFAULT_SIDECAR_ALLOCATION_CONCURRENCY, sidecarOperationTimeoutMs, @@ -355,14 +353,11 @@ export async function createInterchangeHub({ const sidecarPlugins = createSidecarPluginRegistry({ provisioners: sidecarProvisioners, ...(sidecarProvisionerChooser !== undefined - ? { chooser: sidecarProvisionerChooser } + ? { chooser: sidecarProvisionerChooser(db) } : {}), }); const probeSidecarPlugins = createSidecarPluginRegistry({ provisioners: probeSidecarProvisioners, - ...(probeSidecarProvisionerChooser !== undefined - ? { chooser: probeSidecarProvisionerChooser } - : {}), }); const workflowAllocationService = createWorkflowAllocationService({ db, diff --git a/apps/hub/src/manager-state.postgres.test.ts b/apps/hub/src/manager-state.postgres.test.ts index 1cd15ffbe..4763093f3 100644 --- a/apps/hub/src/manager-state.postgres.test.ts +++ b/apps/hub/src/manager-state.postgres.test.ts @@ -4,7 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto"; import { sql } from "drizzle-orm"; import { createDB, runMigrations, schema } from "@intx/db"; import type { Tx } from "./github/admin-github-grant.js"; -import { done, liveListener, liveManager, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js"; +import { done, endNeverRan, liveListener, liveManager, notePlaced, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js"; import { migrateTriageTables } from "./triage-db.js"; const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" }; @@ -411,7 +411,7 @@ test("PostgreSQL: a manager deployed without a sidecar past the startup window d cancellationReason: schema.workflowRun.cancellationReason, }).from(schema.workflowRun).where(sql`${schema.workflowRun.id} = 'run_stuck'`); expect(left?.cancellationRequestedAt).toBeInstanceOf(Date); - expect(left?.cancellationReason).toBe("manager deployed without a sidecar"); + expect(left?.cancellationReason).toBe("manager deployed without a sidecar or never ran"); }); }, 120_000); @@ -464,6 +464,49 @@ test("PostgreSQL: a manager being cancelled is mailed nothing and takes nothing, }); }, 120_000); +async function allocate(db: Db, runId: string, fields: { status: "allocated"; sidecarId: string }): Promise { + await db.insert(schema.sidecar).values({ id: fields.sidecarId, tokenHashSha256: new Uint8Array(32).fill(runId.length), provisionerId: "local-process", provisionerApiVersion: 2, provisionerBindingFingerprint: "fp", adapterManifestDigest: "digest" }).onConflictDoNothing(); + await db.insert(schema.sidecarAllocation).values({ id: `alc_${runId}`, anchorRunId: runId, tenantId: TENANT, provisionerId: "local-process", provisionerApiVersion: 2, provisionerBindingFingerprint: "fp", adapterManifestDigest: "digest", maxDisconnectedMs: 60_000, ...fields }); +} + +test("PostgreSQL: a manager on its sidecar five minutes without running gives up the pull request, and the reaper ends it but never the listener", async () => { + await withDatabase(async function neverRan(db) { + await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" }); + await db.insert(schema.workflowRun).values({ id: "run_listener", anchorRunId: "run_listener", definitionId: "wfd_listener", tenantId: TENANT, status: "deployed", createdAt: new Date(Date.now() - 60 * MINUTE) }); + await allocate(db, "run_listener", { status: "allocated", sidecarId: "sdc_local" }); + await deployedRun(db, "run_quiet", new Date(Date.now() - 6 * MINUTE)); + await allocate(db, "run_quiet", { status: "allocated", sidecarId: "sdc_local" }); + notePlaced({ allocationId: "alc_run_quiet", generation: 0 }, new Date(Date.now() - 6 * MINUTE)); + await claimRun(db, "run_quiet"); + const cancelled: string[][] = []; + const logs: Record[] = []; + + await endNeverRan(db, new Date(), async (...args) => cancelled.push(args), (entry) => logs.push(entry)); + + expect(cancelled).toEqual([[TENANT, "run_quiet", "triage-manager-github deployed but never ran"]]); + expect(logs).toEqual([{ level: "warn", msg: "triage_deployment_never_ran", runId: "run_quiet", workflow: "triage-manager-github" }]); + expect(await liveListener(db, TENANT, "github")).toMatchObject({ runId: "run_listener" }); + await liveRun(db, "run_next"); + expect(await take(db, TENANT, GITHUB, "run_next")).toMatchObject({ outcome: "idle" }); + }); +}, 120_000); + +test("PostgreSQL: a manager that waited past five minutes for a slot is timed from its placement, not its deploy", async () => { + await withDatabase(async function placedLate(db) { + await deployedRun(db, "run_late", new Date(Date.now() - 10 * MINUTE)); + await allocate(db, "run_late", { status: "allocated", sidecarId: "sdc_local" }); + notePlaced({ allocationId: "alc_run_late", generation: 0 }, new Date(Date.now() - MINUTE)); + await claimRun(db, "run_late"); + const cancelled: string[][] = []; + + await endNeverRan(db, new Date(), async (...args) => cancelled.push(args), () => undefined); + + expect(cancelled).toEqual([]); + await liveRun(db, "run_next"); + expect(await take(db, TENANT, GITHUB, "run_next")).toEqual({ outcome: "refused" }); + }); +}, 120_000); + test("PostgreSQL: a listener counts as live as a manager does, and one deployed without a sidecar past the startup window is released", async () => { await withDatabase(async function listener(db) { await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" }); diff --git a/apps/hub/src/manager-state.ts b/apps/hub/src/manager-state.ts index f817d927a..dd50a8ca9 100644 --- a/apps/hub/src/manager-state.ts +++ b/apps/hub/src/manager-state.ts @@ -2,14 +2,14 @@ // pull request when no live run holds it, and done checks for pending events under the claim's lock, so an event is // either taken by the live manager or by the next one. Only the run the claim names takes events or finishes a pass. import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNull, like, or, sql, type SQL } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, isNull, like, lt, or, sql, type SQL } from "drizzle-orm"; import { jsonb, primaryKey, text } from "drizzle-orm/pg-core"; import { type } from "arktype"; import { canExecuteWorkflowRun, schema, type DB } from "@intx/db"; import { asRecord } from "@corbits/common"; -import { APPROVE_DO_KIND, REPO_SOURCES, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, REPO_SOURCES, TRIAGE_JUDGES, TRIAGE_LISTENERS, TRIAGE_MANAGERS, type RepoSource } from "@corbits/triage-contracts"; import type { Tx } from "./github/admin-github-grant.js"; -import { resolveLiveDeployments, type LiveDeployment } from "./github/deployment.js"; +import { resolveLiveDeployments, workflowName, type LiveDeployment } from "./github/deployment.js"; import { triageSchema } from "./triage-db.js"; import { START_DEADLINE_MS } from "./workflow-release/rollout.js"; @@ -195,7 +195,46 @@ export async function recordResult(db: DB["db"], tenantId: string, pr: PullReque */ const MANAGER_START_GRACE_MS = 2 * 60_000; -type ManagerRun = { status: string; createdAt: Date; expiresAt: Date | null; cancellationRequestedAt: Date | null; allocationStatus: string | null; sidecarId: string | null }; +/** + * How long a manager or judge may sit on its sidecar without its first mail starting it. Both are deployed with that + * mail, so one still waiting is stuck. A listener is exempt: it stays deployed until its first pull request arrives. + */ +const NEVER_RAN_MS = 5 * 60_000; + +/** The workflows deployed with the mail that starts them. */ +const MAILED_WORKFLOWS = [...Object.values(TRIAGE_MANAGERS), ...Object.values(TRIAGE_JUDGES)]; + +type Placement = { readonly allocationId: string; readonly generation: number }; + +/** When this hub saw each allocation's current generation reach its sidecar; one placed before the hub started dates from its start. */ +const placements = new Map(); +const hubStartedAt = new Date(); + +/** Records when an allocation's generation reached its sidecar, so a run that waited for a slot is timed from there. */ +export function notePlaced({ allocationId, generation }: Placement, at = new Date()): void { + if (placements.get(allocationId)?.generation === generation) return; + // An entry older than the window no longer keeps any run up, so it can go. + for (const [id, placed] of placements) if (at.getTime() - placed.at.getTime() > NEVER_RAN_MS) placements.delete(id); + placements.set(allocationId, { generation, at }); +} + +type ManagerRun = { + status: string; + createdAt: Date; + expiresAt: Date | null; + cancellationRequestedAt: Date | null; + allocationId: string | null; + generation: number | null; + allocationStatus: string | null; + sidecarId: string | null; +}; + +/** How long the run has sat on its sidecar: from its deploy, or its placement there when that came later. */ +function onSidecarMs(run: Pick, now: Date): number { + const placed = run.allocationId === null ? undefined : placements.get(run.allocationId); + const placedAt = placed !== undefined && placed.generation === run.generation ? placed.at : hubStartedAt; + return now.getTime() - Math.max(run.createdAt.getTime(), placedAt.getTime()); +} async function managerRun(tx: Tx | DB["db"], runId: string): Promise { const [run] = await tx @@ -204,6 +243,8 @@ async function managerRun(tx: Tx | DB["db"], runId: string): Promise { - return !runIsUp(await managerRun(tx, runId), now); + return !runIsUp(await managerRun(tx, runId), now, { startedByMail: true }); } -/** Whether a deployment is up as the hub counts a listener or a manager: running, on its sidecar, or just deployed. */ +/** Whether a deployment is up as the hub counts a listener: running, on its sidecar, or just deployed. */ export async function deploymentUp(db: DB["db"], runId: string): Promise { - return !(await runGone(db, runId, new Date())); + return runIsUp(await managerRun(db, runId), new Date(), { startedByMail: false }); } /** - * Asks the sweep to stop a deploy that still counts as live but never became a process, so a manager cannot start after - * its claim has moved on, and a listener is deployed again by the hub's next roll of the workspace. + * Asks the sweep to stop a deploy that still counts as live but never became a process, or never ran, so a manager + * cannot start after its claim has moved on, and a listener is deployed again by the hub's next roll of the workspace. */ -async function releaseStuck(tx: Tx | DB["db"], runId: string, now: Date, what: "manager" | "listener"): Promise { +async function releaseStuck(tx: Tx | DB["db"], runId: string, now: Date, reason: string): Promise { await tx .update(schema.workflowRun) - .set({ cancellationRequestedAt: now, cancellationDeadline: now, cancellationReason: `${what} deployed without a sidecar` }) + .set({ cancellationRequestedAt: now, cancellationDeadline: now, cancellationReason: reason }) .where(and(eq(schema.workflowRun.id, runId), inArray(schema.workflowRun.status, [...schema.liveWorkflowRunStatuses]), isNull(schema.workflowRun.cancellationRequestedAt))); } +/** Ends every manager and judge that sat on its sidecar past `NEVER_RAN_MS` without its mail starting it. */ +export async function endNeverRan( + db: DB["db"], + now: Date, + cancel: (tenantId: string, runId: string, reason: string) => Promise, + log: (entry: Record) => void, +): Promise { + const { workflowRun, workflowDefinition, workflowRunLaunchSpec, sidecarAllocation } = schema; + const candidates = await db + .select({ tenantId: workflowRun.tenantId, runId: workflowRun.id, workflow: workflowName, createdAt: workflowRun.createdAt, allocationId: sidecarAllocation.id, generation: sidecarAllocation.generation }) + .from(workflowRun) + .innerJoin(workflowDefinition, eq(workflowRun.definitionId, workflowDefinition.id)) + .leftJoin(workflowRunLaunchSpec, eq(workflowRunLaunchSpec.anchorRunId, workflowRun.id)) + .innerJoin(sidecarAllocation, eq(sidecarAllocation.anchorRunId, workflowRun.id)) + .where(and( + eq(workflowRun.id, workflowRun.anchorRunId), + eq(workflowRun.status, "deployed"), + isNull(workflowRun.cancellationRequestedAt), + lt(workflowRun.createdAt, new Date(now.getTime() - NEVER_RAN_MS)), + eq(sidecarAllocation.status, "allocated"), + isNotNull(sidecarAllocation.sidecarId), + inArray(workflowName, MAILED_WORKFLOWS), + )); + for (const run of candidates) { + if (onSidecarMs(run, now) < NEVER_RAN_MS) continue; + await cancel(run.tenantId, run.runId, `${run.workflow} deployed but never ran`); + log({ level: "warn", msg: "triage_deployment_never_ran", runId: run.runId, workflow: run.workflow }); + } +} + /** * The workspace's newest running listener for the host, so pull requests keep going to the one a roll replaces until * the new one runs; else the newest still starting. One that never got a sidecar past its startup window is released. @@ -260,8 +337,8 @@ export async function liveListener(db: DB["db"], tenantId: string, source: RepoS if (listener.cancelling) continue; const run = await managerRun(db, listener.runId); if (runIsRunning(run)) return listener; - if (runIsUp(run, now, LISTENER_START_GRACE_MS)) starting ??= listener; - else await releaseStuck(db, listener.runId, now, "listener"); + if (runIsUp(run, now, { graceMs: LISTENER_START_GRACE_MS, startedByMail: false })) starting ??= listener; + else await releaseStuck(db, listener.runId, now, "listener deployed without a sidecar"); } return starting; } @@ -273,7 +350,7 @@ export async function liveManager(db: DB["db"], tenantId: string, pr: PullReques const claim = await lockClaim(tx, tenantId, pr); if (claim?.status !== "live" || claim.runId === undefined) return undefined; const run = await managerRun(tx, claim.runId); - return run && canExecuteWorkflowRun(run) && runIsUp(run, now) ? claim.runId : undefined; + return run && canExecuteWorkflowRun(run) && runIsUp(run, now, { startedByMail: true }) ? claim.runId : undefined; }); } @@ -294,7 +371,7 @@ async function claimFor(tx: Tx, tenantId: string, pr: PullRequestRef, runId: str if (!claim) throw new Error(`manager_state ${key} vanished under its lock`); if (claim.status === "live" && claim.runId === runId) return claim; if (claim.status === "live" && claim.runId !== undefined && !(await runGone(tx, claim.runId, now))) return undefined; - if (claim.runId !== undefined && claim.runId !== runId) await releaseStuck(tx, claim.runId, now, "manager"); + if (claim.runId !== undefined && claim.runId !== runId) await releaseStuck(tx, claim.runId, now, "manager deployed without a sidecar or never ran"); const taken = { ...claim, ...fresh, starts: claim.starts + 1 }; await writeClaim(tx, tenantId, pr, taken); return taken; diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index db3435787..e18f1a231 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -13,7 +13,7 @@ // a workflow rename or tool-id change breaks the build here. import { resolve } from "node:path"; import { authorize } from "@intx/authz"; -import { createGrantStore, createWorkflowRunLaunchSpecStore, resolveSenderPrincipal, schema, type ResolvedSenderPrincipal } from "@intx/db"; +import { createGrantStore, createWorkflowRunLaunchSpecStore, resolveSenderPrincipal, schema, type DB, type ResolvedSenderPrincipal } from "@intx/db"; import { createEnvKeyCredentialCipher } from "@intx/crypto"; import { hexDecode } from "@intx/types"; import { generateId } from "@intx/hub-common"; @@ -37,7 +37,7 @@ import { asRecord, errorText } from "@corbits/common"; import { createInterchangeHub } from "./interchange-hub.js"; import { createManagerIntake, createManagerWake, createOverdueWake, createSupersede } from "./manager-intake.js"; import { MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; -import { deploymentUp, liveListener, passRequest } from "./manager-state.js"; +import { deploymentUp, endNeverRan, liveListener, notePlaced, passRequest } from "./manager-state.js"; import { mountTenantRoutes } from "./tenant-routes.js"; import { connectDockerSidecars, dockerPlaces, sweepDockerSidecarsOf } from "./sidecar/docker.js"; import { connectE2bSidecars, tendE2bSidecarsOf, type E2bSidecarProvisioner } from "./sidecar/e2b.js"; @@ -45,6 +45,7 @@ import { createLocalProcessSidecarProvisioner } from "./sidecar/local-provisione import { connectRailwaySidecars, sweepRailwaySidecarsOf, type RailwaySidecarProvisioner } from "./sidecar/railway.js"; import { createHostCapacity } from "./sidecar/host-capacity.js"; import { createLocalSidecarManifestStore } from "./sidecar/manifest-store.js"; +import { createPlacementChooser, localSidecarUsage } from "./sidecar/placement.js"; import { restoreLocalSidecars } from "./sidecar/restore.js"; import { createSpawnLocalSidecar } from "./sidecar/spawn.js"; import { SIDECAR_ADAPTER_MANIFEST } from "./sidecar-config.js"; @@ -69,7 +70,7 @@ import { createListenerWatch, releaseWriteOf } from "./workflow-release/triggers import { setUpPublisher, TRIAGE_PUBLISHER } from "./workflow-release/publisher.js"; import { catalogModels, createWorkspaceLock, deploymentState, liveDeployments } from "./workflow-release/reads.js"; import { createRolloutQueue, createWorkflowRollout, type DeployRequest } from "./workflow-release/rollout.js"; -import { bitbucketEgressAllowlist, childProcessEnv, clientIpHeaders, databaseConfig, dockerSettings, e2bSettings, emailSettings, interchangeSettings, githubApiOrigin, loadHubEnv, inviteLinkOrigin, railwaySandboxSettings, signInSettings, triageCoalesceWindow, triageReconcileIntervalMs, triageReconcilePolicy } from "./env.js"; +import { bitbucketEgressAllowlist, childProcessEnv, clientIpHeaders, databaseConfig, dockerSettings, e2bSettings, emailSettings, interchangeSettings, githubApiOrigin, loadHubEnv, inviteLinkOrigin, localSidecarMaxDeployments, railwaySandboxSettings, signInSettings, triageCoalesceWindow, triageReconcileIntervalMs, triageReconcilePolicy } from "./env.js"; import { HOOK_MOUNT_PATH } from "./hooks.js"; import { assertMigrated } from "./migrations.js"; import { createBridgeHandler, logJson, withAppLogin } from "./github/bridge.js"; @@ -155,12 +156,14 @@ await assertMigrated(database); const credentialCipher = createEnvKeyCredentialCipher(hexDecode(env.CREDENTIAL_ENCRYPTION_KEY)); const localSidecarDataRoot = `${env.HUB_DATA_DIR}/local-sidecars`; +const localSidecarCap = localSidecarMaxDeployments(env); const local = createLocalProcessSidecarProvisioner({ dataRoot: localSidecarDataRoot, ...(env.HUB_SIDECAR_STOP_TIMEOUT_MS !== undefined && { stopTimeoutMs: Number(env.HUB_SIDECAR_STOP_TIMEOUT_MS) }), spawnSidecar: createSpawnLocalSidecar({ interchangeDir: V, credentialEncryptionKey: env.SIDECAR_CREDENTIAL_ENCRYPTION_KEY, + maxActiveDeployments: localSidecarCap, processEnv: childProcessEnv(env), }), manifests: createLocalSidecarManifestStore({ @@ -170,6 +173,7 @@ const local = createLocalProcessSidecarProvisioner({ }), capacity: createHostCapacity({ cgroupRoot: "/sys/fs/cgroup", meminfo: "/proc/meminfo" }), adapterManifest: SIDECAR_ADAPTER_MANIFEST, + onAccepted: notePlaced, log: logJson, }); const docker = await connectDockerSidecars({ docker: dockerSettings(env), database, adapterManifest: SIDECAR_ADAPTER_MANIFEST }); @@ -187,6 +191,9 @@ const sidecarProvisioners = [local.provisioner, ...workloadProvisioners]; const codeReviewProbes = [...dockerPlaces(docker), { guide: E2B_PLACE, found: e2b }, { guide: RAILWAY_PLACE, found: railway }]; const noIsolatedPlace = isolatedPlaceBootWarning(workloadProvisioners.length > 0, codeReviewProbes.map((probe) => probe.guide)); if (noIsolatedPlace) logJson(noIsolatedPlace); +function placeDeployments(db: DB["db"]) { + return createPlacementChooser({ db, local: local.provisioner, maxDeployments: localSidecarCap }); +} let shuttingDown = false; let cronTicker: CronTicker | undefined; @@ -233,12 +240,13 @@ const composition = await createInterchangeHub({ credentialCipher, adapterManifest: SIDECAR_ADAPTER_MANIFEST, sidecarProvisioners, + sidecarProvisionerChooser: placeDeployments, probeSidecarProvisioners: sidecarProvisioners, probeSidecarCapabilityRules: workloadProvisioners.length > 0 ? [{ capability: "isolation:workload", effect: "require" }] : [], extraRunTokenRoutes: MANAGER_RUN_TOKEN_ROUTES, }); const stock = composition.server; -composition.sidecarRouter.events.on("sidecar.allocated.connected", local.connected); +composition.sidecarRouter.events.on("sidecar.allocated.connected", notePlaced); /** The hub's credentials, and the push that takes back from sidecars an App key a read moved to the hub. */ const credentialDeps = { @@ -626,6 +634,7 @@ async function reconcileAll(): Promise { await supersedeAll(tenant.id); await wakeOverdue(tenant.id); } + await endNeverRan(composition.db, new Date(), composition.workflowLifecycleService.requestCancellation, logJson); const outcomes: ReconcileOutcome[] = []; for (const reconcile of reconcilers) outcomes.push(await reconcile()); const retry = outcomes.some((outcome) => outcome.retry); @@ -851,6 +860,9 @@ const integrations = createIntegrationsApp({ db: composition.db, getSession: com return deploymentUp(composition.db, anchorRunId); }, rollouts: workflowRollout.rollouts, + sidecar: function sidecarUsage() { + return localSidecarUsage(composition.db, local.provisioner, localSidecarCap); + }, }), codeReviewPlace: createCodeReviewPlace({ authorize: authorizePortal, place: placeCodeReview, probes: codeReviewProbes }), codeReviewRun: createCodeReviewRun({ db: composition.db, authorize: authorizePortal, place: placeCodeReview, readCheckPack, readHead: pullRequests.head, send: sendToListener }), diff --git a/apps/hub/src/sidecar/host-capacity.test.ts b/apps/hub/src/sidecar/host-capacity.test.ts index 55baad81c..40edfdd14 100644 --- a/apps/hub/src/sidecar/host-capacity.test.ts +++ b/apps/hub/src/sidecar/host-capacity.test.ts @@ -52,8 +52,8 @@ test("without a cgroup limit MemAvailable decides, and a box reporting neither h expect(await createHostCapacity(await box({})).freeBytes()).toBeUndefined(); }); -test("each sidecar still starting holds one footprint, and no figure never declines", () => { - expect(shortfall(2 * SIDECAR_BYTES, 1)).toBeUndefined(); - expect(shortfall(2 * SIDECAR_BYTES, 2)).toBe(`${(2 * SIDECAR_BYTES) / MiB} MiB free, ${(3 * SIDECAR_BYTES) / MiB} MiB needed`); - expect(shortfall(undefined, 100)).toBeUndefined(); +test("the sidecar needs one footprint free, and no figure never declines", () => { + expect(shortfall(SIDECAR_BYTES)).toBeUndefined(); + expect(shortfall(SIDECAR_BYTES / 2)).toBe(`${SIDECAR_BYTES / 2 / MiB} MiB free, ${SIDECAR_BYTES / MiB} MiB needed`); + expect(shortfall(undefined)).toBeUndefined(); }); diff --git a/apps/hub/src/sidecar/host-capacity.ts b/apps/hub/src/sidecar/host-capacity.ts index 660a90397..f33890e1b 100644 --- a/apps/hub/src/sidecar/host-capacity.ts +++ b/apps/hub/src/sidecar/host-capacity.ts @@ -1,9 +1,7 @@ -// Whether this box has the memory for one more local sidecar: free memory in +// Whether this box has the memory to start the local sidecar: free memory in // the container's cgroup (v2, else v1) where a limit is set, else Linux's -// MemAvailable, less one footprint for each sidecar started that has not -// connected yet, as its memory is not in use yet. A box with neither, such as -// a macOS dev machine, never declines: os.freemem there leaves out memory the -// system would reclaim. +// MemAvailable. A box with neither, such as a macOS dev machine, never +// declines: os.freemem there leaves out memory the system would reclaim. import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -69,9 +67,8 @@ export function createHostCapacity({ cgroupRoot, meminfo }: HostCapacityOpts): H }; } -/** Why there is no room for one more sidecar beside `starting` ones not connected yet, or undefined when there is. */ -export function shortfall(free: number | undefined, starting: number): string | undefined { - const needed = SIDECAR_BYTES * (starting + 1); - if (free === undefined || free >= needed) return undefined; - return `${Math.floor(free / MiB)} MiB free, ${needed / MiB} MiB needed`; +/** Why there is no room to start the sidecar, or undefined when there is. */ +export function shortfall(free: number | undefined): string | undefined { + if (free === undefined || free >= SIDECAR_BYTES) return undefined; + return `${Math.floor(free / MiB)} MiB free, ${SIDECAR_BYTES / MiB} MiB needed`; } diff --git a/apps/hub/src/sidecar/local-provisioner.conformance.test.ts b/apps/hub/src/sidecar/local-provisioner.conformance.test.ts deleted file mode 100644 index d5d70c26b..000000000 --- a/apps/hub/src/sidecar/local-provisioner.conformance.test.ts +++ /dev/null @@ -1,63 +0,0 @@ -import fs from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; -import { describeProvisionerConformance } from "@corbits/provisioner-test"; -import { createLocalProcessSidecarProvisioner } from "./local-provisioner.js"; -import type { LocalSidecarManifest } from "./manifest-store.js"; - -type FakeProcess = { - readonly generation: number; - readonly sidecarId: string; - readonly allocationId: string; - running: boolean; -}; - -describeProvisionerConformance("local-process", async () => { - const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-conformance-")); - const processes: FakeProcess[] = []; - const manifests = new Map(); - - const local = createLocalProcessSidecarProvisioner({ - dataRoot, - capacity: { freeBytes: async () => undefined }, - adapterManifest: [], - spawnSidecar: function spawnSidecar({ request }) { - const process: FakeProcess = { ...request, running: true }; - processes.push(process); - const exit = Promise.withResolvers(); - return { - pid: processes.length, - exited: exit.promise, - kill() { - process.running = false; - exit.resolve(0); - }, - }; - }, - manifests: { - readAll: async () => [...manifests.values()], - async write(manifest) { - manifests.set(manifest.allocationId, manifest); - }, - readDataDir: async (allocationId) => manifests.get(allocationId)?.dataDir ?? null, - async remove(allocationId) { - manifests.delete(allocationId); - }, - }, - log: () => {}, - }); - - return { - provisioner: local.provisioner, - async live(allocationId) { - return processes - .filter((process) => process.running && process.allocationId === allocationId) - .map(({ generation, sidecarId }) => ({ generation, sidecarId })); - }, - capabilityChecks: {}, - async dispose() { - await local.shutdown(); - await fs.rm(dataRoot, { recursive: true, force: true }); - }, - }; -}); diff --git a/apps/hub/src/sidecar/local-provisioner.test.ts b/apps/hub/src/sidecar/local-provisioner.test.ts index 4b5b9ea9f..e0db68511 100644 --- a/apps/hub/src/sidecar/local-provisioner.test.ts +++ b/apps/hub/src/sidecar/local-provisioner.test.ts @@ -1,133 +1,386 @@ -import { expect, test } from "bun:test"; +import { afterEach, expect, test } from "bun:test"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { createEnvKeyCredentialCipher } from "@intx/crypto"; import type { EnsureSidecarRequest } from "@intx/hub-sessions"; import type { AdapterManifest } from "@intx/types"; -import { CAPACITY_UNAVAILABLE, SIDECAR_BYTES } from "./host-capacity.js"; -import { createLocalProcessSidecarProvisioner } from "./local-provisioner.js"; -import type { LocalSidecarManifest } from "./manifest-store.js"; +import { SIDECAR_BYTES } from "./host-capacity.js"; +import { createLocalProcessSidecarProvisioner, type LocalSidecarRecords } from "./local-provisioner.js"; +import { createLocalSidecarManifestStore, type LocalSidecarHold, type LocalSidecarManifest } from "./manifest-store.js"; const HUB_ADAPTERS: AdapterManifest = [ { provider: "corbits-system-one", specifier: "@corbits/system-one", export: "createSystemOneAdapterFactory" }, ]; -const RECORD: LocalSidecarManifest = { - allocationId: "alloc_1", - generation: 1, - sidecarId: "sc_1", - tenantId: "tnt_1", - anchorRunId: "run_1", - hubWebSocketUrl: "ws://hub", - dataDir: "/data/alloc_1", - token: "token", -}; - -// Written by a hub that loaded a GitHub adapter module it no longer ships. -const STALE_RECORD = { - ...RECORD, - adapterManifest: [...HUB_ADAPTERS, { provider: "corbits-github-app-credential", specifier: "file:///app/apps/hub/src/github/github-app-adapter-module.ts", export: "installGithubAppCredentialAdapter" }], -}; - -test("restore launches a record with the hub's adapters, not the ones it was written with", async () => { - const spawned: EnsureSidecarRequest[] = []; - const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-")); - const local = createLocalProcessSidecarProvisioner({ - dataRoot, - capacity: { freeBytes: async () => undefined }, - adapterManifest: HUB_ADAPTERS, - spawnSidecar: function spawnSidecar({ request }) { - spawned.push(request); - return { pid: 1, exited: new Promise(function never() {}), kill() {} }; - }, - manifests: { - readAll: async () => [STALE_RECORD], - write: async () => {}, - readDataDir: async () => null, - remove: async () => {}, - }, - log: () => {}, - }); +const roots: string[] = []; + +afterEach(async () => { + for (const root of roots.splice(0)) await fs.rm(root, { recursive: true, force: true }); +}); - try { - await local.restore(async () => true); - } finally { - await fs.rm(dataRoot, { recursive: true, force: true }); +function request(allocationId: string, generation = 1): EnsureSidecarRequest { + return { + allocationId, + generation, + tenantId: "tnt_1", + anchorRunId: `run_${allocationId}`, + sidecarId: `sc_${allocationId}`, + token: `token_${allocationId}`, + hubWebSocketUrl: "ws://hub", + }; +} + +async function setup({ freeBytes = async () => undefined }: { freeBytes?: () => Promise } = {}) { + const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-")); + roots.push(dataRoot); + const spawned: { manifest: LocalSidecarManifest; adapterManifest: AdapterManifest }[] = []; + const logs: Record[] = []; + const accepted: { allocationId: string; generation: number }[] = []; + const faults = { holdsWrite: false }; + const manifests = createLocalSidecarManifestStore({ dataRoot, cipher: createEnvKeyCredentialCipher(new Uint8Array(32).fill(7)), log: (entry) => logs.push(entry) }); + function create() { + return createLocalProcessSidecarProvisioner({ + dataRoot, + capacity: { freeBytes }, + adapterManifest: HUB_ADAPTERS, + spawnSidecar: function spawnSidecar(args) { + spawned.push(args); + const exit = Promise.withResolvers<{ code: number | null; signal: string | null }>(); + return { + pid: spawned.length, + exited: exit.promise, + kill(signal) { + exit.resolve({ code: null, signal }); + }, + }; + }, + manifests: { + ...manifests, + async writeHolds(holds) { + if (faults.holdsWrite) throw new Error("disk full"); + await manifests.writeHolds(holds); + }, + }, + onAccepted: (placement) => accepted.push({ ...placement }), + log: (entry) => logs.push(entry), + }); } + async function started(records: LocalSidecarRecords = NOTHING_HELD) { + const local = create(); + await local.restore(records); + return local; + } + return { dataRoot, manifests, spawned, logs, accepted, faults, create, started }; +} + +test("every allocation joins the one sidecar, and destroying any of them leaves it running for the hub to undeploy", async () => { + const { manifests, spawned, started } = await setup(); + const local = await started(); - expect(spawned.map((request) => request.adapterManifest)).toEqual([HUB_ADAPTERS]); + expect(await local.provisioner.ensure(request("a"))).toEqual({ kind: "accepted", externalRef: "1" }); + expect(await local.provisioner.ensure(request("b"))).toEqual({ kind: "accepted", externalRef: "1", sidecarId: "sc_a" }); + expect(await local.provisioner.ensure(request("b"))).toEqual({ kind: "accepted", externalRef: "1", sidecarId: "sc_a" }); + expect(spawned.map(({ manifest }) => manifest.sidecarId)).toEqual(["sc_a"]); + + expect(await local.provisioner.destroy({ allocationId: "a", generation: 2, sidecarId: "sc_a" })).toEqual({ kind: "destroyed", cleanup: "required" }); + // The hub never learned this ensure's outcome, so it names the identity it offered, under which nothing was deployed. + expect(await local.provisioner.destroy({ allocationId: "c", generation: 1, sidecarId: "sc_c" })).toEqual({ kind: "destroyed", cleanup: "confirmed" }); + expect(await local.provisioner.ensure(request("a", 1))).toMatchObject({ kind: "rejected", code: "stale_generation" }); + expect(await manifests.readHolds()).toEqual([{ allocationId: "b", generation: 1, offeredSidecarId: "sc_b" }]); + + await local.shutdown(); }); -test("restore removes a data dir no manifest names and keeps the ones that are named", async () => { - const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-")); - const named = path.join(dataRoot, "sal_1-Ab12Cd"); - const orphan = path.join(dataRoot, "sal_2-Ef34Gh"); - await fs.mkdir(named); - await fs.mkdir(orphan); - const local = createLocalProcessSidecarProvisioner({ - dataRoot, - capacity: { freeBytes: async () => undefined }, - adapterManifest: HUB_ADAPTERS, - spawnSidecar: function spawnSidecar() { - throw new Error("restore has nothing to respawn"); - }, - manifests: { - readAll: async () => [], - write: async () => {}, - readDataDir: async (allocationId) => (allocationId === "sal_1" ? named : null), - remove: async () => {}, - }, - log: () => {}, - }); +test("a probe's start is the one start: its release keeps the sidecar, which the next deployment joins whatever memory is free", async () => { + let free = SIDECAR_BYTES; + const { spawned, started } = await setup({ freeBytes: async () => free }); + const local = await started(); - try { - await local.restore(async () => true); - expect((await fs.readdir(dataRoot)).sort()).toEqual(["sal_1-Ab12Cd"]); - } finally { - await fs.rm(dataRoot, { recursive: true, force: true }); - } + expect((await local.provisioner.ensure(request("probe"))).kind).toBe("accepted"); + free = 0; + expect(await local.provisioner.destroy({ allocationId: "probe", generation: 2, sidecarId: "sc_probe" })).toEqual({ kind: "destroyed", cleanup: "required" }); + expect(await local.provisioner.ensure(request("deployment"))).toEqual({ kind: "accepted", externalRef: "1", sidecarId: "sc_probe" }); + expect(spawned).toHaveLength(1); + + await local.shutdown(); }); -function request(allocationId: string): EnsureSidecarRequest { - return { ...RECORD, allocationId, sidecarId: `sc_${allocationId}` }; +function nothingElse(): Promise { + return Promise.resolve(false); } -test("ensures side by side count each other's sidecars until each connects or exits, and a decline spawns nothing", async () => { - const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-")); - const exits = new Map void>(); - const local = createLocalProcessSidecarProvisioner({ - dataRoot, - capacity: { freeBytes: async () => 2 * SIDECAR_BYTES }, - adapterManifest: HUB_ADAPTERS, - spawnSidecar: function spawnSidecar({ request: spawned }) { - const exited = new Promise(function onExit(resolve) { - exits.set(spawned.allocationId, () => resolve(0)); - }); - return { pid: 1, exited, kill() {} }; - }, - manifests: { - readAll: async () => [], - write: async () => {}, - readDataDir: async () => null, - remove: async () => {}, +async function noneOpen(): Promise> { + return new Set(); +} + +function openOnly(...open: string[]): LocalSidecarRecords["stillOpen"] { + return async function stillOpen(allocationIds) { + return new Set(allocationIds.filter((allocationId) => open.includes(allocationId))); + }; +} + +async function allOpen(allocationIds: readonly string[]): Promise> { + return new Set(allocationIds); +} + +const NOTHING_HELD: LocalSidecarRecords = { isLive: nothingElse, hasWork: nothingElse, stillOpen: allOpen }; + +test("restore respawns the sidecar once for the holds still live and drops the rest", async () => { + const { dataRoot, manifests, spawned, logs, create } = await setup(); + const stored: LocalSidecarManifest = { sidecarId: "sc_first", token: "token_first", hubWebSocketUrl: "ws://hub", dataDir: path.join(dataRoot, "sidecar") }; + const holds: LocalSidecarHold[] = [ + { allocationId: "listener", generation: 1, offeredSidecarId: "sc_first" }, + { allocationId: "manager", generation: 3, offeredSidecarId: "sc_manager" }, + { allocationId: "released", generation: 1, offeredSidecarId: "sc_released" }, + ]; + await manifests.write(stored); + await manifests.writeHolds(holds); + const local = create(); + + const asked: string[] = []; + async function isLive(hold: LocalSidecarHold & { sidecarId: string }): Promise { + asked.push(`${hold.allocationId}@${String(hold.generation)} on ${hold.sidecarId}`); + return hold.allocationId !== "released"; + } + await local.restore({ ...NOTHING_HELD, isLive }); + + expect(asked).toEqual(["listener@1 on sc_first", "manager@3 on sc_first", "released@1 on sc_first"]); + expect(spawned).toEqual([{ manifest: stored, adapterManifest: HUB_ADAPTERS }]); + expect(await manifests.readHolds()).toEqual(holds.slice(0, 2)); + expect(logs).toContainEqual(expect.objectContaining({ msg: "local_sidecar_respawned", sidecarId: "sc_first", holds: 2, dropped: 1 })); + expect(await local.provisioner.ensure(request("next"))).toEqual({ kind: "accepted", externalRef: "1", sidecarId: "sc_first" }); + expect(spawned).toHaveLength(1); + + await local.shutdown(); +}); + +test("restore with no live hold and no copy left to remove forgets the sidecar, and the next ensure starts one under its own identity", async () => { + const { dataRoot, manifests, spawned, create } = await setup(); + const dataDir = path.join(dataRoot, "sidecar"); + await fs.mkdir(dataDir); + await manifests.write({ sidecarId: "sc_first", token: "token_first", hubWebSocketUrl: "ws://hub", dataDir }); + await manifests.writeHolds([{ allocationId: "released", generation: 1, offeredSidecarId: "sc_first" }]); + const local = create(); + + await local.restore(NOTHING_HELD); + + expect(spawned).toEqual([]); + expect(await manifests.read()).toBeNull(); + expect(await fs.readdir(dataRoot)).toEqual(["holds.json"]); + expect(await local.provisioner.ensure(request("next"))).toEqual({ kind: "accepted", externalRef: "1" }); + expect(spawned.map(({ manifest }) => manifest.sidecarId)).toEqual(["sc_next"]); + + await local.shutdown(); +}); + +test("restore brings the sidecar back while the hub still waits for it to remove a released copy", async () => { + const { dataRoot, manifests, spawned, create } = await setup(); + const stored: LocalSidecarManifest = { sidecarId: "sc_first", token: "token_first", hubWebSocketUrl: "ws://hub", dataDir: path.join(dataRoot, "sidecar") }; + await manifests.write(stored); + await manifests.writeHolds([]); + const local = create(); + + await local.restore({ ...NOTHING_HELD, hasWork: async (sidecarId) => sidecarId === "sc_first" }); + + expect(spawned).toEqual([{ manifest: stored, adapterManifest: HUB_ADAPTERS }]); + expect(await manifests.read()).toEqual(stored); + + await local.shutdown(); +}); + +test("a late ensure for a destroyed identity, or another identity at a live generation, is refused", async () => { + const { started } = await setup(); + const local = await started(); + + await local.provisioner.ensure(request("a")); + await local.provisioner.destroy({ allocationId: "a", generation: 2, sidecarId: "sc_a" }); + expect(await local.provisioner.ensure({ ...request("a", 2), sidecarId: "sc_a" })).toMatchObject({ kind: "rejected", code: "generation_destroyed" }); + expect((await local.provisioner.ensure({ ...request("a", 2), sidecarId: "sc_a2" })).kind).toBe("accepted"); + expect(await local.provisioner.ensure({ ...request("a", 2), sidecarId: "sc_other" })).toMatchObject({ kind: "rejected", code: "sidecar_identity_conflict" }); + + await local.shutdown(); +}); + +test("a destroy older than the allocation's hold leaves the live generation held", async () => { + const { manifests, started } = await setup(); + const local = await started(); + + await local.provisioner.ensure(request("listener")); + await local.provisioner.ensure(request("a", 2)); + expect(await local.provisioner.destroy({ allocationId: "a", generation: 1, sidecarId: "sc_listener" })).toEqual({ kind: "destroyed", cleanup: "required" }); + expect(await manifests.readHolds()).toContainEqual({ allocationId: "a", generation: 2, offeredSidecarId: "sc_a" }); + expect(await local.provisioner.ensure(request("a", 2))).toMatchObject({ kind: "accepted", sidecarId: "sc_listener" }); + + await local.shutdown(); +}); + +test("an unreadable holds file is removed and restore goes on without it", async () => { + const { dataRoot, create } = await setup(); + await fs.writeFile(path.join(dataRoot, "holds.json"), "[{\"allocationId\":"); + const local = create(); + + await local.restore(NOTHING_HELD); + + expect(await fs.readdir(dataRoot)).toEqual([]); + + await local.shutdown(); +}); + +test("restore retires the sidecars a hub with one per allocation left and removes their files, keeping one it could not retire for the next boot", async () => { + const { dataRoot, logs, create } = await setup(); + await fs.mkdir(path.join(dataRoot, "sal_orphan-Ab12Cd")); + const untouched = create(); + await untouched.restore(NOTHING_HELD); + expect(await fs.readdir(dataRoot)).toContain("sal_orphan-Ab12Cd"); + + const legacyDir = path.join(dataRoot, "manifests"); + await fs.mkdir(legacyDir); + for (const allocationId of ["sal_old", "sal_stuck"]) { + const legacy = { allocationId, generation: 1, sidecarId: `sc_${allocationId}`, tenantId: "tnt_1", anchorRunId: `run_${allocationId}`, hubWebSocketUrl: "ws://hub", dataDir: path.join(dataRoot, `${allocationId}-Xy34Zw`), token: "sealed" }; + await fs.writeFile(path.join(legacyDir, `${allocationId}.json`), JSON.stringify(legacy)); + await fs.mkdir(legacy.dataDir); + } + await fs.writeFile(path.join(legacyDir, "sal_torn.json"), "{"); + await fs.mkdir(path.join(dataRoot, "sidecar")); + const reported: unknown[] = []; + const local = create(); + local.provisioner.attach?.({ + async sidecarStopped(sidecarId, options) { + reported.push({ sidecarId, options }); + if (sidecarId === "sc_sal_stuck") throw new Error("database unavailable"); + return { kind: "retired" }; }, - log: () => {}, }); - try { - const burst = await Promise.all(["a", "b", "c"].map((id) => local.provisioner.ensure(request(id)))); - const declined = burst.filter((result) => result.kind === "rejected"); - expect(declined).toEqual([expect.objectContaining({ code: CAPACITY_UNAVAILABLE, retryable: true })]); - const [first, second] = [...exits.keys()]; - expect((await fs.readdir(dataRoot)).length).toBe(2); - - local.connected({ allocationId: first!, generation: 1 }); - expect((await local.provisioner.ensure(request("d"))).kind).toBe("accepted"); - expect((await local.provisioner.ensure(request("e"))).kind).toBe("rejected"); - exits.get(second!)!(); - await Bun.sleep(0); - expect((await local.provisioner.ensure(request("e"))).kind).toBe("accepted"); - } finally { - await fs.rm(dataRoot, { recursive: true, force: true }); - } + await local.restore(NOTHING_HELD); + + expect(reported).toEqual([ + { sidecarId: "sc_sal_old", options: { endReconnectWindow: true } }, + { sidecarId: "sc_sal_stuck", options: { endReconnectWindow: true } }, + ]); + expect((await fs.readdir(dataRoot)).sort()).toEqual(["manifests", "sal_stuck-Xy34Zw", "sidecar"]); + expect(await fs.readdir(legacyDir)).toEqual(["sal_stuck.json"]); + expect(logs).toContainEqual(expect.objectContaining({ msg: "local_sidecar_legacy_retire_failed", allocationId: "sal_stuck", sidecarId: "sc_sal_stuck" })); + expect(logs).toContainEqual({ + level: "info", + msg: "local_sidecar_legacy_layout_removed", + retired: 1, + removed: [path.join("manifests", "sal_old.json"), path.join("manifests", "sal_torn.json"), "sal_old-Xy34Zw", "sal_orphan-Ab12Cd"], + }); +}); + +test("a destroy with no hold is fenced across a restart, and the file drops it once stock has ended the allocation", async () => { + const { manifests, create, started } = await setup(); + const before = await started({ ...NOTHING_HELD, stillOpen: openOnly("a") }); + expect(await before.provisioner.destroy({ allocationId: "a", generation: 1, sidecarId: "sc_a" })).toEqual({ kind: "destroyed", cleanup: "confirmed" }); + await before.shutdown(); + + const after = create(); + await after.restore({ ...NOTHING_HELD, stillOpen: openOnly("a") }); + expect(await after.provisioner.ensure(request("a"))).toMatchObject({ kind: "rejected", code: "generation_destroyed" }); + await after.shutdown(); + + const ended = create(); + await ended.restore({ ...NOTHING_HELD, stillOpen: noneOpen }); + expect(await manifests.readFences()).toEqual([]); + await ended.shutdown(); +}); + +test("a corrupt fences file is moved aside, and a late ensure for an allocation stock ended is refused from its record", async () => { + const { dataRoot, logs, started } = await setup(); + await fs.writeFile(path.join(dataRoot, "fences.json"), "[{\"allocationId\":"); + const local = await started({ ...NOTHING_HELD, stillOpen: openOnly("b") }); + + expect(await local.provisioner.ensure({ ...request("a", 2), sidecarId: "sc_late" })).toMatchObject({ kind: "rejected", code: "allocation_ended" }); + expect((await local.provisioner.ensure(request("b"))).kind).toBe("accepted"); + const files = await fs.readdir(dataRoot); + expect(files.filter((name) => name.startsWith("fences.json.corrupt-"))).toHaveLength(1); + expect(logs).toContainEqual(expect.objectContaining({ level: "warn", msg: "local_sidecar_fences_unreadable_moved" })); + await local.shutdown(); +}); + +test("a joined allocation's destroy is required whatever identity it names, and a repeat of it too", async () => { + const { started } = await setup(); + const local = await started(); + await local.provisioner.ensure(request("a")); + await local.provisioner.ensure(request("b")); + + // Stock kept the identity it offered `b`, though `b` joined `sc_a`. + expect(await local.provisioner.destroy({ allocationId: "b", generation: 2, sidecarId: "sc_b" })).toEqual({ kind: "destroyed", cleanup: "required" }); + expect(await local.provisioner.destroy({ allocationId: "b", generation: 2, sidecarId: "sc_b" })).toEqual({ kind: "destroyed", cleanup: "required" }); + expect(await local.provisioner.destroy({ allocationId: "b", generation: 1, sidecarId: "sc_b" })).toEqual({ kind: "destroyed", cleanup: "confirmed" }); + await local.shutdown(); +}); + +test("a repeated destroy keeps fencing the identity the hold was offered under", async () => { + const { started } = await setup(); + const local = await started(); + await local.provisioner.ensure(request("a")); + await local.provisioner.ensure(request("b")); + + await local.provisioner.destroy({ allocationId: "b", generation: 1, sidecarId: "sc_a" }); + await local.provisioner.destroy({ allocationId: "b", generation: 1, sidecarId: "sc_a" }); + + expect(await local.provisioner.ensure(request("b"))).toMatchObject({ kind: "rejected", code: "generation_destroyed" }); + await local.shutdown(); +}); + +test("an ensure that fails after starting the sidecar leaves the hold as it was and its identity unused, so the next ensure mints another", async () => { + const { manifests, spawned, accepted, faults, started } = await setup(); + const local = await started(); + faults.holdsWrite = true; + + await expect(local.provisioner.ensure(request("a"))).rejects.toThrow("disk full"); + faults.holdsWrite = false; + expect(await manifests.read()).toBeNull(); + expect(await local.provisioner.ensure(request("b"))).toEqual({ kind: "accepted", externalRef: "2" }); + + expect(spawned.map(({ manifest }) => manifest.sidecarId)).toEqual(["sc_a", "sc_b"]); + expect(await manifests.readHolds()).toEqual([{ allocationId: "b", generation: 1, offeredSidecarId: "sc_b" }]); + expect(accepted).toEqual([{ allocationId: "b", generation: 1 }]); + await local.shutdown(); +}); + +test("each allocation that joins the sidecar is reported placed, with no connection of its own", async () => { + const { accepted, started } = await setup(); + const local = await started(); + + await local.provisioner.ensure(request("listener")); + await local.provisioner.ensure(request("manager", 3)); + + expect(accepted).toEqual([{ allocationId: "listener", generation: 1 }, { allocationId: "manager", generation: 3 }]); + await local.shutdown(); +}); + +test("an ensure or destroy that arrives before restore waits for it and answers from the restored sidecar", async () => { + const { dataRoot, manifests, spawned, create } = await setup(); + const stored: LocalSidecarManifest = { sidecarId: "sc_first", token: "token_first", hubWebSocketUrl: "ws://hub", dataDir: path.join(dataRoot, "sidecar") }; + await manifests.write(stored); + await manifests.writeHolds([{ allocationId: "listener", generation: 1, offeredSidecarId: "sc_first" }]); + const local = create(); + + const ensured = local.provisioner.ensure(request("early")); + const destroyed = local.provisioner.destroy({ allocationId: "listener", generation: 2, sidecarId: "sc_first" }); + await local.restore({ ...NOTHING_HELD, isLive: async () => true }); + + expect(await ensured).toEqual({ kind: "accepted", externalRef: "1", sidecarId: "sc_first" }); + expect(await destroyed).toEqual({ kind: "destroyed", cleanup: "required" }); + expect(spawned.map(({ manifest }) => manifest.sidecarId)).toEqual(["sc_first"]); + await local.shutdown(); +}); + +test("each fence write drops the fences of allocations stock has ended", async () => { + const { manifests, create } = await setup(); + const local = create(); + const open = new Set(["a", "b"]); + await local.restore({ ...NOTHING_HELD, stillOpen: async (allocationIds) => new Set(allocationIds.filter((allocationId) => open.has(allocationId))) }); + + await local.provisioner.destroy({ allocationId: "a", generation: 1, sidecarId: "sc_a" }); + expect(await manifests.readFences()).toEqual([{ allocationId: "a", generation: 1, sidecarIds: ["sc_a"] }]); + open.delete("a"); + await local.provisioner.destroy({ allocationId: "b", generation: 1, sidecarId: "sc_b" }); + + expect(await manifests.readFences()).toEqual([{ allocationId: "b", generation: 1, sidecarIds: ["sc_b"] }]); + await local.shutdown(); }); diff --git a/apps/hub/src/sidecar/local-provisioner.ts b/apps/hub/src/sidecar/local-provisioner.ts index 3fb9ff165..66af74665 100644 --- a/apps/hub/src/sidecar/local-provisioner.ts +++ b/apps/hub/src/sidecar/local-provisioner.ts @@ -1,8 +1,6 @@ -// Production copy of the provisioning logic in -// vendor/interchange/tests/admin-ui-e2e/harness/local-process-sidecar-provisioner.ts. -// The hub must not import from the vendor test harness: tests/ trees are not -// shipped and may change without notice. Unlike the harness, the caller always -// supplies the spawn so the child env is built from validated hub config. +// One sidecar process on the hub's machine hosts every allocation placed here. The first ensure starts it under the +// identity the hub offered; every later ensure answers with that sidecar, which stock Interchange then deploys into +// over the same connection, up to the active deployments the sidecar advertises in its hello. import fs from "node:fs/promises"; import path from "node:path"; import { errorText } from "@corbits/common"; @@ -12,23 +10,27 @@ import type { EnsureSidecarRequest, EnsureSidecarResult, SidecarProvisioner, + SidecarProvisionerHost, } from "@intx/hub-sessions"; import type { AdapterManifest } from "@intx/types"; import { CAPACITY_UNAVAILABLE, shortfall, type HostCapacity } from "./host-capacity.js"; -import type { LocalSidecarManifest, LocalSidecarManifestStore } from "./manifest-store.js"; +import type { LocalSidecarFence, LocalSidecarHold, LocalSidecarManifest, LocalSidecarManifestStore } from "./manifest-store.js"; const DEFAULT_STOP_TIMEOUT_MS = 25_000; -const DATA_DIR_NAME = /^(.+)-[A-Za-z0-9]{6}$/; +const DATA_DIR_NAME = "sidecar"; + +/** How the process ended: its exit code, or the signal that killed it. */ +export type LocalSidecarExit = { readonly code: number | null; readonly signal: string | null }; interface LocalSidecarProcess { readonly pid: number; - readonly exited: Promise; + readonly exited: Promise; kill(signal: NodeJS.Signals): void; } export type SpawnLocalSidecar = (args: { - readonly request: EnsureSidecarRequest; - readonly dataDir: string; + readonly manifest: LocalSidecarManifest; + readonly adapterManifest: AdapterManifest; }) => LocalSidecarProcess; export type CreateLocalProcessSidecarProvisionerOpts = { @@ -36,20 +38,26 @@ export type CreateLocalProcessSidecarProvisionerOpts = { readonly spawnSidecar: SpawnLocalSidecar; readonly manifests: LocalSidecarManifestStore; readonly capacity: HostCapacity; - /** Operator config, so a restored sidecar loads the hub's adapters, not the ones its record was written with. */ + /** Operator config, so a respawned sidecar loads the hub's adapters, not the ones it first started with. */ readonly adapterManifest: AdapterManifest; readonly stopTimeoutMs?: number; + /** Called when an allocation joins the sidecar, which later joins announce with no connection of their own. */ + readonly onAccepted: (placement: { readonly allocationId: string; readonly generation: number }) => void; readonly log: (entry: Record) => void; }; - -type IsLocalSidecarLive = (manifest: LocalSidecarManifest) => Promise; +export type LocalSidecarRecords = { + /** Whether the hub still holds this allocation, at this generation, on the local sidecar. */ + isLive(hold: LocalSidecarHold & { readonly sidecarId: string }): 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>; +}; export interface LocalProcessSidecarProvisioner { readonly provisioner: SidecarProvisioner; - restore(isLive: IsLocalSidecarLive): Promise; - /** The hub saw this allocation's sidecar connect, so its memory now shows in the box's own figures. */ - connected(allocation: { allocationId: string; generation: number }): void; + restore(records: LocalSidecarRecords): Promise; shutdown(): Promise; } @@ -58,30 +66,12 @@ type ManagedProcess = { exited: boolean; }; -type AllocationState = - | { - readonly kind: "live"; - readonly generation: number; - readonly sidecarId: string; - readonly dataDir: string; - readonly process: ManagedProcess; - } - | { - readonly kind: "destroyed"; - readonly generation: number; - readonly sidecarId: string; - }; - - async function waitForExit(managed: ManagedProcess): Promise { await managed.handle.exited; return true; } -async function exitedWithin( - managed: ManagedProcess, - timeoutMs: number, -): Promise { +async function exitedWithin(managed: ManagedProcess, timeoutMs: number): Promise { if (managed.exited) return true; let timer: ReturnType | undefined; const timedOut = new Promise(function scheduleTimeout(resolve) { @@ -94,20 +84,11 @@ async function exitedWithin( } } -async function trackExit(managed: ManagedProcess, onExit?: () => void): Promise { - try { - await managed.handle.exited; - } finally { - managed.exited = true; - onExit?.(); - } -} - async function settle(operation: Promise): Promise { try { await operation; } catch { - // The caller observes the failure; the per-allocation queue only needs ordering. + // The caller observes the failure; the queue only needs ordering. } } @@ -118,278 +99,307 @@ export function createLocalProcessSidecarProvisioner({ capacity, adapterManifest, stopTimeoutMs = DEFAULT_STOP_TIMEOUT_MS, + onAccepted, log, }: CreateLocalProcessSidecarProvisionerOpts): LocalProcessSidecarProvisioner { if (stopTimeoutMs <= 0) { throw new Error("Local sidecar stop timeout must be positive"); } - function logAllocation(msg: string, manifest: LocalSidecarManifest): void { - log({ - level: "info", - msg, - allocationId: manifest.allocationId, - generation: manifest.generation, - sidecarId: manifest.sidecarId, - }); - } - - const allocations = new Map(); - const operations = new Map>(); - /** Sidecars started and not yet connected or exited, by allocation and generation. */ - const starting = new Set(); + const dataDir = path.join(dataRoot, DATA_DIR_NAME); + const holds = new Map(); + const fences = new Map>(); + let manifest: LocalSidecarManifest | null = null; + let running: ManagedProcess | null = null; + let queue: Promise = Promise.resolve(); let shutdownPromise: Promise | null = null; - - function startingKey({ allocationId, generation }: { allocationId: string; generation: number }): string { - return `${allocationId}@${String(generation)}`; + let host: SidecarProvisionerHost | undefined; + let stillOpen: LocalSidecarRecords["stillOpen"] | 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. + const restored = Promise.withResolvers(); + void settle(restored.promise); + + // Every operation runs alone: the sidecar starts once, and the holds file is written whole. + async function serialize(request: R, run: (request: R) => Promise): Promise { + if (shutdownPromise !== null) { + throw new Error("Local sidecar provisioner is shutting down"); + } + const previous = queue; + const pending = runAfter(previous, request, run); + queue = settle(pending); + return pending; } - async function runAfter( - previous: Promise | undefined, - request: R, - run: (request: R) => Promise, - ): Promise { + async function runAfter(previous: Promise, request: R, run: (request: R) => Promise): Promise { await previous; return run(request); } - async function releaseWhenSettled(allocationId: string, settled: Promise): Promise { - await settled; - if (operations.get(allocationId) === settled) operations.delete(allocationId); + async function persistHolds(): Promise { + await manifests.writeHolds([...holds.values()]); } - async function serialize( - request: R, - run: (request: R) => Promise, - ): Promise { - if (shutdownPromise !== null) { - throw new Error("Local sidecar provisioner is shutting down"); - } - const pending = runAfter(operations.get(request.allocationId), request, run); - const settled = settle(pending); - operations.set(request.allocationId, settled); - void releaseWhenSettled(request.allocationId, settled); - return pending; + function openRecords(): LocalSidecarRecords["stillOpen"] { + if (stillOpen === undefined) throw new Error("Local sidecar provisioner has not been restored"); + return stillOpen; } - async function stopProcess( - state: Extract, - ): Promise { - if (!state.process.exited) { - try { - state.process.handle.kill("SIGTERM"); - } catch (error) { - if (!state.process.exited) throw error; - } - if (!(await exitedWithin(state.process, stopTimeoutMs))) { - state.process.handle.kill("SIGKILL"); - if (!(await exitedWithin(state.process, stopTimeoutMs))) { - throw new Error( - `Local sidecar process ${String(state.process.handle.pid)} did not exit`, - ); - } - } + // 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, sidecarId: string): Promise { + try { + const { code, signal } = await managed.handle.exited; + log({ level: shutdownPromise === null ? "error" : "info", msg: "local_sidecar_exited", sidecarId, pid: managed.handle.pid, code, signal }); + } finally { + managed.exited = true; } } - async function removeStored(allocationId: string, dataDir: string): Promise { - await manifests.remove(allocationId); - await fs.rm(dataDir, { recursive: true, force: true }); + function start(current: LocalSidecarManifest): ManagedProcess { + const managed: ManagedProcess = { handle: spawnSidecar({ manifest: current, adapterManifest }), exited: false }; + void trackExit(managed, current.sidecarId); + running = managed; + return managed; } - async function stopAndRemove( - allocationId: string, - state: Extract, - ): Promise { - await stopProcess(state); - await removeStored(allocationId, state.dataDir); + /** The running sidecar, started under `request`'s identity when none was minted yet. */ + async function sidecarFor(request: EnsureSidecarRequest): Promise< + | { readonly kind: "running"; readonly manifest: LocalSidecarManifest; readonly process: ManagedProcess; readonly minted: boolean } + | Extract + > { + if (manifest !== null && running !== null && !running.exited) { + return { kind: "running", manifest, process: running, minted: false }; + } + const full = shortfall(await capacity.freeBytes()); + if (full !== undefined) { + return { kind: "rejected", code: CAPACITY_UNAVAILABLE, message: full, retryable: true }; + } + request.signal?.throwIfAborted(); + const minted = manifest === null; + const current = manifest ?? { sidecarId: request.sidecarId, token: request.token, hubWebSocketUrl: request.hubWebSocketUrl, dataDir }; + try { + if (minted) { + await fs.mkdir(dataDir, { recursive: true }); + await manifests.write(current); + manifest = current; + } + const child = start(current); + log({ level: "info", msg: minted ? "local_sidecar_started" : "local_sidecar_respawned", sidecarId: current.sidecarId, pid: child.handle.pid }); + return { kind: "running", manifest: current, process: child, minted }; + } catch (error) { + // A rejected ensure retires the identity it was offered, so a sidecar minted from it must not outlive it. + if (minted) { + manifest = null; + await manifests.remove(); + } + return { kind: "rejected", code: "spawn_failed", message: errorText(error), retryable: true }; + } } async function ensureAllocation(request: EnsureSidecarRequest): Promise { request.signal?.throwIfAborted(); - const existing = allocations.get(request.allocationId); - if (existing !== undefined && existing.generation > request.generation) { + const live = holds.get(request.allocationId); + const fence = fences.get(request.allocationId); + const newest = live?.generation ?? fence?.generation; + if (newest !== undefined && newest > request.generation) { return { - kind: "rejected" as const, + kind: "rejected", code: "stale_generation", - message: `Generation ${String(request.generation)} is older than ${String(existing.generation)}`, + message: `Generation ${String(request.generation)} is older than ${String(newest)}`, retryable: false, }; } - // Replacement destroys the old identity at the new generation before - // ensuring a fresh identity at that same generation. Fence only the exact - // identity that was destroyed. - if ( - existing?.kind === "destroyed" && - existing.generation === request.generation && - existing.sidecarId === request.sidecarId - ) { + // 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" as const, + kind: "rejected", code: "generation_destroyed", message: `Generation ${String(request.generation)} was already destroyed`, retryable: false, }; } - if ( - existing?.kind === "live" && - existing.generation === request.generation && - existing.sidecarId !== request.sidecarId - ) { + if (live !== undefined && live.generation === request.generation && live.offeredSidecarId !== request.sidecarId) { return { - kind: "rejected" as const, + kind: "rejected", code: "sidecar_identity_conflict", message: `Generation ${String(request.generation)} already belongs to another sidecar identity`, retryable: false, }; } - if ( - existing?.kind === "live" && - existing.generation === request.generation && - !existing.process.exited - ) { - return { - kind: "accepted" as const, - externalRef: String(existing.process.handle.pid), - }; - } - if (existing?.kind === "live") { - await stopAndRemove(request.allocationId, existing); - } - // Decided and counted in one turn, so ensures running side by side each see the others. - const full = shortfall(await capacity.freeBytes(), starting.size); - if (full !== undefined) { - return { kind: "rejected" as const, code: CAPACITY_UNAVAILABLE, message: full, retryable: true }; - } - const key = startingKey(request); - starting.add(key); - function startedExited(): void { - starting.delete(key); + // 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 }; } - let dataDir: string | undefined; - let live: Extract | undefined; + const sidecar = await sidecarFor(request); + if (sidecar.kind === "rejected") return sidecar; try { - await fs.mkdir(dataRoot, { recursive: true }); - dataDir = await fs.mkdtemp(path.join(dataRoot, `${request.allocationId}-`)); + holds.set(request.allocationId, { allocationId: request.allocationId, generation: request.generation, offeredSidecarId: request.sidecarId }); + fences.delete(request.allocationId); + await persistHolds(); + if (fence !== undefined) await persistFences(); request.signal?.throwIfAborted(); - const handle = spawnSidecar({ request, dataDir }); - const managed: ManagedProcess = { handle, exited: false }; - void trackExit(managed, startedExited); - live = { - kind: "live", - generation: request.generation, - sidecarId: request.sidecarId, - dataDir, - process: managed, - }; - allocations.set(request.allocationId, live); - await manifests.write({ - allocationId: request.allocationId, - generation: request.generation, - sidecarId: request.sidecarId, - tenantId: request.tenantId, - anchorRunId: request.anchorRunId, - hubWebSocketUrl: request.hubWebSocketUrl, - dataDir, - token: request.token, - }); - return { kind: "accepted" as const, externalRef: String(handle.pid) }; } catch (error) { - starting.delete(key); - if (live !== undefined) { - allocations.delete(request.allocationId); - await stopAndRemove(request.allocationId, live); - } else if (dataDir !== undefined) { - await fs.rm(dataDir, { recursive: true, force: true }); - } - return { - kind: "rejected" as const, - code: "spawn_failed", - message: errorText(error), - retryable: true, - }; + await abandonEnsure(request.allocationId, { live, fence }, sidecar); + throw error; } + onAccepted({ allocationId: request.allocationId, generation: request.generation }); + return { + kind: "accepted", + externalRef: String(sidecar.process.handle.pid), + ...(sidecar.manifest.sidecarId !== request.sidecarId && { sidecarId: sidecar.manifest.sidecarId }), + }; } - // Each allocation runs in its own process, and a newer generation or identity - // stopped and removed the older one first, so every destroy confirms cleanup. - async function destroyAllocation(request: DestroySidecarRequest): Promise { - const existing = allocations.get(request.allocationId); - if (existing !== undefined && existing.generation > request.generation) { - return { kind: "destroyed", cleanup: "confirmed" }; - } - // A delayed destroy for the superseded identity must not terminate the - // replacement that already owns this generation. - if (existing !== undefined && existing.sidecarId !== request.sidecarId) { - return { kind: "destroyed", cleanup: "confirmed" }; + // Stock retires the identity of an ensure that failed or that it stopped awaiting, so the hold goes back to what it + // 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 }, + 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 (sidecar.minted) { + await stopProcess(sidecar.process); + running = null; + manifest = null; + await forget(); } - if (existing?.kind === "live") { - await stopAndRemove(request.allocationId, existing); + try { + await persistHolds(); + await persistFences(); + } catch (error) { + log({ level: "error", msg: "local_sidecar_rollback_unpersisted", allocationId, error: errorText(error) }); } - if (existing === undefined) { - // A previous hub process may have left this allocation's state on disk. - const dataDir = await manifests.readDataDir(request.allocationId); - if (dataDir !== null) await removeStored(request.allocationId, dataDir); + } + + // The sidecar outlives every hold, so a copy it hosts is removed by the hub over its connection, which confirms the + // cleanup. Whether it hosts one follows from the hold, whatever identity the destroy names; an identity offered for + // an allocation that never joined had nothing deployed under it. A destroy naming the sidecar's own id is always + // `required`: stock retires an identity whose cleanup a destroy confirms, which would stop every placement on it. + async function destroyAllocation(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 }) }); + holds.delete(request.allocationId); + await pruneFences(); + await persistFences(); + if (live !== undefined) await persistHolds(); } - allocations.set(request.allocationId, { - kind: "destroyed", - generation: request.generation, - sidecarId: request.sidecarId, - }); - return { kind: "destroyed", cleanup: "confirmed" }; + return { kind: "destroyed", cleanup: hosted || request.sidecarId === manifest?.sidecarId ? "required" : "confirmed" }; } - // Ensure or destroy may reach an allocation before restore does; their - // outcome wins. - async function restoreAllocation( - { manifest, isLive }: { readonly allocationId: string; readonly manifest: LocalSidecarManifest; readonly isLive: IsLocalSidecarLive }, - ): Promise { - if (allocations.has(manifest.allocationId)) return; - if (!(await isLive(manifest))) { - await removeStored(manifest.allocationId, manifest.dataDir); - logAllocation("local_sidecar_removed", manifest); - return; + function attached(): SidecarProvisionerHost { + if (host === undefined) throw new Error("Local sidecar provisioner is not attached to the hub"); + return host; + } + + // A hub that ran one sidecar per allocation stopped them all at shutdown, and none comes back. Retiring each identity + // gives its allocations no reconnect window, so the reconciler replaces them at once; a sidecar the hub could not + // retire keeps its files for the next boot to try again. + async function retireLegacySidecars(): Promise { + const legacy = await manifests.legacySidecars(); + const kept = new Set(); + if (legacy.length > 0) { + const hub = attached(); + for (const { allocationId, sidecarId } of legacy) { + try { + await hub.sidecarStopped(sidecarId, { endReconnectWindow: true }); + } catch (error) { + kept.add(allocationId); + log({ level: "warn", msg: "local_sidecar_legacy_retire_failed", allocationId, sidecarId, error: errorText(error) }); + } + } } - const request = { ...manifest, adapterManifest }; - const handle = spawnSidecar({ request, dataDir: manifest.dataDir }); - const managed: ManagedProcess = { handle, exited: false }; - void trackExit(managed); - allocations.set(manifest.allocationId, { - kind: "live", - generation: manifest.generation, - sidecarId: manifest.sidecarId, - dataDir: manifest.dataDir, - process: managed, - }); - logAllocation("local_sidecar_respawned", manifest); + const removed = await manifests.removeLegacyLayout(kept); + if (removed.length > 0) log({ level: "info", msg: "local_sidecar_legacy_layout_removed", retired: legacy.length - kept.size, removed }); } - async function removeOrphan( - { allocationId, dataDir }: { readonly allocationId: string; readonly dataDir: string }, - ): Promise { - const state = allocations.get(allocationId); - if (state?.kind === "live" && state.dataDir === dataDir) return; - if ((await manifests.readDataDir(allocationId)) === dataDir) return; + // 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 forget(): Promise { + await manifests.remove(); + await manifests.writeHolds([]); await fs.rm(dataDir, { recursive: true, force: true }); - log({ level: "info", msg: "local_sidecar_orphan_removed", allocationId, dataDir }); } - // A hub that stopped between creating a data dir and writing its manifest - // leaves a dir no manifest names, which destroy would never remove. - async function removeOrphans(): Promise { - await fs.mkdir(dataRoot, { recursive: true }); - for (const entry of await fs.readdir(dataRoot, { withFileTypes: true })) { - const allocationId = entry.isDirectory() ? DATA_DIR_NAME.exec(entry.name)?.[1] : undefined; - if (allocationId === undefined) continue; - await serialize({ allocationId, dataDir: path.join(dataRoot, entry.name) }, removeOrphan); + // A copy whose hold a destroy released is removed only once the sidecar reconnects and the hub undeploys it, so the + // sidecar comes back for that cleanup too; forgetting it would leave the copy holding a slot for good. + async function restoreSidecar(records: LocalSidecarRecords): Promise { + const { isLive, hasWork } = records; + stillOpen = records.stillOpen; + await retireLegacySidecars(); + await restoreFences(); + const stored = await manifests.read(); + const recorded = await manifests.readHolds(); + if (stored === null) { + if (recorded.length > 0) await forget(); + return; + } + const live: LocalSidecarHold[] = []; + for (const hold of recorded) { + if (await isLive({ ...hold, sidecarId: stored.sidecarId })) live.push(hold); + } + const dropped = recorded.length - live.length; + if (live.length === 0 && !(await hasWork(stored.sidecarId))) { + await forget(); + log({ level: "info", msg: "local_sidecar_removed", sidecarId: stored.sidecarId, dropped }); + return; } + manifest = stored; + for (const hold of live) holds.set(hold.allocationId, hold); + await manifests.writeHolds(live); + const child = start(stored); + log({ level: "info", msg: "local_sidecar_respawned", sidecarId: stored.sidecarId, pid: child.handle.pid, holds: live.length, dropped }); } - async function restore(isLive: IsLocalSidecarLive): Promise { - for (const manifest of await manifests.readAll()) { - await serialize({ allocationId: manifest.allocationId, manifest, isLive }, restoreAllocation); + async function stopProcess(managed: ManagedProcess): Promise { + if (managed.exited) return; + try { + managed.handle.kill("SIGTERM"); + } catch (error) { + if (!managed.exited) throw error; + } + if (await exitedWithin(managed, stopTimeoutMs)) return; + managed.handle.kill("SIGKILL"); + if (!(await exitedWithin(managed, stopTimeoutMs))) { + throw new Error(`Local sidecar process ${String(managed.handle.pid)} did not exit`); } - await removeOrphans(); + } + + // The manifest, holds and data dir survive shutdown so the next hub boot can respawn the sidecar. + async function stopAll(): Promise { + await queue; + if (running !== null) await stopProcess(running); + running = null; } const provisioner: SidecarProvisioner = { @@ -397,37 +407,32 @@ export function createLocalProcessSidecarProvisioner({ apiVersion: 2, bindingFingerprint: "local-process:v1", capabilities: [], - // Keep cleanup behind any late ensure for the same allocation, including - // work the Hub stopped awaiting after an operation timeout. - ensure(request) { + attach(hub) { + host = hub; + }, + async ensure(request) { + await restored.promise; return serialize(request, ensureAllocation); }, - destroy(request) { + async destroy(request) { + await restored.promise; return serialize(request, destroyAllocation); }, }; - // Data dirs and manifests survive shutdown so the next hub boot can respawn. - async function stopAll(): Promise { - await Promise.all(operations.values()); - const live = [...allocations.values()].filter((state) => state.kind === "live"); - const results = await Promise.allSettled(live.map(stopProcess)); - const failures = results.filter((result) => result.status === "rejected").map((result) => result.reason); - allocations.clear(); - if (failures.length > 0) { - throw new AggregateError( - failures, - "Failed to stop every local sidecar process", - ); + async function restoreOnce(records: LocalSidecarRecords): Promise { + try { + await serialize(records, restoreSidecar); + restored.resolve(); + } catch (error) { + restored.reject(error); + throw error; } } return { provisioner, - restore, - connected(allocation) { - starting.delete(startingKey(allocation)); - }, + restore: restoreOnce, shutdown() { shutdownPromise ??= stopAll(); return shutdownPromise; diff --git a/apps/hub/src/sidecar/manifest-store.ts b/apps/hub/src/sidecar/manifest-store.ts index 96c0c99ef..9c1cbc37d 100644 --- a/apps/hub/src/sidecar/manifest-store.ts +++ b/apps/hub/src/sidecar/manifest-store.ts @@ -1,25 +1,54 @@ import { randomBytes } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; +import { type } from "arktype"; import { errorText } from "@corbits/common"; import type { CredentialCipher } from "@intx/types"; +/** The hub's one local sidecar: the identity it authenticates with and where it keeps its state. */ export type LocalSidecarManifest = { - readonly allocationId: string; - readonly generation: number; readonly sidecarId: string; - readonly tenantId: string; - readonly anchorRunId: string; readonly hubWebSocketUrl: string; readonly dataDir: string; readonly token: string; }; +/** An allocation the sidecar hosts, at the generation and under the identity the hub offered for it. */ +const LocalSidecarHold = type({ allocationId: "string > 0", generation: "number.integer >= 0", offeredSidecarId: "string > 0" }); +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(); + +/** A sidecar a hub that ran one per allocation started, as its manifest in `manifests/` names it. */ +const LegacySidecar = type({ allocationId: "string > 0", sidecarId: "string > 0" }); +export type LegacySidecar = typeof LegacySidecar.infer; + export interface LocalSidecarManifestStore { + read(): Promise; write(manifest: LocalSidecarManifest): Promise; - readAll(): Promise; - readDataDir(allocationId: string): Promise; - remove(allocationId: string): 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 sidecars a hub that ran one per allocation left manifests for. */ + legacySidecars(): Promise; + /** + * Removes the manifests and data directories that hub left, tokens with them, but for the allocations in `kept`; + * names what went. + */ + removeLegacyLayout(kept: ReadonlySet): Promise; } export type CreateLocalSidecarManifestStoreOpts = { @@ -28,96 +57,161 @@ export type CreateLocalSidecarManifestStoreOpts = { readonly log: (entry: Record) => void; }; -type StoredManifest = LocalSidecarManifest; +const LEGACY_MANIFESTS = "manifests"; +const LEGACY_MANIFEST = /^(.+)\.json$/; +const LEGACY_DATA_DIR = /^(.+)-[A-Za-z0-9]{6}$/; /** Domain-separated from `credentialAad` so a manifest token never decrypts as a credential secret. */ -function manifestTokenAad(allocationId: string): string { - return JSON.stringify(["sidecar-manifest-token", allocationId]); +function manifestTokenAad(sidecarId: string): string { + return JSON.stringify(["sidecar-manifest-token", sidecarId]); } function isMissing(error: unknown): boolean { return error instanceof Error && "code" in error && error.code === "ENOENT"; } +async function readOptional(file: string): Promise { + try { + return await fs.readFile(file, "utf8"); + } catch (error) { + if (isMissing(error)) return null; + throw error; + } +} + +async function exists(file: string): Promise { + try { + await fs.stat(file); + return true; + } catch (error) { + if (isMissing(error)) return false; + throw error; + } +} + +async function writeAtomically(target: string, text: string): Promise { + const temp = `${target}.${randomBytes(6).toString("hex")}.tmp`; + await fs.mkdir(path.dirname(target), { recursive: true }); + try { + await fs.writeFile(temp, text, { mode: 0o600 }); + await fs.rename(temp, target); + } catch (error) { + await fs.rm(temp, { force: true }); + throw error; + } +} + export function createLocalSidecarManifestStore({ dataRoot, cipher, log, }: CreateLocalSidecarManifestStoreOpts): LocalSidecarManifestStore { - const dir = path.join(dataRoot, "manifests"); + const manifestFile = path.join(dataRoot, "sidecar.json"); + const holdsFile = path.join(dataRoot, "holds.json"); + const fencesFile = path.join(dataRoot, "fences.json"); - function manifestPath(allocationId: string): string { - return path.join(dir, `${allocationId}.json`); + async function write(manifest: LocalSidecarManifest): Promise { + const stored = { ...manifest, token: await cipher.encrypt(manifest.token, manifestTokenAad(manifest.sidecarId)) }; + await writeAtomically(manifestFile, JSON.stringify(stored)); } - async function write(manifest: LocalSidecarManifest): Promise { - const target = manifestPath(manifest.allocationId); - const temp = `${target}.${randomBytes(6).toString("hex")}.tmp`; - const stored: StoredManifest = { - ...manifest, - token: await cipher.encrypt(manifest.token, manifestTokenAad(manifest.allocationId)), - }; - await fs.mkdir(dir, { recursive: true }); + // An unreadable manifest names no sidecar that can be respawned. + async function read(): Promise { + const text = await readOptional(manifestFile); + if (text === null) return null; try { - await fs.writeFile(temp, JSON.stringify(stored), { mode: 0o600 }); - await fs.rename(temp, target); + const stored = JSON.parse(text) as LocalSidecarManifest; + return { ...stored, token: await cipher.decrypt(stored.token, manifestTokenAad(stored.sidecarId)) }; } catch (error) { - await fs.rm(temp, { force: true }); - throw error; + await fs.rm(manifestFile, { force: true }); + log({ level: "warn", msg: "local_sidecar_manifest_unreadable_removed", file: manifestFile, error: errorText(error) }); + return null; } } - async function readAll(): Promise { - let entries: string[]; + async function remove(): Promise { + await fs.rm(manifestFile, { force: true }); + } + + // An unreadable holds file names nothing restore can bring back; the hub places that work again. + async function readHolds(): Promise { + const text = await readOptional(holdsFile); + if (text === null) return []; try { - entries = await fs.readdir(dir); + return LocalSidecarHolds.assert(JSON.parse(text)); } catch (error) { - if (isMissing(error)) return []; - throw error; + await fs.rm(holdsFile, { force: true }); + log({ level: "warn", msg: "local_sidecar_holds_unreadable_removed", file: holdsFile, error: errorText(error) }); + return []; } - const manifests: LocalSidecarManifest[] = []; - for (const entry of entries) { - if (!entry.endsWith(".json")) continue; - const manifest = await readEntry(path.join(dir, entry)); - if (manifest !== null) manifests.push(manifest); - } - return manifests; } - // An unreadable manifest cannot be respawned, so it is treated as not live. - async function readEntry(file: string): Promise { - let stored: StoredManifest | undefined; + async function writeHolds(holds: readonly LocalSidecarHold[]): Promise { + 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 { - stored = JSON.parse(await fs.readFile(file, "utf8")) as StoredManifest; - return { ...stored, token: await cipher.decrypt(stored.token, manifestTokenAad(stored.allocationId)) }; + return LocalSidecarFences.assert(JSON.parse(text)); } catch (error) { - await fs.rm(file, { force: true }); - if (typeof stored?.dataDir === "string" && path.dirname(path.resolve(stored.dataDir)) === path.resolve(dataRoot)) { - await fs.rm(stored.dataDir, { recursive: true, force: true }); - } - log({ - level: "warn", - msg: "local_sidecar_manifest_unreadable_removed", - file, - error: errorText(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 readDataDir(allocationId: string): Promise { + async function writeFences(fences: readonly LocalSidecarFence[]): Promise { + await writeAtomically(fencesFile, JSON.stringify(fences)); + } + + async function legacyEntries(dir: string): Promise { try { - const stored = JSON.parse(await fs.readFile(manifestPath(allocationId), "utf8")) as StoredManifest; - return stored.dataDir; + return (await fs.readdir(dir)).sort(); } catch (error) { - if (isMissing(error)) return null; + if (isMissing(error)) return []; throw error; } } - async function remove(allocationId: string): Promise { - await fs.rm(manifestPath(allocationId), { force: true }); + // A manifest that cannot be read names no sidecar to retire; it goes with the layout. + async function legacySidecars(): Promise { + const manifestsDir = path.join(dataRoot, LEGACY_MANIFESTS); + const sidecars: LegacySidecar[] = []; + for (const name of await legacyEntries(manifestsDir)) { + if (!LEGACY_MANIFEST.test(name)) continue; + try { + sidecars.push(LegacySidecar.assert(JSON.parse(await fs.readFile(path.join(manifestsDir, name), "utf8")))); + } catch (error) { + log({ level: "warn", msg: "local_sidecar_legacy_manifest_unreadable", file: name, error: errorText(error) }); + } + } + return sidecars; + } + + async function removeLegacyLayout(kept: ReadonlySet): Promise { + const manifestsDir = path.join(dataRoot, LEGACY_MANIFESTS); + if (!(await exists(manifestsDir))) return []; + const removed: string[] = []; + for (const name of await legacyEntries(manifestsDir)) { + const allocationId = LEGACY_MANIFEST.exec(name)?.[1]; + if (allocationId !== undefined && kept.has(allocationId)) continue; + removed.push(path.join(LEGACY_MANIFESTS, name)); + } + for (const name of await legacyEntries(dataRoot)) { + const allocationId = LEGACY_DATA_DIR.exec(name)?.[1]; + if (allocationId !== undefined && !kept.has(allocationId) && (await fs.stat(path.join(dataRoot, name))).isDirectory()) removed.push(name); + } + for (const name of removed) await fs.rm(path.join(dataRoot, name), { recursive: true, force: true }); + if ((await legacyEntries(manifestsDir)).length === 0) { + await fs.rm(manifestsDir, { recursive: true, force: true }); + removed.push(LEGACY_MANIFESTS); + } + return removed; } - return { write, readAll, readDataDir, remove }; + return { read, write, remove, readHolds, writeHolds, readFences, writeFences, legacySidecars, removeLegacyLayout }; } diff --git a/apps/hub/src/sidecar/placement.postgres.test.ts b/apps/hub/src/sidecar/placement.postgres.test.ts new file mode 100644 index 000000000..24354fd0c --- /dev/null +++ b/apps/hub/src/sidecar/placement.postgres.test.ts @@ -0,0 +1,151 @@ +// Requires PostgreSQL at TEST_DB. +import { expect, test } from "bun:test"; +import { randomBytes } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { createDB, runMigrations, schema } from "@intx/db"; +import type { SidecarProvisioner } from "@intx/hub-sessions"; +import { createPlacementChooser, localSidecarUsage, WAITING_FOR_SLOT } from "./placement.js"; + +const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" }; +const TENANT = "tnt_placement"; + +type Db = ReturnType["db"]; +type Status = typeof schema.sidecarAllocation.$inferInsert.status; + +function provisioner(id: string): SidecarProvisioner { + return { + id, + apiVersion: 2, + bindingFingerprint: `${id}:v1`, + capabilities: [], + ensure() { + throw new Error("Placement never ensures."); + }, + destroy() { + throw new Error("Placement never destroys."); + }, + }; +} + +const local = provisioner("local-process"); +const sandbox = provisioner("railway-sandbox"); + +async function withDatabase(run: (db: Db) => Promise): Promise { + const database = `corbits_placement_${randomBytes(6).toString("hex")}`; + const admin = createDB(TEST_DB); + await admin.db.execute(sql.raw(`create database ${database}`)); + const config = { ...TEST_DB, database }; + let handle: ReturnType | undefined; + try { + await runMigrations(config, { schema: "public" }); + handle = createDB(config); + await handle.db.insert(schema.tenant).values({ id: TENANT, name: "Placement", slug: "placement", domain: "placement.test" }); + await handle.db.insert(schema.workflowDefinition).values({ id: "wfd_manager", tenantId: TENANT, name: "triage-manager-github" }); + await handle.db.insert(schema.principal).values({ id: "prn_owner", tenantId: TENANT, kind: "user", refId: "usr_owner", status: "active" }); + await run(handle.db); + } finally { + await handle?.close(); + await admin.db.execute(sql.raw(`drop database if exists ${database} with (force)`)); + await admin.close(); + } +} + +let next = 0; + +async function deployed(db: Db, on: SidecarProvisioner, status: Status, { deploymentCleanupConfirmed = false, requires = [] as string[], failureCode = null as string | null, generation = 0 } = {}): Promise { + const id = `run_${String(++next)}`; + await db.insert(schema.workflowRun).values({ id, anchorRunId: id, definitionId: "wfd_manager", tenantId: TENANT, status: "running" }); + await db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: id, + sessionId: `ses_${String(next)}`, + deploymentDomain: "placement.test", + sourceAuthorityPrincipalId: "prn_owner", + frozenApprovalBundle: { projection: { sidecarPlacement: { capabilities: requires.map((capability) => ({ capability, effect: "require" })) } } }, + sourceOfferingIds: [], + defaultSourceOfferingId: "off_default", + deployContent: {}, + }); + await db.insert(schema.sidecarAllocation).values({ + id: `sal_${String(next)}`, + anchorRunId: id, + tenantId: TENANT, + provisionerId: on.id, + provisionerApiVersion: 2, + provisionerBindingFingerprint: on.bindingFingerprint, + adapterManifestDigest: "digest", + status, + deploymentCleanupConfirmed, + failureCode, + generation, + maxDisconnectedMs: 60_000, + }); +} + +async function probing(db: Db, on: SidecarProvisioner, status: typeof schema.workflowProbe.$inferInsert.status): Promise { + const id = `wpr_${String(++next)}`; + await db.insert(schema.asset).values({ id: `ast_${id}`, tenantId: TENANT, kind: "workflow", name: id, creatorPrincipalId: "prn_owner" }); + await db.insert(schema.workflowProbe).values({ + id, + tenantId: TENANT, + definitionAssetId: `ast_${id}`, + source: { kind: "asset", assetId: `ast_${id}`, package: { format: "source", commitSha: "0".repeat(40) } }, + entry: "index.ts", + status, + provisionerId: on.id, + provisionerApiVersion: 2, + provisionerBindingFingerprint: on.bindingFingerprint, + adapterManifestDigest: "digest", + }); +} + +test("a deployment goes to the local sidecar until its active slots are taken, then to the first sandbox, which counts as overflow", async () => { + await withDatabase(async function place(db) { + const choose = createPlacementChooser({ db, local, maxDeployments: 2 }); + await deployed(db, local, "allocated"); + await deployed(db, local, "released", { deploymentCleanupConfirmed: true }); + await deployed(db, local, "releasing", { deploymentCleanupConfirmed: true }); + await deployed(db, sandbox, "allocated"); + await deployed(db, sandbox, "allocated", { requires: ["isolation:workload"] }); + expect(await choose([local, sandbox])).toBe(local); + + // A copy the sidecar has not confirmed removing still holds its slot. + await deployed(db, local, "releasing"); + expect(await choose([local, sandbox])).toBe(sandbox); + expect(await choose([local])).toBe(local); + expect(await choose([sandbox])).toBe(sandbox); + + // Only the sandbox deployment that could have run locally overflowed; the one that requires isolation did not. + expect(await localSidecarUsage(db, local, 2)).toEqual({ active: 2, waiting: 0, cap: 2, overflow: 1 }); + }); +}, 120_000); + +test("placements side by side cannot both take the local sidecar's last slot, which frees once its pick is counted", async () => { + await withDatabase(async function race(db) { + let clock = 0; + const choose = createPlacementChooser({ db, local, maxDeployments: 2, now: () => clock }); + await deployed(db, local, "allocated"); + + const picks = await Promise.all([choose([local, sandbox]), choose([local, sandbox]), choose([local, sandbox])]); + expect(picks).toEqual([local, sandbox, sandbox]); + + clock += 31_000; + expect(await choose([local, sandbox])).toBe(local); + }); +}, 120_000); + +test("probes on the local sidecar take its slots, and placements waiting there for one are counted apart and still fill it", async () => { + await withDatabase(async function shared(db) { + const choose = createPlacementChooser({ db, local, maxDeployments: 3 }); + await deployed(db, local, "allocated"); + await probing(db, local, "probing"); + await probing(db, local, "failed"); + await probing(db, sandbox, "probing"); + expect(await localSidecarUsage(db, local, 3)).toEqual({ active: 2, waiting: 0, cap: 3, overflow: 0 }); + + // What stock writes for a placement refused for room while it waits to try again. + await deployed(db, local, "replacing", { failureCode: WAITING_FOR_SLOT, generation: 1, deploymentCleanupConfirmed: true }); + await deployed(db, local, "failed", { failureCode: WAITING_FOR_SLOT }); + expect(await localSidecarUsage(db, local, 3)).toEqual({ active: 2, waiting: 1, cap: 3, overflow: 0 }); + expect(await choose([local, sandbox])).toBe(sandbox); + }); +}, 120_000); diff --git a/apps/hub/src/sidecar/placement.ts b/apps/hub/src/sidecar/placement.ts new file mode 100644 index 000000000..84b00ce80 --- /dev/null +++ b/apps/hub/src/sidecar/placement.ts @@ -0,0 +1,134 @@ +import { schema, type DB } from "@intx/db"; +import type { SidecarProvisioner, SidecarProvisionerChooser } from "@intx/hub-sessions"; +import type { DeployedVersions } from "@corbits/triage-contracts"; +import { and, eq, inArray, ne, or, sql, type SQL } from "drizzle-orm"; + +export type CreatePlacementChooserOpts = { + readonly db: DB["db"]; + readonly local: SidecarProvisioner; + /** The active deployments the local sidecar advertises. */ + readonly maxDeployments: number; + readonly now?: () => number; +}; + +/** + * How long a local pick counts before its allocation row does. Stock writes that row right after the chooser answers, + * so once it lands the pick counts twice until this passes, which sends a deployment to a sandbox early, never late. + */ +const RESERVATION_MS = 30_000; + +async function settle(decision: Promise): Promise { + try { + await decision; + } catch { + // The caller observes the failure; the queue only needs ordering. + } +} + +const allocation = schema.sidecarAllocation; +const probe = schema.workflowProbe; + +/** Stock's code for a placement refused because the sidecar's active slots were full; it retries the placement. */ +export const WAITING_FOR_SLOT = "sidecar_reuse_rejected"; + +/** Stock's probe statuses short of an outcome, as its probe store counts them. */ +const ACTIVE_PROBE_STATUSES = ["pending", "provisioning", "probing", "releasing"] as const; + +/** + * A placement refused for room, which stock moves to `replacing` with the code until it tries again: it holds no slot, + * its cleanup confirmed since nothing was deployed, but takes the next one that frees. + */ +function waitsForSlot(): SQL | undefined { + return and(eq(allocation.status, "replacing"), eq(allocation.failureCode, WAITING_FOR_SLOT)); +} + +/** An allocation holding an active slot, as stock counts a sidecar's room at placement. */ +function holdsSlot(): SQL | undefined { + return or( + inArray(allocation.status, ["provisioning", "allocated"]), + and(inArray(allocation.status, ["replacing", "releasing", "destroy_failed"]), eq(allocation.deploymentCleanupConfirmed, false)), + ); +} + +async function countAllocations(db: DB["db"], where: SQL | undefined): Promise { + const [row] = await db.select({ count: sql`count(*)::int` }).from(allocation).where(where); + if (row === undefined) throw new Error("placement: count query returned no row"); + return row.count; +} + +async function activeProbes(db: DB["db"], provisionerId: string): Promise { + const [row] = await db + .select({ count: sql`count(*)::int` }) + .from(probe) + .where(and(eq(probe.provisionerId, provisionerId), inArray(probe.status, [...ACTIVE_PROBE_STATUSES]))); + if (row === undefined) throw new Error("placement: count query returned no row"); + return row.count; +} + +/** Deployments and probes on the provisioner's sidecar: probes run there too, so both draw from its slots. */ +async function activeDeployments(db: DB["db"], provisionerId: string): Promise { + return (await activeProbes(db, provisionerId)) + (await countAllocations(db, and(eq(allocation.provisionerId, provisionerId), holdsSlot()))); +} + +function waitingDeployments(db: DB["db"], provisionerId: string): Promise { + return countAllocations(db, and(eq(allocation.provisionerId, provisionerId), waitsForSlot())); +} + +/** Deployments elsewhere whose workflow requires no capability, so the local sidecar was a candidate for them. */ +async function overflowDeployments(db: DB["db"], localId: string): Promise { + const launchSpec = schema.workflowRunLaunchSpec; + const requires = sql`exists (select 1 from jsonb_array_elements(coalesce(${launchSpec.frozenApprovalBundle} -> 'projection' -> 'sidecarPlacement' -> 'capabilities', '[]'::jsonb)) as rule where rule ->> 'effect' = 'require')`; + const [row] = await db + .select({ count: sql`count(*)::int` }) + .from(allocation) + .innerJoin(launchSpec, eq(launchSpec.anchorRunId, allocation.anchorRunId)) + .where(and(ne(allocation.provisionerId, localId), holdsSlot(), sql`not ${requires}`)); + if (row === undefined) throw new Error("placement: count query returned no row"); + return row.count; +} + +export async function localSidecarUsage(db: DB["db"], local: SidecarProvisioner, cap: number): Promise { + return { + active: await activeDeployments(db, local.id), + waiting: await waitingDeployments(db, local.id), + cap, + overflow: await overflowDeployments(db, local.id), + }; +} + +/** + * The local sidecar until it is full, then the first sandbox that matches. Decisions run one at a time, and each local + * pick holds a slot until its allocation is counted, so placements side by side cannot both take the last one. + */ +export function createPlacementChooser({ db, local, maxDeployments, now = Date.now }: CreatePlacementChooserOpts): SidecarProvisionerChooser { + let reservations: number[] = []; + let queue: Promise = Promise.resolve(); + + async function decide(candidates: readonly SidecarProvisioner[]): Promise { + const [first] = candidates; + if (first === undefined) throw new Error("No matching sidecar provisioners are available"); + if (!candidates.includes(local)) return first; + const sandbox = candidates.find((candidate) => candidate !== local); + // With nowhere else to go the deployment still goes local: stock retries a full sidecar's placement with backoff, + // so it waits there until a deployment ends. + if (sandbox === undefined) return local; + const at = now(); + reservations = reservations.filter((until) => until > at); + // A placement waiting for a slot takes the next one that frees, so a new deployment would only queue behind it. + const taken = (await activeDeployments(db, local.id)) + (await waitingDeployments(db, local.id)) + reservations.length; + if (taken >= maxDeployments) return sandbox; + reservations.push(at + RESERVATION_MS); + return local; + } + + async function after(previous: Promise, candidates: readonly SidecarProvisioner[]): Promise { + await previous; + return decide(candidates); + } + + return function choosePlacement(candidates) { + const decision = after(queue, candidates); + queue = settle(decision); + return decision; + }; +} diff --git a/apps/hub/src/sidecar/restore.ts b/apps/hub/src/sidecar/restore.ts index 2652bfa0f..3d2c1c23b 100644 --- a/apps/hub/src/sidecar/restore.ts +++ b/apps/hub/src/sidecar/restore.ts @@ -1,25 +1,43 @@ import { schema, type DB } from "@intx/db"; -import { and, eq, isNull } from "drizzle-orm"; +import { createSidecarCredentialResolver } from "@intx/hub-sessions"; +import { and, eq, inArray, isNull, notInArray } from "drizzle-orm"; import type { LocalProcessSidecarProvisioner } from "./local-provisioner.js"; -import type { LocalSidecarManifest } from "./manifest-store.js"; +import type { LocalSidecarHold } from "./manifest-store.js"; export async function restoreLocalSidecars( db: DB["db"], local: LocalProcessSidecarProvisioner, ): Promise { - async function isLocalSidecarLive(manifest: LocalSidecarManifest): Promise { + const credentials = createSidecarCredentialResolver({ db }); + async function isLive(hold: LocalSidecarHold & { readonly sidecarId: string }): Promise { const row = await db.query.sidecarAllocation.findFirst({ columns: { id: true }, where: and( - eq(schema.sidecarAllocation.id, manifest.allocationId), + eq(schema.sidecarAllocation.id, hold.allocationId), eq(schema.sidecarAllocation.provisionerId, local.provisioner.id), eq(schema.sidecarAllocation.status, "allocated"), - eq(schema.sidecarAllocation.generation, manifest.generation), - eq(schema.sidecarAllocation.sidecarId, manifest.sidecarId), + eq(schema.sidecarAllocation.generation, hold.generation), + eq(schema.sidecarAllocation.sidecarId, hold.sidecarId), isNull(schema.sidecarAllocation.initializationLeaseId), ), }); return row !== undefined; } - await local.restore(isLocalSidecarLive); + 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 }); } diff --git a/apps/hub/src/sidecar/spawn.ts b/apps/hub/src/sidecar/spawn.ts index 66707cd41..c34e9d620 100644 --- a/apps/hub/src/sidecar/spawn.ts +++ b/apps/hub/src/sidecar/spawn.ts @@ -1,28 +1,36 @@ import type { ChildProcessEnv } from "../env.js"; -import type { SpawnLocalSidecar } from "./local-provisioner.js"; +import type { LocalSidecarExit, SpawnLocalSidecar } from "./local-provisioner.js"; export type CreateSpawnLocalSidecarOpts = { readonly interchangeDir: string; readonly credentialEncryptionKey: string; + readonly maxActiveDeployments: number; readonly processEnv: ChildProcessEnv; }; +async function exitOf(child: Bun.Subprocess): Promise { + const code = await child.exited; + return { code: child.signalCode === null ? code : null, signal: child.signalCode }; +} + export function createSpawnLocalSidecar({ interchangeDir, credentialEncryptionKey, + maxActiveDeployments, processEnv, }: CreateSpawnLocalSidecarOpts): SpawnLocalSidecar { - return function spawnSidecar({ request, dataDir }) { + return function spawnSidecar({ manifest, adapterManifest }) { const child = Bun.spawn(["bun", "--conditions=intx-src", "apps/sidecar/src/index.ts"], { cwd: interchangeDir, env: { ...processEnv, - HUB_WS_URL: request.hubWebSocketUrl, - SIDECAR_ID: request.sidecarId, - SIDECAR_TOKEN: request.token, - SIDECAR_DATA_DIR: dataDir, + HUB_WS_URL: manifest.hubWebSocketUrl, + SIDECAR_ID: manifest.sidecarId, + SIDECAR_TOKEN: manifest.token, + SIDECAR_DATA_DIR: manifest.dataDir, SIDECAR_CREDENTIAL_ENCRYPTION_KEY: credentialEncryptionKey, - SIDECAR_ADAPTER_MANIFEST: JSON.stringify(request.adapterManifest ?? []), + SIDECAR_ADAPTER_MANIFEST: JSON.stringify(adapterManifest), + SIDECAR_MAX_ACTIVE_DEPLOYMENTS: String(maxActiveDeployments), }, stdin: "ignore", stdout: "inherit", @@ -30,7 +38,7 @@ export function createSpawnLocalSidecar({ }); return { pid: child.pid, - exited: child.exited, + exited: exitOf(child), kill(signal) { child.kill(signal); }, diff --git a/apps/hub/src/workflow-versions.test.ts b/apps/hub/src/workflow-versions.test.ts index d1b9f8beb..c6df3374a 100644 --- a/apps/hub/src/workflow-versions.test.ts +++ b/apps/hub/src/workflow-versions.test.ts @@ -31,6 +31,7 @@ const SPECS: Record = { }; const ROLLOUT = { source: "github" as const, at: CREATED_AT.toISOString(), error: "Updating Triage failed; see the hub log." }; +const SIDECAR = { active: 3, waiting: 0, cap: 16, overflow: 1 }; const ROWS = [ { runId: "run_tarball", workflow: "pr-triage", domain: "tenant.example", createdAt: CREATED_AT, cancellationRequestedAt: null }, @@ -50,11 +51,11 @@ function db() { } function handler() { - return createWorkflowVersions({ db: db(), up: async (anchorRunId) => anchorRunId !== "run_source", launchSpecs: { get: async (anchorRunId) => (SPECS[anchorRunId] ?? null) as never }, rollouts: (tenantId) => (tenantId === TENANT_ID ? [ROLLOUT] : []) }); + return createWorkflowVersions({ db: db(), up: async (anchorRunId) => anchorRunId !== "run_source", launchSpecs: { get: async (anchorRunId) => (SPECS[anchorRunId] ?? null) as never }, rollouts: (tenantId) => (tenantId === TENANT_ID ? [ROLLOUT] : []), sidecar: async () => SIDECAR }); } describe("createWorkflowVersions", () => { - test("reads each live deployment's workflow and tool versions from its frozen closure, beside the workspace's own rollouts", async () => { + test("reads each live deployment's workflow and tool versions from its frozen closure, beside the workspace's own rollouts and the local sidecar's use", async () => { const res = await handler()({ tenantId: TENANT_ID, principalId: "member-1" }); expect(res.status).toBe(200); expect(await res.json()).toEqual({ @@ -63,6 +64,7 @@ describe("createWorkflowVersions", () => { { id: "run_source", workflow: "pr-triage-historical", createdAt: CREATED_AT.toISOString(), cancelling: true, live: false, package: SOURCE_WORKFLOW, tools: [], source: { assetId: "ast_triage", commitSha: "c0ffee" } }, ], rollouts: [ROLLOUT], + sidecar: SIDECAR, }); }); diff --git a/apps/hub/src/workflow-versions.ts b/apps/hub/src/workflow-versions.ts index 428113aff..17346238a 100644 --- a/apps/hub/src/workflow-versions.ts +++ b/apps/hub/src/workflow-versions.ts @@ -15,6 +15,8 @@ export type WorkflowVersionsDeps = { /** Whether the deployment is up as the hub counts a listener or a manager. */ up: (anchorRunId: string) => Promise; rollouts: (tenantId: string) => DeployedVersions["rollouts"]; + /** The hub's local sidecar: what it runs, its cap, and what overflowed to a sandbox. */ + sidecar: () => Promise; }; export function sourceOf(bundle: FrozenApprovalBundle): DeployedVersions["deployments"][number]["source"] { @@ -39,6 +41,6 @@ export function createWorkflowVersions(deps: WorkflowVersionsDeps) { source: spec ? sourceOf(spec.frozenApprovalBundle) : null, }); } - return Response.json({ deployments, rollouts: deps.rollouts(tenantId) } satisfies DeployedVersions); + return Response.json({ deployments, rollouts: deps.rollouts(tenantId), sidecar: await deps.sidecar() } satisfies DeployedVersions); }; } diff --git a/apps/web/src/components/settings/TriageSection.tsx b/apps/web/src/components/settings/TriageSection.tsx index c90876def..09dc950b9 100644 --- a/apps/web/src/components/settings/TriageSection.tsx +++ b/apps/web/src/components/settings/TriageSection.tsx @@ -1,6 +1,6 @@ import { useMemo, type ReactNode } from "react"; import { Link } from "react-router-dom"; -import { presetFor, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; +import { presetFor, TRIAGE_LISTENERS, type DeployedVersions, type RepoSource } from "@corbits/triage-contracts"; import { useCheckPacks } from "../../lib/check-packs.ts"; import { modelSecrets } from "../../lib/credential-actions.ts"; import { DeniedNotice } from "../../lib/denied.tsx"; @@ -33,6 +33,15 @@ function statusText({ sources, undeployed, model, triaging, total }: Status): st return `Triage is running on ${triaging} of ${counted(total, "repository", "repositories")}.`; } +/** How full this server is, and where pull requests go once it is. */ +function sidecarText({ active, waiting, cap, overflow }: DeployedVersions["sidecar"]): string { + const runs = `This server runs ${active} of up to ${cap} Triage workflows at once`; + const elsewhere = overflow > 0 ? `; ${overflow} more ${overflow === 1 ? "runs" : "run"} in a sandbox because it is full` : ""; + const queued = waiting > 0 ? `; ${waiting} more ${waiting === 1 ? "waits" : "wait"} for one to finish` : ""; + if (active >= cap && overflow === 0 && waiting === 0) return `${runs}: it is full, so a new pull request waits a few minutes for a place, then is tried again on its next update.`; + return `${runs}${elsewhere}${queued}.`; +} + /** A preset by its name; your own provider by its endpoint and model. */ function judgeModel(saved: { endpoint: string; model: string }): ReactNode { const preset = presetFor(saved); @@ -96,6 +105,11 @@ export function TriageSection({ snapshot, deployed }: { snapshot: PortalSnapshot {statusText(status)} Choose repositories + {deployed.sidecar && ( +
+ {sidecarText(deployed.sidecar)} +
+ )} {sources.length > 0 && status.model && } )} diff --git a/apps/web/src/lib/tenant-entities.ts b/apps/web/src/lib/tenant-entities.ts index 58c9dc285..d6673b7c9 100644 --- a/apps/web/src/lib/tenant-entities.ts +++ b/apps/web/src/lib/tenant-entities.ts @@ -86,12 +86,12 @@ export function useRoles(): TenantSection { return useTenantSection(ROLES_QUERY_KEY, listRoles, DIRECTORY_REFRESH_MS); } -export type DeployedSection = TenantSection & { rollouts: DeployedVersions["rollouts"] }; +export type DeployedSection = TenantSection & { rollouts: DeployedVersions["rollouts"]; sidecar: DeployedVersions["sidecar"] | undefined }; -/** The live deployments and the hub's last roll of each host, read often so a roll's outcome shows soon after it ends. */ +/** The live deployments, the hub's last roll of each host and its local sidecar's use, read often so a roll's outcome shows soon after it ends. */ export function useDeployedVersions(): DeployedSection { const query = useTenantRead(DEPLOYED_VERSIONS_QUERY_KEY, readDeployedVersions, LIVE_REFRESH_MS); - return { ...sectionOf(query, (versions) => versions.deployments), rollouts: query.data?.rollouts ?? NO_ROWS }; + return { ...sectionOf(query, (versions) => versions.deployments), rollouts: query.data?.rollouts ?? NO_ROWS, sidecar: query.data?.sidecar }; } export function useAccess(resource: string, action: string): UseQueryResult { diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index fe8be3f08..1c0e055ec 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -95,7 +95,9 @@ GitHub ──webhook──► Hub ──spawn──► Sidecar ──► listene counts as live while it runs or is on its sidecar, and as starting for six minutes after it is deployed, longer than a roll waits for it; the reconciler mails the newest running listener, else the newest starting - one. One deployed without a sidecar past that is released, and a host + one. A listener on its sidecar stays deployed until its first pull request + arrives, so it is never ended for not running. One deployed without a + sidecar past that is released, and a host left with no live listener has its workspace rolled again. The workflow-versions route reports each deployment's `live` as a manager's is counted, which Settings reads. @@ -147,10 +149,28 @@ GitHub ──webhook──► Hub ──spawn──► Sidecar ──► listene is never logged: the manifest callback's carries GitHub's one-time code. A refused workflow deploy also logs `workflow_deploy_refused` with its code and reason. -- **Sidecars**: one per live deployment, started by the hub's - `SidecarProvisioner`s. The local-process provisioner runs child processes - on the hub's machine, restores them on boot and declines a placement while - free memory is short. The Docker provisioner (`packages/provisioner-docker`), +- **Sidecars**: started by the hub's `SidecarProvisioner`s. The local-process + provisioner runs one sidecar, a child process on the hub's machine, that + hosts every deployment placed there, each in a workflow child process of + its own: the first ensure starts it under the + identity Interchange offered, and every later ensure answers with that + sidecar, which stock deploys into over the same connection. It keeps a hold + per allocation beside one process manifest, and persists each destroy's + fence so a late ensure stays refused across a restart; a destroy releases the hold and + leaves the copy for the hub to undeploy, so the process lives as long as the + hub. On boot it respawns the sidecar once if any hold is still allocated or + the hub still waits for it to remove a copy. + It starts the sidecar only with 1 GiB free. The sidecar advertises + `TRIAGE_SIDECAR_MAX_DEPLOYMENTS` (default 16) active deployments in its hello, + and stock refuses to place more there. The hub's placement chooser + (`apps/hub/src/sidecar/placement.ts`) counts the local sidecar's active + allocations and probes, and the placements waiting there for a slot, and + sends a deployment to the first matching sandbox provisioner once the cap is + reached. With no sandbox it stays local, where stock retries its placement + for up to 10 ensure attempts (about two and a half minutes) and then fails + it; the pull request is tried again on its next event. Probes run on the local + sidecar only when no sandbox is registered, since with one they require + `isolation:workload`. The Docker provisioner (`packages/provisioner-docker`), registered after it when a daemon with the sidecar image answers, runs each sidecar in a container of its own. A published hub image carries the digest of the sidecar image CI pushed just before it, and the provisioner pulls that @@ -313,6 +333,12 @@ any other tenant or an unknown sender. the claim: the next manager's take finds the run gone and claims. A run left deployed with no sidecar for two minutes is gone too, and its cancellation is requested so it does not start after the claim has moved on. + A manager or judge is deployed with the mail that starts + it, so one still `deployed` five minutes after it reached its sidecar never + ran: it is gone for the claim, and each reconcile pass ends it and logs + `triage_deployment_never_ran` with its run id and workflow. The hub times + that from the sidecar's connection for the run, or from its own start for a + run placed before it. When the rules need the judge, the host's own checks have finished (or the hub waits for them no longer) and `release.json` publishes a judge for the host, a pass deploys it under `workflow:*` `create` and `read` on exactly diff --git a/docs/Corbits Triage User Manual/Corbits Triage User Manual.md b/docs/Corbits Triage User Manual/Corbits Triage User Manual.md index 1c264254a..f8188ccc5 100644 --- a/docs/Corbits Triage User Manual/Corbits Triage User Manual.md +++ b/docs/Corbits Triage User Manual/Corbits Triage User Manual.md @@ -534,6 +534,8 @@ Everything triage needs, top to bottom: its status, where pull requests come fro The first line says how many repositories Triage is running on, with a **Choose repositories** link. Until a decision model is added, it says **Triage needs a decision model before it can run. Add one below.** Judgement rules are the ones that use the model. Add it on this page: pick a provider, paste the key and save it. The key is stored in the server's vault and you will not see it again. +The next line says how busy this server is, for example **This server runs 3 of up to 16 Triage workflows at once.** Each host's listener takes one of those places, and each pull request being triaged takes another while it runs. When they are all taken and the server has a sandbox set up, the line adds **2 more run in a sandbox because it is full.** Without a sandbox it says **it is full, so a new pull request waits a few minutes for a place, then is tried again on its next update.** Once some are waiting, it says how many instead, for example **3 more wait for one to finish.** The server's operator sets how many there are. + #### Which version runs Once a source and a decision model are set up, a line below the status says which version of Triage runs, one line per connected host (named when there are two): diff --git a/docs/DEV.md b/docs/DEV.md index d74358963..61249141f 100644 --- a/docs/DEV.md +++ b/docs/DEV.md @@ -87,7 +87,7 @@ bun run lint Hub tests connect to `postgres://postgres:postgres@localhost:5432/interchange` and ignore `DATABASE_URL`. -`bun run test:chain` (`tests/chain/`) starts the hub on a database of its own, serving a portal build, sets up a workspace through the portal's routes, waits for the hub to start its listener, and sends GitHub's webhooks to it. GitHub and the decision model are local stand-ins, so it needs no account or key; each sidecar is a local process, and it installs the workflows' native modules from the npm registry as a deploy does. Code review needs an isolated sidecar, which a local process is not, so it is not part of the chain test. +`bun run test:chain` (`tests/chain/`) starts the hub on a database of its own, serving a portal build, sets up a workspace through the portal's routes, waits for the hub to start its listener, and sends GitHub's webhooks to it. GitHub and the decision model are local stand-ins, so it needs no account or key; its one sidecar is a local process, and it installs the workflows' native modules from the npm registry as a deploy does. Code review needs an isolated sidecar, which a local process is not, so it is not part of the chain test. Scripts are listed in [IMPLEMENTATION.md](IMPLEMENTATION.md#scripts). diff --git a/docs/SELF_HOST.md b/docs/SELF_HOST.md index 7aba42421..1d694482a 100644 --- a/docs/SELF_HOST.md +++ b/docs/SELF_HOST.md @@ -27,8 +27,10 @@ Optional, locally or deployed: a Google OAuth client for sign-in, an SMTP account for invite emails, and for [code review](PRODUCT.md#code-review) an Anthropic key and an [isolated place to run it](#where-sidecars-run). -Size the hub for its sidecars: each triage and code review runs in a Bun -process of its own (about 1 GiB at peak). +Size the hub for its sidecar: triage runs in one sidecar process beside the +hub, which starts a workflow process for each of up to +`TRIAGE_SIDECAR_MAX_DEPLOYMENTS` deployments at once. Memory is checked only +when the sidecar starts (1 GiB free), so the cap is what bounds it. ## Run it @@ -149,8 +151,7 @@ the pre-deploy command. Railway Sandboxes need `SANDBOX_REPOSITORY` and 2. Set the [required variables](#required) and any optional ones. 3. Run the [migrations](#migrations) before each start or upgrade. 4. Give the platform a drain time longer than `HUB_SIDECAR_STOP_TIMEOUT_MS` - (default 25 seconds); sidecars stop in parallel, so shutdown takes about - one timeout. + (default 25 seconds), the most the local sidecar takes to stop. 5. Terminate TLS in front of the hub. Sign-in and email rate limits count each caller by address. On Railway the @@ -294,7 +295,8 @@ All optional; the defaults suit most installs. | Variable | Default | Effect | | --- | --- | --- | -| `HUB_SIDECAR_STOP_TIMEOUT_MS` | 25000 | How long each local sidecar gets to finish writing run state after SIGTERM before SIGKILL. | +| `HUB_SIDECAR_STOP_TIMEOUT_MS` | 25000 | How long the local sidecar gets to finish writing run state after SIGTERM before SIGKILL. | +| `TRIAGE_SIDECAR_MAX_DEPLOYMENTS` | 16 | Deployments the hub's local sidecar runs at once: the listener, one manager per pull request being triaged, and the judge. At most 1792. | | `HUB_API_ORIGIN` | the origin of `BETTER_AUTH_BASE_URL` | The origin sidecars reach the hub REST API at, for workflows that call it. An `http` or `https` origin with no path. Set it when sidecars run in Docker on the hub's machine, where `localhost` is the container. | | `HUB_SIDECAR_WEBSOCKET_URL` | `ws://127.0.0.1:/api/sidecars/ws` | The address sidecars dial. Leave it unset for Docker [on another machine](provisioners/docker-remote.md), whose sidecars dial `BETTER_AUTH_BASE_URL` over `wss://`. | | `DOCKER_HOST` | the docker CLI's context, else its default sockets | The daemon for the [Docker provisioner](provisioners/docker.md). | @@ -357,7 +359,7 @@ The hub still publishes the workflows bundled in its own image's ## Where sidecars run -Sidecars run as child processes on the hub's machine by default; triage needs +Triage runs in one sidecar, a child process on the hub's machine; it needs nothing more. Code review needs a sidecar of its own, so it runs only once an isolated place is set up. Settings → Code review → **Where it runs** lists every place this server diff --git a/docs/provisioners/README.md b/docs/provisioners/README.md index 163574d3e..0cfdcb25d 100644 --- a/docs/provisioners/README.md +++ b/docs/provisioners/README.md @@ -4,14 +4,14 @@ Every triage run, code review and workflow probe runs in a sidecar: a process th | Provisioner | Runs sidecars | Isolation | Registered | | --- | --- | --- | --- | -| [Local process](local.md) | as child processes of the hub | none: they share the hub's machine, user and files | always | +| [Local process](local.md) | as one child process of the hub | none: it shares the hub's machine, user and files | always | | [Docker](docker.md) | each in a container of its own, on the hub's machine or [another one over TLS](docker-remote.md) | `isolation:workload` | when a Docker daemon with the sidecar image answers | | [E2B](e2b.md) | each in a sandbox (microVM) of its own in E2B's cloud | `isolation:workload` | when `E2B_API_KEY` is set and E2B answers | | [Railway Sandboxes](railway-sandboxes.md) | each in a Railway sandbox of its own | `isolation:workload` | when `RAILWAY_SANDBOX_TOKEN` and `RAILWAY_SANDBOX_ENVIRONMENT_ID` are set | ## How the hub picks one -Workflows and tenants state the capabilities a sidecar needs, such as `isolation:workload`; provisioners declare the capabilities they offer. The hub tries provisioners in the order it registered them (local process, Docker, E2B, then Railway Sandboxes) and takes the first whose capabilities satisfy every rule. A workflow with no rule therefore runs on the local process provisioner. When no provisioner matches, the deployment fails with `provisioner_no_match` and names what each provisioner lacked. +Workflows and tenants state the capabilities a sidecar needs, such as `isolation:workload`; provisioners declare the capabilities they offer. The hub tries provisioners in the order it registered them (local process, Docker, E2B, then Railway Sandboxes) and takes the first whose capabilities satisfy every rule. A workflow with no rule therefore runs on the local process provisioner, or on the first sandbox provisioner while the local sidecar is full ([Capacity](local.md#capacity)). When no provisioner matches, the deployment fails with `provisioner_no_match` and names what each provisioner lacked. Code review requires `isolation:workload`, so it never runs on the hub's own machine. When several places are ready, it uses the first one registered (Docker, E2B, then Railway Sandboxes); there is no per-workspace choice, because a tenant placement rule would move every workflow, triage included. A place that is full declines the review's sidecar with `capacity_unavailable`; the hub defers the review and retries it on the same place, never the next one, since the place is chosen by capabilities alone. Settings → Code review → **Where it runs** says this in plain words and lists every place this hub knows, in that order, with why to choose it, its state (in use, ready, not set up, or failing with the reason, shown to those who may manage the workspace) and, for a place not set up or failing, the `.env` lines that set it up, with Copy, and its guide. Without a ready place, code review does not run, and the repository page links there. diff --git a/docs/provisioners/local.md b/docs/provisioners/local.md index 917e21d6d..006420385 100644 --- a/docs/provisioners/local.md +++ b/docs/provisioners/local.md @@ -4,14 +4,16 @@ The default provisioner, in `apps/hub/src/sidecar/local-provisioner.ts`. It need ## What it does -It starts one sidecar per allocation as a child process of the hub, with a data directory of its own under `HUB_DATA_DIR/local-sidecars`. On restart, the hub restores the sidecars its database still holds. On shutdown, each sidecar gets `HUB_SIDECAR_STOP_TIMEOUT_MS` (default 25000) to finish writing run state after SIGTERM before SIGKILL. +It runs one sidecar process, a child of the hub, that hosts every deployment placed here: the listener for each host, one manager per pull request being triaged, and the judge. The sidecar runs each deployment in a workflow child process of its own. The first deployment starts the sidecar; every later one joins it. Its state lives under `HUB_DATA_DIR/local-sidecars`: `sidecar/` is its data directory, `sidecar.json` its identity, `holds.json` the deployments it holds and `fences.json` the ended ones a late request must not bring back. Until the hub has restored this state on boot, the provisioner answers no request. A `fences.json` that cannot be read is renamed to `fences.json.corrupt-`. Whatever that file held, a request for a deployment or probe the hub has finished is refused. A deployment that ends is removed from the sidecar, which keeps running for the life of the hub. On restart, the hub starts the sidecar again if its database still holds any of those deployments, or still waits for the sidecar to remove one that ended, and forgets the ones it no longer holds. A hub upgraded from one sidecar per deployment retires each of those sidecars on its first boot, so their deployments are placed again at once, and removes their manifests and data directories. On shutdown, the sidecar gets `HUB_SIDECAR_STOP_TIMEOUT_MS` (default 25000) to finish writing run state after SIGTERM before SIGKILL. ## Isolation it does not give -Sidecars run on the hub's machine as the hub's user. They can read the hub's files and environment, and each other's. The provisioner declares no `isolation:workload`, so a workflow or tenant that requires it is never placed here. Use [Docker](docker.md) for those. +The sidecar runs on the hub's machine as the hub's user. It and its deployments' workflow processes can read the hub's files and environment, and each other's. The provisioner declares no `isolation:workload`, so a workflow or tenant that requires it is never placed here. Use [Docker](docker.md) for those. ## Capacity -It declines a new sidecar while free memory, less 1 GiB for each sidecar still starting, is below 1 GiB. Free memory is the container's cgroup limit less its usage, reclaimable page cache counted free, else Linux's `MemAvailable`. A machine with neither, such as a macOS dev machine, never declines. There is no setting. The hub retries a declined sidecar with backoff, up to Interchange's 10 ensure attempts; after the tenth the deployment fails. +The sidecar runs at most `TRIAGE_SIDECAR_MAX_DEPLOYMENTS` deployments at once (default 16, at most 1792). Interchange counts the deployments placed on it and refuses one more once they fill that; the hub's placement then sends the next deployment to the first sandbox provisioner registered ([Docker](docker.md), [E2B](e2b.md) or [Railway Sandboxes](railway-sandboxes.md)). With none registered, the deployment stays here and Interchange retries its placement with backoff for up to 10 ensure attempts, about two and a half minutes; after the tenth the deployment fails and the pull request is tried again on its next event. A workflow probe runs here only when no sandbox provisioner is registered; with one registered, probes require `isolation:workload` and run there. Interchange counts a probe toward the cap only once its deployment adopts it. Settings → Triage shows how many deployments and probes the sidecar runs, how many wait for a slot, and how many overflowed. A manager or judge that sits on the sidecar five minutes without starting is ended, freeing its slot; the listener is exempt, since it waits for its first pull request. -The 1 GiB was measured on a local stack (2026-10-17, macOS, GitHub emulator) of the replaced `pr-triage-github` workflow: four of those workers peaked at 784 to 860 MiB RSS over a full run, and idle listeners held 471 to 653 MiB. A sidecar connects with about half its peak in use, so a burst right after can still overshoot by that much. +Memory is checked only when the sidecar starts, which needs 1 GiB free; after that the deployment cap is what bounds it, so set `TRIAGE_SIDECAR_MAX_DEPLOYMENTS` for the machine's memory. Free memory is the container's cgroup limit less its usage, reclaimable page cache counted free, else Linux's `MemAvailable`. A machine with neither, such as a macOS dev machine, never declines. The hub retries a declined start with backoff, up to the same 10 attempts. + +The 1 GiB was measured on a local stack (2026-10-17, macOS, GitHub emulator) of the replaced `pr-triage-github` workflow: four of those workers peaked at 784 to 860 MiB RSS over a full run, and idle listeners held 471 to 653 MiB. diff --git a/packages/triage-contracts/src/hub-responses.ts b/packages/triage-contracts/src/hub-responses.ts index 3f75cfe4b..1e65e10f6 100644 --- a/packages/triage-contracts/src/hub-responses.ts +++ b/packages/triage-contracts/src/hub-responses.ts @@ -212,6 +212,12 @@ export const DeployedVersions = type({ }).array(), /** Each host the hub rolled since it started: when the roll ended, and why it failed if it did; the listener before it keeps running. */ rollouts: type({ source: type.enumerated(...REPO_SOURCES), at: "string", "error?": "string" }).array(), + /** + * The hub's local sidecar, across workspaces: deployments and probes it runs (`active`) out of `cap`, deployments + * placed there that wait for a free slot (`waiting`), and deployments that could have run there but run on a sandbox + * because it was full (`overflow`). + */ + sidecar: type({ active: "number", waiting: "number", cap: "number", overflow: "number" }), }); export type DeployedVersions = typeof DeployedVersions.infer; diff --git a/packages/triage-contracts/src/listener.ts b/packages/triage-contracts/src/listener.ts index c090e07ef..9f562f133 100644 --- a/packages/triage-contracts/src/listener.ts +++ b/packages/triage-contracts/src/listener.ts @@ -7,6 +7,9 @@ export const TRIAGE_LISTENERS: Record = { github: "triage-li /** Each host's pull request manager by its definition id, also its mail address. */ export const TRIAGE_MANAGERS: Record = { github: "triage-manager-github", "bitbucket-datacenter": "triage-manager-bitbucket-datacenter" }; +/** Each host's judge by its definition id, also its mail address; a manager deploys one when its rules need the model. */ +export const TRIAGE_JUDGES: Record = { github: "triage-judge-github", "bitbucket-datacenter": "triage-judge-bitbucket-datacenter" }; + /** The system role the hub's publisher holds to publish the listener's chain and deploy it, which every deployment of it then acts under. */ export const DEPLOYER_ROLE = "Deployer"; diff --git a/packages/triage-judge/src/triage-judge-bitbucket-datacenter.ts b/packages/triage-judge/src/triage-judge-bitbucket-datacenter.ts index 0c8feb34b..e87252c6f 100644 --- a/packages/triage-judge/src/triage-judge-bitbucket-datacenter.ts +++ b/packages/triage-judge/src/triage-judge-bitbucket-datacenter.ts @@ -1,5 +1,4 @@ +import { TRIAGE_JUDGES } from "@corbits/triage-contracts"; import { triageJudge } from "./triage-judge.js"; -export const TRIAGE_JUDGE_BITBUCKET_DATACENTER_ADDRESS = "triage-judge-bitbucket-datacenter"; - -export const workflow = triageJudge({ id: TRIAGE_JUDGE_BITBUCKET_DATACENTER_ADDRESS }); +export const workflow = triageJudge({ id: TRIAGE_JUDGES["bitbucket-datacenter"] }); diff --git a/packages/triage-judge/src/triage-judge-github.ts b/packages/triage-judge/src/triage-judge-github.ts index 3df819928..0c05228bd 100644 --- a/packages/triage-judge/src/triage-judge-github.ts +++ b/packages/triage-judge/src/triage-judge-github.ts @@ -1,5 +1,4 @@ +import { TRIAGE_JUDGES } from "@corbits/triage-contracts"; import { triageJudge } from "./triage-judge.js"; -export const TRIAGE_JUDGE_GITHUB_ADDRESS = "triage-judge-github"; - -export const workflow = triageJudge({ id: TRIAGE_JUDGE_GITHUB_ADDRESS }); +export const workflow = triageJudge({ id: TRIAGE_JUDGES.github }); diff --git a/tests/chain/shared-sidecar.test.ts b/tests/chain/shared-sidecar.test.ts new file mode 100644 index 000000000..26a1e3aab --- /dev/null +++ b/tests/chain/shared-sidecar.test.ts @@ -0,0 +1,145 @@ +// One local sidecar process hosts the listener and two managers at once, and +// each manager receives only its own pull request's mail. +import { afterAll, beforeAll, expect, setDefaultTimeout, test } from "bun:test"; +import { mkdtempSync, readdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { sql } from "drizzle-orm"; +import { createDB } from "@intx/db"; +import { fakeGithub, OWNER, type Pull } from "./github.ts"; +import { stubModel } from "./model.ts"; +import { buildPortal, createDatabase, deliver, dropDatabase, listenerStarted, setUpWorkspace, signUp, startHub, type Database, type Hub, type Workspace } from "./stack.ts"; + +setDefaultTimeout(10 * 60_000); + +const FIRST = `${OWNER}/first`; +const SECOND = `${OWNER}/second`; +const POLICY = "rules:\n - check: max-files-changed\n text: Flag pull requests over 30 files changed.\n max: 30\n reason: The maintainers set it.\n evidence:\n - maintainer-stated\n"; + +const github = fakeGithub(); +const model = stubModel(function never(): never { + throw new Error("No rule needs the judge."); +}); +const portalDir = mkdtempSync(join(tmpdir(), "corbits-shared-portal-")); +let database: Database; +let hub: Hub; +let workspace: Workspace; + +beforeAll(async () => { + github.addRepo(FIRST, POLICY); + github.addRepo(SECOND, POLICY); + const [created] = await Promise.all([createDatabase(), buildPortal(portalDir)]); + database = created; + hub = await startHub(database, github, portalDir); + workspace = await signUp(hub, database); + await setUpWorkspace(workspace, github, model.origin); + await listenerStarted(workspace); +}); + +afterAll(async () => { + await hub?.stop(); + if (database) await dropDatabase(database); + github.stop(); + model.stop(); + rmSync(portalDir, { recursive: true, force: true }); +}); + +function sha(seed: string): string { + return new Bun.CryptoHasher("sha1").update(seed).digest("hex"); +} + +function pullOf(head: string): Pull { + return { + number: 1, title: "fix: widget sizing", body: "Fixes the widget's size.", author: "octocat", head, + files: [{ filename: "src/widget.ts", status: "modified", additions: 1, deletions: 1, patch: "@@ -1 +1 @@\n-export const size = 1;\n+export const size = 2;" }], + commits: ["fix: widget sizing"], labels: [], comments: [], + }; +} + +async function pushed(repo: string, head: string, action: "opened" | "synchronize" = "opened"): Promise { + const pull = pullOf(head); + github.open(repo, pull); + await deliver(hub, workspace, "pull_request", { + action, installation: { id: 1 }, repository: { full_name: repo }, + pull_request: { number: 1, state: "open", draft: false, head: { sha: head }, user: { login: pull.author }, title: pull.title, body: pull.body }, + }); +} + +async function rows(query: string): Promise[]> { + const { db, close } = createDB(database.config); + try { + return [...await db.execute(sql.raw(query))] as Record[]; + } finally { + await close(); + } +} + +async function until(what: string, read: () => Promise, ms = 120_000): Promise { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + const value = await read(); + if (value !== undefined) return value; + await Bun.sleep(500); + } + throw new Error(`Timed out waiting for ${what}.\n${hub.logs.slice(-80).join("\n")}`); +} + +type Claim = { status: "live" | "ended"; runId?: string; history: { head: string; runId: string }[] }; + +async function claimOf(repo: string): Promise { + const [row] = await rows(`select state from corbits_triage.manager_state where kind = 'claim' and key = 'mgr:github:${repo}#1'`); + return row?.["state"] as Claim | undefined; +} + +test("one local sidecar process hosts the listener and two live managers, each receiving its own mail", async () => { + const releaseFirst = github.hold(FIRST, 1); + const releaseSecond = github.hold(SECOND, 1); + await pushed(FIRST, sha("first-1")); + await pushed(SECOND, sha("second-1")); + const first = await until("the first manager's claim", async () => { + const claim = await claimOf(FIRST); + return claim?.status === "live" ? claim : undefined; + }); + const second = await until("the second manager's claim", async () => { + const claim = await claimOf(SECOND); + return claim?.status === "live" ? claim : undefined; + }); + expect(first.runId).not.toBe(second.runId); + + const live = await rows(`select a.id, a.sidecar_id, r.status, coalesce(s.frozen_approval_bundle -> 'projection' ->> 'id', d.name) as workflow from sidecar_allocation a join workflow_run r on r.id = a.anchor_run_id join workflow_definition d on d.id = r.definition_id left join workflow_run_launch_spec s on s.anchor_run_id = r.id where a.status = 'allocated'`); + expect(live.filter((row) => row["status"] === "running").map((row) => row["workflow"]).sort()).toEqual(["triage-listener-github", "triage-manager-github", "triage-manager-github"]); + expect(new Set(live.map((row) => row["sidecar_id"])).size).toBe(1); + expect(hub.logs.filter((line) => line.includes('"msg":"local_sidecar_started"'))).toHaveLength(1); + expect(readdirSync(join(hub.dataDir, "local-sidecars"), { withFileTypes: true }).filter((entry) => entry.isDirectory()).map((entry) => entry.name)).toEqual(["sidecar"]); + + await pushed(FIRST, sha("first-2"), "synchronize"); + await until("the first push's second head to be recorded", async () => { + const [row] = await rows(`select count(*)::int as recorded from corbits_triage.manager_state where kind = 'event' and key like 'evt:github:${FIRST}#1:%'`); + return Number(row!["recorded"]) >= 2 || undefined; + }); + releaseFirst(); + releaseSecond(); + const firstDone = await until("the first manager to finish both heads", async () => { + const claim = await claimOf(FIRST); + return claim !== undefined && claim.history.length >= 2 ? claim : undefined; + }); + const secondDone = await until("the second manager to finish", async () => { + const claim = await claimOf(SECOND); + return claim !== undefined && claim.history.length >= 1 ? claim : undefined; + }); + expect(firstDone.history.map((pass) => pass.runId)).toEqual([first.runId!, first.runId!]); + expect(firstDone.history.at(-1)!.head).toBe(sha("first-2")); + expect(secondDone.history).toEqual([expect.objectContaining({ head: sha("second-1"), runId: second.runId })]); + expect(github.pull(FIRST, 1).comments.map((comment) => comment.body)).toContainEqual(expect.stringContaining("**Triage: Pass**")); + expect(github.pull(SECOND, 1).comments.map((comment) => comment.body)).toEqual([expect.stringContaining("**Triage: Pass**")]); + const managerAllocations = live.filter((row) => row["workflow"] === "triage-manager-github").map((row) => `'${String(row["id"])}'`).join(","); + const released = await until("both managers' allocations to be released from the shared sidecar", async () => { + const found = await rows(`select status, deployment_cleanup_confirmed from sidecar_allocation where id in (${managerAllocations})`); + return found.every((row) => row["status"] === "released") ? found : undefined; + }); + expect(released).toEqual([{ status: "released", deployment_cleanup_confirmed: true }, { status: "released", deployment_cleanup_confirmed: true }]); + const [sidecar] = await rows(`select retired_at from sidecar where id = '${String(live[0]!["sidecar_id"])}'`); + expect(sidecar!["retired_at"]).toBeNull(); + const listener = await rows(`select a.status, r.status as run from sidecar_allocation a join workflow_run r on r.id = a.anchor_run_id where a.id = '${String(live.find((row) => row["workflow"] === "triage-listener-github")!["id"])}'`); + expect(listener).toEqual([{ status: "allocated", run: "running" }]); +}); diff --git a/tests/chain/sidecar-cap.test.ts b/tests/chain/sidecar-cap.test.ts new file mode 100644 index 000000000..4a23a1581 --- /dev/null +++ b/tests/chain/sidecar-cap.test.ts @@ -0,0 +1,114 @@ +// The local sidecar's cap with no sandbox to overflow to: a manager placed while the listener and another manager +// fill both slots waits, and runs once the other ends. +import { afterAll, beforeAll, expect, setDefaultTimeout, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { sql } from "drizzle-orm"; +import { createDB } from "@intx/db"; +import { fakeGithub, OWNER, type Pull } from "./github.ts"; +import { stubModel } from "./model.ts"; +import { buildPortal, createDatabase, deliver, dropDatabase, listenerStarted, setUpWorkspace, signUp, startHub, type Database, type Hub, type Workspace } from "./stack.ts"; + +setDefaultTimeout(10 * 60_000); + +const FIRST = `${OWNER}/first`; +const SECOND = `${OWNER}/second`; +const POLICY = "rules:\n - check: max-files-changed\n text: Flag pull requests over 30 files changed.\n max: 30\n reason: The maintainers set it.\n evidence:\n - maintainer-stated\n"; + +const github = fakeGithub(); +const model = stubModel(function never(): never { + throw new Error("No rule needs the judge."); +}); +const portalDir = mkdtempSync(join(tmpdir(), "corbits-cap-portal-")); +let database: Database; +let hub: Hub; +let workspace: Workspace; + +beforeAll(async () => { + github.addRepo(FIRST, POLICY); + github.addRepo(SECOND, POLICY); + const [created] = await Promise.all([createDatabase(), buildPortal(portalDir)]); + database = created; + hub = await startHub(database, github, portalDir, { TRIAGE_SIDECAR_MAX_DEPLOYMENTS: "2" }); + workspace = await signUp(hub, database); + await setUpWorkspace(workspace, github, model.origin); + await listenerStarted(workspace); +}); + +afterAll(async () => { + await hub?.stop(); + if (database) await dropDatabase(database); + github.stop(); + model.stop(); + rmSync(portalDir, { recursive: true, force: true }); +}); + +function sha(seed: string): string { + return new Bun.CryptoHasher("sha1").update(seed).digest("hex"); +} + +async function pushed(repo: string, head: string): Promise { + const pull: Pull = { + number: 1, title: "fix: widget sizing", body: "Fixes the widget's size.", author: "octocat", head, + files: [{ filename: "src/widget.ts", status: "modified", additions: 1, deletions: 1, patch: "@@ -1 +1 @@\n-export const size = 1;\n+export const size = 2;" }], + commits: ["fix: widget sizing"], labels: [], comments: [], + }; + github.open(repo, pull); + await deliver(hub, workspace, "pull_request", { + action: "opened", installation: { id: 1 }, repository: { full_name: repo }, + pull_request: { number: 1, state: "open", draft: false, head: { sha: head }, user: { login: pull.author }, title: pull.title, body: pull.body }, + }); +} + +async function rows(query: string): Promise[]> { + const { db, close } = createDB(database.config); + try { + return [...await db.execute(sql.raw(query))] as Record[]; + } finally { + await close(); + } +} + +async function until(what: string, read: () => Promise, ms = 180_000): Promise { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + const value = await read(); + if (value !== undefined) return value; + await Bun.sleep(500); + } + throw new Error(`Timed out waiting for ${what}.\n${hub.logs.slice(-80).join("\n")}`); +} + +type Claim = { status: "live" | "ended"; runId?: string; history: { head: string; runId: string }[] }; + +async function claimOf(repo: string): Promise { + const [row] = await rows(`select state from corbits_triage.manager_state where kind = 'claim' and key = 'mgr:github:${repo}#1'`); + return row?.["state"] as Claim | undefined; +} + +test("with every local slot taken and no sandbox, a manager waits for one to free and then runs", async () => { + const releaseFirst = github.hold(FIRST, 1); + await pushed(FIRST, sha("first-1")); + const first = await until("the first manager's claim", async () => { + const claim = await claimOf(FIRST); + return claim?.status === "live" ? claim : undefined; + }); + await pushed(SECOND, sha("second-1")); + const waiting = await until("the second manager's placement to be refused for room", async () => { + const [row] = await rows("select failure_code from sidecar_allocation where failure_code = 'sidecar_reuse_rejected' limit 1"); + return row; + }); + expect(waiting).toEqual({ failure_code: "sidecar_reuse_rejected" }); + expect(await claimOf(SECOND)).toBeUndefined(); + + releaseFirst(); + const second = await until("the second manager to post its verdict", async () => { + const claim = await claimOf(SECOND); + return claim !== undefined && claim.history.length === 1 ? claim : undefined; + }); + expect(second.runId ?? second.history[0]!.runId).not.toBe(first.runId); + expect(github.pull(SECOND, 1).comments.map((comment) => comment.body)).toEqual([expect.stringContaining("**Triage: Pass**")]); + const [sidecars] = await rows(`select count(distinct sidecar_id)::int as count from sidecar_allocation where sidecar_id is not null and status in ('allocated', 'released')`); + expect(sidecars).toEqual({ count: 1 }); +}); diff --git a/tests/chain/stack.ts b/tests/chain/stack.ts index d7eac96a4..e2f33190f 100644 --- a/tests/chain/stack.ts +++ b/tests/chain/stack.ts @@ -71,8 +71,8 @@ async function collect(stream: ReadableStream, lines: string[]): Pro export type Hub = { origin: string; dataDir: string; logs: string[]; stop: () => Promise }; -/** The hub as the image starts it, serving the portal build in `portalDir`, with every sidecar a local process under its data directory. */ -export async function startHub(database: Database, github: FakeGithub, portalDir: string): Promise { +/** The hub as the image starts it, serving the portal build in `portalDir`, with its sidecar a local process under its data directory. */ +export async function startHub(database: Database, github: FakeGithub, portalDir: string, env: Record = {}): Promise { const port = freePort(); const origin = `http://127.0.0.1:${port}`; const dataDir = mkdtempSync(join(tmpdir(), "corbits-chain-")); @@ -92,6 +92,7 @@ export async function startHub(database: Database, github: FakeGithub, portalDir TRIAGE_COALESCE_QUIET_MS: "500", // Every sidecar a local process, as on a hub without Docker, so each reaches the fake GitHub and model on loopback. DOCKER_HOST: "unix:///nonexistent/docker.sock", + ...env, }, stdin: "ignore", stdout: "pipe",