mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(plugins): read the lifecycle inbox from per-resource cursors (#20576)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Plugins get agent, project and agent runtime changes from the `resource_lifecycle_events` journal. Each plugin acknowledges events one resource at a time. > - #20405 made the inbox page read linear in the company journal size. The journal has no retention, so each page still reads every event of the company to find the unacknowledged ones: about 1.7 s per 1M events. > - The delivery worker reads a page again after each acknowledgment. The idle-sleep check also reads the whole journal when all events are acknowledged. `acknowledge` checks every older event of the resource. > - On a hosted instance, the page read ran 181 times in one window at about 1 s each. > - This pull request adds a per-plugin, per-resource cursor. Reads start at the cursor, so they do not read acknowledged history again. > - The benefit is that page, idle-check and acknowledgment cost follow the number of resources and pending events, not the journal size. Delivery order and results stay the same. ## Linked Issues or Issue Description Refs #20405 **What happened?** The plugin lifecycle inbox page query (`select ... from "resource_lifecycle_events" ... DISTINCT ON ...`) ran 181 times at about 1 s each on a hosted database in one window. Each acknowledgment causes another full read of the company journal. The idle-sleep check (`hasPending`) reads the full journal when all events are acknowledged. `acknowledge` reads the full history of the resource. **Expected behavior** A page read, the idle-sleep check and an acknowledgment must not read acknowledged history again. Their cost must not grow with the journal size. **Steps to reproduce** 1. Give a company 400 agents with 2,500 `agent_runtime` events each (1M events). 2. Acknowledge the first 2,000 events of each agent for one plugin. 3. Call `pluginLifecycleInbox(db, pluginId).list(companyId, 100, undefined, ["agent", "project", "agent_runtime"])`. 4. The read takes about 1.7 s. `hasPending` takes about 240 ms. **Paperclip version or commit** `8d2bf78323` (master). **Deployment mode** Hosted (PostgreSQL). Local repro uses embedded PostgreSQL 18. ## What Changed - `packages/db/src/schema/plugin_lifecycle_resource_cursors.ts`: new table `plugin_lifecycle_resource_cursors`. It has one row per plugin, company, resource type and resource id. `acknowledged_through_id` means: every event of that resource at or below this id is acknowledged by the plugin. Plugin and company deletion cascade. - Migration `0462_plugin_lifecycle_resource_cursors`: creates the table and carries existing acknowledgments forward (see Risks for the previous-release story). - `server/src/services/plugin-lifecycle-inbox.ts`: - `list` and `hasPending` use a backward loose index scan on `resource_lifecycle_events_resource_order_idx`. It visits each resource stream once and gets its latest event id. A stream whose latest event is at or below its cursor is settled and is not read again. For other streams, the read gets one head: the unacknowledged creation, else the first unacknowledged transition above the cursor. Then it applies `afterId`, id order and the limit, as before. - `acknowledge` reads only the events between the cursor and the acknowledged event. It still writes the per-event `plugin_lifecycle_acknowledgments` row. It moves the cursor only over a contiguous acknowledged prefix. A creation recorded after a pending transition does not move the cursor. - An event counts as acknowledged when it is at or below the cursor, or when it has an acknowledgment row. Thus rows written by an earlier release, and gaps in old history, keep their current behavior. - `server/src/__tests__/resource-lifecycle-events.test.ts`: new database tests for the migration carry-forward, gaps, late creations, cursor reads, the rollback rows and the plugin-delete cascade. All tests from #20405 still pass without change. - `scripts/benchmark-plugin-lifecycle-inbox.ts`: runs the migration backfill on the fixture and times it. Adds an acknowledgment case. Runs `VACUUM ANALYZE`, because hosted tables are autovacuumed. - `doc/DATABASE.md`: describes the cursor, its read cost and the migration. - The delivery worker (`plugin-lifecycle-delivery.ts`) is not changed. ## Verification Thesis: the cost is a read of the whole company journal (page, idle check) or the whole resource history (acknowledgment). If a per-resource cursor marks the acknowledged prefix, each read can skip that prefix. Then the cost follows the number of resource streams and pending events. The rows must stay equal. Local SQL timings: embedded PostgreSQL 18.1 on Apple Silicon, synthetic data, warm cache, one warmup, 7 samples, baseline and treatment order alternate per sample. The machine was shared with other jobs (load average about 35), so spread is wide. Values are median (min to max) in ms. Rows are equal (`assert.deepEqual`) for every list and probe case. Fixture A: 400 agents x 2,500 runtime events, 2,000 acknowledged per agent. 1,020,440 events, 800,420 acknowledgments. | Case | Before | After | Speedup | | --- | --- | --- | --- | | page of 100, all types | 1,700 (1,633-2,438) | 11.5 (9.5-12.2) | 148x | | page of 100, agent and project | 3.3 (2.5-5.4) | 4.2 (3.6-6.7) | 0.8x (slower) | | idle-sleep check | 238 (174-367) | 2.7 (2.1-3.4) | 88x | | acknowledge one head | 15.3 (9.4-30.1) | 20.3 (12.9-29.9) | neutral (noise) | | migration backfill | - | 9.3 s | - | Fixture B: 40 agents x 25,000 runtime events, 24,000 acknowledged per agent (long history). 1,020,080 events. | Case | Before | After | Speedup | | --- | --- | --- | --- | | page of 100, all types | 1,943 (1,642-4,979) | 4.4 (2.9-5.6) | 444x | | page of 100, agent and project | 3.5 (2.1-5.7) | 4.6 (2.8-12.0) | 0.8x (slower) | | idle-sleep check | 234 (220-243) | 2.8 (1.9-6.5) | 85x | | acknowledge one head | 64.6 (48.5-71.6) | 5.0 (2.6-7.1) | 13x | | migration backfill | - | 4.6 s | - | Fixture C: 5,000 agents x 200 runtime events, 190 acknowledged per agent (many resources). 1,025,040 events. | Case | Before | After | Speedup | | --- | --- | --- | --- | | page of 100, all types | 1,867 (1,649-1,975) | 76.9 (66.2-497) | 24x | | page of 100, agent and project | 10.3 (8.7-12.4) | 21.6 (20.2-23.2) | 0.5x (slower) | | idle-sleep check | 220 (201-298) | 2.8 (2.0-5.3) | 78x | | acknowledge one head | 4.9 (3.4-8.6) | 6.0 (5.2-7.6) | neutral | | migration backfill | - | 15.6 s | - | `EXPLAIN (ANALYZE, BUFFERS)` for the all-types page on fixture A: before 2,488 ms, 13,597 shared hits plus temp files; after 7.0 ms, 7,926 shared hits. Commands: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=8d2bf78323 --agents=400 --per-agent=2500 --acked=2000 --samples=7 pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=8d2bf78323 --agents=40 --per-agent=25000 --acked=24000 --samples=7 pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=8d2bf78323 --agents=5000 --per-agent=200 --acked=190 --samples=7 cd server && npx vitest run src/__tests__/resource-lifecycle-events.test.ts src/__tests__/plugin-lifecycle-delivery.test.ts src/__tests__/project-instance-secrets.test.ts src/__tests__/agent-runtime-hooks.test.ts pnpm --filter @paperclipai/db check:migrations ``` - `resource-lifecycle-events.test.ts`, `plugin-lifecycle-delivery.test.ts` and `project-instance-secrets.test.ts`: 86 passed. `agent-runtime-hooks.test.ts`: 22 passed. - `tsc --noEmit -p server`: pass. `check:migrations`: pass. ## Risks - **Schema change.** Migration `0462` adds one table and two foreign keys to `plugins` and `companies`. It adds no index on an existing table. The backfill reads the acknowledgment table once. It took 4.6 s to 15.6 s on 1M-event fixtures with 0.8M to 1M acknowledgments. - **Migration order.** The number `0462` is reserved after open PRs that use `0459` to `0461` (#20571, #20568, #20573). Merge this PR after them. Before merge, rebase onto master and chain the snapshot after the latest migration. - **Previous-release story.** - Upgrade: the backfill stores, per plugin and resource, the end of the contiguous acknowledged prefix. A cursor never passes an unacknowledged event, so no event is lost. Acknowledged events above a gap keep their rows, so no event is delivered again. A test runs the migration statement on rows that only an earlier release wrote and checks that `list` returns the same rows before and after. - Rolling deploy: an earlier release that still runs writes only acknowledgment rows. The new reads honor those rows. - Rollback: the new code still writes every acknowledgment row, so an earlier release sees the same progress. - **Ordering assumption.** The cursor is safe because writers record a resource's events while they hold that resource's row lock (`paperclip_record_agent_runtime` locks the agent row; agent status and project writers document the lock). Thus a later commit for the same resource always has a higher id. A writer that breaks this contract could commit an event below the cursor, and that event would not be delivered. A lower id from a different resource is still delivered (existing test). - **Slower path.** A page that asks only for `agent` and `project` streams is 1 to 11 ms slower. These streams are short, so the old full read was cheap, and the new read visits each stream once. The cost is bounded by the number of resources. - **Many resources.** The page cost grows with the number of resource streams of the requested types. At 5,000 agents a page is about 77 ms, compared with about 1.9 s before. - The hosted planner can choose a different plan. The hosted improvement is not yet measured. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local PostgreSQL benchmarks). ## 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 - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
19e938415a
commit
3bf7af6c77
9 files changed
+546
-53
No files matched your search
+28
-8
@@ -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
|
||||
|
||||
@@ -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;
|
||||
+183
-2
@@ -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": "",
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
@@ -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] }),
|
||||
}));
|
||||
@@ -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=<sha> [--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); }
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<ResourceLifecycleEvent[]> {
|
||||
async list(companyId: string, limit = 50, afterId?: string, resourceTypes?: ResourceType[]): Promise<ResourceLifecycleEvent[]> {
|
||||
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<boolean> {
|
||||
/** Whether `list` would return any event. */
|
||||
async hasPending(companyId: string, resourceTypes?: ResourceType[]): Promise<boolean> {
|
||||
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<void> {
|
||||
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();
|
||||
|
||||
Reference in new issue
Block a user