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 d0a937768d 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 <[email protected]>
Co-Authored-By: Paperclip <[email protected]>

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"],
     [

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;

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();

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,

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,

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;

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();

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,
+    });
+  });
+});

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(

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<Uint8Array>) {
   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<void>((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<void>((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" });

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<T>(operation: () => Promise<T>): Promise<T> {
-  if (activeIO >= 8) {
-    if (waitingIO.length >= 32) throw new Error("object_journal_io_capacity_exceeded");
-    await new Promise<void>((resolve) => waitingIO.push(resolve));
+const waitingIO: Record<ObjectJournalIOPriority, Array<() => void>> = {
+  background: [],
+  interactive: [],
+};
+async function objectIO<T>(
+  operation: () => Promise<T>,
+  priority: ObjectJournalIOPriority = "background",
+  signal?: AbortSignal,
+): Promise<T> {
+  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<void>((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<T>(operation: () => Promise<T>): Promise<T> {
 export interface ObjectByteJournalOptions {
   provider: StorageProvider;
   prefix: string;
+  readPriority?: ObjectJournalIOPriority;
 
   encode?: (bytes: Buffer) => Promise<Buffer>;
   decode?: (bytes: Buffer) => Promise<Buffer>;
@@ -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<NonNullable<StorageProvider["putObjectConditional"]>>[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<ObjectJournalWriter | null> {
     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);

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<Uint8Array>) {
@@ -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<void>((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<typeof store.read> | 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] : [])]);
+    }
+  });
 });

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),

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<string, Buffer>();
+    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<void>((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 });
+    }
+  });
+});

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) {

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) {

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 {

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(() => "") };