diff --git a/server/src/__tests__/heartbeat-run-output.test.ts b/server/src/__tests__/heartbeat-run-output.test.ts
index b9a68d2bbe4..c52bb3df314 100644
--- a/server/src/__tests__/heartbeat-run-output.test.ts
+++ b/server/src/__tests__/heartbeat-run-output.test.ts
@@ -5,6 +5,8 @@ import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRu
import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionContext } from "../adapters/index.js";
import { heartbeatService } from "../services/heartbeat.js";
import { getRunLogStore } from "../services/run-log-store.js";
+import { withS3FailureDiagnostics } from "../storage/s3-failure.js";
+import { sendThroughStorageOutage } from "../storage/storage-unavailable.js";
import { subscribeCompanyLiveEvents } from "../services/live-events.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
@@ -101,6 +103,37 @@ describePostgres("heartbeat output through adapter dispatch and finalization", (
} finally { unsubscribe(); }
});
+ const storeOutage = () => withS3FailureDiagnostics("put_object", async () => {
+ throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
+ });
+
+ it("claims the run log once the store is back and finishes the run (storage chaos ST1)", async () => {
+ const f = await seed(async input => { await input.onLog("stdout", "after the outage\n"); });
+ const store = getRunLogStore();
+ const begin = store.begin.bind(store);
+ let outages = 0;
+ vi.spyOn(store, "begin").mockImplementation(async input => (++outages <= 1 ? storeOutage() : begin(input)));
+ const { finished } = await execute(f);
+ expect(outages).toBeGreaterThanOrEqual(2);
+ expect(finished.status, finished.error ?? undefined).toBe("succeeded");
+ expect(finished.stdoutExcerpt).toContain("after the outage");
+ }, 30_000);
+
+ it("ends a run stopped during a storage outage as cancelled, even in a storage call made without the run's signal", async () => {
+ const f = await seed(async () => undefined);
+ const store = getRunLogStore();
+ const begin = vi.spyOn(store, "begin").mockImplementation(() => sendThroughStorageOutage(storeOutage));
+ const run = await heartbeat.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "manual",
+ payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId } });
+ await vi.waitFor(() => expect(begin).toHaveBeenCalled(), { timeout: 10_000 });
+ const stoppedAt = Date.now();
+ await heartbeat.cancelRun(run!.id, "Cancelled by a board operator");
+ await heartbeat.drainActiveRunExecutions();
+ expect(Date.now() - stoppedAt).toBeLessThan(10_000);
+ const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
+ expect(stopped).toMatchObject({ status: "cancelled" });
+ }, 60_000);
+
it.each([false, true])("retains output arriving during finalization (adapter throws: %s)", async fail => {
let onLog: AdapterExecutionContext["onLog"];
const f = await seed(async input => {
diff --git a/server/src/__tests__/storage-outage-wait.test.ts b/server/src/__tests__/storage-outage-wait.test.ts
new file mode 100644
index 00000000000..3599bea5630
--- /dev/null
+++ b/server/src/__tests__/storage-outage-wait.test.ts
@@ -0,0 +1,219 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { createS3StorageProvider } from "../storage/s3-provider.js";
+import { withS3FailureDiagnostics } from "../storage/s3-failure.js";
+import { endStorageOutageWaitOn, isStorageUnavailableError, retryThroughStorageOutage, runInOwnStorageScope, runSharedStorageWork, waitOutStorageOutages } from "../storage/storage-unavailable.js";
+import { RemoteWorkspaceRestoreError } from "../services/remote-workspace-transport.js";
+
+const OUTAGE = `
+ServiceUnavailableInjected storage outagereq-503`;
+
+let outageUntil = 0;
+let requests = 0;
+const server = createServer((req, res) => {
+ req.resume();
+ req.on("end", () => {
+ requests++;
+ if (Date.now() < outageUntil) {
+ res.writeHead(503, { "content-type": "application/xml" });
+ res.end(OUTAGE);
+ return;
+ }
+ if (req.method === "GET") {
+ res.writeHead(200, { "content-length": "5", etag: '"v1"' });
+ res.end("saved");
+ return;
+ }
+ res.writeHead(200, { etag: '"v2"' });
+ res.end();
+ });
+});
+const savedEnv = { ...process.env };
+let endpoint = "";
+
+beforeAll(async () => {
+ await new Promise(resolve => server.listen(0, "127.0.0.1", resolve));
+ endpoint = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+ process.env.AWS_ACCESS_KEY_ID = "fixture-access-key";
+ process.env.AWS_SECRET_ACCESS_KEY = "fixture-secret-key";
+ delete process.env.AWS_SESSION_TOKEN;
+ delete process.env.AWS_PROFILE;
+});
+afterAll(async () => {
+ process.env = savedEnv;
+ await new Promise(resolve => server.close(() => resolve()));
+});
+beforeEach(() => { requests = 0; outageUntil = 0; });
+
+const provider = () => createS3StorageProvider({ bucket: "bucket", region: "us-east-1", endpoint, forcePathStyle: true });
+
+async function read(stream: NodeJS.ReadableStream) {
+ const chunks: Buffer[] = [];
+ for await (const chunk of stream) chunks.push(Buffer.from(chunk as Buffer));
+ return Buffer.concat(chunks).toString("utf8");
+}
+
+describe("waiting out a storage outage on a run's critical path (Daytona storage chaos ST1)", () => {
+ it("restores a saved object once the store is back instead of failing the run", async () => {
+ outageUntil = Date.now() + 3_000;
+ const object = await waitOutStorageOutages(() => provider().getObject({ objectKey: "company/workspace.pcws" }));
+ expect(await read(object.stream)).toBe("saved");
+ expect(requests).toBeGreaterThan(1);
+ }, 30_000);
+
+ it("claims the run log once the store is back", async () => {
+ outageUntil = Date.now() + 3_000;
+ await expect(waitOutStorageOutages(() => provider().putObjectConditional!({ objectKey: "company/runs/head.json",
+ body: Buffer.from("{}"), contentType: "application/json", contentLength: 2, ifNoneMatch: "*" })))
+ .resolves.toEqual({ etag: '"v2"' });
+ }, 30_000);
+
+ it("keeps failing fast outside a run's critical path", async () => {
+ outageUntil = Date.now() + 60_000;
+ const startedAt = Date.now();
+ await expect(provider().headObject({ objectKey: "company/avatar.png" })).rejects.toMatchObject({ $metadata: { httpStatusCode: 503 } });
+ expect(Date.now() - startedAt).toBeLessThan(10_000);
+ }, 30_000);
+
+ it("stops waiting when the run is stopped", async () => {
+ outageUntil = Date.now() + 60_000;
+ const stop = new AbortController();
+ setTimeout(() => stop.abort(new Error("Stopped by a board operator")), 3_000);
+ const startedAt = Date.now();
+ await expect(waitOutStorageOutages(() => provider().getObject({ objectKey: "company/workspace.pcws", signal: stop.signal })))
+ .rejects.toBeTruthy();
+ expect(Date.now() - startedAt).toBeLessThan(15_000);
+ }, 30_000);
+});
+
+describe("the shared storage outage wait", () => {
+ const outage = () => withS3FailureDiagnostics("get_object", async () => {
+ throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
+ });
+ const policy = { budgetMs: 200, firstDelayMs: 10, maxDelayMs: 40 };
+
+ it("retries until the store answers, within its budget", async () => {
+ let calls = 0;
+ await expect(retryThroughStorageOutage(async () => (++calls < 4 ? outage() : "ok"), undefined, policy)).resolves.toBe("ok");
+ expect(calls).toBe(4);
+ });
+
+ it("gives the outage back to the run's storage wait once the budget is spent", async () => {
+ let calls = 0;
+ await expect(retryThroughStorageOutage(async () => { calls++; return outage(); }, undefined, policy))
+ .rejects.toMatchObject({ name: "ServiceUnavailable" });
+ expect(calls).toBeGreaterThan(1);
+ });
+
+ it("keeps the storage failure when a deadline ends the wait, but not when a Stop does", async () => {
+ const slow = { budgetMs: 60_000, firstDelayMs: 1_000, maxDelayMs: 1_000 };
+ await expect(retryThroughStorageOutage(outage, AbortSignal.timeout(50), slow))
+ .rejects.toSatisfy(isStorageUnavailableError);
+ const stop = new AbortController();
+ setTimeout(() => stop.abort(new Error("Stopped by a board operator")), 50);
+ const stopped = await retryThroughStorageOutage(outage, stop.signal, slow).catch((error: unknown) => error);
+ expect(stopped).toMatchObject({ name: "AbortError" });
+ expect(isStorageUnavailableError(stopped)).toBe(false);
+ });
+
+ it("lets nested and parallel waits ride out one outage shorter than the budget", async () => {
+ const wide = { budgetMs: 600, firstDelayMs: 20, maxDelayMs: 40 };
+ const endsAt = Date.now() + 300;
+ const send = async () => (Date.now() < endsAt ? outage() : "ok");
+ const nested = () => retryThroughStorageOutage(() => retryThroughStorageOutage(send, undefined, wide), undefined, wide);
+ await expect(waitOutStorageOutages(() => Promise.all(Array.from({ length: 8 }, nested))))
+ .resolves.toEqual(Array(8).fill("ok"));
+ });
+
+ it("gives up once one outage has lasted longer than the budget, however the waits overlap", async () => {
+ const startedAt = Date.now();
+ const failing = () => retryThroughStorageOutage(outage, undefined, policy);
+ const results = await waitOutStorageOutages(() => Promise.allSettled(Array.from({ length: 4 }, failing)));
+ expect(results.every(result => result.status === "rejected")).toBe(true);
+ expect(Date.now() - startedAt).toBeGreaterThanOrEqual(policy.budgetMs - policy.maxDelayMs);
+ expect(Date.now() - startedAt).toBeLessThan(policy.budgetMs * 3);
+ });
+
+ it("keeps the limit when only some keys fail and other storage calls succeed", async () => {
+ const startedAt = Date.now();
+ const result = await waitOutStorageOutages(async () => {
+ const reads = setInterval(() => void retryThroughStorageOutage(async () => "read", undefined, policy), 5);
+ try {
+ return await retryThroughStorageOutage(async () => {
+ await retryThroughStorageOutage(async () => "head read", undefined, policy);
+ return outage();
+ }, undefined, policy).catch((error: unknown) => error);
+ } finally { clearInterval(reads); }
+ });
+ expect(result).toMatchObject({ name: "ServiceUnavailable" });
+ expect(Date.now() - startedAt).toBeLessThan(policy.budgetMs * 3);
+ });
+
+ it("keeps one operation's limit when another operation in the same run recovers from its own outage", async () => {
+ const startedAt = Date.now();
+ const result = await waitOutStorageOutages(async () => {
+ let recovering = true;
+ void (async () => {
+ while (recovering) {
+ let failed = false;
+ await retryThroughStorageOutage(async () => (failed ? "recovered" : (failed = true, outage())), undefined, policy).catch(() => undefined);
+ }
+ })();
+ try {
+ return await retryThroughStorageOutage(outage, undefined, policy).catch((error: unknown) => error);
+ } finally { recovering = false; }
+ });
+ expect(result).toMatchObject({ name: "ServiceUnavailable" });
+ expect(Date.now() - startedAt).toBeLessThan(policy.budgetMs * 3);
+ });
+
+ it("gives a later outage its own limit once the waiting work recovered", async () => {
+ await waitOutStorageOutages(async () => {
+ const recoversAt = Date.now() + policy.budgetMs - 60;
+ await expect(retryThroughStorageOutage(async () => (Date.now() < recoversAt ? outage() : "ok"), undefined, policy)).resolves.toBe("ok");
+ const secondRecoversAt = Date.now() + policy.budgetMs - 60;
+ await expect(retryThroughStorageOutage(async () => (Date.now() < secondRecoversAt ? outage() : "again"), undefined, policy)).resolves.toBe("again");
+ });
+ });
+
+ it("gives a run started from a stopped run's cleanup its own scope", async () => {
+ const stopped = new AbortController();
+ stopped.abort(new Error("Stopped by a board operator"));
+ const wide = { budgetMs: 2_000, firstDelayMs: 20, maxDelayMs: 40 };
+ let calls = 0;
+ const next = await waitOutStorageOutages(async () => {
+ endStorageOutageWaitOn(stopped.signal);
+ return runInOwnStorageScope(() => retryThroughStorageOutage(async () => (++calls < 3 ? outage() : "next run"), undefined, wide));
+ });
+ expect(next).toBe("next run");
+ });
+
+ it("lets one run's Stop end only its own wait, not shared work another run waits on", async () => {
+ const endsAt = Date.now() + 150;
+ const wide = { budgetMs: 2_000, firstDelayMs: 20, maxDelayMs: 40 };
+ const stopA = new AbortController();
+ let shared!: Promise;
+ const runA = waitOutStorageOutages(async () => {
+ endStorageOutageWaitOn(stopA.signal);
+ shared = runSharedStorageWork(() => retryThroughStorageOutage(async () => (Date.now() < endsAt ? outage() : "shared"), undefined, wide));
+ return retryThroughStorageOutage(outage, undefined, wide);
+ });
+ setTimeout(() => stopA.abort(new Error("Stopped by a board operator")), 30);
+ await expect(runA).rejects.toMatchObject({ name: "AbortError" });
+ await expect(shared).resolves.toBe("shared");
+ });
+
+ it("never retries a failure that is not an outage", async () => {
+ let calls = 0;
+ await expect(retryThroughStorageOutage(async () => { calls++; throw new Error("object_journal_conflict"); }, undefined, policy))
+ .rejects.toThrow("object_journal_conflict");
+ expect(calls).toBe(1);
+ });
+
+ it("recognizes an outage a workspace restore wrapped (snapshot_download)", async () => {
+ const cause = await outage().catch(error => error);
+ expect(isStorageUnavailableError(new RemoteWorkspaceRestoreError("snapshot_download", true, cause))).toBe(true);
+ expect(isStorageUnavailableError(new RemoteWorkspaceRestoreError("sandbox_restore", true, new Error("reason=occupied")))).toBe(false);
+ });
+});
diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts
index badec8c72b5..f4fbc476821 100644
--- a/server/src/__tests__/task-interruption-query.test.ts
+++ b/server/src/__tests__/task-interruption-query.test.ts
@@ -607,7 +607,7 @@ describe.skipIf(!support.supported)("task interruption latest run query", () =>
});
// N-03: ONB-1 on staging waited 16 minutes on an overdue workspace_busy
// retry. Nothing failed, yet the banner offered a Retry that cannot help.
- it.each(["workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", "runner_revision_wait"])(
+ it.each(["workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", "runner_revision_wait", "storage_unavailable_wait"])(
"does not present an overdue %s wait as a failed attempt", async reason => {
const companyId = randomUUID(), agentId = randomUUID(), parentRunId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Waiting task", issuePrefix: `W${companyId.slice(0, 7)}` });
diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts
index 40bf97bd0d5..754fd0f1712 100644
--- a/server/src/modules/run-dispatch/domain/wake-context.ts
+++ b/server/src/modules/run-dispatch/domain/wake-context.ts
@@ -20,6 +20,7 @@ export const WORKSPACE_BUSY_RETRY_REASON = "workspace_busy";
export const AI_CONNECTION_POOL_WAIT_RETRY_REASON = "ai_connection_pool_wait";
export const AI_CONNECTION_BUSY_RETRY_REASON = "ai_connection_busy";
export const RUNNER_REVISION_WAIT_RETRY_REASON = "runner_revision_wait";
+export const STORAGE_UNAVAILABLE_WAIT_RETRY_REASON = "storage_unavailable_wait";
export const INTERACTION_CONTINUATION_INFRA_RETRY_REASON = "interaction_continuation_infra_retry";
export const INTERACTION_CONTINUATION_INFRA_WAKE_REASON = "interaction_continuation_infra_retry";
export const WAKE_COMMENT_IDS_KEY = "wakeCommentIds";
@@ -204,7 +205,8 @@ export function isResolvedInteractionContinuationWakeContext(contextSnapshot: un
const retryReason = readNonEmptyString(context.retryReason);
return (
(mutation === "interaction" && wakeReason === "issue_commented") ||
- (mutation === "interaction" && retryReason === RUNNER_REVISION_WAIT_RETRY_REASON) ||
+ (mutation === "interaction" && (retryReason === RUNNER_REVISION_WAIT_RETRY_REASON
+ || retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON)) ||
(mutation === "interaction" && context.dependencyBlockedInteraction === true && retryReason !== null
&& RESOLVED_INTERACTION_RESOURCE_WAIT_REASONS.has(retryReason)) ||
wakeReason === INTERACTION_CONTINUATION_INFRA_WAKE_REASON ||
diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts
index 5f391cca374..96523837323 100644
--- a/server/src/modules/run-dispatch/index.ts
+++ b/server/src/modules/run-dispatch/index.ts
@@ -15,6 +15,7 @@ export {
AI_CONNECTION_BUSY_RETRY_REASON,
AI_CONNECTION_POOL_WAIT_RETRY_REASON,
RUNNER_REVISION_WAIT_RETRY_REASON,
+ STORAGE_UNAVAILABLE_WAIT_RETRY_REASON,
INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
WAKE_COMMENT_IDS_KEY,
diff --git a/server/src/services/execution-recovery-attempt.test.ts b/server/src/services/execution-recovery-attempt.test.ts
index a715ffc04fb..1940a71358e 100644
--- a/server/src/services/execution-recovery-attempt.test.ts
+++ b/server/src/services/execution-recovery-attempt.test.ts
@@ -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);
+ });
+});
diff --git a/server/src/services/execution-recovery-attempt.ts b/server/src/services/execution-recovery-attempt.ts
index 977151e5e7b..bff18d9aa6c 100644
--- a/server/src/services/execution-recovery-attempt.ts
+++ b/server/src/services/execution-recovery-attempt.ts
@@ -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 = 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;
}
diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts
index 5e61f756765..a75a065ea27 100644
--- a/server/src/services/explicit-native-continuation.test.ts
+++ b/server/src/services/explicit-native-continuation.test.ts
@@ -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 });
diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts
index bb08881e332..16be14da953 100644
--- a/server/src/services/heartbeat.ts
+++ b/server/src/services/heartbeat.ts
@@ -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;
} = {},
) {
- 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: {
diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts
index dbd8186f3c9..cd94dbfe9d8 100644
--- a/server/src/services/heartbeat/retries.test.ts
+++ b/server/src/services/heartbeat/retries.test.ts
@@ -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"}]};
diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts
index f8738e9d16c..6bc3ad55266 100644
--- a/server/src/services/heartbeat/retries.ts
+++ b/server/src/services/heartbeat/retries.ts
@@ -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[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;
diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts
index 10f480c9760..cd9df2ae118 100644
--- a/server/src/services/heartbeat/run-completion.test.ts
+++ b/server/src/services/heartbeat/run-completion.test.ts
@@ -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"));
diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts
index ebad344fe03..6ab4f3630f8 100644
--- a/server/src/services/heartbeat/run-completion.ts
+++ b/server/src/services/heartbeat/run-completion.ts
@@ -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 ||
diff --git a/server/src/services/heartbeat/run-output.test.ts b/server/src/services/heartbeat/run-output.test.ts
index 6cf9c915416..6bb537af2af 100644
--- a/server/src/services/heartbeat/run-output.test.ts
+++ b/server/src/services/heartbeat/run-output.test.ts
@@ -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;
+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, 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, row: Partial) {
const output = createHeartbeatRunOutput(createRunRowDb(row), {
run: row as Run,
diff --git a/server/src/services/heartbeat/run-output.ts b/server/src/services/heartbeat/run-output.ts
index 33e91bc078a..90a916a6743 100644
--- a/server/src/services/heartbeat/run-output.ts
+++ b/server/src/services/heartbeat/run-output.ts
@@ -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;
assertWriterAuthority?: () => Promise;
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;
diff --git a/server/src/services/object-byte-journal.test.ts b/server/src/services/object-byte-journal.test.ts
index d0eae2640d7..01706bbfa4e 100644
--- a/server/src/services/object-byte-journal.test.ts
+++ b/server/src/services/object-byte-journal.test.ts
@@ -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((_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!;
diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts
index dfbaf23273f..c4c6e160859 100644
--- a/server/src/services/object-byte-journal.ts
+++ b/server/src/services/object-byte-journal.ts
@@ -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(operation: () => Promise, 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 {
@@ -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);
diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts
index 804db520139..59308c169b6 100644
--- a/server/src/services/remote-workspace-snapshot-storage.test.ts
+++ b/server/src/services/remote-workspace-snapshot-storage.test.ts
@@ -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(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));
diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts
index 8044c6aa914..8ec543041c1 100644
--- a/server/src/services/remote-workspace-snapshot-storage.ts
+++ b/server/src/services/remote-workspace-snapshot-storage.ts
@@ -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 {
- return withRequestStats(input.timing, () => uploadWorkspaceSnapshotTimed(input));
+ return withRequestStats(input.timing, () => waitOutStorageOutages(() => uploadWorkspaceSnapshotTimed(input)));
}
async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise {
@@ -370,7 +371,7 @@ export async function loadWorkspaceSnapshotArchive(input: {
/** Receives per-step durations and storage request counts. */
timing?: WorkspaceSnapshotStorageTiming;
}): Promise {
- return withRequestStats(input.timing, () => loadWorkspaceSnapshotArchiveTimed(input));
+ return withRequestStats(input.timing, () => waitOutStorageOutages(() => loadWorkspaceSnapshotArchiveTimed(input)));
}
async function loadWorkspaceSnapshotArchiveTimed(input: Parameters[0]): Promise {
@@ -380,14 +381,20 @@ async function loadWorkspaceSnapshotArchiveTimed(input: Parameters 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([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;
diff --git a/server/src/storage/s3-failure.ts b/server/src/storage/s3-failure.ts
index 3d70a72b1dc..dac19e56a51 100644
--- a/server/src/storage/s3-failure.ts
+++ b/server/src/storage/s3-failure.ts
@@ -25,12 +25,18 @@ function integerIn(value: unknown, min: number, max: number): number | null {
}
/** Record which S3 call failed and the provider's own correlation fields. */
-export function recordS3Failure(error: unknown, operation: StorageFailureOperation): void {
+function deadlineExpired(error: unknown, signal: AbortSignal | undefined) {
+ return ownValue(error, "name") === "AbortError" && signal?.aborted === true
+ && signal.reason instanceof DOMException && signal.reason.name === "TimeoutError";
+}
+
+export function recordS3Failure(error: unknown, operation: StorageFailureOperation, signal?: AbortSignal): void {
const metadata = ownValue(error, "$metadata");
recordStorageFailure(error, {
provider: "s3",
operation,
- errorCode: matching(ownValue(error, "name"), ERROR_CODE) ?? matching(ownValue(error, "code"), ERROR_CODE) ?? "unknown",
+ errorCode: deadlineExpired(error, signal) ? "TimeoutError"
+ : matching(ownValue(error, "name"), ERROR_CODE) ?? matching(ownValue(error, "code"), ERROR_CODE) ?? "unknown",
httpStatus: integerIn(ownValue(metadata, "httpStatusCode"), 100, 599),
requestId: matching(ownValue(metadata, "requestId"), REQUEST_ID),
attempts: integerIn(ownValue(metadata, "attempts"), 1, 100),
@@ -54,14 +60,15 @@ export function recordS3BodyFailure(error: unknown, response: unknown): void {
});
}
-export async function withS3FailureDiagnostics(operation: StorageFailureOperation, send: () => Promise): Promise {
+export async function withS3FailureDiagnostics(operation: StorageFailureOperation, send: () => Promise,
+ signal?: AbortSignal): Promise {
try {
const result = await send();
recordStorageRequest(result, false);
return result;
} catch (error) {
recordStorageRequest(error, true);
- recordS3Failure(error, operation);
+ recordS3Failure(error, operation, signal);
throw error;
}
}
diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts
index 74992e5f1a8..2f64817bc09 100644
--- a/server/src/storage/s3-multipart.ts
+++ b/server/src/storage/s3-multipart.ts
@@ -1,4 +1,5 @@
import { AbortMultipartUploadCommand, CompleteMultipartUploadCommand, CreateMultipartUploadCommand, UploadPartCommand, type S3Client } from "@aws-sdk/client-s3";
+import { sendThroughStorageOutage } from "./storage-unavailable.js";
import { recordStorageBackoff, recordStoragePart, recordStorageThrottle, withStorageRequestStats } from "./request-stats.js";
import { createHash } from "node:crypto";
import { once } from "node:events";
@@ -67,7 +68,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri
const partSize = Math.max(bufferedPartSize, Math.ceil(input.contentLength / 10_000));
try {
check();
- uploadId = (await withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }))).UploadId;
+ uploadId = (await withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }), controller.signal)).UploadId;
if (!uploadId) throw new Error("S3 did not return a multipart upload id");
const parts: { ETag: string; PartNumber: number }[] = [];
let partNumber = 0;
@@ -94,7 +95,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri
if (!body.write(chunk)) await once(body, "drain", { signal: controller.signal });
};
const sending = (async () => {
- const result = await withS3FailureDiagnostics("upload_part", () => client.send(new UploadPartCommand({ Bucket: bucket, Key: key, UploadId: uploadId, PartNumber: number, Body: body, ContentLength: length }), { abortSignal: controller.signal }));
+ const result = await withS3FailureDiagnostics("upload_part", () => client.send(new UploadPartCommand({ Bucket: bucket, Key: key, UploadId: uploadId, PartNumber: number, Body: body, ContentLength: length }), { abortSignal: controller.signal }), controller.signal);
if (remaining || !body.readableEnded || !result.ETag) throw new Error("S3 upload part is incomplete");
parts[number - 1] = { ETag: result.ETag, PartNumber: number };
})().catch(stop);
@@ -115,7 +116,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri
if (pending.length || !(await readNext()).done) throw new Error("S3 upload source exceeds its declared size");
await Promise.all(active);
check();
- await withS3FailureDiagnostics("complete_multipart_upload", () => client.send(new CompleteMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: uploadId, MultipartUpload: { Parts: parts } }), { abortSignal: controller.signal }));
+ await withS3FailureDiagnostics("complete_multipart_upload", () => client.send(new CompleteMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: uploadId, MultipartUpload: { Parts: parts } }), { abortSignal: controller.signal }), controller.signal);
} catch (error) {
stop(error);
await Promise.all(active);
@@ -218,7 +219,7 @@ export async function putS3MultipartFile(
controller.signal.throwIfAborted();
const partSize = Math.max(bufferedPartSize, Math.ceil(input.contentLength / 10_000));
const count = Math.ceil(input.contentLength / partSize);
- uploadId = (await withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }))).UploadId;
+ uploadId = (await sendThroughStorageOutage(() => withS3FailureDiagnostics("create_multipart_upload", () => client.send(new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: input.contentType }), { abortSignal: controller.signal }), controller.signal), controller.signal)).UploadId;
if (!uploadId) throw new Error("S3 did not return a multipart upload id");
const id = uploadId;
const parts: { ETag: string; PartNumber: number }[] = new Array(count);
@@ -236,7 +237,7 @@ export async function putS3MultipartFile(
try {
const result = await withS3FailureDiagnostics("upload_part", () => client.send(new UploadPartCommand({
Bucket: bucket, Key: key, UploadId: id, PartNumber: index + 1, Body: body, ContentLength: length, ContentMD5: contentMd5,
- }), { abortSignal: controller.signal }));
+ }), { abortSignal: controller.signal }), controller.signal);
if (!result.ETag) throw new Error("S3 upload part is incomplete");
parts[index] = { ETag: result.ETag, PartNumber: index + 1 };
recordStoragePart();
@@ -264,7 +265,7 @@ export async function putS3MultipartFile(
const worker = async (slot: number) => {
while (next < count && slot < concurrency) {
const index = next++;
- try { await sendPart(index); }
+ try { await sendThroughStorageOutage(() => sendPart(index), controller.signal); }
catch (error) { if (!controller.signal.aborted) controller.abort(error); throw error; }
}
};
@@ -273,7 +274,7 @@ export async function putS3MultipartFile(
const rejected = settled.find((entry): entry is PromiseRejectedResult => entry.status === "rejected");
if (rejected) throw controller.signal.aborted && controller.signal.reason !== undefined ? controller.signal.reason : rejected.reason;
controller.signal.throwIfAborted();
- await withS3FailureDiagnostics("complete_multipart_upload", () => client.send(new CompleteMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: id, MultipartUpload: { Parts: parts } }), { abortSignal: controller.signal }));
+ await sendThroughStorageOutage(() => withS3FailureDiagnostics("complete_multipart_upload", () => client.send(new CompleteMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: id, MultipartUpload: { Parts: parts } }), { abortSignal: controller.signal }), controller.signal), controller.signal);
uploadId = undefined;
return { integrity: "server_verified_md5" };
} catch (error) {
diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts
index 294afb83a6e..395c4b5e2dc 100644
--- a/server/src/storage/s3-provider.ts
+++ b/server/src/storage/s3-provider.ts
@@ -11,6 +11,7 @@ import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile,
import { createReadStream } from "node:fs";
import { createHash } from "node:crypto";
import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js";
+import { sendThroughStorageOutage } from "./storage-unavailable.js";
import { addAbortSignal, Readable } from "node:stream";
import { retryS3RequestTimeouts } from "./s3-request-timeout-retry.js";
import type { StorageProvider, GetObjectResult, HeadObjectResult } from "./types.js";
@@ -180,7 +181,7 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
])).digest("hex") : undefined,
async putObjectConditional(input) {
- const output = await client.send(new PutObjectCommand({
+ const send = () => withS3FailureDiagnostics("put_object", () => client.send(new PutObjectCommand({
Bucket: bucket,
Key: buildKey(prefix, input.objectKey),
Body: input.body,
@@ -188,7 +189,8 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
ContentLength: input.contentLength,
IfMatch: input.ifMatch,
IfNoneMatch: input.ifNoneMatch,
- }), { abortSignal: input.signal });
+ }), { abortSignal: input.signal }), input.signal);
+ const output = Buffer.isBuffer(input.body) ? await sendThroughStorageOutage(send, input.signal) : await send();
if (!output.ETag) throw new Error("Object storage did not return a publication ETag");
return { etag: output.ETag };
},
@@ -217,22 +219,22 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
if (input.body instanceof Readable) input.body.destroy();
await assertSourceFileSize(sourceFile, input.contentLength);
const contentMd5 = await fileRangeMd5(sourceFile, 0, input.contentLength - 1, input.signal);
- await sendFileRequestWithRetry(input.signal, async () => {
+ await sendThroughStorageOutage(() => sendFileRequestWithRetry(input.signal, async () => {
const body = createReadStream(sourceFile);
try {
await withS3FailureDiagnostics("put_object", () => fileClient.send(new PutObjectCommand({
Bucket: bucket, Key: key, Body: body, ContentType: input.contentType,
ContentLength: input.contentLength, ContentMD5: contentMd5,
- }), { abortSignal: input.signal }));
+ }), { abortSignal: input.signal }), input.signal);
} finally { body.destroy(); }
- });
+ }), input.signal);
return { integrity: "server_verified_md5" } as const;
}
if (input.contentLength > 16 * 1024 * 1024) {
await putS3Multipart(client, bucket, key, input);
return;
}
- await withS3FailureDiagnostics("put_object", () => client.send(
+ const send = () => withS3FailureDiagnostics("put_object", () => client.send(
new PutObjectCommand({
Bucket: bucket,
Key: key,
@@ -241,20 +243,21 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
ContentLength: input.contentLength,
}),
{ abortSignal: input.signal },
- ));
+ ), input.signal);
+ await (Buffer.isBuffer(input.body) ? sendThroughStorageOutage(send, input.signal) : send());
},
async getObject(input): Promise {
const key = buildKey(prefix, input.objectKey);
try {
- const output = await withS3FailureDiagnostics("get_object", () => client.send(
+ const output = await sendThroughStorageOutage(() => withS3FailureDiagnostics("get_object", () => client.send(
new GetObjectCommand({
Bucket: bucket,
Key: key,
Range: input.range ? `bytes=${input.range.start}-${input.range.end}` : undefined,
}),
{ abortSignal: input.signal },
- ));
+ ), input.signal), input.signal);
const first = observeBodyFailures(await toReadableStream(output.Body), output);
const etag = output.ETag;
@@ -264,7 +267,7 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
const resumed = await withS3FailureDiagnostics("get_object", () => client.send(new GetObjectCommand({
Bucket: bucket, Key: key, IfMatch: etag,
Range: `bytes=${start + offset}-${end ?? ""}`,
- }), { abortSignal: resumeSignal }));
+ }), { abortSignal: resumeSignal }), resumeSignal);
return observeBodyFailures(await toReadableStream(resumed.Body), resumed);
};
const stream = etag && end !== undefined
@@ -288,13 +291,13 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
async headObject(input): Promise {
const key = buildKey(prefix, input.objectKey);
try {
- const output = await withS3FailureDiagnostics("head_object", () => client.send(
+ const output = await sendThroughStorageOutage(() => withS3FailureDiagnostics("head_object", () => client.send(
new HeadObjectCommand({
Bucket: bucket,
Key: key,
}),
{ abortSignal: input.signal },
- ));
+ ), input.signal), input.signal);
return {
exists: true,
@@ -313,7 +316,7 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
async deleteObject(input): Promise {
const key = buildKey(prefix, input.objectKey);
if (input.allVersions) {
- await withS3FailureDiagnostics("delete_object", async () => {
+ await sendThroughStorageOutage(() => withS3FailureDiagnostics("delete_object", async () => {
// Finish listing before deleting: removing a cursor version must not skip later pages.
const versions = await listExactVersions(key, input.signal);
for (const versionId of versions) {
@@ -322,16 +325,16 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
{ abortSignal: input.signal });
}
if ((await listExactVersions(key, input.signal)).size) throw new Error("s3_object_versions_still_present");
- });
+ }, input.signal), input.signal);
return;
}
- await withS3FailureDiagnostics("delete_object", () => client.send(
+ await sendThroughStorageOutage(() => withS3FailureDiagnostics("delete_object", () => client.send(
new DeleteObjectCommand({
Bucket: bucket,
Key: key,
}),
{ abortSignal: input.signal },
- ));
+ ), input.signal), input.signal);
},
};
}
diff --git a/server/src/storage/storage-unavailable.test.ts b/server/src/storage/storage-unavailable.test.ts
new file mode 100644
index 00000000000..8fa8deb5a42
--- /dev/null
+++ b/server/src/storage/storage-unavailable.test.ts
@@ -0,0 +1,65 @@
+import { describe, expect, it } from "vitest";
+import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js";
+import { isStorageUnavailableError } from "./storage-unavailable.js";
+
+function s3Error(name: string, status?: number, code?: string) {
+ return Object.assign(new Error(name), { name, ...(code ? { code } : {}),
+ $metadata: status === undefined ? {} : { httpStatusCode: status, attempts: 3 } });
+}
+
+async function recorded(error: unknown, signal?: AbortSignal) {
+ return withS3FailureDiagnostics("put_object", async () => { throw error; }, signal).catch(thrown => thrown);
+}
+
+function sdkAbort() {
+ const aborted = new Error("Request aborted");
+ aborted.name = "AbortError";
+ return aborted;
+}
+
+describe("storage unavailability", () => {
+ it.each([
+ ["503 ServiceUnavailable", s3Error("ServiceUnavailable", 503)],
+ ["503 SlowDown", s3Error("SlowDown", 503)],
+ ["429 throttling", s3Error("TooManyRequests", 429)],
+ ["500 InternalError", s3Error("InternalError", 500)],
+ ["408 RequestTimeout", s3Error("RequestTimeout", 408)],
+ ["a reset recorded under its error name", Object.assign(new Error("socket hang up"), { code: "ECONNRESET" })],
+ ["a refused connection without a status", s3Error("Error", undefined, "ECONNREFUSED")],
+ ])("treats %s from the store as unavailable", async (_label, error) => {
+ expect(isStorageUnavailableError(await recorded(error))).toBe(true);
+ });
+
+ it.each([
+ ["404 NoSuchKey", s3Error("NoSuchKey", 404)],
+ ["403 AccessDenied", s3Error("AccessDenied", 403)],
+ ["412 PreconditionFailed", s3Error("PreconditionFailed", 412)],
+ ["an ordinary error", new Error("adapter crashed")],
+ ])("does not treat %s as an outage", async (_label, error) => {
+ expect(isStorageUnavailableError(await recorded(error))).toBe(false);
+ });
+
+ it("treats a download body reset after a 200 response as unavailable", () => {
+ const reset = Object.assign(new Error("aborted"), { code: "ECONNRESET" });
+ recordS3BodyFailure(reset, { $metadata: { httpStatusCode: 200, requestId: "req-1" } });
+ expect(isStorageUnavailableError(reset)).toBe(true);
+ });
+
+ it("treats a storage call its deadline aborted as unavailable (storage chaos ST2)", async () => {
+ const deadline = AbortSignal.timeout(1);
+ await new Promise(resolve => setTimeout(resolve, 20));
+ expect(isStorageUnavailableError(await recorded(sdkAbort(), deadline))).toBe(true);
+ });
+
+ it("does not treat a run's own Stop abort as a storage outage", async () => {
+ const stop = new AbortController();
+ stop.abort(new Error("Cancelled by a board operator"));
+ expect(isStorageUnavailableError(await recorded(sdkAbort(), stop.signal))).toBe(false);
+ expect(isStorageUnavailableError(await recorded(sdkAbort()))).toBe(false);
+ });
+
+ it("requires the failure to come from a storage call", () => {
+ expect(isStorageUnavailableError(s3Error("ServiceUnavailable", 503))).toBe(false);
+ expect(isStorageUnavailableError(Object.assign(new Error("reset"), { code: "ECONNRESET" }))).toBe(false);
+ });
+});
diff --git a/server/src/storage/storage-unavailable.ts b/server/src/storage/storage-unavailable.ts
new file mode 100644
index 00000000000..395357b9575
--- /dev/null
+++ b/server/src/storage/storage-unavailable.ts
@@ -0,0 +1,111 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { setTimeout as delay } from "node:timers/promises";
+import { storageFailureDiagnosticsFor } from "./failure-diagnostics.js";
+import { recordStorageBackoff } from "./request-stats.js";
+
+export const STORAGE_UNAVAILABLE_CODE = "storage_unavailable";
+
+export const STORAGE_UNAVAILABLE_WAIT_RETRY_REASON = "storage_unavailable_wait";
+
+export const STORAGE_UNAVAILABLE_MESSAGE = "File storage was unavailable, so this attempt stopped before it could save its work. Paperclip will retry automatically when storage is back.";
+
+export const STORAGE_UNAVAILABLE_EXHAUSTED_MESSAGE = "File storage stayed unavailable through every automatic retry, so Paperclip stopped retrying this task. Retry once storage is back.";
+
+export const STORAGE_UNAVAILABLE_RETRY = {
+ retryReason: STORAGE_UNAVAILABLE_WAIT_RETRY_REASON, wakeReason: "storage_unavailable_retry", maxAttempts: 10, delayMs: 60_000,
+} as const;
+
+const UNAVAILABLE_STATUS = new Set([408, 429, 500, 502, 503, 504]);
+const UNAVAILABLE_NAMES = new Set(["ServiceUnavailable", "SlowDown", "InternalError", "RequestTimeout", "TimeoutError", "NetworkingError"]);
+const UNAVAILABLE_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "ETIMEDOUT", "EPIPE", "EAI_AGAIN", "ENOTFOUND",
+ "ERR_STREAM_PREMATURE_CLOSE", "UND_ERR_SOCKET"]);
+
+function ownValue(value: unknown, key: string): unknown {
+ if (!value || typeof value !== "object") return undefined;
+ try {
+ const descriptor = Object.getOwnPropertyDescriptor(value, key);
+ return descriptor && "value" in descriptor ? descriptor.value : undefined;
+ } catch {
+ return undefined;
+ }
+}
+
+function networkFailure(error: unknown): boolean {
+ for (const candidate of [error, ownValue(error, "cause")]) {
+ const name = ownValue(candidate, "name"), code = ownValue(candidate, "code");
+ if ((typeof name === "string" && UNAVAILABLE_NAMES.has(name)) || (typeof code === "string" && UNAVAILABLE_CODES.has(code))) return true;
+ }
+ return false;
+}
+
+function recordedStorageFailure(error: unknown) {
+ for (let candidate = error, depth = 0; candidate && depth < 6; candidate = ownValue(candidate, "cause"), depth++) {
+ const diagnostics = storageFailureDiagnosticsFor(candidate);
+ if (diagnostics) return { error: candidate, diagnostics };
+ }
+ return null;
+}
+
+export function isStorageUnavailableError(error: unknown): boolean {
+ const recorded = recordedStorageFailure(error);
+ if (!recorded) return false;
+ const { diagnostics } = recorded;
+ const status = diagnostics.httpStatus;
+ if (status !== null && (status < 200 || status >= 300)) return UNAVAILABLE_STATUS.has(status);
+ return UNAVAILABLE_NAMES.has(diagnostics.errorCode) || UNAVAILABLE_CODES.has(diagnostics.errorCode) || networkFailure(recorded.error);
+}
+
+export const STORAGE_OUTAGE_WAIT = { budgetMs: 5 * 60_000, firstDelayMs: 2_000, maxDelayMs: 30_000 } as const;
+
+type OutageWaitScope = { stop?: AbortSignal };
+const outageWaitScope = new AsyncLocalStorage();
+type OutageWaitOperation = { firstFailureAt: number | null };
+const outageWaitOperation = new AsyncLocalStorage();
+
+export function waitOutStorageOutages(work: () => Promise): Promise {
+ return outageWaitScope.getStore() ? work() : outageWaitScope.run({}, work);
+}
+
+export function runInOwnStorageScope(work: () => Promise): Promise {
+ return outageWaitScope.run({}, () => outageWaitOperation.run(null, work));
+}
+
+export function runSharedStorageWork(work: () => Promise): Promise {
+ return outageWaitScope.getStore() ? outageWaitScope.run({}, () => outageWaitOperation.run(null, work)) : outageWaitOperation.run(null, work);
+}
+
+export function endStorageOutageWaitOn(stop: AbortSignal): void {
+ const current = outageWaitScope.getStore();
+ if (current) outageWaitScope.enterWith({ stop: current.stop ? AbortSignal.any([current.stop, stop]) : stop });
+}
+
+export function retryThroughStorageOutage(send: () => Promise, signal?: AbortSignal,
+ policy: { budgetMs: number; firstDelayMs: number; maxDelayMs: number } = STORAGE_OUTAGE_WAIT): Promise {
+ const operation = outageWaitOperation.getStore();
+ return operation ? retryWithin(operation, send, signal, policy)
+ : outageWaitOperation.run({ firstFailureAt: null }, () => retryWithin(outageWaitOperation.getStore()!, send, signal, policy));
+}
+
+async function retryWithin(operation: OutageWaitOperation, send: () => Promise, signal: AbortSignal | undefined,
+ policy: { budgetMs: number; firstDelayMs: number; maxDelayMs: number }): Promise {
+ const scopeStop = outageWaitScope.getStore()?.stop;
+ const stop = signal && scopeStop ? AbortSignal.any([signal, scopeStop]) : signal ?? scopeStop;
+ for (let wait = policy.firstDelayMs; ; wait = Math.min(wait * 2, policy.maxDelayMs)) {
+ try {
+ return await send();
+ } catch (error) {
+ if (!isStorageUnavailableError(error)) throw error;
+ const now = Date.now();
+ operation.firstFailureAt ??= now;
+ if (stop?.aborted || now - operation.firstFailureAt + wait > policy.budgetMs) throw error;
+ recordStorageBackoff(wait);
+ await delay(wait, undefined, { signal: stop }).catch((interrupted: unknown) => {
+ throw signal?.aborted && signal.reason instanceof DOMException && signal.reason.name === "TimeoutError" ? error : interrupted;
+ });
+ }
+ }
+}
+
+export function sendThroughStorageOutage(send: () => Promise, signal?: AbortSignal): Promise {
+ return outageWaitScope.getStore() ? retryThroughStorageOutage(send, signal) : send();
+}