Skip to content

fix(recovery): resolve recovery owners on the caller tx; single-flight the recovery sweep chain #1897

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kkroo merged 4 commits into from
Sep 18, 2026
Merged

fix(recovery): resolve recovery owners on the caller tx; single-flight the recovery sweep chain #1897

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d1b5cfa
fix(recovery): resolve recovery owners on the caller tx; single-fligh…
Sep 16, 2026
52c8d7a
fix(recovery): scope the single-flight latch to the lock-taking tail;…
Sep 17, 2026
a1e8a29
fix(issues): keep the tx-scoped settings read read-only
Sep 17, 2026
2410b79
test(recovery): empty the pooled-under-lock allowlist (Ally Important…
kkroo Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 26 additions & 13 deletions server/src/__tests__/issue-recovery-actions.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1319,19 +1319,16 @@ describeEmbeddedPostgres("issue recovery actions", () => {
}

// Fixed here: the escalation body's own `getCompanyIssuePrefix`, `getAgent`
// and `getLatestIssueRun` reads now run on `tx`. Still pooled under the
// lock, tracked on BLO-34207: owner resolution
// (`resolveStrandedIssueRecoveryOwnerAgentId` -> `getAgent` /
// `isAgentInvokable` / `budgets.getInvocationBlock` / instance settings).
// This ratchet fails on any new pooled call site and on a regression of the
// three fixed ones.
const knownPooledUnderLock = [
"resolveStrandedIssueRecoveryOwnerAgentId",
"isAgentInvokable",
"evaluateAgentInvokabilityFromDb",
"getInvocationBlock",
"getOrCreateRow",
];
// and `getLatestIssueRun` reads now run on `tx`, and so does owner
// resolution (BLO-34207: `resolveStrandedIssueRecoveryOwnerAgentId` ->
// `getAgent` / `isAgentInvokable` / `budgets.getInvocationBlock` /
// instance settings). This ratchet fails on any new pooled call site and
// on a regression of any of them.
// Empty on purpose: nothing may run pooled under the lock any more. Master
// dropped `getLatestIssueRun` (29cdd6ab3) and this PR moved the last one,
// `getOrCreateRow`, onto the caller tx (`readInstanceSettingsOn`); it is
// asserted forbidden by name below, so it must not be allowlisted here.
const knownPooledUnderLock: string[] = [];
const unexpected = pooledInsideTransaction.filter(
(entry) => !knownPooledUnderLock.some((name) => entry.includes(name)),
);
Expand All @@ -1343,6 +1340,22 @@ describeEmbeddedPostgres("issue recovery actions", () => {
// Substring match, so this also covers `getLatestIssueRunForAgentStage` and
// `getLatestIssueRunSince` — none of the three may run pooled under the lock.
expect(pooledInsideTransaction.filter((entry) => entry.includes("getLatestIssueRun"))).toEqual([]);
// BLO-34207. Named individually rather than relying on the `unexpected`
// filter above: an allowlist entry is a substring match, so a future entry
// that happens to contain one of these names would silently re-admit it.
for (const name of [
// BLO-34207: `issuesSvc.update` / `addComment` read instance settings on
// the caller handle now (`instanceSettingsOn`), so the singleton read no
// longer takes a second pool connection under the graph lock.
"getOrCreateRow",
"resolveStrandedIssueRecoveryOwnerAgentId",
"resolveInvokableRecoveryAgentId",
"isAgentInvokable",
"evaluateAgentInvokabilityFromDb",
"getInvocationBlock",
]) {
expect(pooledInsideTransaction.filter((entry) => entry.includes(name))).toEqual([]);
}
const actionRows = await db
.select()
.from(issueRecoveryActions)
Expand Down
32 changes: 32 additions & 0 deletions server/src/__tests__/issues-service.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8896,6 +8896,38 @@ describeEmbeddedPostgres("issueService.create workspace inheritance", () => {
});
});

// BLO-34207: issue mutations read instance settings on the caller's handle so
// they do not take a second pool connection while holding the company graph
// lock. That read must stay a pure read: bootstrapping the singleton row from
// inside the caller's tx takes an instance-wide row lock BEFORE
// `lockIssueParentMutationCompany` and holds it to commit, so two callers can
// take the two locks in opposite orders. Assert the write never happens.
it("does not write the instance settings singleton from an update running on a caller transaction", async () => {
const companyId = randomUUID();
const issueId = randomUUID();

await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Settings read must not write",
status: "todo",
priority: "medium",
});
await db.delete(instanceSettings);

await db.transaction(async (tx) => {
await svc.update(issueId, { title: "Renamed under a caller tx" }, tx);
});

expect(await db.select().from(instanceSettings)).toHaveLength(0);
});

