mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(native): bound failover backup restores by inactivity, not total time (#19906)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A native per-turn session that continues an earlier conversation first restores its failover backup from durable storage. This happens before the coordinator claim, so no provider has started. > - That restore ran under a fixed 300 s total deadline (`AbortSignal.timeout(300_000)`). The deadline covers the download and the local decrypt, digest, extraction and provider-artifact steps. > - On staging, restore time for resumed sessions grew during the day. It reached 300 s, and then every resumed run of one review agent failed with the generic "The operation was aborted". Fresh sessions were not affected. > - A failure at this point also enters the saved-work reconciliation path, so the user request stopped. > - This pull request bounds each durable read by inactivity instead of total time. A slow transfer continues while bytes arrive. A read that stops sending bytes fails with a named code. > - The benefit is that resumed sessions no longer fail only because their backup is large or the store is slow. ## Linked Issues or Issue Description Refs #19894 (adds the same idle-deadline rule for workspace snapshot reads; this PR covers the separate native failover-backup reads). **What happened?** On staging, a PR-review agent had 17 runs fail in about 80 minutes. Each failure had the message "The operation was aborted" and the code `native_continuation_requires_reconciliation`. Every failing run resumed an existing session (`sessionIdBefore` set, per-turn lifecycle). Every fresh-session run in the same window succeeded. In each failing run, the error came 304–309 s after the `environment.startup` span, and no `native.coordinator.claim` span was recorded. The run result had `executionRecovery: { kind: "bootstrap", providerWorkStarted: false }`. Earlier resumed runs of the same agent show the time from `environment.startup` to `native.coordinator.claim` growing from about 30 s to about 200 s before the failures started. The only fixed 300 s deadline on that path is the failover-backup restore in `restoreDurableHarnessForExecution`. **Expected behavior** A resumed session restores its backup when the store keeps sending bytes, even when the total time is more than five minutes. A read that stalls fails with a named, diagnosable code. A failed restore publishes no partial backup and keeps the saved backup. **Steps to reproduce** 1. Persist a native failover backup with provider artifacts. 2. Restore it through a storage reader that sends bytes slowly, so the full restore takes longer than the deadline. 3. Before this change, the restore fails with "The operation was aborted". A reader that stops sending bytes, with no caller deadline, waits indefinitely. ## What Changed - `native-harness-backup-storage.ts`: add `downloadIdleTimeoutMs` to `restoreDurableNativeHarnessBackup` and `verifyDurableNativeHarnessBackup`. When set, each durable `getObject` (the backup object and the provider artifacts) runs under an idle timer. Each received chunk renews the timer. The timer is active only while a read is in flight, so local decrypt, digest, extraction and copy are never timed. On expiry the whole restore aborts and throws `native_harness_storage_download_stalled`. Caller cancellation keeps its own error. - `verifyRecoverableNativeHarnessBackup` and the executor's `restoreDurableHarnessForExecution` use the five-minute idle limit (`NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS`) instead of `AbortSignal.timeout(300_000)`. - The whole restore or verification keeps a 30-minute outer limit. It fails with `native_harness_storage_restore_deadline_exceeded`, so a hung local step stays bounded and named. - The executor passes the startup Stop signal (`providerPackSourceAbort`) into the restore, so Stop (also a Stop that came before the restore began) ends it. Only a startup that already completed does not reuse its signal. The restore checks the signal after the local copy and before it publishes `current`. - The wrapped storage is a `Proxy` that replaces only `getObject` and forwards every other member to the real service. - `doc/DEVELOPING.md`: document the idle limit and the named code. Previous-release story (`rules/upgrade-safe-run-state.mdc`): no persisted shape changes. Backups, artifact rows, receipts and checkpoints saved by the previous release restore unchanged. A deploy that lands during a restore stops the process. The new process starts the restore again from the saved durable backup under the new limit. The existing hash, size, encryption and atomic publication checks are unchanged. ## Verification Run from `server/`: - `pnpm exec vitest run src/services/native-runtime/native-harness-backup-download-deadline.integration.test.ts` — 4 tests passed (embedded PostgreSQL, local-disk storage, an OpenCode backup with provider artifacts). The tests are in a focused suite, so the existing 990-line storage suite does not grow. - `pnpm exec vitest run src/services/native-runtime/native-harness-backup-storage.integration.test.ts` — 68 existing tests passed. - New: "keeps restoring a backup whose download outlasts the idle window while bytes keep arriving". With the former total-deadline policy, the same slow transfer fails with "The operation was aborted" and publishes nothing (the staging message). With the idle limit, the restore completes after more than twice the window, restores the session and the OpenCode launch artifacts, and verification also completes. - New: "names a download that stops sending bytes and publishes no partial backup". It checks restore and verify, checks that no `current` backup is published, and then restores the same saved backup successfully. - New: "stops at the outer bound during local copy and publishes nothing". - New: "lets Stop end a progressing download with its own reason, including a Stop before it starts". - RED: with the production files reverted to `origin/master`, the stall test does not finish and times out after 10 s, because no stall is detected. The other two tests pass on the old code, because the total deadline was the executor's caller policy. The first test reproduces that policy directly. - `pnpm exec vitest run` on `native-harness-online-preservation`, `native-harness-filesystem-independence` and `native-controller-migration` integration suites passed. `native-warm-checkpoint-ownership.integration.test.ts` fails locally on master as well, because the local Rust `paperclip-runnerd` binary is not built. - `node scripts/typecheck-layers.mjs` passed for all layers. Not verified: a live resumed run on staging with this build. The staging evidence shows where the time goes (before the coordinator claim, on resumed sessions only) but does not show which restore step was slowest. The fix removes the total deadline for every step on that path. ## Risks - A restore that progresses very slowly can now take longer than five minutes. Before, it failed. The run still stops on caller cancellation. - A local step that hangs (for example a stalled disk) is now cut off by the 30-minute outer limit instead of 5 minutes. A stalled local writer also pauses the read through backpressure, so it can be reported as `native_harness_storage_download_stalled`. - A failure before the coordinator claim still enters the existing saved-work reconciliation path. This PR does not change that classification. It removes the cause of these failures. ## Model Used - Provider: Anthropic. Model: Claude Opus 5.5 (`claude-opus-5-5`), with tool use (shell, file edits, read-only browser for API inspection) 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 - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
57ae2fc309
commit
76994c5027
4 files changed
+271
-5
No files matched your search
@@ -4070,6 +4070,13 @@ Environment overrides:
|
||||
negative, empty, or non-numeric value falls back to the default.
|
||||
`PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_ENABLED=false` turns expiry off, and
|
||||
`PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DRY_RUN=true` only reports candidates.
|
||||
A resumed session restores its failover backup from durable storage before the
|
||||
provider starts. Each read has a five-minute idle limit instead of a short total
|
||||
limit: a large or slow backup keeps restoring while bytes flow. A read that makes
|
||||
no progress for five minutes fails with `native_harness_storage_download_stalled`
|
||||
and publishes no partial backup. The whole restore, including local decrypt,
|
||||
digest and extraction, has a 30-minute outer limit
|
||||
(`native_harness_storage_restore_deadline_exceeded`). Stop ends the restore.
|
||||
- `PAPERCLIP_NATIVE_RUNNER_EVIDENCE_RETENTION_DAYS=<days>` sets how long native
|
||||
session continuity-break archives and quarantined session state are kept (see
|
||||
`doc/recoverable-disk-cleanup.md`). The default is `7` and the minimum is `1`. A
|
||||
|
||||
+173
@@ -0,0 +1,173 @@
|
||||
import fs from "node:fs/promises";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { mkdir, readFile, stat, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { Readable } from "node:stream";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, nativeHarnessBackups, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../../storage/service.js";
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup,
|
||||
verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
|
||||
vi.mock("node:fs/promises", async importOriginal => {
|
||||
const original = await importOriginal<typeof import("node:fs/promises")>();
|
||||
const files = { ...original };
|
||||
return { ...original, default: files, cp: (...args: Parameters<typeof original.cp>) => files.cp(...args) };
|
||||
});
|
||||
|
||||
/** Durable reads that deliver `chunkBytes` every `delayMs`, optionally stopping after `stallAfter` chunks. */
|
||||
function throttledStorage(storage: StorageService, delayMs: number, chunkBytes: number, stallAfter?: number) {
|
||||
let sent = 0;
|
||||
const reads: string[] = [];
|
||||
const throttled = new Proxy(storage, { get(target, property) {
|
||||
if (property !== "getObject") {
|
||||
const value = Reflect.get(target, property, target) as unknown;
|
||||
return typeof value === "function" ? (value as (...args: unknown[]) => unknown).bind(target) : value;
|
||||
}
|
||||
return async (companyId: string, objectKey: string, options?: { signal?: AbortSignal }) => {
|
||||
reads.push(objectKey);
|
||||
const object = await target.getObject(companyId, objectKey, options);
|
||||
async function* slow() {
|
||||
for await (const part of object.stream as AsyncIterable<Buffer>) {
|
||||
for (let offset = 0; offset < part.length; offset += chunkBytes) {
|
||||
if (stallAfter !== undefined && sent >= stallAfter) {
|
||||
await new Promise<void>(resolve => options?.signal?.addEventListener("abort", () => resolve(), { once: true }));
|
||||
options?.signal?.throwIfAborted();
|
||||
}
|
||||
await new Promise(resolve => setTimeout(resolve, delayMs));
|
||||
options?.signal?.throwIfAborted();
|
||||
sent++;
|
||||
yield part.subarray(offset, offset + chunkBytes);
|
||||
}
|
||||
}
|
||||
}
|
||||
return { ...object, stream: Readable.from(slow()) };
|
||||
};
|
||||
} });
|
||||
return { storage: throttled, reads };
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("durable native backup download deadlines", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const ownedHomes: string[] = [];
|
||||
afterEach(async () => {
|
||||
for (const home of ownedHomes.splice(0)) await removeNativeHarnessBackup(home);
|
||||
vi.restoreAllMocks();
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
/** An OpenCode backup with provider artifacts, so restores make two separate durable reads. */
|
||||
async function fixture() {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, runtimeRoot, successor, agentId } = environment.current;
|
||||
await f.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId, controllerGeneration: 1,
|
||||
leaseOwner: "backup-writer", leaseExpiresAt: new Date(Date.now() + 120_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const owner = await nativeHarnessBackupOwner(f.db, { companyId, issueId: f.issueId, runId: f.runId, bootId: successor.bootId });
|
||||
const home = join(runtimeRoot, randomUUID()), root = join(home, "failover-backups", "current");
|
||||
ownedHomes.push(home);
|
||||
await mkdir(join(root, "opencode", "sessions"), { recursive: true });
|
||||
const identity = { normalizedSessionId: randomUUID(), providerSessionId: randomUUID(), provider: "opencode" };
|
||||
// Enough bytes for many throttled chunks, so a restore outlasts the idle window.
|
||||
await writeFile(join(root, "opencode", "sessions", "history.json"), JSON.stringify({ ...identity, turns: "x".repeat(4096) }));
|
||||
const manifest = JSON.stringify({ schema: "paperclip.native-harness-backup.v1", ...identity, directories: ["opencode"] });
|
||||
await writeFile(join(root, "manifest.json"), manifest, { mode: 0o600 });
|
||||
const sourceRoot = join(home, "qualified-provider-pack");
|
||||
await mkdir(sourceRoot);
|
||||
const original = { nodeCommand: "qualified Node executable", opencodeExecutable: "qualified OpenCode executable",
|
||||
opencodeProxy: `qualified proxy bytes ${randomUUID()}` };
|
||||
const artifacts = Object.fromEntries(await Promise.all(Object.entries(original).map(async ([name, bytes]) => {
|
||||
await writeFile(join(sourceRoot, name), bytes, { mode: 0o500 });
|
||||
return [name, { path: name, sha256: `sha256:${createHash("sha256").update(bytes).digest("hex")}` }];
|
||||
})));
|
||||
const payload = { runnerSourceRevision: "a".repeat(40), target: { platform: "linux", architecture: "x64" },
|
||||
pins: { opencode: "1.18.34" }, artifacts: { ...artifacts, opencodeCommand: artifacts.opencodeExecutable } };
|
||||
const packManifest = { schema: "paperclip-runner/remote-provider-pack/v1", payload,
|
||||
digest: `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}` };
|
||||
const launchRoot = `/home/daytona/.paperclip-runtime/paperclip-runner/provider-artifacts/${packManifest.digest.slice(7)}`;
|
||||
await f.db.update(heartbeatRuns).set({ nativeSessionId: identity.normalizedSessionId, driverKind: "opencode_server",
|
||||
runnerProfileJson: { ...f.preparedProfile, nativeProviderPackAdmissionState: {
|
||||
schema: "paperclip.native-provider-pack-admission-state.v1", state: "verified", admissionEpoch: randomUUID() },
|
||||
nativeProviderPackAdmission: { schema: "paperclip.native-provider-pack-admission.v1", stage: "session_admitted",
|
||||
authority: "core_verified_selection", companyId, agentId, runId: f.runId, sessionId: identity.normalizedSessionId,
|
||||
runnerInstanceId: "deadline-fixture-runner", runnerEnvironmentLeaseId: "deadline-fixture-lease",
|
||||
providerPackDigest: packManifest.digest, opencodeProxySha256: artifacts.opencodeProxy.sha256,
|
||||
sourceRevision: payload.runnerSourceRevision } },
|
||||
}).where(eq(heartbeatRuns.id, f.runId));
|
||||
const scopeDigest = createHash("sha256").update(randomUUID()).digest("hex");
|
||||
const storage = createStorageService(createLocalDiskStorageProvider(join(runtimeRoot, "durable-objects")));
|
||||
await persistNativeHarnessBackup({ db: f.db, root, owner, scopeDigest, storage,
|
||||
providerArtifacts: { sourceRoot, manifest: packManifest, launchRoot, accountUid: 1000 } });
|
||||
const [row] = await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, scopeDigest),
|
||||
eq(nativeHarnessBackups.status, "current")));
|
||||
return { db: f.db, companyId, scopeDigest, storage, identity, original, manifestSha256: row.manifestSha256 };
|
||||
}
|
||||
function replacementRoot() {
|
||||
const root = join(environment.current.runtimeRoot, randomUUID(), "failover-backups");
|
||||
ownedHomes.push(root);
|
||||
return root;
|
||||
}
|
||||
|
||||
it("keeps restoring a backup whose downloads outlast the idle window while bytes keep arriving", async () => {
|
||||
const f = await fixture(), idle = 250, slow = throttledStorage(f.storage, 15, 64);
|
||||
// The former total deadline aborted this same progressing transfer with
|
||||
// the generic "The operation was aborted" seen on resumed sessions.
|
||||
const aborted = replacementRoot();
|
||||
await expect(restoreDurableNativeHarnessBackup({ ...f, storage: slow.storage, backupRoot: aborted, signal: AbortSignal.timeout(idle) }))
|
||||
.rejects.toThrow("The operation was aborted");
|
||||
await expect(stat(join(aborted, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const replacement = replacementRoot(), startedAt = Date.now();
|
||||
slow.reads.splice(0);
|
||||
expect(await restoreDurableNativeHarnessBackup({ ...f, storage: slow.storage, backupRoot: replacement,
|
||||
downloadIdleTimeoutMs: idle })).toBe(true);
|
||||
expect(Date.now() - startedAt).toBeGreaterThan(idle * 2);
|
||||
// Both the backup object and the provider artifacts were read through the deadline.
|
||||
expect(slow.reads.some(key => key.includes("/native-harness-backups/"))).toBe(true);
|
||||
expect(slow.reads.some(key => !key.includes("/native-harness-backups/"))).toBe(true);
|
||||
expect(JSON.parse(await readFile(join(replacement, "current/opencode/sessions/history.json"), "utf8")))
|
||||
.toMatchObject(f.identity);
|
||||
expect(await readFile(join(replacement, "current/provider-artifacts/opencodeProxy"), "utf8")).toBe(f.original.opencodeProxy);
|
||||
expect(await verifyDurableNativeHarnessBackup({ ...f, storage: slow.storage, downloadIdleTimeoutMs: idle })).toBe(true);
|
||||
});
|
||||
|
||||
it("names a download that stops sending bytes and publishes no partial backup", { timeout: 10_000 }, async () => {
|
||||
const f = await fixture(), replacement = replacementRoot();
|
||||
await expect(restoreDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 5, 64, 3).storage,
|
||||
backupRoot: replacement, downloadIdleTimeoutMs: 250 })).rejects.toThrow("native_harness_storage_download_stalled");
|
||||
await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(verifyDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 5, 64, 3).storage,
|
||||
downloadIdleTimeoutMs: 250 })).rejects.toThrow("native_harness_storage_download_stalled");
|
||||
// The saved backup is untouched and restores once the store responds again.
|
||||
expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, downloadIdleTimeoutMs: 250 })).toBe(true);
|
||||
});
|
||||
|
||||
it("stops at the outer bound during local copy and publishes nothing", async () => {
|
||||
const f = await fixture(), replacement = replacementRoot(), copy = fs.cp.bind(fs);
|
||||
vi.spyOn(fs, "cp").mockImplementation(async (source, destination, options) => {
|
||||
if (String(destination).includes(".durable-")) await new Promise(resolve => setTimeout(resolve, 400));
|
||||
return copy(source, destination, options);
|
||||
});
|
||||
await expect(restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, downloadIdleTimeoutMs: 5_000,
|
||||
restoreMaxMs: 200 })).rejects.toThrow("native_harness_storage_restore_deadline_exceeded");
|
||||
await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("lets Stop end a progressing download with its own reason, including a Stop before it starts", async () => {
|
||||
const f = await fixture(), replacement = replacementRoot(), stop = new AbortController();
|
||||
setTimeout(() => stop.abort(new Error("caller_stopped")), 50);
|
||||
const failure = await restoreDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 15, 64).storage,
|
||||
backupRoot: replacement, signal: stop.signal, downloadIdleTimeoutMs: 5_000 }).then(() => null, (error: unknown) => error);
|
||||
const reason = failure as Error & { cause?: unknown };
|
||||
expect(failure).toBeInstanceOf(Error);
|
||||
expect([reason.message, (reason.cause as Error | undefined)?.message]).toContain("caller_stopped");
|
||||
await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, signal: stop.signal,
|
||||
downloadIdleTimeoutMs: 5_000 })).rejects.toThrow();
|
||||
await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
});
|
||||
@@ -57,7 +57,7 @@ export async function verifyRecoverableNativeHarnessBackup(input: {
|
||||
const storage = durableNativeHarnessStorage();
|
||||
if (!storage) return verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId);
|
||||
const durable = await verifyDurableNativeHarnessBackup({ db: input.db, storage, companyId: input.companyId,
|
||||
scopeDigest: stamp.sessionScopeSha256, manifestSha256: stamp.manifestSha256, signal: AbortSignal.timeout(300_000) });
|
||||
scopeDigest: stamp.sessionScopeSha256, manifestSha256: stamp.manifestSha256, downloadIdleTimeoutMs: NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS });
|
||||
return durable || process.env.PAPERCLIP_EPHEMERAL_SERVER !== "true" && verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId);
|
||||
}
|
||||
function reference(row: Row): Reference {
|
||||
@@ -287,7 +287,82 @@ export async function persistNativeHarnessBackup(input: {
|
||||
} finally { key.fill(0); await removeNativeHarnessBackup(scratch); }
|
||||
}
|
||||
|
||||
/** How long a durable read may go without progress. */
|
||||
export const NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS = 300_000;
|
||||
/** Outer bound for one whole restore or verification, including local steps. */
|
||||
export const NATIVE_HARNESS_RESTORE_MAX_MS = 30 * 60_000;
|
||||
export const NATIVE_HARNESS_DOWNLOAD_STALLED = "native_harness_storage_download_stalled";
|
||||
export const NATIVE_HARNESS_RESTORE_DEADLINE_EXCEEDED = "native_harness_storage_restore_deadline_exceeded";
|
||||
|
||||
/**
|
||||
* Bound each durable read by inactivity, not by a short total time. A large or
|
||||
* slow backup keeps restoring while bytes flow through the read; a read that
|
||||
* makes no progress for the idle window aborts the whole restore with a named
|
||||
* error. A generous outer bound keeps a hung local step diagnosable. Caller
|
||||
* cancellation (Stop) keeps its own reason.
|
||||
*/
|
||||
async function withDownloadProgressDeadline<T>(
|
||||
input: { storage: StorageService; signal?: AbortSignal; downloadIdleTimeoutMs?: number; restoreMaxMs?: number },
|
||||
run: (storage: StorageService, signal: AbortSignal | undefined) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const idleMs = input.downloadIdleTimeoutMs;
|
||||
if (idleMs === undefined) return run(input.storage, input.signal);
|
||||
if (!Number.isSafeInteger(idleMs) || idleMs <= 0) throw new Error("native_harness_storage_idle_timeout_invalid");
|
||||
const stall = new AbortController();
|
||||
const outer = setTimeout(() => stall.abort(new Error(NATIVE_HARNESS_RESTORE_DEADLINE_EXCEEDED)),
|
||||
input.restoreMaxMs ?? Math.max(idleMs, NATIVE_HARNESS_RESTORE_MAX_MS));
|
||||
outer.unref?.();
|
||||
const signal = input.signal ? AbortSignal.any([input.signal, stall.signal]) : stall.signal;
|
||||
const base = input.storage;
|
||||
const getObject: StorageService["getObject"] = async (companyId, objectKey, options) => {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const disarm = () => { if (timer) clearTimeout(timer); timer = undefined; };
|
||||
const arm = () => {
|
||||
disarm();
|
||||
timer = setTimeout(() => stall.abort(new Error(NATIVE_HARNESS_DOWNLOAD_STALLED)), idleMs);
|
||||
timer.unref?.();
|
||||
};
|
||||
arm();
|
||||
let object: Awaited<ReturnType<StorageService["getObject"]>>;
|
||||
try {
|
||||
object = await base.getObject(companyId, objectKey, {
|
||||
...options, signal: options?.signal ? AbortSignal.any([options.signal, signal]) : signal });
|
||||
} catch (error) { disarm(); throw error; }
|
||||
arm();
|
||||
// Chunks pass through as the consumer reads them, so a read is idle when
|
||||
// neither the store nor the local writer makes progress.
|
||||
const observed = new Transform({
|
||||
transform(bytes, _encoding, callback) { arm(); callback(null, bytes); },
|
||||
flush(callback) { disarm(); callback(); },
|
||||
});
|
||||
observed.once("close", disarm);
|
||||
pipeline(object.stream, observed, { signal }).catch(() => disarm());
|
||||
return { ...object, stream: observed };
|
||||
};
|
||||
// Forward every other member to the real service so its own state stays intact.
|
||||
const storage = new Proxy(base, {
|
||||
get(target, property) {
|
||||
if (property === "getObject") return getObject;
|
||||
const value = Reflect.get(target, property, target) as unknown;
|
||||
return typeof value === "function" ? (value as (...args: unknown[]) => unknown).bind(target) : value;
|
||||
},
|
||||
});
|
||||
try {
|
||||
return await run(storage, signal);
|
||||
} catch (error) {
|
||||
if (stall.signal.aborted && !input.signal?.aborted) throw stall.signal.reason;
|
||||
throw error;
|
||||
} finally { clearTimeout(outer); }
|
||||
}
|
||||
|
||||
export async function restoreDurableNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string;
|
||||
backupRoot: string; signal?: AbortSignal; downloadIdleTimeoutMs?: number; restoreMaxMs?: number;
|
||||
}): Promise<boolean> {
|
||||
return withDownloadProgressDeadline(input, (storage, signal) => restoreCurrentNativeHarnessBackup({ ...input, storage, signal }));
|
||||
}
|
||||
|
||||
async function restoreCurrentNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string;
|
||||
backupRoot: string; signal?: AbortSignal;
|
||||
}): Promise<boolean> {
|
||||
@@ -332,6 +407,8 @@ export async function restoreDurableNativeHarnessBackup(input: {
|
||||
const pending = join(input.backupRoot, `.durable-${randomUUID()}`);
|
||||
try {
|
||||
await cp(restored, pending, { recursive: true, preserveTimestamps: true, verbatimSymlinks: true, errorOnExist: true, force: false });
|
||||
// A deadline or Stop that fired during the local copy must not publish.
|
||||
input.signal?.throwIfAborted();
|
||||
await publishNativeHarnessBackup({ db: input.db, companyId: input.companyId, scopeDigest: input.scopeDigest,
|
||||
backupRoot: input.backupRoot, pendingRoot: pending });
|
||||
} finally { await removeNativeHarnessBackup(pending); }
|
||||
@@ -340,13 +417,14 @@ export async function restoreDurableNativeHarnessBackup(input: {
|
||||
}
|
||||
|
||||
export async function verifyDurableNativeHarnessBackup(input: {
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string; manifestSha256: string; backupId?: string; signal?: AbortSignal;
|
||||
db: Db; storage: StorageService; companyId: string; scopeDigest: string; manifestSha256: string; backupId?: string;
|
||||
signal?: AbortSignal; downloadIdleTimeoutMs?: number;
|
||||
}): Promise<boolean> {
|
||||
const [row] = await input.db.select().from(nativeHarnessBackups).where(and(scope(input),
|
||||
eq(nativeHarnessBackups.manifestSha256, input.manifestSha256), input.backupId ? eq(nativeHarnessBackups.id, input.backupId) : undefined,
|
||||
inArray(nativeHarnessBackups.status, ["current", "superseded"]))).limit(1);
|
||||
if (!row) return false;
|
||||
await receive(input.storage, reference(row), undefined, input.signal);
|
||||
await withDownloadProgressDeadline(input, (storage, signal) => receive(storage, reference(row), undefined, signal));
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
@@ -13,7 +13,8 @@ import { readProviderFailure } from "@paperclipai/shared";
|
||||
import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js";
|
||||
import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js";
|
||||
import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js";
|
||||
import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS, nativeHarnessBackupOwner, persistNativeHarnessBackup,
|
||||
restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js";
|
||||
import { restoreNativeProviderArtifacts, retainNativeProviderArtifacts, retainExistingNativeProviderArtifacts,
|
||||
type NativeProviderArtifactReceipt } from "./native-provider-artifacts.js";
|
||||
import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./remote-opencode-artifacts.js";
|
||||
@@ -7275,9 +7276,16 @@ function harnessBackupRoot(root: string): string {
|
||||
async function restoreDurableHarnessForExecution(db: Db, execution: NativeExecutionInput, root: string) {
|
||||
const storage = durableNativeHarnessStorage();
|
||||
if (!storage) return;
|
||||
// Stop or lease loss during startup aborts this source, including before this
|
||||
// restore begins. Only a completed startup's signal is not reused.
|
||||
const startupSignal = nativeSessionStartups.get(execution.binding.runId)?.providerPackSourceAbort.signal;
|
||||
const completed = startupSignal?.aborted && startupSignal.reason instanceof Error
|
||||
&& startupSignal.reason.message === "native_execution_startup_completed";
|
||||
const stop = completed ? undefined : startupSignal;
|
||||
await restoreDurableNativeHarnessBackup({ db, storage, companyId: execution.binding.companyId,
|
||||
scopeDigest: createHash("sha256").update(nativeSessionScopeKey(execution)).digest("hex"),
|
||||
backupRoot: harnessBackupRoot(root), signal: AbortSignal.timeout(300_000) });
|
||||
backupRoot: harnessBackupRoot(root), signal: stop,
|
||||
downloadIdleTimeoutMs: NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS });
|
||||
}
|
||||
|
||||
function harnessBackupCandidates(root: string): string[] {
|
||||
|
||||
Reference in new issue
Block a user