diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 80f140e0264..2ab0bbb301c 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -1596,6 +1596,18 @@ Lifecycle polls can page past failed resources using the last returned event id cursor, never a persisted high-water mark. This retries failures and includes transactions that commit later with lower ids. +A lifecycle page reads the company's unacknowledged events for the requested +types once. It keeps one head for each resource: the unacknowledged creation, +else the oldest unacknowledged event. Then it applies `afterId`, id order and +the limit. Cost grows with the company's journal size, not with pending events +times acknowledged history. The idle-sleep check only asks whether any +unacknowledged event exists, because each such resource has a head. No index +or migration is needed. + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=02bfe71a6b +``` + ## Legacy controller ownership Legacy run claims atomically record `controller_boot_id`, a database-clock diff --git a/scripts/benchmark-plugin-lifecycle-inbox.ts b/scripts/benchmark-plugin-lifecycle-inbox.ts new file mode 100644 index 00000000000..6ebd8e26909 --- /dev/null +++ b/scripts/benchmark-plugin-lifecycle-inbox.ts @@ -0,0 +1,127 @@ +// Compares the plugin lifecycle inbox page query before and after a change. +// Usage: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts \ +// --baseline-ref= [--agents=40] [--per-agent=2500] [--acked=2000] [--samples=5] [--treatment-only] +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createRequire } from "node:module"; +import * as schema from "../packages/db/src/schema/index.js"; +import { companies, plugins, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import * as current from "../server/src/services/plugin-lifecycle-inbox.js"; + +const arg = (name: string, fallback: string) => process.argv.find(a => a.startsWith(`--${name}=`))?.slice(name.length + 3) ?? fallback; +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${arg("baseline-ref", "origin/master")}^{commit}`], { encoding: "utf8" }).trim(); +const agentCount = Number(arg("agents", "40")); +const perAgent = Number(arg("per-agent", "2500")); +const acked = Number(arg("acked", "2000")); +const sampleCount = Number(arg("samples", "5")); +const timeoutMs = Number(arg("timeout-ms", "600000")); +// The baseline is quadratic; at large fixtures time only the treatment. +const treatmentOnly = process.argv.includes("--treatment-only"); +// Resolve database drivers through the db package; the scripts folder has no own dependencies. +const dbRequire = createRequire(new URL("../packages/db/package.json", import.meta.url)); +const postgres = (await import(pathToFileURL(dbRequire.resolve("postgres")).href)).default as typeof import("postgres").default; +const { drizzle } = await import(pathToFileURL(dbRequire.resolve("drizzle-orm/postgres-js")).href) as typeof import("drizzle-orm/postgres-js"); +type Inbox = typeof current.pluginLifecycleInbox; +type Query = { sql: string; params: unknown[] }; + +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let client: import("postgres").Sql | undefined; +let baselineDir: string | undefined; +try { + database = await startEmbeddedPostgresTestDatabase("lifecycle-inbox-benchmark-"); + let recorded: Query[] = []; + client = postgres(database.connectionString, { + max: 1, onnotice: () => {}, connection: { statement_timeout: timeoutMs }, + debug: (_connection, query, params) => { recorded.push({ sql: query, params }); }, + }); + const db = drizzle(client, { schema }) as unknown as Db; + + baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".lifecycle-baseline-")); + const modulePath = "server/src/services/plugin-lifecycle-inbox.ts"; + const original = new URL(`../${modulePath}`, import.meta.url); + const source = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" }) + .replace(/from "(\.{1,2}\/[^"]+)"/g, (_m, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, original)))}`); + const target = path.join(baselineDir, "plugin-lifecycle-inbox.ts"); + await fs.writeFile(target, source); + const baseline = (await import(pathToFileURL(target).href)).pluginLifecycleInbox as Inbox; + + // Fixture: agent_runtime history interleaved across agents. Each agent has `acked` + // acknowledged revisions followed by a backlog of unacknowledged revisions. + const companyId = randomUUID(), decoyId = randomUUID(); + await db.insert(companies).values([{ id: companyId, name: "Bench", issuePrefix: "BNCH" }, { id: decoyId, name: "Decoy", issuePrefix: "DCOY" }]); + const [plugin] = await db.insert(plugins).values({ pluginKey: randomUUID(), packageName: "bench", version: "1.0.0", status: "ready", manifestJson: {} as never }).returning(); + const agentIds = Array.from({ length: agentCount }, () => randomUUID()); + const projectIds = Array.from({ length: 20 }, () => randomUUID()); + await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + SELECT $1::uuid, 'agent', a, 'create' FROM unnest($2::uuid[]) a + UNION ALL SELECT $1::uuid, 'project', p, 'create' FROM unnest($3::uuid[]) p`, [companyId, agentIds, projectIds]); + await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + SELECT $1::uuid, 'agent_runtime', ($2::uuid[])[(n % $3) + 1], 'update' FROM generate_series(0, $4::int - 1) n ORDER BY n`, + [companyId, agentIds, agentCount, agentCount * perAgent]); + // A few project updates after the agent backlog, so heads are spread across the id range. + await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + SELECT $1::uuid, 'project', p, 'update' FROM unnest($2::uuid[]) p`, [companyId, projectIds]); + await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + SELECT $1::uuid, 'agent_runtime', gen_random_uuid(), 'update' FROM generate_series(1, 20000)`, [decoyId]); + await client.unsafe(`INSERT INTO plugin_lifecycle_acknowledgments (plugin_id, event_id) + SELECT $1::uuid, id FROM ( + SELECT id, resource_type, row_number() OVER (PARTITION BY resource_type, resource_id ORDER BY id) AS rank + FROM resource_lifecycle_events WHERE company_id = $2::uuid) e + WHERE resource_type = 'agent' OR (resource_type = 'project' AND rank = 1) OR (resource_type = 'agent_runtime' AND rank <= $3)`, + [plugin.id, companyId, acked]); + await client.unsafe("ANALYZE"); + const [{ events }] = await client.unsafe(`SELECT count(*)::int AS events FROM resource_lifecycle_events`); + const [{ acks }] = await client.unsafe(`SELECT count(*)::int AS acks FROM plugin_lifecycle_acknowledgments`); + const [{ server_version: version }] = await client.unsafe(`SHOW server_version`); + + const cases = [ + { name: "page-100-all-types", limit: 100, types: ["agent", "project", "agent_runtime"] }, + { name: "page-100-agent-project", limit: 100, types: ["agent", "project"] }, + // Idle-sleep check: baseline read one listed event; the treatment asks only whether one exists. + { name: "idle-probe-all-types", limit: 1, types: ["agent", "project", "agent_runtime"], probe: true }, + ] as const; + async function run(inbox: Inbox, c: (typeof cases)[number]) { + recorded = []; + const start = performance.now(); + const service = inbox(db, plugin.id) as ReturnType; + const rows = "probe" in c + ? [("hasPending" in service ? await service.hasPending(companyId, [...c.types]) : (await service.list(companyId, 1, undefined, [...c.types])).length > 0)] + : await service.list(companyId, c.limit, undefined, [...c.types]); + const ms = performance.now() - start; + const query = recorded.find(q => q.sql.includes('"resource_lifecycle_events"')); + return { ms, rows, query }; + } + const results = []; + for (const c of cases) { + const samples = { baseline: [] as number[], treatment: [] as number[] }; + let baseRows: unknown, treatRows: unknown, baseQuery: Query | undefined, treatQuery: Query | undefined; + // One warmup each, then alternate order per sample to limit cache bias. + for (let n = 0; n <= sampleCount; n++) { + const order = n % 2 === 0 ? (["baseline", "treatment"] as const) : (["treatment", "baseline"] as const); + for (const which of order.filter(w => !treatmentOnly || w === "treatment")) { + const r = await run(which === "baseline" ? baseline : current.pluginLifecycleInbox, c); + if (n > 0) samples[which].push(r.ms); + if (which === "baseline") { baseRows = r.rows; baseQuery = r.query; } else { treatRows = r.rows; treatQuery = r.query; } + } + } + if (!treatmentOnly) assert.deepEqual(treatRows, baseRows, `${c.name}: results differ`); + const explain = async (q?: Query) => q ? (await client!.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) ${q.sql}`, q.params as never[])).map(r => r["QUERY PLAN"]).join("\n") : ""; + const stats = (s: number[]) => { const v = [...s].sort((a, b) => a - b); return { median: v[Math.floor(v.length / 2)], min: v[0], max: v.at(-1), samples: s.map(x => Math.round(x * 10) / 10) }; }; + const b = treatmentOnly ? undefined : stats(samples.baseline), t = stats(samples.treatment); + results.push({ case: c.name, rows: (treatRows as unknown[]).length, equal: !treatmentOnly, baselineMs: b, treatmentMs: t, speedup: b ? Math.round(b.median / t.median * 10) / 10 : undefined, + baselinePlan: await explain(baseQuery), treatmentPlan: await explain(treatQuery) }); + } + // Control: the agent runtime revision read (plugin-agent-runtime-state.ts, agent-file-store.ts) is unchanged. + const revisionPlan = (await client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) SELECT id FROM resource_lifecycle_events + WHERE company_id = $1 AND resource_type = 'agent_runtime' AND resource_id = $2 ORDER BY id DESC LIMIT 1`, [companyId, agentIds[0]])) + .map(r => r["QUERY PLAN"]).join("\n"); + console.log(JSON.stringify({ revisionPlan, baselineRevision, postgres: version, agents: agentCount, perAgent, acked, events, acks, samples: sampleCount, results }, null, 2)); +} finally { + try { await client?.end({ timeout: 0 }); await database?.cleanup(); if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/resource-lifecycle-events.test.ts b/server/src/__tests__/resource-lifecycle-events.test.ts index 3d58050664b..0a98414630d 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -462,6 +462,53 @@ describePostgres("Resource lifecycle events", () => { await expect(inbox.list(companyId, 100, "invalid")).rejects.toMatchObject({ status: 400 }); }); + it("lists exactly one head per resource across acknowledged history, late creations, gaps, and pages", async () => { + const plugin = await createPlugin(); + const other = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + const decoyCompanyId = randomUUID(); + await db.insert(companies).values({ id: decoyCompanyId, name: "Lifecycle decoy", issuePrefix: `D${decoyCompanyId.replaceAll("-", "").slice(0, 6)}` }); + const agentId = randomUUID(), lateCreateId = randomUUID(), gapId = randomUUID(), doneId = randomUUID(), projectId = randomUUID(); + type Row = typeof resourceLifecycleEvents.$inferInsert; + const insert = async (rows: Row[]) => db.insert(resourceLifecycleEvents).values(rows).returning(); + const ack = async (ids: number[], pluginId = plugin.id) => { + if (ids.length) await db.insert(pluginLifecycleAcknowledgments).values(ids.map(eventId => ({ pluginId, eventId }))); + }; + const [agentCreate] = await insert([{ companyId, resourceType: "agent", resourceId: agentId, action: "create" }]); + // The same id as a different resource type is an independent stream. + const runtime = await insert(Array.from({ length: 12 }, () => ({ companyId, resourceType: "agent_runtime" as const, resourceId: agentId, action: "update" as const }))); + const [latePause] = await insert([{ companyId, resourceType: "agent", resourceId: lateCreateId, action: "pause" }]); + const gap = await insert([0, 1, 2].map(() => ({ companyId, resourceType: "project" as const, resourceId: gapId, action: "update" as const }))); + const done = await insert([{ companyId, resourceType: "agent", resourceId: doneId, action: "create" }, { companyId, resourceType: "agent", resourceId: doneId, action: "pause" }]); + const [lateCreate] = await insert([{ companyId, resourceType: "agent", resourceId: lateCreateId, action: "create" }]); + const [projectCreate, projectArchive] = await insert([ + { companyId, resourceType: "project", resourceId: projectId, action: "create" }, { companyId, resourceType: "project", resourceId: projectId, action: "archive" }]); + await insert([{ companyId: decoyCompanyId, resourceType: "agent", resourceId: randomUUID(), action: "create" }]); + await ack([agentCreate.id, ...runtime.slice(0, 9).map(e => e.id), gap[1].id, ...done.map(e => e.id), projectCreate.id]); + await ack(runtime.map(e => e.id), other.id); + + const all = ["agent", "project", "agent_runtime"] as const; + expect((await inbox.list(companyId, 100, undefined, [...all])).map(e => Number(e.id))).toEqual( + [runtime[9].id, gap[0].id, lateCreate.id, projectArchive.id].sort((a, b) => a - b)); + expect((await inbox.list(companyId)).map(e => [e.resourceId, e.action])).toEqual( + [[gapId, "update"], [lateCreateId, "create"], [projectId, "archive"]]); + expect(await inbox.list(companyId, 2, undefined, [...all])).toEqual((await inbox.list(companyId, 100, undefined, [...all])).slice(0, 2)); + expect((await inbox.list(companyId, 100, String(gap[0].id), [...all])).map(e => Number(e.id))).toEqual([lateCreate.id, projectArchive.id]); + expect(await pluginLifecycleInbox(db, other.id).list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]); + expect(await pluginLifecycleInbox(db, other.id).hasPending(companyId, ["agent_runtime"])).toBe(false); + expect(await inbox.hasPending(companyId, ["agent_runtime"])).toBe(true); + expect(await inbox.hasPending(companyId)).toBe(true); + expect(await inbox.list(decoyCompanyId, 100, undefined, [...all])).toEqual([expect.objectContaining({ companyId: decoyCompanyId })]); + const [head] = await inbox.list(companyId, 1, undefined, ["project"]); + expect(head).toEqual({ id: String(gap[0].id), companyId, resourceType: "project", resourceId: gapId, action: "update", createdAt: gap[0].createdAt.toISOString() }); + + // Acknowledging a head exposes that resource's next unacknowledged event, skipping acknowledged gaps. + await inbox.acknowledge(companyId, String(gap[0].id)); + await inbox.acknowledge(companyId, String(lateCreate.id)); + expect((await inbox.list(companyId)).map(e => Number(e.id))).toEqual([latePause.id, gap[2].id, projectArchive.id].sort((a, b) => a - b)); + await expect(inbox.acknowledge(companyId, String(runtime[10].id))).rejects.toMatchObject({ status: 409 }); + }); + it("never skips a lower event id that commits after a higher id is acknowledged", async () => { const plugin = await createPlugin(); const inbox = pluginLifecycleInbox(db, plugin.id); diff --git a/server/src/services/plugin-lifecycle-delivery.ts b/server/src/services/plugin-lifecycle-delivery.ts index f765865a1ad..0db0a038e01 100644 --- a/server/src/services/plugin-lifecycle-delivery.ts +++ b/server/src/services/plugin-lifecycle-delivery.ts @@ -236,9 +236,9 @@ export function registerPluginLifecycleDelivery( if (!current()) return "unknown"; if (!types) continue; const { plugin, companyId } = target; - const events = await pluginLifecycleInbox(db, plugin.id).list(companyId, 1, undefined, types); + const hasEvents = await pluginLifecycleInbox(db, plugin.id).hasPending(companyId, types); if (!current()) return "unknown"; - if (events.length) { wake(); inspectionFailure = false; return "present"; } + if (hasEvents) { wake(); inspectionFailure = false; return "present"; } } inspectionFailure = false; return "none"; diff --git a/server/src/services/plugin-lifecycle-inbox.ts b/server/src/services/plugin-lifecycle-inbox.ts index 44bc8c290ea..9872bb5f7bd 100644 --- a/server/src/services/plugin-lifecycle-inbox.ts +++ b/server/src/services/plugin-lifecycle-inbox.ts @@ -38,11 +38,26 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) { if (!Number.isInteger(limit) || limit < 1 || limit > 100) throw badRequest("limit must be an integer from 1 to 100"); if (afterId !== undefined && (typeof afterId !== "string" || !/^[1-9]\d*$/.test(afterId) || !Number.isSafeInteger(Number(afterId)))) throw badRequest("Invalid lifecycle page id"); await assertAvailable(db, companyId); - const rows = await db.select().from(resourceLifecycleEvents) - .where(and(eq(resourceLifecycleEvents.companyId, companyId), unacknowledged, firstForResource, inArray(resourceLifecycleEvents.resourceType, resourceTypes ?? ["agent", "project"]), afterId === undefined ? undefined : gt(resourceLifecycleEvents.id, Number(afterId)))) - .orderBy(asc(resourceLifecycleEvents.id)).limit(limit); + // One pass picks each resource's deliverable head: its unacknowledged creation, else its + // oldest unacknowledged event. This is the set `firstForResource` admits, without probing + // every acknowledged older event once per pending candidate. The page filter stays outside. + const heads = db.selectDistinctOn([resourceLifecycleEvents.resourceType, resourceLifecycleEvents.resourceId]).from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), inArray(resourceLifecycleEvents.resourceType, resourceTypes ?? ["agent", "project"]), unacknowledged)) + .orderBy(resourceLifecycleEvents.resourceType, resourceLifecycleEvents.resourceId, sql`${resourceLifecycleEvents.action} <> 'create'`, resourceLifecycleEvents.id) + .as("heads"); + const rows = await db.select().from(heads) + .where(afterId === undefined ? undefined : gt(heads.id, Number(afterId))) + .orderBy(asc(heads.id)).limit(limit); return rows.map(row => ({ ...row, id: String(row.id), createdAt: row.createdAt.toISOString() })) as ResourceLifecycleEvent[]; }, + /** Whether `list` would return any event. Every resource with an unacknowledged event has a head. */ + async hasPending(companyId: string, resourceTypes?: ResourceLifecycleEvent["resourceType"][]): Promise { + await assertAvailable(db, companyId); + const [row] = await db.select({ id: resourceLifecycleEvents.id }).from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), inArray(resourceLifecycleEvents.resourceType, resourceTypes ?? ["agent", "project"]), unacknowledged)) + .orderBy(asc(resourceLifecycleEvents.id)).limit(1); + return row !== undefined; + }, async acknowledge(companyId: string, eventId: string, credentials?: { snapshot: ProjectSecretSnapshot; changes: unknown }): Promise { if (typeof eventId !== "string" || !/^[1-9]\d*$/.test(eventId) || !Number.isSafeInteger(Number(eventId))) throw badRequest("Invalid lifecycle event id"); const commit = () => db.transaction(async tx => {