feat(run-logs): durable run-log store via object-storage mirror (#8984)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Every agent run streams its stdout/stderr/system output into the run-log store (`server/src/services/run-log-store.ts`), and the run-log API serves those logs back for review and debugging > - The only store implementation is `local_file`: logs live on the server pod's filesystem under `PAPERCLIP_HOME` > - In hardened / ephemeral deployments, `PAPERCLIP_HOME` is an `emptyDir` with no persistent volume, so every pod restart wipes the log files while the DB row still references them — the run-log API then returns "Run log not found" for every completed run after any redeploy > - Run logs are the primary audit/debugging trail for agent work; losing them on routine redeploys undermines trust in the platform > - This pull request adds transparent durability: when `RUN_LOG_S3_BUCKET` is set, the store mirrors each completed log to object storage on `finalize` (same `logRef` key) and falls back to it on `read` when the local file is gone; live append/tail stays on the fast pod-local file > - The benefit is that completed run logs survive pod restarts and redeploys with zero changes for existing deployments (unset bucket = today's behaviour) and zero downstream changes (store id stays `local_file`) ## Linked Issues or Issue Description No existing public issue — inline description following the bug report template: **What happened?** After any server pod restart/redeploy, the run-log API returns "Run log not found" for all previously completed runs. The DB still references the log file, but the file is gone because run logs are written only to the pod-local filesystem. **Expected behavior:** Completed run logs remain readable across pod restarts and redeploys. **Steps to reproduce:** 1. Deploy the server with `PAPERCLIP_HOME` on an `emptyDir` (no persistent volume — common in hardened/ephemeral Kubernetes deployments). 2. Complete an agent run and confirm its log is readable via the run-log API. 3. Restart or redeploy the server pod. 4. Request the same run's log — the API throws "Run log not found". **Paperclip version or commit:** reproducible on current `master`. **Deployment mode:** Kubernetes (server pod without persistent volume). **Agent adapter(s) involved:** Not adapter-specific (core bug). Supersedes #8795. ## What Changed - `server/src/services/run-log-store.ts`: the local-file store becomes a durable store with an optional object-storage mirror - `finalize` mirrors the completed NDJSON log to S3-compatible object storage (keyed by the same `logRef`), best-effort so a failed upload can never break run finalization; upload failures are logged via `console.warn` so operators can detect a persistently broken mirror before a pod roll makes logs unreadable - `read` serves the pod-local file when present and falls back to a ranged object-storage read (with correct `nextOffset`) when the local file is gone - Live `append`/tail stays on the pod-local file — fast path unchanged, no per-chunk PUT - Store id stays `local_file`, so nothing downstream changes (feedback pipeline, read casts, fixtures untouched) - New optional config, all read at store construction: `RUN_LOG_S3_BUCKET`, `RUN_LOG_S3_ENDPOINT`, `RUN_LOG_S3_REGION` (default `us-east-1`), `RUN_LOG_S3_PREFIX` (default `run-logs`), `RUN_LOG_S3_FORCE_PATH_STYLE` (default `true`); credentials via the standard AWS env chain; works with any S3-compatible endpoint - Reuses the existing `createS3StorageProvider`; deliberately independent from `PAPERCLIP_STORAGE_PROVIDER` so enabling durable logs does not redirect workspace/file storage - `server/src/services/run-log-store.test.ts` (new): 7 tests with an in-memory `StorageProvider` mock ## Verification - `npx vitest run src/services/run-log-store.test.ts` in `server/` — 7/7 pass locally: - store id stays `local_file` - live read served from the local file (no S3 round-trip) - `finalize` uploads the completed log to the mirror - read falls back to S3 after a simulated pod roll (local file deleted) - ranged S3 read returns correct slice + `nextOffset` - not-found when neither local nor mirror has the log - local-only safe degrade when no bucket is configured - `npx tsc --noEmit -p server` — clean for the touched files - Manual: set `RUN_LOG_S3_*` against any S3-compatible endpoint (e.g. MinIO), complete a run, delete the local `.ndjson` file, and re-request the log via the run-log API — it is served from the mirror ## Risks - Low risk: with `RUN_LOG_S3_BUCKET` unset (the default), behaviour is byte-for-byte today's local-only store - Mirror upload is best-effort by design — a misconfigured bucket loses durability (not correctness) for affected runs; failures are now surfaced via a `console.warn` per failed upload - No DB migration, no API shape change, no change to the persisted `store`/`logRef` handle format ## Model Used - Claude (Anthropic), model ID `claude-fable-5` (Fable 5), via Claude Code with extended thinking and tool use (code execution, file editing). Original implementation TDD-authored with the same tooling. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jannes Stubbemann
GitHub
parent
543de323f6
commit
b4e7ba5143
|
|
@ -0,0 +1,168 @@
|
|||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import os from "node:os";
|
||||
import { Readable } from "node:stream";
|
||||
import { createDurableRunLogStore } from "./run-log-store.js";
|
||||
import type { StorageProvider } from "../storage/types.js";
|
||||
|
||||
// In-memory StorageProvider stand-in: durable, survives the "pod roll" (local
|
||||
// dir wipe) the same way Cubbit does. Records calls so we can assert behaviour.
|
||||
function createMemoryProvider() {
|
||||
const objects = new Map<string, Buffer>();
|
||||
const calls = { put: 0, get: 0, head: 0 };
|
||||
const provider: StorageProvider = {
|
||||
id: "s3",
|
||||
async putObject(input) {
|
||||
calls.put++;
|
||||
if (Buffer.isBuffer(input.body)) {
|
||||
objects.set(input.objectKey, Buffer.from(input.body));
|
||||
return;
|
||||
}
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of input.body) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
}
|
||||
objects.set(input.objectKey, Buffer.concat(chunks));
|
||||
},
|
||||
async getObject(input) {
|
||||
calls.get++;
|
||||
const buf = objects.get(input.objectKey);
|
||||
if (!buf) {
|
||||
const err = new Error("Object not found") as Error & { name: string };
|
||||
err.name = "NoSuchKey";
|
||||
throw err;
|
||||
}
|
||||
const slice = input.range ? buf.subarray(input.range.start, input.range.end + 1) : buf;
|
||||
return { stream: Readable.from(slice), contentLength: slice.length };
|
||||
},
|
||||
async headObject(input) {
|
||||
calls.head++;
|
||||
const buf = objects.get(input.objectKey);
|
||||
return buf ? { exists: true, contentLength: buf.length } : { exists: false };
|
||||
},
|
||||
async deleteObject(input) {
|
||||
objects.delete(input.objectKey);
|
||||
},
|
||||
};
|
||||
return { provider, objects, calls };
|
||||
}
|
||||
|
||||
let baseDir: string;
|
||||
beforeEach(async () => {
|
||||
baseDir = await fs.mkdtemp(path.join(os.tmpdir(), "run-log-store-test-"));
|
||||
});
|
||||
afterEach(async () => {
|
||||
await fs.rm(baseDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
const begin = { companyId: "co1", agentId: "ag1", runId: "run1" };
|
||||
|
||||
describe("createDurableRunLogStore", () => {
|
||||
it("keeps store id 'local_file' so downstream coupling (feedback, casts) is unchanged", async () => {
|
||||
const { provider } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } });
|
||||
const handle = await store.begin(begin);
|
||||
expect(handle.store).toBe("local_file");
|
||||
});
|
||||
|
||||
it("appends locally and reads back during a run WITHOUT hitting S3 (fast live tail)", async () => {
|
||||
const { provider, calls } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } });
|
||||
const handle = await store.begin(begin);
|
||||
await store.append(handle, { stream: "stdout", chunk: "hello", ts: "t1" });
|
||||
const res = await store.read(handle);
|
||||
expect(res.content).toContain("hello");
|
||||
expect(calls.get).toBe(0); // local file present -> no S3 read
|
||||
});
|
||||
|
||||
it("uploads the complete log to S3 on finalize", async () => {
|
||||
const { provider, objects, calls } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider, keyPrefix: "run-logs" } });
|
||||
const handle = await store.begin(begin);
|
||||
await store.append(handle, { stream: "stdout", chunk: "line-A", ts: "t1" });
|
||||
await store.append(handle, { stream: "stdout", chunk: "line-B", ts: "t2" });
|
||||
const summary = await store.finalize(handle);
|
||||
expect(calls.put).toBe(1);
|
||||
expect(summary.bytes).toBeGreaterThan(0);
|
||||
// keyed by prefix + the handle's logRef so read can find it later
|
||||
const key = `run-logs/${handle.logRef}`;
|
||||
expect(objects.has(key)).toBe(true);
|
||||
expect(objects.get(key)!.toString("utf8")).toContain("line-A");
|
||||
expect(objects.get(key)!.toString("utf8")).toContain("line-B");
|
||||
});
|
||||
|
||||
it("falls back to S3 when the local file is gone (the pod-roll case that caused 'Run log not found')", async () => {
|
||||
const { provider } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider, keyPrefix: "run-logs" } });
|
||||
const handle = await store.begin(begin);
|
||||
await store.append(handle, { stream: "stdout", chunk: "persisted-line", ts: "t1" });
|
||||
await store.finalize(handle);
|
||||
// Simulate a pod restart wiping the emptyDir.
|
||||
await fs.rm(baseDir, { recursive: true, force: true });
|
||||
const res = await store.read(handle);
|
||||
expect(res.content).toContain("persisted-line");
|
||||
});
|
||||
|
||||
it("S3 fallback honours offset/limitBytes (range read) and reports nextOffset", async () => {
|
||||
const { provider } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider, keyPrefix: "p" } });
|
||||
const handle = await store.begin(begin);
|
||||
// one line; persisted bytes = JSON line + "\n"
|
||||
await store.append(handle, { stream: "stdout", chunk: "0123456789", ts: "t" });
|
||||
await store.finalize(handle);
|
||||
const full = await store.read(handle); // from local, to learn total size
|
||||
const total = Buffer.byteLength(full.content, "utf8");
|
||||
await fs.rm(baseDir, { recursive: true, force: true }); // force S3 path
|
||||
const firstHalf = await store.read(handle, { offset: 0, limitBytes: 5 });
|
||||
expect(Buffer.byteLength(firstHalf.content, "utf8")).toBe(5);
|
||||
expect(firstHalf.nextOffset).toBe(5);
|
||||
const tail = await store.read(handle, { offset: total - 3, limitBytes: 100 });
|
||||
expect(Buffer.byteLength(tail.content, "utf8")).toBe(3);
|
||||
expect(tail.nextOffset).toBeUndefined();
|
||||
});
|
||||
|
||||
it("falls back to S3 when the local file vanishes between stat() and open (TOCTOU race)", async () => {
|
||||
const { provider } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider, keyPrefix: "run-logs" } });
|
||||
const handle = await store.begin(begin);
|
||||
await store.append(handle, { stream: "stdout", chunk: "raced-line", ts: "t1" });
|
||||
await store.finalize(handle);
|
||||
// Delete the local file DURING stat(), i.e. after it reports the file
|
||||
// present but before createReadStream opens it -> the open hits ENOENT.
|
||||
const realStat = fs.stat.bind(fs);
|
||||
const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, ...rest) => {
|
||||
const result = await realStat(target as Parameters<typeof realStat>[0], ...(rest as []));
|
||||
if (String(target).endsWith(".ndjson")) {
|
||||
await fs.rm(target as string, { force: true });
|
||||
}
|
||||
return result;
|
||||
});
|
||||
try {
|
||||
const res = await store.read(handle);
|
||||
expect(res.content).toContain("raced-line");
|
||||
} finally {
|
||||
statSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("throws notFound when neither local nor S3 has the log (pre-S3 run after a roll)", async () => {
|
||||
const { provider } = createMemoryProvider();
|
||||
const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } });
|
||||
const handle = await store.begin(begin);
|
||||
await fs.rm(baseDir, { recursive: true, force: true }); // never finalized -> never uploaded
|
||||
await expect(store.read(handle)).rejects.toThrow(/not found/i);
|
||||
});
|
||||
|
||||
it("without S3 configured behaves exactly like the local-only store (safe degrade)", async () => {
|
||||
const store = createDurableRunLogStore({ basePath: baseDir });
|
||||
const handle = await store.begin(begin);
|
||||
await store.append(handle, { stream: "stdout", chunk: "local-only", ts: "t1" });
|
||||
await store.finalize(handle);
|
||||
const res = await store.read(handle);
|
||||
expect(res.content).toContain("local-only");
|
||||
// and a roll loses it (documented limitation; this is the pre-fix behaviour)
|
||||
await fs.rm(baseDir, { recursive: true, force: true });
|
||||
await expect(store.read(handle)).rejects.toThrow(/not found/i);
|
||||
});
|
||||
});
|
||||
|
|
@ -3,6 +3,8 @@ import path from "node:path";
|
|||
import { createHash } from "node:crypto";
|
||||
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";
|
||||
|
||||
export type RunLogStoreType = "local_file";
|
||||
|
||||
|
|
@ -50,38 +52,100 @@ function resolveWithin(basePath: string, relativePath: string) {
|
|||
return resolved;
|
||||
}
|
||||
|
||||
function createLocalFileRunLogStore(basePath: string): RunLogStore {
|
||||
function normalizeKeyPrefix(prefix: string | undefined): string {
|
||||
if (!prefix) return "";
|
||||
return prefix.trim().replace(/^\/+/, "").replace(/\/+$/, "");
|
||||
}
|
||||
|
||||
export interface DurableRunLogStoreOptions {
|
||||
basePath: string;
|
||||
// When provided, completed logs are mirrored to object storage on finalize and
|
||||
// served from there on read whenever the local file is missing (e.g. the pod
|
||||
// rolled and wiped the emptyDir). When omitted, the store is local-only (the
|
||||
// historical behaviour: a restart loses the log).
|
||||
s3?: { provider: StorageProvider; keyPrefix?: string };
|
||||
}
|
||||
|
||||
// Run-log store with TRANSPARENT durability. The store id stays "local_file" so
|
||||
// nothing downstream (feedback.ts, the heartbeat read cast, fixtures) changes;
|
||||
// the S3 mirror is keyed by the same logRef and is purely an implementation
|
||||
// detail. Live append/tail stays on the pod-local file (fast, no per-chunk PUT);
|
||||
// on finalize the complete .ndjson is uploaded to object storage; on read we try
|
||||
// local first and fall back to S3 when the local file is gone. This is the fix
|
||||
// for "Run log not found" after a deploy/restart (the /paperclip data dir is an
|
||||
// emptyDir in cloud_tenant mode -- persistence is disabled to avoid the
|
||||
// operator's privileged selinux-relabel init container in our hardened ns).
|
||||
export function createDurableRunLogStore(options: DurableRunLogStoreOptions): RunLogStore {
|
||||
const { basePath } = options;
|
||||
const s3 = options.s3;
|
||||
const s3Prefix = normalizeKeyPrefix(s3?.keyPrefix);
|
||||
|
||||
function s3Key(logRef: string): string {
|
||||
return s3Prefix ? `${s3Prefix}/${logRef}` : logRef;
|
||||
}
|
||||
|
||||
async function ensureDir(relativeDir: string) {
|
||||
const dir = resolveWithin(basePath, relativeDir);
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
}
|
||||
|
||||
async function readFileRange(filePath: string, offset: number, limitBytes: number): Promise<RunLogReadResult> {
|
||||
async function readLocalRange(
|
||||
filePath: string,
|
||||
offset: number,
|
||||
limitBytes: number,
|
||||
): Promise<RunLogReadResult | null> {
|
||||
const stat = await fs.stat(filePath).catch(() => null);
|
||||
if (!stat) throw notFound("Run log not found");
|
||||
|
||||
if (!stat) return null;
|
||||
const start = Math.max(0, Math.min(offset, stat.size));
|
||||
const end = Math.max(start, Math.min(start + limitBytes - 1, stat.size - 1));
|
||||
|
||||
if (start > end) {
|
||||
return { content: "", nextOffset: start };
|
||||
}
|
||||
if (start > end) return { content: "", nextOffset: start };
|
||||
|
||||
const chunks: Buffer[] = [];
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const stream = createReadStream(filePath, { start, end });
|
||||
stream.on("data", (chunk) => {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const stream = createReadStream(filePath, { start, end });
|
||||
stream.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
|
||||
stream.on("error", reject);
|
||||
stream.on("end", () => resolve());
|
||||
});
|
||||
stream.on("error", reject);
|
||||
stream.on("end", () => resolve());
|
||||
});
|
||||
|
||||
} catch (err) {
|
||||
// File deleted between stat() and open (pod-roll cleanup racing a read):
|
||||
// treat as missing so the caller falls through to the S3 mirror instead
|
||||
// of surfacing the very "Run log not found" this store exists to prevent.
|
||||
if ((err as NodeJS.ErrnoException | null)?.code === "ENOENT") return null;
|
||||
throw err;
|
||||
}
|
||||
const content = Buffer.concat(chunks).toString("utf8");
|
||||
const nextOffset = end + 1 < stat.size ? end + 1 : undefined;
|
||||
return { content, nextOffset };
|
||||
}
|
||||
|
||||
async function readS3Range(
|
||||
logRef: string,
|
||||
offset: number,
|
||||
limitBytes: number,
|
||||
): Promise<RunLogReadResult> {
|
||||
if (!s3) throw notFound("Run log not found");
|
||||
const key = s3Key(logRef);
|
||||
const head = await s3.provider.headObject({ objectKey: key });
|
||||
if (!head.exists) throw notFound("Run log not found");
|
||||
const total = head.contentLength ?? 0;
|
||||
const start = Math.max(0, Math.min(offset, total));
|
||||
const end = Math.max(start, Math.min(start + limitBytes - 1, total - 1));
|
||||
if (start > end || total === 0) return { content: "", nextOffset: start < total ? start : undefined };
|
||||
|
||||
const result = await s3.provider.getObject({ objectKey: key, range: { start, end } });
|
||||
const chunks: Buffer[] = [];
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
result.stream.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
|
||||
result.stream.on("error", reject);
|
||||
result.stream.on("end", () => resolve());
|
||||
});
|
||||
const content = Buffer.concat(chunks).toString("utf8");
|
||||
const nextOffset = end + 1 < total ? end + 1 : undefined;
|
||||
return { content, nextOffset };
|
||||
}
|
||||
|
||||
async function sha256File(filePath: string): Promise<string> {
|
||||
return new Promise<string>((resolve, reject) => {
|
||||
const hash = createHash("sha256");
|
||||
|
|
@ -99,10 +163,8 @@ function createLocalFileRunLogStore(basePath: string): RunLogStore {
|
|||
const relDir = path.join(companyId, agentId);
|
||||
const relPath = path.join(relDir, `${runId}.ndjson`);
|
||||
await ensureDir(relDir);
|
||||
|
||||
const absPath = resolveWithin(basePath, relPath);
|
||||
await fs.writeFile(absPath, "", "utf8");
|
||||
|
||||
return { store: "local_file", logRef: relPath };
|
||||
},
|
||||
|
||||
|
|
@ -124,38 +186,79 @@ function createLocalFileRunLogStore(basePath: string): RunLogStore {
|
|||
},
|
||||
|
||||
async finalize(handle) {
|
||||
if (handle.store !== "local_file") {
|
||||
return { bytes: 0, compressed: false };
|
||||
}
|
||||
if (handle.store !== "local_file") return { bytes: 0, compressed: false };
|
||||
const absPath = resolveWithin(basePath, handle.logRef);
|
||||
const stat = await fs.stat(absPath).catch(() => null);
|
||||
if (!stat) throw notFound("Run log not found");
|
||||
|
||||
const hash = await sha256File(absPath);
|
||||
return {
|
||||
bytes: stat.size,
|
||||
sha256: hash,
|
||||
compressed: false,
|
||||
};
|
||||
|
||||
// Mirror the completed log to object storage so it survives a pod roll.
|
||||
// Best-effort upload failures must NOT fail run finalization (which also
|
||||
// records cost/usage); the local copy still serves reads until the pod
|
||||
// rolls, and a failed mirror only loses durability for that one run.
|
||||
if (s3) {
|
||||
try {
|
||||
// Stream from disk instead of buffering the whole .ndjson in the
|
||||
// heap; long agent sessions can produce large logs. The file is
|
||||
// complete at this point, so stat.size is the exact content length.
|
||||
await s3.provider.putObject({
|
||||
objectKey: s3Key(handle.logRef),
|
||||
body: createReadStream(absPath),
|
||||
contentType: "application/x-ndjson",
|
||||
contentLength: stat.size,
|
||||
});
|
||||
} catch (err) {
|
||||
// Best-effort: finalization must not break, but a persistently
|
||||
// failing mirror (bad creds/bucket/endpoint) should be visible to
|
||||
// operators before a pod roll makes the logs unreadable.
|
||||
console.warn(
|
||||
`[run-log-store] Failed to mirror run log to object storage (key: ${s3Key(handle.logRef)}):`,
|
||||
err,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { bytes: stat.size, sha256: hash, compressed: false };
|
||||
},
|
||||
|
||||
async read(handle, opts) {
|
||||
if (handle.store !== "local_file") {
|
||||
throw notFound("Run log not found");
|
||||
}
|
||||
if (handle.store !== "local_file") throw notFound("Run log not found");
|
||||
const absPath = resolveWithin(basePath, handle.logRef);
|
||||
const offset = opts?.offset ?? 0;
|
||||
const limitBytes = opts?.limitBytes ?? 256_000;
|
||||
return readFileRange(absPath, offset, limitBytes);
|
||||
const local = await readLocalRange(absPath, offset, limitBytes);
|
||||
if (local) return local;
|
||||
// Local file gone (pod rolled) -> serve from the S3 mirror if configured.
|
||||
return readS3Range(handle.logRef, offset, limitBytes);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// Build the run-log S3 mirror from dedicated RUN_LOG_S3_* env. Deliberately
|
||||
// separate from PAPERCLIP_STORAGE_PROVIDER so enabling durable run logs does
|
||||
// NOT redirect the product's workspace/file storage (smaller blast radius).
|
||||
// Unset RUN_LOG_S3_BUCKET -> no mirror -> local-only (safe degrade). Creds come
|
||||
// from the standard AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY chain.
|
||||
function resolveRunLogS3(): { provider: StorageProvider; keyPrefix?: string } | undefined {
|
||||
const bucket = process.env.RUN_LOG_S3_BUCKET?.trim();
|
||||
if (!bucket) return undefined;
|
||||
const provider = createS3StorageProvider({
|
||||
bucket,
|
||||
region: process.env.RUN_LOG_S3_REGION?.trim() || "us-east-1",
|
||||
endpoint: process.env.RUN_LOG_S3_ENDPOINT?.trim() || undefined,
|
||||
prefix: undefined, // prefixing is handled by keyPrefix below (kept off the provider)
|
||||
forcePathStyle: process.env.RUN_LOG_S3_FORCE_PATH_STYLE
|
||||
? process.env.RUN_LOG_S3_FORCE_PATH_STYLE === "true"
|
||||
: true, // Cubbit (and most S3-compatible endpoints) need path-style
|
||||
});
|
||||
return { provider, keyPrefix: process.env.RUN_LOG_S3_PREFIX?.trim() || "run-logs" };
|
||||
}
|
||||
|
||||
let cachedStore: RunLogStore | null = null;
|
||||
|
||||
export function getRunLogStore() {
|
||||
if (cachedStore) return cachedStore;
|
||||
const basePath = process.env.RUN_LOG_BASE_PATH ?? path.resolve(resolvePaperclipInstanceRoot(), "data", "run-logs");
|
||||
cachedStore = createLocalFileRunLogStore(basePath);
|
||||
cachedStore = createDurableRunLogStore({ basePath, s3: resolveRunLogS3() });
|
||||
return cachedStore;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,9 @@ import type { Readable } from "node:stream";
|
|||
|
||||
export interface PutObjectInput {
|
||||
objectKey: string;
|
||||
body: Buffer;
|
||||
// Readable bodies stream straight to the backend (contentLength must be the
|
||||
// exact byte size); Buffer stays supported for small payloads.
|
||||
body: Buffer | Readable;
|
||||
contentType: string;
|
||||
contentLength: number;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue