mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(server): read plugin lifecycle inbox heads in one pass (#20405)
## 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. > - The inbox page query checked each unacknowledged event against all older events of the same resource. Each check walked the acknowledged history of that resource. > - Agent runtime streams get a new event for each configuration change, so one resource can have thousands of acknowledged events. The cost grew as pending events times acknowledged history. > - On a hosted instance, three copies of this read ran for 601 seconds each, and one earlier copy ran for 19 minutes. The delivery worker runs it again after each acknowledgment. > - This pull request selects the head of each resource in one pass and keeps the same result. > - The benefit is that the page read is linear in the company journal size. In local tests it is 250x to 660x faster, and the result rows are equal. ## Linked Issues or Issue Description **What happened?** The plugin lifecycle inbox page query (`select "id", "company_id", "resource_type", "resource_id", "action", "created_at" from "resource_lifecycle_events" where ...`) ran for 601 seconds on a hosted database. Three copies were active at the same time. **Expected behavior** A lifecycle page read must stay fast when a resource has a long acknowledged history and a pending backlog. **Steps to reproduce** 1. Give a company 40 agents with 1,500 `agent_runtime` events each. 2. Acknowledge the first 1,200 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 10 seconds for 60,000 events of that company. At 200,000 events it takes about 6 minutes. **Paperclip version or commit** `02bfe71a6b` (master). **Deployment mode** Hosted (PostgreSQL). Local repro uses embedded PostgreSQL 18. ## What Changed - `server/src/services/plugin-lifecycle-inbox.ts`: `list` now selects one head per resource with `DISTINCT ON (resource_type, resource_id)`. The head is the unacknowledged creation, else the oldest unacknowledged event. Then the query applies `afterId`, id order and the limit. This is the same set that the old `firstForResource` predicate admits. `acknowledge` keeps the old predicate for its single-event check. - `plugin-lifecycle-inbox.ts`: new `hasPending(companyId, types)`. It returns whether any unacknowledged event exists. Each resource with an unacknowledged event has a head, so this gives the same answer as `list(companyId, 1).length > 0`. - `server/src/services/plugin-lifecycle-delivery.ts`: the idle-sleep journal check uses `hasPending`. This keeps its early exit. A one-row `list` page would read the full journal. - `server/src/__tests__/resource-lifecycle-events.test.ts`: new database test for heads across acknowledged history, a creation recorded after transitions, acknowledged gaps, shared ids across resource types, per-plugin progress, another company, `afterId` pages, limits, `hasPending`, and the next head after an acknowledgment. The test also passes on the old query, so it locks the contract. - `scripts/benchmark-plugin-lifecycle-inbox.ts`: new benchmark. It loads the baseline service from git, runs both on the same fixture, alternates order, asserts equal rows, and prints plans. - `doc/DATABASE.md`: describes the read cost and the benchmark command. - No schema change, index or migration. ## Verification Thesis: the old read cost is (unacknowledged candidates) x (acknowledged older events per resource), because each candidate walks the acknowledged prefix of its resource. One pass that keeps the per-resource head removes that product and returns equal rows. Baseline plan at 220,080 events: 40,020 candidates each walked about 3,999 older rows, which gave 160,039,980 acknowledgment probes and 562,140,833 buffer hits. The treatment plan reads the journal once with a hash anti join and sorts the 40,020 unacknowledged rows (3,925 buffer hits). Local SQL timings (embedded PostgreSQL 18, Apple Silicon; event counts include a 20,000-event second company, synthetic data, warm cache, one warmup, order alternates): | Fixture (events / acknowledged) | Case | Before median | After median | Samples (after) | Speedup | | --- | --- | --- | --- | --- | --- | | 40,060 / 16,040 | page of 100, all types | 3,263 ms | 13.1 ms | 3 (12.0-14.1) | 250x | | 80,080 / 48,060 | page of 100, all types | 10,137 ms | 23.9 ms | 5 (22.7-32.6) | 424x | | 220,080 / 160,060 | page of 100, all types | 355,718 ms | 538 ms | 3 (72-785) | 662x | | 1,020,080 / 960,060 | page of 100, all types | not run (est. > 30 min) | 1,682 ms | 7 (1,203-2,384) | - | | 80,080 / 48,060 | page of 100, agent and project only | 1.5 ms | 1.6 ms | 5 | neutral | | 80,080 / 48,060 | idle-sleep check | 14.5 ms | 13.8 ms | 5 | neutral | | 1,020,080 / 960,060 | idle-sleep check | not run | 406 ms | 7 | - | - Result rows are equal (`assert.deepEqual` on all mapped rows, in order) for each case where both versions ran. - The agent runtime revision read in `plugin-agent-runtime-state.ts` and `agent-file-store.ts` (`select id ... order by id desc limit 1`) uses a backward index-only scan and reads 5 buffers at 1M events. It is not changed. Commands: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=02bfe71a6b --agents=40 --per-agent=1500 --acked=1200 --samples=5 pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=02bfe71a6b --agents=40 --per-agent=25000 --acked=24000 --samples=7 --treatment-only cd server && npx vitest run src/__tests__/resource-lifecycle-events.test.ts src/__tests__/plugin-lifecycle-delivery.test.ts src/__tests__/agent-runtime-hooks.test.ts ``` - `resource-lifecycle-events.test.ts`: 28 passed. `plugin-lifecycle-delivery.test.ts`: 40 passed (includes idle-sleep "none" and "present"). `agent-runtime-hooks.test.ts`: 22 passed. - `tsc --noEmit -p server`: pass. ## Risks - The page read is still linear in the company journal size, because acknowledgments are per plugin and the journal has no retention yet. At 1M events a page takes about 1.7 s locally. Each acknowledgment wakes another full pass. A per-plugin cursor or retention can remove this later. - The planner can choose a different plan on the hosted database. The local plans use a hash anti join and one sort. The hosted improvement is not yet measured. - `acknowledge` still checks one event against older events of its resource. That check is linear in that resource's history per call, as before. - Low behavior risk: the new test passes on both the old and new queries. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local PostgreSQL). Extended reasoning on. ## 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
17014cd7db
commit
c57cd1bdb9
5 files changed
+206
-5
No files matched your search
@@ -1596,6 +1596,18 @@ Lifecycle polls can page past failed resources using the last returned event id
|
||||
cursor, never a persisted high-water mark. This retries failures and includes
|
||||
transactions that commit later with lower ids.
|
||||
|
||||
A lifecycle page reads the company's unacknowledged events for the requested
|
||||
types once. It keeps one head for each resource: the unacknowledged creation,
|
||||
else the oldest unacknowledged event. Then it applies `afterId`, id order and
|
||||
the limit. Cost grows with the company's journal size, not with pending events
|
||||
times acknowledged history. The idle-sleep check only asks whether any
|
||||
unacknowledged event exists, because each such resource has a head. No index
|
||||
or migration is needed.
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts --baseline-ref=02bfe71a6b
|
||||
```
|
||||
|
||||
## Legacy controller ownership
|
||||
|
||||
Legacy run claims atomically record `controller_boot_id`, a database-clock
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
// Compares the plugin lifecycle inbox page query before and after a change.
|
||||
// Usage: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-plugin-lifecycle-inbox.ts \
|
||||
// --baseline-ref=<sha> [--agents=40] [--per-agent=2500] [--acked=2000] [--samples=5] [--treatment-only]
|
||||
import assert from "node:assert/strict";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath, pathToFileURL } from "node:url";
|
||||
import { createRequire } from "node:module";
|
||||
import * as schema from "../packages/db/src/schema/index.js";
|
||||
import { companies, plugins, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js";
|
||||
import * as current from "../server/src/services/plugin-lifecycle-inbox.js";
|
||||
|
||||
const arg = (name: string, fallback: string) => process.argv.find(a => a.startsWith(`--${name}=`))?.slice(name.length + 3) ?? fallback;
|
||||
const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${arg("baseline-ref", "origin/master")}^{commit}`], { encoding: "utf8" }).trim();
|
||||
const agentCount = Number(arg("agents", "40"));
|
||||
const perAgent = Number(arg("per-agent", "2500"));
|
||||
const acked = Number(arg("acked", "2000"));
|
||||
const sampleCount = Number(arg("samples", "5"));
|
||||
const timeoutMs = Number(arg("timeout-ms", "600000"));
|
||||
// The baseline is quadratic; at large fixtures time only the treatment.
|
||||
const treatmentOnly = process.argv.includes("--treatment-only");
|
||||
// Resolve database drivers through the db package; the scripts folder has no own dependencies.
|
||||
const dbRequire = createRequire(new URL("../packages/db/package.json", import.meta.url));
|
||||
const postgres = (await import(pathToFileURL(dbRequire.resolve("postgres")).href)).default as typeof import("postgres").default;
|
||||
const { drizzle } = await import(pathToFileURL(dbRequire.resolve("drizzle-orm/postgres-js")).href) as typeof import("drizzle-orm/postgres-js");
|
||||
type Inbox = typeof current.pluginLifecycleInbox;
|
||||
type Query = { sql: string; params: unknown[] };
|
||||
|
||||
const keepAlive = setInterval(() => {}, 1000);
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let client: import("postgres").Sql | undefined;
|
||||
let baselineDir: string | undefined;
|
||||
try {
|
||||
database = await startEmbeddedPostgresTestDatabase("lifecycle-inbox-benchmark-");
|
||||
let recorded: Query[] = [];
|
||||
client = postgres(database.connectionString, {
|
||||
max: 1, onnotice: () => {}, connection: { statement_timeout: timeoutMs },
|
||||
debug: (_connection, query, params) => { recorded.push({ sql: query, params }); },
|
||||
});
|
||||
const db = drizzle(client, { schema }) as unknown as Db;
|
||||
|
||||
baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".lifecycle-baseline-"));
|
||||
const modulePath = "server/src/services/plugin-lifecycle-inbox.ts";
|
||||
const original = new URL(`../${modulePath}`, import.meta.url);
|
||||
const source = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" })
|
||||
.replace(/from "(\.{1,2}\/[^"]+)"/g, (_m, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, original)))}`);
|
||||
const target = path.join(baselineDir, "plugin-lifecycle-inbox.ts");
|
||||
await fs.writeFile(target, source);
|
||||
const baseline = (await import(pathToFileURL(target).href)).pluginLifecycleInbox as Inbox;
|
||||
|
||||
// Fixture: agent_runtime history interleaved across agents. Each agent has `acked`
|
||||
// acknowledged revisions followed by a backlog of unacknowledged revisions.
|
||||
const companyId = randomUUID(), decoyId = randomUUID();
|
||||
await db.insert(companies).values([{ id: companyId, name: "Bench", issuePrefix: "BNCH" }, { id: decoyId, name: "Decoy", issuePrefix: "DCOY" }]);
|
||||
const [plugin] = await db.insert(plugins).values({ pluginKey: randomUUID(), packageName: "bench", version: "1.0.0", status: "ready", manifestJson: {} as never }).returning();
|
||||
const agentIds = Array.from({ length: agentCount }, () => randomUUID());
|
||||
const projectIds = Array.from({ length: 20 }, () => randomUUID());
|
||||
await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action)
|
||||
SELECT $1::uuid, 'agent', a, 'create' FROM unnest($2::uuid[]) a
|
||||
UNION ALL SELECT $1::uuid, 'project', p, 'create' FROM unnest($3::uuid[]) p`, [companyId, agentIds, projectIds]);
|
||||
await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action)
|
||||
SELECT $1::uuid, 'agent_runtime', ($2::uuid[])[(n % $3) + 1], 'update' FROM generate_series(0, $4::int - 1) n ORDER BY n`,
|
||||
[companyId, agentIds, agentCount, agentCount * perAgent]);
|
||||
// A few project updates after the agent backlog, so heads are spread across the id range.
|
||||
await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action)
|
||||
SELECT $1::uuid, 'project', p, 'update' FROM unnest($2::uuid[]) p`, [companyId, projectIds]);
|
||||
await client.unsafe(`INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action)
|
||||
SELECT $1::uuid, 'agent_runtime', gen_random_uuid(), 'update' FROM generate_series(1, 20000)`, [decoyId]);
|
||||
await client.unsafe(`INSERT INTO plugin_lifecycle_acknowledgments (plugin_id, event_id)
|
||||
SELECT $1::uuid, id FROM (
|
||||
SELECT id, resource_type, row_number() OVER (PARTITION BY resource_type, resource_id ORDER BY id) AS rank
|
||||
FROM resource_lifecycle_events WHERE company_id = $2::uuid) e
|
||||
WHERE resource_type = 'agent' OR (resource_type = 'project' AND rank = 1) OR (resource_type = 'agent_runtime' AND rank <= $3)`,
|
||||
[plugin.id, companyId, acked]);
|
||||
await client.unsafe("ANALYZE");
|
||||
const [{ events }] = await client.unsafe(`SELECT count(*)::int AS events FROM resource_lifecycle_events`);
|
||||
const [{ acks }] = await client.unsafe(`SELECT count(*)::int AS acks FROM plugin_lifecycle_acknowledgments`);
|
||||
const [{ server_version: version }] = await client.unsafe(`SHOW server_version`);
|
||||
|
||||
const cases = [
|
||||
{ name: "page-100-all-types", limit: 100, types: ["agent", "project", "agent_runtime"] },
|
||||
{ name: "page-100-agent-project", limit: 100, types: ["agent", "project"] },
|
||||
// Idle-sleep check: baseline read one listed event; the treatment asks only whether one exists.
|
||||
{ name: "idle-probe-all-types", limit: 1, types: ["agent", "project", "agent_runtime"], probe: true },
|
||||
] as const;
|
||||
async function run(inbox: Inbox, c: (typeof cases)[number]) {
|
||||
recorded = [];
|
||||
const start = performance.now();
|
||||
const service = inbox(db, plugin.id) as ReturnType<Inbox>;
|
||||
const rows = "probe" in c
|
||||
? [("hasPending" in service ? await service.hasPending(companyId, [...c.types]) : (await service.list(companyId, 1, undefined, [...c.types])).length > 0)]
|
||||
: await service.list(companyId, c.limit, undefined, [...c.types]);
|
||||
const ms = performance.now() - start;
|
||||
const query = recorded.find(q => q.sql.includes('"resource_lifecycle_events"'));
|
||||
return { ms, rows, query };
|
||||
}
|
||||
const results = [];
|
||||
for (const c of cases) {
|
||||
const samples = { baseline: [] as number[], treatment: [] as number[] };
|
||||
let baseRows: unknown, treatRows: unknown, baseQuery: Query | undefined, treatQuery: Query | undefined;
|
||||
// One warmup each, then alternate order per sample to limit cache bias.
|
||||
for (let n = 0; n <= sampleCount; n++) {
|
||||
const order = n % 2 === 0 ? (["baseline", "treatment"] as const) : (["treatment", "baseline"] as const);
|
||||
for (const which of order.filter(w => !treatmentOnly || w === "treatment")) {
|
||||
const r = await run(which === "baseline" ? baseline : current.pluginLifecycleInbox, c);
|
||||
if (n > 0) samples[which].push(r.ms);
|
||||
if (which === "baseline") { baseRows = r.rows; baseQuery = r.query; } else { treatRows = r.rows; treatQuery = r.query; }
|
||||
}
|
||||
}
|
||||
if (!treatmentOnly) assert.deepEqual(treatRows, baseRows, `${c.name}: results differ`);
|
||||
const explain = async (q?: Query) => q ? (await client!.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) ${q.sql}`, q.params as never[])).map(r => r["QUERY PLAN"]).join("\n") : "";
|
||||
const stats = (s: number[]) => { const v = [...s].sort((a, b) => a - b); return { median: v[Math.floor(v.length / 2)], min: v[0], max: v.at(-1), samples: s.map(x => Math.round(x * 10) / 10) }; };
|
||||
const b = treatmentOnly ? undefined : stats(samples.baseline), t = stats(samples.treatment);
|
||||
results.push({ case: c.name, rows: (treatRows as unknown[]).length, equal: !treatmentOnly, baselineMs: b, treatmentMs: t, speedup: b ? Math.round(b.median / t.median * 10) / 10 : undefined,
|
||||
baselinePlan: await explain(baseQuery), treatmentPlan: await explain(treatQuery) });
|
||||
}
|
||||
// Control: the agent runtime revision read (plugin-agent-runtime-state.ts, agent-file-store.ts) is unchanged.
|
||||
const revisionPlan = (await client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) SELECT id FROM resource_lifecycle_events
|
||||
WHERE company_id = $1 AND resource_type = 'agent_runtime' AND resource_id = $2 ORDER BY id DESC LIMIT 1`, [companyId, agentIds[0]]))
|
||||
.map(r => r["QUERY PLAN"]).join("\n");
|
||||
console.log(JSON.stringify({ revisionPlan, baselineRevision, postgres: version, agents: agentCount, perAgent, acked, events, acks, samples: sampleCount, results }, null, 2));
|
||||
} finally {
|
||||
try { await client?.end({ timeout: 0 }); await database?.cleanup(); if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); }
|
||||
finally { clearInterval(keepAlive); }
|
||||
}
|
||||
@@ -462,6 +462,53 @@ describePostgres("Resource lifecycle events", () => {
|
||||
await expect(inbox.list(companyId, 100, "invalid")).rejects.toMatchObject({ status: 400 });
|
||||
});
|
||||
|
||||
it("lists exactly one head per resource across acknowledged history, late creations, gaps, and pages", async () => {
|
||||
const plugin = await createPlugin();
|
||||
const other = await createPlugin();
|
||||
const inbox = pluginLifecycleInbox(db, plugin.id);
|
||||
const decoyCompanyId = randomUUID();
|
||||
await db.insert(companies).values({ id: decoyCompanyId, name: "Lifecycle decoy", issuePrefix: `D${decoyCompanyId.replaceAll("-", "").slice(0, 6)}` });
|
||||
const agentId = randomUUID(), lateCreateId = randomUUID(), gapId = randomUUID(), doneId = randomUUID(), projectId = randomUUID();
|
||||
type Row = typeof resourceLifecycleEvents.$inferInsert;
|
||||
const insert = async (rows: Row[]) => db.insert(resourceLifecycleEvents).values(rows).returning();
|
||||
const ack = async (ids: number[], pluginId = plugin.id) => {
|
||||
if (ids.length) await db.insert(pluginLifecycleAcknowledgments).values(ids.map(eventId => ({ pluginId, eventId })));
|
||||
};
|
||||
const [agentCreate] = await insert([{ companyId, resourceType: "agent", resourceId: agentId, action: "create" }]);
|
||||
// The same id as a different resource type is an independent stream.
|
||||
const runtime = await insert(Array.from({ length: 12 }, () => ({ companyId, resourceType: "agent_runtime" as const, resourceId: agentId, action: "update" as const })));
|
||||
const [latePause] = await insert([{ companyId, resourceType: "agent", resourceId: lateCreateId, action: "pause" }]);
|
||||
const gap = await insert([0, 1, 2].map(() => ({ companyId, resourceType: "project" as const, resourceId: gapId, action: "update" as const })));
|
||||
const done = await insert([{ companyId, resourceType: "agent", resourceId: doneId, action: "create" }, { companyId, resourceType: "agent", resourceId: doneId, action: "pause" }]);
|
||||
const [lateCreate] = await insert([{ companyId, resourceType: "agent", resourceId: lateCreateId, action: "create" }]);
|
||||
const [projectCreate, projectArchive] = await insert([
|
||||
{ companyId, resourceType: "project", resourceId: projectId, action: "create" }, { companyId, resourceType: "project", resourceId: projectId, action: "archive" }]);
|
||||
await insert([{ companyId: decoyCompanyId, resourceType: "agent", resourceId: randomUUID(), action: "create" }]);
|
||||
await ack([agentCreate.id, ...runtime.slice(0, 9).map(e => e.id), gap[1].id, ...done.map(e => e.id), projectCreate.id]);
|
||||
await ack(runtime.map(e => e.id), other.id);
|
||||
|
||||
const all = ["agent", "project", "agent_runtime"] as const;
|
||||
expect((await inbox.list(companyId, 100, undefined, [...all])).map(e => Number(e.id))).toEqual(
|
||||
[runtime[9].id, gap[0].id, lateCreate.id, projectArchive.id].sort((a, b) => a - b));
|
||||
expect((await inbox.list(companyId)).map(e => [e.resourceId, e.action])).toEqual(
|
||||
[[gapId, "update"], [lateCreateId, "create"], [projectId, "archive"]]);
|
||||
expect(await inbox.list(companyId, 2, undefined, [...all])).toEqual((await inbox.list(companyId, 100, undefined, [...all])).slice(0, 2));
|
||||
expect((await inbox.list(companyId, 100, String(gap[0].id), [...all])).map(e => Number(e.id))).toEqual([lateCreate.id, projectArchive.id]);
|
||||
expect(await pluginLifecycleInbox(db, other.id).list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]);
|
||||
expect(await pluginLifecycleInbox(db, other.id).hasPending(companyId, ["agent_runtime"])).toBe(false);
|
||||
expect(await inbox.hasPending(companyId, ["agent_runtime"])).toBe(true);
|
||||
expect(await inbox.hasPending(companyId)).toBe(true);
|
||||
expect(await inbox.list(decoyCompanyId, 100, undefined, [...all])).toEqual([expect.objectContaining({ companyId: decoyCompanyId })]);
|
||||
const [head] = await inbox.list(companyId, 1, undefined, ["project"]);
|
||||
expect(head).toEqual({ id: String(gap[0].id), companyId, resourceType: "project", resourceId: gapId, action: "update", createdAt: gap[0].createdAt.toISOString() });
|
||||
|
||||
// Acknowledging a head exposes that resource's next unacknowledged event, skipping acknowledged gaps.
|
||||
await inbox.acknowledge(companyId, String(gap[0].id));
|
||||
await inbox.acknowledge(companyId, String(lateCreate.id));
|
||||
expect((await inbox.list(companyId)).map(e => Number(e.id))).toEqual([latePause.id, gap[2].id, projectArchive.id].sort((a, b) => a - b));
|
||||
await expect(inbox.acknowledge(companyId, String(runtime[10].id))).rejects.toMatchObject({ status: 409 });
|
||||
});
|
||||
|
||||
it("never skips a lower event id that commits after a higher id is acknowledged", async () => {
|
||||
const plugin = await createPlugin();
|
||||
const inbox = pluginLifecycleInbox(db, plugin.id);
|
||||
|
||||
@@ -236,9 +236,9 @@ export function registerPluginLifecycleDelivery(
|
||||
if (!current()) return "unknown";
|
||||
if (!types) continue;
|
||||
const { plugin, companyId } = target;
|
||||
const events = await pluginLifecycleInbox(db, plugin.id).list(companyId, 1, undefined, types);
|
||||
const hasEvents = await pluginLifecycleInbox(db, plugin.id).hasPending(companyId, types);
|
||||
if (!current()) return "unknown";
|
||||
if (events.length) { wake(); inspectionFailure = false; return "present"; }
|
||||
if (hasEvents) { wake(); inspectionFailure = false; return "present"; }
|
||||
}
|
||||
inspectionFailure = false;
|
||||
return "none";
|
||||
|
||||
@@ -38,11 +38,26 @@ export function pluginLifecycleInbox(db: Db, pluginId: string) {
|
||||
if (!Number.isInteger(limit) || limit < 1 || limit > 100) throw badRequest("limit must be an integer from 1 to 100");
|
||||
if (afterId !== undefined && (typeof afterId !== "string" || !/^[1-9]\d*$/.test(afterId) || !Number.isSafeInteger(Number(afterId)))) throw badRequest("Invalid lifecycle page id");
|
||||
await assertAvailable(db, companyId);
|
||||
const rows = await db.select().from(resourceLifecycleEvents)
|
||||
.where(and(eq(resourceLifecycleEvents.companyId, companyId), unacknowledged, firstForResource, inArray(resourceLifecycleEvents.resourceType, resourceTypes ?? ["agent", "project"]), afterId === undefined ? undefined : gt(resourceLifecycleEvents.id, Number(afterId))))
|
||||
.orderBy(asc(resourceLifecycleEvents.id)).limit(limit);
|
||||
// One pass picks each resource's deliverable head: its unacknowledged creation, else its
|
||||
// oldest unacknowledged event. This is the set `firstForResource` admits, without probing
|
||||
// every acknowledged older event once per pending candidate. The page filter stays outside.
|
||||
const heads = db.selectDistinctOn([resourceLifecycleEvents.resourceType, resourceLifecycleEvents.resourceId]).from(resourceLifecycleEvents)
|
||||
.where(and(eq(resourceLifecycleEvents.companyId, companyId), inArray(resourceLifecycleEvents.resourceType, resourceTypes ?? ["agent", "project"]), unacknowledged))
|
||||
.orderBy(resourceLifecycleEvents.resourceType, resourceLifecycleEvents.resourceId, sql`${resourceLifecycleEvents.action} <> 'create'`, resourceLifecycleEvents.id)
|
||||
.as("heads");
|
||||
const rows = await db.select().from(heads)
|
||||
.where(afterId === undefined ? undefined : gt(heads.id, Number(afterId)))
|
||||
.orderBy(asc(heads.id)).limit(limit);
|
||||
return rows.map(row => ({ ...row, id: String(row.id), createdAt: row.createdAt.toISOString() })) as ResourceLifecycleEvent[];
|
||||
},
|
||||
/** Whether `list` would return any event. Every resource with an unacknowledged event has a head. */
|
||||
async hasPending(companyId: string, resourceTypes?: ResourceLifecycleEvent["resourceType"][]): Promise<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;
|
||||
},
|
||||
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");
|
||||
const commit = () => db.transaction(async tx => {
|
||||
|
||||
Reference in new issue
Block a user