mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-14 17:22:00 +01:00
fix(storage): wait out a store outage on a run's critical path instead of failing it (#20299)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A cloud instance keeps run logs and evidence in object storage, so a run touches the store as soon as it starts > - In storage chaos wave `w-20261109T145628-s9902` (ST1: 75–176 s windows where every S3 call returns 503), only 1 of 8 tasks finished. A run starting or resuming during an outage failed within a second as `adapter_failed`, generic recovery burned its attempts seconds apart, the task was escalated to the board, and the user saw the raw S3 error > - This pull request treats a transient storage error as an infrastructure wait: the run is retried on a minute-scale schedule, rapid recovery is suppressed, the board shows it as working, and the run error says storage was unavailable > - The benefit is that a store outage of several minutes delays work instead of failing and stranding it ## Linked Issues or Issue Description Refs #19831. Storage chaos wave `w-20261109T145628-s9902` (master `610b54f551`, `--storage s3`, ST1). **Where the run fails.** At run start or resume, `run-output` `begin` opens the run log: `getRunLogStore().begin` → object run log store → object byte journal `claim` → S3 conditional PUT. The journal retries only 3 times within about 300 ms, so a 503 escapes and fails the run as `adapter_failed`. The async stack ends in the SDK middleware with no app frame. `log_store` stays null on the failed run, which pins the claim. **What followed.** - `failRun` released the issue into generic immediate recovery. Recovery retried 1–2 s apart inside the outage (t01: 14:58:33, :35, :59, 14:59:01), then stopped and opened a board-owned `stranded_assigned_issue` action. These are the open recovery actions on CHA-1, CHA-4, CHA-7 and CHA-8; a later Retry hit the outage again and escalated the same way. - The run error was the raw S3 text ("Injected storage outage"). **Slow store (ST2, `w-20261109T151946-s9903`).** With about 9.7 s of added latency per call, the same claim hit the journal's 15 s per-operation deadline (`AbortSignal.timeout(15_000)`). The run failed after about 36 s with `AbortError: Request aborted` and burned 3 runs the same way; 4 of 8 tasks needed a Retry. t02/t03 (Retry parked with `process_identity_missing`) are a separate native-recovery class and are handled in a follow-up. **Daytona storage chaos ST1 (`w-20261109T161202-s9912`, 1/8):** with S3 returning 503 for 72 to 154 s at a time: - Workspace restore failed at once (`remote_workspace_restore_failed: step=snapshot_download; storage=provider_error`), three times in 40 s, then `setup_failed`. It left an active recovery action behind. - Runs reattached after a restart died within 2 s on the run-log claim, then became reconciliation-required. - Setup failed on other storage reads. - 6 of 7 failures needed a UI Retry. So the rule here is one class rule: a transient store error on a run's critical path waits, it never fails the run. ## What Changed - New `server/src/storage/storage-unavailable.ts`: - `isStorageUnavailableError` reads the recorded storage failure diagnostics first, then the SDK error shape. Unavailable means 429/500/502/503/504, the S3 throttling and timeout names, network codes (`ECONNRESET`, `ECONNREFUSED`, `ETIMEDOUT`, `EPIPE`, `EAI_AGAIN`, `ENOTFOUND`), or an `AbortError` whose cause is a `TimeoutError` (a per-operation deadline). 403, 404 and 412 are not outages, and neither is an abort caused by the run's own Stop. - It defines the `storage_unavailable` code, the `storage_unavailable_wait` retry reason, the retry policy (60 s × 10 attempts, about 10 minutes) and the user-facing message. - `server/src/services/heartbeat/run-completion.ts`: - `failRun` classifies such a failure as `storage_unavailable` with the plain message. - Unless native recovery owns the run, it schedules the storage wait with `scheduleBoundedRetryForRun`, skips the interaction-continuation retry, and releases the issue with `suppressImmediateRecovery`. - `failRunSetup` uses the same policy for this code. - `server/src/services/heartbeat/retries.ts`: `storage_unavailable` joins the transient infrastructure codes. It gets its own retry budget and policy, like `runner_revision_mismatch`. - `server/src/services/task-interruption-presentation.ts`: `storage_unavailable_wait` is a resource wait, so the board shows the task as working with no Retry while it waits. - **One shared outage wait for the run's critical path** (`server/src/storage/storage-unavailable.ts`): - `retryThroughStorageOutage` retries a transient store failure with backoff (2 s doubling to 30 s) until that storage operation's outage has lasted 5 minutes, then gives the error back. The outermost wait of an operation records its first failure, and nested and parallel waits inside it share that start; no other request's success extends it. Shared probes and shared reads run under their own scope, so one run's Stop never cancels work another run waits on. A stop signal ends the wait, and non-outage errors are never retried. - Inside `waitOutStorageOutages`, every replayable S3 request waits instead of failing: get, head, delete, buffered and file puts, and multipart create, file parts and complete. A consumed upload stream is never resent, and requests outside the scope keep failing fast. - The scope covers run execution (`heartbeat.ts` `executeRun`: setup, workspace restore, run-log claim and the run itself), plus `loadWorkspaceSnapshotArchive` and `uploadWorkspaceSnapshot`, which copy-back and recovery workers use too. - The run-log claim retries as a whole, because the journal bounds each operation with a 15 s deadline. - Past the budget, the run takes the bounded `storage_unavailable` wait below. - The classifier follows the error's `cause` chain, so a restore failure wrapping an outage counts. `failRunSetup` checks storage before restore failures, so that case takes the storage wait instead of two quick retries and an escalation (the R6 leak). - `object-byte-journal.ts`: when its deadline expires, the journal rethrows the recorded storage error instead of a bare `TimeoutError`. ## Verification Red/green (each new test fails against master and passes with this PR): - `run-completion.test.ts` "waits a minute and suppresses rapid recovery when a run fails because storage is unavailable (storage chaos ST1)". An S3-shaped 503 records `storage_unavailable` with the plain message (no raw S3 text), schedules `STORAGE_UNAVAILABLE_RETRY`, releases with `suppressImmediateRecovery: true`, and skips the continuation retry. "keeps an ordinary adapter failure on the normal recovery path" guards the default. - `task-interruption-query.test.ts` "does not present an overdue storage_unavailable_wait wait as a failed attempt": working, `canRetry: false`. - `storage-unavailable.test.ts`: which errors count as an outage and which do not, including recorded diagnostics, a deadline-aborted call (ST2), and a Stop-aborted call that must not count. Static checks: - `tsc --noEmit` passes for `server`. Live verification: Phase 3 will rerun ST1 (seed 9902) with the MinIO fault proxy after merge. - `server/src/__tests__/storage-outage-wait.test.ts` runs against a local S3 that answers 503 for 3 s: - Inside the scope, a snapshot `getObject` and a run-log conditional PUT succeed once the store is back. - Outside it, a request fails fast. - A stop ends the wait. - Unit cases cover the budget, non-outage errors, and a `RemoteWorkspaceRestoreError` wrapping an outage. - `run-completion.test.ts`: a setup failure from a restore that wrapped an outage takes the storage wait (red before). - `object-byte-journal.test.ts`: a real `claim` whose head PUT hits its deadline is classified as a storage outage (red before). ## Risks - A storage outage now holds a task for up to about 10 minutes of automatic waits instead of escalating within seconds. A longer outage still ends in the existing board recovery. - A run whose failure is misclassified as storage-unavailable waits instead of using fast recovery. The classifier accepts only server-side or network failures from the storage SDK, never 4xx answers. - No schema change. - A run can now wait until a storage outage has lasted 5 minutes before it fails over to the bounded wait. It shows as working throughout, and Stop ends the wait. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits). ## 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
0116e45958
commit
4a2eb920cf
25 files changed
+821
-80
No files matched your search
@@ -29,7 +29,7 @@ describe("failure attempts across resource waits", () => {
|
||||
});
|
||||
|
||||
describe("persisted independent accounting", () => {
|
||||
it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", "runner_revision_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => {
|
||||
it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", "runner_revision_wait", "storage_unavailable_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => {
|
||||
expect(executionFailureRetryCount({ scheduledRetryReason, scheduledRetryAttempt: 20,
|
||||
contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 1 } },
|
||||
})).toBe(2);
|
||||
@@ -107,3 +107,17 @@ describe("interrupted preparation retries", () => {
|
||||
expect(accountingForScheduledRetry(retry, "transient_failure", 2).failureRetries).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("storage outage waits", () => {
|
||||
it("keep their count through an intervening workspace wait", () => {
|
||||
const afterStorage = { scheduledRetryReason: "storage_unavailable_wait", scheduledRetryAttempt: 7,
|
||||
contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 1, maxTurnContinuations: 0 } } };
|
||||
const ledger = accountingForScheduledRetry(afterStorage, "workspace_busy", 1);
|
||||
expect(ledger).toMatchObject({ storageWaits: 7, failureRetries: 1 });
|
||||
const afterWorkspaceWait = { scheduledRetryReason: "workspace_busy", scheduledRetryAttempt: 1,
|
||||
contextSnapshot: { executionRetryAccounting: ledger } };
|
||||
expect(executionRetryAttemptCount(afterWorkspaceWait, "storage_unavailable_wait")).toBe(7);
|
||||
expect(accountingForScheduledRetry(afterWorkspaceWait, "storage_unavailable_wait", 8)).toMatchObject({ storageWaits: 8, failureRetries: 1 });
|
||||
expect(executionFailureRetryCount(afterWorkspaceWait)).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -1,4 +1,5 @@
|
||||
import { RUNNER_REVISION_WAIT_RETRY_REASON } from "./runner-revision-mismatch.js";
|
||||
import { STORAGE_UNAVAILABLE_WAIT_RETRY_REASON } from "../storage/storage-unavailable.js";
|
||||
import { PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT } from "./native-runtime/provider-attempt-outcome.js";
|
||||
|
||||
/**
|
||||
@@ -14,7 +15,7 @@ export const INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS = PROVIDER_ATTEMPT_CREDI
|
||||
/** Retry lanes that keep their own counter and never spend failures. */
|
||||
const UNCHARGED_RETRY_LANES: ReadonlySet<string> = new Set([
|
||||
"workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", RUNNER_REVISION_WAIT_RETRY_REASON,
|
||||
INTERRUPTED_PREPARATION_RETRY_REASON,
|
||||
INTERRUPTED_PREPARATION_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON,
|
||||
]);
|
||||
|
||||
type RetryRun = {
|
||||
@@ -34,6 +35,8 @@ export interface ExecutionRetryAccounting {
|
||||
* wait retry in between never resets it. Omitted while zero.
|
||||
*/
|
||||
interruptedPreparations?: number;
|
||||
/** Storage-outage waits in this chain, kept across intervening resource waits. Omitted while zero. */
|
||||
storageWaits?: number;
|
||||
}
|
||||
|
||||
function count(value: unknown): number | null {
|
||||
@@ -48,8 +51,10 @@ function savedAccounting(run: RetryRun): ExecutionRetryAccounting | null {
|
||||
const maxTurnContinuations = count(saved.maxTurnContinuations);
|
||||
if (saved.version !== 1 || failureRetries === null || maxTurnContinuations === null) return null;
|
||||
const interruptedPreparations = count(saved.interruptedPreparations) ?? 0;
|
||||
const storageWaits = count(saved.storageWaits) ?? 0;
|
||||
return { version: 1, failureRetries, maxTurnContinuations,
|
||||
...(interruptedPreparations > 0 ? { interruptedPreparations } : {}) };
|
||||
...(interruptedPreparations > 0 ? { interruptedPreparations } : {}),
|
||||
...(storageWaits > 0 ? { storageWaits } : {}) };
|
||||
}
|
||||
|
||||
function historicalFailureCount(run: RetryRun): number {
|
||||
@@ -72,12 +77,15 @@ export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccountin
|
||||
const nonFailureLane = reason === "max_turns_continuation" || reason === "issue_disposition_repair" || UNCHARGED_RETRY_LANES.has(reason);
|
||||
const interruptedPreparations = Math.max(saved?.interruptedPreparations ?? 0,
|
||||
reason === INTERRUPTED_PREPARATION_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0);
|
||||
const storageWaits = Math.max(saved?.storageWaits ?? 0,
|
||||
reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0);
|
||||
return {
|
||||
version: 1,
|
||||
failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)),
|
||||
maxTurnContinuations: Math.max(saved?.maxTurnContinuations ?? 0,
|
||||
run.scheduledRetryReason === "max_turns_continuation" ? count(run.scheduledRetryAttempt) ?? 0 : 0),
|
||||
...(interruptedPreparations > 0 ? { interruptedPreparations } : {}),
|
||||
...(storageWaits > 0 ? { storageWaits } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -88,6 +96,7 @@ export function executionFailureRetryCount(run: RetryRun): number {
|
||||
|
||||
export function executionRetryAttemptCount(run: RetryRun, reason: string): number {
|
||||
if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) return executionRetryAccounting(run).interruptedPreparations ?? 0;
|
||||
if (reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) return executionRetryAccounting(run).storageWaits ?? 0;
|
||||
if (UNCHARGED_RETRY_LANES.has(reason)) {
|
||||
return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0;
|
||||
}
|
||||
@@ -99,6 +108,7 @@ export function accountingForScheduledRetry(run: RetryRun, reason: string, attem
|
||||
const accounting = executionRetryAccounting(run);
|
||||
if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt;
|
||||
else if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) accounting.interruptedPreparations = attempt;
|
||||
else if (reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) accounting.storageWaits = attempt;
|
||||
else if (!UNCHARGED_RETRY_LANES.has(reason)) accounting.failureRetries = attempt;
|
||||
return accounting;
|
||||
}
|
||||
@@ -1019,6 +1019,17 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps the user's continuation while it waits out a storage outage", async () => {
|
||||
const f = await seedTimedOutExplicitTurn();
|
||||
await db.update(heartbeatRuns).set({ status: "failed", errorCode: "storage_unavailable" }).where(eq(heartbeatRuns.id, f.parent.id));
|
||||
const scheduled = await heartbeatService(db).scheduleBoundedRetry(f.parent.id, { now: f.now });
|
||||
expect(scheduled.outcome).toBe("scheduled");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.parent.id));
|
||||
expect(successor).toMatchObject({ scheduledRetryReason: "storage_unavailable_wait",
|
||||
contextSnapshot: { explicitUserContinuation: { previousRunId: f.parent.id } } });
|
||||
await expect(dispatchExplicitRetry(f, successor!)).resolves.toBeTruthy();
|
||||
});
|
||||
|
||||
it("binds the initial user-message admission before any automatic retry", async () => {
|
||||
const f = await seedTimedOutExplicitTurn();
|
||||
await expect(dispatchExplicitRetry(f, f.parent)).resolves.toMatchObject({ interruptedRunId: f.sourceRunId });
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js";
|
||||
import { endStorageOutageWaitOn, runInOwnStorageScope } from "../storage/storage-unavailable.js";
|
||||
import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js";
|
||||
import { isProcessRunAdmissionClosed } from "./heartbeat/run-admission.js";
|
||||
import type { StopInvocationOptions } from "./heartbeat/run-control.js";
|
||||
@@ -1873,7 +1874,7 @@ export function heartbeatService(
|
||||
onNativeExecutionLeaseAcquired?: () => Promise<void>;
|
||||
} = {},
|
||||
) {
|
||||
return withRunExecutionTiming(runId, () => executeRunAttempt(runId, runOptions));
|
||||
return withRunExecutionTiming(runId, () => runInOwnStorageScope(() => executeRunAttempt(runId, runOptions)));
|
||||
}
|
||||
|
||||
const { deferRunForProviderLimit } = createProviderLimitWait(db, {
|
||||
@@ -2012,6 +2013,7 @@ export function heartbeatService(
|
||||
activeRunExecutions.add(run.id);
|
||||
const timePreparation = createHeartbeatPreparationTiming(run.id);
|
||||
const executionControl = createAdapterExecutionControl();
|
||||
endStorageOutageWaitOn(executionControl.controller.signal);
|
||||
// This coarse scope also covers host finalization after the adapter returns.
|
||||
// Nested scopes refine the label without making any termination claim.
|
||||
executionControl.phases.enter("host_execution");
|
||||
@@ -3230,6 +3232,7 @@ export function heartbeatService(
|
||||
appendRunEvent, getCurrentUserRedactionOptions,
|
||||
assertWriterAuthority: createRunLogWriterAuthority(db, run, runOptions.nativeLeaseOwner),
|
||||
runtimeLogSecretValues: observabilitySecretValues(observability),
|
||||
signal: executionControl.controller.signal,
|
||||
});
|
||||
const goalCheckpointSession: {
|
||||
current: {
|
||||
|
||||
@@ -128,6 +128,18 @@ describe("heartbeat retry module callbacks", () => {
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ends an answered interaction's storage wait at its limit and escalates it", async () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks();
|
||||
const run = { ...unitRun(), status: "failed", errorCode: "storage_unavailable", scheduledRetryReason: "storage_unavailable_wait", scheduledRetryAttempt: 10,
|
||||
contextSnapshot: { issueId: "issue", interactionId: "interaction-1", interactionStatus: "answered", interactionKind: "ask_user_questions",
|
||||
mutation: "interaction", retryReason: "storage_unavailable_wait",
|
||||
executionRetryAccounting: { version: 1, failureRetries: 0, maxTurnContinuations: 0 } } } as Run;
|
||||
const retry = await extracted.createHeartbeatRetries(database.db, deps).scheduleBoundedRetryForRun(run, { adapterType: "codex_local" } as Agent);
|
||||
expect(retry).toMatchObject({ outcome: "retry_exhausted", maxAttempts: 10 });
|
||||
expect(deps.escalatePlanApprovalResumeFailureNeedsAttention).toHaveBeenCalledWith(expect.objectContaining({ run, maxAttempts: 10 }));
|
||||
});
|
||||
|
||||
it("leaves successor work to the committed case handoff without retry escalation", async () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks();
|
||||
@@ -391,6 +403,40 @@ describePostgres("heartbeat retry module database wiring", () => {
|
||||
expect(wake!.payload).toMatchObject({ mutation: "interaction", interactionId: "interaction-1", retryReason: "runner_revision_wait" });
|
||||
});
|
||||
|
||||
it("waits out a storage outage in its own lane without spending failure retries", async () => {
|
||||
const { run, agent } = await fixture();
|
||||
const ledger = { executionRetryAccounting: { version: 1, failureRetries: 1, maxTurnContinuations: 0 } };
|
||||
let [failed] = await db.update(heartbeatRuns).set({ status: "failed", errorCode: "storage_unavailable",
|
||||
contextSnapshot: { ...run.contextSnapshot, ...ledger } }).where(eq(heartbeatRuns.id, run.id)).returning();
|
||||
for (let attempt = 1; attempt <= 10; attempt++) {
|
||||
const result = await boundRetries().retries.scheduleBoundedRetryForRun(failed!, agent, { now });
|
||||
expect(result.outcome).toBe("scheduled");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, failed!.id));
|
||||
expect(successor).toMatchObject({ scheduledRetryReason: "storage_unavailable_wait", scheduledRetryAttempt: attempt });
|
||||
expect(successor!.contextSnapshot).toMatchObject({ executionRetryAccounting: { failureRetries: 1 } });
|
||||
[failed] = await db.update(heartbeatRuns).set({ status: "failed", errorCode: "storage_unavailable", finishedAt: now,
|
||||
resultJson: run.resultJson })
|
||||
.where(eq(heartbeatRuns.id, successor!.id)).returning();
|
||||
await db.update(issues).set({ executionRunId: successor!.id, checkoutRunId: successor!.id }).where(eq(issues.id, run.issueId!));
|
||||
}
|
||||
expect(await boundRetries().retries.scheduleBoundedRetryForRun(failed!, agent, { now }))
|
||||
.toMatchObject({ outcome: "retry_exhausted", maxAttempts: 10 });
|
||||
});
|
||||
|
||||
it("keeps an answered interaction through a storage outage wait", async () => {
|
||||
const { run, agent } = await fixture();
|
||||
const snapshot = { ...run.contextSnapshot, interactionId: "interaction-1", interactionStatus: "answered", interactionKind: "ask_user_questions",
|
||||
mutation: "interaction", wakeReason: "issue_commented" };
|
||||
const [failed] = await db.update(heartbeatRuns).set({ errorCode: "storage_unavailable", contextSnapshot: snapshot })
|
||||
.where(eq(heartbeatRuns.id, run.id)).returning();
|
||||
expect((await boundRetries().retries.scheduleBoundedRetryForRun(failed!, agent, { now })).outcome).toBe("scheduled");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run.id));
|
||||
expect(successor).toMatchObject({ scheduledRetryReason: "storage_unavailable_wait" });
|
||||
expect(isResolvedInteractionContinuationWakeContext(successor!.contextSnapshot)).toBe(true);
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, successor!.wakeupRequestId!));
|
||||
expect(wake!.payload).toMatchObject({ mutation: "interaction", interactionId: "interaction-1", retryReason: "storage_unavailable_wait" });
|
||||
});
|
||||
|
||||
it("resolves current skills on a retry while retaining the failed attempt snapshot", async () => {
|
||||
const {run,agent}=await fixture();
|
||||
const snapshot={...run.contextSnapshot,skillVersions:{"company/review":"v1"},skillSelectionSnapshot:true,shadowedSkills:[{key:"old",shadowedBy:"company/review"}]};
|
||||
|
||||
@@ -33,6 +33,7 @@ import { ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE } from "../environment-provider-u
|
||||
import { REPOSITORY_CATALOG_UNAVAILABLE_CODE } from "../repository-catalog-unavailable-error.js";
|
||||
import { REVIEWED_CHAT_BINDING_NOT_READY } from "../reviewed-chat-binding-diagnostics.js";
|
||||
import { RUNNER_REVISION_MISMATCH_CODE, RUNNER_REVISION_MISMATCH_RETRY, RUNNER_REVISION_WAIT_RETRY_REASON } from "../runner-revision-mismatch.js";
|
||||
import { STORAGE_UNAVAILABLE_CODE, STORAGE_UNAVAILABLE_RETRY, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON } from "../../storage/storage-unavailable.js";
|
||||
import { CONTROLLER_LEASE_LOST_FAILURE_CODE } from "../legacy-controller-lease.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import {
|
||||
@@ -148,7 +149,8 @@ export function isTransientWorkspacePreparationCode(code: string | null | undefi
|
||||
|| code === REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE || code === REUSABLE_SANDBOX_RESUME_FAILED_CODE
|
||||
|| code === REMOTE_WORKSPACE_RESTORE_FAILED_CODE || code === PLUGIN_WORKER_UNAVAILABLE_CODE
|
||||
|| code === RUNNER_REVISION_MISMATCH_CODE || code === ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE
|
||||
|| code === REPOSITORY_CATALOG_UNAVAILABLE_CODE || code === REVIEWED_CHAT_BINDING_NOT_READY;
|
||||
|| code === REPOSITORY_CATALOG_UNAVAILABLE_CODE || code === REVIEWED_CHAT_BINDING_NOT_READY
|
||||
|| code === STORAGE_UNAVAILABLE_CODE;
|
||||
}
|
||||
|
||||
export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS =
|
||||
@@ -157,7 +159,9 @@ export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS =
|
||||
export function failedRunRetryBudgetSpent(run: Parameters<typeof executionFailureRetryCount>[0] & { errorCode?: string | null }): boolean {
|
||||
return run.errorCode === RUNNER_REVISION_MISMATCH_CODE
|
||||
? executionRetryAttemptCount(run, RUNNER_REVISION_WAIT_RETRY_REASON) >= RUNNER_REVISION_MISMATCH_RETRY.maxAttempts
|
||||
: executionFailureRetryCount(run) >= BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS;
|
||||
: run.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
? executionRetryAttemptCount(run, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) >= STORAGE_UNAVAILABLE_RETRY.maxAttempts
|
||||
: executionFailureRetryCount(run) >= BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS;
|
||||
}
|
||||
|
||||
const INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS = 2;
|
||||
@@ -365,7 +369,8 @@ function isRetryableInteractionContinuationInfrastructureFailure(
|
||||
run.errorCode === ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE ||
|
||||
run.errorCode === REPOSITORY_CATALOG_UNAVAILABLE_CODE ||
|
||||
run.errorCode === REVIEWED_CHAT_BINDING_NOT_READY ||
|
||||
run.errorCode === RUNNER_REVISION_MISMATCH_CODE
|
||||
run.errorCode === RUNNER_REVISION_MISMATCH_CODE ||
|
||||
run.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
@@ -481,6 +486,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
}
|
||||
}
|
||||
if (!opts?.retryReason && run.errorCode === RUNNER_REVISION_MISMATCH_CODE) opts = { ...RUNNER_REVISION_MISMATCH_RETRY, ...opts };
|
||||
if (!opts?.retryReason && run.errorCode === STORAGE_UNAVAILABLE_CODE) opts = { ...STORAGE_UNAVAILABLE_RETRY, ...opts };
|
||||
if (await hasCaseHandoffForRun(run)) return { outcome: "not_scheduled" as const,
|
||||
reason: "The durable case handoff owns successor work.", issueId: readNonEmptyString(run.contextSnapshot?.issueId) };
|
||||
if (["openai_dot", "muse"].includes(String(parseObject(agent.adapterConfig).provider))
|
||||
@@ -578,7 +584,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
payload: exhaustion,
|
||||
retryExhaustion: exhaustion,
|
||||
});
|
||||
if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) {
|
||||
if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON || (retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON
|
||||
&& isResolvedInteractionContinuationWakeContext(run.contextSnapshot))) {
|
||||
await escalatePlanApprovalResumeFailureNeedsAttention({
|
||||
run,
|
||||
issueId,
|
||||
@@ -689,7 +696,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
}
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null);
|
||||
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
||||
const interactionRunnerWait = retryReason === RUNNER_REVISION_WAIT_RETRY_REASON
|
||||
const interactionRunnerWait = (retryReason === RUNNER_REVISION_WAIT_RETRY_REASON
|
||||
|| retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON)
|
||||
&& isResolvedInteractionContinuationWakeContext(run.contextSnapshot);
|
||||
const interactionContinuationPayload =
|
||||
retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON || interactionRunnerWait
|
||||
@@ -1079,7 +1087,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
|
||||
const scheduledRunId = randomUUID();
|
||||
if (contextSnapshot.explicitUserContinuation) {
|
||||
const continuation = issueId && (retryReason === "transient_failure" || retryReason === RUNNER_REVISION_WAIT_RETRY_REASON) ? await admitExplicitContinuationRetry({
|
||||
const continuation = issueId && (retryReason === "transient_failure" || retryReason === RUNNER_REVISION_WAIT_RETRY_REASON
|
||||
|| retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) ? await admitExplicitContinuationRetry({
|
||||
db: tx as unknown as Db, companyId: run.companyId, issueId, agentId: run.agentId,
|
||||
parentRunId: run.id, successorRunId: scheduledRunId, now,
|
||||
}) : null;
|
||||
|
||||
@@ -33,6 +33,8 @@ import { PluginWorkerUnavailableError } from "../plugin-worker-unavailable-error
|
||||
import { ReviewedChatBindingDeniedError, ReviewedChatBindingNotReadyError } from "../reviewed-chat-binding-diagnostics.js";
|
||||
import { environmentTaskErrorCodeData, JsonRpcCallError, PluginEnvironmentTaskError, PluginWorkspaceOperationError, workspaceOperationErrorData } from "@paperclipai/plugin-sdk";
|
||||
import { RUNNER_REVISION_MISMATCH_RETRY } from "../runner-revision-mismatch.js";
|
||||
import { STORAGE_UNAVAILABLE_CODE, STORAGE_UNAVAILABLE_EXHAUSTED_MESSAGE, STORAGE_UNAVAILABLE_MESSAGE, STORAGE_UNAVAILABLE_RETRY } from "../../storage/storage-unavailable.js";
|
||||
import { withS3FailureDiagnostics } from "../../storage/s3-failure.js";
|
||||
import { workspaceProviderCallFailure } from "../workspace-provider-failure.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { unprocessable } from "../../errors.js";
|
||||
@@ -786,6 +788,50 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
|
||||
resultJson: { providerFailure } });
|
||||
});
|
||||
|
||||
it.each(["run", "setup", "restore"] as const)("waits a minute and suppresses rapid recovery when %s fails because storage is unavailable (storage chaos ST1)", async phase => {
|
||||
const f = await fixture();
|
||||
const outage = await withS3FailureDiagnostics("put_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable",
|
||||
$fault: "server", $metadata: { httpStatusCode: 503, attempts: 3 } });
|
||||
}).catch(error => error as Error);
|
||||
f.deps.scheduleBoundedRetryForRun.mockResolvedValueOnce({ outcome: "scheduled", run: { id: "wait" } } as never);
|
||||
if (phase === "restore") await f.completion.failRunSetup(f.setup(new RemoteWorkspaceRestoreError("snapshot_download", true, outage)));
|
||||
else if (phase === "setup") await f.completion.failRunSetup(f.setup(outage));
|
||||
else await f.completion.failRun(f.failure(outage));
|
||||
const failed = await read(f.run.id);
|
||||
expect(failed).toMatchObject({ status: "failed", errorCode: STORAGE_UNAVAILABLE_CODE, error: STORAGE_UNAVAILABLE_MESSAGE });
|
||||
expect(failed.error).not.toContain("Injected storage outage");
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent, STORAGE_UNAVAILABLE_RETRY);
|
||||
expect(f.deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }),
|
||||
expect.objectContaining({ suppressImmediateRecovery: true }));
|
||||
expect(f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["run", "setup"] as const)("ends the task's automatic recovery once its storage wait is exhausted (%s)", async phase => {
|
||||
const f = await fixture();
|
||||
const outage = await withS3FailureDiagnostics("put_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
|
||||
}).catch(error => error as Error);
|
||||
f.deps.scheduleBoundedRetryForRun.mockResolvedValueOnce({ outcome: "retry_exhausted", attempt: 11, maxAttempts: 10 } as never);
|
||||
if (phase === "setup") await f.completion.failRunSetup(f.setup(outage));
|
||||
else await f.completion.failRun(f.failure(outage));
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent, STORAGE_UNAVAILABLE_RETRY);
|
||||
expect(f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible).not.toHaveBeenCalled();
|
||||
expect(f.deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }),
|
||||
expect.objectContaining({ suppressImmediateRecovery: true }));
|
||||
expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: STORAGE_UNAVAILABLE_CODE,
|
||||
error: STORAGE_UNAVAILABLE_EXHAUSTED_MESSAGE });
|
||||
});
|
||||
|
||||
it("keeps an ordinary adapter failure on the normal recovery path", async () => {
|
||||
const f = await fixture();
|
||||
await f.completion.failRun(f.failure(new Error("adapter crashed")));
|
||||
expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: "adapter_failed" });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
expect(f.deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }),
|
||||
expect.objectContaining({ suppressImmediateRecovery: false }));
|
||||
});
|
||||
|
||||
it("still records failure and releases issue ownership when log finalization and progress flushing fail", async () => {
|
||||
const f = await fixture();
|
||||
f.deps.runLogStore.finalize.mockRejectedValueOnce(new Error("log unavailable"));
|
||||
|
||||
@@ -19,6 +19,7 @@ import { ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE, findEnvironmentProviderUnavailab
|
||||
import { REPOSITORY_CATALOG_UNAVAILABLE_CODE, findRepositoryCatalogUnavailableError } from "../repository-catalog-unavailable-error.js";
|
||||
import { findWorkspaceProviderFailure } from "../workspace-provider-failure.js";
|
||||
import { isRunnerRevisionMismatch, RUNNER_REVISION_MISMATCH_CODE, RUNNER_REVISION_MISMATCH_RETRY } from "../runner-revision-mismatch.js";
|
||||
import { isStorageUnavailableError, STORAGE_UNAVAILABLE_CODE, STORAGE_UNAVAILABLE_EXHAUSTED_MESSAGE, STORAGE_UNAVAILABLE_MESSAGE, STORAGE_UNAVAILABLE_RETRY } from "../../storage/storage-unavailable.js";
|
||||
import { type RunSessionOutcome, resolveNextSessionState, normalizeUsageTotals, resolveCacheAdjustedCostUsd, resolveLedgerBiller, resolveLedgerCostStatus, normalizeLedgerBillingType } from "./run-state.js";
|
||||
import { readGitConnectionFailure } from "../git-connection-failure.js";
|
||||
import { CONFIGURATION_INCOMPLETE_FAILURE_CODE, ConfigurationIncompleteFailure } from "./run-preparation.js";
|
||||
@@ -1145,13 +1146,14 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
: null;
|
||||
const message = interruptedExactResume
|
||||
? describeRefusedExactResume(interruptedExactResume, thrownMessage)
|
||||
: thrownMessage;
|
||||
: isStorageUnavailableError(err) ? STORAGE_UNAVAILABLE_MESSAGE : thrownMessage;
|
||||
const failureErrorCode =
|
||||
workspaceValidationFailure?.code ??
|
||||
configurationIncompleteFailure?.code ??
|
||||
nonRetryablePreflightFailureCode(err) ??
|
||||
recordedResponsibleUserDenialCode ??
|
||||
nativeTerminalFailureCode ??
|
||||
(isStorageUnavailableError(err) ? STORAGE_UNAVAILABLE_CODE : null) ??
|
||||
"adapter_failed";
|
||||
logger.error({ err: identityRedactor.redact({ message: err instanceof Error ? err.message : String(err), stack: err instanceof Error ? err.stack : undefined }), runId }, "heartbeat execution failed");
|
||||
|
||||
@@ -1286,16 +1288,26 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
) {
|
||||
await finalizeIssueCommentPolicy(livenessRun, agent);
|
||||
}
|
||||
await scheduleInteractionContinuationInfrastructureRetryIfEligible(
|
||||
livenessRun,
|
||||
agent,
|
||||
);
|
||||
const storageWait = livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE && nativeTerminalFailureCode === null
|
||||
? await scheduleBoundedRetryForRun(livenessRun, agent, STORAGE_UNAVAILABLE_RETRY).catch((retryError) => {
|
||||
logger.warn({ err: retryError, runId: livenessRun.id }, "failed to schedule a storage wait after a storage outage");
|
||||
return null;
|
||||
})
|
||||
: null;
|
||||
if (storageWait?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun);
|
||||
if (storageWait?.outcome !== "scheduled" && storageWait?.outcome !== "retry_exhausted") {
|
||||
await scheduleInteractionContinuationInfrastructureRetryIfEligible(
|
||||
livenessRun,
|
||||
agent,
|
||||
);
|
||||
}
|
||||
await releaseIssueExecutionAndPromote(livenessRun, {
|
||||
// Native recovery owns the original heartbeat run through
|
||||
// exhaustion. Once its durable coordinator has classified a
|
||||
// terminal failure, generic issue recovery must not create a
|
||||
// replacement retryOfRunId chain for the same provider work.
|
||||
suppressImmediateRecovery: nativeTerminalFailureCode !== null,
|
||||
suppressImmediateRecovery: nativeTerminalFailureCode !== null || storageWait?.outcome === "scheduled"
|
||||
|| storageWait?.outcome === "retry_exhausted",
|
||||
});
|
||||
await handleIssueReviewPathDisposition(livenessRun);
|
||||
|
||||
@@ -1350,6 +1362,12 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
}
|
||||
}
|
||||
|
||||
async function markStorageWaitExhausted(run: typeof heartbeatRuns.$inferSelect) {
|
||||
await db.update(heartbeatRuns).set({ error: STORAGE_UNAVAILABLE_EXHAUSTED_MESSAGE, updatedAt: new Date() })
|
||||
.where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.errorCode, STORAGE_UNAVAILABLE_CODE)))
|
||||
.catch((error: unknown) => logger.warn({ err: error, runId: run.id }, "failed to record an exhausted storage wait"));
|
||||
}
|
||||
|
||||
async function failRunSetup(input: FailHeartbeatRunSetupInput) {
|
||||
const { run, runId, outerErr, signal, providerDispatchStarted, identityRedactor, readFailureReportSecrets } = input;
|
||||
// The lease watchdog stopped preparation because this host no longer owns
|
||||
@@ -1362,7 +1380,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
// The inner catch did not fire, so we must record the failure here.
|
||||
const providerFailure = providerFailureFromError(outerErr);
|
||||
const currentUserRedactionOptions = await getCurrentUserRedactionOptions();
|
||||
const message = redactCurrentUserText(
|
||||
const storageUnavailable = isStorageUnavailableError(outerErr);
|
||||
const message = storageUnavailable ? STORAGE_UNAVAILABLE_MESSAGE : redactCurrentUserText(
|
||||
identityRedactor.redact(outerErr instanceof Error
|
||||
? outerErr.message
|
||||
: "Unknown setup failure"),
|
||||
@@ -1431,6 +1450,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
&& run.nativeSessionId === null && run.runnerInstanceId === null && isRunnerRevisionMismatch(outerErr);
|
||||
const setupFailureErrorCode =
|
||||
(controllerLeaseLost ? CONTROLLER_LEASE_LOST_FAILURE_CODE : null) ??
|
||||
(storageUnavailable ? STORAGE_UNAVAILABLE_CODE : null) ??
|
||||
remoteWorkspacePreparationTimeout?.code ??
|
||||
reusableSandboxResumeFailure?.code ??
|
||||
remoteWorkspaceRestoreFailure?.code ??
|
||||
@@ -1562,6 +1582,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
const livenessRun = await classifyAndPersistRunLiveness(
|
||||
failedRun,
|
||||
).catch(() => failedRun);
|
||||
let storageWaitScheduled = false;
|
||||
const setupFailureIssueId = readNonEmptyString(
|
||||
parseObject(livenessRun.contextSnapshot).issueId,
|
||||
);
|
||||
@@ -1622,7 +1643,9 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
const retry = await (isTransientWorkspacePreparationCode(livenessRun.errorCode) && !interactionOwnsRetry
|
||||
? livenessRun.errorCode === RUNNER_REVISION_MISMATCH_CODE
|
||||
? scheduleBoundedRetryForRun(livenessRun, failedAgent, RUNNER_REVISION_MISMATCH_RETRY)
|
||||
: scheduleBoundedRetryForRun(livenessRun, failedAgent)
|
||||
: livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
? scheduleBoundedRetryForRun(livenessRun, failedAgent, STORAGE_UNAVAILABLE_RETRY)
|
||||
: scheduleBoundedRetryForRun(livenessRun, failedAgent)
|
||||
: scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent)
|
||||
).catch((retryError) => {
|
||||
logger.warn(
|
||||
@@ -1632,10 +1655,13 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
return null;
|
||||
});
|
||||
reviewedChatRetryScheduled = retry?.outcome === "scheduled";
|
||||
if (livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE && retry?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun);
|
||||
storageWaitScheduled = livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
&& (retry?.outcome === "scheduled" || retry?.outcome === "retry_exhausted");
|
||||
}
|
||||
}
|
||||
await releaseIssueExecutionAndPromote(livenessRun, {
|
||||
suppressImmediateRecovery:
|
||||
suppressImmediateRecovery: storageWaitScheduled ||
|
||||
readNonEmptyString(
|
||||
parseObject(livenessRun.contextSnapshot).goalControlRequestId,
|
||||
) !== null ||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { NativeSessionDiagnostics } from "../native-runtime/native-session-diagnostics.js";
|
||||
import { withS3FailureDiagnostics } from "../../storage/s3-failure.js";
|
||||
import { objectLogFixture } from "./__tests__/object-log-fixture.js";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
@@ -71,6 +72,33 @@ const newRunRow = {
|
||||
lastOutputAt: null,
|
||||
} as Partial<Run>;
|
||||
|
||||
describe("claiming the run log during a storage outage (storage chaos ST1)", () => {
|
||||
const outage = () => withS3FailureDiagnostics("put_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
|
||||
});
|
||||
const output = (begin: () => Promise<never | { store: "s3_segments"; logRef: string }>, signal?: AbortSignal) =>
|
||||
createHeartbeatRunOutput(createRunRowDb({ ...newRunRow }), { run: newRunRow as Run, runId: "run-1", issueId: null,
|
||||
runLogStore: { begin, append: async () => 0, enqueue: async () => ({ committed: Promise.resolve(0) }), flush: async () => undefined },
|
||||
identityRedactor: { chunk: (_stream, chunk) => chunk, redact: value => value }, secretKeys: new Set(),
|
||||
appendRunEvent: vi.fn(), getCurrentUserRedactionOptions: async () => ({ enabled: false }), signal });
|
||||
|
||||
it("claims the log once the store is back instead of failing the run", async () => {
|
||||
let calls = 0;
|
||||
const begin = vi.fn(async () => (++calls < 2 ? outage() : { store: "s3_segments" as const, logRef: "company/agent/run" }));
|
||||
await output(begin).begin({ ...newRunRow } as Run);
|
||||
expect(begin).toHaveBeenCalledTimes(2);
|
||||
}, 30_000);
|
||||
|
||||
it("stops waiting as soon as the run is stopped", async () => {
|
||||
const stop = new AbortController();
|
||||
const begin = vi.fn(async () => outage());
|
||||
setTimeout(() => stop.abort(new Error("Stopped by a board operator")), 100);
|
||||
const startedAt = Date.now();
|
||||
await expect(output(begin, stop.signal).begin({ ...newRunRow } as Run)).rejects.toBeTruthy();
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_500);
|
||||
}, 30_000);
|
||||
});
|
||||
|
||||
async function beginAttempt(store: ReturnType<typeof createDurableRunLogStore>, row: Partial<Run>) {
|
||||
const output = createHeartbeatRunOutput(createRunRowDb(row), {
|
||||
run: row as Run,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { notifyActiveRunSilence } from "../live-run-recovery-notifications.js";
|
||||
import { retryThroughStorageOutage } from "../../storage/storage-unavailable.js";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import type { AdapterInvocationMeta, AdapterRuntimeEvent } from "../../adapters/index.js";
|
||||
@@ -34,6 +35,7 @@ export interface HeartbeatRunOutputInput {
|
||||
getCurrentUserRedactionOptions: () => Promise<CurrentUserRedactionOptions>;
|
||||
assertWriterAuthority?: () => Promise<void>;
|
||||
runtimeLogSecretValues?: readonly string[];
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -124,14 +126,16 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
|
||||
const run = currentRun;
|
||||
if (run.logStore != null && !isRunLogStoreType(run.logStore))
|
||||
throw new Error("Run log backend is not supported");
|
||||
handle = await runLogStore.begin({
|
||||
const previousStore = run.logStore ?? undefined;
|
||||
handle = await retryThroughStorageOutage(() => runLogStore.begin({
|
||||
companyId: run.companyId,
|
||||
agentId: run.agentId,
|
||||
runId,
|
||||
newRun: isNewRunLog(run),
|
||||
assertWriterAuthority: input.assertWriterAuthority,
|
||||
previousStore: run.logStore ?? undefined,
|
||||
});
|
||||
previousStore,
|
||||
signal: input.signal,
|
||||
}), input.signal);
|
||||
|
||||
outputSeq = handle.initialSequence ?? outputSeq;
|
||||
persistedLogBytes = handle.initialBytes ?? persistedLogBytes;
|
||||
|
||||
@@ -2,6 +2,8 @@ import { createHash } from "node:crypto";
|
||||
import { Readable } from "node:stream";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { StorageProvider } from "../storage/types.js";
|
||||
import { recordS3BodyFailure, withS3FailureDiagnostics } from "../storage/s3-failure.js";
|
||||
import { endStorageOutageWaitOn, isStorageUnavailableError, sendThroughStorageOutage, waitOutStorageOutages } from "../storage/storage-unavailable.js";
|
||||
import {
|
||||
createObjectByteJournal,
|
||||
ObjectJournalBusyError,
|
||||
@@ -69,6 +71,68 @@ const authority = async () => undefined;
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
describe("object byte journal", () => {
|
||||
it("keeps another run's shared probe going when one run is stopped during a storage outage", async () => {
|
||||
const s = storage();
|
||||
const put = s.provider.putObjectConditional!;
|
||||
const recoversAt = Date.now() + 1_000;
|
||||
s.provider.putObjectConditional = (input) => sendThroughStorageOutage(async () => {
|
||||
if (input.objectKey.includes("conditional-probe-") && Date.now() < recoversAt)
|
||||
return withS3FailureDiagnostics("put_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
|
||||
});
|
||||
return put(input);
|
||||
}, input.signal);
|
||||
const stopA = new AbortController();
|
||||
const claimA = waitOutStorageOutages(async () => {
|
||||
endStorageOutageWaitOn(stopA.signal);
|
||||
return createObjectByteJournal({ provider: s.provider, prefix: "company/a/log" })
|
||||
.claim({ assertAuthority: authority, signal: stopA.signal });
|
||||
});
|
||||
const claimB = waitOutStorageOutages(() => createObjectByteJournal({ provider: s.provider, prefix: "company/b/log" })
|
||||
.claim({ assertAuthority: authority }));
|
||||
void claimA.catch(() => undefined);
|
||||
setTimeout(() => stopA.abort(new Error("Stopped by a board operator")), 200);
|
||||
await expect(claimA).rejects.toBeTruthy();
|
||||
await expect(claimB).resolves.toMatchObject({ head: { generation: 1 } });
|
||||
}, 30_000);
|
||||
|
||||
it("rereads a head whose download was cut off during a storage outage", async () => {
|
||||
const s = storage();
|
||||
const journal = createObjectByteJournal({ provider: s.provider, prefix: "company/reset/log" });
|
||||
const first = await journal.claim({ assertAuthority: authority });
|
||||
const get = s.provider.getObject;
|
||||
let cut = false;
|
||||
s.provider.getObject = async (input) => {
|
||||
const result = await get(input);
|
||||
if (cut || !input.objectKey.endsWith("/head.json")) return result;
|
||||
cut = true;
|
||||
const reset = Object.assign(new Error("aborted"), { code: "ECONNRESET" });
|
||||
recordS3BodyFailure(reset, { $metadata: { httpStatusCode: 200 } });
|
||||
return { ...result, stream: new Readable({ read() { this.destroy(reset); } }) };
|
||||
};
|
||||
const next = await waitOutStorageOutages(() => journal.claim({ assertAuthority: authority }));
|
||||
expect(cut).toBe(true);
|
||||
expect(next.head.generation).toBe(first.head.generation + 1);
|
||||
}, 30_000);
|
||||
|
||||
it("keeps a slow store's head write recognizable as a storage outage when its deadline expires", async () => {
|
||||
const s = storage();
|
||||
const put = s.provider.putObjectConditional!;
|
||||
s.provider.putObjectConditional = async (input) => {
|
||||
if (!input.objectKey.endsWith("/head.json")) return put(input);
|
||||
return withS3FailureDiagnostics("put_object", () => new Promise<never>((_resolve, reject) => {
|
||||
input.signal!.addEventListener("abort", () => {
|
||||
const aborted = new Error("Request aborted");
|
||||
aborted.name = "AbortError";
|
||||
reject(aborted);
|
||||
}, { once: true });
|
||||
}), input.signal);
|
||||
};
|
||||
const journal = createObjectByteJournal({ provider: s.provider, prefix: "company/slow/log" });
|
||||
const failure = await journal.claim({ assertAuthority: authority }).catch((error: unknown) => error);
|
||||
expect(isStorageUnavailableError(failure)).toBe(true);
|
||||
}, 40_000);
|
||||
|
||||
it("keeps a shared capability probe alive for another writer when one claim cancels", async () => {
|
||||
const s = storage();
|
||||
const put = s.provider.putObjectConditional!;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { runSharedStorageWork, sendThroughStorageOutage } from "../storage/storage-unavailable.js";
|
||||
import { z } from "zod";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import type { StorageProvider } from "../storage/types.js";
|
||||
@@ -83,7 +84,7 @@ async function retryObjectOperation<T>(operation: () => Promise<T>, signal?: Abo
|
||||
try {
|
||||
return await operation();
|
||||
} catch (error) {
|
||||
if (attempt >= 2 || !isRetryable(error)) throw error;
|
||||
if (attempt >= 2 || !isRetryable(error) || signal?.aborted) throw error;
|
||||
await delay(100 * 2 ** attempt, undefined, { signal });
|
||||
}
|
||||
}
|
||||
@@ -118,7 +119,7 @@ async function qualifyConditionalWrites(
|
||||
const probeSignal = AbortSignal.any([controller.signal, AbortSignal.timeout(15_000)]);
|
||||
const current: ConditionalQualification = { controller, waiters: 0, work: Promise.resolve() };
|
||||
qualification = current;
|
||||
current.work = (async () => {
|
||||
current.work = runSharedStorageWork(async () => {
|
||||
const objectKey = `${prefix}/conditional-probe-${randomUUID()}`;
|
||||
const body = Buffer.from("paperclip-conditional-probe-v1");
|
||||
const put = provider.putObjectConditional!.bind(provider);
|
||||
@@ -164,7 +165,7 @@ async function qualifyConditionalWrites(
|
||||
}
|
||||
probeSignal.throwIfAborted();
|
||||
qualifiedProviders.add(provider);
|
||||
})().finally(() => {
|
||||
}).finally(() => {
|
||||
if (qualifications.get(provider) === current) qualifications.delete(provider);
|
||||
});
|
||||
qualifications.set(provider, current);
|
||||
@@ -394,24 +395,27 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
|
||||
signal = signal
|
||||
? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
|
||||
: AbortSignal.timeout(15_000);
|
||||
const result = await retryObjectOperation(
|
||||
() => provider.getObject({ objectKey: key, signal }),
|
||||
signal,
|
||||
);
|
||||
const chunks: Buffer[] = [];
|
||||
let size = 0;
|
||||
try {
|
||||
for await (const chunk of result.stream) {
|
||||
const bytes = Buffer.from(chunk);
|
||||
size += bytes.length;
|
||||
if (size > limit) throw new Error("object_journal_object_too_large");
|
||||
chunks.push(bytes);
|
||||
const read = signal;
|
||||
return sendThroughStorageOutage(async () => {
|
||||
const result = await retryObjectOperation(
|
||||
() => provider.getObject({ objectKey: key, signal: read }),
|
||||
read,
|
||||
);
|
||||
const chunks: Buffer[] = [];
|
||||
let size = 0;
|
||||
try {
|
||||
for await (const chunk of result.stream) {
|
||||
const bytes = Buffer.from(chunk);
|
||||
size += bytes.length;
|
||||
if (size > limit) throw new Error("object_journal_object_too_large");
|
||||
chunks.push(bytes);
|
||||
}
|
||||
} finally {
|
||||
result.stream.destroy();
|
||||
}
|
||||
} finally {
|
||||
result.stream.destroy();
|
||||
}
|
||||
signal?.throwIfAborted();
|
||||
return { bytes: Buffer.concat(chunks), etag: result.etag };
|
||||
read?.throwIfAborted();
|
||||
return { bytes: Buffer.concat(chunks), etag: result.etag };
|
||||
}, read);
|
||||
}, readPriority, signal);
|
||||
}
|
||||
async function head(signal?: AbortSignal): Promise<ObjectJournalWriter | null> {
|
||||
@@ -522,6 +526,7 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
|
||||
});
|
||||
return { head: value, etag: result.etag };
|
||||
} catch (error) {
|
||||
if (signal.aborted) throw error;
|
||||
const observed = await head(signal);
|
||||
if (observed && JSON.stringify(observed.head) === JSON.stringify(value)) return observed;
|
||||
if (isObjectPreconditionFailure(error) || (observed?.etag ?? null) !== etag)
|
||||
@@ -571,14 +576,14 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
|
||||
let shared = readsInFlight.get(cacheKey);
|
||||
if (!shared) {
|
||||
const controller = new AbortController();
|
||||
const work = (async () => {
|
||||
const work = runSharedStorageWork(async () => {
|
||||
const stored = await getBytes(ref.key, MAX_OBJECT_BYTES * 2, controller.signal);
|
||||
const bytes = options.decode ? await options.decode(stored.bytes) : stored.bytes;
|
||||
if (bytes.length > MAX_OBJECT_BYTES || sha256(bytes) !== ref.sha256)
|
||||
throw new Error("object_journal_integrity_error");
|
||||
cacheObject(cacheKey, bytes);
|
||||
return bytes;
|
||||
})();
|
||||
});
|
||||
const entry: SharedObjectRead = { work, controller, waiters: 0 };
|
||||
const release = () => {
|
||||
if (readsInFlight.get(cacheKey) === entry) readsInFlight.delete(cacheKey);
|
||||
|
||||
@@ -16,6 +16,8 @@ import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.j
|
||||
import { createS3StorageProvider } from "../storage/s3-provider.js";
|
||||
import { createStorageService } from "../storage/service.js";
|
||||
import type { StorageService } from "../storage/types.js";
|
||||
import { withS3FailureDiagnostics } from "../storage/s3-failure.js";
|
||||
import { isStorageUnavailableError } from "../storage/storage-unavailable.js";
|
||||
import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
|
||||
import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
|
||||
import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, newWorkspaceSnapshotStorageTiming, recoverWorkspaceSnapshotArchive, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js";
|
||||
@@ -453,6 +455,28 @@ suite("remote workspace snapshot storage", () => {
|
||||
await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally { temp.mockRestore(); }
|
||||
});
|
||||
it("keeps a storage outage as the cause when the download deadline ends its wait", async () => {
|
||||
const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload);
|
||||
let requested!: () => void;
|
||||
const firstRequest = new Promise<void>(resolve => { requested = resolve; });
|
||||
const unavailable = { ...storage, getObject: vi.fn(async () => {
|
||||
requested();
|
||||
return withS3FailureDiagnostics("get_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
|
||||
});
|
||||
}) } as StorageService;
|
||||
const scratch = path.join(root, "outage-scratch"); await fs.mkdir(scratch);
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
const result = loadWorkspaceSnapshotArchive({ storage: unavailable, reference: ref, targetPath: path.join(root, "outage.pcws"),
|
||||
temporaryParent: scratch }).then(() => ({ succeeded: true }), (error: unknown) => error);
|
||||
try {
|
||||
await firstRequest;
|
||||
await vi.advanceTimersByTimeAsync(300_001);
|
||||
const outcome = await Promise.race([result, wait(2_000).then(() => ({ pending: true }))]);
|
||||
expect(isStorageUnavailableError(outcome)).toBe(true);
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it.each(["headers", "body"])("closes a stalled S3 %s request before returning a retryable download failure", async phase => {
|
||||
const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload);
|
||||
const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey));
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { createCipheriv, createDecipheriv, createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { isStorageUnavailableError, sendThroughStorageOutage, waitOutStorageOutages } from "../storage/storage-unavailable.js";
|
||||
import { measureRunPhase, type ExecutionPhase } from "./execution-preparation-timing.js";
|
||||
import { collectStorageRequestStats, newStorageRequestStats, type StorageRequestStats } from "../storage/request-stats.js";
|
||||
import { constants, createReadStream, createWriteStream } from "node:fs";
|
||||
@@ -174,7 +175,7 @@ type WorkspaceSnapshotUpload = {
|
||||
|
||||
/** Upload attempts are journaled before remote I/O. The caller owns publication. */
|
||||
export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): Promise<WorkspaceSnapshotRef> {
|
||||
return withRequestStats(input.timing, () => uploadWorkspaceSnapshotTimed(input));
|
||||
return withRequestStats(input.timing, () => waitOutStorageOutages(() => uploadWorkspaceSnapshotTimed(input)));
|
||||
}
|
||||
|
||||
async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise<WorkspaceSnapshotRef> {
|
||||
@@ -370,7 +371,7 @@ export async function loadWorkspaceSnapshotArchive(input: {
|
||||
/** Receives per-step durations and storage request counts. */
|
||||
timing?: WorkspaceSnapshotStorageTiming;
|
||||
}): Promise<void> {
|
||||
return withRequestStats(input.timing, () => loadWorkspaceSnapshotArchiveTimed(input));
|
||||
return withRequestStats(input.timing, () => waitOutStorageOutages(() => loadWorkspaceSnapshotArchiveTimed(input)));
|
||||
}
|
||||
|
||||
async function loadWorkspaceSnapshotArchiveTimed(input: Parameters<typeof loadWorkspaceSnapshotArchive>[0]): Promise<void> {
|
||||
@@ -380,14 +381,20 @@ async function loadWorkspaceSnapshotArchiveTimed(input: Parameters<typeof loadWo
|
||||
try {
|
||||
input.signal?.throwIfAborted();
|
||||
const ciphertext = path.join(scratch, "ciphertext"), plaintext = path.join(scratch, "plaintext");
|
||||
let received = 0; const digest = createHash("sha256");
|
||||
let received = 0, digest = createHash("sha256");
|
||||
const stalled = new DOMException("remote_workspace_snapshot_download_stalled", "TimeoutError");
|
||||
const idle = new AbortController();
|
||||
const signal = input.signal ? AbortSignal.any([input.signal, idle.signal]) : idle.signal;
|
||||
const timer = setTimeout(() => idle.abort(stalled), DOWNLOAD_IDLE_TIMEOUT_MS);
|
||||
timer.unref();
|
||||
try {
|
||||
await timedStep(input.timing, "downloadMs", "storage.snapshot.download", async () => {
|
||||
let attempts = 0;
|
||||
await timedStep(input.timing, "downloadMs", "storage.snapshot.download", () => sendThroughStorageOutage(async () => {
|
||||
if (attempts++ > 0) {
|
||||
timer.refresh();
|
||||
received = 0; digest = createHash("sha256");
|
||||
await rm(ciphertext, { force: true });
|
||||
}
|
||||
const object = await input.storage.getObject(ref.companyId, ref.objectKey, { signal });
|
||||
const bounded = new Transform({ transform(bytes, _encoding, callback) {
|
||||
if (bytes.length > 0) timer.refresh();
|
||||
@@ -396,10 +403,11 @@ async function loadWorkspaceSnapshotArchiveTimed(input: Parameters<typeof loadWo
|
||||
} });
|
||||
try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal }); }
|
||||
finally { object.stream.destroy(); }
|
||||
});
|
||||
}, signal));
|
||||
if (input.timing) input.timing.bytes = received;
|
||||
} catch (error) {
|
||||
if (signal.aborted && signal.reason === stalled) throw stalled;
|
||||
if (signal.aborted && signal.reason === stalled) throw isStorageUnavailableError(error)
|
||||
? Object.assign(new DOMException(stalled.message, stalled.name), { cause: error }) : stalled;
|
||||
throw error;
|
||||
} finally { clearTimeout(timer); }
|
||||
if (received !== ref.byteSize || digest.digest("hex") !== ref.sha256) failure("object_hash_mismatch");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { STORAGE_UNAVAILABLE_WAIT_RETRY_REASON } from "../storage/storage-unavailable.js";
|
||||
import { and, desc, eq, inArray, isNull, sql, type SQL } from "drizzle-orm";
|
||||
import { unidentifiedLocalLaunchRefusals } from "./native-local-process-stop.js";
|
||||
import { agents, agentWakeupRequests, approvals, chatActions, companies, heartbeatRuns, issueApprovals, issueRecoveryActions,
|
||||
@@ -25,7 +26,7 @@ import { AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON,
|
||||
// attempt. Their dispatcher re-checks the resource, so the run can stay
|
||||
// scheduled past its retry time. Retry cannot free the resource either.
|
||||
const resourceWaitRetryReasons = new Set<string>([WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON,
|
||||
AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON]);
|
||||
AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON]);
|
||||
|
||||
// A due same-run resume dispatches within seconds; past this, offer Retry.
|
||||
const NATIVE_RESUME_DISPATCH_GRACE_MS = 2 * 60_000;
|
||||
|
||||
Reference in new issue
Block a user