From 31266d7159304c3497b85ac5f3971b2ede2a7201 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:34:36 -0600 Subject: [PATCH] fix(logs): drain existing admissions before finalization (#19055) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Run logs retain execution output for operators. > - Object storage uses bounded admission and asynchronous uploads. > - A diagnostic can wait for capacity when its native execution ends. > - Finalization closed the writer before that existing admission could finish. > - This pull request joins those admissions before it closes the log. > - The benefit is complete accepted output with the existing storage bounds. ## Linked Issues or Issue Description Refs #18944 and #18327. **What happened?** A runner close diagnostic began object-log admission before its recorder retired. The call waited at the 8 MiB pending-byte bound. Completion then closed the writer. The waiting call failed with `Object run log writer is closed`. A valid final log hash covered the stored prefix, but the diagnostic was absent. **Expected behavior** Finalization must retain calls that started before closure. It must reject new calls after closure. Cancellation and storage failures must still settle without a success claim. **Steps to reproduce** 1. Use the actual recorder and segmented object store with a controlled conditional upload. 2. Fill the pending-byte bound. Start native warm-session retirement while the diagnostic waits for capacity. 3. Complete the run through the real completion handler. Release the upload. 4. Check the diagnostic bytes, final hash, source coverage, and warning rows. ## What Changed - Track active admission promises in each object-log writer. - Reject new producers when finalization starts. Join existing bounded admissions, then flush their tail before manifest closure. - Add actual completion coverage and store controls for failure, Stop, global capacity, and independent writers. - Clarify the finalization contract in the storage documentation. Keep feature maps unchanged for this routine fix. ## Verification - Before the fix, the actual completion probe lost a pre-retirement close diagnostic and emitted `object_log_persistence_failed`. Useful canonical output, source coverage, and the final prefix hash remained valid. - After the fix, the exact probe retains the diagnostic and checks the stored bytes and full hash. Ordinary completion and stalled local append controls also pass. - Focused store controls pass, including shared 64 MiB capacity, new-producer rejection, storage failure, Stop, and independent writer progress. - Production and test source at `bc03f8d53a721f01f80a12105a69d8583865e651`: 3 actual completion controls and 53 store, recorder, journal, and diagnostic tests pass. Run with `node node_modules/vitest/vitest.mjs run` from `server/`; select the four safety files plus the completion file with `-t "retired recorder"` in a separate invocation. - Initial treatment repository typechecks and build pass. Current map and module checks pass. A broader four-suite run had 99 passing assertions but failed its existing 30-second completion teardown hook. A later isolated completion run passed all 61 cases at the same limits. The original teardown cause remains unresolved; no timeout was changed. - Repository typechecks and build pass at that source commit. Current head `6652adc7f3379ae250f7b726dad6c58c8482410f` removes only the incidental map paragraph; production and test source are identical. All current-head checks are terminal: 54 successful check runs, one successful legacy status, and three skipped checks. Required `ci / verify` and `ci / e2e` pass. Greptile reviewed exact current head at 5/5. Both review threads are resolved. GitHub reports the PR clean and mergeable. Full repository tests stopped in chat shard 2: a GitHub recovery fair-batch test expected no provider requests but observed `GET /app/hook/config`. Its exact unchanged case passes in isolation. The original cause remains unresolved; this full-suite run is red. First-head remote server lane 2 failed a native-reassessment backlog test: its quiet check observed two active work items instead of zero. The exact unchanged case passes locally. That original CI cause remains unresolved; no job retry or timeout change was made. Other first-head CI lanes passed. The first native review rated the earlier source head 4/5 for the same removed map note. That is not a current-head assessment. Current GitHub rules require zero approving reviews and only the two CI checks above; a fresh native advisory re-review is optional and remains unavailable while its earlier run has an unresolved workspace/accounting obligation. That obligation is tracked separately and is not claimed recovered. - Removed the incidental map paragraph identified by Greptile and the native review. The follow-up contains no runtime or test changes. - This proves the local production-path mechanism. It does not attribute any specific hosted warning to this cause. Hosted replay and served revision checks remain separate acceptance work. ## Risks Finalization now waits for existing bounded capacity admissions. Storage cancellation and the existing operation deadlines still apply. Tests check that failed uploads and Stop settle, and that a separate writer can make progress. No capacity limit, timeout, or execution authority changes. ## Model Used OpenAI `gpt-6.1-sol`, through Codex, with reasoning and tool execution. Context window size is not exposed by the host. ## 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 (56 focused checks; the full-suite red remains documented above) - [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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/run-evidence-storage.md | 5 +- .../heartbeat/__tests__/object-log-fixture.ts | 3 +- .../services/heartbeat/run-completion.test.ts | 96 ++++++++++++++++--- .../src/services/object-run-log-store.test.ts | 87 +++++++++++++++++ server/src/services/object-run-log-store.ts | 12 ++- 5 files changed, 186 insertions(+), 17 deletions(-) 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)