diff --git a/scripts/benchmark-run-liveness-backfill.ts b/scripts/benchmark-run-liveness-backfill.ts new file mode 100644 index 00000000000..f822b8f1a3e --- /dev/null +++ b/scripts/benchmark-run-liveness-backfill.ts @@ -0,0 +1,176 @@ +// Local SQL benchmark for the run liveness backfill candidate lookup. +// Usage: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-run-liveness-backfill.ts --baseline-ref= +// The baseline query is the pre-fix builder from activity.ts, copied verbatim. +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { fileURLToPath } from "node:url"; +import { and, eq, isNull, or, sql } from "../server/node_modules/drizzle-orm/index.js"; +import { activityLog, agents, companies, createDb, heartbeatRuns, issues, startEmbeddedPostgresTestDatabase, type Db } + from "../packages/db/src/index.js"; +import { missingRunLivenessCandidatesQuery } from "../server/src/services/run-liveness-backfill-query.js"; + +type Query = { sql: string; params: unknown[] }; +const repoRoot = fileURLToPath(new URL("../", import.meta.url)); +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref"); +const git = (...args: string[]) => execFileSync("git", args, { cwd: repoRoot, encoding: "utf8" }).trim(); +const baselineRevision = git("rev-parse", "--verify", `${baselineRef}^{commit}`); +const baselineSource = git("show", `${baselineRevision}:server/src/services/activity.ts`); +assert.ok(baselineSource.includes("${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,\n sql`exists ("), + "Baseline revision must contain the pre-fix candidate query"); + +function baselineQuery(db: Db, companyId: string, issueId: string, limit: number) { + return db + .select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, status: heartbeatRuns.status, + contextSnapshot: heartbeatRuns.contextSnapshot, resultJson: heartbeatRuns.resultJson, + stdoutExcerpt: heartbeatRuns.stdoutExcerpt, stderrExcerpt: heartbeatRuns.stderrExcerpt, error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, continuationAttempt: heartbeatRuns.continuationAttempt, + }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + isNull(heartbeatRuns.livenessState), + sql`${heartbeatRuns.status} not in ('queued', 'running')`, + or( + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + sql`exists ( + select 1 + from ${activityLog} + where ${activityLog.companyId} = ${companyId} + and ${activityLog.entityType} = 'issue' + and ${activityLog.entityId} = ${issueId} + and ${activityLog.runId} = ${heartbeatRuns.id} + )`, + ), + )) + .limit(limit); +} + +const fixtures = [ + { name: "clean issue, many unclassified runs elsewhere", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 0 }, + { name: "issue with 3 missing, many unclassified runs elsewhere", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 3 }, + { name: "issue with 25 missing (more than one batch)", unclassifiedElsewhere: 6000, classified: 30000, activity: 300000, targetMissing: 25 }, + { name: "small control", unclassifiedElsewhere: 50, classified: 500, activity: 2000, targetMissing: 3 }, +]; + +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined; +try { + database = await startEmbeddedPostgresTestDatabase("run-liveness-backfill-benchmark-"); + db = createDb(database.connectionString); + const client = db.$client; + const [settings] = await client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit"); + const results = []; + for (const fixture of fixtures) { + await client.unsafe("set client_min_messages = warning"); + await client.unsafe("truncate activity_log, heartbeat_runs, issues, agents, companies cascade"); + const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); + const issueId = randomUUID(); + await db.insert(companies).values([ + { id: companyId, name: "Fixture", issuePrefix: `L${companyId.replaceAll("-", "").slice(0, 8)}` }, + { id: otherCompanyId, name: "Decoy", issuePrefix: `D${otherCompanyId.replaceAll("-", "").slice(0, 8)}` }, + ]); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }, + { id: otherAgentId, companyId: otherCompanyId, name: "Decoy", adapterType: "codex_local" }, + ]); + await db.insert(issues).values({ id: issueId, companyId, title: "Target", status: "in_progress", assigneeAgentId: agentId }); + // 400 other issue ids in the same company. + await client.unsafe(`insert into issues (id, company_id, title, status) + select gen_random_uuid(), $1, 'Other ' || g, 'todo' from generate_series(1, 400) g`, [companyId]); + // Wide, mostly incompressible context: hex digests, about 20 KiB each. + const blob = `(select string_agg(md5(random()::text || g || s), '') from generate_series(1, 640) s)`; + const otherIssues = `(select array_agg(id::text order by id) ids from issues where company_id = $1 and id::text <> $4::text)`; + // Classified runs across the company (the common case). + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot, created_at) + select $1, $2, 'succeeded', 'advanced', jsonb_build_object('issueId', other.ids[1 + g % 400]), + now() - (g || ' seconds')::interval + from generate_series(1, $3::int) g, ${otherIssues} other`, [companyId, agentId, fixture.classified, issueId]); + // Unclassified finished runs of other issues and of no issue. They never match the target issue. + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot, created_at) + select $1, $2, case when g % 3 = 0 then 'cancelled' when g % 3 = 1 then 'failed' else 'timed_out' end, null, + case when g % 4 = 0 then jsonb_build_object('blob', ${blob}) + else jsonb_build_object('issueId', other.ids[1 + g % 400], 'blob', ${blob}) end, + now() - (g || ' minutes')::interval + from generate_series(1, $3::int) g, ${otherIssues} other`, [companyId, agentId, fixture.unclassifiedElsewhere, issueId]); + // Target issue: classified history, missing rows via context, missing rows linked only by activity, + // one running run (excluded), and a cross-company decoy naming the same issue id. + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot) + select $1, $2, 'succeeded', 'completed', jsonb_build_object('issueId', $3::text, 'blob', ${blob}) from generate_series(1, 40) g`, + [companyId, agentId, issueId]); + const viaContext = Math.ceil(fixture.targetMissing / 2), viaActivity = fixture.targetMissing - viaContext; + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot) + select $1, $2, 'failed', null, jsonb_build_object('issueId', $3::text, 'blob', ${blob}) from generate_series(1, $4::int) g`, + [companyId, agentId, issueId, viaContext]); + await client.unsafe(`with runs as ( + insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot) + select $1, $2, 'cancelled', null, jsonb_build_object('blob', ${blob}) from generate_series(1, $4::int) g returning id) + insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id) + select $1, 'agent', 'issue.updated', 'issue', $3, id from runs`, [companyId, agentId, issueId, viaActivity]); + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot) + values ($1, $2, 'running', null, jsonb_build_object('issueId', $3::text))`, [companyId, agentId, issueId]); + await client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, liveness_state, context_snapshot) + values ($1, $2, 'failed', null, jsonb_build_object('issueId', $3::text))`, [otherCompanyId, otherAgentId, issueId]); + // Activity log: about half the rows carry a run id of a run that is not the target issue's. + // Issue entities spread over the other issues; the target issue has its own history too. + await client.unsafe(`insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id, created_at) + select $1, 'agent', 'issue.updated', 'issue', other.ids[1 + g % 400], + case when g % 2 = 0 then runs.ids[1 + g % array_length(runs.ids, 1)] end, + now() - (g || ' seconds')::interval + from generate_series(1, $2::int) g, ${otherIssues.replace("$4", "$3")} other, + (select array_agg(id order by id) ids from heartbeat_runs + where company_id = $1 and context_snapshot ->> 'issueId' is distinct from $3) runs`, + [companyId, fixture.activity, issueId]); + await client.unsafe(`insert into activity_log (company_id, actor_id, action, entity_type, entity_id, run_id) + select $1, 'agent', 'issue.comment_added', 'issue', $2, case when g % 2 = 0 then runs.ids[1 + g % 40] end + from generate_series(1, 200) g, (select array_agg(id order by id) ids from heartbeat_runs + where company_id = $1 and context_snapshot ->> 'issueId' = $2 and liveness_state = 'completed') runs`, + [companyId, issueId]); + await client.unsafe("vacuum analyze"); + + const before = baselineQuery(db, companyId, issueId, 20).toSQL() as Query; + const after = missingRunLivenessCandidatesQuery(db, companyId, issueId, 20).toSQL() as Query; + const execute = async (query: Query) => client.unsafe(query.sql, query.params as never[]).values(); + // Equivalence on the complete candidate set (no limit), every column, stable order. + const sortRows = (rows: unknown[][]) => rows.map(row => JSON.stringify(row)).sort(); + const fullBefore = sortRows(await execute(baselineQuery(db, companyId, issueId, 100000).toSQL() as Query)); + const fullAfter = sortRows(await execute(missingRunLivenessCandidatesQuery(db, companyId, issueId, 100000).toSQL() as Query)); + assert.deepEqual(fullAfter, fullBefore, "Full candidate set must match on every column"); + assert.equal(fullAfter.length, fixture.targetMissing); + const limitedAfter = sortRows(await execute(after)); + assert.equal(limitedAfter.length, Math.min(20, fixture.targetMissing)); + assert.ok(limitedAfter.every(row => fullBefore.includes(row)), "Limited batch must come from the baseline candidate set"); + + const samples = { baseline: [] as number[], treatment: [] as number[] }; + for (let round = 0; round < 13; round++) { + for (const arm of round % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const) { + const start = performance.now(); + await execute(arm === "baseline" ? before : after); + if (round >= 2) samples[arm].push(performance.now() - start); + } + } + const explain = async (query: Query) => { + const [row] = await client.unsafe(`explain (analyze, buffers, format json) ${query.sql}`, query.params as never[]); + const plan = (row!["QUERY PLAN"] as Array<{ Plan: Record; "Execution Time": number }>)[0]!; + return { executionMs: plan["Execution Time"], sharedHit: plan.Plan["Shared Hit Blocks"], sharedRead: plan.Plan["Shared Read Blocks"] }; + }; + const summarize = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + const median = sorted[Math.floor(sorted.length / 2)]!; + return { medianMs: +median.toFixed(3), minMs: +sorted[0]!.toFixed(3), maxMs: +sorted.at(-1)!.toFixed(3) }; + }; + const b = summarize(samples.baseline), t = summarize(samples.treatment); + results.push({ fixture, candidates: fullAfter.length, fullSetEquality: true, + baseline: { ...b, plan: await explain(before) }, treatment: { ...t, plan: await explain(after) }, + speedup: +(b.medianMs / t.medianMs).toFixed(1) }); + } + console.log(JSON.stringify({ baselineRevision, treatmentRevision: git("rev-parse", "HEAD"), + treatmentDirty: git("status", "--porcelain", "--", "server/src/services/activity.ts", "server/src/services/run-liveness-backfill-query.ts") !== "", + settings, node: process.version, warmupsPerArm: 2, measuredSamplesPerArm: 11, results }, null, 2)); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/activity-service.test.ts b/server/src/__tests__/activity-service.test.ts index d2ed798d5c7..713054457bf 100644 --- a/server/src/__tests__/activity-service.test.ts +++ b/server/src/__tests__/activity-service.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, agentWakeupRequests, @@ -693,4 +693,98 @@ describeEmbeddedPostgres("activity service", () => { lastUsefulActionAt: null, }); }); + + describe("run liveness backfill", () => { + async function seedIssue() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Backfill", issuePrefix: `B${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", adapterType: "codex_local" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Backfill target", status: "done", assigneeAgentId: agentId }); + return { companyId, agentId, issueId }; + } + + async function insertMissingRun(companyId: string, agentId: string, contextSnapshot: Record, status = "succeeded") { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status, contextSnapshot, livenessState: null }); + return runId; + } + + async function livenessOf(runIds: string[]) { + const rows = await db.select({ id: heartbeatRuns.id, livenessState: heartbeatRuns.livenessState }).from(heartbeatRuns); + return runIds.map((id) => rows.find((row) => row.id === id)?.livenessState ?? null); + } + + async function waitForLiveness(service: ActivityService, companyId: string, issueId: string, runIds: string[]) { + const deadline = performance.now() + 3_000; + while (performance.now() < deadline) { + await service.runsForIssue(companyId, issueId); + if ((await livenessOf(runIds)).every((state) => state !== null)) { + // Let the pass finish its bookkeeping after the last update. + await new Promise((resolve) => setTimeout(resolve, 50)); + return; + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error(`Timed out waiting for liveness: ${JSON.stringify(await livenessOf(runIds))}`); + } + + it("backfills only this issue's runs, found by context or by issue activity", async () => { + const { companyId, agentId, issueId } = await seedIssue(); + const viaContext = await insertMissingRun(companyId, agentId, { issueId }); + const viaActivity = await insertMissingRun(companyId, agentId, {}, "failed"); + await db.insert(activityLog).values({ companyId, actorType: "agent", actorId: agentId, action: "issue.updated", + entityType: "issue", entityId: issueId, runId: viaActivity }); + const running = await insertMissingRun(companyId, agentId, { issueId }, "running"); + const otherIssueRun = await insertMissingRun(companyId, agentId, { issueId: randomUUID() }); + const other = await seedIssue(); + const otherCompanyRun = await insertMissingRun(other.companyId, other.agentId, { issueId }); + + await waitForLiveness(activityService(db), companyId, issueId, [viaContext, viaActivity]); + + expect(await livenessOf([running, otherIssueRun, otherCompanyRun])).toEqual([null, null, null]); + }); + + it("does not repeat the lookup for a clean issue until the skip window ends", async () => { + const { companyId, agentId, issueId } = await seedIssue(); + const first = await insertMissingRun(companyId, agentId, { issueId }); + const service = activityService(db); + await waitForLiveness(service, companyId, issueId, [first]); + + const late = await insertMissingRun(companyId, agentId, { issueId }); + for (let read = 0; read < 3; read++) { + await service.runsForIssue(companyId, issueId); + await new Promise((resolve) => setTimeout(resolve, 50)); + } + expect(await livenessOf([late])).toEqual([null]); + + const now = Date.now(); + const clock = vi.spyOn(Date, "now").mockReturnValue(now + 11 * 60_000); + try { + await waitForLiveness(service, companyId, issueId, [late]); + } finally { + clock.mockRestore(); + } + }); + + it("keeps backfilling on later reads while an issue has more than one batch", async () => { + const { companyId, agentId, issueId } = await seedIssue(); + const runIds: string[] = []; + for (let index = 0; index < 25; index++) runIds.push(await insertMissingRun(companyId, agentId, { issueId })); + + await waitForLiveness(activityService(db), companyId, issueId, runIds); + }); + + it("leaves the backfill to the task page when a read skips it", async () => { + const { companyId, agentId, issueId } = await seedIssue(); + const runId = await insertMissingRun(companyId, agentId, { issueId }); + const service = activityService(db); + await service.runsForIssue(companyId, issueId, null, 2, undefined, { skipLivenessBackfill: true }); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(await livenessOf([runId])).toEqual([null]); + + await waitForLiveness(service, companyId, issueId, [runId]); + }); + }); }); diff --git a/server/src/services/activity.ts b/server/src/services/activity.ts index 36ed18b205e..c4975395ed8 100644 --- a/server/src/services/activity.ts +++ b/server/src/services/activity.ts @@ -24,6 +24,7 @@ import { hasWorkspaceRestoreFailure, safeWorkspaceRestorePath, ISSUE_CONTINUATIO import { logger } from "../middleware/logger.js"; import { visibleIssueCondition } from "./issue-visibility.js"; import { classifyRunLiveness } from "./run-liveness.js"; +import { missingRunLivenessCandidatesQuery } from "./run-liveness-backfill-query.js"; export interface ActivityFilters { companyId: string; @@ -43,8 +44,27 @@ export function normalizeActivityLimit(limit: number | undefined) { return Math.max(1, Math.min(MAX_ACTIVITY_LIMIT, Math.floor(limit ?? DEFAULT_ACTIVITY_LIMIT))); } +const RUN_LIVENESS_BACKFILL_BATCH = 20; +// Runs finalize with a liveness classification. The backfill only repairs rows +// that missed it (older rows or a failed finalization), so a delay is safe. +const RUN_LIVENESS_CLEAN_TTL_MS = 10 * 60_000; +const RUN_LIVENESS_CLEAN_MAX_ENTRIES = 10_000; + export function activityService(db: Db) { const scheduledLivenessBackfills = new Set(); + // Issues whose last backfill pass left no missing liveness, keyed like + // `scheduledLivenessBackfills`. Values are the time the skip window ends. + // The map keeps insertion order, so the first entry is the oldest. + const livenessCleanUntil = new Map(); + function rememberLivenessClean(key: string) { + livenessCleanUntil.delete(key); + livenessCleanUntil.set(key, Date.now() + RUN_LIVENESS_CLEAN_TTL_MS); + while (livenessCleanUntil.size > RUN_LIVENESS_CLEAN_MAX_ENTRIES) { + const oldest = livenessCleanUntil.keys().next().value; + if (oldest === undefined) break; + livenessCleanUntil.delete(oldest); + } + } const issueIdAsText = sql`${issues.id}::text`; const summarizedUsageJson = sql | null>` case @@ -192,41 +212,9 @@ export function activityService(db: Db) { } async function backfillMissingRunLivenessForIssue(companyId: string, issueId: string) { - const runs = await db - .select({ - id: heartbeatRuns.id, - companyId: heartbeatRuns.companyId, - status: heartbeatRuns.status, - contextSnapshot: heartbeatRuns.contextSnapshot, - resultJson: heartbeatRuns.resultJson, - stdoutExcerpt: heartbeatRuns.stdoutExcerpt, - stderrExcerpt: heartbeatRuns.stderrExcerpt, - error: heartbeatRuns.error, - errorCode: heartbeatRuns.errorCode, - continuationAttempt: heartbeatRuns.continuationAttempt, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, companyId), - isNull(heartbeatRuns.livenessState), - sql`${heartbeatRuns.status} not in ('queued', 'running')`, - or( - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, - sql`exists ( - select 1 - from ${activityLog} - where ${activityLog.companyId} = ${companyId} - and ${activityLog.entityType} = 'issue' - and ${activityLog.entityId} = ${issueId} - and ${activityLog.runId} = ${heartbeatRuns.id} - )`, - ), - ), - ) - .limit(20); + const runs = await missingRunLivenessCandidatesQuery(db, companyId, issueId, RUN_LIVENESS_BACKFILL_BATCH); - if (runs.length === 0) return; + if (runs.length === 0) return 0; const issue = await db .select({ @@ -357,13 +345,24 @@ export function activityService(db: Db) { }) .where(and(eq(heartbeatRuns.id, run.id), isNull(heartbeatRuns.livenessState))); } + return runs.length; } function scheduleRunLivenessBackfill(companyId: string, issueId: string) { const key = `${companyId}:${issueId}`; if (scheduledLivenessBackfills.has(key)) return; + const cleanUntil = livenessCleanUntil.get(key); + if (cleanUntil !== undefined) { + if (cleanUntil > Date.now()) return; + livenessCleanUntil.delete(key); + } scheduledLivenessBackfills.add(key); void backfillMissingRunLivenessForIssue(companyId, issueId) + .then((found) => { + // A partial batch means this pass classified every remaining candidate. + // Skip this issue for a while, so page reads do not repeat the lookup. + if (found < RUN_LIVENESS_BACKFILL_BATCH) rememberLivenessClean(key); + }) .catch((err: unknown) => { logger.warn({ err, companyId, issueId }, "run liveness backfill failed"); }) diff --git a/server/src/services/run-liveness-backfill-query.ts b/server/src/services/run-liveness-backfill-query.ts new file mode 100644 index 00000000000..a7fd2e10f22 --- /dev/null +++ b/server/src/services/run-liveness-backfill-query.ts @@ -0,0 +1,48 @@ +import { and, eq, isNull, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { activityLog, heartbeatRuns } from "@paperclipai/db"; + +/** + * Finished runs of one issue that have no liveness classification yet. + * A run belongs to the issue when its context names the issue or when an + * issue activity row records the run. Each branch reads only this issue's + * rows through its own index, so the cost does not grow with the number of + * unclassified runs in other issues of the company. + */ +export function missingRunLivenessCandidatesQuery(db: Db, companyId: string, issueId: string, limit: number) { + return db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + status: heartbeatRuns.status, + contextSnapshot: heartbeatRuns.contextSnapshot, + resultJson: heartbeatRuns.resultJson, + stdoutExcerpt: heartbeatRuns.stdoutExcerpt, + stderrExcerpt: heartbeatRuns.stderrExcerpt, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + continuationAttempt: heartbeatRuns.continuationAttempt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + isNull(heartbeatRuns.livenessState), + sql`${heartbeatRuns.status} not in ('queued', 'running')`, + sql`${heartbeatRuns.id} in ( + select issue_run.id + from ${heartbeatRuns} as issue_run + where issue_run.company_id = ${companyId} + and issue_run.context_snapshot ->> 'issueId' = ${issueId} + union all + select ${activityLog.runId} + from ${activityLog} + where ${activityLog.companyId} = ${companyId} + and ${activityLog.entityType} = 'issue' + and ${activityLog.entityId} = ${issueId} + and ${activityLog.runId} is not null + )`, + ), + ) + .limit(limit); +}