mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 11:20:42 +01:00
perf(issues): filter queued comment wakes by task (#17613)
Apply the existing typed task identity filter in PostgreSQL before returning queued-comment wake controls. Preserve queue selection, ordering, payloads, and current run ownership checks. All 84 route tests, build, typecheck, exact-head Greptile 5/5, and required CI pass. The old source fails the new deferred and queued byte budgets; corrected reads preserve the complete response and all saved rows. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
a09ef344b3
commit
bf9a8551c3
3 files changed
+45
No files matched your search
@@ -238,6 +238,11 @@ for chain checks. The selected agent's full configuration, current monthly spend
|
||||
and existing response redaction remain available. Every call reads current data;
|
||||
these lookups do not cache authorization or agent configuration.
|
||||
|
||||
Queued-comment display filters wake requests by the current company, assigned
|
||||
agent, status, and exact JSON string task identity in SQL. Other tasks' comment
|
||||
identifiers stay in the database. The display projection, ordering, current run
|
||||
ownership checks, and persisted wake payloads retain their existing contract.
|
||||
|
||||
## 1. Embedded PostgreSQL — zero config
|
||||
|
||||
If you don't set `DATABASE_URL`, the server automatically starts an embedded PostgreSQL instance and manages a local data directory.
|
||||
|
||||
@@ -252,6 +252,45 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
||||
} finally { capture.flush(); await observed.$client.end(); }
|
||||
});
|
||||
|
||||
it.each(["deferred", "queued"])("filters unrelated task wakes before reading the %s comment queue", async state => {
|
||||
const f = await seedQueue();
|
||||
if (state === "queued") await promoteQueue(f);
|
||||
const path = `/api/issues/${f.issueId}/queued-comments`;
|
||||
const expected = await request(app(f.companyId)).get(path).expect(200);
|
||||
const unrelated = Array.from({ length: 80 }, (_, index) => ({
|
||||
companyId: f.companyId, agentId: f.agentId, source: "issue_comment",
|
||||
status: index % 2 ? "queued" : "deferred_issue_execution",
|
||||
requestedAt: new Date("2020-01-01T00:00:00Z"),
|
||||
payload: { issueId: randomUUID(), _paperclipWakeContext: {
|
||||
wakeCommentIds: Array.from({ length: 60 }, () => randomUUID()),
|
||||
} },
|
||||
}));
|
||||
const malformed = [null, [f.issueId], { id: f.issueId }, f.issueId.toUpperCase()];
|
||||
await db.insert(agentWakeupRequests).values([
|
||||
...unrelated,
|
||||
...malformed.map(issueId => ({ companyId: f.companyId, agentId: f.agentId,
|
||||
source: "issue_comment", status: "deferred_issue_execution",
|
||||
requestedAt: new Date("2020-01-01T00:00:00Z"),
|
||||
payload: { issueId, _paperclipWakeContext: { wakeCommentIds: f.commentIds } },
|
||||
})),
|
||||
]);
|
||||
const observed = createDb(tempDb!.connectionString), events: Array<Record<string, unknown>> = [], now = Date.now();
|
||||
const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
|
||||
installDatabaseResponseCapture(observed, { agentWakeupRequests }, capture);
|
||||
try {
|
||||
const response = await request(app(f.companyId, "queue-owner", undefined, observed)).get(path).expect(200);
|
||||
expect(response.body).toEqual(expected.body);
|
||||
capture.flush();
|
||||
const reads = events.filter(event => event.table === "agent_wakeup_requests" && event.operation === "select");
|
||||
expect(reads.length).toBeGreaterThan(0);
|
||||
expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(4096);
|
||||
expect(reads.reduce((rows, event) => rows + Number(event.rows), 0)).toBe(1);
|
||||
expect(await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(85);
|
||||
await request(app(randomUUID(), "queue-owner", undefined, observed)).get(path).expect(404);
|
||||
} finally { capture.flush(); await observed.$client.end(); }
|
||||
});
|
||||
|
||||
it.each(["cancelled", "running"])("rejects a late Done from an interrupted %s task run at the write boundary", async status => {
|
||||
const seeded = await seedQueue();
|
||||
await db.update(heartbeatRuns).set({ status, resultJson: { executionCancellation: { state: "requested" } } })
|
||||
|
||||
@@ -7320,6 +7320,7 @@ export function issueRoutes(
|
||||
and(
|
||||
eq(agentWakeupRequests.companyId, issue.companyId),
|
||||
eq(agentWakeupRequests.agentId, issue.assigneeAgentId),
|
||||
sql`${agentWakeupRequests.payload}->'issueId' = to_jsonb(${issue.id}::text)`,
|
||||
inArray(agentWakeupRequests.status, [
|
||||
"deferred_issue_execution",
|
||||
"queued",
|
||||
|
||||
Reference in new issue
Block a user