diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts index 646b76c9974..38cd52c527b 100644 --- a/server/src/services/object-byte-journal.ts +++ b/server/src/services/object-byte-journal.ts @@ -255,6 +255,109 @@ async function objectIO( } } +// Segments and index nodes are content-addressed: each key ends in the SHA-256 +// of the object's bytes, and load() verifies that digest. Their bytes cannot +// change, so repeated reads of a run's tail (live polling, reopening a task) +// reuse them instead of paying one object-store round trip per object. Only the +// mutable head is read again. The cache is bounded by total bytes (payload plus +// per-entry overhead) and evicts the least recently used objects first. Readers authorize before any read. +// Entries are scoped to the storage provider instance, so a different store +// (or a test's fresh store) never sees objects that another store returned. +const READ_CACHE_MAX_BYTES = 32 * 1024 * 1024; +const READ_CACHE_MAX_OBJECT_BYTES = 1024 * 1024; +const readCache = new Map(); +let readCacheBytes = 0; +const providerScopes = new WeakMap(); +let nextProviderScope = 0; +function providerCacheScope(provider: StorageProvider) { + let scope = providerScopes.get(provider); + if (scope === undefined) providerScopes.set(provider, (scope = nextProviderScope++)); + return `${scope}:`; +} + +// Concurrent readers of one object share one fetch. The fetch has its own +// controller: a reader that aborts only stops its own wait, and the fetch is +// aborted only when no reader is still waiting for it. +interface SharedObjectRead { + work: Promise; + controller: AbortController; + waiters: number; +} +const readsInFlight = new Map(); +function cachedObject(key: string) { + const bytes = readCache.get(key); + if (bytes) { + readCache.delete(key); + readCache.set(key, bytes); + } + return bytes; +} +// The budget counts each entry's key and a fixed bookkeeping overhead with its +// bytes, so many small segments cannot hold much more memory than the cap. +const READ_CACHE_ENTRY_OVERHEAD_BYTES = 256; +const readCacheEntryCost = (key: string, bytes: Buffer) => + bytes.length + key.length * 2 + READ_CACHE_ENTRY_OVERHEAD_BYTES; +function cacheObject(key: string, bytes: Buffer) { + if (bytes.length > READ_CACHE_MAX_OBJECT_BYTES || readCache.has(key)) return; + readCache.set(key, bytes); + readCacheBytes += readCacheEntryCost(key, bytes); + for (const [oldestKey, oldest] of readCache) { + if (readCacheBytes <= READ_CACHE_MAX_BYTES) break; + readCache.delete(oldestKey); + readCacheBytes -= readCacheEntryCost(oldestKey, oldest); + } +} +/** Bytes the read cache counts against its budget, including entry overhead. */ +export function objectJournalReadCacheBytes() { + return readCacheBytes; +} +export function clearObjectJournalReadCache() { + readCache.clear(); + readCacheBytes = 0; + // Readers already waiting keep their fetch; new readers start a new one. + readsInFlight.clear(); +} + +// One range read fetches its objects with bounded overlap. The shared objectIO +// limiter still caps the process; this keeps one tail read from taking all of +// its slots or filling the interactive wait queue. +const READ_RANGE_CONCURRENCY = 4; +function concurrencyLimit(max: number) { + let active = 0; + const waiting: Array<() => void> = []; + return async (operation: () => Promise): Promise => { + if (active >= max) await new Promise((resolve) => waiting.push(resolve)); + else active++; + try { + return await operation(); + } finally { + const next = waiting.shift(); + if (next) next(); + else active--; + } + }; +} + +// Wait for a shared operation, but stop waiting when this caller's signal aborts. +function untilAborted(operation: Promise, signal?: AbortSignal): Promise { + if (!signal) return operation; + signal.throwIfAborted(); + return new Promise((resolve, reject) => { + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + operation.then( + (value) => { + signal.removeEventListener("abort", onAbort); + resolve(value); + }, + (error) => { + signal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }); +} + export interface ObjectByteJournalOptions { provider: StorageProvider; prefix: string; @@ -280,6 +383,7 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { const headKey = `${prefix}/head.json`; const objectPrefix = `${prefix}/objects/`; const readPriority = options.readPriority ?? "background"; + const cacheScope = providerCacheScope(provider); const putConditional = ( input: Parameters>[0], ) => objectIO(() => provider.putObjectConditional!(input), "background", input.signal); @@ -462,11 +566,42 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { } async function load(ref: Reference, signal?: AbortSignal) { validateReference(ref); - const stored = await getBytes(ref.key, MAX_OBJECT_BYTES * 2, signal); - const bytes = options.decode ? await options.decode(stored.bytes) : stored.bytes; - if (bytes.length > MAX_OBJECT_BYTES || sha256(bytes) !== ref.sha256) - throw new Error("object_journal_integrity_error"); - return bytes; + signal?.throwIfAborted(); + const cacheKey = `${cacheScope}${ref.key}`; + const cached = cachedObject(cacheKey); + if (cached) return cached; + let shared = readsInFlight.get(cacheKey); + if (!shared) { + const controller = new AbortController(); + const work = (async () => { + const stored = await getBytes(ref.key, MAX_OBJECT_BYTES * 2, controller.signal); + const bytes = options.decode ? await options.decode(stored.bytes) : stored.bytes; + if (bytes.length > MAX_OBJECT_BYTES || sha256(bytes) !== ref.sha256) + throw new Error("object_journal_integrity_error"); + cacheObject(cacheKey, bytes); + return bytes; + })(); + const entry: SharedObjectRead = { work, controller, waiters: 0 }; + const release = () => { + if (readsInFlight.get(cacheKey) === entry) readsInFlight.delete(cacheKey); + }; + void work.then(release, release); + readsInFlight.set(cacheKey, entry); + shared = entry; + } + const entry = shared; + entry.waiters++; + try { + return await untilAborted(entry.work, signal); + } finally { + entry.waiters--; + // The last reader left before the fetch finished: stop the fetch, and + // let a later reader start a new one. + if (!entry.waiters && signal?.aborted) { + if (readsInFlight.get(cacheKey) === entry) readsInFlight.delete(cacheKey); + entry.controller.abort(signal.reason); + } + } } async function save( bytes: Buffer, @@ -540,33 +675,39 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { if ((!snapshot.root && !snapshot.roots?.length) || limit === 0 || start >= snapshot.end) return Buffer.alloc(0); const end = Math.min(snapshot.end, start + limit), - chunks: Buffer[] = []; + pieces: Array<{ start: number; bytes: Buffer }> = [], + limited = concurrencyLimit(READ_RANGE_CONCURRENCY), + failed = new AbortController(), + readSignal = signal ? AbortSignal.any([signal, failed.signal]) : failed.signal; + const overlaps = (ref: Reference) => ref.end > start && ref.start < end; + // Siblings are fetched concurrently; pieces are reassembled by offset. async function readPayload(ref: Reference) { - const bytes = await load(ref, signal); + const bytes = await limited(() => load(ref, readSignal)); if (bytes.length !== ref.end - ref.start) throw new Error("object_journal_invalid_segment_length"); - chunks.push( - bytes.subarray(Math.max(0, start - ref.start), Math.min(bytes.length, end - ref.start)), - ); + pieces.push({ + start: ref.start, + bytes: bytes.subarray(Math.max(0, start - ref.start), Math.min(bytes.length, end - ref.start)), + }); } - async function visit(ref: Reference, expectedLevel?: number) { - const node = await loadNode(ref, signal); + async function visit(ref: Reference, expectedLevel?: number): Promise { + const node = await limited(() => loadNode(ref, readSignal)); if (expectedLevel !== undefined && node.level !== expectedLevel) throw new Error("object_journal_invalid_index_level"); - for (const child of node.children) { - if (child.end <= start || child.start >= end) continue; - if (node.level > 0) await visit(child, node.level - 1); - else await readPayload(child); - } + await Promise.all(node.children.filter(overlaps).map((child) => + node.level > 0 ? visit(child, node.level - 1) : readPayload(child))); } - if (snapshot.roots) { - for (const root of snapshot.roots) { - if (root.end <= start || root.start >= end) continue; - if (root.level === 0) await readPayload(root); - else await visit(root, root.level - 1); - } - } else if (snapshot.root) await visit(snapshot.root); - const bytes = Buffer.concat(chunks); + try { + if (snapshot.roots) { + await Promise.all(snapshot.roots.filter(overlaps).map((root) => + root.level === 0 ? readPayload(root) : visit(root, root.level - 1))); + } else if (snapshot.root) await visit(snapshot.root); + } catch (error) { + failed.abort(error); + throw error; + } + pieces.sort((a, b) => a.start - b.start); + const bytes = Buffer.concat(pieces.map((piece) => piece.bytes)); if (bytes.length !== end - start) throw new Error("object_journal_missing_range"); return bytes; } diff --git a/server/src/services/object-run-log-store-tail-read.test.ts b/server/src/services/object-run-log-store-tail-read.test.ts new file mode 100644 index 00000000000..e2b4c2b5361 --- /dev/null +++ b/server/src/services/object-run-log-store-tail-read.test.ts @@ -0,0 +1,288 @@ +import { createHash } from "node:crypto"; +import { Readable } from "node:stream"; +import { setTimeout as delay } from "node:timers/promises"; +import { describe, expect, it, vi } from "vitest"; +import type { StorageProvider } from "../storage/types.js"; +import { clearObjectJournalReadCache, objectJournalReadCacheBytes } from "./object-byte-journal.js"; +import { createObjectRunLogStore } from "./object-run-log-store.js"; + +// A live run flushes its log in many small segments, so the task page's tail +// read (the last 32 KB, aligned to a record) crosses many objects. Each object +// GET costs a storage round trip, so the number of GETs that must run one after +// another decides how long the chat waits for the run's latest output. +function countingStorage(getLatencyMs: number | ((objectKey: string) => number) = 0) { + const objects = new Map(); + const stats = { gets: 0, inFlight: 0, maxInFlight: 0, completed: 0 }; + const etag = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex"); + const provider: StorageProvider = { + id: "s3", + 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 Object.assign(new Error("conflict"), { name: "PreconditionFailed" }); + const chunks: Buffer[] = []; + if (Buffer.isBuffer(input.body)) chunks.push(input.body); + else for await (const chunk of input.body) chunks.push(Buffer.from(chunk)); + const bytes = Buffer.concat(chunks); + objects.set(input.objectKey, bytes); + return { etag: etag(bytes) }; + }, + async getObject({ objectKey }) { + stats.gets++; + stats.inFlight++; + stats.maxInFlight = Math.max(stats.maxInFlight, stats.inFlight); + try { + await delay(typeof getLatencyMs === "number" ? getLatencyMs : getLatencyMs(objectKey)); + const bytes = objects.get(objectKey); + if (!bytes) throw Object.assign(new Error("missing"), { name: "NoSuchKey" }); + return { stream: Readable.from([bytes]), etag: etag(bytes), contentLength: bytes.length }; + } finally { + stats.inFlight--; + stats.completed++; + } + }, + async headObject({ objectKey }) { + return { exists: objects.has(objectKey) }; + }, + async deleteObject({ objectKey }) { + objects.delete(objectKey); + }, + async listObjects({ prefix }) { + return { keys: [...objects.keys()].filter((key) => key.startsWith(prefix)) }; + }, + }; + return { provider, stats, objects }; +} + +// Holds every content-addressed object GET until released; the head answers. +function holdObjectGets(provider: StorageProvider) { + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const gate = { keys: [] as string[], signals: [] as Array, headReads: 0, release: () => release() }; + const getObject = provider.getObject.bind(provider); + provider.getObject = async (input) => { + if (input.objectKey.includes("/objects/")) { + gate.keys.push(input.objectKey); + gate.signals.push(input.signal); + await held; + } + const result = await getObject(input); + if (input.objectKey.endsWith("/head.json")) gate.headReads++; + return result; + }; + return gate; +} + +const TAIL_BYTES = 32_000; + +async function writeManySegmentLog( + provider: StorageProvider, + records: number, + chunk = (seq: number) => `line ${seq} ${"x".repeat(240)}\n`, +) { + const store = createObjectRunLogStore({ + provider, + keyPrefix: "logs", + assertWriterAuthority: async () => undefined, + }); + const handle = await store.begin({ companyId: "company", agentId: "agent", runId: "run" }); + for (let seq = 1; seq <= records; seq++) { + // One append is one flushed segment, like a live run's frequent flushes. + void store.enqueue!(handle, { + stream: "stdout", + chunk: chunk(seq), + ts: "2026-11-09T00:00:00Z", + seq, + }); + await store.flush!(handle); + } + const full = Buffer.from((await store.read(handle)).content); + return { store, handle, full }; +} + +describe("object run log tail reads", () => { + it("reads an aligned tail across many segments in parallel and without a second traversal", async () => { + clearObjectJournalReadCache(); + const { provider, stats } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 400); + const offset = full.length - TAIL_BYTES; + clearObjectJournalReadCache(); + stats.gets = 0; + stats.maxInFlight = 0; + + const tail = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + + // Correctness: the tail starts at the first complete record before offset. + const recordStart = full.lastIndexOf(10, offset - 1) + 1; + expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart)); + expect(tail.nextOffset).toBe(full.length); + // Every object in the aligned range is fetched once, and fetches overlap. + const singlePassGets = stats.gets; + expect(stats.maxInFlight).toBeGreaterThan(1); + + clearObjectJournalReadCache(); + stats.gets = 0; + const unaligned = await store.read(handle, { offset, limitBytes: TAIL_BYTES }); + expect(Buffer.from(unaligned.content)).toEqual(full.subarray(offset)); + // Alignment adds only a short record prefix, not a second walk of the index. + expect(singlePassGets).toBeLessThanOrEqual(stats.gets + 30); + }); + + it("serves a repeated tail read from immutable objects without fetching them again", async () => { + clearObjectJournalReadCache(); + const { provider, stats } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 200); + const offset = full.length - TAIL_BYTES; + const first = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + stats.gets = 0; + + const second = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + + expect(second).toEqual(first); + // Only the mutable head is read again; segments and index nodes are + // content-addressed by their SHA-256, so they cannot change. + expect(stats.gets).toBe(1); + }); + + it("keeps appended output visible to a tail read after earlier reads were cached", async () => { + clearObjectJournalReadCache(); + const { provider } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 50); + await store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES, alignStartToRecord: true }); + void store.enqueue!(handle, { stream: "stdout", chunk: "newest line\n", ts: "2026-11-09T00:00:01Z", seq: 51 }); + await store.flush!(handle); + + const tail = await store.read(handle, { offset: full.length - 300, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + + expect(tail.content).toContain("newest line"); + }); + + it("stops a reader that navigates away while it waits on another reader's fetch", async () => { + clearObjectJournalReadCache(); + const { provider } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 60); + clearObjectJournalReadCache(); + const gets = holdObjectGets(provider); + const offset = full.length - TAIL_BYTES / 2; + const leaving = new AbortController(); + + const staying = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0)); + const joining = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true, signal: leaving.signal }); + // Let the joining reader read the head and reach the held shared fetch. + await vi.waitFor(() => expect(gets.headReads).toBe(2)); + await new Promise((resolve) => setImmediate(resolve)); + leaving.abort(new Error("navigated away")); + + // Every object GET is still held, so only the reader's own abort can end it. + await expect(joining).rejects.toThrow("navigated away"); + expect(gets.signals.every((signal) => !signal?.aborted)).toBe(true); + gets.release(); + const tail = await staying; + const recordStart = full.lastIndexOf(10, offset - 1) + 1; + expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart)); + expect(new Set(gets.keys).size).toBe(gets.keys.length); + }); + + it("keeps one fetch per object when the reader that started it navigates away", async () => { + clearObjectJournalReadCache(); + const { provider } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 60); + clearObjectJournalReadCache(); + const gets = holdObjectGets(provider); + const offset = full.length - TAIL_BYTES / 2; + const leaving = new AbortController(); + + const first = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true, signal: leaving.signal }); + await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0)); + const joining = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + // Let the joining reader read the head and reach the held shared fetch. + await vi.waitFor(() => expect(gets.headReads).toBe(2)); + await new Promise((resolve) => setImmediate(resolve)); + leaving.abort(new Error("navigated away")); + await expect(first).rejects.toThrow("navigated away"); + // The remaining reader still waits on the started fetch, so it stays open. + expect(gets.signals.every((signal) => !signal?.aborted)).toBe(true); + gets.release(); + + const tail = await joining; + const recordStart = full.lastIndexOf(10, offset - 1) + 1; + expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart)); + // No object was fetched twice: the joining reader did not start its own GET. + expect(new Set(gets.keys).size).toBe(gets.keys.length); + }); + + it("stops a shared fetch when its last reader navigates away", async () => { + clearObjectJournalReadCache(); + const { provider } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 20); + clearObjectJournalReadCache(); + const gets = holdObjectGets(provider); + const leaving = new AbortController(); + + const read = store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES, signal: leaving.signal }); + await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0)); + leaving.abort(new Error("navigated away")); + + await expect(read).rejects.toThrow("navigated away"); + expect(gets.signals[0]?.aborted).toBe(true); + gets.release(); + }); + + it.each([ + ["just over the first lookback window", 6_000], + ["across several backward windows", 90_000], + ])("aligns a tail that starts inside a record %s", async (_label, recordBytes) => { + clearObjectJournalReadCache(); + const { provider } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 6, (seq) => + seq === 5 ? `long ${"y".repeat(recordBytes)}\n` : `line ${seq}\n`); + const longStart = full.indexOf("long "); + const longEnd = full.indexOf(10, longStart); + // Start near the end of the long record, so the 4 KB lookback has no newline. + const offset = longEnd - 100; + expect(offset - longStart).toBeGreaterThan(4096); + const recordStart = full.lastIndexOf(10, offset - 1) + 1; + clearObjectJournalReadCache(); + + const tail = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true }); + + expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart)); + expect(tail.nextOffset).toBe(full.length); + }); + + it("counts per-entry overhead against the cache budget, not only the object bytes", async () => { + clearObjectJournalReadCache(); + const { provider, objects } = countingStorage(); + const { store, handle, full } = await writeManySegmentLog(provider, 100); + clearObjectJournalReadCache(); + const gets = holdObjectGets(provider); + gets.release(); + + await store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES }); + + const cached = [...new Set(gets.keys)]; + const payload = cached.reduce((sum, key) => sum + objects.get(key)!.length, 0); + // Many small segments: the bookkeeping per entry is a real part of memory. + expect(objectJournalReadCacheBytes()).toBeGreaterThanOrEqual(payload + cached.length * 256); + clearObjectJournalReadCache(); + expect(objectJournalReadCacheBytes()).toBe(0); + }); + + it("does not serve one store's cached objects to a reader of another store", async () => { + clearObjectJournalReadCache(); + const first = countingStorage(); + const { handle, full } = await writeManySegmentLog(first.provider, 20); + const offset = Math.max(0, full.length - TAIL_BYTES); + // A second store has the same head, but none of the objects. + const second = countingStorage(); + for (const [key, bytes] of first.objects) if (!key.includes("/objects/")) second.objects.set(key, bytes); + const secondStore = createObjectRunLogStore({ provider: second.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined }); + + await expect(secondStore.read(handle, { offset, limitBytes: TAIL_BYTES })).rejects.toThrow(); + expect(second.stats.gets).toBeGreaterThan(1); + }); +}); diff --git a/server/src/services/object-run-log-store.ts b/server/src/services/object-run-log-store.ts index b7a9135718a..d37f4f700de 100644 --- a/server/src/services/object-run-log-store.ts +++ b/server/src/services/object-run-log-store.ts @@ -8,6 +8,10 @@ import { MAX_RUN_LOG_RECORD_BYTES } from "./run-log-record-limits.js"; const MAX_SEGMENT_BYTES = 1024 * 1024; const MAX_PENDING_BYTES = MAX_RUN_LOG_RECORD_BYTES; const MAX_READ_BYTES = 4 * MAX_SEGMENT_BYTES; +const ALIGN_WINDOW_BYTES = 32_000; +// Most records are short, so the first alignment window rides along with the +// body read. Longer records continue with ALIGN_WINDOW_BYTES windows. +const ALIGN_FIRST_LOOKBACK_BYTES = 4096; export interface ObjectRunLogStoreOptions { provider: StorageProvider; keyPrefix: string; @@ -360,12 +364,20 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL limit = opts.limitBytes ?? 256_000; if (!Number.isSafeInteger(limit) || limit < 1 || limit > MAX_READ_BYTES) throw new Error("Invalid object log read limit"); - const fetched = await store.readRange( + const bodyLimit = Math.min(MAX_READ_BYTES, limit + 3); + // Read the alignment window and the body in one range, so a tail read + // walks the index once instead of once for each part. + const lookback = opts.alignStartToRecord && offset > 0 + ? Math.min(offset, ALIGN_FIRST_LOOKBACK_BYTES, MAX_READ_BYTES - bodyLimit) + : 0; + const window = await store.readRange( snapshot.head, - offset, - Math.min(MAX_READ_BYTES, limit + 3), + offset - lookback, + lookback + bodyLimit, opts.signal, ); + const lookbackBytes = window.subarray(0, lookback); + const fetched = window.subarray(lookback); let end = Math.min(limit, fetched.length); while (end < fetched.length && (fetched[end]! & 0xc0) === 0x80) end++; if ( @@ -386,9 +398,16 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL return { content: bytes.toString("utf8"), nextOffset }; const parts = [bytes]; let before = offset; + if (lookback > 0) { + const boundary = lookbackBytes.lastIndexOf(10); + parts.unshift(lookbackBytes.subarray(boundary + 1)); + before = boundary >= 0 ? 0 : offset - lookback; + if (before > 0 && parts.reduce((n, part) => n + part.length, 0) > MAX_PENDING_BYTES) + throw new Error("Object log record exceeds read alignment limit"); + } while (before > 0) { - const start = Math.max(0, before - 32_000); + const start = Math.max(0, before - ALIGN_WINDOW_BYTES); const prefix = await store.readRange(snapshot.head, start, before - start, opts.signal); const boundary = prefix.lastIndexOf(10); parts.unshift(prefix.subarray(boundary + 1));