diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index e18f1a231..12dd381d5 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -68,8 +68,8 @@ import { triageTenants } from "./deployer-role.js"; import { bundleAt, bundled, type Publisher } from "./workflow-release/publish.js"; import { createListenerWatch, releaseWriteOf } from "./workflow-release/triggers.js"; 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 { catalogModels, createWorkspaceLock, deployedAllocation, deploymentState, liveDeployments } from "./workflow-release/reads.js"; +import { createRolloutQueue, createWorkflowRollout, stopGivenUp, type DeployRequest } from "./workflow-release/rollout.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"; @@ -570,6 +570,9 @@ const workflowRollout = createWorkflowRollout({ async cancel(tenantId, runId) { await composition.workflowLifecycleService.requestCancellation(tenantId, runId, "replaced"); }, + status(tenantId, runId) { + return composition.workflowLifecycleService.getStatus(tenantId, runId); + }, lock: createWorkspaceLock(composition.db.$client), sleep: Bun.sleep, now, @@ -583,14 +586,35 @@ function rollHost(source: RepoSource) { if (workflowsBundled) workflowReleases.request(tenantId, [source]); }; } +/** The host's listeners still live after stock gave up stopping them. */ +async function stuckListeners(tenantId: string, source: RepoSource): Promise { + const stuck: string[] = []; + for (const listener of await resolveLiveDeployments(composition.db, tenantId, TRIAGE_LISTENERS[source])) { + if (!listener.cancelling) continue; + const run = await composition.workflowLifecycleService.getStatus(tenantId, listener.runId); + if (run && stopGivenUp(run)) stuck.push(listener.runId); + } + return stuck; +} const watchListeners = createListenerWatch({ connectedSources: function sourcesOf(tenantId) { return connectedSources(composition.db, tenantId); }, listener: liveListenerOf, + stuck: stuckListeners, + allocation: function allocationOf(runId) { + return deployedAllocation(composition.db, runId); + }, + routable: function routableOf(target) { + return composition.sidecarRouter.isAllocatedWorkflowActive(target); + }, ensure: function rollWorkspace(tenantId) { if (workflowsBundled) workflowReleases.ensure(tenantId); }, + replace: function replaceListener(tenantId, source) { + return workflowsBundled && workflowReleases.replace(tenantId, source); + }, + log: logJson, }); const managerRuns = createManagerRuns(composition.db, verdictRecordStore, now); const tenantOpenHeads = createTenantOpenHeads(tenantSources); diff --git a/apps/hub/src/workflow-release/reads.ts b/apps/hub/src/workflow-release/reads.ts index 07889fc03..be7625453 100644 --- a/apps/hub/src/workflow-release/reads.ts +++ b/apps/hub/src/workflow-release/reads.ts @@ -1,7 +1,8 @@ // What a roll reads from the hub's database: the workspace's catalog, its live deployments and how a deploy is starting, // and the lock that keeps one roll per workspace. -import { eq } from "drizzle-orm"; +import { and, eq, isNotNull, isNull } from "drizzle-orm"; import { listVisibleOfferings, schema, type createDB, type DB, type WorkflowRunLaunchSpecStore } from "@intx/db"; +import type { AllocatedSidecarTarget } from "@intx/hub-sessions"; import type { ModelInfo } from "@corbits/triage-contracts"; import { resolveLiveDeployments } from "../github/deployment.js"; import { sourceOf } from "../workflow-versions.js"; @@ -74,3 +75,26 @@ export async function deploymentState(db: DB["db"], runId: string): Promise { + const [row] = await db + .select({ allocationId: schema.sidecarAllocation.id, generation: schema.sidecarAllocation.generation, status: schema.workflowRun.status }) + .from(schema.workflowRun) + .innerJoin(schema.sidecarAllocation, eq(schema.sidecarAllocation.anchorRunId, schema.workflowRun.id)) + .where( + and( + eq(schema.workflowRun.id, runId), + isNull(schema.workflowRun.cancellationRequestedAt), + isNotNull(schema.workflowRun.publicKey), + eq(schema.sidecarAllocation.status, "allocated"), + isNotNull(schema.sidecarAllocation.sidecarId), + eq(schema.sidecarAllocation.ensureAcceptedGeneration, schema.sidecarAllocation.generation), + ), + ); + if (!row || !schema.isLiveWorkflowRunStatus(row.status)) return undefined; + return { allocationId: row.allocationId, generation: row.generation }; +} diff --git a/apps/hub/src/workflow-release/rollout.postgres.test.ts b/apps/hub/src/workflow-release/rollout.postgres.test.ts index 86838a418..d6bbcefaf 100644 --- a/apps/hub/src/workflow-release/rollout.postgres.test.ts +++ b/apps/hub/src/workflow-release/rollout.postgres.test.ts @@ -64,7 +64,7 @@ afterAll(async () => { await rm(dataDir, { recursive: true, force: true }); }); -type Run = Deployment & { workflow: string; state: DeployState }; +type Run = Deployment & { workflow: string; state: DeployState; polls: number }; /** A workspace with its publisher, rolled against fake deployments. */ async function workspace(name: string) { @@ -77,6 +77,9 @@ async function workspace(name: string) { models: [DECISION] as ModelInfo[], starts: "live" as DeployState, deployFails: false, + stopHangs: false, + releasedAfterPolls: Infinity, + statusFails: 0, clock: Date.now(), runs: [] as Run[], requests: [] as DeployRequest[], @@ -101,7 +104,7 @@ async function workspace(name: string) { if (fake.deployFails) throw new Error("no sidecar capacity"); fake.requests.push(request); const runId = `run_${fake.requests.length}`; - fake.runs.push({ runId, workflow: LISTENER, state: fake.starts, createdAt: new Date(fake.clock + fake.requests.length), cancelling: false, source: { assetId: request.assetId, commitSha: request.commitSha } }); + fake.runs.push({ runId, workflow: LISTENER, state: fake.starts, createdAt: new Date(fake.clock + fake.requests.length), cancelling: false, polls: 0, source: { assetId: request.assetId, commitSha: request.commitSha } }); return runId; }, async state(runId) { @@ -111,6 +114,20 @@ async function workspace(name: string) { const run = fake.runs.find((row) => row.runId === runId); if (run) run.cancelling = true; }, + async status(_tenant, runId) { + if (fake.statusFails > 0) { + fake.statusFails -= 1; + throw new Error("status read failed"); + } + const run = fake.runs.find((row) => row.runId === runId); + if (!run) return null; + run.polls += 1; + const released = run.cancelling && run.polls >= fake.releasedAfterPolls; + return { + status: run.cancelling && !fake.stopHangs ? "cancelled" : "running", + allocation: { id: `alc_${runId}`, generation: 1, status: released ? "releasing" : "allocated", failureCode: null, failureMessage: null }, + }; + }, lock: createWorkspaceLock(handle.db.$client), async sleep(ms) { fake.clock += ms; @@ -219,9 +236,9 @@ test("PostgreSQL: once the newest listener runs the release, the listeners befor await rollout.roll(tenantId, NO_REDEPLOY); const release = { assetId: await listenerAsset(), commitSha: fake.requests[0]!.commitSha }; fake.runs.push( - { runId: "run_older", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, source: { assetId: "ast_old", commitSha: "old" } }, - { runId: "run_replaced", workflow: "pr-triage-github", state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, source: null }, - { runId: "run_cut_short", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock + 60_000), cancelling: false, source: release }, + { runId: "run_older", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, polls: 0, source: { assetId: "ast_old", commitSha: "old" } }, + { runId: "run_replaced", workflow: "pr-triage-github", state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, polls: 0, source: null }, + { runId: "run_cut_short", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock + 60_000), cancelling: false, polls: 0, source: release }, ); expect(await rollout.roll(tenantId, NO_REDEPLOY)).toEqual(SUCCEEDED); expect(fake.requests).toHaveLength(1); @@ -229,10 +246,49 @@ test("PostgreSQL: once the newest listener runs the release, the listeners befor expect(live("pr-triage-github")).toEqual([]); }, 120_000); +test("PostgreSQL: a replaced listener or workflow that stays live once stock gave up stopping it is logged as stuck, and the roll succeeds", async () => { + const { tenantId, fake, rollout, live } = await workspace("stuck"); + await rollout.roll(tenantId, NO_REDEPLOY); + fake.runs.push({ runId: "run_historical", workflow: "pr-triage-historical", state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, polls: 0, source: null }); + fake.stopHangs = true; + fake.releasedAfterPolls = 3; + expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); + expect(live()).toEqual(["run_2"]); + const stuck = fake.logs.filter((entry) => String(entry["msg"]).endsWith("_cancel_stuck")); + expect(stuck).toHaveLength(2); + expect(stuck).toEqual(expect.arrayContaining([ + { level: "warn", msg: "listener_cancel_stuck", tenantId, source: "github", deployment: "run_1" }, + { level: "warn", msg: "rollout_cancel_stuck", tenantId, source: "github", workflow: "pr-triage-historical", deployment: "run_historical" }, + ])); +}, 120_000); + +test("PostgreSQL: a replaced listener stock neither stops nor gives up on is waited on five minutes from the stop request, then logged as stuck, and the roll succeeds", async () => { + const { tenantId, fake, rollout, live } = await workspace("unanswered"); + await rollout.roll(tenantId, NO_REDEPLOY); + fake.stopHangs = true; + const before = fake.clock; + expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); + expect(fake.clock - before).toBeGreaterThanOrEqual(5 * 60_000); + expect(live()).toEqual(["run_2"]); + expect(fake.logs.filter((entry) => entry["msg"] === "listener_cancel_stuck")).toEqual([{ level: "warn", msg: "listener_cancel_stuck", tenantId, source: "github", deployment: "run_1" }]); +}, 120_000); + +test("PostgreSQL: a stop wait retries a failed status read three times, and one that keeps failing is logged as stuck while the roll succeeds", async () => { + const { tenantId, fake, rollout, live } = await workspace("unread"); + await rollout.roll(tenantId, NO_REDEPLOY); + fake.statusFails = 3; + expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); + expect(fake.logs.filter((entry) => String(entry["msg"]).endsWith("_cancel_stuck"))).toEqual([]); + fake.statusFails = 4; + expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); + expect(live()).toEqual(["run_3"]); + expect(fake.logs.filter((entry) => entry["msg"] === "listener_cancel_stuck")).toEqual([{ level: "warn", msg: "listener_cancel_stuck", tenantId, source: "github", deployment: "run_2", error: "status read failed" }]); +}, 120_000); + test("PostgreSQL: a roll neither waits on nor stops a listener it did not deploy that runs the release, however long it takes to start", async () => { const { tenantId, fake, rollout, live } = await workspace("replacing"); await rollout.roll(tenantId, NO_REDEPLOY); - fake.runs.push({ runId: "run_older", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, source: { assetId: "ast_old", commitSha: "old" } }); + fake.runs.push({ runId: "run_older", workflow: LISTENER, state: "live", createdAt: new Date(fake.clock - 60_000), cancelling: false, polls: 0, source: { assetId: "ast_old", commitSha: "old" } }); fake.runs[0]!.state = "starting"; fake.clock += START_DEADLINE_MS * 2; const before = fake.clock; @@ -365,9 +421,10 @@ test("a roll that fails while a request waits out its quiet window rolls again w expect(rolls).toBe(2); }); -test("a connected host found with no live listener rolls its workspace", async () => { +test("a connected host found with no live listener rolls its workspace, and every host's listener stock gave up stopping is reported on every pass", async () => { const ensured: string[] = []; - const listeners: Partial> = { github: "run_listener" }; + const logged: Record[] = []; + const listeners: Partial> = { "bitbucket-datacenter": { runId: "run_bitbucket" } }; const watch = createListenerWatch({ async connectedSources() { return ["github", "bitbucket-datacenter"]; @@ -375,15 +432,100 @@ test("a connected host found with no live listener rolls its workspace", async ( async listener(_tenantId, source) { return listeners[source]; }, + async stuck(_tenantId, source) { + return source === "bitbucket-datacenter" ? ["run_stuck"] : []; + }, + async allocation() { + return undefined; + }, + async routable() { + return true; + }, ensure(tenantId) { ensured.push(tenantId); }, + replace() { + return true; + }, + log(entry) { + logged.push(entry); + }, }); await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); - listeners["bitbucket-datacenter"] = "run_bitbucket"; + listeners.github = { runId: "run_github" }; await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); + const stuck = { level: "warn", msg: "listener_cancel_stuck", tenantId: "tnt_watch", source: "bitbucket-datacenter", deployment: "run_stuck" }; + expect(logged).toEqual([stuck, stuck]); +}); + +test("a listener whose connected sidecar the router holds no route to is started again once, and one whose sidecar is not connected is left to stock", async () => { + const listeners: Record = { github: { runId: "run_github" }, "bitbucket-datacenter": { runId: "run_bitbucket" } }; + const routes: Record boolean> = { + alc_github: () => false, + alc_bitbucket: () => { + throw new Error("Allocation alc_bitbucket is not connected"); + }, + }; + const rolled: Array = []; + const entered = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const replaced = Promise.withResolvers(); + const rollout = { + async roll(_tenantId: string, redeploy: ReadonlySet): Promise { + rolled.push([...redeploy].sort()); + entered.resolve(); + await gate.promise; + for (const source of redeploy) listeners[source] = { runId: `${listeners[source].runId}_next` }; + replaced.resolve(); + return SUCCEEDED; + }, + }; + const logged: Record[] = []; + const queue = createRolloutQueue(rollout, { quietMs: 0, retryMs: 60_000, concurrency: 2, random: Math.random }, function ignore() {}); + const watch = createListenerWatch({ + async connectedSources() { + return ["github", "bitbucket-datacenter"]; + }, + async listener(_tenantId, source) { + return listeners[source]; + }, + async stuck() { + return []; + }, + async allocation(runId) { + return runId.endsWith("_next") ? { allocationId: "alc_next", generation: 1 } : { allocationId: `alc_${runId.slice(4)}`, generation: 1 }; + }, + async routable(target) { + return routes[target.allocationId]?.() ?? true; + }, + ensure() { + throw new Error("Every host has a listener."); + }, + replace: queue.replace, + log(entry) { + logged.push(entry); + }, + }); + await watch("tnt_unroutable"); + await watch("tnt_unroutable"); + await entered.promise; + await watch("tnt_unroutable"); + gate.resolve(); + await replaced.promise; + await Bun.sleep(10); + await watch("tnt_unroutable"); + await Bun.sleep(10); + expect(rolled).toEqual([["github"]]); + const disconnected = { level: "debug", msg: "listener_disconnected", tenantId: "tnt_unroutable", source: "bitbucket-datacenter", deployment: "run_bitbucket", allocationId: "alc_bitbucket", generation: 1, reason: "Allocation alc_bitbucket is not connected" }; + expect(logged).toEqual([ + { level: "warn", msg: "listener_unroutable", tenantId: "tnt_unroutable", source: "github", deployment: "run_github", allocationId: "alc_github", generation: 1, reason: "its sidecar is connected but the hub holds no route to it" }, + disconnected, + disconnected, + disconnected, + disconnected, + ]); }); test("PostgreSQL: a person keeps the Deployer role until a roll leaves every host's listener running", async () => { diff --git a/apps/hub/src/workflow-release/rollout.ts b/apps/hub/src/workflow-release/rollout.ts index 1fb0bbff1..69726fbad 100644 --- a/apps/hub/src/workflow-release/rollout.ts +++ b/apps/hub/src/workflow-release/rollout.ts @@ -1,5 +1,7 @@ // Rolls a workspace onto the hub build's release: publishes it as the workspace's system principal, deploys each // connected host's listener from it once the running one differs, waits for it to start and stops what it replaces. +import { schema } from "@intx/db"; +import { isSidecarAllocationDispatchable, type WorkflowLifecycleResponse } from "@intx/types"; import { errorText } from "@corbits/common"; import { chainOf, offeringsIn, publishable, TRIAGE_LISTENERS, type DeployedVersions, type ModelInfo, type PublishedSource, type RepoSource } from "@corbits/triage-contracts"; import { revokeHumanDeployers } from "../deployer-role.js"; @@ -13,6 +15,8 @@ export type DeployRequest = { tenantId: string; principalId: string; assetId: st /** `live` once the deployment runs, `ended` once it stopped or failed without running. */ export type DeployState = "starting" | "live" | "ended"; +export type RunStop = Pick; + /** Releases a held workspace lock. */ export type Unlock = () => Promise; @@ -31,6 +35,8 @@ type RolloutDeps = PublishDeps & { deploy: (request: DeployRequest) => Promise; state: (runId: string) => Promise; cancel: (tenantId: string, runId: string) => Promise; + /** Stock's lifecycle status of the deployment; null once it is gone. */ + status: (tenantId: string, runId: string) => Promise; /** Takes the workspace's lock across hub processes; undefined while another holds it. */ lock: (tenantId: string) => Promise; sleep: (ms: number) => Promise; @@ -42,7 +48,19 @@ const ROLL_FAILED = "Updating Triage failed; see the hub log."; /** A new listener that is not live by then is stopped, and the one before it keeps running. */ export const START_DEADLINE_MS = 5 * 60_000; -const START_POLL_MS = 2_000; +const POLL_MS = 2_000; + +/** A replaced deployment stock has neither stopped nor given up on by then is logged as stuck, and the roll goes on. */ +const STOP_WAIT_MS = 5 * 60_000; +const STATUS_RETRIES = 3; + +/** + * Whether stock gave up stopping a deployment whose stop was requested: it stays live while stock reclaims its sidecar, + * as stock's forced release does once the stop goes unconfirmed past its grace. + */ +export function stopGivenUp(run: RunStop): boolean { + return schema.isLiveWorkflowRunStatus(run.status) && run.allocation !== null && !isSidecarAllocationDispatchable(run.allocation.status); +} /** Each host's workflows from before the listener, which nothing mails any more. */ const REPLACED: Record = { @@ -78,19 +96,64 @@ export function createWorkflowRollout(deps: RolloutDeps) { const state = await deps.state(runId); if (state === "live") return; if (state === "ended") throw new Error(`Listener ${runId} ended before it started.`); - await deps.sleep(START_POLL_MS); + await deps.sleep(POLL_MS); } await deps.cancel(tenantId, runId); throw new Error(`Listener ${runId} did not start within five minutes and was stopped.`); } + function logStuck(tenantId: string, source: RepoSource, workflow: string, runId: string, error?: unknown): void { + const reason = error === undefined ? {} : { error: errorText(error) }; + if (workflow === TRIAGE_LISTENERS[source]) deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId, ...reason }); + else deps.log({ level: "warn", msg: "rollout_cancel_stuck", tenantId, source, workflow, deployment: runId, ...reason }); + } + + /** Stock's status of the deployment, read again after a failure until the retries are spent. */ + async function statusOf(tenantId: string, runId: string): Promise { + for (let attempt = 1; ; attempt += 1) { + try { + return await deps.status(tenantId, runId); + } catch (error) { + if (attempt > STATUS_RETRIES) throw error; + await deps.sleep(POLL_MS * attempt); + } + } + } + + /** + * Waits until a deployment whose stop was requested ends. One stock gave up stopping, or that outlasts the wait, stays + * live and is only logged: stopping it is stock's. + */ + async function waitUntilStopped(tenantId: string, source: RepoSource, workflow: string, runId: string): Promise { + const deadline = deps.now().getTime() + STOP_WAIT_MS; + for (;;) { + let run: RunStop | null; + try { + run = await statusOf(tenantId, runId); + } catch (error) { + logStuck(tenantId, source, workflow, runId, error); + return; + } + if (!run || !schema.isLiveWorkflowRunStatus(run.status)) return; + if (stopGivenUp(run) || deps.now().getTime() >= deadline) { + logStuck(tenantId, source, workflow, runId); + return; + } + await deps.sleep(POLL_MS); + } + } + /** The host's other listeners and the workflows they replaced stop, and the host is recorded as on the release. */ async function settle(tenantId: string, source: RepoSource, kept: string, msg: string): Promise { + const stopped: Array<{ workflow: string; runId: string }> = []; for (const workflow of [TRIAGE_LISTENERS[source], ...REPLACED[source]]) { for (const live of await liveOf(tenantId, workflow)) { - if (live.runId !== kept) await deps.cancel(tenantId, live.runId); + if (live.runId === kept) continue; + await deps.cancel(tenantId, live.runId); + stopped.push({ workflow, runId: live.runId }); } } + await Promise.all(stopped.map((run) => waitUntilStopped(tenantId, source, run.workflow, run.runId))); record(tenantId, { source, at: deps.now().toISOString() }); deps.log({ level: "info", msg, tenantId, source, deployment: kept }); } @@ -277,5 +340,16 @@ export function createRolloutQueue(rollout: { roll: (tenantId: string, redeploy: schedule(tenantId, [], timing.quietMs, true); } - return { run, request, ensure }; + /** + * Starts the host's listener again unless a roll is running, which may already be replacing it, or one due already + * starts it again. Whether this asked for it. + */ + function replace(tenantId: string, source: RepoSource): boolean { + const state = stateOf(tenantId); + if (state.running || state.redeploy.has(source)) return false; + schedule(tenantId, [source], timing.quietMs, false); + return true; + } + + return { run, request, ensure, replace }; } diff --git a/apps/hub/src/workflow-release/triggers.ts b/apps/hub/src/workflow-release/triggers.ts index d53bf7c6f..e0d4fd0e0 100644 --- a/apps/hub/src/workflow-release/triggers.ts +++ b/apps/hub/src/workflow-release/triggers.ts @@ -1,7 +1,9 @@ -// What starts a roll besides the hub's own start: a write through a stock route, and a host found with no listener. +// What starts a roll besides the hub's own start: a write through a stock route, a host found with no listener, and one whose +// listener the hub cannot route mail to. import { and, eq } from "drizzle-orm"; import { schema, type DB } from "@intx/db"; -import { asRecord } from "@corbits/common"; +import { SidecarIdentityValidationError, type AllocatedSidecarTarget } from "@intx/hub-sessions"; +import { asRecord, errorText } from "@corbits/common"; import { REPO_SOURCES, type RepoSource } from "@corbits/triage-contracts"; /** A write that can change a workspace's release: its catalog decides which workflows publish and the offerings they run on. */ @@ -55,17 +57,48 @@ export async function releaseWriteOf(db: DB["db"], req: Request, path: string): export type ListenerWatchDeps = { connectedSources: (tenantId: string) => Promise; - listener: (tenantId: string, source: RepoSource) => Promise; + listener: (tenantId: string, source: RepoSource) => Promise<{ runId: string } | undefined>; + /** The host's listeners whose stop stock gave up on while they stay live. */ + stuck: (tenantId: string, source: RepoSource) => Promise; + /** The allocation generation the listener runs on once deployed; undefined while it starts. */ + allocation: (runId: string) => Promise; + /** Stock's router: whether it routes the generation's mail. */ + routable: (target: AllocatedSidecarTarget) => Promise; ensure: (tenantId: string) => void; + replace: (tenantId: string, source: RepoSource) => boolean; + log: (entry: Record) => void; }; -/** Rolls a workspace once a connected host has no live listener, as after one stopped on its own. */ +/** + * Rolls a workspace once a connected host has no live listener, as after one stopped on its own, and starts a host's + * listener again once its sidecar is connected but the hub holds no route to it. Stock acts on a lost route only when a + * sidecar connects or reports a stop, so such a listener would otherwise stay live and unreachable. A listener whose + * sidecar is not connected is left to stock, which fails it once it stays away past its deployment's limit. A listener + * whose stop stock gave up on is reported on every pass. + */ export function createListenerWatch(deps: ListenerWatchDeps) { return async function watch(tenantId: string): Promise { for (const source of await deps.connectedSources(tenantId)) { - if (await deps.listener(tenantId, source)) continue; - deps.ensure(tenantId); - return; + for (const runId of await deps.stuck(tenantId, source)) deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId }); + const listener = await deps.listener(tenantId, source); + if (!listener) { + deps.ensure(tenantId); + continue; + } + const target = await deps.allocation(listener.runId); + if (!target) continue; + const entry = { tenantId, source, deployment: listener.runId, allocationId: target.allocationId, generation: target.generation }; + let routed: boolean; + try { + routed = await deps.routable(target); + } catch (error) { + // A failed identity lookup leaves whether the sidecar is connected unknown, not absent. + if (error instanceof SidecarIdentityValidationError) deps.log({ level: "warn", msg: "listener_identity_unknown", ...entry, reason: errorText(error) }); + else deps.log({ level: "debug", msg: "listener_disconnected", ...entry, reason: errorText(error) }); + continue; + } + if (routed || !deps.replace(tenantId, source)) continue; + deps.log({ level: "warn", msg: "listener_unroutable", ...entry, reason: "its sidecar is connected but the hub holds no route to it" }); } }; } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 1c0e055ec..5646746da 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -241,7 +241,8 @@ any other tenant or an unknown sender. is connected or its key replaced, a Bitbucket Data Center instance is connected, its token replaced or its webhook secret rotated, and when GitHub installations sync; and when a reconcile pass finds a connected host - with no live listener. At most two rolls run at once across the hub, and + with no live listener, or with one whose connected sidecar the hub holds no + route to. At most two rolls run at once across the hub, and one per workspace across hub processes, under a Postgres session advisory lock held on a connection reserved for it, so a roll holds no transaction while it waits. A hub whose build has no `packages/sources/index.json` @@ -298,6 +299,33 @@ any other tenant or an unknown sender. removed. A running listener keeps the credentials it launched with, which is why a credential write deploys its host's listener again; managers and judges launch with the current value. + + Stock routes a deployment's mail through the sidecar router's address + index, per allocation generation, and acts on a lost route only when a + sidecar connects or reports a stop, or once it stayed disconnected past the + deployment's `maxDisconnected`. Each reconcile pass therefore asks the + router, through `isAllocatedWorkflowActive`, stock's deploy-readiness check, + which reads the same per-connection route table mail is routed by, whether + it routes the generation of each host's live listener whose first deploy + completed. Like stock's own reconciler calling it, the check undeploys a + sidecar whose identity is no longer current. When the listener's sidecar is + connected but the router holds no route to it, the hub starts that host's + listener again exactly as a credential write does and logs + `listener_unroutable` with the workspace, host, run, allocation, generation + and reason. A host whose roll is running or already due to redeploy it is + left to that roll, so one unroutable listener is replaced once. A listener whose sidecar is not + connected is left to stock, which fails it once it stays away past + `maxDisconnected`; each pass logs `listener_disconnected` at debug level + for it, or `listener_identity_unknown` when the router could not validate + the sidecar's identity. After asking stock to stop the listeners and + workflows a roll replaces, the roll polls stock's lifecycle status until + each ends. One that stays live once stock gave up stopping it, which + stock's lifecycle status shows as its allocation no longer dispatchable, + one still live five minutes after the stop was requested, and one whose + status could not be read after three retries are logged and the roll goes + on: a listener as `listener_cancel_stuck`, which every reconcile pass logs + again while stock has given up on it, and a replaced workflow once as + `rollout_cancel_stuck` with its name. - `triage-manager-github` and `triage-manager-bitbucket-datacenter` (`triageManager`): one live run per pull request, which runs only while it has work, and the only workflow that writes to the pull request. It takes diff --git a/tests/chain/chain.test.ts b/tests/chain/chain.test.ts index 152a7a703..3d77d78c6 100644 --- a/tests/chain/chain.test.ts +++ b/tests/chain/chain.test.ts @@ -8,11 +8,11 @@ import { join } from "node:path"; import { sql } from "drizzle-orm"; import { createDB } from "@intx/db"; import { listWorkflowDeployments, listWorkflowRuns, readWorkflowRunEvents } from "@intx/hub-client"; -import { VERDICT_LABELS, WAITING_LABEL } from "@corbits/triage-contracts"; +import { TRIAGE_LISTENERS, VERDICT_LABELS, WAITING_LABEL } from "@corbits/triage-contracts"; import { approveHeldDo } from "../../apps/web/src/lib/hub-api.ts"; import { fakeGithub, OWNER, type Pull, type Write } from "./github.ts"; import { stubModel, type Answer, type Question } from "./model.ts"; -import { buildPortal, createDatabase, deliver, dropDatabase, listenerStarted, setUpWorkspace, signUp, startHub, type Database, type Hub, type Workspace } from "./stack.ts"; +import { buildPortal, createDatabase, deliver, dropDatabase, listenerStarted, RECONCILE_INTERVAL_MS, setUpWorkspace, signUp, startHub, type Database, type Hub, type Workspace } from "./stack.ts"; setDefaultTimeout(20 * 60_000); @@ -20,6 +20,8 @@ const MANAGER = "triage-manager-github"; const SETTLE_MS = 90_000; // The listener waits two minutes for a manager's key before its wake mail fails; stock then stops it within its 30-second cancel grace. const UNWOKEN_MS = 5 * 60_000; +// The hub drops a silent sidecar's socket after stock's one-minute ping timeout, and the next reconcile pass checks it. +const DISCONNECT_MS = 60_000 + 2 * RECONCILE_INTERVAL_MS + SETTLE_MS; const DETERMINISTIC = `${OWNER}/deterministic`; const JUDGED = `${OWNER}/judged`; @@ -28,6 +30,7 @@ const PUSHED = `${OWNER}/pushed`; const HELD = `${OWNER}/held`; const WOKEN = `${OWNER}/woken`; const UNWOKEN = `${OWNER}/unwoken`; +const DISCONNECTED = `${OWNER}/disconnected`; // Interchange's `liveWorkflowRunStatuses`: an anchor is born `deployed` and flips to `running` on its first trigger. const LIVE_DEPLOYMENT_STATUSES = new Set(["deployed", "running"]); @@ -67,6 +70,7 @@ beforeAll(async () => { github.addRepo(HELD, policy([MAX_FILES], ASSIGN_ON_PASS)); github.addRepo(WOKEN, policy([MAX_FILES, FOCUSED])); github.addRepo(UNWOKEN, policy([MAX_FILES])); + github.addRepo(DISCONNECTED, policy([MAX_FILES])); const [created] = await Promise.all([createDatabase(), buildPortal(portalDir)]); database = created; hub = await startHub(database, github, portalDir); @@ -437,3 +441,43 @@ test("a manager whose wake mail failed is cancelled, and the next event deploys expect(claim).toMatchObject({ status: "ended", history: [{ head: second }] }); expect(comments(writes)).toEqual([expect.stringContaining("**Triage: Pass**")]); }); + +/** Every GitHub listener deployment's anchor run, oldest first, and the pid of the local sidecar it runs on. */ +async function listeners(): Promise<{ id: string; status: string; pid: number }[]> { + const found = await rows(`select r.id, r.status, a.external_ref from workflow_run r join workflow_definition d on d.id = r.definition_id left join workflow_run_launch_spec s on s.anchor_run_id = r.id join sidecar_allocation a on a.anchor_run_id = r.id where r.id = r.anchor_run_id and coalesce(s.frozen_approval_bundle -> 'projection' ->> 'id', d.name) = '${TRIAGE_LISTENERS.github}' order by r.created_at`); + return found.map((row) => ({ id: String(row["id"]), status: String(row["status"]), pid: Number(row["external_ref"]) })); +} + +function logged(msg: string): Record[] { + return hub.logs.filter((line) => line.includes(`"msg":"${msg}"`)).map((line) => JSON.parse(line) as Record); +} + +const disconnected = once(async function disconnected() { + await after(approved); + const [listener] = (await listeners()).filter((row) => isLiveDeployment(row.status)); + if (!listener) throw new Error("The GitHub listener is not live."); + // A frozen sidecar keeps its process and stops pinging, so stock drops its socket, and with it the listener's route, + // while the run stays live and its allocation allocated. + process.kill(listener.pid, "SIGSTOP"); + try { + await until("a pass to find the listener's sidecar disconnected", function seen() { + return logged("listener_disconnected").find((entry) => entry["deployment"] === listener.id); + }, DISCONNECT_MS); + } finally { + process.kill(listener.pid, "SIGCONT"); + } + const head = sha("disconnected-1"); + const missed = github.misses.length; + await pushed(DISCONNECTED, pullOf(1, head)); + await settled(DISCONNECTED, 1); + return { listener, unroutable: logged("listener_unroutable"), writes: github.writesTo(DISCONNECTED, 1), misses: github.misses.slice(missed) }; +}); + +test("a listener whose sidecar disconnected is left to stock, and once it reconnects the next pull request gets its verdict", async () => { + const { listener, unroutable, writes, misses } = await disconnected(); + expect(unroutable).toEqual([]); + expect((await listeners()).filter((row) => isLiveDeployment(row.status)).map((row) => row.id)).toEqual([listener.id]); + expect(misses).toEqual([]); + expect(comments(writes)).toEqual([expect.stringContaining("**Triage: Pass**")]); + expect(labelled(writes)).toEqual([VERDICT_LABELS.pass]); +}); diff --git a/tests/chain/stack.ts b/tests/chain/stack.ts index e2f33190f..1ee232c19 100644 --- a/tests/chain/stack.ts +++ b/tests/chain/stack.ts @@ -27,6 +27,8 @@ const SECRETS = { export const HOOK_SECRET = "chain-hook-secret"; const READY_MS = 60_000; const LISTENER_START_MS = 10 * 60_000; +/** The reconcile pass, short so a scenario waits on it for seconds rather than the default five minutes. */ +export const RECONCILE_INTERVAL_MS = 15_000; export type Database = { name: string; url: string; config: typeof ADMIN_DB }; @@ -90,6 +92,7 @@ export async function startHub(database: Database, github: FakeGithub, portalDir BETTER_AUTH_BASE_URL: origin, GITHUB_API_ORIGIN: github.origin, TRIAGE_COALESCE_QUIET_MS: "500", + TRIAGE_RECONCILE_INTERVAL_MS: String(RECONCILE_INTERVAL_MS), // 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,