mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
e482f0f632
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 <noreply@anthropic.com>
Co-Authored-By: Paperclip <noreply@paperclip.ing>
387 lines
17 KiB
TypeScript
387 lines
17 KiB
TypeScript
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,
|
|
ObjectJournalBusyError,
|
|
ObjectJournalConflictError,
|
|
} from "./object-byte-journal.js";
|
|
|
|
async function readBody(body: AsyncIterable<Uint8Array>) {
|
|
const chunks: Buffer[] = [];
|
|
for await (const bytes of body) chunks.push(Buffer.from(bytes));
|
|
return Buffer.concat(chunks);
|
|
}
|
|
|
|
function storage() {
|
|
const objects = new Map<string, Buffer>();
|
|
const calls = { get: 0, put: 0, maxPut: 0 };
|
|
let loseHeadResponse = false;
|
|
const etag = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
|
|
const missing = () => Object.assign(new Error("missing"), { name: "NoSuchKey" });
|
|
const conflict = () => Object.assign(new Error("conflict"), { name: "PreconditionFailed" });
|
|
const provider: StorageProvider = {
|
|
id: "s3",
|
|
async getObject({ objectKey, signal }) {
|
|
signal?.throwIfAborted();
|
|
calls.get++;
|
|
const bytes = objects.get(objectKey);
|
|
if (!bytes) throw missing();
|
|
return { stream: Readable.from([bytes]), etag: etag(bytes), contentLength: bytes.length };
|
|
},
|
|
async putObject() {
|
|
throw new Error("unconditional PUT forbidden");
|
|
},
|
|
async putObjectConditional(input) {
|
|
const old = objects.get(input.objectKey);
|
|
if (input.ifNoneMatch ? old : !old || etag(old) !== input.ifMatch) throw conflict();
|
|
const bytes = Buffer.isBuffer(input.body) ? input.body : await readBody(input.body);
|
|
objects.set(input.objectKey, Buffer.from(bytes));
|
|
calls.put++;
|
|
calls.maxPut = Math.max(calls.maxPut, bytes.length);
|
|
if (loseHeadResponse && input.objectKey.endsWith("/head.json")) {
|
|
loseHeadResponse = false;
|
|
throw new Error("response lost after PUT");
|
|
}
|
|
return { etag: etag(bytes) };
|
|
},
|
|
async headObject({ objectKey }) {
|
|
return { exists: objects.has(objectKey) };
|
|
},
|
|
async deleteObject({ objectKey }) {
|
|
objects.delete(objectKey);
|
|
},
|
|
async listObjects({ prefix }) {
|
|
return { keys: [...objects.keys()].filter((k) => k.startsWith(prefix)) };
|
|
},
|
|
};
|
|
return {
|
|
provider,
|
|
objects,
|
|
calls,
|
|
loseResponse: () => {
|
|
loseHeadResponse = true;
|
|
},
|
|
};
|
|
}
|
|
const authority = async () => undefined;
|
|
afterEach(() => vi.useRealTimers());
|
|
|
|
describe("object byte journal", () => {
|
|
it("keeps a shared capability probe alive for another writer when one claim cancels", async () => {
|
|
const s = storage();
|
|
const put = s.provider.putObjectConditional!;
|
|
let release!: () => void, started!: () => void, admitted!: () => void;
|
|
let probeSignal: AbortSignal | undefined;
|
|
const probing = new Promise<void>((resolve) => {
|
|
started = resolve;
|
|
});
|
|
const joined = new Promise<void>((resolve) => {
|
|
admitted = resolve;
|
|
});
|
|
s.provider.putObjectConditional = async (input) => {
|
|
if (input.objectKey.includes("conditional-probe-") && !probeSignal) {
|
|
probeSignal = input.signal;
|
|
await new Promise<void>((resolve, reject) => {
|
|
release = resolve;
|
|
input.signal!.addEventListener("abort", () => reject(input.signal!.reason), {
|
|
once: true,
|
|
});
|
|
started();
|
|
});
|
|
}
|
|
return put(input);
|
|
};
|
|
const controller = new AbortController();
|
|
const a = createObjectByteJournal({ provider: s.provider, prefix: "company/a/log" });
|
|
const first = a.claim({
|
|
expectedOwner: null,
|
|
assertAuthority: authority,
|
|
signal: controller.signal,
|
|
});
|
|
void first.catch(() => undefined);
|
|
await probing;
|
|
const b = createObjectByteJournal({ provider: s.provider, prefix: "company/b/log" });
|
|
const second = b.claim({
|
|
expectedOwner: null,
|
|
assertAuthority: async () => {
|
|
admitted();
|
|
},
|
|
});
|
|
await joined;
|
|
await Promise.resolve();
|
|
controller.abort();
|
|
await expect(first).rejects.toMatchObject({ name: "AbortError" });
|
|
expect(probeSignal!.aborted).toBe(false);
|
|
release();
|
|
const writer = await second;
|
|
expect((await b.head())!.head.owner).toBe(writer.head.owner);
|
|
});
|
|
it("recovers committed bytes and stable ranges with no local filesystem", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "company/run/log" });
|
|
let writer = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
const expected = Buffer.from("first\n再開\n".repeat(20));
|
|
writer = await a.append(writer, expected.subarray(0, 17));
|
|
writer = await a.append(writer, expected.subarray(17));
|
|
const b = createObjectByteJournal({ provider: s.provider, prefix: "company/run/log" });
|
|
const snapshot = (await b.head())!;
|
|
expect(await b.readRange(snapshot.head, 0, expected.length)).toEqual(expected);
|
|
expect(await b.readRange(snapshot.head, 7, 23)).toEqual(expected.subarray(7, 30));
|
|
expect(await b.readRange(snapshot.head, expected.length, 100)).toEqual(Buffer.alloc(0));
|
|
});
|
|
it("fences stale appends/finalization and preserves a pinned reader snapshot", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let old = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
old = await a.append(old, Buffer.from("before"));
|
|
const snapshot = old.head;
|
|
let next = await a.claim({ expectedOwner: old.head.owner, assertAuthority: authority });
|
|
next = await a.append(next, Buffer.from("after"));
|
|
await expect(a.append(old, Buffer.from("lost"))).rejects.toBeInstanceOf(
|
|
ObjectJournalConflictError,
|
|
);
|
|
await expect(a.close(old)).rejects.toBeInstanceOf(ObjectJournalConflictError);
|
|
expect(await a.readRange(snapshot, 0, 100)).toEqual(Buffer.from("before"));
|
|
expect(await a.readRange(next.head, 0, 100)).toEqual(Buffer.from("beforeafter"));
|
|
});
|
|
it("reconciles an uncertain manifest commit without duplicating output", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
s.loseResponse();
|
|
writer = await a.append(writer, Buffer.from("once"));
|
|
expect(writer.head.end).toBe(4);
|
|
expect(await a.readRange((await a.head())!.head, 0, 100)).toEqual(Buffer.from("once"));
|
|
});
|
|
it("keeps index objects bounded over multiple tree levels", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
for (let i = 0; i < 180; i++)
|
|
writer = await a.append(writer, Buffer.from(String(i).padStart(4, "0")));
|
|
expect(await a.readRange(writer.head, 252, 16)).toEqual(Buffer.from("0063006400650066"));
|
|
expect(await a.readRange(writer.head, 700, 20)).toEqual(Buffer.from("01750176017701780179"));
|
|
expect(s.calls.maxPut).toBeLessThan(20_000);
|
|
expect(s.calls.put).toBeLessThan(180 * 5);
|
|
const reachable = new Set<string>();
|
|
function visit(ref: { key: string }, level: number) {
|
|
reachable.add(ref.key);
|
|
if (level === 0) return;
|
|
const node = JSON.parse(s.objects.get(ref.key)!.toString());
|
|
for (const child of node.children) visit(child, level - 1);
|
|
}
|
|
for (const root of writer.head.roots!) visit(root, root.level);
|
|
expect([...reachable].sort()).toEqual(
|
|
[...s.objects.keys()].filter((key) => key.includes("/objects/")).sort(),
|
|
);
|
|
});
|
|
it("retries a transient head conflict without fencing its current writer", async () => {
|
|
const s = storage(),
|
|
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority });
|
|
const original = s.provider.putObjectConditional!;
|
|
let conflicts = 0;
|
|
s.provider.putObjectConditional = async (input) => {
|
|
if (input.objectKey.endsWith("/head.json") && conflicts++ === 0)
|
|
throw Object.assign(new Error("in-flight conflict"), {
|
|
name: "ConditionalRequestConflict",
|
|
$metadata: { httpStatusCode: 409 },
|
|
});
|
|
return original(input);
|
|
};
|
|
writer = await journal.append(writer, Buffer.from("once"));
|
|
expect(await journal.readRange(writer.head, 0, 100)).toEqual(Buffer.from("once"));
|
|
expect(conflicts).toBe(2);
|
|
});
|
|
it("does not accept a provider's 409 response as proof of conditional enforcement", async () => {
|
|
const s = storage(),
|
|
original = s.provider.putObjectConditional!;
|
|
s.provider.putObjectConditional = async (input) => {
|
|
if (s.objects.has(input.objectKey))
|
|
throw Object.assign(new Error("conflict"), { $metadata: { httpStatusCode: 409 } });
|
|
return original(input);
|
|
};
|
|
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
await expect(
|
|
journal.claim({ expectedOwner: null, assertAuthority: authority }),
|
|
).rejects.toThrow("conflict");
|
|
expect(s.objects.size).toBe(0);
|
|
});
|
|
it("retries transient segment failures and acknowledges exactly one batch", async () => {
|
|
const s = storage(),
|
|
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority });
|
|
const original = s.provider.putObjectConditional!;
|
|
let failures = 2;
|
|
s.provider.putObjectConditional = async (input) => {
|
|
if (input.objectKey.includes("/objects/") && failures-- > 0)
|
|
throw Object.assign(new Error("temporary outage"), { $metadata: { httpStatusCode: 503 } });
|
|
return original(input);
|
|
};
|
|
writer = await journal.append(writer, Buffer.from("one batch"));
|
|
expect(await journal.readRange(writer.head, 0, 100)).toEqual(Buffer.from("one batch"));
|
|
});
|
|
it("rejects corrupt published data and cancelled reads", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
writer = await a.append(writer, Buffer.from("data"));
|
|
const data = [...s.objects].find(([, bytes]) => bytes.toString() === "data")!;
|
|
s.objects.set(data[0], Buffer.from("oops"));
|
|
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" });
|
|
let writer = await a.claim({ expectedOwner: null, assertAuthority: authority });
|
|
writer = await a.append(writer, Buffer.from("private"));
|
|
await a.close(writer);
|
|
await a.remove();
|
|
expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]);
|
|
expect((await a.head())!.head.deleted).toBe(true);
|
|
await expect(
|
|
a.claim({ expectedOwner: writer.head.owner, assertAuthority: authority }),
|
|
).rejects.toThrow();
|
|
});
|
|
it("does not claim ownership when execution authority rejects", async () => {
|
|
const s = storage(),
|
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
await expect(
|
|
a.claim({
|
|
expectedOwner: null,
|
|
assertAuthority: async () => {
|
|
throw new Error("revoked");
|
|
},
|
|
}),
|
|
).rejects.toThrow("revoked");
|
|
expect(s.calls.put).toBe(0);
|
|
});
|
|
it("fences a pending first claim when deletion wins before journal creation", async () => {
|
|
const s = storage(),
|
|
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let enter!: () => void, release!: () => void;
|
|
const entered = new Promise<void>((resolve) => {
|
|
enter = resolve;
|
|
});
|
|
const blocked = new Promise<void>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const pending = journal.claim({
|
|
expectedOwner: null,
|
|
assertAuthority: async () => {
|
|
enter();
|
|
await blocked;
|
|
},
|
|
});
|
|
await entered;
|
|
await journal.remove();
|
|
release();
|
|
await expect(pending).rejects.toBeInstanceOf(ObjectJournalConflictError);
|
|
expect((await journal.head())!.head.deleted).toBe(true);
|
|
expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]);
|
|
});
|
|
it("rejects a compatible provider which silently ignores conditional writes", async () => {
|
|
const s = storage();
|
|
s.provider.putObjectConditional = async (input) => {
|
|
const bytes = Buffer.from(input.body as Buffer);
|
|
s.objects.set(input.objectKey, bytes);
|
|
return { etag: "ignored-condition" };
|
|
};
|
|
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
await expect(
|
|
journal.claim({ expectedOwner: null, assertAuthority: authority }),
|
|
).rejects.toThrow("does not enforce conditional");
|
|
expect(s.objects.size).toBe(0);
|
|
});
|
|
it("fences an active writer before cleanup and preserves the retry obligation", async () => {
|
|
vi.useFakeTimers({ toFake: ["Date"] });
|
|
const s = storage(),
|
|
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority });
|
|
writer = await journal.append(writer, Buffer.from("private"));
|
|
await expect(journal.remove()).rejects.toThrow("cleanup_pending");
|
|
expect((await journal.head())!.head.deleted).toBe(true);
|
|
await expect(journal.append(writer, Buffer.from("late"))).rejects.toBeInstanceOf(
|
|
ObjectJournalConflictError,
|
|
);
|
|
vi.setSystemTime(Date.now() + 30_001);
|
|
await journal.remove();
|
|
expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]);
|
|
});
|
|
it("does not publish a partial batch when a later segment fails", async () => {
|
|
const s = storage(),
|
|
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
|
const writer = await journal.claim({ expectedOwner: null, assertAuthority: authority });
|
|
const original = s.provider.putObjectConditional!;
|
|
const failed = Buffer.alloc(1024 * 1024, 2);
|
|
const key = createHash("sha256").update(failed).digest("hex");
|
|
s.provider.putObjectConditional = (input) => {
|
|
if (input.objectKey.endsWith(key)) throw Error("second segment failed");
|
|
return original(input);
|
|
};
|
|
await expect(
|
|
journal.append(writer, Buffer.concat([Buffer.alloc(1024 * 1024, 1), failed])),
|
|
).rejects.toThrow("second segment failed");
|
|
expect((await journal.head())!.head.end).toBe(0);
|
|
});
|
|
});
|