mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-15 20:53:23 +02:00
fix(routines): reject lock contention before receipt mutations
Keep owned-routine lock acquisition nonblocking after run authority locks. A scheduled firing can finish, and the same mutation may retry without a saved partial receipt. The real concurrency regression is added; local PostgreSQL bootstrap failed before assertions, so its execution still requires CI. Server typecheck passes. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
79e5d38139
commit
c7afea8dab
2 files changed
+85
-3
No files matched your search
@@ -5,7 +5,7 @@ import { createRoutineSchema, updateRoutineSchema, createRoutineTriggerSchema, u
|
||||
import { routineService } from "../routines.js";
|
||||
import { documentAnnotationService } from "../document-annotations.js";
|
||||
import { persistActivity } from "../activity-log.js";
|
||||
import { forbidden, notFound } from "../../errors.js";
|
||||
import { conflict, forbidden, notFound } from "../../errors.js";
|
||||
|
||||
export const manageRoutineInputSchema = z.object({
|
||||
idempotencyKey: z.string().trim().min(1).max(240),
|
||||
@@ -33,7 +33,20 @@ export const manageRoutineInputSchema = z.object({
|
||||
type Binding = { companyId: string; agentId: string; issueId: string; runId: string };
|
||||
|
||||
export async function lockOwnedRoutine(db: Db, binding: Binding, id: string) {
|
||||
const [routine] = await db.select().from(routines).where(and(eq(routines.id, id), eq(routines.companyId, binding.companyId))).for("update");
|
||||
// Execution authorization already holds the agent and issue locks. A firing
|
||||
// holds the routine while assigning its issue and waking the agent, so waiting
|
||||
// here would form a cycle. Reject contention and roll back the receipt before
|
||||
// the agent retries; this also covers saved create receipts.
|
||||
let routine: typeof routines.$inferSelect | undefined;
|
||||
try {
|
||||
[routine] = await db.select().from(routines).where(and(eq(routines.id, id), eq(routines.companyId, binding.companyId))).for("update", { noWait: true });
|
||||
} catch (error) {
|
||||
const failure = error as { code?: string; cause?: { code?: string } } | null;
|
||||
if (failure?.code === "55P03" || failure?.cause?.code === "55P03") {
|
||||
throw conflict("Routine is busy. Retry the same request after the current firing or edit finishes.", { code: "routine_busy", retryable: true });
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
if (!routine) throw notFound("Routine not found");
|
||||
if (routine.assigneeAgentId !== binding.agentId) throw forbidden("Agents can only manage routines assigned to themselves");
|
||||
return routine;
|
||||
|
||||
@@ -2,7 +2,7 @@ import * as cloudIdentity from "../cloud-runtime-identity.js";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { upsertIssueDocumentSchema } from "@paperclipai/shared";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import {
|
||||
activityLog,
|
||||
agents,
|
||||
@@ -192,6 +192,75 @@ describe("PaperclipRunnerToolAuthority", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects a contended routine edit, lets its scheduled firing finish, and retries exactly once", async () => {
|
||||
const userId = "routine-concurrent-owner";
|
||||
await db.insert(authUsers).values({ id: userId, name: "Routine concurrency owner", email: "routine-concurrent@example.test", createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" });
|
||||
await db.update(issues).set({ responsibleUserId: userId }).where(eq(issues.id, issueId));
|
||||
try {
|
||||
const authority = new PaperclipRunnerToolAuthority(db, { companyId, agentId, issueId, runId });
|
||||
const created = await authority.execute({ tool: "manage_routine", callId: "concurrent-create", arguments: {
|
||||
idempotencyKey: "concurrent-create", action: "create", title: "Concurrent routine",
|
||||
schedule: { cronExpression: "0 9 * * 1-5", timezone: "America/Chicago" },
|
||||
} }) as { routineId: string; baseRevisionId: string };
|
||||
// A routine can outlive its original task. Avoid the parent's foreign-key
|
||||
// lock so this exercises the agent/routine cycle independently.
|
||||
await db.update(routines).set({ parentIssueId: null }).where(eq(routines.id, created.routineId));
|
||||
const due = new Date("2026-10-07T14:00:00Z");
|
||||
await db.update(routineTriggers).set({ nextRunAt: due }).where(eq(routineTriggers.routineId, created.routineId));
|
||||
let enteredWake = false;
|
||||
let releaseWake!: () => void;
|
||||
const wakeGate = new Promise<void>(resolve => { releaseWake = resolve; });
|
||||
const scheduler = routineService(db, { runtimeEnv: {}, heartbeat: { wakeup: async () => {
|
||||
enteredWake = true;
|
||||
await wakeGate;
|
||||
// The real assignment path needs this agent row while holding the routine.
|
||||
await db.update(agents).set({ status: "active" }).where(eq(agents.id, agentId));
|
||||
return null;
|
||||
} } });
|
||||
const firing = scheduler.tickScheduledTriggers(due);
|
||||
void firing.catch(() => undefined);
|
||||
const guardedDb = new Proxy(db, { get(target, key, receiver) {
|
||||
if (key !== "transaction") return Reflect.get(target, key, receiver);
|
||||
return (effect: (tx: unknown) => Promise<unknown>) => target.transaction(async tx => {
|
||||
// Bound failure cleanup even on the old blocking implementation. The
|
||||
// shorter race below must succeed before this diagnostic limit fires.
|
||||
await tx.execute(sql`set local lock_timeout = '3000ms'`);
|
||||
return effect(tx);
|
||||
});
|
||||
} });
|
||||
const editingAuthority = new PaperclipRunnerToolAuthority(guardedDb, { companyId, agentId, issueId, runId });
|
||||
const input = { idempotencyKey: "concurrent-edit", action: "update", routineId: created.routineId,
|
||||
baseRevisionId: created.baseRevisionId, title: "After concurrent firing" };
|
||||
let edit: Promise<unknown> | undefined;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await vi.waitFor(() => expect(enteredWake).toBe(true), { timeout: 5000 });
|
||||
edit = editingAuthority.execute({ tool: "manage_routine", callId: "concurrent-edit", arguments: input });
|
||||
const outcome = await Promise.race([
|
||||
edit.then(value => ({ kind: "applied", value }), error => ({ kind: "rejected", error })),
|
||||
new Promise<{ kind: "timeout" }>(resolve => { timer = setTimeout(() => resolve({ kind: "timeout" }), 2000); }),
|
||||
]);
|
||||
expect(outcome).toMatchObject({ kind: "rejected", error: { status: 409, details: { code: "routine_busy", retryable: true } } });
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect((run!.resultJson as { semanticToolReceipts: object }).semanticToolReceipts).not.toHaveProperty(input.idempotencyKey);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
releaseWake();
|
||||
await Promise.allSettled([firing, ...(edit ? [edit] : [])]);
|
||||
}
|
||||
expect(await firing).toEqual({ triggered: 1 });
|
||||
const updated = await authority.execute({ tool: "manage_routine", callId: "concurrent-retry", arguments: input });
|
||||
expect(updated).toMatchObject({ title: input.title });
|
||||
expect(await authority.execute({ tool: "manage_routine", callId: "concurrent-replay", arguments: input })).toEqual(updated);
|
||||
expect(await db.select().from(routineRuns).where(eq(routineRuns.routineId, created.routineId))).toMatchObject([
|
||||
{ status: "issue_created", source: "schedule" },
|
||||
]);
|
||||
} finally {
|
||||
await db.update(issues).set({ responsibleUserId: null }).where(eq(issues.id, issueId));
|
||||
}
|
||||
});
|
||||
|
||||
it("advertises only real bindings and reads the bound task", async () => {
|
||||
const authority = new PaperclipRunnerToolAuthority(db, {
|
||||
companyId,
|
||||
|
||||
Reference in new issue
Block a user