From d40da00f0c165d75d355f7501a0836f28ad79d67 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 27 Oct 2026 08:25:13 -0700 Subject: [PATCH 1/5] feat(hub): replace a listener whose mail the hub cannot route --- apps/hub/src/server.ts | 10 ++- apps/hub/src/workflow-release/reads.ts | 26 ++++++- .../workflow-release/rollout.postgres.test.ts | 75 ++++++++++++++++++- apps/hub/src/workflow-release/rollout.ts | 14 +++- apps/hub/src/workflow-release/triggers.ts | 41 ++++++++-- docs/ARCHITECTURE.md | 16 +++- tests/chain/chain.test.ts | 60 ++++++++++++++- 7 files changed, 228 insertions(+), 14 deletions(-) diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index e18f1a231..01f3e0bc8 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -68,7 +68,7 @@ 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 { catalogModels, createWorkspaceLock, deployedAllocation, 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, localSidecarMaxDeployments, railwaySandboxSettings, signInSettings, triageCoalesceWindow, triageReconcileIntervalMs, triageReconcilePolicy } from "./env.js"; import { HOOK_MOUNT_PATH } from "./hooks.js"; @@ -588,9 +588,17 @@ const watchListeners = createListenerWatch({ return connectedSources(composition.db, tenantId); }, listener: liveListenerOf, + allocation: function allocationOf(runId) { + return deployedAllocation(composition.db, runId); + }, + routable: composition.sidecarRouter.isAllocatedWorkflowActive, 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..2db49adb3 100644 --- a/apps/hub/src/workflow-release/rollout.postgres.test.ts +++ b/apps/hub/src/workflow-release/rollout.postgres.test.ts @@ -367,7 +367,7 @@ test("a roll that fails while a request waits out its quiet window rolls again w test("a connected host found with no live listener rolls its workspace", async () => { const ensured: string[] = []; - const listeners: Partial> = { github: "run_listener" }; + const listeners: Partial> = { github: { runId: "run_listener" } }; const watch = createListenerWatch({ async connectedSources() { return ["github", "bitbucket-datacenter"]; @@ -375,17 +375,88 @@ test("a connected host found with no live listener rolls its workspace", async ( async listener(_tenantId, source) { return listeners[source]; }, + async allocation() { + return undefined; + }, + async routable() { + return true; + }, ensure(tenantId) { ensured.push(tenantId); }, + replace() { + return true; + }, + log() {}, }); await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); - listeners["bitbucket-datacenter"] = "run_bitbucket"; + listeners["bitbucket-datacenter"] = { runId: "run_bitbucket" }; await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); }); +test("a listener whose mail the router cannot route, or whose check throws, is started again once, and each is logged", 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 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([["bitbucket-datacenter", "github"]]); + expect(logged).toEqual([ + { level: "warn", msg: "listener_unroutable", tenantId: "tnt_unroutable", source: "github", deployment: "run_github", allocationId: "alc_github", generation: 1, reason: "the hub holds no route to it" }, + { level: "warn", msg: "listener_unroutable", tenantId: "tnt_unroutable", source: "bitbucket-datacenter", deployment: "run_bitbucket", allocationId: "alc_bitbucket", generation: 1, reason: "Allocation alc_bitbucket is not connected" }, + ]); +}); + test("PostgreSQL: a person keeps the Deployer role until a roll leaves every host's listener running", async () => { const { tenantId, principalId, fake, rollout } = await workspace("revoke"); await rollout.roll(tenantId, NO_REDEPLOY); diff --git a/apps/hub/src/workflow-release/rollout.ts b/apps/hub/src/workflow-release/rollout.ts index 1fb0bbff1..9ce97b352 100644 --- a/apps/hub/src/workflow-release/rollout.ts +++ b/apps/hub/src/workflow-release/rollout.ts @@ -277,5 +277,17 @@ 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; + if (state.timer) state.redeploy.add(source); + else schedule(tenantId, [source], timing.quietMs, true); + 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..dba32acc3 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 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,42 @@ 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 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 the hub can no longer route its mail. Stock acts on a lost route only when a sidecar connects or + * reports a stop, or once it stayed disconnected past its deployment's limit, and until then the listener stays live. + */ export function createListenerWatch(deps: ListenerWatchDeps) { + async function unroutable(target: AllocatedSidecarTarget): Promise { + try { + return (await deps.routable(target)) ? undefined : "the hub holds no route to it"; + } catch (error) { + return errorText(error); + } + } + 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; + const listener = await deps.listener(tenantId, source); + if (!listener) { + deps.ensure(tenantId); + return; + } + const target = await deps.allocation(listener.runId); + if (!target) continue; + const reason = await unroutable(target); + if (reason === undefined || !deps.replace(tenantId, source)) continue; + deps.log({ level: "warn", msg: "listener_unroutable", tenantId, source, deployment: listener.runId, allocationId: target.allocationId, generation: target.generation, reason }); } }; } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 1c0e055ec..e2d672255 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -241,7 +241,7 @@ 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 the hub cannot route mail 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 +298,20 @@ 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`, so until then a listener stays live while + the hub holds no route to it. Each reconcile + pass therefore asks the router, through `isAllocatedWorkflowActive`, the + check stock's mail routing uses, whether it routes the generation of each + host's live listener whose first deploy completed. A `false` answer or a + check that throws, as for a sidecar whose socket closed, starts that host's + listener again exactly as a credential write does, and the hub 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. - `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..2506a6934 100644 --- a/tests/chain/chain.test.ts +++ b/tests/chain/chain.test.ts @@ -8,7 +8,7 @@ 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"; @@ -20,6 +20,9 @@ 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 reconcile pass that finds the +// listener unroutable comes up to five minutes later. +const REPLACE_MS = 9 * 60_000; const DETERMINISTIC = `${OWNER}/deterministic`; const JUDGED = `${OWNER}/judged`; @@ -28,6 +31,7 @@ const PUSHED = `${OWNER}/pushed`; const HELD = `${OWNER}/held`; const WOKEN = `${OWNER}/woken`; const UNWOKEN = `${OWNER}/unwoken`; +const UNROUTABLE = `${OWNER}/unroutable`; // 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 +71,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(UNROUTABLE, policy([MAX_FILES])); const [created] = await Promise.all([createDatabase(), buildPortal(portalDir)]); database = created; hub = await startHub(database, github, portalDir); @@ -437,3 +442,56 @@ 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 loggedUnroutable(): Record[] { + return hub.logs.filter((line) => line.includes(`"msg":"listener_unroutable"`)).map((line) => JSON.parse(line) as Record); +} + +/** Wakes the frozen sidecar so the hub can stop it, unless the hub already killed it. */ +function resume(pid: number): void { + try { + process.kill(pid, "SIGCONT"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; + } +} + +const unroutable = once(async function unroutable() { + await after(approved); + const [old] = (await listeners()).filter((row) => isLiveDeployment(row.status)); + if (!old) 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(old.pid, "SIGSTOP"); + let fresh: { id: string; status: string; pid: number }; + try { + fresh = await until("a fresh listener", async function rolled() { + const now = await listeners(); + const newest = now.at(-1)!; + const replaced = now.find((row) => row.id === old.id); + return newest.id !== old.id && isLiveDeployment(newest.status) && !isLiveDeployment(replaced!.status) ? newest : undefined; + }, REPLACE_MS); + } finally { + resume(old.pid); + } + const head = sha("unroutable-1"); + const missed = github.misses.length; + await pushed(UNROUTABLE, pullOf(1, head)); + await settled(UNROUTABLE, 1); + return { old, fresh, logged: loggedUnroutable(), writes: github.writesTo(UNROUTABLE, 1), misses: github.misses.slice(missed) }; +}); + +test("a listener whose mail the hub can no longer route is replaced once, and the next pull request gets its verdict", async () => { + const { old, fresh, logged, writes, misses } = await unroutable(); + expect(logged).toEqual([expect.objectContaining({ tenantId: workspace.tenantId, source: "github", deployment: old.id })]); + expect((await listeners()).filter((row) => isLiveDeployment(row.status)).map((row) => row.id)).toEqual([fresh.id]); + expect(misses).toEqual([]); + expect(comments(writes)).toEqual([expect.stringContaining("**Triage: Pass**")]); + expect(labelled(writes)).toEqual([VERDICT_LABELS.pass]); +}); From f638ea40b06b3a9b2de408f2e57eda9e6463dc87 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 27 Oct 2026 08:36:03 -0700 Subject: [PATCH 2/5] fix(hub): leave a listener whose sidecar disconnected to stock --- .../workflow-release/rollout.postgres.test.ts | 12 ++-- apps/hub/src/workflow-release/triggers.ts | 26 ++++---- docs/ARCHITECTURE.md | 25 ++++---- tests/chain/chain.test.ts | 62 +++++++------------ tests/chain/stack.ts | 3 + 5 files changed, 62 insertions(+), 66 deletions(-) diff --git a/apps/hub/src/workflow-release/rollout.postgres.test.ts b/apps/hub/src/workflow-release/rollout.postgres.test.ts index 2db49adb3..00f889e28 100644 --- a/apps/hub/src/workflow-release/rollout.postgres.test.ts +++ b/apps/hub/src/workflow-release/rollout.postgres.test.ts @@ -396,7 +396,7 @@ test("a connected host found with no live listener rolls its workspace", async ( expect(ensured).toEqual(["tnt_watch"]); }); -test("a listener whose mail the router cannot route, or whose check throws, is started again once, and each is logged", async () => { +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, @@ -450,10 +450,14 @@ test("a listener whose mail the router cannot route, or whose check throws, is s await Bun.sleep(10); await watch("tnt_unroutable"); await Bun.sleep(10); - expect(rolled).toEqual([["bitbucket-datacenter", "github"]]); + 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: "the hub holds no route to it" }, - { level: "warn", msg: "listener_unroutable", tenantId: "tnt_unroutable", source: "bitbucket-datacenter", deployment: "run_bitbucket", allocationId: "alc_bitbucket", generation: 1, reason: "Allocation alc_bitbucket is not connected" }, + { 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, ]); }); diff --git a/apps/hub/src/workflow-release/triggers.ts b/apps/hub/src/workflow-release/triggers.ts index dba32acc3..c2b723429 100644 --- a/apps/hub/src/workflow-release/triggers.ts +++ b/apps/hub/src/workflow-release/triggers.ts @@ -69,18 +69,11 @@ export type ListenerWatchDeps = { /** * 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 the hub can no longer route its mail. Stock acts on a lost route only when a sidecar connects or - * reports a stop, or once it stayed disconnected past its deployment's limit, and until then the listener stays live. + * 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. */ export function createListenerWatch(deps: ListenerWatchDeps) { - async function unroutable(target: AllocatedSidecarTarget): Promise { - try { - return (await deps.routable(target)) ? undefined : "the hub holds no route to it"; - } catch (error) { - return errorText(error); - } - } - return async function watch(tenantId: string): Promise { for (const source of await deps.connectedSources(tenantId)) { const listener = await deps.listener(tenantId, source); @@ -90,9 +83,16 @@ export function createListenerWatch(deps: ListenerWatchDeps) { } const target = await deps.allocation(listener.runId); if (!target) continue; - const reason = await unroutable(target); - if (reason === undefined || !deps.replace(tenantId, source)) continue; - deps.log({ level: "warn", msg: "listener_unroutable", tenantId, source, deployment: listener.runId, allocationId: target.allocationId, generation: target.generation, reason }); + const entry = { tenantId, source, deployment: listener.runId, allocationId: target.allocationId, generation: target.generation }; + let routed: boolean; + try { + routed = await deps.routable(target); + } catch (error) { + 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 e2d672255..145ff7463 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, or with one the hub cannot route mail to. 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` @@ -302,16 +303,18 @@ any other tenant or an unknown sender. 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`, so until then a listener stays live while - the hub holds no route to it. Each reconcile - pass therefore asks the router, through `isAllocatedWorkflowActive`, the - check stock's mail routing uses, whether it routes the generation of each - host's live listener whose first deploy completed. A `false` answer or a - check that throws, as for a sidecar whose socket closed, starts that host's - listener again exactly as a credential write does, and the hub 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. + deployment's `maxDisconnected`. Each reconcile pass therefore asks the + router, through `isAllocatedWorkflowActive`, the check stock's mail routing + uses, whether it routes the generation of each host's live listener whose + first deploy completed. 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. - `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 2506a6934..3d77d78c6 100644 --- a/tests/chain/chain.test.ts +++ b/tests/chain/chain.test.ts @@ -12,7 +12,7 @@ import { TRIAGE_LISTENERS, VERDICT_LABELS, WAITING_LABEL } from "@corbits/triage 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,9 +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 reconcile pass that finds the -// listener unroutable comes up to five minutes later. -const REPLACE_MS = 9 * 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`; @@ -31,7 +30,7 @@ const PUSHED = `${OWNER}/pushed`; const HELD = `${OWNER}/held`; const WOKEN = `${OWNER}/woken`; const UNWOKEN = `${OWNER}/unwoken`; -const UNROUTABLE = `${OWNER}/unroutable`; +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"]); @@ -71,7 +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(UNROUTABLE, 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); @@ -449,48 +448,35 @@ async function listeners(): Promise<{ id: string; status: string; pid: number }[ return found.map((row) => ({ id: String(row["id"]), status: String(row["status"]), pid: Number(row["external_ref"]) })); } -function loggedUnroutable(): Record[] { - return hub.logs.filter((line) => line.includes(`"msg":"listener_unroutable"`)).map((line) => JSON.parse(line) as Record); +function logged(msg: string): Record[] { + return hub.logs.filter((line) => line.includes(`"msg":"${msg}"`)).map((line) => JSON.parse(line) as Record); } -/** Wakes the frozen sidecar so the hub can stop it, unless the hub already killed it. */ -function resume(pid: number): void { - try { - process.kill(pid, "SIGCONT"); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; - } -} - -const unroutable = once(async function unroutable() { +const disconnected = once(async function disconnected() { await after(approved); - const [old] = (await listeners()).filter((row) => isLiveDeployment(row.status)); - if (!old) 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, + 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(old.pid, "SIGSTOP"); - let fresh: { id: string; status: string; pid: number }; + process.kill(listener.pid, "SIGSTOP"); try { - fresh = await until("a fresh listener", async function rolled() { - const now = await listeners(); - const newest = now.at(-1)!; - const replaced = now.find((row) => row.id === old.id); - return newest.id !== old.id && isLiveDeployment(newest.status) && !isLiveDeployment(replaced!.status) ? newest : undefined; - }, REPLACE_MS); + 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 { - resume(old.pid); + process.kill(listener.pid, "SIGCONT"); } - const head = sha("unroutable-1"); + const head = sha("disconnected-1"); const missed = github.misses.length; - await pushed(UNROUTABLE, pullOf(1, head)); - await settled(UNROUTABLE, 1); - return { old, fresh, logged: loggedUnroutable(), writes: github.writesTo(UNROUTABLE, 1), misses: github.misses.slice(missed) }; + 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 mail the hub can no longer route is replaced once, and the next pull request gets its verdict", async () => { - const { old, fresh, logged, writes, misses } = await unroutable(); - expect(logged).toEqual([expect.objectContaining({ tenantId: workspace.tenantId, source: "github", deployment: old.id })]); - expect((await listeners()).filter((row) => isLiveDeployment(row.status)).map((row) => row.id)).toEqual([fresh.id]); +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, From 7cf9fa5157c626424a2c36759ff5d2741ef1b871 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 27 Oct 2026 08:50:05 -0700 Subject: [PATCH 3/5] fix(hub): wait for a replaced listener to stop and report one that stays live --- apps/hub/src/server.ts | 14 ++++++- .../workflow-release/rollout.postgres.test.ts | 38 +++++++++++++++++-- apps/hub/src/workflow-release/rollout.ts | 38 ++++++++++++++++++- apps/hub/src/workflow-release/triggers.ts | 12 ++++-- docs/ARCHITECTURE.md | 14 +++++-- 5 files changed, 103 insertions(+), 13 deletions(-) diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index 01f3e0bc8..d01111e87 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -69,7 +69,7 @@ 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, deployedAllocation, deploymentState, liveDeployments } from "./workflow-release/reads.js"; -import { createRolloutQueue, createWorkflowRollout, type DeployRequest } from "./workflow-release/rollout.js"; +import { createRolloutQueue, createWorkflowRollout, stopOverdue, 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,7 @@ const workflowRollout = createWorkflowRollout({ async cancel(tenantId, runId) { await composition.workflowLifecycleService.requestCancellation(tenantId, runId, "replaced"); }, + stop: composition.workflowLifecycleService.getStatus, lock: createWorkspaceLock(composition.db.$client), sleep: Bun.sleep, now, @@ -583,11 +584,22 @@ 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 && stopOverdue(run, now())) 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); }, diff --git a/apps/hub/src/workflow-release/rollout.postgres.test.ts b/apps/hub/src/workflow-release/rollout.postgres.test.ts index 00f889e28..f36e205eb 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; deadline?: string }; /** A workspace with its publisher, rolled against fake deployments. */ async function workspace(name: string) { @@ -77,6 +77,7 @@ async function workspace(name: string) { models: [DECISION] as ModelInfo[], starts: "live" as DeployState, deployFails: false, + stopHangs: false, clock: Date.now(), runs: [] as Run[], requests: [] as DeployRequest[], @@ -109,7 +110,14 @@ async function workspace(name: string) { }, async cancel(_tenant, runId) { const run = fake.runs.find((row) => row.runId === runId); - if (run) run.cancelling = true; + if (!run) return; + run.cancelling = true; + run.deadline = new Date(fake.clock + 30_000).toISOString(); + }, + async stop(_tenant, runId) { + const run = fake.runs.find((row) => row.runId === runId); + if (!run) return null; + return { status: run.cancelling && !fake.stopHangs ? "cancelled" : "running", cancellationDeadline: run.deadline ?? null }; }, lock: createWorkspaceLock(handle.db.$client), async sleep(ms) { @@ -229,6 +237,17 @@ 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 that stock never stops is waited on until its stop is overdue, then logged as stuck", async () => { + const { tenantId, fake, rollout, live } = await workspace("stuck"); + 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(90_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 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); @@ -365,8 +384,9 @@ 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 a listener stock gave up stopping is reported on every pass", async () => { const ensured: string[] = []; + const logged: Record[] = []; const listeners: Partial> = { github: { runId: "run_listener" } }; const watch = createListenerWatch({ async connectedSources() { @@ -375,6 +395,9 @@ 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 === "github" ? ["run_stuck"] : []; + }, async allocation() { return undefined; }, @@ -387,13 +410,17 @@ test("a connected host found with no live listener rolls its workspace", async ( replace() { return true; }, - log() {}, + log(entry) { + logged.push(entry); + }, }); await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); listeners["bitbucket-datacenter"] = { runId: "run_bitbucket" }; await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); + const stuck = { level: "warn", msg: "listener_cancel_stuck", tenantId: "tnt_watch", source: "github", 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 () => { @@ -427,6 +454,9 @@ test("a listener whose connected sidecar the router holds no route to is started 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 }; }, diff --git a/apps/hub/src/workflow-release/rollout.ts b/apps/hub/src/workflow-release/rollout.ts index 9ce97b352..261b3e41a 100644 --- a/apps/hub/src/workflow-release/rollout.ts +++ b/apps/hub/src/workflow-release/rollout.ts @@ -1,5 +1,6 @@ // 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 { 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 +14,9 @@ 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"; +/** A deployment as stock's lifecycle status reports it; `cancellationDeadline` is set once its stop was requested. */ +export type RunStop = { status: string; cancellationDeadline: string | null }; + /** 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. */ + stop: (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; @@ -44,6 +50,15 @@ const ROLL_FAILED = "Updating Triage failed; see the hub log."; export const START_DEADLINE_MS = 5 * 60_000; const START_POLL_MS = 2_000; +/** Stock's `unreachableStopGraceMs`, which the hub leaves at its default: how long past its deadline stock tries a stop. */ +const STOP_GRACE_MS = 60_000; + +/** Whether a deployment whose stop was requested is still live after stock stopped trying to stop it. */ +export function stopOverdue(run: RunStop, now: Date): boolean { + if (!schema.isLiveWorkflowRunStatus(run.status) || run.cancellationDeadline === null) return false; + return now.getTime() >= Date.parse(run.cancellationDeadline) + STOP_GRACE_MS; +} + /** Each host's workflows from before the listener, which nothing mails any more. */ const REPLACED: Record = { github: ["pr-triage", "pr-triage-github", "pr-triage-historical", "pr-code-review-github", "pr-workflow-anthropic-github"], @@ -84,13 +99,34 @@ export function createWorkflowRollout(deps: RolloutDeps) { throw new Error(`Listener ${runId} did not start within five minutes and was stopped.`); } + /** + * Waits until a deployment whose stop was requested ends. One stock gave up stopping stays live, and is only logged: + * nothing the hub may do stops it. + */ + async function waitUntilStopped(tenantId: string, source: RepoSource, runId: string): Promise { + for (;;) { + const run = await deps.stop(tenantId, runId); + if (!run || !schema.isLiveWorkflowRunStatus(run.status)) return; + if (run.cancellationDeadline === null) throw new Error(`Deployment ${runId} is live with no stop requested.`); + if (stopOverdue(run, deps.now())) { + deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId }); + return; + } + await deps.sleep(START_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: 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(live.runId); } } + await Promise.all(stopped.map((runId) => waitUntilStopped(tenantId, source, runId))); record(tenantId, { source, at: deps.now().toISOString() }); deps.log({ level: "info", msg, tenantId, source, deployment: kept }); } diff --git a/apps/hub/src/workflow-release/triggers.ts b/apps/hub/src/workflow-release/triggers.ts index c2b723429..053cd7fd5 100644 --- a/apps/hub/src/workflow-release/triggers.ts +++ b/apps/hub/src/workflow-release/triggers.ts @@ -2,7 +2,7 @@ // listener the hub cannot route mail to. import { and, eq } from "drizzle-orm"; import { schema, type DB } from "@intx/db"; -import type { AllocatedSidecarTarget } from "@intx/hub-sessions"; +import { SidecarIdentityValidationError, type AllocatedSidecarTarget } from "@intx/hub-sessions"; import { asRecord, errorText } from "@corbits/common"; import { REPO_SOURCES, type RepoSource } from "@corbits/triage-contracts"; @@ -58,6 +58,8 @@ export async function releaseWriteOf(db: DB["db"], req: Request, path: string): export type ListenerWatchDeps = { connectedSources: (tenantId: string) => 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. */ @@ -71,11 +73,13 @@ export type ListenerWatchDeps = { * 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. + * 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)) { + 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); @@ -88,7 +92,9 @@ export function createListenerWatch(deps: ListenerWatchDeps) { try { routed = await deps.routable(target); } catch (error) { - deps.log({ level: "debug", msg: "listener_disconnected", ...entry, reason: errorText(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; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 145ff7463..24c4e0f8a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -304,9 +304,10 @@ any other tenant or an unknown sender. 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`, the check stock's mail routing - uses, whether it routes the generation of each host's live listener whose - first deploy completed. When the listener's sidecar is connected but 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. 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 @@ -314,7 +315,12 @@ any other tenant or an unknown sender. 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. + for it, or `listener_identity_unknown` when the router could not validate + the sidecar's identity. After asking stock to stop the listeners a roll + replaces, the roll waits for each to end, until the stop's deadline plus + the minute stock keeps trying an unreachable stop; one still live then is + logged as `listener_cancel_stuck`, and every reconcile pass logs it again + while it stays live. - `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 From 7ed87e19c28fddd527dc8b65ca285d3f0527b48b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 27 Oct 2026 10:19:04 -0700 Subject: [PATCH 4/5] fix(hub): keep the rollout's lifecycle calls bound and wait out a stop deadline stock has not set yet Pass the lifecycle status and router readiness calls bound to their owners. A replaced deployment whose stop deadline is not set yet is still waited on instead of failing the roll, and a replaced workflow that stays live is logged as rollout_cancel_stuck with its name. --- apps/hub/src/server.ts | 4 +-- .../workflow-release/rollout.postgres.test.ts | 27 +++++++++++++++++-- apps/hub/src/workflow-release/rollout.ts | 21 ++++++++------- docs/ARCHITECTURE.md | 15 ++++++----- 4 files changed, 48 insertions(+), 19 deletions(-) diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index d01111e87..c678a808f 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -570,7 +570,7 @@ const workflowRollout = createWorkflowRollout({ async cancel(tenantId, runId) { await composition.workflowLifecycleService.requestCancellation(tenantId, runId, "replaced"); }, - stop: composition.workflowLifecycleService.getStatus, + stop: (tenantId, runId) => composition.workflowLifecycleService.getStatus(tenantId, runId), lock: createWorkspaceLock(composition.db.$client), sleep: Bun.sleep, now, @@ -603,7 +603,7 @@ const watchListeners = createListenerWatch({ allocation: function allocationOf(runId) { return deployedAllocation(composition.db, runId); }, - routable: composition.sidecarRouter.isAllocatedWorkflowActive, + routable: (target) => composition.sidecarRouter.isAllocatedWorkflowActive(target), ensure: function rollWorkspace(tenantId) { if (workflowsBundled) workflowReleases.ensure(tenantId); }, diff --git a/apps/hub/src/workflow-release/rollout.postgres.test.ts b/apps/hub/src/workflow-release/rollout.postgres.test.ts index f36e205eb..878f890a1 100644 --- a/apps/hub/src/workflow-release/rollout.postgres.test.ts +++ b/apps/hub/src/workflow-release/rollout.postgres.test.ts @@ -78,6 +78,7 @@ async function workspace(name: string) { starts: "live" as DeployState, deployFails: false, stopHangs: false, + deadlineAfterPolls: 0, clock: Date.now(), runs: [] as Run[], requests: [] as DeployRequest[], @@ -112,11 +113,12 @@ async function workspace(name: string) { const run = fake.runs.find((row) => row.runId === runId); if (!run) return; run.cancelling = true; - run.deadline = new Date(fake.clock + 30_000).toISOString(); + if (fake.deadlineAfterPolls === 0) run.deadline = new Date(fake.clock + 30_000).toISOString(); }, async stop(_tenant, runId) { const run = fake.runs.find((row) => row.runId === runId); if (!run) return null; + if (run.cancelling && !run.deadline && --fake.deadlineAfterPolls === 0) run.deadline = new Date(fake.clock + 30_000).toISOString(); return { status: run.cancelling && !fake.stopHangs ? "cancelled" : "running", cancellationDeadline: run.deadline ?? null }; }, lock: createWorkspaceLock(handle.db.$client), @@ -237,14 +239,35 @@ 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 that stock never stops is waited on until its stop is overdue, then logged as stuck", async () => { +test("PostgreSQL: a replaced listener or workflow that stock never stops is waited on until its stop is overdue, then logged as stuck", 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, source: null }); fake.stopHangs = true; const before = fake.clock; expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); expect(fake.clock - before).toBeGreaterThanOrEqual(90_000); 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 whose stop deadline stock sets late is waited on until 60 s past that deadline, and the roll succeeds", async () => { + const { tenantId, fake, rollout, live } = await workspace("late"); + await rollout.roll(tenantId, NO_REDEPLOY); + fake.stopHangs = true; + fake.deadlineAfterPolls = 3; + const before = fake.clock; + expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); + const deadline = Date.parse(fake.runs.find((run) => run.runId === "run_1")!.deadline!); + expect(deadline).toBe(before + 4_000 + 30_000); + expect(fake.clock).toBeGreaterThanOrEqual(deadline + 60_000); + expect(fake.clock).toBeLessThan(deadline + 60_000 + 2_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); diff --git a/apps/hub/src/workflow-release/rollout.ts b/apps/hub/src/workflow-release/rollout.ts index 261b3e41a..21e602ad0 100644 --- a/apps/hub/src/workflow-release/rollout.ts +++ b/apps/hub/src/workflow-release/rollout.ts @@ -53,7 +53,10 @@ const START_POLL_MS = 2_000; /** Stock's `unreachableStopGraceMs`, which the hub leaves at its default: how long past its deadline stock tries a stop. */ const STOP_GRACE_MS = 60_000; -/** Whether a deployment whose stop was requested is still live after stock stopped trying to stop it. */ +/** + * Whether a deployment whose stop was requested is still live after stock stopped trying to stop it, counted from the + * cancellation deadline stock set rather than from when the hub asked. + */ export function stopOverdue(run: RunStop, now: Date): boolean { if (!schema.isLiveWorkflowRunStatus(run.status) || run.cancellationDeadline === null) return false; return now.getTime() >= Date.parse(run.cancellationDeadline) + STOP_GRACE_MS; @@ -100,16 +103,16 @@ export function createWorkflowRollout(deps: RolloutDeps) { } /** - * Waits until a deployment whose stop was requested ends. One stock gave up stopping stays live, and is only logged: - * nothing the hub may do stops it. + * Waits until a deployment whose stop was requested ends; one whose deadline stock has not set yet is still being + * stopped. One stock gave up stopping stays live, and is only logged: nothing the hub may do stops it. */ - async function waitUntilStopped(tenantId: string, source: RepoSource, runId: string): Promise { + async function waitUntilStopped(tenantId: string, source: RepoSource, workflow: string, runId: string): Promise { for (;;) { const run = await deps.stop(tenantId, runId); if (!run || !schema.isLiveWorkflowRunStatus(run.status)) return; - if (run.cancellationDeadline === null) throw new Error(`Deployment ${runId} is live with no stop requested.`); if (stopOverdue(run, deps.now())) { - deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId }); + if (workflow === TRIAGE_LISTENERS[source]) deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId }); + else deps.log({ level: "warn", msg: "rollout_cancel_stuck", tenantId, source, workflow, deployment: runId }); return; } await deps.sleep(START_POLL_MS); @@ -118,15 +121,15 @@ export function createWorkflowRollout(deps: RolloutDeps) { /** 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: string[] = []; + 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) continue; await deps.cancel(tenantId, live.runId); - stopped.push(live.runId); + stopped.push({ workflow, runId: live.runId }); } } - await Promise.all(stopped.map((runId) => waitUntilStopped(tenantId, source, 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 }); } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 24c4e0f8a..daa184c6a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -307,7 +307,8 @@ any other tenant or an unknown sender. 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. When the listener's sidecar is connected but the + 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 @@ -316,11 +317,13 @@ any other tenant or an unknown sender. 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 a roll - replaces, the roll waits for each to end, until the stop's deadline plus - the minute stock keeps trying an unreachable stop; one still live then is - logged as `listener_cancel_stuck`, and every reconcile pass logs it again - while it stays live. + the sidecar's identity. After asking stock to stop the listeners and + workflows a roll replaces, the roll waits for each to end, until the + deadline stock sets for the stop plus the minute stock keeps trying an + unreachable stop, and keeps waiting while that deadline is not set yet. A + listener still live then is logged as `listener_cancel_stuck`, and every + reconcile pass logs it again while it stays live; a replaced workflow is + logged 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 From ac998cbc1a44605da0c36ad2a8cf2392ee8dab85 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 27 Oct 2026 10:49:02 -0700 Subject: [PATCH 5/5] fix(hub): judge a stuck stop by stock's forced release and keep a roll bounded A replaced deployment counts as stuck once stock's lifecycle status shows its allocation no longer dispatchable, instead of the hub re-timing stock's deadline and grace. The stop wait is bounded at five minutes from the request, retries a failed status read three times, and logs rather than fails the roll. The listener watch checks every connected host on each pass. The rollout's status dependency is typed from stock's lifecycle response, its callbacks are named, and the queue reuses schedule for a replace. --- apps/hub/src/server.ts | 12 ++-- .../workflow-release/rollout.postgres.test.ts | 72 +++++++++++-------- apps/hub/src/workflow-release/rollout.ts | 65 +++++++++++------ apps/hub/src/workflow-release/triggers.ts | 2 +- docs/ARCHITECTURE.md | 26 +++---- 5 files changed, 110 insertions(+), 67 deletions(-) diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index c678a808f..12dd381d5 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -69,7 +69,7 @@ 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, deployedAllocation, deploymentState, liveDeployments } from "./workflow-release/reads.js"; -import { createRolloutQueue, createWorkflowRollout, stopOverdue, type DeployRequest } from "./workflow-release/rollout.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,7 +570,9 @@ const workflowRollout = createWorkflowRollout({ async cancel(tenantId, runId) { await composition.workflowLifecycleService.requestCancellation(tenantId, runId, "replaced"); }, - stop: (tenantId, runId) => composition.workflowLifecycleService.getStatus(tenantId, runId), + status(tenantId, runId) { + return composition.workflowLifecycleService.getStatus(tenantId, runId); + }, lock: createWorkspaceLock(composition.db.$client), sleep: Bun.sleep, now, @@ -590,7 +592,7 @@ async function stuckListeners(tenantId: string, source: RepoSource): Promise composition.sidecarRouter.isAllocatedWorkflowActive(target), + routable: function routableOf(target) { + return composition.sidecarRouter.isAllocatedWorkflowActive(target); + }, ensure: function rollWorkspace(tenantId) { if (workflowsBundled) workflowReleases.ensure(tenantId); }, diff --git a/apps/hub/src/workflow-release/rollout.postgres.test.ts b/apps/hub/src/workflow-release/rollout.postgres.test.ts index 878f890a1..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; deadline?: string }; +type Run = Deployment & { workflow: string; state: DeployState; polls: number }; /** A workspace with its publisher, rolled against fake deployments. */ async function workspace(name: string) { @@ -78,7 +78,8 @@ async function workspace(name: string) { starts: "live" as DeployState, deployFails: false, stopHangs: false, - deadlineAfterPolls: 0, + releasedAfterPolls: Infinity, + statusFails: 0, clock: Date.now(), runs: [] as Run[], requests: [] as DeployRequest[], @@ -103,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,15 +112,21 @@ async function workspace(name: string) { }, async cancel(_tenant, runId) { const run = fake.runs.find((row) => row.runId === runId); - if (!run) return; - run.cancelling = true; - if (fake.deadlineAfterPolls === 0) run.deadline = new Date(fake.clock + 30_000).toISOString(); + if (run) run.cancelling = true; }, - async stop(_tenant, runId) { + 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; - if (run.cancelling && !run.deadline && --fake.deadlineAfterPolls === 0) run.deadline = new Date(fake.clock + 30_000).toISOString(); - return { status: run.cancelling && !fake.stopHangs ? "cancelled" : "running", cancellationDeadline: run.deadline ?? 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) { @@ -229,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); @@ -239,14 +246,13 @@ 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 stock never stops is waited on until its stop is overdue, then logged as stuck", async () => { +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, source: null }); + 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; - const before = fake.clock; + fake.releasedAfterPolls = 3; expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); - expect(fake.clock - before).toBeGreaterThanOrEqual(90_000); expect(live()).toEqual(["run_2"]); const stuck = fake.logs.filter((entry) => String(entry["msg"]).endsWith("_cancel_stuck")); expect(stuck).toHaveLength(2); @@ -256,25 +262,33 @@ test("PostgreSQL: a replaced listener or workflow that stock never stops is wait ])); }, 120_000); -test("PostgreSQL: a replaced listener whose stop deadline stock sets late is waited on until 60 s past that deadline, and the roll succeeds", async () => { - const { tenantId, fake, rollout, live } = await workspace("late"); +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; - fake.deadlineAfterPolls = 3; const before = fake.clock; expect(await rollout.roll(tenantId, REDEPLOY_GITHUB)).toEqual(SUCCEEDED); - const deadline = Date.parse(fake.runs.find((run) => run.runId === "run_1")!.deadline!); - expect(deadline).toBe(before + 4_000 + 30_000); - expect(fake.clock).toBeGreaterThanOrEqual(deadline + 60_000); - expect(fake.clock).toBeLessThan(deadline + 60_000 + 2_000); + 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; @@ -407,10 +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, and a listener stock gave up stopping is reported on every pass", 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 logged: Record[] = []; - const listeners: Partial> = { github: { runId: "run_listener" } }; + const listeners: Partial> = { "bitbucket-datacenter": { runId: "run_bitbucket" } }; const watch = createListenerWatch({ async connectedSources() { return ["github", "bitbucket-datacenter"]; @@ -419,7 +433,7 @@ test("a connected host found with no live listener rolls its workspace, and a li return listeners[source]; }, async stuck(_tenantId, source) { - return source === "github" ? ["run_stuck"] : []; + return source === "bitbucket-datacenter" ? ["run_stuck"] : []; }, async allocation() { return undefined; @@ -439,10 +453,10 @@ test("a connected host found with no live listener rolls its workspace, and a li }); await watch("tnt_watch"); expect(ensured).toEqual(["tnt_watch"]); - listeners["bitbucket-datacenter"] = { runId: "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: "github", deployment: "run_stuck" }; + const stuck = { level: "warn", msg: "listener_cancel_stuck", tenantId: "tnt_watch", source: "bitbucket-datacenter", deployment: "run_stuck" }; expect(logged).toEqual([stuck, stuck]); }); diff --git a/apps/hub/src/workflow-release/rollout.ts b/apps/hub/src/workflow-release/rollout.ts index 21e602ad0..69726fbad 100644 --- a/apps/hub/src/workflow-release/rollout.ts +++ b/apps/hub/src/workflow-release/rollout.ts @@ -1,6 +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"; @@ -14,8 +15,7 @@ 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"; -/** A deployment as stock's lifecycle status reports it; `cancellationDeadline` is set once its stop was requested. */ -export type RunStop = { status: string; cancellationDeadline: string | null }; +export type RunStop = Pick; /** Releases a held workspace lock. */ export type Unlock = () => Promise; @@ -36,7 +36,7 @@ type RolloutDeps = PublishDeps & { state: (runId: string) => Promise; cancel: (tenantId: string, runId: string) => Promise; /** Stock's lifecycle status of the deployment; null once it is gone. */ - stop: (tenantId: string, runId: string) => Promise; + 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; @@ -48,18 +48,18 @@ 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; -/** Stock's `unreachableStopGraceMs`, which the hub leaves at its default: how long past its deadline stock tries a stop. */ -const STOP_GRACE_MS = 60_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 a deployment whose stop was requested is still live after stock stopped trying to stop it, counted from the - * cancellation deadline stock set rather than from when the hub asked. + * 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 stopOverdue(run: RunStop, now: Date): boolean { - if (!schema.isLiveWorkflowRunStatus(run.status) || run.cancellationDeadline === null) return false; - return now.getTime() >= Date.parse(run.cancellationDeadline) + STOP_GRACE_MS; +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. */ @@ -96,26 +96,50 @@ 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 whose deadline stock has not set yet is still being - * stopped. One stock gave up stopping stays live, and is only logged: nothing the hub may do stops it. + * 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 (;;) { - const run = await deps.stop(tenantId, runId); + 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 (stopOverdue(run, deps.now())) { - if (workflow === TRIAGE_LISTENERS[source]) deps.log({ level: "warn", msg: "listener_cancel_stuck", tenantId, source, deployment: runId }); - else deps.log({ level: "warn", msg: "rollout_cancel_stuck", tenantId, source, workflow, deployment: runId }); + if (stopGivenUp(run) || deps.now().getTime() >= deadline) { + logStuck(tenantId, source, workflow, runId); return; } - await deps.sleep(START_POLL_MS); + await deps.sleep(POLL_MS); } } @@ -323,8 +347,7 @@ export function createRolloutQueue(rollout: { roll: (tenantId: string, redeploy: function replace(tenantId: string, source: RepoSource): boolean { const state = stateOf(tenantId); if (state.running || state.redeploy.has(source)) return false; - if (state.timer) state.redeploy.add(source); - else schedule(tenantId, [source], timing.quietMs, true); + schedule(tenantId, [source], timing.quietMs, false); return true; } diff --git a/apps/hub/src/workflow-release/triggers.ts b/apps/hub/src/workflow-release/triggers.ts index 053cd7fd5..e0d4fd0e0 100644 --- a/apps/hub/src/workflow-release/triggers.ts +++ b/apps/hub/src/workflow-release/triggers.ts @@ -83,7 +83,7 @@ export function createListenerWatch(deps: ListenerWatchDeps) { const listener = await deps.listener(tenantId, source); if (!listener) { deps.ensure(tenantId); - return; + continue; } const target = await deps.allocation(listener.runId); if (!target) continue; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index daa184c6a..5646746da 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -308,22 +308,24 @@ any other tenant or an unknown sender. 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 + 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 waits for each to end, until the - deadline stock sets for the stop plus the minute stock keeps trying an - unreachable stop, and keeps waiting while that deadline is not set yet. A - listener still live then is logged as `listener_cancel_stuck`, and every - reconcile pass logs it again while it stays live; a replaced workflow is - logged once as `rollout_cancel_stuck` with its name. + 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