From e482f0f63273f773bbd7ba1a084d2d1754fb14b4 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 6 Nov 2026 13:47:25 -0800 Subject: [PATCH] fix(server): shed only retryable object-journal reads under load (#19176) Client polling can exhaust the shared object journal queue and reject log persistence. Bound ordinary object-log and trace reads, keep persistence and internal verification queued, and return retryable 503 responses when reads are busy. Leave feedback exports unsent until every log page is available. Validation: 72 latest-master affected tests and 10 workflow tests pass locally, with the workflow fixture passing strict TypeScript checks. The preceding broader run passed 280 server tests with two opt-in S3 skips. All applicable checks on d0a937768da9be297485323c5b905eb63371ec15 are green (56 success, three skipped); exact-head Apex review is 5/5 with no unresolved findings. Background persistence still has no fixed queue bound during a prolonged storage outage. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Paperclip --- .../__tests__/agent-live-run-routes.test.ts | 22 +++++ server/src/__tests__/error-handler.test.ts | 18 +++++ server/src/__tests__/feedback-service.test.ts | 35 ++++++++ server/src/middleware/error-handler.ts | 11 +++ server/src/routes/agents.ts | 6 +- server/src/services/feedback.ts | 6 +- .../native-runtime/native-run-log-export.ts | 2 +- .../native-run-log-recovery.test.ts | 20 +++++ .../native-runtime/native-run-log-recovery.ts | 2 +- .../src/services/object-byte-journal.test.ts | 62 +++++++++++++- server/src/services/object-byte-journal.ts | 65 ++++++++++++--- .../src/services/object-run-log-store.test.ts | 37 +++++++++ server/src/services/object-run-log-store.ts | 7 +- .../provider-trace-object-store.test.ts | 81 ++++++++++++++++++- .../services/provider-trace-object-store.ts | 20 +++-- server/src/services/provider-trace-store.ts | 16 ++-- server/src/services/run-log-store.ts | 2 + tests/runner-e2e/workflow-target.test.ts | 3 +- 18 files changed, 382 insertions(+), 33 deletions(-) create mode 100644 server/src/services/native-runtime/native-run-log-recovery.test.ts diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 67daeba44c8..f902d888c03 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -1940,6 +1940,28 @@ describe("agent live run routes", () => { ); }); + it("returns a retryable busy response for workspace-diff repair and permits a later retry", async () => { + const app = await createApp(); + const { ObjectJournalBusyError } = await import("../services/object-byte-journal.js"); + mockProviderTraceStore.getByRun.mockResolvedValue({ + id: "trace-1", status: "complete", deletedAt: null, + expiresAt: new Date("2099-01-01T00:00:00.000Z"), + }); + mockProviderTraceStore.readExactEntries.mockRejectedValueOnce(new ObjectJournalBusyError()); + const route = "/api/heartbeat-runs/aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa/provider-trace/reproject-workspace-diffs"; + const busy = await requestApp(app, baseUrl => request(baseUrl).post(route)); + expect(busy.status).toBe(503); + expect(busy.headers["retry-after"]).toBe("2"); + expect(busy.body.code).toBe("object_journal_io_busy"); + expect(mockWorkspaceDiffReprojection.project).not.toHaveBeenCalled(); + expect(mockWorkspaceDiffReprojection.persist).not.toHaveBeenCalled(); + expect(mockLogActivity).not.toHaveBeenCalled(); + + const retried = await requestApp(app, baseUrl => request(baseUrl).post(route)); + expect(retried.status).toBe(200); + expect(mockWorkspaceDiffReprojection.persist).toHaveBeenCalledOnce(); + }); + it.each([ ["unavailable", null, "trace_unavailable"], [ diff --git a/server/src/__tests__/error-handler.test.ts b/server/src/__tests__/error-handler.test.ts index a2d36a49765..e50866249ca 100644 --- a/server/src/__tests__/error-handler.test.ts +++ b/server/src/__tests__/error-handler.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { HttpError } from "../errors.js"; import { errorHandler } from "../middleware/error-handler.js"; import { mcpDiscoveryHttpFailure, retainMcpConnectionFailure } from "../services/mcp-connection-failure.js"; +import { ObjectJournalBusyError } from "../services/object-byte-journal.js"; const recordResponsibleUserDenialOnActiveRunMock = vi.hoisted(() => vi.fn()); const captureExceptionMock = vi.hoisted(() => vi.fn()); @@ -115,6 +116,23 @@ describe("errorHandler", () => { expect(telemetryMocks.trackErrorHandlerCrash).not.toHaveBeenCalled(); }); + it("answers a shed object-storage read with a retryable 503 without reporting it", () => { + const res = { ...makeRes(), set: vi.fn() } as any; + const next = vi.fn() as unknown as NextFunction; + + errorHandler(new ObjectJournalBusyError(), makeReq(), res, next); + + expect(res.set).toHaveBeenCalledWith("Retry-After", "2"); + expect(res.status).toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith({ + error: "Object storage is busy; retry shortly", + code: "object_journal_io_busy", + }); + expect(res.err).toBeUndefined(); + expect(captureExceptionMock).not.toHaveBeenCalled(); + expect(telemetryMocks.trackErrorHandlerCrash).not.toHaveBeenCalled(); + }); + it("only ends the response when headers were already sent during database shutdown", () => { const req = { ...makeReq(), app: { locals: { paperclipDatabaseClosing: true } } } as unknown as Request; const res = { ...makeRes(), end: vi.fn(), headersSent: true } as any; diff --git a/server/src/__tests__/feedback-service.test.ts b/server/src/__tests__/feedback-service.test.ts index 08a44442038..dae4d117124 100644 --- a/server/src/__tests__/feedback-service.test.ts +++ b/server/src/__tests__/feedback-service.test.ts @@ -24,6 +24,8 @@ import { issues, } from "@paperclipai/db"; import { feedbackService } from "../services/feedback.ts"; +import { getRunLogStore } from "../services/run-log-store.js"; +import { ObjectJournalBusyError } from "../services/object-byte-journal.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -1056,6 +1058,39 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => { }); }); + it.each([0, 1])("retries an unsent feedback export when log page %s is busy", async busyPage => { + const { companyId, issueId, targetCommentId, runId } = await seedIssueWithRichAgentComment(); + await db.update(heartbeatRuns).set({ logStore: "local_file", logRef: "fixture/run.ndjson" }).where(eq(heartbeatRuns.id, runId)); + const uploadTraceBundle = vi.fn().mockResolvedValue({ objectKey: "feedback-traces/recovered.json" }); + const flushingSvc = feedbackService(db, { shareClient: { uploadTraceBundle } }); + await flushingSvc.saveIssueVote({ issueId, targetType: "issue_comment", targetId: targetCommentId, vote: "up", authorUserId: "user-1", allowSharing: true }); + const first = JSON.stringify({ stream: "stdout", chunk: "first log record" }) + "\n"; + const second = JSON.stringify({ stream: "stdout", chunk: "second log record" }) + "\n"; + const read = vi.spyOn(getRunLogStore(), "read"); + if (busyPage) read.mockResolvedValueOnce({ content: first, nextOffset: Buffer.byteLength(first) }); + read.mockRejectedValueOnce(new ObjectJournalBusyError()); + try { + expect(await flushingSvc.flushPendingFeedbackTraces()).toMatchObject({ attempted: 1, sent: 0, failed: 1 }); + expect(uploadTraceBundle).not.toHaveBeenCalled(); + const [pending] = await flushingSvc.listFeedbackTraces({ companyId, issueId }); + expect(pending?.status).toBe("failed"); + expect(pending?.exportedAt).toBeNull(); + expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(true); + read.mockReset().mockResolvedValue({ content: first + second }); + expect(await flushingSvc.flushPendingFeedbackTraces()).toMatchObject({ attempted: 1, sent: 1, failed: 0 }); + expect(uploadTraceBundle).toHaveBeenCalledOnce(); + const bundle = JSON.stringify(uploadTraceBundle.mock.calls[0]?.[0]); + expect(bundle).toContain("first log record"); + expect(bundle).toContain("second log record"); + const [sent] = await flushingSvc.listFeedbackTraces({ companyId, issueId }); + expect(sent?.status).toBe("sent"); + expect(sent?.attemptCount).toBe(2); + expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(false); + } finally { + read.mockRestore(); + } + }); + it("can flush a single shared trace immediately by trace id", async () => { const { companyId, issueId, commentId: firstCommentId } = await seedIssueWithAgentComment(); const secondCommentId = randomUUID(); diff --git a/server/src/middleware/error-handler.ts b/server/src/middleware/error-handler.ts index 7907f60c733..438a27acda6 100644 --- a/server/src/middleware/error-handler.ts +++ b/server/src/middleware/error-handler.ts @@ -18,6 +18,7 @@ import { redactSensitiveValueOccurrences, } from "./redact-sensitive.js"; import { recordResponsibleUserDenialOnActiveRun } from "../services/responsible-user-denial-run-outcomes.js"; +import { isObjectJournalBusyError } from "../services/object-byte-journal.js"; import { isExpectedMcpConnectionFailure } from "../services/mcp-connection-failure.js"; import { isEndedDbPoolError } from "../database-retry.js"; import { ToolGatewayHttpError } from "../services/tool-gateway.js"; @@ -323,6 +324,16 @@ export function errorHandler( return; } + if (isObjectJournalBusyError(rootError)) { + if (!res.headersSent) { + res.set("Retry-After", "2"); + res.status(503).json({ error: rootError.message, code: rootError.code }); + } else { + res.end(); + } + return; + } + const reportableError = sanitizeSecretSensitiveError(req, rootError); attachErrorContext( req, diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index ad0d748e57a..ccdb43ab484 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -1,4 +1,5 @@ import { assertProviderApiPlacementPolicy, managedCursorApiDefaultEnabled, resolveProviderApiPlacement } from "../services/native-runtime/provider-api-placement.js"; +import { isObjectJournalBusyError } from "../services/object-byte-journal.js"; import { readTaskPanelActivations } from "../services/plugin-task-panels.js"; import { enqueueNativeRuntimeResponse } from "../services/native-runtime/native-runtime-response-delivery.js"; import { latestFailedRunsByAgent } from "@paperclipai/shared"; @@ -7547,7 +7548,10 @@ export function agentRoutes( const entries = await providerTraces .readExactEntries(run.id, run.companyId) - .catch(() => null); + .catch(error => { + if (isObjectJournalBusyError(error)) throw error; + return null; + }); if (entries === null) { res.json({ created: 0, diff --git a/server/src/services/feedback.ts b/server/src/services/feedback.ts index e3e4d70b766..5a202a84860 100644 --- a/server/src/services/feedback.ts +++ b/server/src/services/feedback.ts @@ -48,6 +48,7 @@ import { sha256Digest, } from "./feedback-redaction.js"; import { getRunLogStore, isRunLogStoreType } from "./run-log-store.js"; +import { isObjectJournalBusyError } from "./object-byte-journal.js"; import { recoverSettledNativeRunLog } from "./native-runtime/native-run-log-recovery.js"; import { getOperatorSettingDefaults } from "./setting-defaults.js"; @@ -333,7 +334,10 @@ async function readFullRunLog(db: Db, run: { const result = await store.read({ store: run.logStore, logRef: run.logRef }, { offset, limitBytes: 512_000, - }).catch(() => null); + }).catch((error) => { + if (isObjectJournalBusyError(error)) throw error; + return null; + }); if (!result) return combined || null; combined += result.content; if (result.nextOffset == null) break; diff --git a/server/src/services/native-runtime/native-run-log-export.ts b/server/src/services/native-runtime/native-run-log-export.ts index ec588fc4153..b0457764fa5 100644 --- a/server/src/services/native-runtime/native-run-log-export.ts +++ b/server/src/services/native-runtime/native-run-log-export.ts @@ -309,7 +309,7 @@ export async function nativeRunLogPrefixCursor(input: { while (offset < bytes) { input.signal?.throwIfAborted(); const page = await input.store.read(input.handle, { - offset, limitBytes: Math.min(256_000, bytes - offset), signal: input.signal, + offset, limitBytes: Math.min(256_000, bytes - offset), signal: input.signal, priority: "background", }); const content = Buffer.from(page.content); if (!content.length || offset + content.length > bytes) throw unproven(); diff --git a/server/src/services/native-runtime/native-run-log-recovery.test.ts b/server/src/services/native-runtime/native-run-log-recovery.test.ts new file mode 100644 index 00000000000..e484be0ca33 --- /dev/null +++ b/server/src/services/native-runtime/native-run-log-recovery.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import type { RunLogStore } from "../run-log-store.js"; +import { readRecoverableRunLog } from "./native-run-log-recovery.js"; + +describe("recoverable run log reads", () => { + it("lets the client log poll be shed under load and keeps the caller's options", async () => { + const read = vi.fn(async () => ({ content: "", nextOffset: 0 })); + const store = { read } as unknown as RunLogStore; + const handle = { store: "local_file" as const, logRef: "logs/run.ndjson" }; + const signal = new AbortController().signal; + await readRecoverableRunLog({} as Db, store, handle, { offset: 12, limitBytes: 64, signal }); + expect(read).toHaveBeenCalledWith(handle, { + priority: "interactive", + offset: 12, + limitBytes: 64, + signal, + }); + }); +}); diff --git a/server/src/services/native-runtime/native-run-log-recovery.ts b/server/src/services/native-runtime/native-run-log-recovery.ts index 388f17ae389..084084d87be 100644 --- a/server/src/services/native-runtime/native-run-log-recovery.ts +++ b/server/src/services/native-runtime/native-run-log-recovery.ts @@ -73,7 +73,7 @@ export async function readRecoverableRunLog( ); }, ); - return store.read(handle, options); + return store.read(handle, { priority: "interactive", ...options }); } export async function recoverSettledNativeRunLog( diff --git a/server/src/services/object-byte-journal.test.ts b/server/src/services/object-byte-journal.test.ts index 3caddc32c08..4412249353c 100644 --- a/server/src/services/object-byte-journal.test.ts +++ b/server/src/services/object-byte-journal.test.ts @@ -2,7 +2,11 @@ import { createHash } from "node:crypto"; import { Readable } from "node:stream"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { StorageProvider } from "../storage/types.js"; -import { createObjectByteJournal, ObjectJournalConflictError } from "./object-byte-journal.js"; +import { + createObjectByteJournal, + ObjectJournalBusyError, + ObjectJournalConflictError, +} from "./object-byte-journal.js"; async function readBody(body: AsyncIterable) { const chunks: Buffer[] = []; @@ -229,6 +233,62 @@ describe("object byte journal", () => { await expect(a.readRange(writer.head, 0, 4)).rejects.toThrow("integrity"); await expect(a.readRange(writer.head, 0, 4, AbortSignal.abort())).rejects.toThrow(); }); + it("sheds only interactive reads under saturated object IO and never a writer", async () => { + const s = storage(), + a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/busy" }); + let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + writer = await a.append(writer, Buffer.from("data")); + const get = s.provider.getObject; + let open!: () => void; + const gate = new Promise((resolve) => { + open = resolve; + }); + s.provider.getObject = async (input) => { + await gate; + return get(input); + }; + const reader = createObjectByteJournal({ + provider: s.provider, + prefix: "c/r/busy", + readPriority: "interactive", + }); + const reads = Array.from({ length: 40 }, () => reader.head()); + await expect(reader.head()).rejects.toBeInstanceOf(ObjectJournalBusyError); + const other = createObjectByteJournal({ provider: s.provider, prefix: "c/r/writer" }); + const claimed = other.claim({ expectedOwner: null, assertAuthority: authority }); + open(); + const otherWriter = await other.append(await claimed, Buffer.from("more")); + expect(otherWriter.head.end).toBe(4); + expect((await Promise.all(reads)).every((snapshot) => snapshot?.head.end === 4)).toBe(true); + }); + it("releases a queued read's place when its caller aborts", async () => { + const s = storage(), + a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/abort" }); + let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + writer = await a.append(writer, Buffer.from("data")); + const get = s.provider.getObject; + let open!: () => void; + const gate = new Promise((resolve) => { + open = resolve; + }); + s.provider.getObject = async (input) => { + await gate; + return get(input); + }; + const reader = createObjectByteJournal({ + provider: s.provider, + prefix: "c/r/abort", + readPriority: "interactive", + }); + const reads = Array.from({ length: 39 }, () => reader.head()); + const controller = new AbortController(); + const cancelled = reader.head(controller.signal); + controller.abort(new Error("client_disconnected")); + await expect(cancelled).rejects.toThrow("client_disconnected"); + const replacement = reader.head(); + open(); + expect((await Promise.all([...reads, replacement])).every((snapshot) => snapshot?.head.end === 4)).toBe(true); + }); it("retains a deletion tombstone and denies a subsequent claim", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts index 3a2ece6f3d9..646b76c9974 100644 --- a/server/src/services/object-byte-journal.ts +++ b/server/src/services/object-byte-journal.ts @@ -204,17 +204,52 @@ async function qualifyConditionalWrites( }); } +export class ObjectJournalBusyError extends Error { + readonly code = "object_journal_io_busy"; + constructor() { + super("Object storage is busy; retry shortly"); + } +} +export function isObjectJournalBusyError(error: unknown): error is ObjectJournalBusyError { + return error instanceof ObjectJournalBusyError; +} + +export type ObjectJournalIOPriority = "background" | "interactive"; +const MAX_ACTIVE_IO = 8; +const MAX_WAITING_INTERACTIVE_IO = 32; let activeIO = 0; -const waitingIO: Array<() => void> = []; -async function objectIO(operation: () => Promise): Promise { - if (activeIO >= 8) { - if (waitingIO.length >= 32) throw new Error("object_journal_io_capacity_exceeded"); - await new Promise((resolve) => waitingIO.push(resolve)); +const waitingIO: Record void>> = { + background: [], + interactive: [], +}; +async function objectIO( + operation: () => Promise, + priority: ObjectJournalIOPriority = "background", + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted(); + if (activeIO >= MAX_ACTIVE_IO) { + const queue = waitingIO[priority]; + if (priority === "interactive" && queue.length >= MAX_WAITING_INTERACTIVE_IO) + throw new ObjectJournalBusyError(); + await new Promise((resolve, reject) => { + const grant = () => { + signal?.removeEventListener("abort", abort); + resolve(); + }; + const abort = () => { + const index = queue.indexOf(grant); + if (index >= 0) queue.splice(index, 1); + reject(signal!.reason); + }; + queue.push(grant); + signal?.addEventListener("abort", abort, { once: true }); + }); } else activeIO++; try { return await operation(); } finally { - const next = waitingIO.shift(); + const next = waitingIO.background.shift() ?? waitingIO.interactive.shift(); if (next) next(); else activeIO--; } @@ -223,6 +258,7 @@ async function objectIO(operation: () => Promise): Promise { export interface ObjectByteJournalOptions { provider: StorageProvider; prefix: string; + readPriority?: ObjectJournalIOPriority; encode?: (bytes: Buffer) => Promise; decode?: (bytes: Buffer) => Promise; @@ -243,9 +279,10 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { throw new Error("Object storage requires conditional writes and scoped listing"); const headKey = `${prefix}/head.json`; const objectPrefix = `${prefix}/objects/`; + const readPriority = options.readPriority ?? "background"; const putConditional = ( input: Parameters>[0], - ) => objectIO(() => provider.putObjectConditional!(input)); + ) => objectIO(() => provider.putObjectConditional!(input), "background", input.signal); async function getBytes(key: string, limit: number, signal?: AbortSignal) { return objectIO(async () => { @@ -271,7 +308,7 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { } signal?.throwIfAborted(); return { bytes: Buffer.concat(chunks), etag: result.etag }; - }); + }, readPriority, signal); } async function head(signal?: AbortSignal): Promise { let result; @@ -630,12 +667,18 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { signal ? AbortSignal.any([signal, AbortSignal.timeout(15_000)]) : AbortSignal.timeout(15_000); - const page = await objectIO(() => - provider.listObjects!({ prefix: objectPrefix, continuationToken, signal: deadline() }), + const page = await objectIO( + () => provider.listObjects!({ prefix: objectPrefix, continuationToken, signal: deadline() }), + "background", + signal, ); for (const key of page.keys) { if (!key.startsWith(objectPrefix)) throw new Error("object_journal_invalid_listing"); - await objectIO(() => provider.deleteObject({ objectKey: key, signal: deadline() })); + await objectIO( + () => provider.deleteObject({ objectKey: key, signal: deadline() }), + "background", + signal, + ); } continuationToken = page.continuationToken; } while (continuationToken); diff --git a/server/src/services/object-run-log-store.test.ts b/server/src/services/object-run-log-store.test.ts index 5467801b77a..6016a931ce2 100644 --- a/server/src/services/object-run-log-store.test.ts +++ b/server/src/services/object-run-log-store.test.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { Readable } from "node:stream"; import { describe, expect, it, vi } from "vitest"; import type { StorageProvider } from "../storage/types.js"; +import { ObjectJournalBusyError } from "./object-byte-journal.js"; import { createObjectRunLogStore } from "./object-run-log-store.js"; async function readBody(body: AsyncIterable) { @@ -482,4 +483,40 @@ describe("S3 run log store", () => { await store.append(h, event(3)); expect((await store.finalize(h)).bytes).toBeGreaterThan(9 * 1024 * 1024); }); + it("bounds ordinary reads while explicit internal reads keep queuing", async () => { + const m = memory(), + store = createObjectRunLogStore({ + provider: m.provider, + keyPrefix: "logs", + assertWriterAuthority: async () => undefined, + }); + const handle = await store.begin(input); + await store.append(handle, event(1)); + await store.finalize(handle); + const get = m.provider.getObject; + let open!: () => void; + const gate = new Promise((resolve) => { + open = resolve; + }); + m.provider.getObject = async (i) => { + await gate; + return get(i); + }; + const polls = Array.from({ length: 40 }, () => store.read(handle)); + const excess = store.read(handle); + let rejected: unknown; + void excess.catch((error) => { rejected = error; }); + let internal: ReturnType | undefined; + try { + await vi.waitFor(() => expect(rejected).toBeInstanceOf(ObjectJournalBusyError)); + internal = store.read(handle, { priority: "background" }); + open(); + for (const poll of await Promise.allSettled(polls)) + if (poll.status === "rejected") expect(poll.reason).toBeInstanceOf(ObjectJournalBusyError); + expect((await internal).content.trim().split("\n")).toHaveLength(1); + } finally { + open(); + await Promise.allSettled([...polls, excess, ...(internal ? [internal] : [])]); + } + }); }); diff --git a/server/src/services/object-run-log-store.ts b/server/src/services/object-run-log-store.ts index 2598d1d3921..b7a9135718a 100644 --- a/server/src/services/object-run-log-store.ts +++ b/server/src/services/object-run-log-store.ts @@ -2,7 +2,7 @@ import { beginIdleTrackedWork } from "./task-admission.js"; import { createHash, randomUUID } from "node:crypto"; import type { StorageProvider } from "../storage/types.js"; import type { RunLogStore, RunLogHandle, RunLogBeginInput } from "./run-log-store.js"; -import { createObjectByteJournal, ObjectJournalSourcePrefixProofUnavailableError, type ObjectJournalWriter } from "./object-byte-journal.js"; +import { createObjectByteJournal, ObjectJournalSourcePrefixProofUnavailableError, type ObjectJournalIOPriority, type ObjectJournalWriter } from "./object-byte-journal.js"; import { MAX_RUN_LOG_RECORD_BYTES } from "./run-log-record-limits.js"; const MAX_SEGMENT_BYTES = 1024 * 1024; @@ -85,12 +85,13 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL else activeUploads--; }; } - function journal(logRef: string) { + function journal(logRef: string, readPriority?: ObjectJournalIOPriority) { if (!/^[a-zA-Z0-9._-]+\/[a-zA-Z0-9._-]+\/[a-zA-Z0-9._-]+$/.test(logRef)) throw new Error("Invalid object log reference"); return createObjectByteJournal({ provider: options.provider, prefix: `${options.keyPrefix}/segments-v1/${logRef}`, + readPriority, }); } function schedule(state: WriterState) { @@ -352,7 +353,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL }, async read(handle, opts = {}) { if (handle.store !== "s3_segments" && handle.store !== "s3_archive") throw new Error("Invalid object run log backend"); - const store = journal(handle.logRef), + const store = journal(handle.logRef, opts.priority ?? "interactive"), snapshot = await store.head(opts.signal); if (!snapshot || snapshot.head.deleted) throw new Error("Object run log not found"); const offset = Math.max(0, opts.offset ?? 0), diff --git a/server/src/services/provider-trace-object-store.test.ts b/server/src/services/provider-trace-object-store.test.ts index 955252dcbc5..f42077fc237 100644 --- a/server/src/services/provider-trace-object-store.test.ts +++ b/server/src/services/provider-trace-object-store.test.ts @@ -1,9 +1,12 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; +import { Readable } from "node:stream"; import { promises as fs } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createS3StorageProvider } from "../storage/s3-provider.js"; +import type { StorageProvider } from "../storage/types.js"; +import { ObjectJournalBusyError } from "./object-byte-journal.js"; import { createProviderTraceObjectStore } from "./provider-trace-object-store.js"; import { S3Client, @@ -93,3 +96,79 @@ describe.skipIf(!endpoint)("encrypted object provider traces (S3 integration)", 30_000, ); }); + +describe("encrypted object provider trace reads under load", () => { + function memory() { + const objects = new Map(); + const etag = (b: Buffer) => createHash("sha256").update(b).digest("hex"); + const provider: StorageProvider = { + id: "s3", + async putObject() { + throw Error("unconditional"); + }, + async putObjectConditional(i) { + const old = objects.get(i.objectKey); + if (i.ifNoneMatch ? old : !old || etag(old) !== i.ifMatch) + throw Object.assign(Error("conflict"), { name: "PreconditionFailed" }); + const chunks: Buffer[] = []; + if (Buffer.isBuffer(i.body)) chunks.push(i.body); + else for await (const bytes of i.body) chunks.push(Buffer.from(bytes)); + const b = Buffer.concat(chunks); + objects.set(i.objectKey, b); + return { etag: etag(b) }; + }, + async getObject(i) { + const b = objects.get(i.objectKey); + if (!b) throw Object.assign(Error("missing"), { name: "NoSuchKey" }); + return { stream: Readable.from([b]), etag: etag(b) }; + }, + async headObject(i) { + return { exists: objects.has(i.objectKey) }; + }, + async deleteObject(i) { + objects.delete(i.objectKey); + }, + async listObjects(i) { + return { keys: [...objects.keys()].filter((k) => k.startsWith(i.prefix)) }; + }, + }; + return provider; + } + + it("never sheds a finalization read while client reads are shed", async () => { + const directory = await fs.mkdtemp(path.join(tmpdir(), "object-trace-load-")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64)); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY_FILE", path.join(directory, "unused.key")); + try { + const provider = memory(), + store = createProviderTraceObjectStore(provider), + file = path.join(directory, "trace.ndjson"), + ref = `s3-segments-v1:company/run/${randomUUID()}.ndjson`, + bytes = Buffer.from('{"kind":"frame","frameId":1}\n'); + await store.open(ref, file, async () => undefined); + await fs.appendFile(file, bytes); + await store.finish(ref); + const get = provider.getObject; + let open!: () => void; + const gate = new Promise((resolve) => { + open = resolve; + }); + provider.getObject = async (i) => { + await gate; + return get(i); + }; + const clients = Array.from({ length: 40 }, () => store.read(ref, undefined, "interactive")); + await expect(store.read(ref, undefined, "interactive")).rejects.toBeInstanceOf( + ObjectJournalBusyError, + ); + const finalization = store.read(ref); + open(); + for (const read of await Promise.allSettled(clients)) + if (read.status === "rejected") expect(read.reason).toBeInstanceOf(ObjectJournalBusyError); + else expect(read.value).toEqual(bytes); + expect(await finalization).toEqual(bytes); + } finally { + await fs.rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/server/src/services/provider-trace-object-store.ts b/server/src/services/provider-trace-object-store.ts index f122a5938eb..308afdbf8c1 100644 --- a/server/src/services/provider-trace-object-store.ts +++ b/server/src/services/provider-trace-object-store.ts @@ -5,6 +5,7 @@ import { promises as fs } from "node:fs"; import { createObjectByteJournal, type ObjectByteJournal, + type ObjectJournalIOPriority, type ObjectJournalWriter, } from "./object-byte-journal.js"; import { deriveLocalEncryptedInstanceSecret } from "../secrets/local-encrypted-provider.js"; @@ -64,7 +65,7 @@ export function createProviderTraceObjectStore( }, 1000); capture.timer.unref(); } - function channelsFor(ref: string, filePath = "") { + function channelsFor(ref: string, filePath = "", readPriority?: ObjectJournalIOPriority) { const match = objectProviderTraceRef(ref); if (!match) throw new Error("invalid_object_trace_ref"); return ["native", "rehydration"].map((channel, i) => { @@ -79,6 +80,7 @@ export function createProviderTraceObjectStore( const journal = createObjectByteJournal({ provider, prefix: root, + readPriority, async encode(bytes) { const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); @@ -238,9 +240,13 @@ export function createProviderTraceObjectStore( capture.finishIdle(); } } - async function* readChunks(ref: string, signal?: AbortSignal) { + async function* readChunks( + ref: string, + signal?: AbortSignal, + priority: ObjectJournalIOPriority = "background", + ) { let nativeTail: number | undefined; - for (const [index, { journal }] of channelsFor(ref).entries()) { + for (const [index, { journal }] of channelsFor(ref, "", priority).entries()) { const snapshot = await journal.head(signal); if (!snapshot || snapshot.head.deleted) throw new Error("provider_trace_payload_unavailable"); if (snapshot.head.end > MAX_CAPTURE_BYTES) @@ -254,9 +260,13 @@ export function createProviderTraceObjectStore( } } } - async function read(ref: string, signal?: AbortSignal) { + async function read( + ref: string, + signal?: AbortSignal, + priority: ObjectJournalIOPriority = "background", + ) { const chunks: Buffer[] = []; - for await (const bytes of readChunks(ref, signal)) chunks.push(bytes); + for await (const bytes of readChunks(ref, signal, priority)) chunks.push(bytes); return Buffer.concat(chunks); } async function stop(ref: string) { diff --git a/server/src/services/provider-trace-store.ts b/server/src/services/provider-trace-store.ts index be89478ba3a..c69977d63d3 100644 --- a/server/src/services/provider-trace-store.ts +++ b/server/src/services/provider-trace-store.ts @@ -10,6 +10,7 @@ import type { } from "@paperclipai/shared"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { createProviderTraceObjectStore, objectProviderTraceRef, importingProviderTraceRef, providerTraceCleanupRef, type ProviderTraceObjectStore } from "./provider-trace-object-store.js"; +import { isObjectJournalBusyError, type ObjectJournalIOPriority } from "./object-byte-journal.js"; import { resolveRunEvidenceStorage } from "./run-evidence-storage.js"; import { loadStorageConfig } from "../storage/config.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -268,11 +269,11 @@ export function providerTraceStore(db: Db, options: { objectStore?: ProviderTrac traceObjectStores.set(db, external); return external; } - async function readBytes(traceRef: string) { - return objectProviderTraceRef(traceRef) ? objectStore().read(traceRef) : readTraceBytes(traceRef); + async function readBytes(traceRef: string, priority: ObjectJournalIOPriority) { + return objectProviderTraceRef(traceRef) ? objectStore().read(traceRef, undefined, priority) : readTraceBytes(traceRef); } async function entriesFor(traceRef: string) { - return (await readBytes(traceRef)).toString("utf8").split("\n").flatMap(line => { + return (await readBytes(traceRef, "interactive")).toString("utf8").split("\n").flatMap(line => { if (!line.trim()) return []; try { return [record(JSON.parse(line))]; } catch { return [{ kind: "trace_status", status: "incomplete", reason: "invalid_sidecar_record" }]; } @@ -508,7 +509,7 @@ export function providerTraceStore(db: Db, options: { objectStore?: ProviderTrac await objectStore().finish(row.traceRef); if (!await objectStore().settled(row.traceRef)) throw new Error("trace_capture_unsettled"); } - bytes = await readBytes(row.traceRef); + bytes = await readBytes(row.traceRef, "background"); } catch { const updatedAt = new Date(); const finalized = await db @@ -575,7 +576,10 @@ export function providerTraceStore(db: Db, options: { objectStore?: ProviderTrac async function inspect(runId: string, companyId: string) { const row = await getReadableByRun(runId, companyId); if (!row) return { trace: null, entries: [] }; - const entries = await entriesFor(row.traceRef).catch(() => []); + const entries = await entriesFor(row.traceRef).catch((error) => { + if (isObjectJournalBusyError(error)) throw error; + return []; + }); return { trace: asMetadata(row), entries: entries.map((entry) => @@ -609,7 +613,7 @@ export function providerTraceStore(db: Db, options: { objectStore?: ProviderTrac async function download(runId: string, companyId: string) { const row = await getReadableByRun(runId, companyId); if (!row) return null; - return { row, bytes: await readBytes(row.traceRef) }; + return { row, bytes: await readBytes(row.traceRef, "interactive") }; } async function remove(runId: string, companyId: string) { diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 5912f26a3a6..1e9120a21b8 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -7,6 +7,7 @@ import { notFound } from "../errors.js"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { createS3StorageProvider } from "../storage/s3-provider.js"; import type { StorageProvider } from "../storage/types.js"; +import type { ObjectJournalIOPriority } from "./object-byte-journal.js"; import { createObjectRunLogStore } from "./object-run-log-store.js"; import { resolveRunEvidenceStorage } from "./run-evidence-storage.js"; @@ -32,6 +33,7 @@ export interface RunLogReadOptions { /** Include the start of the first NDJSON record when reading a log tail. */ alignStartToRecord?: boolean; signal?: AbortSignal; + priority?: ObjectJournalIOPriority; } export interface RunLogReadResult { diff --git a/tests/runner-e2e/workflow-target.test.ts b/tests/runner-e2e/workflow-target.test.ts index 7f7aab50686..d02c01d647b 100644 --- a/tests/runner-e2e/workflow-target.test.ts +++ b/tests/runner-e2e/workflow-target.test.ts @@ -18,8 +18,7 @@ async function resolveTarget(event: string, cadence: string, branch = "master") encoding: "utf8", env: { ...process.env, EVENT_NAME: event, EVAL_CADENCE: cadence, IMAGE_ONLY: "false", TESTOPS_REQUESTED: "false", TARGET_BRANCH: branch, DEFAULT_BRANCH: "master", GITHUB_SHA: queuedSha, - MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip", - IMAGE_ONLY: "false", TESTOPS_REQUESTED: "false" }, + MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip" }, }); return { status: result.status, stderr: result.stderr, output: await readFile(output, "utf8").catch(() => "") };