diff --git a/doc/run-evidence-storage.md b/doc/run-evidence-storage.md index ff2662df10c..b74e10e852a 100644 --- a/doc/run-evidence-storage.md +++ b/doc/run-evidence-storage.md @@ -89,8 +89,9 @@ the bounded waiting allowance is rejected without poisoning accepted batches. Transient network, 409, 429 and server errors receive bounded retries. A 412 or a changed head fences the stale writer; a 409 alone does not prove ownership loss or conditional-write support. Live log messages and progress are published -after storage confirmation. Finalization drains admitted writes before closing -and hashing the stream. Logging failure reports incomplete persistence and +after storage confirmation. Finalization rejects new enqueue calls and drains +already-started bounded admissions, including calls waiting for capacity, before +closing and hashing the stream. Logging failure reports incomplete persistence and leaves the final size/digest unknown rather than claiming a complete snapshot. The original progress projection remains coarsely throttled; concurrent batch completions share a single progress write. diff --git a/server/src/services/heartbeat/__tests__/object-log-fixture.ts b/server/src/services/heartbeat/__tests__/object-log-fixture.ts index 4980bee8605..d7febf1a62e 100644 --- a/server/src/services/heartbeat/__tests__/object-log-fixture.ts +++ b/server/src/services/heartbeat/__tests__/object-log-fixture.ts @@ -3,13 +3,14 @@ import { Readable } from "node:stream"; import type { StorageProvider } from "../../../storage/types.js"; import { createObjectRunLogStore } from "../../object-run-log-store.js"; -export function objectLogFixture() { +export function objectLogFixture(beforeConditionalPut?: (input: Parameters>[0]) => Promise) { const objects = new Map(); const digest = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex"); const provider: StorageProvider = { id: "s3", async putObject() { throw new Error("unexpected unconditional write"); }, async putObjectConditional(input) { + await beforeConditionalPut?.(input); const old = objects.get(input.objectKey); if (input.ifNoneMatch ? old : !old || digest(old) !== input.ifMatch) throw Object.assign(new Error("conflict"), { name: "PreconditionFailed" }); diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 7c210a1d97b..046d190f90f 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -44,6 +44,7 @@ vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); const diagnosticNativeFixture = vi.hoisted(() => ({ enabled: false, retireWarm: false, + beforeWarmClose: undefined as (() => Promise) | undefined, backendOptions: undefined as Parameters[1], })); vi.mock("../native-runtime/native-codex-runner.js", async importOriginal => { @@ -63,6 +64,7 @@ vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => { const transport = diagnosticNativeFixture.backendOptions!.codexTransportFactory!(); const session = { close: async ({ reason }: { reason: string }) => { await transport.close(reason); } }; await options.onSession?.(session as never); + await diagnosticNativeFixture.beforeWarmClose?.(); if (diagnosticNativeFixture.retireWarm) await closeIdleWarmNativeSessionsForRestart(); return { result: { summary: "accepted result" }, terminal: { runTerminalState: "succeeded" }, turnId: "test-turn", normalizedSessionId: options.input.session.normalizedSessionId, providerSessionId: "synthetic-provider", @@ -173,12 +175,52 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( } const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); - it.each([false, true])("finishes accepted output before ordinary lease teardown without writing to a retired recorder (stalled append=%s)", async stalledAppend => { + it.each(["ordinary", "stalled append", "object admission"] as const)("finishes accepted output before ordinary lease teardown without writing to a retired recorder (%s)", async mode => { + const stalledAppend = mode === "stalled append", stalledAdmission = mode === "object admission"; const f = await fixture({ runtimeMode: "native", resultJson: { durable: "accepted result" } }); const workspace = await realpath(await mkdtemp(join(tmpdir(), "diagnostic-completion-"))); onTestFinished(() => rm(workspace, { recursive: true, force: true })); - const store = stalledAppend ? createDurableRunLogStore({ basePath: join(workspace, "run-logs") }) : objectLogFixture(); + let holdUploads = false, uploadEntered = false, finalizeEntered = false; + let releaseUpload!: () => void; + const uploadGate = new Promise(resolve => { releaseUpload = resolve; }); + const store = stalledAppend ? createDurableRunLogStore({ basePath: join(workspace, "run-logs") }) : objectLogFixture(async input => { + if (stalledAdmission && holdUploads && input.objectKey.includes("/objects/")) { + uploadEntered = true; + await uploadGate; + } + }); + let latestOutputSeq = 0, admittedBytes = 0; + const diagnosticAdmissions: Array<{ chunk: string; admitted?: boolean; error?: string }> = []; + if (store.enqueue) { + const originalEnqueue = store.enqueue.bind(store); + vi.spyOn(store, "enqueue").mockImplementation(async (handle, event) => { + latestOutputSeq = event.seq ?? latestOutputSeq; + const observation = event.chunk.includes("runnerd diagnostic") ? { chunk: event.chunk } as typeof diagnosticAdmissions[number] : undefined; + if (observation) diagnosticAdmissions.push(observation); + let receipt; + try { receipt = await originalEnqueue(handle, event); if (observation) observation.admitted = true; } + catch (error) { if (observation) observation.error = error instanceof Error ? error.message : String(error); throw error; } + if (holdUploads) admittedBytes += Buffer.byteLength(JSON.stringify({ ts: event.ts, + attemptId: handle.attemptId, stream: event.stream, chunk: event.chunk, seq: event.seq }) + "\n"); + return receipt; + }); + } + const originalFinalize = store.finalize.bind(store); + vi.spyOn(store, "finalize").mockImplementation((...args) => { + finalizeEntered = true; + return originalFinalize(...args); + }); + const readLog = async () => { + let content = "", offset = 0; + do { + const page = await store.read(output.handle!, { offset, limitBytes: 4 * 1024 * 1024 }); + content += page.content; + if (page.nextOffset === undefined) break; + offset = page.nextOffset; + } while (true); + return { content }; + }; let releaseAppend!: () => void, observeAppend!: () => void; const appendGate = new Promise(resolve => { releaseAppend = resolve; }); const appendStarted = new Promise(resolve => { observeAppend = resolve; }); @@ -207,17 +249,23 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( await new Promise((resolve, reject) => { api.once("error", reject); api.listen(0, "127.0.0.1", resolve); }); onTestFinished(() => new Promise(resolve => api.close(() => resolve()))); diagnosticNativeFixture.enabled = true; - diagnosticNativeFixture.retireWarm = stalledAppend; + diagnosticNativeFixture.retireWarm = stalledAppend || stalledAdmission; + if (stalledAdmission) diagnosticNativeFixture.beforeWarmClose = async () => { + await store.flush!(output.handle!); + holdUploads = true; + const target = 8 * 1024 * 1024 - 1; + while (admittedBytes < target) { + const overhead = Buffer.byteLength(JSON.stringify({ ts: new Date().toISOString(), + attemptId: output.handle!.attemptId, stream: "stdout", chunk: "", seq: latestOutputSeq + 1 }) + "\n"); + const chunk = "x".repeat(Math.min(64 * 1024, target - admittedBytes - overhead)); + await handlers.onLog("stdout", chunk); + } + expect(admittedBytes).toBe(target); + expect(uploadEntered).toBe(true); + }; let completed = false; const diagnosticWrites: Promise[] = []; const finishing = (async () => { - const adapter = await executePaperclipNativeSession({ db, execution: execution as never, - runnerInstanceId: "synthetic-controller", useRunnerd: true, onLog: (stream, chunk) => { - const write = handlers.onLog(stream, chunk); - if (chunk.includes("runnerd diagnostic")) diagnosticWrites.push(write); - return write; - }, - runnerEnvironment: { HOME: workspace, CODEX_HOME: join(workspace, "synthetic-codex-home") } }); const event: PrpEvent = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceInstanceId: randomUUID(), sourceEventId: randomUUID(), sourceSeq: 1, sourceKind: "runner", runId: f.run.id, normalizedSessionId: execution.session.normalizedSessionId, @@ -227,6 +275,13 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( nativeSource: { sourceInstanceId: event.sourceInstanceId, sourceEventId: event.sourceEventId, sourceSeq: event.sourceSeq, protocolSchemaVersion: 1, canonicalPayload: event as unknown as Record } }); if (!stalledAppend) await handlers.onPersistedNativeLog!(persisted.row.seq, formatNativeRunLogEvent(event)); + const adapter = await executePaperclipNativeSession({ db, execution: execution as never, + runnerInstanceId: "synthetic-controller", useRunnerd: true, onLog: (stream, chunk) => { + const write = handlers.onLog(stream, chunk); + if (chunk.includes("runnerd diagnostic")) diagnosticWrites.push(write); + return write; + }, + runnerEnvironment: { HOME: workspace, CODEX_HOME: join(workspace, "synthetic-codex-home") } }); await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "committed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); Object.assign(f.run, await read(f.run.id)); f.agent.adapterType = "paperclip_runner"; @@ -241,12 +296,17 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( })(); void finishing.catch(() => undefined); try { + if (stalledAdmission) { + await vi.waitFor(() => expect(finalizeEntered).toBe(true), { timeout: 10_000 }); + expect(diagnosticWrites.length).toBeGreaterThan(0); + releaseUpload(); + } if (stalledAppend) { await appendStarted; await vi.waitFor(() => expect(completed).toBe(true), { timeout: 10_000 }); } const { persisted, event } = await finishing; - const before = await store.read(output.handle!, { limitBytes: 65536 }); + const before = await readLog(); expect(before.content).toContain("accepted output for this exact run"); const settled = await read(f.run.id); expect(settled).toMatchObject({ companyId: f.run.companyId, status: "succeeded", resultJson: { durable: "accepted result" } }); @@ -270,23 +330,35 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( await createHeartbeatRunControl(db, deps).releaseEnvironmentLeasesForRun({ runId: f.run.id, companyId: f.run.companyId, agentId: f.run.agentId, status: "succeeded", providerResourceDisposition: "destroy" }); expect(deps.options.closeWarmNativeSessionsForRun).toHaveBeenCalledOnce(); - const after = await store.read(output.handle!, { limitBytes: 65536 }); + const after = await readLog(); expect(after.content).toBe(before.content); if (!stalledAppend) { const coverage = createNativeRunLogCoverageVerifier(db, f.run.companyId, f.run.id, { redactChunks: chunk => [chunk] }); await coverage.write(Buffer.from(after.content)); expect(await coverage.finish()).toBe(persisted.row.seq); } + if (stalledAdmission) { + await Promise.allSettled(diagnosticWrites); + expect(diagnosticAdmissions.length).toBeGreaterThan(0); + const storedChunks = before.content.trimEnd().split("\n").map(line => JSON.parse(line).chunk); + for (const admission of diagnosticAdmissions) { + expect(admission).toMatchObject({ admitted: true }); + expect(admission.error).toBeUndefined(); + expect(storedChunks).toContain(admission.chunk); + } + } const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); expect(events.filter(event => (event.payload as { reason?: string } | null)?.reason === "object_log_persistence_failed")).toEqual([]); } finally { releaseAppend(); + releaseUpload(); await Promise.allSettled(diagnosticWrites); await finishing; if (stalledAppend) expect(await read(f.run.id)).toMatchObject({ logBytes: null, logSha256: null }); await closeIdleWarmNativeSessionsForRestart(); diagnosticNativeFixture.enabled = false; diagnosticNativeFixture.retireWarm = false; + diagnosticNativeFixture.beforeWarmClose = undefined; } }); diff --git a/server/src/services/object-run-log-store.test.ts b/server/src/services/object-run-log-store.test.ts index dc315aae15a..5467801b77a 100644 --- a/server/src/services/object-run-log-store.test.ts +++ b/server/src/services/object-run-log-store.test.ts @@ -318,6 +318,93 @@ describe("S3 run log store", () => { const [size, result] = await Promise.all([pending, finalizing]); expect(result.bytes).toBe(size); }); + it.each(["commit", "storage failure", "Stop"] as const)("settles pre-close bounded admission through %s and fences new producers", async mode => { + const m = memory(), original = m.provider.putObjectConditional!; + let release!: () => void, entered = false; + const gate = new Promise(resolve => { release = resolve; }); + const cancellation = new AbortController(); + m.provider.putObjectConditional = async value => { + if (value.objectKey.includes("/objects/") && !entered) { + entered = true; + await gate; + value.signal?.throwIfAborted(); + if (mode === "storage failure") throw new Error("controlled upload failure"); + } + return original(value); + }; + const store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined }); + const handle = await store.begin({ ...input, signal: cancellation.signal }); + const first = await store.enqueue!(handle, { ...event(1), chunk: "a".repeat(7 * 1024 * 1024) }); + const committed = first.committed.catch(error => error); + await vi.waitFor(() => expect(entered).toBe(true)); + let admitted = false; + const waiting = store.enqueue!(handle, { ...event(2), chunk: "b".repeat(2 * 1024 * 1024) }) + .then(receipt => { admitted = true; return receipt.committed; }).catch(error => error); + let finalized = false; + const finalizing = store.finalize(handle).then(value => { finalized = true; return value; }); + try { + await expect(store.enqueue!(handle, event(3))).rejects.toThrow("closed"); + const peer = await store.begin({ ...input, runId: "peer" }); + await store.append(peer, { ...event(1), chunk: "peer output" }); + expect((await store.finalize(peer)).bytes).toBeGreaterThan(0); + expect(admitted).toBe(false); + expect(finalized).toBe(false); + if (mode === "Stop") cancellation.abort(new Error("controlled Stop")); + } finally { release(); } + const [firstResult, secondResult, summary] = await Promise.all([committed, waiting, finalizing]); + if (mode === "commit") { + expect(admitted).toBe(true); + expect(summary.bytes).toBe(Number(firstResult) + Number(secondResult)); + const hash = createHash("sha256"); + let offset = 0, chunks = 0; + for (;;) { + const page = await store.read(handle, { offset, limitBytes: 1024 * 1024 }); + hash.update(page.content); + chunks += (page.content.match(/"chunk":/g) ?? []).length; + if (page.nextOffset === undefined) break; + offset = page.nextOffset; + } + expect(chunks).toBe(2); + expect(summary.sha256).toBe(hash.digest("hex")); + } else { + expect(firstResult).toBeInstanceOf(Error); + expect(secondResult).toBeInstanceOf(Error); + expect(admitted).toBe(false); + expect(summary).toMatchObject({ bytes: null, sha256: null }); + } + }); + it("drains an admission waiting on global capacity while other writers flush", async () => { + const m = memory(), original = m.provider.putObjectConditional!; + let release!: () => void, uploads = 0; + const gate = new Promise(resolve => { release = resolve; }); + m.provider.putObjectConditional = async value => { + if (value.objectKey.includes("/objects/")) { uploads++; await gate; } + return original(value); + }; + const store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined }); + const handles = [], committed = []; + for (let index = 0; index < 9; index++) { + const handle = await store.begin({ ...input, runId: `full-${index}` }); + handles.push(handle); + committed.push((await store.enqueue!(handle, { ...event(1), chunk: "x".repeat(7 * 1024 * 1024) })).committed); + } + await vi.waitFor(() => expect(uploads).toBe(4)); + const target = await store.begin({ ...input, runId: "waiting" }); + let admitted = false, finalized = false; + const waiting = store.enqueue!(target, { ...event(1), chunk: "y".repeat(2 * 1024 * 1024) }) + .then(receipt => { admitted = true; return receipt.committed; }); + const finalizing = store.finalize(target).then(value => { finalized = true; return value; }); + try { + await expect(store.enqueue!(target, event(2))).rejects.toThrow("closed"); + expect(admitted).toBe(false); + expect(finalized).toBe(false); + } finally { release(); } + const [bytes, summary] = await Promise.all([waiting, finalizing]); + expect(admitted).toBe(true); + expect(summary.bytes).toBe(bytes); + await Promise.all(committed); + await Promise.all(handles.map(handle => store.finalize(handle))); + }); it("does not acknowledge a batch when storage fails", async () => { const m = memory(), store = createObjectRunLogStore({ diff --git a/server/src/services/object-run-log-store.ts b/server/src/services/object-run-log-store.ts index 1c9efae921a..2598d1d3921 100644 --- a/server/src/services/object-run-log-store.ts +++ b/server/src/services/object-run-log-store.ts @@ -31,6 +31,7 @@ interface WriterState { timer?: NodeJS.Timeout; flushing?: Promise; closed: boolean; + admitting: Set>; failure?: unknown; signal?: AbortSignal; sourcePrefixVerified: boolean; @@ -167,6 +168,12 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL async function enqueue(handle: RunLogHandle, event: Parameters[1]) { const state = writers.get(handle); if (!state || state.closed) throw new Error("Object run log writer is closed"); + const admission = admit(state, handle, event); + state.admitting.add(admission); + try { return await admission; } + finally { state.admitting.delete(admission); } + } + async function admit(state: WriterState, handle: RunLogHandle, event: Parameters[1]) { state.signal?.throwIfAborted(); if (state.failure) throw state.failure; const bytes = Buffer.from( @@ -186,7 +193,6 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL state.pendingBytes + bytes.length > MAX_PENDING_BYTES || totalPendingBytes + bytes.length > 64 * MAX_SEGMENT_BYTES ) { - if (state.closed) throw new Error("Object run log writer is closed"); if (state.failure) throw state.failure; if (!waiting) { if (admissions >= 32 || admissionBytes + bytes.length > MAX_PENDING_BYTES) @@ -210,7 +216,6 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL capacityWaiters.add(wake); }); } - if (state.closed) throw new Error("Object run log writer is closed"); state.signal?.throwIfAborted(); if (state.failure) throw state.failure; if (event.preservesSourceCoverage !== true || event.sourceCursor !== undefined && @@ -267,6 +272,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL pending: [], pendingBytes: 0, closed: false, + admitting: new Set(), signal: input.signal, sourcePrefixVerified, coverageEligible: true, @@ -322,6 +328,8 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL state.closed = true; capacityChanged(); try { + await flush(state); + await Promise.allSettled([...state.admitting]); await flush(state); state.writer = await state.journal.close(state.writer, state.signal); if (finalizeOptions?.computeDigest === false)