fix: retry transient continuation admission locks (#13290)
## Thinking Path > - Paperclip manages AI agents and the tasks that they execute. > - The run scheduler checks continuation authority before it starts a provider. > - This check uses database locks to order execution against conversation closure. > - A short lock conflict could fail a valid user follow-up before the provider started. > - This pull request retries the admission transaction after the locks are released. > - Valid work can start after normal contention, while closure and cancellation still stop execution. ## Linked Issues or Issue Description Refs #13038. Related continuation work: #13270 and #13239. **What happened?** A user comment started a run through the automation queue. Its source records and admission marker were valid. A database lock conflict at dispatch caused `chat_control_recovery_proof_unresolved` and stopped automatic recovery. The provider received no work. **Expected behavior** Retry short database lock conflicts before failing admission. Read current ownership and conversation-close evidence on each attempt. Do not retry provider execution. **Steps to reproduce** 1. Queue a user follow-up through the automation transport. 2. Hold the task row lock in a separate transaction at the dispatch boundary. 3. Release the lock after 250 ms. 4. Before this fix, the run fails before provider dispatch. With this fix, the run passes admission once the lock is released. **Paperclip version or commit** Reproduced on base commit `1c4bcff2b`. Disabling the new retry reproduces the original error in the regression test. **Deployment mode** Self-hosted server with PostgreSQL. Regression tests use embedded PostgreSQL. ## What Changed - Retry rolled-back admission transactions after lock conflicts, with up to 50 waits of 100 ms. - Keep queue claims nonblocking. Keep provider dispatch outside the retried transaction. - Recheck current run state and committed close evidence after every conflict. - Explain persistent database contention in the exhausted admission error. - Add real database contention tests and bounded retry tests. Document the behavior. ## Verification - `pnpm exec vitest run server/src/__tests__/heartbeat-process-recovery.test.ts server/src/services/chat-control-admission-retry.test.ts`: 273 passed. This includes task, wake, and run locks, the native runner, close/cancel races, unrelated failures, and retry exhaustion. - Regression proof: disabling retries makes the user-follow-up test fail with `chat_control_recovery_proof_unresolved`. - `pnpm -r typecheck`: passed. - `pnpm build`: passed. - `pnpm test:run`: stopped after all equivalent CI server/workspace shards passed. The local run exposed a missing `fake-codex-app-server` fixture binary in the fresh worktree; after `pnpm --filter @paperclipai/paperclip-runner run build:rust`, the complete affected `native-session-resume.test.ts` suite passes (37 tests). - Greptile: 5/5, no findings, on commit `c974a496a`. - CI: all 31 checks passed on commit `c974a496a`, including all server/workspace test shards, browser tests, typecheck, runner verification, build, and the release dry run. [CI run](https://github.com/paperclipai/paperclip/actions/runs/34654770074). ## Risks - A contended dispatch can wait through 50 short delays, plus transaction time. - Persistent contention still fails closed after the retry budget. Invalid source evidence fails without retrying admission. - No schema, permission, provider retry budget, or queue-claim behavior changes. ## Model Used OpenAI Codex, GPT-6. The exact deployed model ID and context-window size are not exposed in this session. Used reasoning, repository inspection, code editing, and command execution. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
9031516a7e
commit
a38ccf9972
|
|
@ -837,6 +837,14 @@ Every continuation carries the triggering request, ordered user direction, inter
|
|||
|
||||
### Interrupted conversation continuation
|
||||
|
||||
Before provider dispatch, chat-control admission retries transient database lock
|
||||
contention with up to 50 waits of 100 ms. Each attempt starts a new transaction
|
||||
and rechecks the current run and committed conversation-close evidence. No lock
|
||||
is held between attempts, and no provider call is retried. Queue claims remain
|
||||
nonblocking. Persistent contention retains the bounded admission failure, with
|
||||
an explicit database-lock error; missing or invalid source evidence still stops
|
||||
the run without retrying the admission check.
|
||||
|
||||
An interrupted conversation does not permanently block its task. For local conversational adapters, Paperclip starts a new bounded turn with the existing session when compatible, or the full task conversation when the session is unavailable. The prompt says: “Your previous run was interrupted. Continue from where you left off.” The agent decides what remains from the history and latest user request. Paperclip never automatically replays recorded tool calls. Unknown past action outcomes are not a task-wide execution gate, and no action-reconciliation questionnaire is required.
|
||||
|
||||
Shutdown, process loss, and provider failure use the existing durable failure retry counter and delay. Ordinary failure recovery permits at most two automatic retries in a failure chain. Accepted-interaction infrastructure recovery retains its existing bounded policy. Repeated scheduler visits reuse the same successor; restarting the server does not reset the counter. After exhaustion, automatic attempts stop. A new explicit user message can start a fresh run and failure budget. Productive max-turn continuation and confirmed workspace waits keep their separate existing semantics.
|
||||
|
|
|
|||
|
|
@ -11335,6 +11335,96 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
},
|
||||
);
|
||||
|
||||
it.each(["issue", "wake", "run", "native", "close", "cancel"] as const)(
|
||||
"rechecks admission after transient database contention: %s",
|
||||
async (mode) => {
|
||||
const source = await seedCommittedChatControlStop();
|
||||
await db.update(chatPublications).set({ state: "pending" })
|
||||
.where(eq(chatPublications.id, source.publicationId));
|
||||
const child = await seedChatAutomaticChild(source);
|
||||
// Board comments use the automation transport but are fresh user work.
|
||||
if (!["close", "cancel"].includes(mode)) {
|
||||
await db.update(agentWakeupRequests).set({
|
||||
requestedByActorType: "user", requestedByActorId: "responsible-user",
|
||||
reason: "issue_commented",
|
||||
}).where(eq(agentWakeupRequests.id, child.wakeupRequestId));
|
||||
await db.update(heartbeatRuns).set({ retryOfRunId: null })
|
||||
.where(eq(heartbeatRuns.id, child.runId));
|
||||
}
|
||||
if (mode === "native") {
|
||||
await db.update(agents).set({
|
||||
adapterType: "paperclip_runner",
|
||||
adapterConfig: { provider: "codex", model: "gpt-5.6-luna" },
|
||||
}).where(eq(agents.id, source.agentId));
|
||||
}
|
||||
const factory = vi.fn(() => { throw new NativeRunnerOwnershipUnverifiedError(); });
|
||||
let release!: () => void;
|
||||
let locked: Promise<unknown> | undefined;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const heartbeat = heartbeatService(db, {
|
||||
nativeSessionBackendFactory: factory,
|
||||
beforeChatControlRecoveryCheck: async ({ stage }) => {
|
||||
if (stage !== "dispatch") return;
|
||||
let ready!: () => void;
|
||||
const acquired = new Promise<void>((resolve) => { ready = resolve; });
|
||||
const held = new Promise<void>((resolve) => { release = resolve; });
|
||||
locked = db.transaction(async (tx) => {
|
||||
if (mode === "wake") {
|
||||
await tx.select().from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, child.wakeupRequestId)).for("update");
|
||||
} else if (mode === "run" || mode === "cancel") {
|
||||
await tx.select().from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, child.runId)).for("update");
|
||||
} else if (mode === "close") {
|
||||
await tx.select().from(chatConversations)
|
||||
.where(eq(chatConversations.id, source.conversationId)).for("update");
|
||||
} else {
|
||||
await tx.select().from(issues)
|
||||
.where(eq(issues.id, source.issueId)).for("update");
|
||||
}
|
||||
ready();
|
||||
await held;
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
expect(factory).not.toHaveBeenCalled();
|
||||
if (mode === "close") {
|
||||
await tx.update(chatPublications).set({ state: "published" })
|
||||
.where(eq(chatPublications.id, source.publicationId));
|
||||
} else if (mode === "cancel") {
|
||||
await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, child.runId));
|
||||
}
|
||||
});
|
||||
await acquired;
|
||||
timer = setTimeout(release, 250);
|
||||
},
|
||||
});
|
||||
try {
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
release?.();
|
||||
await locked;
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
}
|
||||
expect(locked).toBeDefined();
|
||||
const settled = await heartbeat.getRun(child.runId);
|
||||
expect(settled?.errorCode).not.toBe(CHAT_CONTROL_RECOVERY_UNRESOLVED_CODE);
|
||||
if (mode === "native") {
|
||||
expect(factory).toHaveBeenCalledTimes(1);
|
||||
expect(readChatControlRecoveryAdmission(settled!)).toBe("admitted");
|
||||
} else if (mode === "close" || mode === "cancel") {
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
expect(settled?.status).toBe("cancelled");
|
||||
if (mode === "close") expect(settled?.errorCode).toBe(CHAT_CONTROL_RECOVERY_STOP_CODE);
|
||||
} else {
|
||||
expect(mockAdapterExecute).toHaveBeenCalledTimes(1);
|
||||
expect(settled?.status).toBe("succeeded");
|
||||
expect(readChatControlRecoveryAdmission(settled!)).toBe("admitted");
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("defers unresolved automatic ancestry at claim and records a distinct nonretrying failure after claim", async () => {
|
||||
const source = await seedCommittedChatControlStop();
|
||||
await db
|
||||
|
|
|
|||
|
|
@ -0,0 +1,33 @@
|
|||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { retryChatControlAdmission } from "./chat-control-admission-retry.js";
|
||||
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
it("retries a rolled-back lock conflict and returns the fresh admission result", async () => {
|
||||
vi.useFakeTimers();
|
||||
const attempt = vi.fn()
|
||||
.mockRejectedValueOnce(new Error("query failed", { cause: { code: "55P03" } }))
|
||||
.mockResolvedValueOnce(null);
|
||||
const result = retryChatControlAdmission(attempt);
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
await expect(result).resolves.toBeNull();
|
||||
expect(attempt).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not retry unrelated database failures", async () => {
|
||||
const error = new Error("constraint violation", { cause: { code: "23505" } });
|
||||
const attempt = vi.fn().mockRejectedValue(error);
|
||||
await expect(retryChatControlAdmission(attempt)).rejects.toBe(error);
|
||||
expect(attempt).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("stops persistent contention after fifty delays", async () => {
|
||||
vi.useFakeTimers();
|
||||
const error = new Error("query failed", { cause: { code: "55P03" } });
|
||||
const attempt = vi.fn().mockRejectedValue(error);
|
||||
const rejected = expect(retryChatControlAdmission(attempt)).rejects.toBe(error);
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
await rejected;
|
||||
expect(attempt).toHaveBeenCalledTimes(51);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
import { isExternalChatWaitAuthorizationContention } from "./native-runtime/chat-attachment-reuse.js";
|
||||
|
||||
/** Retry only a rolled-back admission transaction, never provider execution. */
|
||||
export async function retryChatControlAdmission<T>(attempt: () => Promise<T>): Promise<T> {
|
||||
for (let retry = 0; ; retry += 1) {
|
||||
try {
|
||||
return await attempt();
|
||||
} catch (error) {
|
||||
if (retry >= 50 || !isExternalChatWaitAuthorizationContention(error)) throw error;
|
||||
}
|
||||
// The previous transaction has released all locks. The next attempt must
|
||||
// read current run ownership and close evidence again before admission.
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
}
|
||||
}
|
||||
|
|
@ -547,6 +547,7 @@ import { extractSkillMentionIds, isUuidLike } from "@paperclipai/shared";
|
|||
import { evaluateCodexCredentialReadiness } from "@paperclipai/adapter-codex-local/server";
|
||||
import { environmentService } from "./environments.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js";
|
||||
import { retryChatControlAdmission } from "./chat-control-admission-retry.js";
|
||||
import {
|
||||
environmentRuntimeService,
|
||||
type ProviderResourceDisposition,
|
||||
|
|
@ -799,7 +800,7 @@ function nonRetryablePreflightFailureCode(error: unknown): string | null {
|
|||
class ChatControlRecoveryUnresolvedError extends Error {
|
||||
constructor() {
|
||||
super(
|
||||
"Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry.",
|
||||
"Run admission could not acquire its database locks after bounded retries. No provider work started. Review database contention and send a fresh request; this attempt will not automatically retry.",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -16401,9 +16402,11 @@ export function heartbeatService(
|
|||
}
|
||||
let terminal: typeof heartbeatRuns.$inferSelect | null = null;
|
||||
try {
|
||||
const result = await db.transaction(async (tx) => {
|
||||
const attempt = () => db.transaction(async (tx) => {
|
||||
terminal = null;
|
||||
// Same queue-edit lock order, then the close committer's conversation
|
||||
// row. NOWAIT makes contention a scoped deferral, never authority.
|
||||
// row. NOWAIT releases partial locks on contention. Claim defers to the
|
||||
// queue; dispatch retries this transaction before considering failure.
|
||||
const [issue] = await tx
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
|
|
@ -16567,6 +16570,9 @@ export function heartbeatService(
|
|||
);
|
||||
return null;
|
||||
});
|
||||
const result = stage === "dispatch"
|
||||
? await retryChatControlAdmission(attempt)
|
||||
: await attempt();
|
||||
if (terminal) {
|
||||
const settled = terminal as typeof heartbeatRuns.$inferSelect;
|
||||
publishLiveEvent({
|
||||
|
|
|
|||
Loading…
Reference in New Issue