Skip to content

Commit 950ffe8

Browse files
committed
refactor(listener): let stock's mail hold replace the manager ready window
The listener mails a just-deployed manager at once. Interchange's hub-sessions sidecar handler parks mail to a run whose deploy is not recorded, redrives it when the deploy records the run's key, and drops it after HUB_MAIL_HOLD_TTL_MS. The listener's ready window, wake poll and its unwoken, retry and unwokenAgain steps are gone, as are the listener and manager waits #713 published in release.json. HUB_MAIL_HOLD_TTL_MS keeps its fifteen-minute default and is the one source for the hub's start grace, the time a manager may sit on its sidecar unstarted and the hold on a new deploy ask; a queued head waits the hold plus one reconcile interval before it is queued again, and a manager start that can no longer run is asked again on the next event.
1 parent 8f6e849 commit 950ffe8

33 files changed

Lines changed: 343 additions & 372 deletions

‎.env.example‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -52,20 +52,12 @@ HUB_DATA_DIR=./tmp/hub-data
5252
# TRIAGE_RECONCILE_INTERVAL_MS=
5353
# Milliseconds before a pass that could not reach a listener tries again, doubling to two minutes. Default 15000.
5454
# TRIAGE_RECONCILE_RETRY_MS=
55-
# Milliseconds a listener waits for a just-deployed manager to reach a sidecar before it gives up on it. Default 120000.
56-
# TRIAGE_MANAGER_READY_MS=
57-
# Milliseconds a just-deployed manager may have no sidecar and still hold its pull request. Default 120000.
58-
# TRIAGE_MANAGER_START_GRACE_MS=
59-
# Milliseconds a manager waiting on CI or a reviewer waits between checks. Default 30000.
60-
# TRIAGE_MANAGER_POLL_MS=
61-
# Milliseconds between a listener's checks on whether its manager reached a sidecar. Default 2000.
62-
# TRIAGE_WAKE_POLL_MS=
6355

6456
# Pull request heads the reconciler lets a tenant have queued or running at once. Default 5.
6557
# TRIAGE_MAX_IN_FLIGHT=
6658
# Milliseconds after a head is capped by runs that never started before it gets a fresh set of attempts. Default 21600000.
6759
# TRIAGE_CAPPED_RETRY_MS=
68-
# Milliseconds without a new webhook event before a pull request's events go out as one triage run; at most four times this after the first. Default 30000.
60+
# Milliseconds without a new webhook event before a pull request's events go out as one triage run; at most four times this after the first, which must stay under HUB_MAIL_HOLD_TTL_MS plus TRIAGE_RECONCILE_INTERVAL_MS. Default 30000.
6961
# TRIAGE_COALESCE_QUIET_MS=
7062
# Milliseconds a pull request may wait for Triage to start before the hub logs triage_stalled, shows it in the portal and emails the workspace's owners and admins. Default 300000.
7163
# TRIAGE_STALLED_AFTER_MS=
@@ -75,6 +67,8 @@ HUB_DATA_DIR=./tmp/hub-data
7567
# Interchange settings. Unset keeps the default shown: Interchange's stock default, except
7668
# HUB_MAIL_HOLD_TTL_MS, HUB_MAIL_ACK_RETRY_INTERVAL_MS and HUB_MAIL_ACK_MAX_RETRIES,
7769
# which the hub raises so routed triage mail survives a long pack receive.
70+
# HUB_MAIL_HOLD_TTL_MS is also how long a just-deployed manager that has not come up keeps
71+
# its pull request; a queued head may wait that plus TRIAGE_RECONCILE_INTERVAL_MS to start.
7872
# Durations are whole numbers ending in s, m, h or d; the max lifetime and max disconnected
7973
# must be above zero.
8074
# HUB_MAX_TARBALL_BYTES=10485760

‎.github/workflows/ci.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ jobs:
9292
matrix:
9393
include:
9494
- shard: 1/5
95-
files: tests/chain/unwoken.test.ts
95+
files: tests/chain/unstarted.test.ts
9696
- shard: 2/5
9797
files: tests/chain/woken.test.ts tests/chain/local-sidecars.test.ts tests/chain/deterministic.test.ts tests/chain/unset.test.ts
9898
- shard: 3/5

