fix(hub): hold webhook mail while the listener is being replaced

apps/hub/src/github/backlog-triage.test.ts

@@ -5,6 +5,25 @@ import { checkPackName, emptyPack } from "@corbits/triage-contracts";
 const PACK = { ...emptyPack("octocat/hello"), checks: { "no-merge-conflicts": { enabled: true } } };
 import { createBacklogStart, createGithubBacklog } from "./backlog-triage.js";
 import { NoLiveDeploymentError } from "./deployment.js";
+import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js";
+
+/** A listener takes every backlog here, so no head is held. */
+const NOT_HELD = {
+  async openHeads(): Promise<never> {
+    throw new Error("no head is held");
+  },
+  store: {
+    async load(): Promise<never> {
+      throw new Error("no head is held");
+    },
+    async save(): Promise<never> {
+      throw new Error("no head is held");
+    },
+  },
+  policy: DEFAULT_RECONCILE_POLICY,
+  now: () => new Date(),
+  retry() {},
+};
 
 /** No deployment holds the credential in these tests. */
 async function noKeyPush(): Promise<void> {}
@@ -32,6 +51,7 @@ function backlog(repo: Record<string, unknown>, sent: unknown[], sendMail = asyn
       return { status: "ok", pack: PACK, automatic: ["labels"], source: SOURCE };
     },
     sendMail,
+    ...NOT_HELD,
     log() {},
   });
 }
@@ -60,6 +80,7 @@ test("a Bitbucket Data Center repository's backlog goes to its listener like a G
     async sendMail(_tenant, payload) {
       sent.push(payload);
     },
+    ...NOT_HELD,
     log() {},
   });
   const started = await start(MEMBER.tenantId, { name: "TRI/widgets", source: "bitbucket-datacenter", connected: true, enabled: true });

apps/hub/src/github/backlog-triage.ts

@@ -4,20 +4,29 @@
 import { type } from "arktype";
 import { errorText } from "@corbits/common";
 import { triageStopped } from "@corbits/triage-contracts";
-import { isRunTriggerUnroutable } from "@corbits/webhooks";
 import { mailPayload } from "./bridge.js";
 import type { PolicyReader } from "./repo-policy.js";
-import { DeploymentNotReadyError, NoLiveDeploymentError } from "./deployment.js";
+import { ListenerUnroutedError, NoLiveDeploymentError } from "./deployment.js";
+import type { ReconcilePolicy } from "./reconcile-plan.js";
+import { holdHead } from "./triage-queue.js";
+import type { TriageStateStore } from "./triage-state-store.js";
 import { failure, type PortalCredentialDeps } from "./portal-credential.js";
 import { grantedTriageRecord, triageInputs } from "./pr-triage.js";
-import { sourceOf, type RepoRecord } from "./tenant-config.js";
+import { sourceOf, triageStateKey, type RepoRecord } from "./tenant-config.js";
 import type { Member } from "../integrations/app.js";
 
 export const BacklogBody = type({ repo: /^[\w.-]+\/[\w.-]+$/ });
 
 export type BacklogStartDeps = {
   readCheckPack: PolicyReader;
   sendMail: (tenantId: string, payload: unknown) => Promise<void>;
+  /** The repository's open pull requests and their heads, held when the listener takes no mail. */
+  openHeads: (tenantId: string, record: RepoRecord) => Promise<{ number: number; headSha: string; draft: boolean }[]>;
+  store: TriageStateStore;
+  policy: ReconcilePolicy;
+  now: () => Date;
+  /** Asks the reconciler for a pass soon. */
+  retry: () => void;
   log: (entry: Record<string, unknown>) => void;
 };
 
