diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 7c24efe7ba1..66fdcd6852c 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -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 ." or, when storage stayed unavailable, "Waited Ns +for file storage before ; 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. diff --git a/packages/shared/src/types/heartbeat.ts b/packages/shared/src/types/heartbeat.ts index a14cefbbdd1..e5449f8a58a 100644 --- a/packages/shared/src/types/heartbeat.ts +++ b/packages/shared/src/types/heartbeat.ts @@ -256,6 +256,7 @@ export type HeartbeatRunStatusPhase = | "restore" | "export" | "finalize" + | "waiting_for_storage" | "run_activity"; export type HeartbeatRunOutputSilenceLevel = diff --git a/server/src/__tests__/storage-outage-wait.test.ts b/server/src/__tests__/storage-outage-wait.test.ts index 3599bea5630..fdf54784c42 100644 --- a/server/src/__tests__/storage-outage-wait.test.ts +++ b/server/src/__tests__/storage-outage-wait.test.ts @@ -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 = ` @@ -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)) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 16be14da953..dbb1836d81b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -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"); diff --git a/server/src/services/heartbeat/run-output.test.ts b/server/src/services/heartbeat/run-output.test.ts index 6bb537af2af..26becfaf358 100644 --- a/server/src/services/heartbeat/run-output.test.ts +++ b/server/src/services/heartbeat/run-output.test.ts @@ -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; + const writes: Partial[] = []; + 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) => unknown })(table); + return { set: (values: Partial) => { 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 }) => { + 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()); diff --git a/server/src/services/heartbeat/run-output.ts b/server/src/services/heartbeat/run-output.ts index 90a916a6743..06163b15cb7 100644 --- a/server/src/services/heartbeat/run-output.ts +++ b/server/src/services/heartbeat/run-output.ts @@ -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; diff --git a/server/src/services/heartbeat/storage-wait-notice.ts b/server/src/services/heartbeat/storage-wait-notice.ts new file mode 100644 index 00000000000..4ea24a1f5d0 --- /dev/null +++ b/server/src/services/heartbeat/storage-wait-notice.ts @@ -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): string | null { + const issueId = (run.contextSnapshot as Record | null)?.issueId; + return typeof issueId === "string" && issueId.trim() ? issueId : null; +} + +export function createStorageWaitNotice(input: { + run: Run; + appendRunEvent: ReturnType["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); + }, + }; +} diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 59308c169b6..bf273ae940b 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -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[0]) => sendThroughStorageOutage(async () => (++puts < 2 ? outage() : storage.putFile(input))), + getObject: vi.fn(async (...args: Parameters) => (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; diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index b79a1dcf21e..07cc627cdab 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -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 { - return withRequestStats(input.timing, () => waitOutStorageOutages(() => uploadWorkspaceSnapshotTimed(input))); + return withRequestStats(input.timing, () => waitOutStorageOutages(() => storageOutageWaitFor("saving the workspace", () => uploadWorkspaceSnapshotTimed(input)))); } async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise { @@ -396,7 +396,7 @@ export async function loadWorkspaceSnapshotArchive(input: { /** Receives per-step durations and storage request counts. */ timing?: WorkspaceSnapshotStorageTiming; }): Promise { - return withRequestStats(input.timing, () => waitOutStorageOutages(() => loadWorkspaceSnapshotArchiveTimed(input))); + return withRequestStats(input.timing, () => waitOutStorageOutages(() => storageOutageWaitFor("restoring the workspace", () => loadWorkspaceSnapshotArchiveTimed(input)))); } async function loadWorkspaceSnapshotArchiveTimed(input: Parameters[0]): Promise { diff --git a/server/src/storage/storage-unavailable.ts b/server/src/storage/storage-unavailable.ts index 865f341643e..2772bf7701e 100644 --- a/server/src/storage/storage-unavailable.ts +++ b/server/src/storage/storage-unavailable.ts @@ -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(); +const outageWaitPurpose = new AsyncLocalStorage(); type OutageWaitOperation = { firstFailureAt: number | null }; const outageWaitOperation = new AsyncLocalStorage(); @@ -97,14 +105,41 @@ export function runSharedStorageWork(work: () => Promise): Promise { 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(purpose: string, work: () => Promise): Promise { + return outageWaitPurpose.run(purpose, work); +} + +function notify(report: () => void) { + try { report(); } catch { return; } } 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)); + 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(operation: OutageWaitOperation, send: () => Promise, signal: AbortSignal | undefined, @@ -120,6 +155,11 @@ async function retryWithin(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; }); diff --git a/tests/restart-chaos/live-status-evidence.test.ts b/tests/restart-chaos/live-status-evidence.test.ts new file mode 100644 index 00000000000..3da5da6f796 --- /dev/null +++ b/tests/restart-chaos/live-status-evidence.test.ts @@ -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 => ({ + 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 => { throw new Error("unreachable"); }) }; + expect(await withLiveStatusEvidence(failing as never, "issue-1", [run()])).toEqual([run()]); + }); +}); diff --git a/tests/restart-chaos/live-status-evidence.ts b/tests/restart-chaos/live-status-evidence.ts new file mode 100644 index 00000000000..e82f1e724c8 --- /dev/null +++ b/tests/restart-chaos/live-status-evidence.ts @@ -0,0 +1,38 @@ +import type { RunSample } from "./types.js"; + +type LiveRun = { id?: unknown; currentStatusUpdatedAt?: unknown; lastEventAt?: unknown }; + +function latest(values: ReadonlyArray): 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 | 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(path: string, timeoutMs?: number): Promise }; +const seen = new WeakMap>(); + +export async function withLiveStatusEvidence(api: LiveApi, issueId: string, runs: RunSample[]): Promise { + if (!runs.some(run => run.status === "running")) return runs; + const live = await api.get(`/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 }; + }); +} diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index 15abf96dfd2..0bbcdc4fabd 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -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 }; }