diff --git a/server/src/services/object-byte-journal.test.ts b/server/src/services/object-byte-journal.test.ts index 01706bbfa4e..aab25cef77b 100644 --- a/server/src/services/object-byte-journal.test.ts +++ b/server/src/services/object-byte-journal.test.ts @@ -257,6 +257,64 @@ describe("object byte journal", () => { expect(await journal.readRange(writer.head, 0, 100)).toEqual(Buffer.from("once")); expect(conflicts).toBe(2); }); + it("waits for a just-written head to become readable instead of failing the claim as another writer's", async () => { + const s = storage(), + journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); + let writer = await journal.claim({ assertAuthority: authority }); + writer = await journal.append(writer, Buffer.from("kept")); + const original = s.provider.getObject; + let hidden = 3; + s.provider.getObject = async (input) => { + if (input.objectKey.endsWith("/head.json") && hidden > 0) { + hidden--; + throw Object.assign(new Error("not yet visible"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 } }); + } + return original(input); + }; + const next = await journal.claim({ assertAuthority: authority }); + expect(next.head).toMatchObject({ generation: writer.head.generation + 1, end: 4 }); + expect(next.head.owner).not.toBe(writer.head.owner); + expect(hidden).toBe(0); + expect(await journal.readRange(next.head, 0, 100)).toEqual(Buffer.from("kept")); + }); + function hideHead(s: ReturnType, onRead: () => void = () => undefined) { + const original = s.provider.getObject; + let reads = 0; + s.provider.getObject = async (input) => { + if (input.objectKey.endsWith("/head.json")) { + reads++; + onRead(); + throw Object.assign(new Error("not yet visible"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 } }); + } + return original(input); + }; + return { reads: () => reads }; + } + it("stops waiting for an unreadable head at the deadline and makes no further storage calls", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const s = storage(), + journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); + await journal.claim({ assertAuthority: authority }); + const hidden = hideHead(s, () => vi.setSystemTime(Date.now() + 15_000)); + await expect(journal.claim({ assertAuthority: authority })).rejects.toBeInstanceOf(ObjectJournalConflictError); + const reads = hidden.reads(), puts = s.calls.put; + expect(reads).toBeGreaterThan(1); + await new Promise((resolve) => setTimeout(resolve, 1_500)); + expect([hidden.reads(), s.calls.put]).toEqual([reads, puts]); + }); + it("stops a claim that waits for an unreadable head when its caller aborts", async () => { + const s = storage(), + journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); + await journal.claim({ assertAuthority: authority }); + const hidden = hideHead(s); + const controller = new AbortController(); + const claiming = journal.claim({ assertAuthority: authority, signal: controller.signal }); + setTimeout(() => controller.abort(new Error("caller stopped")), 100); + await expect(claiming).rejects.toMatchObject({ name: "AbortError" }); + const reads = hidden.reads(), puts = s.calls.put; + await new Promise((resolve) => setTimeout(resolve, 600)); + expect([hidden.reads(), s.calls.put]).toEqual([reads, puts]); + }); it("does not accept a provider's 409 response as proof of conditional enforcement", async () => { const s = storage(), original = s.provider.putObjectConditional!; diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts index c4c6e160859..7ece64a9f32 100644 --- a/server/src/services/object-byte-journal.ts +++ b/server/src/services/object-byte-journal.ts @@ -6,6 +6,7 @@ import type { StorageProvider } from "../storage/types.js"; const MAX_INDEX_LEVEL = 52; const MAX_OBJECT_BYTES = 4 * 1024 * 1024; +const HEAD_VISIBILITY_WAIT_MS = 60_000; const offset = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER); const referenceSchema = z .object({ @@ -539,7 +540,21 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { assertAuthority: () => Promise; signal?: AbortSignal; }) { - const current = await head(input.signal); + const deadline = Date.now() + HEAD_VISIBILITY_WAIT_MS; + for (let wait = 250; ; wait = Math.min(wait * 2, 4_000)) { + const current = await head(input.signal); + try { + return await claimFrom(current, input); + } catch (error) { + if (current || !(error instanceof ObjectJournalConflictError) || Date.now() + wait > deadline) throw error; + await delay(wait, undefined, { signal: input.signal }); + } + } + } + async function claimFrom(current: ObjectJournalWriter | null, input: { + assertAuthority: () => Promise; + signal?: AbortSignal; + }) { if (current?.head.deleted) throw new ObjectJournalConflictError(); await input.assertAuthority(); await qualifyConditionalWrites(provider, prefix, input.signal);