feat(hub): replace a listener whose mail the hub cannot route

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<string[]> {
+  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);

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<Depl
   if (row.allocation === null || (row.allocation === "allocated" && row.nextAttemptAt === null)) return "live";
   return "starting";
 }
+
+/**
+ * The allocation generation a deployment runs on once its first deploy completed, which stock's router then routes its
+ * mail to; undefined while it starts or once it stops.
+ */
+export async function deployedAllocation(db: DB["db"], runId: string): Promise<AllocatedSidecarTarget | undefined> {
+  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 };
+}

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,20 +236,59 @@ 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);
   expect(live()).toEqual(["run_cut_short"]);
   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,25 +421,111 @@ 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<Record<RepoSource, string>> = { github: "run_listener" };
+  const logged: Record<string, unknown>[] = [];
+  const listeners: Partial<Record<RepoSource, { runId: string }>> = { "bitbucket-datacenter": { runId: "run_bitbucket" } };
   const watch = createListenerWatch({
     async connectedSources() {
       return ["github", "bitbucket-datacenter"];
     },
     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<RepoSource, { runId: string }> = { github: { runId: "run_github" }, "bitbucket-datacenter": { runId: "run_bitbucket" } };
+  const routes: Record<string, () => boolean> = {
+    alc_github: () => false,
+    alc_bitbucket: () => {
+      throw new Error("Allocation alc_bitbucket is not connected");
+    },
+  };
+  const rolled: Array<RepoSource[]> = [];
+  const entered = Promise.withResolvers<void>();
+  const gate = Promise.withResolvers<void>();
+  const replaced = Promise.withResolvers<void>();
+  const rollout = {
+    async roll(_tenantId: string, redeploy: ReadonlySet<RepoSource>): Promise<RollOutcome> {
+      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<string, unknown>[] = [];
+  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 () => {