mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(heartbeat): show a storage outage wait as live status, then record it once (#20460)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - In cloud mode a run reads and writes file storage on its critical path: it starts its run log (`s3_segments`), restores its saved workspace, and saves it again > - Since #20299 a run waits out a storage outage there for up to 5 minutes instead of failing > - That wait was silent. The board showed "running" with no sign of life, so a storage wait looked like a dead run (restart-chaos R5 `silent_dead_run`) > - The first version of this PR wrote a run event on every wait. That writes the run row (`next_event_seq`, `updated_at`) while the run-log claim is still in progress, and it covered only the run-log begin > - This PR shows the wait as live run status instead, writes nothing during the wait, and records one lifecycle event when the wait ends. It covers every critical-path wait in the run's storage scope > - The benefit is that a storage outage shows on the board as waiting, the run's history keeps one honest line about it, and the claim path is left alone ## Linked Issues or Issue Description Refs #20299 (wait out a store outage on a run's critical path). **What happened?** - `w-20261110T043533-s11936` (Daytona R1, ST1 194 s, K2): run `d49f58d8` waited 186 s in `heartbeat.prepare.run_output` with no event. The grader failed it as `R5:silent_dead_run`. - `w-20261110T105903-s11936`: the restore download (`storage.snapshot.download`) waited 182 s with no sign of life. **Expected:** the board shows the run waiting for file storage, and nothing about the notice can disturb run-log ownership. **On the failed `run_output` seen on builds of the first version** (`w-20261110T101517-s11936` t01 run `9d39550a`, `w-20261110T081647-s11936` t02 run `9da10424`): the evidence points to an expired restart lease, not to the notice's row writes. Neither the writer-authority check (`run-log-authority.ts`) nor the journal claim compares `next_event_seq` or `updated_at`. A remote `bootstrap_incomplete` restart claim gets a 60 s lease (`REMOTE_CONTROLLER_LEASE_MS`), and preparation renews it only for `reattach_remote_runner` claims. In t01 the claim was at 10:19:55, the outage ended at 10:22:18.15, and `run_output` failed at 10:22:18.17, on the first authority check after the store came back. That fix ships separately as its own PR. ## What Changed - `storage/storage-unavailable.ts`: - The run's storage scope can carry a wait observer (`reportStorageOutageWaitsTo`). - `retryThroughStorageOutage` tells it about each backoff (`waiting`). The operation that owns the wait reports one `ended` (`recovered` true or false), so nested waits do not double-report. - `storageOutageWaitFor(purpose, work)` names the step. - An observer that throws never changes the wait. `endStorageOutageWaitOn` keeps the observer. - `heartbeat/storage-wait-notice.ts` (new): - Each wait sets live runtime status (phase `waiting_for_storage`, "Waiting for file storage before starting the run log (45s so far)") and publishes `heartbeat.run.progress`. Nothing goes to the database. - When the wait ends, it writes one `lifecycle` warn event: "Waited Ns for file storage before …", with payload `reason: "storage_outage_wait"`, `purpose`, `waitedMs` and `recovered`. - `heartbeat.ts`: installs the notice next to `endStorageOutageWaitOn` (one line). - Purposes: - run-log begin: "starting the run log"; - `loadWorkspaceSnapshotArchive`: "restoring the workspace"; - `uploadWorkspaceSnapshot`: "saving the workspace". - `packages/shared`: adds `waiting_for_storage` to `HeartbeatRunStatusPhase`. - `tests/restart-chaos/live-status-evidence.ts` (new): R5 also counts the board's live run status (`/api/issues/:id/live-runs` `currentStatusUpdatedAt` / `lastEventAt`) as evidence that a running run is alive. Wiring it into `observer.ts` takes one import and one changed line. - `doc/run-log-events.md`: documents the live status and the one event. - Removed: the first version's per-wait run event. ## Verification - `cd server && npx vitest run src/services/heartbeat/run-output.test.ts src/__tests__/storage-outage-wait.test.ts src/services/remote-workspace-snapshot-storage.test.ts src/services/object-byte-journal.test.ts src/__tests__/storage-s3-credential-rotation.test.ts`: 103 pass. - Red/green: with the observer disabled (master behaviour), these new tests fail. With this change they pass: - run-output "shows the wait as live status and writes nothing to the run while its log claim waits": no run-row write and no event during the claim's waits; the live status is `waiting_for_storage`; the claim's writer-authority check still passes; one lifecycle event afterwards; - the helper's observer tests (one end per wait; gave-up reporting; a throwing observer; the Stop signal keeps the observer); - snapshot storage "tells the run it is waiting for storage while a save or a restore waits out an outage". - `PAPERCLIP_IN_WORKTREE=false npx vitest run --config tests/restart-chaos/vitest.config.ts tests/restart-chaos/live-status-evidence.test.ts tests/restart-chaos/observer.test.ts tests/restart-chaos/triggers.test.ts`: 65 pass. `node scripts/typecheck-test-projects.mjs tests/restart-chaos`: pass. `cd server && npx tsc --noEmit`: pass. - Live restart-chaos waves: results in a comment below. ## Risks - Low. - The retry policy, its budget and Stop handling are unchanged. - During a wait, the only new effect is an in-memory status and a live event. - Afterwards, one event per wait, written without blocking the run. - Runtime status is per server process. A board on another instance sees the waiting line only after the closing event. - Shared journal work (`runSharedStorageWork`) keeps its own scope without the observer. Its waits surface through the enclosing run-scoped wait, as before. - Previous-release state: this persists no new run or session state. A run saved by the previous release resumes unchanged. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
bf514759f9
commit
15a734d068
13 files changed
+315
-15
No files matched your search
@@ -627,6 +627,22 @@ settlement record. The record does not imply exactly-once emission across
|
||||
controller recovery, and failure to write it is logged without changing the
|
||||
already-settled outcome.
|
||||
|
||||
## Storage outage waits
|
||||
|
||||
A run waits out a file-storage outage on its critical path: when it starts its
|
||||
run log, restores its saved workspace, or saves it. While it waits, the run's live
|
||||
status says so, with phase `waiting_for_storage`, for example "Waiting for file
|
||||
storage before starting the run log (45s so far)". That status lives in server
|
||||
memory and on the live-event stream only. It writes nothing to the run row or
|
||||
the run log during the wait.
|
||||
|
||||
When the wait ends, the controller writes one record with `eventType:
|
||||
"lifecycle"`, `stream: "system"`, and `level: "warn"`. Its message is "Waited Ns
|
||||
for file storage before <step>." or, when storage stayed unavailable, "Waited Ns
|
||||
for file storage before <step>; it was still unavailable." Its payload holds
|
||||
`reason: "storage_outage_wait"`, `purpose` (the step), `waitedMs`, and
|
||||
`recovered`. The record stays in the instance run log.
|
||||
|
||||
## Durable storage timing
|
||||
|
||||
A saved remote workspace is one encrypted archive in durable object storage.
|
||||
|
||||
@@ -256,6 +256,7 @@ export type HeartbeatRunStatusPhase =
|
||||
| "restore"
|
||||
| "export"
|
||||
| "finalize"
|
||||
| "waiting_for_storage"
|
||||
| "run_activity";
|
||||
|
||||
export type HeartbeatRunOutputSilenceLevel =
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import { createServer } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } 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 { endStorageOutageWaitOn, isStorageUnavailableError, reportStorageOutageWaitsTo, retryThroughStorageOutage, runInOwnStorageScope, runSharedStorageWork, storageOutageWaitFor, waitOutStorageOutages } from "../storage/storage-unavailable.js";
|
||||
import { RemoteWorkspaceRestoreError } from "../services/remote-workspace-transport.js";
|
||||
|
||||
const OUTAGE = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
@@ -204,6 +204,43 @@ describe("the shared storage outage wait", () => {
|
||||
await expect(shared).resolves.toBe("shared");
|
||||
});
|
||||
|
||||
it("tells the run's wait observer about each wait and reports one end per wait", async () => {
|
||||
const wide = { budgetMs: 2_000, firstDelayMs: 20, maxDelayMs: 40 };
|
||||
const waiting = vi.fn(), ended = vi.fn();
|
||||
const endsAt = Date.now() + 150;
|
||||
const send = async () => (Date.now() < endsAt ? outage() : "ok");
|
||||
await runInOwnStorageScope(async () => {
|
||||
reportStorageOutageWaitsTo({ waiting, ended });
|
||||
await expect(storageOutageWaitFor("restoring the workspace",
|
||||
() => retryThroughStorageOutage(() => retryThroughStorageOutage(send, undefined, wide), undefined, wide))).resolves.toBe("ok");
|
||||
await expect(retryThroughStorageOutage(async () => "no wait", undefined, wide)).resolves.toBe("no wait");
|
||||
});
|
||||
expect(waiting.mock.calls.length).toBeGreaterThan(1);
|
||||
expect(waiting).toHaveBeenCalledWith({ purpose: "restoring the workspace", waitedMs: expect.any(Number) });
|
||||
expect(ended).toHaveBeenCalledTimes(1);
|
||||
expect(ended).toHaveBeenCalledWith({ purpose: "restoring the workspace", waitedMs: expect.any(Number), recovered: true });
|
||||
});
|
||||
|
||||
it("reports a wait that gave up, and keeps waiting when the observer throws", async () => {
|
||||
const ended = vi.fn();
|
||||
await runInOwnStorageScope(async () => {
|
||||
reportStorageOutageWaitsTo({ waiting: () => { throw new Error("status unavailable"); }, ended });
|
||||
await expect(retryThroughStorageOutage(outage, undefined, policy)).rejects.toMatchObject({ name: "ServiceUnavailable" });
|
||||
});
|
||||
expect(ended).toHaveBeenCalledWith({ purpose: "continuing", waitedMs: expect.any(Number), recovered: false });
|
||||
});
|
||||
|
||||
it("keeps a run's wait observer when its Stop signal is attached", async () => {
|
||||
const waiting = vi.fn();
|
||||
let calls = 0;
|
||||
await runInOwnStorageScope(async () => {
|
||||
reportStorageOutageWaitsTo({ waiting, ended: vi.fn() });
|
||||
endStorageOutageWaitOn(new AbortController().signal);
|
||||
await retryThroughStorageOutage(async () => (++calls < 2 ? outage() : "ok"), undefined, policy);
|
||||
});
|
||||
expect(waiting).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
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))
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js";
|
||||
import { endStorageOutageWaitOn, runInOwnStorageScope } from "../storage/storage-unavailable.js";
|
||||
import { endStorageOutageWaitOn, reportStorageOutageWaitsTo, runInOwnStorageScope } from "../storage/storage-unavailable.js";
|
||||
import { createStorageWaitNotice } from "./heartbeat/storage-wait-notice.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";
|
||||
@@ -2014,6 +2015,7 @@ export function heartbeatService(
|
||||
const timePreparation = createHeartbeatPreparationTiming(run.id);
|
||||
const executionControl = createAdapterExecutionControl();
|
||||
endStorageOutageWaitOn(executionControl.controller.signal);
|
||||
reportStorageOutageWaitsTo(createStorageWaitNotice({ run, appendRunEvent }));
|
||||
// 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");
|
||||
|
||||
@@ -13,6 +13,9 @@ import { createHeartbeatRunOutput, isNewRunLog } from "./run-output.js";
|
||||
import { createAgentIdentityRedactor } from "../agent-identity-redaction.js";
|
||||
import { formatNativeRunLogEvent } from "../native-runtime/native-run-log-export.js";
|
||||
import { generateKeyPairSync } from "node:crypto";
|
||||
import { reportStorageOutageWaitsTo, runInOwnStorageScope } from "../../storage/storage-unavailable.js";
|
||||
import { clearAllHeartbeatRunRuntimeStatuses, getHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js";
|
||||
import { createStorageWaitNotice, STORAGE_WAIT_PHASE } from "./storage-wait-notice.js";
|
||||
|
||||
vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
|
||||
|
||||
@@ -89,6 +92,51 @@ describe("claiming the run log during a storage outage (storage chaos ST1)", ()
|
||||
expect(begin).toHaveBeenCalledTimes(2);
|
||||
}, 30_000);
|
||||
|
||||
it("shows the wait as live status and writes nothing to the run while its log claim waits", async () => {
|
||||
clearAllHeartbeatRunRuntimeStatuses();
|
||||
const row = { ...newRunRow, contextSnapshot: { issueId: "issue-1" } } as Partial<Run>;
|
||||
const writes: Partial<Run>[] = [];
|
||||
const db = createRunRowDb(row);
|
||||
const update = db.update.bind(db);
|
||||
(db as { update: unknown }).update = (table: unknown) => {
|
||||
const pending = (update as (t: unknown) => { set: (v: Partial<Run>) => unknown })(table);
|
||||
return { set: (values: Partial<Run>) => { writes.push(values); return pending.set(values); } };
|
||||
};
|
||||
const appendRunEvent = vi.fn(async () => undefined);
|
||||
const authority = vi.fn(async () => undefined);
|
||||
const seen: Array<{ writes: number; events: number; phase: string | undefined; message: string | undefined }> = [];
|
||||
let calls = 0;
|
||||
const begin = vi.fn(async (input: { assertWriterAuthority?: () => Promise<void> }) => {
|
||||
const status = getHeartbeatRunRuntimeStatus("run-1", { companyId: "company-1", agentId: "agent-1" });
|
||||
seen.push({ writes: writes.length, events: appendRunEvent.mock.calls.length, phase: status?.phase, message: status?.message });
|
||||
if (++calls < 3) return outage();
|
||||
await input.assertWriterAuthority?.();
|
||||
return { store: "s3_segments" as const, logRef: "company/agent/run" };
|
||||
});
|
||||
const recorder = createHeartbeatRunOutput(db, { run: row as Run, runId: "run-1", issueId: "issue-1",
|
||||
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: appendRunEvent as never, getCurrentUserRedactionOptions: async () => ({ enabled: false }),
|
||||
assertWriterAuthority: authority });
|
||||
await runInOwnStorageScope(async () => {
|
||||
reportStorageOutageWaitsTo(createStorageWaitNotice({ run: row as Run, appendRunEvent: appendRunEvent as never }));
|
||||
await recorder.begin(row as Run);
|
||||
});
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
expect(begin).toHaveBeenCalledTimes(3);
|
||||
expect(authority).toHaveBeenCalledTimes(1);
|
||||
expect(seen.slice(1)).toEqual([
|
||||
expect.objectContaining({ writes: 0, events: 0, phase: STORAGE_WAIT_PHASE, message: expect.stringContaining("before starting the run log") }),
|
||||
expect.objectContaining({ writes: 0, events: 0, phase: STORAGE_WAIT_PHASE }),
|
||||
]);
|
||||
expect(writes).toEqual([expect.objectContaining({ logStore: "s3_segments", logRef: "company/agent/run" })]);
|
||||
expect(appendRunEvent).toHaveBeenCalledTimes(1);
|
||||
expect(appendRunEvent).toHaveBeenCalledWith(row, expect.objectContaining({ eventType: "lifecycle",
|
||||
message: expect.stringMatching(/^Waited \d+s for file storage before starting the run log\.$/),
|
||||
payload: expect.objectContaining({ reason: "storage_outage_wait", recovered: true }) }));
|
||||
expect(getHeartbeatRunRuntimeStatus("run-1", { companyId: "company-1", agentId: "agent-1" })?.phase).toBe("run_activity");
|
||||
}, 30_000);
|
||||
|
||||
it("stops waiting as soon as the run is stopped", async () => {
|
||||
const stop = new AbortController();
|
||||
const begin = vi.fn(async () => outage());
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { notifyActiveRunSilence } from "../live-run-recovery-notifications.js";
|
||||
import { retryThroughStorageOutage } from "../../storage/storage-unavailable.js";
|
||||
import { retryThroughStorageOutage, storageOutageWaitFor } 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";
|
||||
@@ -127,7 +127,7 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
|
||||
if (run.logStore != null && !isRunLogStoreType(run.logStore))
|
||||
throw new Error("Run log backend is not supported");
|
||||
const previousStore = run.logStore ?? undefined;
|
||||
handle = await retryThroughStorageOutage(() => runLogStore.begin({
|
||||
handle = await storageOutageWaitFor("starting the run log", () => retryThroughStorageOutage(() => runLogStore.begin({
|
||||
companyId: run.companyId,
|
||||
agentId: run.agentId,
|
||||
runId,
|
||||
@@ -135,7 +135,7 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
|
||||
assertWriterAuthority: input.assertWriterAuthority,
|
||||
previousStore,
|
||||
signal: input.signal,
|
||||
}), input.signal);
|
||||
}), input.signal));
|
||||
|
||||
outputSeq = handle.initialSequence ?? outputSeq;
|
||||
persistedLogBytes = handle.initialBytes ?? persistedLogBytes;
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import type { heartbeatRuns } from "@paperclipai/db";
|
||||
import type { StorageOutageWaitObserver } from "../../storage/storage-unavailable.js";
|
||||
import { setHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js";
|
||||
import { publishHeartbeatRunRuntimeProgress, type createHeartbeatLifecycle } from "./run-lifecycle.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
export const STORAGE_WAIT_PHASE = "waiting_for_storage";
|
||||
|
||||
const seconds = (ms: number) => Math.max(1, Math.round(ms / 1000));
|
||||
|
||||
function runIssueId(run: Pick<Run, "contextSnapshot">): string | null {
|
||||
const issueId = (run.contextSnapshot as Record<string, unknown> | null)?.issueId;
|
||||
return typeof issueId === "string" && issueId.trim() ? issueId : null;
|
||||
}
|
||||
|
||||
export function createStorageWaitNotice(input: {
|
||||
run: Run;
|
||||
appendRunEvent: ReturnType<typeof createHeartbeatLifecycle>["appendRunEvent"];
|
||||
}): StorageOutageWaitObserver {
|
||||
const { run, appendRunEvent } = input;
|
||||
const status = (phase: typeof STORAGE_WAIT_PHASE | "run_activity", message: string) => {
|
||||
const at = new Date();
|
||||
const published = setHeartbeatRunRuntimeStatus({
|
||||
companyId: run.companyId, issueId: runIssueId(run), agentId: run.agentId, runId: run.id,
|
||||
phase, message, updatedAt: at, lastEventAt: at,
|
||||
});
|
||||
if (published) publishHeartbeatRunRuntimeProgress(published);
|
||||
};
|
||||
return {
|
||||
waiting({ purpose, waitedMs }) {
|
||||
status(STORAGE_WAIT_PHASE, `Waiting for file storage before ${purpose} (${seconds(waitedMs)}s so far)`);
|
||||
},
|
||||
ended({ purpose, waitedMs, recovered }) {
|
||||
const waited = seconds(waitedMs);
|
||||
status("run_activity", recovered
|
||||
? `File storage is back after ${waited}s`
|
||||
: `Stopped waiting for file storage after ${waited}s`);
|
||||
void Promise.resolve().then(() => appendRunEvent(run, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: recovered
|
||||
? `Waited ${waited}s for file storage before ${purpose}.`
|
||||
: `Waited ${waited}s for file storage before ${purpose}; it was still unavailable.`,
|
||||
payload: { reason: "storage_outage_wait", purpose, waitedMs, recovered },
|
||||
})).catch(() => undefined);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -17,7 +17,7 @@ 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 { isStorageUnavailableError, reportStorageOutageWaitsTo, runInOwnStorageScope, sendThroughStorageOutage } 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";
|
||||
@@ -455,6 +455,26 @@ suite("remote workspace snapshot storage", () => {
|
||||
await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally { temp.mockRestore(); }
|
||||
});
|
||||
it("tells the run it is waiting for storage while a save or a restore waits out an outage", async () => {
|
||||
const f = await fixture();
|
||||
const outage = () => withS3FailureDiagnostics("put_object", async () => {
|
||||
throw Object.assign(new Error("Injected storage outage"), { name: "ServiceUnavailable", $metadata: { httpStatusCode: 503 } });
|
||||
});
|
||||
let puts = 0, gets = 0, restoring = false;
|
||||
const flaky = { ...storage,
|
||||
putFile: (input: Parameters<StorageService["putFile"]>[0]) => sendThroughStorageOutage(async () => (++puts < 2 ? outage() : storage.putFile(input))),
|
||||
getObject: vi.fn(async (...args: Parameters<StorageService["getObject"]>) => (restoring && ++gets < 2 ? outage() : storage.getObject(...args))),
|
||||
} as StorageService;
|
||||
const waiting = vi.fn(), ended = vi.fn();
|
||||
await runInOwnStorageScope(async () => {
|
||||
reportStorageOutageWaitsTo({ waiting, ended });
|
||||
const ref = await uploadWorkspaceSnapshot({ ...f.upload, storage: flaky });
|
||||
restoring = true;
|
||||
await loadWorkspaceSnapshotArchive({ storage: flaky, reference: ref, targetPath: path.join(root, "waited.pcws") });
|
||||
});
|
||||
expect(waiting.mock.calls.map(([wait]) => wait.purpose)).toEqual(["saving the workspace", "restoring the workspace"]);
|
||||
expect(ended.mock.calls.map(([wait]) => [wait.purpose, wait.recovered])).toEqual([["saving the workspace", true], ["restoring the workspace", true]]);
|
||||
}, 30_000);
|
||||
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;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createCipheriv, createDecipheriv, createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { isStorageUnavailableError, sendThroughStorageOutage, waitOutStorageOutages } from "../storage/storage-unavailable.js";
|
||||
import { isStorageUnavailableError, sendThroughStorageOutage, storageOutageWaitFor, 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";
|
||||
@@ -200,7 +200,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, () => waitOutStorageOutages(() => uploadWorkspaceSnapshotTimed(input)));
|
||||
return withRequestStats(input.timing, () => waitOutStorageOutages(() => storageOutageWaitFor("saving the workspace", () => uploadWorkspaceSnapshotTimed(input))));
|
||||
}
|
||||
|
||||
async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise<WorkspaceSnapshotRef> {
|
||||
@@ -396,7 +396,7 @@ export async function loadWorkspaceSnapshotArchive(input: {
|
||||
/** Receives per-step durations and storage request counts. */
|
||||
timing?: WorkspaceSnapshotStorageTiming;
|
||||
}): Promise<void> {
|
||||
return withRequestStats(input.timing, () => waitOutStorageOutages(() => loadWorkspaceSnapshotArchiveTimed(input)));
|
||||
return withRequestStats(input.timing, () => waitOutStorageOutages(() => storageOutageWaitFor("restoring the workspace", () => loadWorkspaceSnapshotArchiveTimed(input))));
|
||||
}
|
||||
|
||||
async function loadWorkspaceSnapshotArchiveTimed(input: Parameters<typeof loadWorkspaceSnapshotArchive>[0]): Promise<void> {
|
||||
|
||||
@@ -78,8 +78,16 @@ export function isStorageUnavailableError(error: unknown): boolean {
|
||||
|
||||
export const STORAGE_OUTAGE_WAIT = { budgetMs: 5 * 60_000, firstDelayMs: 2_000, maxDelayMs: 30_000 } as const;
|
||||
|
||||
type OutageWaitScope = { stop?: AbortSignal };
|
||||
export type StorageOutageWaitObserver = {
|
||||
waiting(wait: { purpose: string; waitedMs: number }): void;
|
||||
ended(wait: { purpose: string; waitedMs: number; recovered: boolean }): void;
|
||||
};
|
||||
|
||||
const DEFAULT_WAIT_PURPOSE = "continuing";
|
||||
|
||||
type OutageWaitScope = { stop?: AbortSignal; observer?: StorageOutageWaitObserver };
|
||||
const outageWaitScope = new AsyncLocalStorage<OutageWaitScope>();
|
||||
const outageWaitPurpose = new AsyncLocalStorage<string>();
|
||||
type OutageWaitOperation = { firstFailureAt: number | null };
|
||||
const outageWaitOperation = new AsyncLocalStorage<OutageWaitOperation | null>();
|
||||
|
||||
@@ -97,14 +105,41 @@ export function runSharedStorageWork<T>(work: () => Promise<T>): Promise<T> {
|
||||
|
||||
export function endStorageOutageWaitOn(stop: AbortSignal): void {
|
||||
const current = outageWaitScope.getStore();
|
||||
if (current) outageWaitScope.enterWith({ stop: current.stop ? AbortSignal.any([current.stop, stop]) : stop });
|
||||
if (current) outageWaitScope.enterWith({ ...current, stop: current.stop ? AbortSignal.any([current.stop, stop]) : stop });
|
||||
}
|
||||
|
||||
export function reportStorageOutageWaitsTo(observer: StorageOutageWaitObserver): void {
|
||||
const current = outageWaitScope.getStore();
|
||||
if (current) outageWaitScope.enterWith({ ...current, observer });
|
||||
}
|
||||
|
||||
export function storageOutageWaitFor<T>(purpose: string, work: () => Promise<T>): Promise<T> {
|
||||
return outageWaitPurpose.run(purpose, work);
|
||||
}
|
||||
|
||||
function notify(report: () => void) {
|
||||
try { report(); } catch { return; }
|
||||
}
|
||||
|
||||
export function retryThroughStorageOutage<T>(send: () => Promise<T>, signal?: AbortSignal,
|
||||
policy: { budgetMs: number; firstDelayMs: number; maxDelayMs: number } = STORAGE_OUTAGE_WAIT): Promise<T> {
|
||||
const operation = outageWaitOperation.getStore();
|
||||
return operation ? retryWithin(operation, send, signal, policy)
|
||||
: outageWaitOperation.run({ firstFailureAt: null }, () => retryWithin(outageWaitOperation.getStore()!, send, signal, policy));
|
||||
if (operation) return retryWithin(operation, send, signal, policy);
|
||||
const owned: OutageWaitOperation = { firstFailureAt: null };
|
||||
return outageWaitOperation.run(owned, async () => {
|
||||
let recovered = false;
|
||||
try {
|
||||
const result = await retryWithin(owned, send, signal, policy);
|
||||
recovered = true;
|
||||
return result;
|
||||
} finally {
|
||||
const observer = outageWaitScope.getStore()?.observer;
|
||||
if (observer && owned.firstFailureAt !== null) {
|
||||
const ended = { purpose: outageWaitPurpose.getStore() ?? DEFAULT_WAIT_PURPOSE, waitedMs: Date.now() - owned.firstFailureAt, recovered };
|
||||
notify(() => observer.ended(ended));
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function retryWithin<T>(operation: OutageWaitOperation, send: () => Promise<T>, signal: AbortSignal | undefined,
|
||||
@@ -120,6 +155,11 @@ async function retryWithin<T>(operation: OutageWaitOperation, send: () => Promis
|
||||
operation.firstFailureAt ??= now;
|
||||
if (stop?.aborted || now - operation.firstFailureAt + wait > policy.budgetMs) throw error;
|
||||
recordStorageBackoff(wait);
|
||||
const observer = outageWaitScope.getStore()?.observer;
|
||||
if (observer) {
|
||||
const waiting = { purpose: outageWaitPurpose.getStore() ?? DEFAULT_WAIT_PURPOSE, waitedMs: now - operation.firstFailureAt };
|
||||
notify(() => observer.waiting(waiting));
|
||||
}
|
||||
await delay(wait, undefined, { signal: stop }).catch((interrupted: unknown) => {
|
||||
throw signal?.aborted && signal.reason instanceof DOMException && signal.reason.name === "TimeoutError" ? error : interrupted;
|
||||
});
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { deadRunningRun, STALL_LIMIT_MS } from "./grader.js";
|
||||
import { mergeLiveStatusEvidence, withLiveStatusEvidence } from "./live-status-evidence.js";
|
||||
import type { RunSample, TaskSample } from "./types.js";
|
||||
|
||||
const startedAt = Date.parse("2026-11-10T10:19:56.915Z");
|
||||
const now = startedAt + STALL_LIMIT_MS + 30_000;
|
||||
const run = (overrides: Partial<RunSample> = {}): RunSample => ({
|
||||
id: "9d39550a", status: "running", createdAt: null, startedAt: new Date(startedAt).toISOString(), finishedAt: null,
|
||||
errorCode: null, retryOfRunId: null, wakeCommentIds: [], livenessState: null,
|
||||
lastEvidenceAt: new Date(startedAt).toISOString(), ...overrides,
|
||||
});
|
||||
const sample = (runs: RunSample[]): TaskSample => ({ at: now, reachable: true, serverReady: true, issueStatus: "in_progress", runs }) as TaskSample;
|
||||
const waiting = { id: "9d39550a", currentStatusMessage: "Waiting for file storage before starting the run log (200s so far)",
|
||||
currentStatusUpdatedAt: new Date(now - 10_000).toISOString(), lastEventAt: new Date(now - 10_000).toISOString() };
|
||||
|
||||
describe("a run the board shows as waiting for storage (R5)", () => {
|
||||
it("is not a silent dead run while its live status stays fresh", () => {
|
||||
expect(deadRunningRun(sample([run()]), startedAt)).toMatchObject({ id: "9d39550a" });
|
||||
expect(deadRunningRun(sample(mergeLiveStatusEvidence([run()], [waiting])), startedAt)).toBeNull();
|
||||
});
|
||||
|
||||
it("still fails a run whose live status went stale too", () => {
|
||||
const stale = { ...waiting, currentStatusUpdatedAt: new Date(startedAt).toISOString(), lastEventAt: null };
|
||||
expect(deadRunningRun(sample(mergeLiveStatusEvidence([run()], [stale])), startedAt)).toMatchObject({ id: "9d39550a" });
|
||||
});
|
||||
|
||||
it("keeps live activity it already saw when a later live read fails", async () => {
|
||||
const api = { get: vi.fn(async () => [waiting]) };
|
||||
await withLiveStatusEvidence(api as never, "issue-1", [run()]);
|
||||
api.get.mockRejectedValueOnce(new Error("unreachable"));
|
||||
const [later] = await withLiveStatusEvidence(api as never, "issue-1", [run()]);
|
||||
expect(later.lastEvidenceAt).toBe(waiting.currentStatusUpdatedAt);
|
||||
expect(deadRunningRun(sample([later]), startedAt)).toBeNull();
|
||||
});
|
||||
|
||||
it("reads live status only while a run is running, and tolerates a failed read", async () => {
|
||||
const get = vi.fn(async () => [waiting]);
|
||||
expect(await withLiveStatusEvidence({ get } as never, "issue-1", [run({ status: "succeeded" })])).toHaveLength(1);
|
||||
expect(get).not.toHaveBeenCalled();
|
||||
const [merged] = await withLiveStatusEvidence({ get } as never, "issue-1", [run()]);
|
||||
expect(get).toHaveBeenCalledWith("/api/issues/issue-1/live-runs", 5_000);
|
||||
expect(merged.lastEvidenceAt).toBe(waiting.currentStatusUpdatedAt);
|
||||
const failing = { get: vi.fn(async (): Promise<unknown> => { throw new Error("unreachable"); }) };
|
||||
expect(await withLiveStatusEvidence(failing as never, "issue-1", [run()])).toEqual([run()]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,38 @@
|
||||
import type { RunSample } from "./types.js";
|
||||
|
||||
type LiveRun = { id?: unknown; currentStatusUpdatedAt?: unknown; lastEventAt?: unknown };
|
||||
|
||||
function latest(values: ReadonlyArray<unknown>): string | null {
|
||||
let best: number | null = null;
|
||||
for (const value of values) {
|
||||
const at = typeof value === "string" ? Date.parse(value) : NaN;
|
||||
if (Number.isFinite(at) && (best === null || at > best)) best = at;
|
||||
}
|
||||
return best === null ? null : new Date(best).toISOString();
|
||||
}
|
||||
|
||||
export function mergeLiveStatusEvidence(runs: RunSample[], live: ReadonlyArray<LiveRun> | null): RunSample[] {
|
||||
if (!live?.length) return runs;
|
||||
const byId = new Map(live.filter(run => typeof run.id === "string").map(run => [run.id as string, run]));
|
||||
return runs.map(run => {
|
||||
const status = run.status === "running" ? byId.get(run.id) : undefined;
|
||||
if (!status) return run;
|
||||
return { ...run, lastEvidenceAt: latest([run.lastEvidenceAt, status.currentStatusUpdatedAt, status.lastEventAt]) };
|
||||
});
|
||||
}
|
||||
|
||||
type LiveApi = { get<T>(path: string, timeoutMs?: number): Promise<T> };
|
||||
const seen = new WeakMap<LiveApi, Map<string, string>>();
|
||||
|
||||
export async function withLiveStatusEvidence(api: LiveApi, issueId: string, runs: RunSample[]): Promise<RunSample[]> {
|
||||
if (!runs.some(run => run.status === "running")) return runs;
|
||||
const live = await api.get<LiveRun[]>(`/api/issues/${issueId}/live-runs`, 5_000).catch(() => null);
|
||||
const merged = mergeLiveStatusEvidence(runs, Array.isArray(live) ? live : null);
|
||||
let remembered = seen.get(api);
|
||||
if (!remembered) { remembered = new Map(); seen.set(api, remembered); }
|
||||
return merged.map(run => {
|
||||
const lastEvidenceAt = latest([run.lastEvidenceAt, remembered!.get(run.id)]);
|
||||
if (lastEvidenceAt) remembered!.set(run.id, lastEvidenceAt);
|
||||
return lastEvidenceAt === (run.lastEvidenceAt ?? null) ? run : { ...run, lastEvidenceAt };
|
||||
});
|
||||
}
|
||||
@@ -20,6 +20,7 @@ import { captureRemoteWorkspaceFiles, readRemoteWorkspaceFile } from "./remote-w
|
||||
import { taskFaultSchedule } from "./scheduler.js";
|
||||
import { g1ChildTitle, g1Decision, workloadDefinitions, type WorkloadAgents } from "./workloads.js";
|
||||
import { databaseUrlFromConfig, journeyDriverFor, type JourneyDriver } from "./journey-driver.js";
|
||||
import { withLiveStatusEvidence } from "./live-status-evidence.js";
|
||||
|
||||
const IGNORED_DIRECTORIES = new Set([".git", ".bench", "__pycache__", "node_modules"]);
|
||||
const MAX_SNAPSHOT_FILES = 2_000;
|
||||
@@ -548,7 +549,7 @@ export class TaskObserver {
|
||||
interruption: toInterruption(issue.interruption), activeRecoveryAction: Boolean(issue.activeRecoveryAction),
|
||||
recoveryOwner: issue.activeRecoveryAction?.ownerType ?? null,
|
||||
executionBlocker: Boolean(issue.executionBlocker), scheduledRetry: Boolean(issue.scheduledRetry),
|
||||
runs: await Promise.all((runs ?? []).map(toRunSample).map(run => this.runEvidence(run))) };
|
||||
runs: await withLiveStatusEvidence(this.api, this.record.issueId!, await Promise.all((runs ?? []).map(toRunSample).map(run => this.runEvidence(run)))) };
|
||||
} catch {
|
||||
return { at: now, reachable: false, serverReady: false };
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user