mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-20 06:02:18 +02:00
perf(server): skip settled committed finalizations in unscoped passes (#16485)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - On restart, `/api/health` reports `starting` until startup recovery finishes, and deploys wait for it > - Native recovery and the orphan reap each run an unscoped native finalization reconciliation pass, and readiness waits for both > - The pass selects every committed finalization with a result, including runs that ended long ago, and spends dozens of sequential queries on each one > - On a busy instance this is about one minute per pass, so about two minutes of every restart, and the periodic recovery repeats it on every tick > - This pull request skips settled committed finalizations in unscoped passes, and keeps a small rotating upkeep page so their upkeep still runs > - The benefit is faster restarts and less steady database load, with no change for unsettled work ## Linked Issues or Issue Description Related: #16330 and #16329 (startup and reconciliation timing). **What existing behavior does this improve?** Startup recovery time and steady-state recovery load. **Current behavior** - The unscoped candidate scan in `reconcileNativeFinalizations` returns every native finalization with a result and phase other than `terminal_failure`. Nothing excludes rows that are already settled. - Each committed row then takes the full per-row path: pending-effect, assessment, evidence and decision reads, presentation repair (twice), the workspace barrier read, a `finalizeNativeRun` re-read of the committed coordinator, and workspace settlement (lease release check and agent status update). That's about 40–60 sequential round trips per row. - Startup runs the unscoped pass twice (native recovery, then the orphan reap), and readiness waits for both. The periodic orphan reap runs it again on every tick. - On one instance, all 268 candidates were committed finalizations whose runs had ended. Native recovery took 50–80 s and the orphan reap 47–80 s on every restart, with database round trips of about 3.4 ms. **Proposed behavior** An unscoped pass skips a committed finalization when: - its run is `succeeded`, `failed`, `cancelled` or `timed_out`, and - the run finished more than an hour ago, and - its decision has no `pending` or `failed` status-decision effect. Each unscoped pass also includes up to 10 such settled rows in a rotating order, so the presentation and evidence upkeep still reaches every row over time. Scoped passes (explicit run IDs) are unchanged. **Reason and benefit** These settled rows have nothing left to finish. Recent rows, unfinished runs and rows with undelivered effects stay in every pass, so crash-window repairs and effect delivery behave as before. Startup and periodic passes no longer scale with the full history of committed runs. **Breaking changes** None. A late repair on a settled row can be delayed until its turn in the upkeep rotation instead of the next pass. ## What Changed - `server/src/services/native-runtime/native-finalization-reconciler.ts`: - `settledCommittedFinalization(now)`: the settled predicate above, including a `NOT EXISTS` on undelivered `status_decision_effects` for the same company, issue and decision. - The unscoped candidate scan adds `NOT settled OR run in the upkeep page`. - `nextSettledCommittedUpkeep`: up to `NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS` (10) settled run IDs after an in-process cursor, ordered by run ID, wrapping at the end. - New constants `NATIVE_SETTLED_COMMITTED_GRACE_MS` (1 hour) and `NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS`, plus `resetSettledCommittedUpkeepForTests`. - New `server/src/__tests__/native-finalization-settled-candidates.test.ts`. - `doc/execution-semantics.md`: describes the settled rule and the upkeep rotation. ## Verification - `pnpm exec vitest run src/__tests__/native-finalization-settled-candidates.test.ts` (in `server/`): 4 passed. The test mocks the timing hook to capture the candidate count and stop the pass after the scan: - 25 settled rows give 10, 10, 5, then 10 again over four passes (rotation and wrap). - A row finished 10 minutes ago, a running row, and a terminal row without `finished_at` are all candidates. - A settled row whose decision has a `failed` effect is a candidate. - A scoped pass with an explicit run ID still selects a settled row. - `pnpm exec vitest run` for `native-finalization-recovery`, `legacy-finalization-regression`, `native-run-finalizer-telemetry`, `native-workspace-finalizer`, `native-finalization-reconciler-timing`, `external-chat-wait.integration` and `native-status-arbiter-corpus`: 316 passed. - `heartbeat/recovery.test.ts`: the 8 tests that do not need a database passed. The 20 database tests were skipped locally because embedded PostgreSQL could not start (the host's shared-memory segments were exhausted by other processes). CI runs them. - `pnpm typecheck` in `server/`: passed. ## Risks - A settled row that needs a late repair (for example a missing chat presentation) waits for its turn in the upkeep rotation, up to `ceil(settled rows / 10)` passes, instead of the next pass. - The upkeep cursor is in memory. After a restart the rotation starts from the beginning; the work is idempotent. - A run left `running` with a committed finalization is never considered settled, so it is still projected before the reaper's active-run loop. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code with tool use (shell, file edits, test runs, database and log analysis). A separate Claude agent mapped the recovery call tree. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
c3ffa3f0dd
commit
c7ea5c0216
3 files changed
+250
-1
No files matched your search
@@ -0,0 +1,175 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { sql } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
statusDecisionEffects,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
|
||||
// The pass records its candidate count on its timing object, then starts the
|
||||
// per-row loop with phase("candidates"). The mock captures the count and stops
|
||||
// the pass there, so these tests check candidate selection only.
|
||||
const captured = vi.hoisted(() => ({ counts: [] as number[] }));
|
||||
class StopAfterCandidateScan extends Error {}
|
||||
vi.mock("../services/native-runtime/native-reconciliation-timing.js", () => ({
|
||||
createNativeReconciliationPassTiming: () => ({
|
||||
phase: (name: string) => {
|
||||
if (name === "candidates") throw new StopAfterCandidateScan();
|
||||
},
|
||||
candidates: (count: number) => { captured.counts.push(count); },
|
||||
finish: () => undefined,
|
||||
}),
|
||||
}));
|
||||
|
||||
const {
|
||||
NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS,
|
||||
reconcileNativeFinalizations,
|
||||
resetSettledCommittedUpkeepForTests,
|
||||
} = await import("../services/native-runtime/native-finalization-reconciler.js");
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
const HOUR_MS = 60 * 60 * 1000;
|
||||
|
||||
async function candidateCount(pass: Promise<unknown>) {
|
||||
await expect(pass).rejects.toBeInstanceOf(StopAfterCandidateScan);
|
||||
return captured.counts.at(-1);
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("native finalization candidate selection", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let companyId: string;
|
||||
let agentId: string;
|
||||
let issueId: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-settled-candidates-");
|
||||
db = createDb(temporary.connectionString);
|
||||
}, 30_000);
|
||||
|
||||
beforeEach(async () => {
|
||||
captured.counts = [];
|
||||
resetSettledCommittedUpkeepForTests();
|
||||
companyId = randomUUID();
|
||||
agentId = randomUUID();
|
||||
issueId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Settled", issuePrefix: `S${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({
|
||||
id: agentId, companyId, name: "Native", role: "engineer", status: "active",
|
||||
adapterType: "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {},
|
||||
});
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Settled work", status: "done" });
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(statusDecisionEffects);
|
||||
await db.delete(nativeRunFinalizations);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(issues);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await temporary?.cleanup();
|
||||
});
|
||||
|
||||
async function committedRun(options: { status?: string; finishedAgoMs?: number | null; decisionId?: string } = {}) {
|
||||
const runId = randomUUID();
|
||||
const finishedAgoMs = options.finishedAgoMs === undefined ? 2 * HOUR_MS : options.finishedAgoMs;
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId, companyId, agentId,
|
||||
status: options.status ?? "succeeded",
|
||||
runtimeMode: "native",
|
||||
nativeIssueId: issueId,
|
||||
finishedAt: finishedAgoMs === null ? null : new Date(Date.now() - finishedAgoMs),
|
||||
contextSnapshot: { issueId },
|
||||
});
|
||||
// Result, assessment and decision rows need full contract chains. These
|
||||
// tests check candidate selection only, so foreign-key triggers are off
|
||||
// for this insert; check constraints still apply.
|
||||
await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`set local session_replication_role = replica`);
|
||||
await tx.insert(nativeRunFinalizations).values({
|
||||
runId, companyId, issueId, phase: "committed",
|
||||
resultId: randomUUID(),
|
||||
...(options.decisionId ? { assessmentId: randomUUID(), decisionId: options.decisionId } : {}),
|
||||
});
|
||||
});
|
||||
return runId;
|
||||
}
|
||||
|
||||
it("skips settled committed rows outside the upkeep rotation and rotates through them", async () => {
|
||||
const settled = NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS * 2 + 5;
|
||||
for (let index = 0; index < settled; index++) await committedRun();
|
||||
|
||||
const counts = [];
|
||||
for (let pass = 0; pass < 4; pass++) counts.push(await candidateCount(reconcileNativeFinalizations(db)));
|
||||
|
||||
// Two full upkeep pages, the remainder, then the rotation wraps.
|
||||
expect(counts).toEqual([
|
||||
NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS,
|
||||
NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS,
|
||||
5,
|
||||
NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS,
|
||||
]);
|
||||
});
|
||||
|
||||
// More settled rows than one upkeep page, so the page is full and a
|
||||
// protected row can only be a candidate through its own protection.
|
||||
async function fillUpkeepPage() {
|
||||
for (let index = 0; index < NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS + 2; index++) await committedRun();
|
||||
}
|
||||
|
||||
it("keeps committed rows that finished within the grace window or have not finished", async () => {
|
||||
await fillUpkeepPage();
|
||||
const protectedRows = 12;
|
||||
for (let index = 0; index < protectedRows; index++) {
|
||||
const variant = index % 3;
|
||||
if (variant === 0) await committedRun({ finishedAgoMs: 10 * 60 * 1000 });
|
||||
else if (variant === 1) await committedRun({ status: "running", finishedAgoMs: null });
|
||||
else await committedRun({ status: "succeeded", finishedAgoMs: null });
|
||||
}
|
||||
|
||||
expect(await candidateCount(reconcileNativeFinalizations(db)))
|
||||
.toBe(NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS + protectedRows);
|
||||
});
|
||||
|
||||
it("keeps settled committed rows whose decision has an undelivered effect", async () => {
|
||||
await fillUpkeepPage();
|
||||
const protectedRows = 12;
|
||||
for (let index = 0; index < protectedRows; index++) {
|
||||
const decisionId = randomUUID();
|
||||
await committedRun({ decisionId });
|
||||
// As above, the decision row is not created; foreign-key triggers are off.
|
||||
await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`set local session_replication_role = replica`);
|
||||
await tx.insert(statusDecisionEffects).values({
|
||||
companyId, issueId, decisionId, ordinal: 0, effectKind: "wake", targetType: "agent",
|
||||
idempotencyKey: randomUUID(), payload: {}, deliveryState: index % 2 === 0 ? "failed" : "pending",
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
expect(await candidateCount(reconcileNativeFinalizations(db)))
|
||||
.toBe(NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS + protectedRows);
|
||||
});
|
||||
|
||||
it("reconciles a settled row when a pass names it explicitly", async () => {
|
||||
const runId = await committedRun();
|
||||
await committedRun();
|
||||
|
||||
expect(await candidateCount(reconcileNativeFinalizations(db, [runId]))).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -3,7 +3,7 @@ import { hasPendingNativeChildCompletion } from "./native-child-completion-deliv
|
||||
import { dismissAutomaticCompletionReviews, decisionHasRetiredAutomaticReview } from "./automatic-completion-reviews.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, notInArray, or, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, ne, not, notInArray, or, sql, type SQL } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
completionContracts,
|
||||
@@ -634,6 +634,62 @@ type NativeReconciliationOptions = {
|
||||
|
||||
type NativeReconciliationPassTiming = ReturnType<typeof createNativeReconciliationPassTiming>;
|
||||
|
||||
// A committed finalization whose run ended this long ago, with no undelivered
|
||||
// decision effect, is settled: its result is projected and its effects are
|
||||
// delivered, so an unscoped pass has nothing to finish. Re-reading every such
|
||||
// row on every pass cost dozens of sequential queries per row, and startup
|
||||
// runs the pass twice before readiness. Recent rows stay in every pass, so the
|
||||
// crash-window repairs still run promptly.
|
||||
export const NATIVE_SETTLED_COMMITTED_GRACE_MS = 60 * 60 * 1000;
|
||||
// Settled rows still get the presentation and evidence upkeep, a few per pass
|
||||
// in a rotating order, so a late repair is delayed, never skipped.
|
||||
export const NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS = 10;
|
||||
const NATIVE_SETTLED_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"];
|
||||
let settledCommittedUpkeepCursor: string | null = null;
|
||||
|
||||
/** Test hook: restart the settled-row upkeep rotation from the beginning. */
|
||||
export function resetSettledCommittedUpkeepForTests() {
|
||||
settledCommittedUpkeepCursor = null;
|
||||
}
|
||||
|
||||
function settledCommittedFinalization(now: Date): SQL {
|
||||
return and(
|
||||
eq(nativeRunFinalizations.phase, "committed"),
|
||||
inArray(heartbeatRuns.status, NATIVE_SETTLED_RUN_STATUSES),
|
||||
isNotNull(heartbeatRuns.finishedAt),
|
||||
lt(heartbeatRuns.finishedAt, new Date(now.getTime() - NATIVE_SETTLED_COMMITTED_GRACE_MS)),
|
||||
sql<boolean>`not exists (
|
||||
select 1 from ${statusDecisionEffects} effect
|
||||
where effect.company_id = ${nativeRunFinalizations.companyId}
|
||||
and effect.issue_id = ${nativeRunFinalizations.issueId}
|
||||
and effect.decision_id = ${nativeRunFinalizations.decisionId}
|
||||
and effect.delivery_state in ('pending', 'failed')
|
||||
)`,
|
||||
)!;
|
||||
}
|
||||
|
||||
/** The next settled committed runs in the upkeep rotation, wrapping at the end. */
|
||||
async function nextSettledCommittedUpkeep(db: Db, now: Date): Promise<string[]> {
|
||||
const page = (after: string | null) => db
|
||||
.select({ runId: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.where(and(
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
isNotNull(nativeRunFinalizations.resultId),
|
||||
settledCommittedFinalization(now),
|
||||
...(after ? [gt(heartbeatRuns.id, after)] : []),
|
||||
))
|
||||
.orderBy(asc(heartbeatRuns.id))
|
||||
.limit(NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS);
|
||||
let rows = await page(settledCommittedUpkeepCursor);
|
||||
if (rows.length === 0 && settledCommittedUpkeepCursor !== null) rows = await page(null);
|
||||
settledCommittedUpkeepCursor = rows.length === NATIVE_SETTLED_COMMITTED_UPKEEP_PER_PASS
|
||||
? rows[rows.length - 1]!.runId
|
||||
: null;
|
||||
return rows.map((row) => row.runId);
|
||||
}
|
||||
|
||||
/** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */
|
||||
export async function reconcileNativeFinalizations(
|
||||
db: Db,
|
||||
@@ -675,6 +731,11 @@ async function reconcileNativeFinalizationsPass(
|
||||
await dismissAutomaticCompletionReviews(db);
|
||||
}
|
||||
timing.phase("candidate_scan");
|
||||
const now = new Date();
|
||||
// An unscoped pass skips settled committed rows, except the few in this
|
||||
// pass's upkeep rotation. A scoped pass reconciles exactly the given runs.
|
||||
const unscoped = !(runIds && runIds.length > 0);
|
||||
const upkeepRunIds = unscoped ? await nextSettledCommittedUpkeep(db, now) : [];
|
||||
const rows = await db
|
||||
.select({
|
||||
runId: heartbeatRuns.id,
|
||||
@@ -711,6 +772,11 @@ async function reconcileNativeFinalizationsPass(
|
||||
lte(nativeRunFinalizations.leaseExpiresAt, new Date()),
|
||||
),
|
||||
...(runIds && runIds.length > 0 ? [inArray(heartbeatRuns.id, runIds)] : []),
|
||||
...(unscoped
|
||||
? [upkeepRunIds.length > 0
|
||||
? or(not(settledCommittedFinalization(now)), inArray(heartbeatRuns.id, upkeepRunIds))!
|
||||
: not(settledCommittedFinalization(now))]
|
||||
: []),
|
||||
));
|
||||
timing.candidates(rows.length);
|
||||
timing.phase("candidates");
|
||||
|
||||
Reference in new issue
Block a user