it("returns cycle validation instead of deadlocking intersecting multi-level reparent updates", async () => {
const companyId = randomUUID();
const issueAId = randomUUID();
Expand Down
242 changes: 158 additions & 84 deletions server/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1086,6 +1086,29 @@ export async function startServer(): Promise<StartedServer> {
// crashed run and handing that lock to the retry — exactly the interleaving
// the in-tick `await` was added to remove.
let crashReconcileSweepInFlight = false;
// BLO-34207: the same latch, for the reap → retry-promotion → queued-resume →
// stranded-reconcile → … chain below. That chain is sequential over every
// stranded candidate in the estate (147 distinct issues per pass, measured
// 2026-09-16) and each candidate takes the company-wide issue-graph advisory
// lock. One slow candidate makes the pass outlive the 30 s interval, the next
// tick starts a second pass, and the passes then contend with EACH OTHER on
// that lock: 8 waiters against `POSTGRES_POOL_MAX=10` starved the pool, so
// the holder could not get the second connection it needed to finish and only
// a waiter's 15 s `lock_timeout` broke the cycle. Every pass in the chain is
// idempotent, so a tick that finds one still running skips it.
//
// The latch inherits `crashReconcileSweepInFlight`'s failure mode: a pass that
// HANGS rather than rejects never clears it, and periodic recovery then stops
// estate-wide until restart with no output at all — the symptom is an absence.
// `heartbeatRecoveryChainStartedAt` makes that absence visible. It is not a
// timeout: clearing the latch without cancelling the work would re-admit the
// overlap the latch exists to remove, so this reports and does not act.
let heartbeatRecoveryChainInFlight = false;
let heartbeatRecoveryChainStartedAt = 0;
// 10 ticks. Above the normal case (a sweep routinely outlives one interval —
// that is what the latch is for) and far below the hours a wedged chain would
// otherwise sit silent.
const HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS = 10 * config.heartbeatSchedulerIntervalMs;
const heartbeatSchedulerInFlight = new Set<Promise<void>>();
const trackHeartbeatSchedulerWork = (work: Promise<unknown>) => {
let tracked: Promise<void>;
Expand Down Expand Up @@ -1648,102 +1671,153 @@ export async function startServer(): Promise<StartedServer> {

// Periodically reap orphaned runs (5-min staleness threshold) and make sure
// persisted queued work is still being driven forward.
//
// Deliberately NOT under `heartbeatRecoveryChainInFlight`. These four
// passes are the dispatch path — `resumeQueuedRuns` is what actually
// starts a queued run — and they do not take
// `lockIssueParentMutationCompany`, which is the contention the latch
// exists to remove. Gating them on the latched tail would couple
// dispatch to that tail's slowest pass: it ends in
// `reconcileContendedPrReviewerWakes`, which crosses the
// plugin-worker RPC bridge that logged ~976 x 30 s timeouts in the
// BLO-34207 incident. Resumption would then run once per whole chain
// instead of once per 30 s tick, reproducing the very symptom the
// latch is deployed against ("runs sat in `running` 15-20 min before
// their k8s Job was created").
trackHeartbeatSchedulerWork(heartbeat
.resumeRunningExternalRuntimeRuns()
.then(() => heartbeat.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }))
.then(() => heartbeat.promoteDueScheduledRetries())
.then(async (promotion) => {
await heartbeat.resumeQueuedRuns();
const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
if (
promotion.promoted > 0 ||
reconciled.assignmentDispatched > 0 ||
reconciled.dispatchRequeued > 0 ||
reconciled.continuationRequeued > 0 ||
reconciled.successfulRunHandoffEscalated > 0 ||
reconciled.reviewWaitingParked > 0 ||
reconciled.waitingOnReviewResolved > 0 ||
reconciled.escalated > 0
) {
logger.warn(
{ promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
"periodic heartbeat recovery changed assigned issue state",
);
}
})
.then(async () => {
const reconciled = await heartbeat.reconcileIssueGraphLiveness();
// BLO-29601: auto-resolving a dead escalation is a change to the issue
// graph too. Without it in this gate the drain runs silently and the only
// evidence it happened at all is the cancelled rows themselves.
if (
reconciled.escalationsCreated > 0 ||
reconciled.dependencyWakesHealed > 0 ||
reconciled.staleEscalationsAutoResolved > 0
) {
logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state");
}
})
.then(async () => {
const reconciled = await heartbeat.reconcileTaskWatchdogs();
if (reconciled.triggered > 0) {
logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work");
}
})
.then(async () => {
const scanned = await heartbeat.scanSilentActiveRuns();
if (scanned.created > 0 || scanned.escalated > 0) {
logger.warn({ ...scanned }, "periodic active-run output watchdog created review work");
}
})
.then(async () => {
const reviewed = await heartbeat.reconcileProductivityReviews();
// BLO-30303 AC4: unconditional — see the startup pass above.
logger.info({ ...reviewed }, "periodic productivity reconciliation funnel");
if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) {
logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work");
}
})
.then(async () => {
const swept = await heartbeat.reconcileResolvedBlockerDependents();
if (swept.woken > 0 || swept.failed > 0) {
logger.warn({ ...swept }, "periodic resolved-blocker-dependents sweep enqueued wakes");
}
})
.then(async () => {
const failedWakeDispatches = await heartbeat.reconcileFailedWakeDispatches();
if (failedWakeDispatches.recovered > 0 || failedWakeDispatches.exhausted > 0) {
if (promotion.promoted > 0) {
logger.warn(
{ ...failedWakeDispatches },
"periodic failed-wake-dispatch reconciliation retried durable wake failures (BLO-14395)",
);
}
})
.then(async () => {
// BLO-21995: replay PR-reviewer wakes that lost their PR-scope
// advisory lock at webhook time. GitHub never redelivers a 200,
// so this pass is the only path back for a sanctioned review
// request that lost that race.
const contendedReviewerWakes = await reconcileContendedPrReviewerWakes(db as any, {
webhookSecret: config.githubWebhookSecret || null,
pluginWorkerManager,
heartbeatOptions: { paperclipNodeRole: config.paperclipNodeRole },
prReviewerAgentIds: config.githubPrReviewerAgentIds,
prReviewerBotLogin: config.prReviewerBotLogin || null,
});
if (
contendedReviewerWakes.recovered > 0 ||
contendedReviewerWakes.exhausted > 0
) {
logger.warn(
{ ...contendedReviewerWakes },
"periodic contended PR-reviewer wake reconciliation replayed lock-contended review requests (BLO-21995)",
{ promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds },
"periodic heartbeat dispatch promoted due scheduled retries",
);
}
})
.catch((err) => {
logger.error({ err }, "periodic heartbeat recovery failed");
logger.error({ err }, "periodic heartbeat dispatch resumption failed");
}));

if (heartbeatSchedulerStopped) return;

// The lock-taking tail, single-flighted across ticks. Block form (not
// an early `return`) so a pass appended after it still runs while a
// chain is in flight — at 147 sequential candidates against a 30 s
// tick, "in flight" is the steady state, so an early return would
// make any later sibling silently dead. Matches
// `crashReconcileSweepInFlight` above.
if (!heartbeatRecoveryChainInFlight) {
heartbeatRecoveryChainInFlight = true;
heartbeatRecoveryChainStartedAt = Date.now();
trackHeartbeatSchedulerWork(heartbeat
.reconcileStrandedAssignedIssues()
.then(async (reconciled) => {
if (
reconciled.assignmentDispatched > 0 ||
reconciled.dispatchRequeued > 0 ||
reconciled.continuationRequeued > 0 ||
reconciled.successfulRunHandoffEscalated > 0 ||
reconciled.reviewWaitingParked > 0 ||
reconciled.waitingOnReviewResolved > 0 ||
reconciled.escalated > 0
) {
logger.warn({ ...reconciled }, "periodic heartbeat recovery changed assigned issue state");
}
})
.then(async () => {
const reconciled = await heartbeat.reconcileIssueGraphLiveness();
// BLO-29601: auto-resolving a dead escalation is a change to the issue
// graph too. Without it in this gate the drain runs silently and the only
// evidence it happened at all is the cancelled rows themselves.
if (
reconciled.escalationsCreated > 0 ||
reconciled.dependencyWakesHealed > 0 ||
reconciled.staleEscalationsAutoResolved > 0
) {
logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state");
}
})
.then(async () => {
const reconciled = await heartbeat.reconcileTaskWatchdogs();
if (reconciled.triggered > 0) {
logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work");
}
})
.then(async () => {
const scanned = await heartbeat.scanSilentActiveRuns();
if (scanned.created > 0 || scanned.escalated > 0) {
logger.warn({ ...scanned }, "periodic active-run output watchdog created review work");
}
})
.then(async () => {
const reviewed = await heartbeat.reconcileProductivityReviews();
// BLO-30303 AC4: unconditional — see the startup pass above.
logger.info({ ...reviewed }, "periodic productivity reconciliation funnel");
if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) {
logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work");
}
})
.then(async () => {
const swept = await heartbeat.reconcileResolvedBlockerDependents();
if (swept.woken > 0 || swept.failed > 0) {
logger.warn({ ...swept }, "periodic resolved-blocker-dependents sweep enqueued wakes");
}
})
.then(async () => {
const failedWakeDispatches = await heartbeat.reconcileFailedWakeDispatches();
if (failedWakeDispatches.recovered > 0 || failedWakeDispatches.exhausted > 0) {
logger.warn(
{ ...failedWakeDispatches },
"periodic failed-wake-dispatch reconciliation retried durable wake failures (BLO-14395)",
);
}
})
.then(async () => {
// BLO-21995: replay PR-reviewer wakes that lost their PR-scope
// advisory lock at webhook time. GitHub never redelivers a 200,
// so this pass is the only path back for a sanctioned review
// request that lost that race.
const contendedReviewerWakes = await reconcileContendedPrReviewerWakes(db as any, {
webhookSecret: config.githubWebhookSecret || null,
pluginWorkerManager,
heartbeatOptions: { paperclipNodeRole: config.paperclipNodeRole },
prReviewerAgentIds: config.githubPrReviewerAgentIds,
prReviewerBotLogin: config.prReviewerBotLogin || null,
});
if (
contendedReviewerWakes.recovered > 0 ||
contendedReviewerWakes.exhausted > 0
) {
logger.warn(
{ ...contendedReviewerWakes },
"periodic contended PR-reviewer wake reconciliation replayed lock-contended review requests (BLO-21995)",
);
}
})
.catch((err) => {
logger.error({ err }, "periodic heartbeat recovery failed");
})
.finally(() => {
heartbeatRecoveryChainInFlight = false;
heartbeatRecoveryChainStartedAt = 0;
}));
} else if (
Date.now() - heartbeatRecoveryChainStartedAt >
HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS
) {
// Skipping is normal and silent; a chain that has been in flight for
// many ticks is not. Reported, not acted on — see the latch decl.
logger.warn(
{
inFlightMs: Date.now() - heartbeatRecoveryChainStartedAt,
warnAfterMs: HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS,
},
"periodic heartbeat recovery chain still in flight across many ticks; recovery passes are not running",
);
}
}
})();
}, config.heartbeatSchedulerIntervalMs);
Expand Down
5 changes: 4 additions & 1 deletion server/src/services/agent-invokability.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import { getAgentWorkEligibility, type AgentEligibilityAgent, type AgentOrgChain
import { eq } from "drizzle-orm";

type AgentStatus = (typeof agents.$inferSelect)["status"];
type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];

export type AgentOrgRow = Pick<
typeof agents.$inferSelect,
Expand Down Expand Up @@ -116,7 +117,9 @@ export function evaluateAgentInvokability(
}

export async function evaluateAgentInvokabilityFromDb(
db: Db,
// Callers holding an advisory lock MUST pass their own transaction — a second
// pool connection here is what convoyed on BLO-34207.
db: Db | DbTransaction,
agent: AgentOrgRow | null | undefined,
): Promise<AgentInvokability> {
if (!agent) return evaluateAgentInvokability(agent, []);
Expand Down
Loading
Loading