mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
perf(server): stop readiness waiting on startup housekeeping (#16329)
## 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 deploy tooling waits for `ok` > - After heartbeat recovery, readiness also waits for housekeeping that every scheduler tick repeats and that serving requests does not need > - #16330 added OpenTelemetry spans for the main startup phases; native recovery and orphan reaping are the slow ones, and both run the native finalization reconciliation pass, which has no inner breakdown > - This pull request stops readiness waiting on that housekeeping, and records the phases of slow reconciliation passes as spans > - The benefit is faster readiness on every restart, and traces that show where a slow reconciliation pass spends its time ## Linked Issues or Issue Description Builds on #16330, which records `controller.startup.<phase>` spans. Related to #16328. **What existing behavior does this improve?** Server readiness time after a restart, and diagnosis of slow native recovery. **Current behavior** - After heartbeat recovery, readiness waits for the custom-image setup cleanup, tool connection health (remote MCP checks), decision expiry, the adapter and setup-token login reapers, environment lease cleanup (provider calls) and the retention sweep. Each also runs on every scheduler tick. On one instance they added 11–28 s to every measured restart. A throw in any of them fails startup. - On the same instance, native recovery took from 33 s to 11.5 min and orphan reaping from 30 s to 4 min across restarts in one hour. Both run the native finalization reconciliation pass. The #16330 spans time each step as a whole, but not the phases inside that pass. **Proposed behavior** - The housekeeping steps run after readiness. The scheduler interval still starts after them. - With an OTLP endpoint, a reconciliation pass of at least one second records a span with one child span per phase. **Reason and benefit** Readiness depends only on what serving requests and run recovery need. Slow reconciliation passes become attributable to a phase in the existing traces, without adding trace volume for the periodic pass. **Breaking changes** None. A failed housekeeping step no longer fails startup; it is logged and the next step runs. ## What Changed - `server/src/index.ts`: moves the housekeeping steps into `runStartupHousekeeping`, which runs after readiness under `trackHeartbeatSchedulerWork` (idle sleep and shutdown wait for it). Each step catches and logs its own failure (`startup housekeeping step failed`). The heartbeat scheduler interval starts when housekeeping finishes, unless the scheduler was stopped, so the reapers still finish before the first timer tick. - New `server/src/services/native-runtime/native-reconciliation-timing.ts`: records the phase boundaries of one pass. When the pass took at least one second and an OTLP endpoint is configured, it emits a `controller.native_reconciliation` root span (`paperclip.native-reconciliation`) with child spans `legacy_workspace_exports`, `obsolete_policy_reviews`, `automatic_completion_reviews`, `candidate_scan` and `candidates`, using the recorded start and end times. It records `scoped` and the candidate count only. A failed pass marks the pass span and the running phase. Tracing failures are swallowed. - `native-finalization-reconciler.ts`: the exported function wraps the pass in that timing and finishes it in `finally`, so failed passes are recorded; the original error is rethrown. - `doc/execution-semantics.md`: what readiness waits for, and the housekeeping order. `doc/observability.md`: the reconciliation spans. ## Verification - `pnpm exec vitest run src/__tests__/server-startup-feedback-export.test.ts src/services/native-runtime/native-reconciliation-timing.test.ts src/services/native-runtime/native-finalization-reconciler-timing.test.ts src/startup-recovery-timing.test.ts` (in `server/`): 52 passed. New tests: - Readiness is `ready` while a housekeeping step hangs, and the heartbeat interval starts only after the step settles. - A failed housekeeping step is logged, and the scheduler still starts after the remaining steps. - Span timing: nothing for a fast pass or without an endpoint; a slow pass emits the pass span and one child per phase with the recorded times; a failed pass marks the running phase and the pass; a failing tracer does not throw. - The reconciler records its phases and a failed outcome when a pass throws, and rethrows. - Three existing startup tests settle housekeeping before they read the scheduler interval or the setup cleanup. - `pnpm exec vitest run src/services/native-runtime`: 4,399 passed. - `pnpm typecheck` in `server/`: passed. ## Risks - Housekeeping now runs concurrently with the first requests and with `waitForExternalAdapters`. None of these steps is needed to serve requests, and each one also runs on every scheduler tick. - Readiness is reported earlier; the heartbeat timer starts at the same point relative to housekeeping as before. - The new spans exist only with an OTLP endpoint and only for passes of at least one second. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code with tool use (shell, file edits, test runs). ## 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
aa04c5c2a7
commit
cda77edc60
8 files changed
+391
-44
No files matched your search
@@ -841,6 +841,15 @@ On startup and on the periodic recovery loop, Paperclip performs the following r
|
||||
|
||||
The stranded-work pass closes the gap where issue state survives a crash but the wake/run path does not. The silent-run scan covers the separate case where a live process exists but has stopped producing observable output.
|
||||
|
||||
Startup health (`/api/health`) reports `starting` until these recovery passes and
|
||||
the other startup steps that serving requests depends on finish. Housekeeping
|
||||
that every scheduler tick repeats does not delay it: the custom-image setup
|
||||
cleanup, tool connection health, decision expiry, adapter and setup-token login
|
||||
reapers, environment lease cleanup, and the retention sweep run after readiness.
|
||||
A failed housekeeping step is logged as `startup housekeeping step failed` and
|
||||
the next step still runs. The heartbeat scheduler interval starts after
|
||||
housekeeping, so the reapers still finish before the first timer tick.
|
||||
|
||||
The stale-lock sweep leaves native runs with their finalization coordinator while
|
||||
same-session recovery or accepted-result finalization is pending. A provider can
|
||||
exit before workspace copy-back, assessment, or arbitration completes. An expired
|
||||
|
||||
@@ -229,6 +229,17 @@ provider startup; they are not agent-run durations. A disabled endpoint takes
|
||||
the direct work path without a tracer lookup or additional log messages.
|
||||
Tracing failures cannot repeat recovery or replace its result or error.
|
||||
|
||||
Native recovery and orphan reaping each run a native finalization
|
||||
reconciliation pass; the periodic recovery loop runs it on every scheduler tick.
|
||||
With an endpoint, a pass that takes at least one second records, after it ends,
|
||||
a `controller.native_reconciliation` root span from `paperclip.native-reconciliation`
|
||||
with one child span per phase: `legacy_workspace_exports`,
|
||||
`obsolete_policy_reviews`, `automatic_completion_reviews`, `candidate_scan` and
|
||||
`candidates`. The pass span carries `paperclip.native_reconciliation.scoped` and,
|
||||
once known, `paperclip.native_reconciliation.candidates` (a count). A failed
|
||||
pass marks the pass span and the phase that was running with an error status.
|
||||
Faster passes record nothing.
|
||||
|
||||
## Native Runner Trace Spans
|
||||
|
||||
Paperclip Runner task runs emit a single foldable OpenTelemetry trace. This is
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { getStartupRecoveryState } from "../startup-recovery-state.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
|
||||
import * as idleLocalWork from "../services/idle-local-work.js";
|
||||
@@ -445,6 +447,14 @@ import { startServer } from "../index.ts";
|
||||
import { reconcileSafeNativeReplacements } from "../services/native-runtime/native-safe-replacement.js";
|
||||
import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "../services/execution-control-deadline.js";
|
||||
|
||||
// Startup housekeeping runs after readiness and starts the heartbeat scheduler
|
||||
// interval when it finishes. Let that background work settle.
|
||||
async function settleStartupHousekeeping() {
|
||||
for (let index = 0; index < 10; index += 1) {
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
}
|
||||
}
|
||||
|
||||
describe("startServer feedback export wiring", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
@@ -721,6 +731,7 @@ describe("startServer feedback export wiring", () => {
|
||||
}) as typeof setInterval);
|
||||
try {
|
||||
await startServer();
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(1);
|
||||
expect(intervalCallback).not.toBeNull();
|
||||
@@ -782,6 +793,7 @@ describe("startServer feedback export wiring", () => {
|
||||
|
||||
try {
|
||||
await startServer();
|
||||
await settleStartupHousekeeping();
|
||||
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled();
|
||||
expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled();
|
||||
@@ -849,6 +861,7 @@ describe("startServer feedback export wiring", () => {
|
||||
}) as typeof setInterval);
|
||||
try {
|
||||
expect((await startServer()).server).toBe(fakeServer);
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(tick).toBeDefined();
|
||||
tick?.();
|
||||
@@ -859,6 +872,67 @@ describe("startServer feedback export wiring", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("reports ready before startup housekeeping finishes and starts the scheduler after it", async () => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({
|
||||
heartbeatSchedulerEnabled: true,
|
||||
heartbeatSchedulerIntervalMs: 30000,
|
||||
}));
|
||||
let releaseCleanup: (() => void) | undefined;
|
||||
environmentCustomImagesServiceMock.cleanupExpiredSetupSessions.mockImplementationOnce(() =>
|
||||
new Promise((resolve) => {
|
||||
releaseCleanup = () => resolve({ scanned: 0, timedOut: 0, failed: 0 });
|
||||
}));
|
||||
const heartbeatIntervals: Array<() => void> = [];
|
||||
const setIntervalSpy = vi.spyOn(globalThis, "setInterval").mockImplementation(((
|
||||
callback: () => void,
|
||||
interval: number,
|
||||
) => {
|
||||
if (interval === 30000) heartbeatIntervals.push(callback);
|
||||
return 1 as unknown as ReturnType<typeof setInterval>;
|
||||
}) as typeof setInterval);
|
||||
try {
|
||||
await startServer();
|
||||
// Readiness does not wait for the hung housekeeping step.
|
||||
expect(getStartupRecoveryState().phase).toBe("ready");
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatIntervals).toHaveLength(0);
|
||||
|
||||
releaseCleanup?.();
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatIntervals).toHaveLength(1);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("runs the remaining startup housekeeping steps when one fails", async () => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({
|
||||
heartbeatSchedulerEnabled: true,
|
||||
heartbeatSchedulerIntervalMs: 30000,
|
||||
}));
|
||||
environmentCustomImagesServiceMock.cleanupExpiredSetupSessions.mockRejectedValueOnce(new Error("provider outage"));
|
||||
const heartbeatIntervals: Array<() => void> = [];
|
||||
const setIntervalSpy = vi.spyOn(globalThis, "setInterval").mockImplementation(((
|
||||
callback: () => void,
|
||||
interval: number,
|
||||
) => {
|
||||
if (interval === 30000) heartbeatIntervals.push(callback);
|
||||
return 1 as unknown as ReturnType<typeof setInterval>;
|
||||
}) as typeof setInterval);
|
||||
try {
|
||||
await expect(startServer()).resolves.toBeDefined();
|
||||
await settleStartupHousekeeping();
|
||||
expect(vi.mocked(logger.error)).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ startupStep: "custom_image_setup_cleanup" }),
|
||||
"startup housekeeping step failed",
|
||||
);
|
||||
// The scheduler starts only after every housekeeping step settled.
|
||||
expect(heartbeatIntervals).toHaveLength(1);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not replay hot-restart adoption when the orphan reaper retries", async () => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({
|
||||
heartbeatSchedulerEnabled: true,
|
||||
|
||||
+39
-35
@@ -1624,39 +1624,6 @@ async function startServerWithDatabaseTeardown(
|
||||
await startupHeartbeatRecovery;
|
||||
}
|
||||
|
||||
const setupCleanup = await environmentCustomImages.cleanupExpiredSetupSessions();
|
||||
if (setupCleanup.timedOut > 0 || setupCleanup.failed > 0) {
|
||||
logger.warn({ ...setupCleanup }, "startup environment customImage setup cleanup changed sessions");
|
||||
}
|
||||
|
||||
const toolHealthSweep = await tools.sweepConnectionHealth();
|
||||
if (toolHealthSweep.failed > 0) {
|
||||
logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections");
|
||||
}
|
||||
await decisionExecutor.sweepExpired();
|
||||
|
||||
// Run the adapter login reaper once at startup, so a login sandbox that
|
||||
// outlived a server restart is deleted before timer ticks start.
|
||||
await adapterLoginReaper
|
||||
.sweep()
|
||||
.then(logAdapterLoginReaperResult)
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "startup adapter login reaper sweep failed");
|
||||
});
|
||||
|
||||
// Run the setup-token login reaper once at startup, so a login sandbox lease
|
||||
// that outlived a server restart releases before timer ticks start.
|
||||
await setupTokenReaper
|
||||
.sweep()
|
||||
.then(logSetupTokenReaperResult)
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "startup setup-token login reaper sweep failed");
|
||||
});
|
||||
|
||||
// Retry any orphan sandbox teardown left by a failed acquire before a server
|
||||
// restart, so a leaked sandbox does not stay allocated across the restart.
|
||||
await runEnvironmentLeaseCleanupSweep(0);
|
||||
|
||||
const runRetentionSweep = async () => {
|
||||
const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
|
||||
let archived = 0;
|
||||
@@ -1674,9 +1641,43 @@ async function startServerWithDatabaseTeardown(
|
||||
const notifications = await retentionExecutor.deliverNotifications();
|
||||
return { archived, ...notifications };
|
||||
};
|
||||
await runRetentionSweep();
|
||||
// Startup housekeeping: provider and connection sweeps that every timer tick
|
||||
// repeats and that neither serving requests nor run recovery depends on.
|
||||
// Readiness does not wait for it. It is tracked as scheduler work, so idle
|
||||
// sleep and shutdown wait for it. Each step logs its own failure and the
|
||||
// next step still runs. The scheduler interval starts after it, so the login
|
||||
// reapers and the lease cleanup still finish before the first timer tick.
|
||||
const runStartupHousekeeping = async () => {
|
||||
const step = (name: string, work: () => Promise<unknown>) =>
|
||||
work().catch((err: unknown) => {
|
||||
logger.error({ err, startupStep: name }, "startup housekeeping step failed");
|
||||
});
|
||||
await step("custom_image_setup_cleanup", async () => {
|
||||
const setupCleanup = await environmentCustomImages.cleanupExpiredSetupSessions();
|
||||
if (setupCleanup.timedOut > 0 || setupCleanup.failed > 0) {
|
||||
logger.warn({ ...setupCleanup }, "startup environment customImage setup cleanup changed sessions");
|
||||
}
|
||||
});
|
||||
await step("tool_connection_health", async () => {
|
||||
const toolHealthSweep = await tools.sweepConnectionHealth();
|
||||
if (toolHealthSweep.failed > 0) {
|
||||
logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections");
|
||||
}
|
||||
});
|
||||
await step("decision_expiry", () => decisionExecutor.sweepExpired());
|
||||
// A login sandbox that outlived a server restart is deleted before timer
|
||||
// ticks start.
|
||||
await step("adapter_login_reaper", () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult));
|
||||
// A setup-token login sandbox lease that outlived a server restart
|
||||
// releases before timer ticks start.
|
||||
await step("setup_token_reaper", () => setupTokenReaper.sweep().then(logSetupTokenReaperResult));
|
||||
// Retry any orphan sandbox teardown left by a failed acquire before a
|
||||
// server restart, so a leaked sandbox does not stay allocated.
|
||||
await step("environment_lease_cleanup", () => runEnvironmentLeaseCleanupSweep(0));
|
||||
await step("retention_sweep", runRetentionSweep);
|
||||
};
|
||||
|
||||
startHeartbeatSchedulerInterval(() => {
|
||||
const startHeartbeatScheduler = () => startHeartbeatSchedulerInterval(() => {
|
||||
// Track the outer async callback as well as the work it starts. Shutdown
|
||||
// can then wait through an already-running suppression check before it
|
||||
// captures the authoritative set of running heartbeat rows.
|
||||
@@ -1855,6 +1856,9 @@ async function startServerWithDatabaseTeardown(
|
||||
logger.error({ err }, "heartbeat scheduler tick failed");
|
||||
}));
|
||||
});
|
||||
trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => {
|
||||
if (!heartbeatSchedulerStopped) startHeartbeatScheduler();
|
||||
}));
|
||||
} else {
|
||||
// The heartbeat scheduler is disabled, but the orphan-sandbox cleanup sweep
|
||||
// is still required. A failed acquire can leak a paid provider sandbox, so
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
const timing = vi.hoisted(() => ({
|
||||
phase: vi.fn(),
|
||||
candidates: vi.fn(),
|
||||
finish: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("./native-reconciliation-timing.js", () => ({
|
||||
createNativeReconciliationPassTiming: () => timing,
|
||||
}));
|
||||
vi.mock("./native-workspace-export-recovery.js", () => ({
|
||||
recoverLegacyUnsafeWorkspaceExports: vi.fn(async () => undefined),
|
||||
}));
|
||||
vi.mock("./obsolete-policy-reviews.js", () => ({
|
||||
dismissObsoleteNativePolicyReviews: vi.fn(async () => undefined),
|
||||
}));
|
||||
vi.mock("./automatic-completion-reviews.js", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("./automatic-completion-reviews.js")>()),
|
||||
dismissAutomaticCompletionReviews: vi.fn(async () => {
|
||||
throw new Error("review lookup failed");
|
||||
}),
|
||||
}));
|
||||
|
||||
const { reconcileNativeFinalizations } = await import("./native-finalization-reconciler.js");
|
||||
|
||||
describe("native finalization reconciliation timing", () => {
|
||||
it("records the phases and a failed outcome when a pass fails, and rethrows", async () => {
|
||||
await expect(reconcileNativeFinalizations({} as never)).rejects.toThrow("review lookup failed");
|
||||
|
||||
expect(timing.phase.mock.calls.map(([phase]) => phase)).toEqual([
|
||||
"legacy_workspace_exports",
|
||||
"obsolete_policy_reviews",
|
||||
"automatic_completion_reviews",
|
||||
]);
|
||||
expect(timing.finish).toHaveBeenCalledExactlyOnceWith({ outcome: "failed", scoped: false });
|
||||
});
|
||||
});
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
} from "./native-workspace-sync.js";
|
||||
import type { EnvironmentRuntimeService } from "../environment-runtime.js";
|
||||
import { classifyNativeEvidence } from "./evidence-classifier.js";
|
||||
import { createNativeReconciliationPassTiming } from "./native-reconciliation-timing.js";
|
||||
import { recordNativeWorkAssessment } from "./work-assessments.js";
|
||||
import {
|
||||
isNativeRunnerOwnershipHeld,
|
||||
@@ -546,26 +547,51 @@ export async function claimNativeSessionResumptions(input: {
|
||||
return claims;
|
||||
}
|
||||
|
||||
type NativeReconciliationOptions = {
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
onWorkspaceSettled?: (input: {
|
||||
runId: string;
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
succeeded: boolean;
|
||||
}) => Promise<void>;
|
||||
};
|
||||
|
||||
type NativeReconciliationPassTiming = ReturnType<typeof createNativeReconciliationPassTiming>;
|
||||
|
||||
/** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */
|
||||
export async function reconcileNativeFinalizations(
|
||||
db: Db,
|
||||
runIds?: string[],
|
||||
options: {
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
onWorkspaceSettled?: (input: {
|
||||
runId: string;
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
succeeded: boolean;
|
||||
}) => Promise<void>;
|
||||
} = {},
|
||||
options: NativeReconciliationOptions = {},
|
||||
) {
|
||||
const timing = createNativeReconciliationPassTiming();
|
||||
let outcome: "ok" | "failed" = "ok";
|
||||
try {
|
||||
return await reconcileNativeFinalizationsPass(db, runIds, options, timing);
|
||||
} catch (error) {
|
||||
outcome = "failed";
|
||||
throw error;
|
||||
} finally {
|
||||
timing.finish({ outcome, scoped: Boolean(runIds?.length) });
|
||||
}
|
||||
}
|
||||
|
||||
async function reconcileNativeFinalizationsPass(
|
||||
db: Db,
|
||||
runIds: string[] | undefined,
|
||||
options: NativeReconciliationOptions,
|
||||
timing: NativeReconciliationPassTiming,
|
||||
) {
|
||||
timing.phase("legacy_workspace_exports");
|
||||
await recoverLegacyUnsafeWorkspaceExports(db, runIds).catch((err) => {
|
||||
logger.warn({ err }, "Historical unsafe export recovery remains pending");
|
||||
});
|
||||
timing.phase("obsolete_policy_reviews");
|
||||
await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => {
|
||||
logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation");
|
||||
});
|
||||
timing.phase("automatic_completion_reviews");
|
||||
if (runIds?.length) {
|
||||
const scopes = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations)
|
||||
.where(inArray(nativeRunFinalizations.runId, runIds));
|
||||
@@ -573,6 +599,7 @@ export async function reconcileNativeFinalizations(
|
||||
} else {
|
||||
await dismissAutomaticCompletionReviews(db);
|
||||
}
|
||||
timing.phase("candidate_scan");
|
||||
const rows = await db
|
||||
.select({
|
||||
runId: heartbeatRuns.id,
|
||||
@@ -610,6 +637,8 @@ export async function reconcileNativeFinalizations(
|
||||
),
|
||||
...(runIds && runIds.length > 0 ? [inArray(heartbeatRuns.id, runIds)] : []),
|
||||
));
|
||||
timing.candidates(rows.length);
|
||||
timing.phase("candidates");
|
||||
const results = [];
|
||||
for (const row of rows) {
|
||||
const pendingEffects = row.decisionId
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const handles = vi.hoisted(() => ({
|
||||
startSpan: vi.fn(),
|
||||
getTracer: vi.fn(),
|
||||
setSpan: vi.fn((_context: unknown, span: unknown) => ({ parent: span })),
|
||||
}));
|
||||
vi.mock("@opentelemetry/api", () => ({
|
||||
ROOT_CONTEXT: "root",
|
||||
SpanStatusCode: { ERROR: 2 },
|
||||
trace: { getTracer: handles.getTracer, setSpan: handles.setSpan },
|
||||
}));
|
||||
|
||||
const { createNativeReconciliationPassTiming } = await import("./native-reconciliation-timing.js");
|
||||
|
||||
function clock() {
|
||||
let at = 10_000;
|
||||
return { now: () => at, advance: (ms: number) => { at += ms; } };
|
||||
}
|
||||
|
||||
function spanStub(name: string) {
|
||||
return { name, end: vi.fn(), setStatus: vi.fn() };
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
vi.clearAllMocks();
|
||||
handles.getTracer.mockReturnValue({ startSpan: handles.startSpan });
|
||||
handles.startSpan.mockImplementation((name: string) => spanStub(name));
|
||||
});
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
|
||||
describe("native reconciliation pass timing", () => {
|
||||
it("records nothing for a fast pass", () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
const c = clock();
|
||||
const timing = createNativeReconciliationPassTiming({ now: c.now });
|
||||
timing.phase("candidate_scan");
|
||||
c.advance(999);
|
||||
timing.finish({ outcome: "ok", scoped: false });
|
||||
expect(handles.getTracer).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("records nothing without an OTLP endpoint", () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "");
|
||||
const c = clock();
|
||||
const timing = createNativeReconciliationPassTiming({ now: c.now });
|
||||
timing.phase("candidate_scan");
|
||||
c.advance(5_000);
|
||||
timing.finish({ outcome: "ok", scoped: false });
|
||||
expect(handles.getTracer).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("emits a pass span with one child span per phase for a slow pass", () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
const c = clock();
|
||||
const timing = createNativeReconciliationPassTiming({ now: c.now });
|
||||
timing.phase("legacy_workspace_exports");
|
||||
c.advance(100);
|
||||
timing.phase("candidate_scan");
|
||||
c.advance(300);
|
||||
timing.candidates(187);
|
||||
timing.phase("candidates");
|
||||
c.advance(2_000);
|
||||
timing.finish({ outcome: "ok", scoped: false });
|
||||
|
||||
const calls = handles.startSpan.mock.calls;
|
||||
expect(calls[0]).toEqual([
|
||||
"controller.native_reconciliation",
|
||||
{ startTime: 10_000, attributes: { "paperclip.native_reconciliation.scoped": false, "paperclip.native_reconciliation.candidates": 187 } },
|
||||
"root",
|
||||
]);
|
||||
const pass = handles.startSpan.mock.results[0]!.value;
|
||||
expect(calls.slice(1).map(([name, options, context]) => [name, options.startTime, context])).toEqual([
|
||||
["controller.native_reconciliation.legacy_workspace_exports", 10_000, { parent: pass }],
|
||||
["controller.native_reconciliation.candidate_scan", 10_100, { parent: pass }],
|
||||
["controller.native_reconciliation.candidates", 10_400, { parent: pass }],
|
||||
]);
|
||||
const children = handles.startSpan.mock.results.slice(1).map(result => result.value);
|
||||
expect(children.map(span => span.end.mock.calls[0]?.[0])).toEqual([10_100, 10_400, 12_400]);
|
||||
expect(pass.end).toHaveBeenCalledWith(12_400);
|
||||
expect(pass.setStatus).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("marks the failed phase and the pass when a slow pass fails", () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
const c = clock();
|
||||
const timing = createNativeReconciliationPassTiming({ now: c.now });
|
||||
timing.phase("obsolete_policy_reviews");
|
||||
c.advance(10);
|
||||
timing.phase("automatic_completion_reviews");
|
||||
c.advance(2_000);
|
||||
timing.finish({ outcome: "failed", scoped: true });
|
||||
|
||||
const [pass, reviews, completion] = handles.startSpan.mock.results.map(result => result.value);
|
||||
expect(pass.setStatus).toHaveBeenCalledWith({ code: 2 });
|
||||
expect(reviews.setStatus).not.toHaveBeenCalled();
|
||||
expect(completion.name).toBe("controller.native_reconciliation.automatic_completion_reviews");
|
||||
expect(completion.setStatus).toHaveBeenCalledWith({ code: 2 });
|
||||
});
|
||||
|
||||
it("never throws when tracing fails", () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
handles.getTracer.mockImplementation(() => { throw new Error("tracer unavailable"); });
|
||||
const c = clock();
|
||||
const timing = createNativeReconciliationPassTiming({ now: c.now });
|
||||
timing.phase("candidates");
|
||||
c.advance(5_000);
|
||||
expect(() => timing.finish({ outcome: "ok", scoped: false })).not.toThrow();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,71 @@
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { ROOT_CONTEXT, SpanStatusCode, trace } from "@opentelemetry/api";
|
||||
|
||||
export type NativeReconciliationPhase = "legacy_workspace_exports" | "obsolete_policy_reviews"
|
||||
| "automatic_completion_reviews" | "candidate_scan" | "candidates";
|
||||
|
||||
// A pass at least this slow records its phases. Startup recovery runs the pass
|
||||
// twice and readiness waits for both; the periodic pass runs on every
|
||||
// scheduler tick, so fast passes record nothing.
|
||||
export const NATIVE_RECONCILIATION_SLOW_PASS_MS = 1_000;
|
||||
|
||||
type RecordedPhase = { phase: NativeReconciliationPhase; startedAt: number; endedAt: number };
|
||||
|
||||
// Records phase boundaries of one native finalization reconciliation pass.
|
||||
// When the pass is slow and an OTLP endpoint is configured, `finish` emits a
|
||||
// `controller.native_reconciliation` span with one child span per phase, using
|
||||
// the recorded start and end times. It records counts and outcome only.
|
||||
export function createNativeReconciliationPassTiming(options: { now?: () => number } = {}) {
|
||||
const now = options.now ?? (() => performance.timeOrigin + performance.now());
|
||||
const passStartedAt = now();
|
||||
const phases: RecordedPhase[] = [];
|
||||
let current: { phase: NativeReconciliationPhase; startedAt: number } | null = null;
|
||||
let candidates: number | null = null;
|
||||
|
||||
const closeCurrent = () => {
|
||||
if (!current) return;
|
||||
phases.push({ ...current, endedAt: now() });
|
||||
current = null;
|
||||
};
|
||||
|
||||
return {
|
||||
phase(phase: NativeReconciliationPhase) {
|
||||
closeCurrent();
|
||||
current = { phase, startedAt: now() };
|
||||
},
|
||||
candidates(count: number) {
|
||||
candidates = count;
|
||||
},
|
||||
finish(details: { outcome: "ok" | "failed"; scoped: boolean }) {
|
||||
const failedPhase = details.outcome === "failed" ? current?.phase ?? null : null;
|
||||
closeCurrent();
|
||||
const passEndedAt = now();
|
||||
if (passEndedAt - passStartedAt < NATIVE_RECONCILIATION_SLOW_PASS_MS) return;
|
||||
if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim()) return;
|
||||
try {
|
||||
const tracer = trace.getTracer("paperclip.native-reconciliation");
|
||||
const pass = tracer.startSpan("controller.native_reconciliation", {
|
||||
startTime: passStartedAt,
|
||||
attributes: {
|
||||
"paperclip.native_reconciliation.scoped": details.scoped,
|
||||
...(candidates === null ? {} : { "paperclip.native_reconciliation.candidates": candidates }),
|
||||
},
|
||||
}, ROOT_CONTEXT);
|
||||
const passContext = trace.setSpan(ROOT_CONTEXT, pass);
|
||||
for (const recorded of phases) {
|
||||
const span = tracer.startSpan(
|
||||
`controller.native_reconciliation.${recorded.phase}`,
|
||||
{ startTime: recorded.startedAt },
|
||||
passContext,
|
||||
);
|
||||
if (recorded.phase === failedPhase) span.setStatus({ code: SpanStatusCode.ERROR });
|
||||
span.end(recorded.endedAt);
|
||||
}
|
||||
if (details.outcome === "failed") pass.setStatus({ code: SpanStatusCode.ERROR });
|
||||
pass.end(passEndedAt);
|
||||
} catch {
|
||||
// Tracing must never change reconciliation.
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user