mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(run-log): read a resumed run's own log while the store still hides its new head (#20523)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - In cloud mode, run logs use the `s3_segments` store: an object byte journal whose `head.json` names the current writer and the committed bytes > - When a run resumes after a restart, `runOutput.begin` claims the journal (this publishes a new `head.json`) and then reads the existing log prefix to find where native log export continues (`nativeRunLogPrefixCursor`) > - On an S3-compatible store with read-after-write lag, the `head.json` that the claim just published can read as `NoSuchKey` for some seconds > - The read then throws "Object run log not found", and the resumed run settles as `adapter_failed` and needs a Retry > - This pull request makes a read through a live writer's own handle use that writer's head when the store does not show the head yet, and makes the one-time source-prefix proof step treat a hidden head as "proof unavailable" instead of "another writer" > - The benefit is that a run that resumes right after a restart survives a store that shows new objects late, with no added wait ## Linked Issues or Issue Description Refs #20509 (the same lag on the journal claim). **What happened?** In restart-chaos wave w-20261110T095205-s701 (local, `--storage s3 --storage-faults ST7`, K1/K2, on master with #20504, #20509 and #20514), two resumed runs (t03 run 4f4fc7bc, t04 run c92c53d8) failed in `runOutput.begin`. The error was "Object run log not found" from `object-run-log-store.ts` `read`, called by `nativeRunLogPrefixCursor` (`native-run-log-export.ts:314`). Both runs settled `adapter_failed`, and the grade was R3 `retry_needed`. ST7 makes a GET or HEAD of a just-written key answer 404 `NoSuchKey` for 5–30 s. **Expected behavior** A resumed run reads the log prefix that it just claimed and continues. A store that shows a new object late must not look like a missing log. **Steps to reproduce** Write a run log, begin a successor writer on a new store instance, then make `head.json` read as `NoSuchKey`. `read(successor)` throws "Object run log not found". ## What Changed - `server/src/services/object-run-log-store.ts`: `read()` reads the head from the store as before. When the store says the head does not exist and this process has a live writer for that handle, it uses the writer's head. The writer published that head, so it proves the log exists. A handle without a live writer still fails at once with "Object run log not found". - `server/src/services/object-byte-journal.ts`: `recordVerifiedSourcePrefix` reads the head again to confirm ownership before it publishes the optional source-prefix proof. When that head reads as missing, it now throws `ObjectJournalSourcePrefixProofUnavailableError` (the existing "proof is optional" result). The store then continues without the proof, and the next resume scans the prefix again. A visible head with a different etag, a closed head or a deleted head still fails as `ObjectJournalConflictError`. Later appends still publish the head with a conditional write against the writer's etag, so another writer is still detected. - `server/src/services/object-run-log-store.test.ts`: a test hides the successor's just-published `head.json` for the full resume sequence that `nativeRunLogPrefixCursor` runs against the store (`read`, then `recordVerifiedSourcePrefix`). Before this change, `read` fails with "Object run log not found"; with only the read fix, the proof step fails with "The output journal is owned by another writer". After both, the read returns records 1–3, the proof is marked unavailable, and a later append and read return records 1–4. A second test checks that a read of a log that was never written still fails at once. ## Verification - `cd server && npx vitest run src/services/object-byte-journal.test.ts src/services/object-run-log-store.test.ts src/services/object-run-log-store-tail-read.test.ts src/services/run-log-store.test.ts src/services/run-log-store-cancellation.test.ts`: 87 pass. The new resume test fails with "Object run log not found" without the read fix, and with "The output journal is owned by another writer" without the proof fix. - `cd server && npx vitest run src/services/native-runtime/native-run-log-recovery.test.ts src/services/native-runtime/native-run-log-watermark.test.ts`: 27 pass. - `cd server && npx tsc --noEmit -p tsconfig.json` passes. - Live: a local ST7 rerun (seed 701) on a build with this fix follows in a comment. ## Risks - Low risk. The read change only affects a read whose store head is missing and whose handle has a live writer in this process. All other reads behave as before, including fetch counts and the cache. - When the head is hidden during the proof step, the proof is skipped for this attempt. The next resume of the same run scans the log prefix once more. This is the existing degraded path for a proof outage. - The fix does not wait for lagged segment objects. A segment that the previous server wrote less than the lag window before a restart can still read as `NoSuchKey` during the prefix scan. The wave evidence shows only the head lag. A broad segment wait broke existing failure tests, so it is not part of this change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
bf45700ca3
commit
adb0e77a4c
3 files changed
+37
-2
No files matched your search
@@ -476,7 +476,8 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
|
|||||||
if (!Number.isSafeInteger(sourceCursor) || sourceCursor < writer.head.sourceCursor)
|
if (!Number.isSafeInteger(sourceCursor) || sourceCursor < writer.head.sourceCursor)
|
||||||
throw new Error("object_journal_invalid_source_prefix_cursor");
|
throw new Error("object_journal_invalid_source_prefix_cursor");
|
||||||
const current = await head(signal);
|
const current = await head(signal);
|
||||||
if (!current || current.etag !== writer.etag || current.head.closed || current.head.deleted)
|
if (!current) throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
|
||||||
|
if (current.etag !== writer.etag || current.head.closed || current.head.deleted)
|
||||||
throw new ObjectJournalConflictError();
|
throw new ObjectJournalConflictError();
|
||||||
// One-time verification can repair legacy max-cursor metadata, never payload bytes.
|
// One-time verification can repair legacy max-cursor metadata, never payload bytes.
|
||||||
if (sourceCursor !== writer.head.sourceCursor)
|
if (sourceCursor !== writer.head.sourceCursor)
|
||||||
|
|||||||
@@ -289,6 +289,40 @@ describe("S3 run log store", () => {
|
|||||||
.rejects.toThrow("object_journal_source_prefix_proof_invalid");
|
.rejects.toThrow("object_journal_source_prefix_proof_invalid");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("resumes a run log through the prefix read and proof while the store still hides the head it just published", async () => {
|
||||||
|
const m = memory(),
|
||||||
|
options = { provider: m.provider, keyPrefix: "logs", flushMs: 10, assertWriterAuthority: async () => undefined };
|
||||||
|
const lagged = { ...input, runId: "lagged-run" };
|
||||||
|
const a = createObjectRunLogStore(options),
|
||||||
|
handle = await a.begin(lagged);
|
||||||
|
await Promise.all([1, 2, 3].map((seq) => a.append(handle, event(seq))));
|
||||||
|
const b = createObjectRunLogStore(options),
|
||||||
|
successor = await b.begin(lagged);
|
||||||
|
const original = m.provider.getObject;
|
||||||
|
let hiddenHeadReads = 0;
|
||||||
|
m.provider.getObject = async (i) => {
|
||||||
|
if (i.objectKey === "logs/segments-v1/company/agent/lagged-run/head.json") {
|
||||||
|
hiddenHeadReads++;
|
||||||
|
throw Object.assign(Error("not yet visible"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 } });
|
||||||
|
}
|
||||||
|
return original(i);
|
||||||
|
};
|
||||||
|
const read = await b.read(successor, { limitBytes: successor.initialBytes });
|
||||||
|
expect(read.content.trim().split("\n").map((line) => JSON.parse(line).seq)).toEqual([1, 2, 3]);
|
||||||
|
await b.recordVerifiedSourcePrefix!(successor, 0);
|
||||||
|
expect(successor.initialSourcePrefixVerified).toBe(false);
|
||||||
|
expect(hiddenHeadReads).toBeGreaterThanOrEqual(2);
|
||||||
|
m.provider.getObject = original;
|
||||||
|
await b.append(successor, event(4));
|
||||||
|
const resumed = await b.read(successor);
|
||||||
|
expect(resumed.content.trim().split("\n").map((line) => JSON.parse(line).seq)).toEqual([1, 2, 3, 4]);
|
||||||
|
});
|
||||||
|
it("fails a read of a run log that was never written", async () => {
|
||||||
|
const m = memory(),
|
||||||
|
store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined });
|
||||||
|
await expect(store.read({ store: "s3_segments", logRef: "company/agent/never-written", attemptId: "attempt" }))
|
||||||
|
.rejects.toThrow("Object run log not found");
|
||||||
|
});
|
||||||
it("batches concurrent appends, confirms durability, and resumes offsets/sequence on a replacement", async () => {
|
it("batches concurrent appends, confirms durability, and resumes offsets/sequence on a replacement", async () => {
|
||||||
const m = memory(),
|
const m = memory(),
|
||||||
options = {
|
options = {
|
||||||
|
|||||||
@@ -356,7 +356,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
|
|||||||
async read(handle, opts = {}) {
|
async read(handle, opts = {}) {
|
||||||
if (handle.store !== "s3_segments" && handle.store !== "s3_archive") throw new Error("Invalid object run log backend");
|
if (handle.store !== "s3_segments" && handle.store !== "s3_archive") throw new Error("Invalid object run log backend");
|
||||||
const store = journal(handle.logRef, opts.priority ?? "interactive"),
|
const store = journal(handle.logRef, opts.priority ?? "interactive"),
|
||||||
snapshot = await store.head(opts.signal);
|
snapshot = await store.head(opts.signal) ?? writers.get(handle)?.writer;
|
||||||
if (!snapshot || snapshot.head.deleted) throw new Error("Object run log not found");
|
if (!snapshot || snapshot.head.deleted) throw new Error("Object run log not found");
|
||||||
const offset = Math.max(0, opts.offset ?? 0),
|
const offset = Math.max(0, opts.offset ?? 0),
|
||||||
limit = opts.limitBytes ?? 256_000;
|
limit = opts.limitBytes ?? 256_000;
|
||||||
|
|||||||
Reference in new issue
Block a user