feat(hub): run every Triage deployment in one local sidecar

.env.example

@@ -25,9 +25,12 @@ HUB_DATA_DIR=./tmp/hub-data
 # The origin sidecars reach the hub REST API at, for workflows that call it. Unset = the origin of BETTER_AUTH_BASE_URL
 # HUB_API_ORIGIN=
 
-# Milliseconds a local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000.
+# Milliseconds the local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000.
 # HUB_SIDECAR_STOP_TIMEOUT_MS=
 
+# Deployments the hub's local sidecar runs at once: the listener, one manager per pull request and the judge. Default 16, at most 1792.
+# TRIAGE_SIDECAR_MAX_DEPLOYMENTS=
+
 # Docker daemon for workflows that require isolation:workload, registered when it answers
 # and holds the image from: docker build --target sidecar -t corbits-triage-sidecar .
 # Unset: the docker CLI's context (DOCKER_CONTEXT, else its current one), else
@@ -48,7 +51,7 @@ HUB_DATA_DIR=./tmp/hub-data
 # Milliseconds between passes that re-queue open pull requests whose head is not triaged. Default 300000.
 # TRIAGE_RECONCILE_INTERVAL_MS=
 
-# Pull request heads the reconciler lets a tenant have queued or running at once, each in a sidecar of its own. Default 5.
+# Pull request heads the reconciler lets a tenant have queued or running at once. Default 5.
 # TRIAGE_MAX_IN_FLIGHT=
 # Milliseconds after a head is capped by runs that never started before it gets a fresh set of attempts. Default 21600000.
 # TRIAGE_CAPPED_RETRY_MS=

apps/hub/src/env.ts

@@ -3,6 +3,7 @@ import { homedir } from "node:os";
 import { type, type ArkError } from "arktype";
 import type { DockerTls, FindDockerHostOpts } from "@corbits/provisioner-docker";
 import { LifecycleDuration, lifecycleDurationMs, type ResolvedWorkflowLifecyclePolicy } from "@intx/types";
+import { DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS, MAX_SIDECAR_INCARNATIONS } from "@intx/types/sidecar";
 import { DEFAULT_GITHUB_API_ORIGIN } from "./github/github-app-credential-adapter.js";
 import type { CoalesceWindow } from "./github/coalescer.js";
 import { DEFAULT_RECONCILE_POLICY, type ReconcilePolicy } from "./github/reconcile-plan.js";