‎apps/hub/src/env.ts‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@ import type { DockerTls, FindDockerHostOpts } from "@corbits/provisioner-docker"
55
import { LifecycleDuration, lifecycleDurationMs, type ResolvedWorkflowLifecyclePolicy } from "@intx/types";
66
import { DEFAULT_GITHUB_API_ORIGIN } from "./github/github-app-credential-adapter.js";
77
import type { CoalesceWindow } from "./github/coalescer.js";
8-
import { DEFAULT_RECONCILE_POLICY, type ReconcilePolicy } from "./github/reconcile-plan.js";
8+
import { DEFAULT_RECONCILE_POLICY, unstartedAfterMs, type ReconcilePolicy } from "./github/reconcile-plan.js";
9+
import { DEFAULT_MAIL_HOLD_MS } from "./timings.js";
910

1011
const secret = type("string").narrow((value, ctx) =>
1112
/^[0-9a-fA-F]{64}$/.test(value) || ctx.mustBe("32 bytes encoded as 64 hexadecimal characters"));
@@ -73,10 +74,6 @@ const HubEnvSchema = type({
7374
"RAILWAY_GIT_REPO_OWNER?": "string",
7475
"RAILWAY_GIT_REPO_NAME?": "string",
7576
"RAILWAY_ENVIRONMENT_ID?": "string",
76-
"TRIAGE_MANAGER_READY_MS?": positiveInteger,
77-
"TRIAGE_MANAGER_START_GRACE_MS?": positiveInteger,
78-
"TRIAGE_MANAGER_POLL_MS?": positiveInteger,
79-
"TRIAGE_WAKE_POLL_MS?": positiveInteger,
8077
"TRIAGE_RECONCILE_INTERVAL_MS?": positiveInteger,
8178
"TRIAGE_RECONCILE_RETRY_MS?": positiveInteger,
8279
"TRIAGE_COALESCE_QUIET_MS?": positiveInteger,
@@ -273,10 +270,11 @@ export function migrationEnv(db: DatabaseConfig): Record<string, string> {
273270
};
274271
}
275272

276-
/** Heads the reconciler lets a tenant have queued or running at once. */
277-
export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" | "TRIAGE_CAPPED_RETRY_MS">): ReconcilePolicy {
273+
/** Heads the reconciler lets a tenant have queued or running at once, and how long a queued head may wait to start. */
274+
export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" | "TRIAGE_CAPPED_RETRY_MS">, { mailHoldMs, reconcileIntervalMs }: { mailHoldMs: number; reconcileIntervalMs: number }): ReconcilePolicy {
278275
return {
279276
...DEFAULT_RECONCILE_POLICY,
277+
unstartedAfterMs: unstartedAfterMs(mailHoldMs, reconcileIntervalMs),
280278
maxInFlight: numberOr(env.TRIAGE_MAX_IN_FLIGHT, DEFAULT_RECONCILE_POLICY.maxInFlight),
281279
cappedRetryAfterMs: numberOr(env.TRIAGE_CAPPED_RETRY_MS, DEFAULT_RECONCILE_POLICY.cappedRetryAfterMs),
282280
};
@@ -386,9 +384,10 @@ export function interchangeSettings(env: HubEnv): InterchangeSettings {
386384
...(env.HUB_API_ORIGIN !== undefined && { hubApiOrigin: env.HUB_API_ORIGIN }),
387385
// Triage mail is routed, not dispatched, so only the router's ack retries keep it alive. A
388386
// pack receive holds the connection queue for minutes and the stock 5 x 10 s give-up drops it.
387+
// The mail hold is the hub's start grace for a just-deployed manager too, and how long a queued head may wait to start.
389388
sidecarLink: {
390389
...(env.HUB_SIDECAR_REQUEST_TIMEOUT_MS !== undefined && { requestTimeoutMs: Number(env.HUB_SIDECAR_REQUEST_TIMEOUT_MS) }),
391-
mailHoldTTLMs: numberOr(env.HUB_MAIL_HOLD_TTL_MS, 15 * 60_000),
390+
mailHoldTTLMs: numberOr(env.HUB_MAIL_HOLD_TTL_MS, DEFAULT_MAIL_HOLD_MS),
392391
mailAckRetryIntervalMs: numberOr(env.HUB_MAIL_ACK_RETRY_INTERVAL_MS, 30_000),
393392
mailAckMaxRetries: numberOr(env.HUB_MAIL_ACK_MAX_RETRIES, 10),
394393
},

‎apps/hub/src/github/reconcile-plan.test.ts‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@ import { DEFAULT_RECONCILE_POLICY, planTenant, pullKey, requeue, runKey, type Ob
55
const MINUTE = 60_000;
66
const T0 = new Date("2026-10-07T00:00:00.000Z");
77
const policy = DEFAULT_RECONCILE_POLICY;
8+
/** Minutes a queued head may wait to start. */
9+
const UNSTARTED = policy.unstartedAfterMs / MINUTE;
810

911
function at(minutes: number): Date {
1012
return new Date(T0.getTime() + minutes * MINUTE);
@@ -77,8 +79,8 @@ describe("reconcile plan", () => {
7779

7880
test("a queued mail lost before any run started is queued again", () => {
7981
const queued = plan([pr(1)], [], at(0)).rows;
80-
expect(enqueued(plan([pr(1)], queued, at(9)))).toEqual([]);
81-
expect(enqueued(plan([pr(1)], queued, at(11)))).toEqual([1]);
82+
expect(enqueued(plan([pr(1)], queued, at(UNSTARTED - 1)))).toEqual([]);
83+
expect(enqueued(plan([pr(1)], queued, at(UNSTARTED + 1)))).toEqual([1]);
8284
});
8385

8486
test("failed runs retry with backoff and stop at the attempt cap", () => {
@@ -160,7 +162,7 @@ describe("reconcile plan", () => {
160162
const queued = plan([pr(1)], [], at(0)).rows;
161163
const result = plan([pr(1)], queued, at(3), [[pr(1), [{ ...run("run_a", "failed", 0), startedAt: at(-0.5).toISOString() }]]]);
162164
expect(status(result, 1)).toMatchObject({ status: "failed", runId: "run_a", attempts: 1, error: "run failed" });
163-
const before = plan([pr(1)], queued, at(11), [[pr(1), [run("run_old", "failed", -5)]]]);
165+
const before = plan([pr(1)], queued, at(UNSTARTED + 1), [[pr(1), [run("run_old", "failed", -5)]]]);
164166
expect(status(before, 1)).toMatchObject({ status: "queued", attempts: 2 });
165167
});
166168

@@ -295,8 +297,8 @@ describe("a head whose manager was refused for capacity", () => {
295297
const placed = plan([pr(1)], waiting, at(31));
296298
expect(enqueued(placed)).toEqual([]);
297299
expect(status(placed, 1)).toEqual(expect.not.objectContaining({ error: expect.anything() }));
298-
expect(status(plan([pr(1)], placed.rows, at(41)), 1)).toMatchObject({ status: "queued", attempts: 1 });
299-
expect(status(plan([pr(1)], placed.rows, at(42)), 1)).toMatchObject({ status: "queued", attempts: 2 });
300+
expect(status(plan([pr(1)], placed.rows, at(31 + UNSTARTED)), 1)).toMatchObject({ status: "queued", attempts: 1 });
301+
expect(status(plan([pr(1)], placed.rows, at(32 + UNSTARTED)), 1)).toMatchObject({ status: "queued", attempts: 2 });
300302
});
301303
});
302304

‎apps/hub/src/github/reconcile-plan.ts‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
// Decides, from what the host and the managers' runs say now, which open pull
22
// requests to queue again. Pure, so every recovery path is decided in one place.
3-
import { PR_TRIAGE_STUCK_RUN_MS, WAITING_FOR_CAPACITY, type AwaitedStage, type PrTriageRow, type RepoSource } from "@corbits/triage-contracts";
3+
import { DEFAULT_TIMINGS, PR_TRIAGE_STUCK_RUN_MS, WAITING_FOR_CAPACITY, type AwaitedStage, type PrTriageRow, type RepoSource } from "@corbits/triage-contracts";
44
import { MODEL_NOT_ASKED } from "@corbits/triage-workflows/logic/settled";
5+
import { DEFAULT_MAIL_HOLD_MS } from "../timings.js";
56

67
/** `labels` only where the host lists them. */
78
export type OpenPr = { number: number; headSha: string; updatedAt: string; draft: boolean; labels?: string[] };
@@ -51,9 +52,18 @@ export type ReconcilePolicy = {
5152

5253
const MINUTE = 60_000;
5354

55+
/**
56+
* A queued head counts as never started one reconcile pass after stock's mail hold, `mailHoldMs`, has dropped its
57+
* manager's wake: by then the hub's start grace and deploy-ask hold, both the hold, have lapsed for a manager deployed
58+
* after the head was queued.
59+
*/
60+
export function unstartedAfterMs(mailHoldMs: number, reconcileIntervalMs: number): number {
61+
return mailHoldMs + reconcileIntervalMs;
62+
}
63+
5464
export const DEFAULT_RECONCILE_POLICY: ReconcilePolicy = {
5565
webhookGraceMs: 2 * MINUTE,
56-
unstartedAfterMs: 10 * MINUTE,
66+
unstartedAfterMs: unstartedAfterMs(DEFAULT_MAIL_HOLD_MS, DEFAULT_TIMINGS.reconcileIntervalMs),
5767
stuckAfterMs: PR_TRIAGE_STUCK_RUN_MS,
5868
clockSkewMs: MINUTE,
5969
backoffBaseMs: 5 * MINUTE,

‎apps/hub/src/github/triage-reconciler.test.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
import { describe, expect, test } from "bun:test";
22
import { formatRunAddress } from "@intx/types";
3-
import { emptyPack, type ActionKind, type CheckPack, type PrTriageRow } from "@corbits/triage-contracts";
3+
import { DEFAULT_TIMINGS, emptyPack, type ActionKind, type CheckPack, type PrTriageRow } from "@corbits/triage-contracts";
44
import type { CheckRun } from "@corbits/github-reads";
55
import { unsettledBy } from "@corbits/triage-workflows/logic/settled";
66
import { DEFAULT_RECONCILE_POLICY, planTenant, runKey, type ObservedRun, type OpenPr } from "./reconcile-plan.js";
77
import { triageReconcilePolicy } from "../env.js";
8+
import { DEFAULT_MAIL_HOLD_MS } from "../timings.js";
89
import { TriageStateConflictError, type TriageStateStore } from "./triage-state-store.js";
910
import { ListenerUnroutedError, type LiveDeployment } from "./deployment.js";
1011
import type { HostedVerdict } from "./verdict-record-store.js";
@@ -154,7 +155,7 @@ function harness(runs: Runs | (() => Runs), prs: OpenPr[] | Record<string, OpenP
154155
delivered.push({ payload: payload as Record<string, unknown> });
155156
},
156157
unroutedTries,
157-
policy: triageReconcilePolicy({ ...(options.maxInFlight !== undefined && { TRIAGE_MAX_IN_FLIGHT: options.maxInFlight }) }),
158+
policy: triageReconcilePolicy({ ...(options.maxInFlight !== undefined && { TRIAGE_MAX_IN_FLIGHT: options.maxInFlight }) }, { mailHoldMs: DEFAULT_MAIL_HOLD_MS, reconcileIntervalMs: DEFAULT_TIMINGS.reconcileIntervalMs }),
158159
now,
159160
log: (entry) => logged.push(entry),
160161
});

‎apps/hub/src/interchange-hub.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,8 @@
1212
// the hub, so an ended run's sidecar has nothing left to keep. `sidecarLink`
1313
// is spread into createSidecarRouter with longer mail hold and ack retries
1414
// than the router's defaults: triage mail is routed, not dispatched, so only
15-
// those retries keep it alive while a pack receive holds the connection.
15+
// those retries keep it alive while a pack receive holds the connection. The
16+
// hub's start grace for a just-deployed manager is that same mail hold.
1617
// - `adapterManifest` replaces stock's `sidecarAdapterManifest` and has no
1718
// SIDECAR_ADAPTER_MANIFEST fallback: the hub's adapters are fixed in
1819
// sidecar-config.ts, not operator configuration.

‎apps/hub/src/manager-intake.postgres.test.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import { createPauseGate } from "./github/pr-workflow.js";
1010
import { createManagerIntake, createManagerWake, createOverdueWake, createSupersede } from "./manager-intake.js";
1111
import { ListenerUnroutedError, RunUnavailableError } from "./github/deployment.js";
1212
import { recordEvent } from "./manager-state.js";
13+
import { timings } from "./timings.js";
1314
import { migrateTriageTables } from "./triage-db.js";
1415

1516
const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -282,7 +283,7 @@ test("PostgreSQL: the reconcile pass wakes a manager for a pull request whose re
282283
expect(logged).toEqual([{ level: "info", msg: "manager_woken_overdue", tenantId: TENANT, source: "github", repo: RECORD.name, number: 7, head: "h1" }]);
283284

284285
// A manager the wake started took the claim and went away before done; past the start grace the next pass wakes again.
285-
const earlier = new Date(Date.now() - 3 * 60_000);
286+
const earlier = new Date(Date.now() - timings().mailHoldMs - 60_000);
286287
await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || ${JSON.stringify({ overdueWokenAt: earlier.toISOString(), deployAskedAt: earlier.toISOString(), claimedAt: new Date(earlier.getTime() + 1_000).toISOString() })}::jsonb WHERE key = ${`mgr:github:${RECORD.name}#7`}`);
287288
await wakeOverdue(TENANT);
288289
expect(mailed).toEqual([woken, woken]);
@@ -332,7 +333,7 @@ test("PostgreSQL: an overdue wake is recorded only once it succeeded, and one pu
332333

333334
test("PostgreSQL: a manager deployed without a sidecar past the startup window is not mailed, and the listener is", async () => {
334335
await withDatabase(async function stuck(db) {
335-
await claimManager(db, "run_stuck", 7, "deployed", new Date(Date.now() - 3 * 60_000));
336+
await claimManager(db, "run_stuck", 7, "deployed", new Date(Date.now() - timings().mailHoldMs - 60_000));
336337
const { toListener, mailed } = stack(db, LISTENER);
337338
expect(await toListener(TENANT, requeued(7, "h1"))).toBe(true);
338339
expect(mailed).toEqual([{ address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } }]);

‎apps/hub/src/manager-routes.postgres.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import { MAX_BODY_BYTES } from "./read-json.js";
1616
import { askDeploy, managedPulls, passRequest, recordEvent, recordRefused, type PullRequestRef } from "./manager-state.js";
1717
import { createVerdictRecordStore } from "./github/verdict-record-store.js";
1818
import { mountTenantRoutes } from "./tenant-routes.js";
19+
import { timings } from "./timings.js";
1920
import { migrateTriageTables } from "./triage-db.js";
2021

2122
const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -243,7 +244,7 @@ test("PostgreSQL: a result whose wake fails stays recorded with nothing else, th
243244
const wakeOverdue = createOverdueWake({ db, wake: waker(woken), log });
244245
await wakeOverdue(TENANT);
245246
expect(woken).toEqual([]);
246-
const earlier = new Date(Date.now() - 2 * 60_000 - 1_000).toISOString();
247+
const earlier = new Date(Date.now() - timings().mailHoldMs - 1_000).toISOString();
247248
await db.execute(sql`UPDATE corbits_triage.manager_state SET state = jsonb_set(state, '{receivedAt}', to_jsonb(${earlier}::text)) WHERE key LIKE '%:run_reviewer'`);
248249
await wakeOverdue(TENANT);
249250
expect(woken).toEqual([{ tenantId: TENANT, pr: GITHUB, head: "h1" }]);

0 commit comments

Comments
 (0)