@@ -26,15 +35,28 @@ export type GithubBacklogDeps = PortalCredentialDeps & BacklogStartDeps;
 export type BacklogStart = (tenantId: string, record: RepoRecord) => Promise<Response>;
 
 export function createBacklogStart(deps: BacklogStartDeps): BacklogStart {
+  /** Each open head goes out once a listener is up, as the backlog mail would have sent it. */
+  async function holdOpenHeads(tenantId: string, record: RepoRecord, err: ListenerUnroutedError): Promise<void> {
+    const now = deps.now();
+    for (const pull of (await deps.openHeads(tenantId, record)).filter((open) => !open.draft)) {
+      await holdHead(deps.store, { tenantId, repo: triageStateKey(record), number: pull.number, headSha: pull.headSha }, err, now, deps.policy);
+    }
+    deps.retry();
+    deps.log({ level: "info", msg: "backlog_held", tenantId, repo: record.name, error: String(err) });
+  }
+
   return async function startBacklog(tenantId, record) {
     const repo = record.name;
     const inputs = await triageInputs(deps, tenantId, record);
     if (inputs instanceof Response) return inputs;
     try {
       await deps.sendMail(tenantId, { ...mailPayload(record, inputs.policy, inputs.pack, {}), kind: "backlog" });
     } catch (err) {
-      if (err instanceof DeploymentNotReadyError) return failure(409, err.code, err.message);
-      if (err instanceof NoLiveDeploymentError || isRunTriggerUnroutable(err)) return failure(503, "service_not_running", triageStopped(sourceOf(record)));
+      if (err instanceof ListenerUnroutedError) {
+        await holdOpenHeads(tenantId, record, err);
+        return Response.json({ status: "started" }, { status: 202 });
+      }
+      if (err instanceof NoLiveDeploymentError) return failure(503, "service_not_running", triageStopped(sourceOf(record)));
       deps.log({ level: "error", msg: "backlog_request_failed", tenantId, repo, error: String(err) });
       return failure(502, "hub_unavailable", `Could not start triage of ${repo}: ${errorText(err)}. Try again.`);
     }

apps/hub/src/github/coalescer.test.ts

@@ -97,6 +97,7 @@ function harness(rows: PrTriageRow[] = [], overrides: Partial<CoalescerDeps> = {
     async sendMail(job) {
       mails.push(job.payload as Record<string, unknown>);
     },
+    retry() {},
     async openHeadFor() {
       return async function openHead() {
         return { headSha: "sha1", draft: false };

apps/hub/src/github/coalescer.ts

@@ -22,7 +22,7 @@ import { GithubRateLimited } from "./github-usage.js";
 import type { PrMail } from "./normalize.js";
 import { triageStateKey, type RepoRecord } from "./tenant-config.js";
 import type { OpenHeadReader } from "./tenant-open-heads.js";
-import { deliveryFailure, failHead, queueHead, restoreHead, type QueuedHead, type Remembered } from "./triage-queue.js";
+import { failDelivery, failHead, queueHead, restoreHead, type QueuedHead, type Remembered } from "./triage-queue.js";
 import { TriageStateConflictError, type TriageStateStore } from "./triage-state-store.js";
 import { createKeyedSerial } from "../keyed-serial.js";
 
@@ -57,6 +57,8 @@ export type Accepted = "queued" | "coalesced";
 export type CoalescerDeps = {
   store: TriageStateStore;
   sendMail: (job: TriageJob) => Promise<unknown>;
+  /** Asks the reconciler for a pass soon, after a head's mail reached no listener. */
+  retry: () => void;
   openHeadFor: (tenantId: string) => Promise<OpenHeadReader | undefined>;
   window: CoalesceWindow;
   setTimer: SetTimer;
@@ -164,8 +166,9 @@ export function createCoalescer(deps: CoalescerDeps): Coalescer {
       await deps.sendMail({ ...head, record: latest.record, payload });
     } catch (err) {
       if (err instanceof GithubRateLimited) return drop(open, entry, "github_rate_limited");
-      deps.log({ level: "error", msg: "forward_failed", ...entry, error: String(err) });
-      await failHead(deps.store, head, deliveryFailure(err), deps.now());
+      const unrouted = await failDelivery(deps.store, head, err, deps.now());
+      deps.log({ level: unrouted ? "info" : "error", msg: "forward_failed", ...entry, error: String(err) });
+      if (unrouted) deps.retry();
       return;
     }
     deps.log({ level: "info", msg: "forwarded", ...entry, workflow: latest.workflow });

apps/hub/src/github/deployment.ts

@@ -21,22 +21,42 @@ export class TriageStoppedError extends NoLiveDeploymentError {
 }
 
 export class DeploymentNotReadyError extends Error {
-  readonly code = "deployment_not_ready";
   constructor() {
     super("The workflow deployment is still starting.");
   }
 }
 
+/** Interchange refused mail to a run that is stopping, ended or gone, so none was sent. */
+export class RunUnavailableError extends Error {}
+
+/** The host's listener took no mail: it was still starting, stopping or gone, or the hub held no route to it. Nothing was sent. */
+export class ListenerUnroutedError extends NoLiveDeploymentError {
+  constructor(source: RepoSource, address: string, cause?: unknown) {
+    super(TRIAGE_LISTENERS[source]);
+    this.message = cause === undefined ? `listener ${address} is not routable` : `listener ${address} took no mail: ${String(cause)}`;
+    this.cause = cause;
+  }
+}
+
+/** Run states Interchange refuses mail in. */
+const UNAVAILABLE_RUN_CODES = new Set(["workflow_run_stopping", "workflow_run_terminal"]);
+
 /** Interchange's sender gate refused the mail; a worker copied from a listener deployed before the gate keeps refusing until the listener is redeployed. */
 export class SenderRefusedError extends Error {
   readonly code = MAIL_ADMISSION_DENIED_CODE;
 }
 
-/** @corbits/webhooks 0.3.0 ignores `notReady` and routes the mail without grants; this fails it so the caller retries. Drop once the package handles it. */
+/**
+ * @corbits/webhooks 0.3.0 ignores `notReady` and routes the mail without grants; this fails it so the caller retries.
+ * Drop once the package handles it. A run Interchange refuses mail to as stopping, ended or gone fails as
+ * `RunUnavailableError`, since the package reports it only as a plain error.
+ */
 export function readyMaterializer(materialize: RunTriggerMaterialize): RunTriggerMaterialize {
   return async function materializeReady(args) {
     const grants = await materialize(args);
     if (grants.outcome === "notReady") throw new DeploymentNotReadyError();
+    if (grants.outcome === "skip") throw new RunUnavailableError(`${args.agentAddress} is not a live deployment`);
+    if (grants.outcome === "rejected" && UNAVAILABLE_RUN_CODES.has(grants.code)) throw new RunUnavailableError(grants.message);
     return grants;
   };
 }

apps/hub/src/github/github-read-access.postgres.test.ts

@@ -6,11 +6,14 @@ import { HUB_CONDITION_REGISTRY } from "@intx/hub-api";
 import { createEnvKeyCredentialCipher } from "@intx/crypto";
 import { createDB, createGrantStore, dropSchema, runMigrations, schema } from "@intx/db";
 import { credentialAad } from "@intx/types";
-import { TRIAGE_ROLE, triageStopped, type ActionKind } from "@corbits/triage-contracts";
+import { TRIAGE_ROLE, triageStopped, type ActionKind, type PrTriageRow } from "@corbits/triage-contracts";
 import { doEffectId } from "@corbits/triage-workflows/logic/actions";
 import { guardedApp } from "../integrations/test-app.js";
 import { createAuditLog } from "../audit-log.js";
+import { ListenerUnroutedError } from "./deployment.js";
 import { createGithubHeldDos } from "./held-dos.js";
+import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js";
+import type { TriageStateStore } from "./triage-state-store.js";
 import { createGithubOpenPulls, createOpenPullsCache } from "./open-pulls.js";
 import { createTenantOpenPrs } from "./tenant-open-heads.js";
 import { createGithubPrActions } from "./pr-actions.js";
@@ -106,6 +109,24 @@ async function authorizePortal(principalId: string, tenantId: string, resource:
 }
 
 const deps = { db: handle.db, cipher, onKeyMoved: noKeyPush, authorize: authorizePortal, githubApiOrigin };
+/** Triage rows kept in memory, with the version the store would give each write. */
+function triageRows(): TriageStateStore & { rows: PrTriageRow[] } {
+  let version = 0;
+  const kept = {
+    rows: [] as PrTriageRow[],
+    async load() {
+      return { rows: kept.rows, version: version === 0 ? null : { artifactId: "art_state", version } };
+    },
+    async save(_tenantId: string, _repo: string, rows: PrTriageRow[]) {
+      kept.rows = rows;
+      version += 1;
+      return { artifactId: "art_state", version };
+    },
+  };
+  return kept;
+}
+/** The listener takes every approval here, so no head is held. */
+const NOT_HELD = { store: triageRows(), policy: DEFAULT_RECONCILE_POLICY, now: () => new Date(), retry() {} };
 const list = createTenantOpenPrs({ ...deps, tenantBitbucket: async () => undefined });
 const openPulls = createGithubOpenPulls({ ...deps, list, cache: createOpenPullsCache(() => new Date()) });
 const reads = createRepoReads(() => new Date());
@@ -341,6 +362,7 @@ test("PostgreSQL: approving a held Do needs its kind's grant and hands the liste
   let recorded: Array<Pick<VerdictRecord, "runId" | "degraded" | "runStatus">> = [];
   const heldDos = createGithubHeldDos({
     ...deps,
+    ...NOT_HELD,
     runReader: { async readRunEvents() { return events; } },
     verdicts: { async forPull() { return recorded as VerdictRecord[]; } },
     async toListener(_tenantId, payload) {
@@ -376,6 +398,38 @@ test("PostgreSQL: approving a held Do needs its kind's grant and hands the liste
   expect(await approve("usr_approver", ship)).toMatchObject({ status: 503, body: { error: { code: "service_not_running", message: triageStopped("github") } } });
 });
 
+test("PostgreSQL: a held Do approved while the listener is starting answers 202, and its head is queued for the reconciler to mail once the listener is up", async () => {
+  const target = { labels: ["api"] };
+  const tag = { list: "pass" as const, index: 0, kind: "labels" as const, target, held: true, effectId: doEffectId({ repo: REPO, number: 8, headSha: HEAD, list: "pass", index: 0, kind: "labels", target }) };
+  const verdict = { repo: REPO, number: 8, headSha: HEAD, request: { repo: REPO, number: 8, labels: [], owned: [], comment: "" }, actions: [tag] };
+  const events = [{ seq: 1, type: "StepCompleted", body: { type: "StepCompleted", seq: 1, stepId: "evaluate", output: { ref: `inline:${JSON.stringify(verdict)}` } } }];
+  let kicked = 0;
+  const store = triageRows();
+  const heldDos = createGithubHeldDos({
+    ...deps,
+    ...NOT_HELD,
+    store,
+    retry() {
+      kicked += 1;
+    },
+    runReader: { async readRunEvents() { return events; } },
+    verdicts: { async forPull() { return [{ runId: "wfr_gap", degraded: null, runStatus: "completed" }] as VerdictRecord[]; } },
+    async toListener() {
+      throw new ListenerUnroutedError("github", "wfr_listener@tenant.example");
+    },
+    async passRequest() {
+      return undefined;
+    },
+    log() {},
+  });
+  await handle.db.insert(schema.workflowDefinition).values({ id: "wfd_gap", tenantId: TENANT, name: "pr-triage", status: "stopped" });
+  await handle.db.insert(schema.workflowRun).values({ id: "wfr_gap", definitionId: "wfd_gap", anchorRunId: "wfr_gap", tenantId: TENANT, status: "completed" });
+  const res = await heldDos({ runId: "wfr_gap", repo: REPO, number: 8, effectId: tag.effectId, kind: "labels" }, member("usr_approver"));
+  expect({ status: res.status, body: await res.json() }).toEqual({ status: 202, body: { status: "sent" } });
+  expect(kicked).toBe(1);
+  expect(store.rows.map((row) => `${row.number}:${row.headSha}:${row.status}:${row.unrouted === true}`)).toEqual([`8:${HEAD}:failed:true`]);
+});
+
 test("PostgreSQL: a held Do of a verdict a manager reported is approved from the recorded verdict and the mirror request its pass kept, and goes to the listener; once the run's newer verdict replaced it, it is superseded", async () => {
   await handle.db.insert(schema.workflowDefinition).values({ id: "wfd_manager", tenantId: TENANT, name: "triage-manager-github", status: "stopped" });
   await handle.db.insert(schema.workflowRun).values({ id: "wfr_manager", definitionId: "wfd_manager", anchorRunId: "wfr_manager", tenantId: TENANT, status: "running" });
@@ -387,6 +441,7 @@ test("PostgreSQL: a held Do of a verdict a manager reported is approved from the
   const listened: unknown[] = [];
   const heldDos = createGithubHeldDos({
     ...deps,
+    ...NOT_HELD,
     runReader: { async readRunEvents() { return [{ seq: 0, type: "RunStarted", body: { type: "RunStarted", seq: 0 } }]; } },
     verdicts: { async forPull() { return recorded as VerdictRecord[]; } },
     async toListener(_tenantId, payload) {

apps/hub/src/github/held-dos.ts

@@ -4,7 +4,6 @@
 // the pull request's manager writes it at most once under its effect id.
 import { and, eq } from "drizzle-orm";
 import { type } from "arktype";
-import { isRunTriggerUnroutable } from "@corbits/webhooks";
 import { ACTION_KINDS, AGENT_DO, APPROVE_DO_KIND, DoList, repoPolicy, triageStopped, type ActionKind, type ResolvedTarget } from "@corbits/triage-contracts";
 import { schema } from "@intx/db";
 import { formatRunAddress } from "@intx/types";
@@ -16,8 +15,11 @@ import { doRefused, failure, type PortalCredentialDeps } from "./portal-credenti
 import type { Member } from "../integrations/app.js";
 import type { PullRequestRef } from "../manager-state.js";
 import type { VerdictRecordStore } from "./verdict-record-store.js";
-import { DeploymentNotReadyError } from "./deployment.js";
-import { repoRecords, sourceOf, triageNs, type RepoRecord } from "./tenant-config.js";
+import { ListenerUnroutedError } from "./deployment.js";
+import type { ReconcilePolicy } from "./reconcile-plan.js";
+import { holdHead } from "./triage-queue.js";
+import type { TriageStateStore } from "./triage-state-store.js";
+import { repoRecords, sourceOf, triageNs, triageStateKey, type RepoRecord } from "./tenant-config.js";
 import { verdictFor } from "./triage-runs.js";
 
 export const GITHUB_HELD_DOS_PATH = "/api/integrations/github-held-dos";
@@ -45,6 +47,12 @@ export type GithubHeldDosDeps = PortalCredentialDeps & {
   toListener: (tenantId: string, payload: unknown) => Promise<boolean>;
   /** The mirror request a manager's pass kept with the verdict it posted on the head. */
   passRequest: (tenantId: string, pr: PullRequestRef, runId: string, head: string) => Promise<Record<string, unknown> | undefined>;
+  /** Holds the head of an approval its listener took no mail for, which the reconciler mails once one is up. */
+  store: TriageStateStore;
+  policy: ReconcilePolicy;
+  now: () => Date;
+  /** Asks the reconciler for a pass soon. */
+  retry: () => void;
   log: (entry: Record<string, unknown>) => void;
 };
 
@@ -139,10 +147,14 @@ export function createGithubHeldDos(deps: GithubHeldDosDeps) {
     try {
       if (!(await deps.toListener(tenantId, mailed))) return failure(503, "service_not_running", triageStopped(repoSource));
     } catch (err) {
-      deps.log({ level: "error", msg: "held_do_approval_failed", ...logged, error: String(err) });
-      if (err instanceof DeploymentNotReadyError) return failure(409, err.code, err.message);
-      const unroutable = isRunTriggerUnroutable(err);
-      return failure(unroutable ? 503 : 502, unroutable ? "service_not_running" : "hub_unavailable", `Could not send the approval: ${errorText(err)}. Try again.`);
+      if (!(err instanceof ListenerUnroutedError)) {
+        deps.log({ level: "error", msg: "held_do_approval_failed", ...logged, error: String(err) });
+        return failure(502, "hub_unavailable", `Could not send the approval: ${errorText(err)}. Try again.`);
+      }
+      // The approval is recorded; the head goes out with it once a listener is up.
+      await holdHead(deps.store, { tenantId, repo: triageStateKey(record), number, headSha: approval.step.headSha }, err, deps.now(), deps.policy);
+      deps.retry();
+      deps.log({ level: "info", msg: "held_do_approval_held", ...logged, error: String(err) });
     }
     deps.log({ level: "info", msg: "held_do_approved", ...logged });
     return Response.json({ status: "sent", ...(warning && { warning }) }, { status: 202 });