@@ -13,6 +14,10 @@ const secret = type("string").narrow((value, ctx) =>
 const positiveInteger = type("string.integer").narrow((value, ctx) =>
   Number(value) > 0 || ctx.mustBe("a positive integer"));
 
+/** The sidecar holds its active and kept deployments within Interchange's combined limit; the hub leaves it the default kept ones. */
+const MAX_LOCAL_SIDECAR_DEPLOYMENTS = MAX_SIDECAR_INCARNATIONS - DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS;
+const DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS = 16;
+
 const lifecycleDuration = type("string.trim").pipe(LifecycleDuration);
 
 const positiveDuration = lifecycleDuration.narrow((value, ctx) =>
@@ -75,6 +80,8 @@ const HubEnvSchema = type({
   "TRIAGE_MAX_IN_FLIGHT?": positiveInteger,
   "TRIAGE_CAPPED_RETRY_MS?": positiveInteger,
   "TRIAGE_COALESCE_QUIET_MS?": positiveInteger,
+  "TRIAGE_SIDECAR_MAX_DEPLOYMENTS?": positiveInteger.narrow((value, ctx) =>
+    Number(value) <= MAX_LOCAL_SIDECAR_DEPLOYMENTS || ctx.mustBe(`at most ${String(MAX_LOCAL_SIDECAR_DEPLOYMENTS)}`)),
   "HUB_AGENT_GC_PACK_THRESHOLD?": positiveInteger,
   "HUB_AGENT_GC_LOOSE_THRESHOLD?": positiveInteger,
   "HUB_AGENT_GC_WARN_BYTES?": positiveInteger,
@@ -267,7 +274,7 @@ export function triageReconcileIntervalMs(env: HubEnv): number {
   return env.TRIAGE_RECONCILE_INTERVAL_MS === undefined ? 5 * 60_000 : Number(env.TRIAGE_RECONCILE_INTERVAL_MS);
 }
 
-/** Heads the reconciler lets a tenant have queued or running at once; each runs in a worker sidecar of its own. */
+/** Heads the reconciler lets a tenant have queued or running at once. */
 export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" | "TRIAGE_CAPPED_RETRY_MS">): ReconcilePolicy {
   return {
     ...DEFAULT_RECONCILE_POLICY,
@@ -276,6 +283,11 @@ export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" |
   };
 }
 
+/** Deployments the hub's local sidecar runs at once. */
+export function localSidecarMaxDeployments(env: Pick<HubEnv, "TRIAGE_SIDECAR_MAX_DEPLOYMENTS">): number {
+  return numberOr(env.TRIAGE_SIDECAR_MAX_DEPLOYMENTS, DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS);
+}
+
 const COALESCE_BURST_MS = 3_000;
 
 /**

apps/hub/src/interchange-hub.ts

@@ -47,6 +47,7 @@ import {
   resolveSenderKey,
   resolveSenderPrincipal,
   withExecutableWorkflowRun,
+  type DB,
 } from "@intx/db";
 import { createEnvKeyCredentialCipher } from "@intx/crypto";
 import {
@@ -112,12 +113,10 @@ export type CreateHubServerOpts = {
   readonly credentialCipher: CredentialCipher;
   /** Provisioners eligible to host frozen workflow deployments. */
   readonly sidecarProvisioners?: readonly SidecarProvisioner[];
-  /** Selects among matching deployment provisioners. Defaults to the first. */
-  readonly sidecarProvisionerChooser?: SidecarProvisionerChooser;
+  /** Selects among matching deployment provisioners, reading the hub's database. Defaults to the first. */
+  readonly sidecarProvisionerChooser?: (db: DB["db"]) => SidecarProvisionerChooser;
   /** Provisioners eligible to evaluate workflow source code. */
   readonly probeSidecarProvisioners?: readonly SidecarProvisioner[];
-  /** Selects among matching probe provisioners. Defaults to the first. */
-  readonly probeSidecarProvisionerChooser?: SidecarProvisionerChooser;
   readonly probeSidecarCapabilityRules?: readonly SidecarCapabilityRule[];
   /** Maximum simultaneous allocation reconciliations. Defaults to eight. */
   readonly sidecarAllocationConcurrency?: number;
@@ -138,7 +137,6 @@ export async function createInterchangeHub({
   sidecarProvisioners = [],
   sidecarProvisionerChooser,
   probeSidecarProvisioners = [],
-  probeSidecarProvisionerChooser,
   probeSidecarCapabilityRules = [],
   sidecarAllocationConcurrency = DEFAULT_SIDECAR_ALLOCATION_CONCURRENCY,
   sidecarOperationTimeoutMs,
@@ -355,14 +353,11 @@ export async function createInterchangeHub({
   const sidecarPlugins = createSidecarPluginRegistry({
     provisioners: sidecarProvisioners,
     ...(sidecarProvisionerChooser !== undefined
-      ? { chooser: sidecarProvisionerChooser }
+      ? { chooser: sidecarProvisionerChooser(db) }
       : {}),
   });
   const probeSidecarPlugins = createSidecarPluginRegistry({
     provisioners: probeSidecarProvisioners,
-    ...(probeSidecarProvisionerChooser !== undefined
-      ? { chooser: probeSidecarProvisionerChooser }
-      : {}),
   });
   const workflowAllocationService = createWorkflowAllocationService({
     db,

apps/hub/src/manager-state.postgres.test.ts

@@ -4,7 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto";
 import { sql } from "drizzle-orm";
 import { createDB, runMigrations, schema } from "@intx/db";
 import type { Tx } from "./github/admin-github-grant.js";
-import { done, liveListener, liveManager, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js";
+import { done, endNeverRan, liveListener, liveManager, notePlaced, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js";
 import { migrateTriageTables } from "./triage-db.js";
 
 const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -411,7 +411,7 @@ test("PostgreSQL: a manager deployed without a sidecar past the startup window d
       cancellationReason: schema.workflowRun.cancellationReason,
     }).from(schema.workflowRun).where(sql`${schema.workflowRun.id} = 'run_stuck'`);
     expect(left?.cancellationRequestedAt).toBeInstanceOf(Date);
-    expect(left?.cancellationReason).toBe("manager deployed without a sidecar");
+    expect(left?.cancellationReason).toBe("manager deployed without a sidecar or never ran");
   });
 }, 120_000);
 
@@ -464,6 +464,49 @@ test("PostgreSQL: a manager being cancelled is mailed nothing and takes nothing,
   });
 }, 120_000);
 
+async function allocate(db: Db, runId: string, fields: { status: "allocated"; sidecarId: string }): Promise<void> {
+  await db.insert(schema.sidecar).values({ id: fields.sidecarId, tokenHashSha256: new Uint8Array(32).fill(runId.length), provisionerId: "local-process", provisionerApiVersion: 2, provisionerBindingFingerprint: "fp", adapterManifestDigest: "digest" }).onConflictDoNothing();
+  await db.insert(schema.sidecarAllocation).values({ id: `alc_${runId}`, anchorRunId: runId, tenantId: TENANT, provisionerId: "local-process", provisionerApiVersion: 2, provisionerBindingFingerprint: "fp", adapterManifestDigest: "digest", maxDisconnectedMs: 60_000, ...fields });
+}
+
+test("PostgreSQL: a manager on its sidecar five minutes without running gives up the pull request, and the reaper ends it but never the listener", async () => {
+  await withDatabase(async function neverRan(db) {
+    await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" });
+    await db.insert(schema.workflowRun).values({ id: "run_listener", anchorRunId: "run_listener", definitionId: "wfd_listener", tenantId: TENANT, status: "deployed", createdAt: new Date(Date.now() - 60 * MINUTE) });
+    await allocate(db, "run_listener", { status: "allocated", sidecarId: "sdc_local" });
+    await deployedRun(db, "run_quiet", new Date(Date.now() - 6 * MINUTE));
+    await allocate(db, "run_quiet", { status: "allocated", sidecarId: "sdc_local" });
+    notePlaced({ allocationId: "alc_run_quiet", generation: 0 }, new Date(Date.now() - 6 * MINUTE));
+    await claimRun(db, "run_quiet");
+    const cancelled: string[][] = [];
+    const logs: Record<string, unknown>[] = [];
+
+    await endNeverRan(db, new Date(), async (...args) => cancelled.push(args), (entry) => logs.push(entry));
+
+    expect(cancelled).toEqual([[TENANT, "run_quiet", "triage-manager-github deployed but never ran"]]);
+    expect(logs).toEqual([{ level: "warn", msg: "triage_deployment_never_ran", runId: "run_quiet", workflow: "triage-manager-github" }]);
+    expect(await liveListener(db, TENANT, "github")).toMatchObject({ runId: "run_listener" });
+    await liveRun(db, "run_next");
+    expect(await take(db, TENANT, GITHUB, "run_next")).toMatchObject({ outcome: "idle" });
+  });
+}, 120_000);
+
+test("PostgreSQL: a manager that waited past five minutes for a slot is timed from its placement, not its deploy", async () => {
+  await withDatabase(async function placedLate(db) {
+    await deployedRun(db, "run_late", new Date(Date.now() - 10 * MINUTE));
+    await allocate(db, "run_late", { status: "allocated", sidecarId: "sdc_local" });
+    notePlaced({ allocationId: "alc_run_late", generation: 0 }, new Date(Date.now() - MINUTE));
+    await claimRun(db, "run_late");
+    const cancelled: string[][] = [];
+
+    await endNeverRan(db, new Date(), async (...args) => cancelled.push(args), () => undefined);
+
+    expect(cancelled).toEqual([]);
+    await liveRun(db, "run_next");
+    expect(await take(db, TENANT, GITHUB, "run_next")).toEqual({ outcome: "refused" });
+  });
+}, 120_000);
+
 test("PostgreSQL: a listener counts as live as a manager does, and one deployed without a sidecar past the startup window is released", async () => {
   await withDatabase(async function listener(db) {
     await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" });