diff --git a/doc/DATABASE.md b/doc/DATABASE.md index bcd663b1803..c5822df0f02 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -1798,16 +1798,36 @@ 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. +`plugin_lifecycle_resource_cursors` keeps one row for each plugin and resource +stream. Its `acknowledged_through_id` says that every event of that resource at +or below the value is acknowledged. Acknowledgments above the cursor use +`plugin_lifecycle_acknowledgments`. Acknowledgment writes both records, so an +earlier release reads the same progress after a rollback. The cursor moves only +over a contiguous acknowledged prefix. A creation recorded after transitions +that are still pending does not move it. Writers record a resource's events +while they hold that resource's row lock. Thus an event that commits later for +the same resource always has a higher id than the cursor. A lower id from a +different resource still stays deliverable. + +A lifecycle page visits each resource stream of the requested types once with a +backward loose index scan, which also gives the stream's latest event id. A +stream whose latest event is at or below its cursor is settled and is not read +again. For each other stream the page reads one head: the unacknowledged +creation, else the first unacknowledged transition above the cursor. Then it +applies `afterId`, id order and the limit. Cost grows with the number of +resource streams and pending events, not with the journal size. Acknowledgment +reads only the events between the cursor and the acknowledged event. The +idle-sleep check asks whether any stream has a head. + +Migration `0462` creates the cursor table and carries existing acknowledgments +forward. For each plugin and resource it stores the end of the contiguous +acknowledged prefix of events. A cursor never passes an unacknowledged event, so +a gap is still delivered. Acknowledged events after the gap are still skipped +through their acknowledgment rows. An earlier release that runs during +a rolling deploy writes only acknowledgment rows, and the new reads honor them. ```sh -pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=02bfe71a6b +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=8d2bf78323 --agents=400 --per-agent=2500 --acked=2000 ``` ## Legacy controller ownership diff --git a/packages/db/src/migrations/0462_plugin_lifecycle_resource_cursors.sql b/packages/db/src/migrations/0462_plugin_lifecycle_resource_cursors.sql new file mode 100644 index 00000000000..e9ba8e6f51b --- /dev/null +++ b/packages/db/src/migrations/0462_plugin_lifecycle_resource_cursors.sql @@ -0,0 +1,31 @@ +CREATE TABLE "plugin_lifecycle_resource_cursors" ( + "plugin_id" uuid NOT NULL, + "company_id" uuid NOT NULL, + "resource_type" text NOT NULL, + "resource_id" uuid NOT NULL, + "acknowledged_through_id" bigint NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "plugin_lifecycle_resource_cursors_pk" PRIMARY KEY("plugin_id","company_id","resource_type","resource_id") +); +--> statement-breakpoint +ALTER TABLE "plugin_lifecycle_resource_cursors" ADD CONSTRAINT "plugin_lifecycle_resource_cursors_plugin_id_plugins_id_fk" FOREIGN KEY ("plugin_id") REFERENCES "public"."plugins"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "plugin_lifecycle_resource_cursors" ADD CONSTRAINT "plugin_lifecycle_resource_cursors_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +-- Carry acknowledgments forward. Each cursor ends at the contiguous acknowledged prefix of a +-- resource's events, so an unacknowledged gap or late creation is still delivered, and +-- acknowledged events after it are still skipped through their acknowledgment rows. +INSERT INTO "plugin_lifecycle_resource_cursors" ("plugin_id", "company_id", "resource_type", "resource_id", "acknowledged_through_id") +SELECT progress."plugin_id", progress."company_id", progress."resource_type", progress."resource_id", + COALESCE(( + SELECT min(pending."id") - 1 FROM "resource_lifecycle_events" pending + WHERE pending."company_id" = progress."company_id" AND pending."resource_type" = progress."resource_type" + AND pending."resource_id" = progress."resource_id" + AND NOT EXISTS (SELECT 1 FROM "plugin_lifecycle_acknowledgments" ack + WHERE ack."plugin_id" = progress."plugin_id" AND ack."event_id" = pending."id") + ), progress."last_acknowledged_id") +FROM ( + SELECT ack."plugin_id", acked_event."company_id", acked_event."resource_type", acked_event."resource_id", max(acked_event."id") AS "last_acknowledged_id" + FROM "plugin_lifecycle_acknowledgments" ack + JOIN "resource_lifecycle_events" acked_event ON acked_event."id" = ack."event_id" + GROUP BY ack."plugin_id", acked_event."company_id", acked_event."resource_type", acked_event."resource_id" +) progress +ON CONFLICT DO NOTHING; diff --git a/packages/db/src/migrations/meta/0457_snapshot.json b/packages/db/src/migrations/meta/0462_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0457_snapshot.json rename to packages/db/src/migrations/meta/0462_snapshot.json index 7372eea79c5..e374e515147 100644 --- a/packages/db/src/migrations/meta/0457_snapshot.json +++ b/packages/db/src/migrations/meta/0462_snapshot.json @@ -1,6 +1,6 @@ { - "id": "109f5f96-1597-40da-9e08-6ed488f8ad27", - "prevId": "a2606b41-36c2-4eb5-a59d-28cc730bc455", + "id": "0287d884-1fee-4e8d-9ae0-2be0aac45b71", + "prevId": "f409684f-c629-4648-9123-a5e1064308ef", "version": "7", "dialect": "postgresql", "tables": { @@ -10188,6 +10188,40 @@ "method": "btree", "with": {} }, + "chat_actions_company_issue_kind_idx": { + "name": "chat_actions_company_issue_kind_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\"->>'issueId')", + "asc": true, + "isExpression": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" IN ('inbound_wakeup', 'failed_run_retry')", + "concurrently": true, + "method": "btree", + "with": {} + }, "chat_actions_github_stale_head_idx": { "name": "chat_actions_github_stale_head_idx", "columns": [ @@ -29133,6 +29167,22 @@ "concurrently": false, "method": "btree", "with": {} + }, + "environment_leases_native_backup_scope_idx": { + "name": "environment_leases_native_backup_scope_idx", + "columns": [ + { + "expression": "(\"metadata\"->'nativeHarnessBackup'->>'sessionScopeSha256')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "(\"environment_leases\".\"metadata\"->'nativeHarnessBackup'->>'sessionScopeSha256') is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35967,6 +36017,28 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_ctx_execution_workspace_idx": { + "name": "heartbeat_runs_company_ctx_execution_workspace_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"context_snapshot\" ->> 'executionWorkspaceId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"context_snapshot\" ? 'executionWorkspaceId'", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_company_ctx_paperclip_issue_created_idx": { "name": "heartbeat_runs_company_ctx_paperclip_issue_created_idx", "columns": [ @@ -43508,6 +43580,28 @@ "method": "btree", "with": {} }, + "issues_company_legacy_origin_run_idx": { + "name": "issues_company_legacy_origin_run_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "origin_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"issues\".\"origin_issue_id\" is null and \"issues\".\"origin_run_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} + }, "issues_company_project_workspace_idx": { "name": "issues_company_project_workspace_idx", "columns": [ @@ -49013,6 +49107,93 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.plugin_lifecycle_resource_cursors": { + "name": "plugin_lifecycle_resource_cursors", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "resource_type": { + "name": "resource_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "resource_id": { + "name": "resource_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "acknowledged_through_id": { + "name": "acknowledged_through_id", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "plugin_lifecycle_resource_cursors_plugin_id_plugins_id_fk": { + "name": "plugin_lifecycle_resource_cursors_plugin_id_plugins_id_fk", + "tableFrom": "plugin_lifecycle_resource_cursors", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_lifecycle_resource_cursors_company_id_companies_id_fk": { + "name": "plugin_lifecycle_resource_cursors_company_id_companies_id_fk", + "tableFrom": "plugin_lifecycle_resource_cursors", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "plugin_lifecycle_resource_cursors_pk": { + "name": "plugin_lifecycle_resource_cursors_pk", + "columns": [ + "plugin_id", + "company_id", + "resource_type", + "resource_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.plugin_logs": { "name": "plugin_logs", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index dde23060e66..d259e9c0e83 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3221,6 +3221,13 @@ "when": 1794334724911, "tag": "0461_created_from_legacy_origin_run_index", "breakpoints": true + }, + { + "idx": 462, + "version": "7", + "when": 1794335976242, + "tag": "0462_plugin_lifecycle_resource_cursors", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index 33ebbd85998..7812dfde547 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -221,6 +221,7 @@ export { runIdentityContexts } from "./run_identity_contexts.js"; export { connectionIntentDeliveries } from "./connection_intent_deliveries.js"; export { resourceLifecycleEvents } from "./resource_lifecycle_events.js"; export { pluginLifecycleAcknowledgments } from "./plugin_lifecycle_acknowledgments.js"; +export { pluginLifecycleResourceCursors } from "./plugin_lifecycle_resource_cursors.js"; export { toolActionDeliveries } from "./tool_action_deliveries.js"; export { chatTeamsFileTransfers } from "./chat_teams_file_transfers.js"; diff --git a/packages/db/src/schema/plugin_lifecycle_resource_cursors.ts b/packages/db/src/schema/plugin_lifecycle_resource_cursors.ts new file mode 100644 index 00000000000..3ee43e1b75e --- /dev/null +++ b/packages/db/src/schema/plugin_lifecycle_resource_cursors.ts @@ -0,0 +1,20 @@ +import { bigint, pgTable, primaryKey, text, timestamp, uuid } from "drizzle-orm/pg-core"; +import { companies } from "./companies.js"; +import { plugins } from "./plugins.js"; + +/** + * Per-plugin delivery progress for one lifecycle resource stream. Every event of the resource with + * an id at or below `acknowledgedThroughId` is acknowledged by the plugin. Acknowledgments above + * the cursor still use `plugin_lifecycle_acknowledgments`, which remains the record of each + * acknowledgment. + */ +export const pluginLifecycleResourceCursors = pgTable("plugin_lifecycle_resource_cursors", { + pluginId: uuid("plugin_id").notNull().references(() => plugins.id, { onDelete: "cascade" }), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + resourceType: text("resource_type").$type<"agent" | "project" | "agent_runtime">().notNull(), + resourceId: uuid("resource_id").notNull(), + acknowledgedThroughId: bigint("acknowledged_through_id", { mode: "number" }).notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), +}, table => ({ + pk: primaryKey({ name: "plugin_lifecycle_resource_cursors_pk", columns: [table.pluginId, table.companyId, table.resourceType, table.resourceId] }), +})); diff --git a/scripts/benchmark-plugin-lifecycle-inbox.ts b/scripts/benchmark-plugin-lifecycle-inbox.ts index 6ebd8e26909..0b4c0dafcef 100644 --- a/scripts/benchmark-plugin-lifecycle-inbox.ts +++ b/scripts/benchmark-plugin-lifecycle-inbox.ts @@ -1,4 +1,4 @@ -// Compares the plugin lifecycle inbox page query before and after a change. +// Compares the plugin lifecycle inbox page, idle probe and acknowledgment 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"; @@ -74,7 +74,15 @@ try { 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"); + // Upgrade path: carry the acknowledgment rows into resource cursors as the migration does. + const cursorMigration = await fs.readFile(new URL("../packages/db/src/migrations/0462_plugin_lifecycle_resource_cursors.sql", import.meta.url), "utf8"); + const backfill = cursorMigration.split("--> statement-breakpoint").find(statement => statement.includes('INSERT INTO "plugin_lifecycle_resource_cursors"')); + assert.ok(backfill, "cursor backfill statement not found"); + const backfillStart = performance.now(); + await client.unsafe(backfill); + const backfillMs = Math.round(performance.now() - backfillStart); + // Hosted tables are autovacuumed, so index-only scans skip most heap visits. + await client.unsafe("VACUUM 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`); @@ -116,11 +124,39 @@ try { 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) }); } + // Acknowledge one resource's head, then undo it, so every sample sees the same state. + { + const ackSamples = { baseline: [] as number[], treatment: [] as number[] }; + const [head] = await current.pluginLifecycleInbox(db, plugin.id).list(companyId, 1, undefined, ["agent_runtime"]); + // A fresh plugin (--acked=0) has no cursor row; restore that absence between samples. + const [cursorRow] = await client.unsafe(`SELECT acknowledged_through_id::text AS through FROM plugin_lifecycle_resource_cursors + WHERE plugin_id = $1 AND company_id = $2 AND resource_type = 'agent_runtime' AND resource_id = $3`, [plugin.id, companyId, head.resourceId]); + const through: string | undefined = cursorRow?.through; + 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 start = performance.now(); + await (which === "baseline" ? baseline : current.pluginLifecycleInbox)(db, plugin.id).acknowledge(companyId, head.id); + if (n > 0) ackSamples[which].push(performance.now() - start); + await client.unsafe(`DELETE FROM plugin_lifecycle_acknowledgments WHERE plugin_id = $1 AND event_id = $2`, [plugin.id, head.id]); + if (through === undefined) { + await client.unsafe(`DELETE FROM plugin_lifecycle_resource_cursors + WHERE plugin_id = $1 AND company_id = $2 AND resource_type = 'agent_runtime' AND resource_id = $3`, [plugin.id, companyId, head.resourceId]); + } else { + await client.unsafe(`UPDATE plugin_lifecycle_resource_cursors SET acknowledged_through_id = $4 + WHERE plugin_id = $1 AND company_id = $2 AND resource_type = 'agent_runtime' AND resource_id = $3`, [plugin.id, companyId, head.resourceId, through]); + } + } + } + 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(ackSamples.baseline), t = stats(ackSamples.treatment); + results.push({ case: "acknowledge-head", rows: 1, equal: !treatmentOnly, baselineMs: b, treatmentMs: t, speedup: b ? Math.round(b.median / t.median * 10) / 10 : undefined, baselinePlan: "", treatmentPlan: "" }); + } // 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)); + console.log(JSON.stringify({ revisionPlan, baselineRevision, postgres: version, agents: agentCount, perAgent, acked, events, acks, backfillMs, 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 dba0da19fb1..d5addd9741a 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -1,8 +1,8 @@ import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent-lifecycle.js"; import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; -import { and, eq, ne, sql } from "drizzle-orm"; -import { agentApiKeys, agents, budgetPolicies, companies, costEvents, createDb, pluginCompanySettings, pluginLifecycleAcknowledgments, plugins, projects, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import { and, eq, inArray, ne, sql } from "drizzle-orm"; +import { agentApiKeys, agents, budgetPolicies, companies, costEvents, createDb, pluginCompanySettings, pluginLifecycleAcknowledgments, pluginLifecycleResourceCursors, plugins, projects, resourceLifecycleEvents, type Db } from "@paperclipai/db"; import { createHostClientHandlers } from "@paperclipai/plugin-sdk"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -547,6 +547,140 @@ describePostgres("Resource lifecycle events", () => { await expect(inbox.acknowledge(companyId, String(runtime[10].id))).rejects.toMatchObject({ status: 409 }); }); + it("carries acknowledgments from an earlier release into resource cursors without redelivery or loss", async () => { + const plugin = await createPlugin(); + const other = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + const all = ["agent", "project", "agent_runtime"] as const; + const agentId = randomUUID(), gapId = randomUUID(), runtimeId = randomUUID(), createdOnlyId = randomUUID(); + type Row = typeof resourceLifecycleEvents.$inferInsert; + const insert = async (rows: Row[]) => db.insert(resourceLifecycleEvents).values(rows).returning(); + // An earlier release wrote only per-event acknowledgment rows. + const ack = async (ids: number[], pluginId = plugin.id) => { + if (ids.length) await db.insert(pluginLifecycleAcknowledgments).values(ids.map(eventId => ({ pluginId, eventId }))); + }; + const [agentCreate, ...agentTransitions] = await insert([ + { companyId, resourceType: "agent", resourceId: agentId, action: "create" }, + ...Array.from({ length: 6 }, (_, index) => ({ companyId, resourceType: "agent" as const, resourceId: agentId, action: index % 2 ? "resume" as const : "pause" as const })), + ]); + const gap = await insert([0, 1, 2].map(() => ({ companyId, resourceType: "project" as const, resourceId: gapId, action: "update" as const }))); + const runtime = await insert(Array.from({ length: 5 }, () => ({ companyId, resourceType: "agent_runtime" as const, resourceId: runtimeId, action: "update" as const }))); + const [createdOnly] = await insert([{ companyId, resourceType: "agent", resourceId: createdOnlyId, action: "create" }]); + await ack([agentCreate.id, ...agentTransitions.slice(0, 3).map(e => e.id), gap[1].id, ...runtime.map(e => e.id), createdOnly.id]); + await ack([gap[0].id], other.id); + + const before = await inbox.list(companyId, 100, undefined, [...all]); + const otherBefore = await pluginLifecycleInbox(db, other.id).list(companyId, 100, undefined, [...all]); + expect(before.map(e => Number(e.id))).toEqual([agentTransitions[3].id, gap[0].id]); + + const migration = readFileSync(new URL("../../../packages/db/src/migrations/0462_plugin_lifecycle_resource_cursors.sql", import.meta.url), "utf8"); + const backfill = migration.split("--> statement-breakpoint").filter(statement => statement.includes('INSERT INTO "plugin_lifecycle_resource_cursors"')); + expect(backfill).toHaveLength(1); + await db.execute(sql.raw(backfill[0])); + await db.execute(sql.raw(backfill[0])); + + const cursors = await db.select().from(pluginLifecycleResourceCursors).where(eq(pluginLifecycleResourceCursors.companyId, companyId)); + expect(cursors.map(c => [c.pluginId, c.resourceType, c.resourceId, c.acknowledgedThroughId]).sort()).toEqual([ + [plugin.id, "agent", agentId, agentTransitions[2].id], + // The cursor stops before the unacknowledged gap, so the gap is still delivered. + [plugin.id, "project", gapId, gap[0].id - 1], + [plugin.id, "agent_runtime", runtimeId, runtime[4].id], + [plugin.id, "agent", createdOnlyId, createdOnly.id], + [other.id, "project", gapId, gap[0].id], + ].sort()); + expect(await inbox.list(companyId, 100, undefined, [...all])).toEqual(before); + expect(await pluginLifecycleInbox(db, other.id).list(companyId, 100, undefined, [...all])).toEqual(otherBefore); + expect(await inbox.hasPending(companyId, ["agent_runtime"])).toBe(false); + + // Reads use the cursor: history at or below it stays acknowledged without its rows. + await db.delete(pluginLifecycleAcknowledgments).where(and(eq(pluginLifecycleAcknowledgments.pluginId, plugin.id), inArray(pluginLifecycleAcknowledgments.eventId, runtime.map(e => e.id)))); + expect(await inbox.hasPending(companyId, ["agent_runtime"])).toBe(false); + await inbox.acknowledge(companyId, String(runtime[2].id)); + expect(await inbox.list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]); + + // Acknowledging the gap moves the cursor and skips the event that the earlier release acknowledged. + await expect(inbox.acknowledge(companyId, String(gap[2].id))).rejects.toMatchObject({ status: 409 }); + await inbox.acknowledge(companyId, String(gap[0].id)); + expect((await inbox.list(companyId, 100, undefined, ["project"])).map(e => Number(e.id))).toEqual([gap[2].id]); + await inbox.acknowledge(companyId, String(gap[2].id)); + expect(await inbox.list(companyId, 100, undefined, ["project"])).toEqual([]); + const [gapCursor] = await db.select().from(pluginLifecycleResourceCursors) + .where(and(eq(pluginLifecycleResourceCursors.pluginId, plugin.id), eq(pluginLifecycleResourceCursors.resourceId, gapId))); + expect(gapCursor.acknowledgedThroughId).toBe(gap[2].id); + // An earlier release that reads only acknowledgment rows sees the same progress after a rollback. + const rows = await db.select({ eventId: pluginLifecycleAcknowledgments.eventId }).from(pluginLifecycleAcknowledgments) + .where(and(eq(pluginLifecycleAcknowledgments.pluginId, plugin.id), inArray(pluginLifecycleAcknowledgments.eventId, gap.map(e => e.id)))); + expect(rows.map(r => r.eventId).sort((a, b) => a - b)).toEqual(gap.map(e => e.id)); + + // A new transition after the cursor is delivered, and a later one waits behind it. + const [late, later] = await insert([{ companyId, resourceType: "agent_runtime", resourceId: runtimeId, action: "update" }, { companyId, resourceType: "agent_runtime", resourceId: runtimeId, action: "delete" }]); + expect((await inbox.list(companyId, 100, undefined, ["agent_runtime"])).map(e => Number(e.id))).toEqual([late.id]); + await expect(inbox.acknowledge(companyId, String(later.id))).rejects.toMatchObject({ status: 409 }); + await inbox.acknowledge(companyId, String(late.id)); + expect((await inbox.list(companyId, 100, undefined, ["agent_runtime"])).map(e => Number(e.id))).toEqual([later.id]); + + await db.delete(plugins).where(eq(plugins.id, plugin.id)); + expect(await db.select().from(pluginLifecycleResourceCursors).where(eq(pluginLifecycleResourceCursors.pluginId, plugin.id))).toEqual([]); + }); + + it("returns a full page when a manifest repeats a resource type", async () => { + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + await db.insert(resourceLifecycleEvents).values(Array.from({ length: 101 }, () => ({ + companyId, resourceType: "agent" as const, resourceId: randomUUID(), action: "create" as const, + }))); + const page = await inbox.list(companyId, 100, undefined, ["agent", "agent"]); + expect(page).toHaveLength(100); + expect(page).toEqual(await inbox.list(companyId, 100, undefined, ["agent"])); + expect(await inbox.list(companyId, 100, page.at(-1)!.id, ["agent", "agent"])).toHaveLength(1); + }); + + it("keeps the cursor behind a pending transition when a late creation is acknowledged first", async () => { + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + const lateId = randomUUID(), freshId = randomUUID(); + const through = async (resourceId: string) => (await db.select().from(pluginLifecycleResourceCursors) + .where(and(eq(pluginLifecycleResourceCursors.pluginId, plugin.id), eq(pluginLifecycleResourceCursors.resourceId, resourceId))))[0]?.acknowledgedThroughId; + const [pause] = await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: lateId, action: "pause" }).returning(); + const [create] = await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: lateId, action: "create" }).returning(); + const [freshCreate] = await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: freshId, action: "create" }).returning(); + expect((await inbox.list(companyId)).map(e => Number(e.id))).toEqual([create.id, freshCreate.id]); + await inbox.acknowledge(companyId, String(create.id)); + await inbox.acknowledge(companyId, String(freshCreate.id)); + // The earlier pause is still pending, so the creation does not move the cursor past it. + expect(await through(lateId)).toBeUndefined(); + expect(await through(freshId)).toBe(freshCreate.id); + expect((await inbox.list(companyId)).map(e => Number(e.id))).toEqual([pause.id]); + await inbox.acknowledge(companyId, String(pause.id)); + expect(await through(lateId)).toBe(pause.id); + // The creation above the cursor stays acknowledged through its row. + expect(await inbox.list(companyId)).toEqual([]); + expect(await inbox.hasPending(companyId)).toBe(false); + const [resume] = await db.insert(resourceLifecycleEvents).values({ companyId, resourceType: "agent", resourceId: lateId, action: "resume" }).returning(); + expect(await inbox.hasPending(companyId)).toBe(true); + expect((await inbox.list(companyId)).map(e => Number(e.id))).toEqual([resume.id]); + await inbox.acknowledge(companyId, String(resume.id)); + expect(await through(lateId)).toBe(resume.id); + expect(await inbox.list(companyId)).toEqual([]); + }); + + it("acknowledges a head without reading the resource's acknowledged history", async () => { + // Before the cursor, eligibility read every older event of the resource. Remove the + // acknowledgment rows below the cursor: a history scan would now see them as pending. + const plugin = await createPlugin(); + const inbox = pluginLifecycleInbox(db, plugin.id); + const resourceId = randomUUID(); + const rows = await db.insert(resourceLifecycleEvents).values(Array.from({ length: 50 }, () => ({ + companyId, resourceType: "agent_runtime" as const, resourceId, action: "update" as const, + }))).returning({ id: resourceLifecycleEvents.id }); + for (const row of rows.slice(0, 40)) await inbox.acknowledge(companyId, String(row.id)); + await db.delete(pluginLifecycleAcknowledgments).where(eq(pluginLifecycleAcknowledgments.pluginId, plugin.id)); + const all = ["agent_runtime"] as const; + expect((await inbox.list(companyId, 100, undefined, [...all])).map(e => Number(e.id))).toEqual([rows[40].id]); + await inbox.acknowledge(companyId, String(rows[40].id)); + expect((await inbox.list(companyId, 100, undefined, [...all])).map(e => Number(e.id))).toEqual([rows[41].id]); + }); + 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-inbox.ts b/server/src/services/plugin-lifecycle-inbox.ts index 9872bb5f7bd..c59df5056a0 100644 --- a/server/src/services/plugin-lifecycle-inbox.ts +++ b/server/src/services/plugin-lifecycle-inbox.ts @@ -1,29 +1,73 @@ import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server"; import { projectInstanceSecretService, type ProjectSecretSnapshot } from "./project-instance-secrets.js"; import { publishActivity, type ActivityPublication } from "./activity-log.js"; -import { pluginCompanySettings, pluginLifecycleAcknowledgments, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db"; +import { pluginCompanySettings, pluginLifecycleAcknowledgments, pluginLifecycleResourceCursors, plugins, resourceLifecycleEvents, type Db } from "@paperclipai/db"; import type { ResourceLifecycleEvent } from "@paperclipai/plugin-sdk"; import { z } from "zod"; -import { and, asc, eq, gt, inArray, sql } from "drizzle-orm"; +import { and, asc, eq, sql, type SQL } from "drizzle-orm"; import { badRequest, conflict, forbidden, notFound } from "../errors.js"; -/** Durable delivery inbox. Provider work stays outside tenant transactions. */ +type ResourceType = ResourceLifecycleEvent["resourceType"]; + +/** + * Durable delivery inbox. Provider work stays outside tenant transactions. + * + * Progress for each resource stream has two parts. The resource cursor says that every event of + * the stream at or below `acknowledged_through_id` is acknowledged. Rows in + * `plugin_lifecycle_acknowledgments` record each acknowledgment, including those above the + * cursor, such as one written by an earlier release. An event is acknowledged when either part + * says so. Writers record a resource's events while they hold that resource's row lock, so an + * event that commits later for the same resource always has a higher id than the cursor. + */ export function pluginLifecycleInbox(db: Db, pluginId: string) { - const unacknowledged = sql`NOT EXISTS ( - SELECT 1 FROM plugin_lifecycle_acknowledgments ack - WHERE ack.plugin_id = ${pluginId} AND ack.event_id = ${resourceLifecycleEvents.id} - )`; - const firstForResource = sql`NOT EXISTS ( - SELECT 1 FROM resource_lifecycle_events older - WHERE older.company_id = ${resourceLifecycleEvents.companyId} - AND older.resource_type = ${resourceLifecycleEvents.resourceType} - AND older.resource_id = ${resourceLifecycleEvents.resourceId} - -- Backfilled creation must precede transitions captured before this baseline. - AND ${resourceLifecycleEvents.action} <> 'create' - AND (older.action = 'create' OR older.id < ${resourceLifecycleEvents.id}) - AND NOT EXISTS (SELECT 1 FROM plugin_lifecycle_acknowledgments ack - WHERE ack.plugin_id = ${pluginId} AND ack.event_id = older.id) - )`; + const acknowledged = (event: SQL) => sql`EXISTS ( + SELECT 1 FROM plugin_lifecycle_acknowledgments ack WHERE ack.plugin_id = ${pluginId} AND ack.event_id = ${event})`; + const unacknowledgedCreation = (companyId: string, type: SQL, resource: SQL, through: SQL) => sql` + SELECT e.id FROM resource_lifecycle_events e + WHERE e.company_id = ${companyId} AND e.resource_type = ${type} AND e.resource_id = ${resource} + AND e.action = 'create' AND e.id > ${through} AND NOT ${acknowledged(sql`e.id`)}`; + + /** + * Ids of each resource's deliverable head: its unacknowledged creation, else its oldest + * unacknowledged transition. A backward loose index scan visits each resource stream once and + * reads its latest event id. A stream whose latest event is at or below its cursor is settled, + * so only streams with newer events are probed. Cost follows the number of resource streams + * and pending events, not the length of acknowledged history. + */ + function heads(companyId: string, types: ResourceType[]) { + // A manifest may repeat a type. Duplicate streams would repeat heads and shorten the page. + const typeArray = sql`ARRAY[${sql.join([...new Set(types)].map(type => sql`${type}`), sql`, `)}]::text[]`; + const through = sql`COALESCE(stream_cursor.acknowledged_through_id, 0)`; + return sql` + WITH RECURSIVE streams(resource_type, resource_id, last_id) AS ( + SELECT t.resource_type, first_stream.resource_id, first_stream.id FROM unnest(${typeArray}) AS t(resource_type) + CROSS JOIN LATERAL ( + SELECT e.resource_id, e.id FROM resource_lifecycle_events e + WHERE e.company_id = ${companyId} AND e.resource_type = t.resource_type + ORDER BY e.resource_id DESC, e.id DESC LIMIT 1) first_stream + UNION ALL + SELECT s.resource_type, next_stream.resource_id, next_stream.id FROM streams s + CROSS JOIN LATERAL ( + SELECT e.resource_id, e.id FROM resource_lifecycle_events e + WHERE e.company_id = ${companyId} AND e.resource_type = s.resource_type AND e.resource_id < s.resource_id + ORDER BY e.resource_id DESC, e.id DESC LIMIT 1) next_stream + ) + SELECT head.id FROM streams s + LEFT JOIN plugin_lifecycle_resource_cursors stream_cursor ON stream_cursor.plugin_id = ${pluginId} + AND stream_cursor.company_id = ${companyId} AND stream_cursor.resource_type = s.resource_type + AND stream_cursor.resource_id = s.resource_id + CROSS JOIN LATERAL ( + SELECT COALESCE( + (${unacknowledgedCreation(companyId, sql`s.resource_type`, sql`s.resource_id`, through)}), + (SELECT e.id FROM resource_lifecycle_events e + WHERE e.company_id = ${companyId} AND e.resource_type = s.resource_type AND e.resource_id = s.resource_id + AND e.id > ${through} AND e.action <> 'create' AND NOT ${acknowledged(sql`e.id`)} + ORDER BY e.id LIMIT 1)) AS id + -- Keep the head probes evaluated once per stream. + OFFSET 0 + ) head + WHERE s.last_id > ${through} AND head.id IS NOT NULL`; + } async function assertAvailable(database: Db, companyId: string) { if (!z.string().uuid().safeParse(companyId).success) throw badRequest("companyId must be a UUID"); @@ -34,29 +78,23 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) { } return { - async list(companyId: string, limit = 50, afterId?: string, resourceTypes?: ResourceLifecycleEvent["resourceType"][]): Promise { + async list(companyId: string, limit = 50, afterId?: string, resourceTypes?: ResourceType[]): Promise { 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); - // 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); + const page = sql`SELECT page.id FROM (${heads(companyId, resourceTypes ?? ["agent", "project"])}) page + ${afterId === undefined ? sql`` : sql`WHERE page.id > ${Number(afterId)}`} + ORDER BY page.id LIMIT ${limit}`; + const rows = await db.select().from(resourceLifecycleEvents) + .where(and(eq(resourceLifecycleEvents.companyId, companyId), sql`${resourceLifecycleEvents.id} IN (${page})`)) + .orderBy(asc(resourceLifecycleEvents.id)); 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 { + /** Whether `list` would return any event. */ + async hasPending(companyId: string, resourceTypes?: 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; + const [row] = await db.execute<{ pending: boolean }>(sql`SELECT EXISTS (${heads(companyId, resourceTypes ?? ["agent", "project"])}) AS pending`); + return row?.pending === true; }, 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"); @@ -67,16 +105,41 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) { const [event] = await tx.select().from(resourceLifecycleEvents) .where(and(eq(resourceLifecycleEvents.companyId, companyId), eq(resourceLifecycleEvents.id, Number(eventId)))).for("update"); if (!event) throw notFound("Lifecycle event not found"); + const [cursor] = await tx.select({ through: pluginLifecycleResourceCursors.acknowledgedThroughId }).from(pluginLifecycleResourceCursors) + .where(and(eq(pluginLifecycleResourceCursors.pluginId, pluginId), eq(pluginLifecycleResourceCursors.companyId, companyId), + eq(pluginLifecycleResourceCursors.resourceType, event.resourceType), eq(pluginLifecycleResourceCursors.resourceId, event.resourceId))); + const through = cursor?.through ?? 0; + if (event.id <= through) return publications; const [acked] = await tx.select().from(pluginLifecycleAcknowledgments).where(and(eq(pluginLifecycleAcknowledgments.pluginId, pluginId), eq(pluginLifecycleAcknowledgments.eventId, event.id))); if (acked) return publications; - const [eligible] = await tx.select({ id: resourceLifecycleEvents.id }).from(resourceLifecycleEvents) - .where(and(eq(resourceLifecycleEvents.id, event.id), firstForResource)); - if (!eligible) throw conflict("Acknowledge earlier lifecycle events for this resource first"); + // Events at or below the cursor are acknowledged, so each check reads only the range above it. + const earlierTransition = sql`(SELECT e.id FROM resource_lifecycle_events e + WHERE e.company_id = ${companyId} AND e.resource_type = ${event.resourceType} AND e.resource_id = ${event.resourceId} + AND e.id > ${through} AND e.id < ${event.id} AND e.action <> 'create' AND NOT ${acknowledged(sql`e.id`)} + ORDER BY e.id LIMIT 1)`; + const creation = sql`(${unacknowledgedCreation(companyId, sql`${event.resourceType}`, sql`${event.resourceId}::uuid`, sql`${through}`)})`; + // A creation is always deliverable. A transition waits for the creation and every earlier + // transition. The earlier pending transition takes precedence, so for a creation the + // blocker is the creation itself only when no earlier transition is pending. + const [row] = await tx.execute<{ id: number | string | null }>(sql`SELECT COALESCE(${earlierTransition}, ${creation}) AS id`); + const blocker = row?.id == null ? undefined : { id: row.id }; + if (event.action !== "create" && blocker) throw conflict("Acknowledge earlier lifecycle events for this resource first"); if (credentials) { if (event.resourceType !== "project") throw forbidden("Project instance secret event required"); publications.push(...await projectInstanceSecretService(txDb, pluginId).apply({ ...event, id: String(event.id), createdAt: event.createdAt.toISOString() } as ResourceLifecycleEvent, credentials.snapshot, credentials.changes)); } + // Keep the per-event row as well, so an earlier release reads the same progress after a rollback. await tx.insert(pluginLifecycleAcknowledgments).values({ pluginId, eventId: event.id }).onConflictDoNothing(); + // The cursor moves only over a contiguous acknowledged prefix. A creation recorded after + // transitions that are still pending leaves the cursor in place. + if (event.action !== "create" || Number(blocker?.id) === event.id) { + await tx.insert(pluginLifecycleResourceCursors) + .values({ pluginId, companyId, resourceType: event.resourceType, resourceId: event.resourceId, acknowledgedThroughId: event.id }) + .onConflictDoUpdate({ + target: [pluginLifecycleResourceCursors.pluginId, pluginLifecycleResourceCursors.companyId, pluginLifecycleResourceCursors.resourceType, pluginLifecycleResourceCursors.resourceId], + set: { acknowledgedThroughId: sql`GREATEST(${pluginLifecycleResourceCursors.acknowledgedThroughId}, excluded.acknowledged_through_id)`, updatedAt: sql`now()` }, + }); + } return publications; }); const publications = credentials ? await withAccountHomeSecretMutationLock(undefined, companyId, commit) : await commit();