PaperClipAI/server/src/services/object-byte-journal.ts

import { createHash, randomUUID } from "node:crypto";
import { runSharedStorageWork, sendThroughStorageOutage } from "../storage/storage-unavailable.js";
import { z } from "zod";
import { setTimeout as delay } from "node:timers/promises";
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({
    key: z.string(),
    start: offset,
    end: offset,
    sha256: z.string().regex(/^[a-f0-9]{64}$/),
  })
  .strict();
type Reference = z.infer<typeof referenceSchema>;
const nodeSchema = z
  .object({
    level: z.number().int().min(0).max(MAX_INDEX_LEVEL),
    children: z.array(referenceSchema).min(1).max(64),
  })
  .strict();
type Node = z.infer<typeof nodeSchema>;
const rootSchema = referenceSchema.extend({ level: z.number().int().min(0).max(MAX_INDEX_LEVEL) });
type Root = z.infer<typeof rootSchema>;
const headSchema = z
  .object({
    v: z.literal(1),
    owner: z.string().uuid(),
    generation: offset,
    end: offset,
    sequence: offset.default(0),
    sourceCursor: offset.default(0),
    root: referenceSchema.nullable(),
    roots: z
      .array(rootSchema)
      .max(MAX_INDEX_LEVEL + 1)
      .optional(),
    closed: z.boolean(),
    deleted: z.boolean(),
    cleanupAfter: offset.default(0),
  })
  .strict();
export type ObjectJournalHead = z.infer<typeof headSchema>;
export interface ObjectJournalWriter {
  head: ObjectJournalHead;
  etag: string;
}
export class ObjectJournalConflictError extends Error {
  readonly code = "object_journal_writer_conflict";
  constructor() {
    super("The output journal is owned by another writer");
  }
}
export class ObjectJournalSourcePrefixProofUnavailableError extends Error {
  constructor(readonly writer: ObjectJournalWriter) {
    super("The committed log prefix has no available source proof");
  }
}
class ObjectJournalSourcePrefixProofInvalidError extends Error {
  constructor() { super("object_journal_source_prefix_proof_invalid"); }
}
export function isObjectPreconditionFailure(error: unknown) {
  const e = error as { name?: string; $metadata?: { httpStatusCode?: number } };
  return e?.name === "PreconditionFailed" || e?.$metadata?.httpStatusCode === 412;
}
function isRetryable(error: unknown) {
  const e = error as { name?: string; code?: string; $metadata?: { httpStatusCode?: number } };
  const status = e?.$metadata?.httpStatusCode;
  return (
    e?.name === "ConditionalRequestConflict" ||
    status === 409 ||
    status === 429 ||
    (status !== undefined && status >= 500) ||
    ["TimeoutError", "RequestTimeout", "NetworkingError"].includes(e?.name ?? "") ||
    ["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "EPIPE"].includes(e?.code ?? "")
  );
}
async function retryObjectOperation<T>(operation: () => Promise<T>, signal?: AbortSignal) {
  for (let attempt = 0; ; attempt++) {
    signal?.throwIfAborted();
    try {
      return await operation();
    } catch (error) {
      if (attempt >= 2 || !isRetryable(error) || signal?.aborted) throw error;
      await delay(100 * 2 ** attempt, undefined, { signal });
    }
  }
}
function isMissing(error: unknown) {
  const e = error as { name?: string; status?: number; $metadata?: { httpStatusCode?: number } };
  return (
    e?.name === "NoSuchKey" ||
    e?.name === "NotFound" ||
    e?.status === 404 ||
    e?.$metadata?.httpStatusCode === 404
  );
}
const sha256 = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
interface ConditionalQualification {
  work: Promise<void>;
  controller: AbortController;
  waiters: number;
}
const qualifiedProviders = new WeakSet<StorageProvider>();
const qualifications = new WeakMap<StorageProvider, ConditionalQualification>();
async function qualifyConditionalWrites(
  provider: StorageProvider,
  prefix: string,
  signal?: AbortSignal,
) {
  signal?.throwIfAborted();
  if (qualifiedProviders.has(provider)) return;
  let qualification = qualifications.get(provider);
  if (!qualification || qualification.controller.signal.aborted) {
    const controller = new AbortController();
    const probeSignal = AbortSignal.any([controller.signal, AbortSignal.timeout(15_000)]);
    const current: ConditionalQualification = { controller, waiters: 0, work: Promise.resolve() };
    qualification = current;
    current.work = runSharedStorageWork(async () => {
      const objectKey = `${prefix}/conditional-probe-${randomUUID()}`;
      const body = Buffer.from("paperclip-conditional-probe-v1");
      const put = provider.putObjectConditional!.bind(provider);
      try {
        await retryObjectOperation(
          () =>
            put({
              objectKey,
              body,
              contentLength: body.length,
              contentType: "text/plain",
              ifNoneMatch: "*",
              signal: probeSignal,
            }),
          probeSignal,
        );
        for (const condition of [
          { ifNoneMatch: "*" as const },
          { ifMatch: '"paperclip-impossible-etag"' },
        ]) {
          let rejected = false;
          try {
            await retryObjectOperation(
              () =>
                put({
                  objectKey,
                  body,
                  contentLength: body.length,
                  contentType: "text/plain",
                  ...condition,
                  signal: probeSignal,
                }),
              probeSignal,
            );
          } catch (error) {
            if (!isObjectPreconditionFailure(error)) throw error;
            rejected = true;
          }
          if (!rejected) throw new Error("Object storage does not enforce conditional publication");
        }
      } finally {
        await provider.deleteObject({ objectKey, signal: probeSignal });
      }
      probeSignal.throwIfAborted();
      qualifiedProviders.add(provider);
    }).finally(() => {
      if (qualifications.get(provider) === current) qualifications.delete(provider);
    });
    qualifications.set(provider, current);
    void current.work.catch(() => undefined);
  }
  const pending = qualification;
  pending.waiters++;
  await new Promise<void>((resolve, reject) => {
    let finished = false;
    const release = () => {
      if (finished) return false;
      finished = true;
      signal?.removeEventListener("abort", abort);
      pending.waiters--;
      return true;
    };
    const abort = () => {
      if (!release()) return;
      if (!pending.waiters) {
        pending.controller.abort(signal!.reason);
        void pending.work.then(
          () => reject(signal!.reason),
          () => reject(signal!.reason),
        );
      } else reject(signal!.reason);
    };
    signal?.addEventListener("abort", abort, { once: true });
    if (signal?.aborted) abort();
    pending.work.then(
      () => {
        if (release()) resolve();
      },
      (error) => {
        if (release()) reject(error);
      },
    );
  });
}

export class ObjectJournalBusyError extends Error {
  readonly code = "object_journal_io_busy";
  constructor() {
    super("Object storage is busy; retry shortly");
  }
}
export function isObjectJournalBusyError(error: unknown): error is ObjectJournalBusyError {
  return error instanceof ObjectJournalBusyError;
}

export type ObjectJournalIOPriority = "background" | "interactive";
const MAX_ACTIVE_IO = 8;
const MAX_WAITING_INTERACTIVE_IO = 32;
let activeIO = 0;
const waitingIO: Record<ObjectJournalIOPriority, Array<() => void>> = {
  background: [],
  interactive: [],
};
async function objectIO<T>(
  operation: () => Promise<T>,
  priority: ObjectJournalIOPriority = "background",
  signal?: AbortSignal,
): Promise<T> {
  signal?.throwIfAborted();
  if (activeIO >= MAX_ACTIVE_IO) {
    const queue = waitingIO[priority];
    if (priority === "interactive" && queue.length >= MAX_WAITING_INTERACTIVE_IO)
      throw new ObjectJournalBusyError();
    await new Promise<void>((resolve, reject) => {
      const grant = () => {
        signal?.removeEventListener("abort", abort);
        resolve();
      };
      const abort = () => {
        const index = queue.indexOf(grant);
        if (index >= 0) queue.splice(index, 1);
        reject(signal!.reason);
      };
      queue.push(grant);
      signal?.addEventListener("abort", abort, { once: true });
    });
  } else activeIO++;
  try {
    return await operation();
  } finally {
    const next = waitingIO.background.shift() ?? waitingIO.interactive.shift();
    if (next) next();
    else activeIO--;
  }
}

// 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<string, Buffer>();
let readCacheBytes = 0;
const providerScopes = new WeakMap<StorageProvider, number>();
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<Buffer>;
  controller: AbortController;
  waiters: number;
}
const readsInFlight = new Map<string, SharedObjectRead>();
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 <T>(operation: () => Promise<T>): Promise<T> => {
    if (active >= max) await new Promise<void>((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<T>(operation: Promise<T>, signal?: AbortSignal): Promise<T> {
  if (!signal) return operation;
  signal.throwIfAborted();
  return new Promise<T>((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;
  readPriority?: ObjectJournalIOPriority;

  encode?: (bytes: Buffer) => Promise<Buffer>;
  decode?: (bytes: Buffer) => Promise<Buffer>;
}

export function createObjectByteJournal(options: ObjectByteJournalOptions) {
  const { provider } = options;
  const prefix = options.prefix.replace(/\/+$/, "");
  if (
    !prefix ||
    prefix.length > 512 ||
    prefix.startsWith("/") ||
    prefix.split("/").some((p) => !/^[a-zA-Z0-9._-]+$/.test(p))
  ) {
    throw new Error("invalid_object_journal_prefix");
  }
  if (!provider.putObjectConditional || !provider.listObjects)
    throw new Error("Object storage requires conditional writes and scoped listing");
  const headKey = `${prefix}/head.json`;
  const objectPrefix = `${prefix}/objects/`;
  const readPriority = options.readPriority ?? "background";
  const cacheScope = providerCacheScope(provider);
  const putConditional = (
    input: Parameters<NonNullable<StorageProvider["putObjectConditional"]>>[0],
  ) => objectIO(() => provider.putObjectConditional!(input), "background", input.signal);

  async function getBytes(key: string, limit: number, signal?: AbortSignal) {
    return objectIO(async () => {
      signal?.throwIfAborted();
      signal = signal
        ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
        : AbortSignal.timeout(15_000);
      const read = signal;
      return sendThroughStorageOutage(async () => {
        const result = await retryObjectOperation(
          () => provider.getObject({ objectKey: key, signal: read }),
          read,
        );
        const chunks: Buffer[] = [];
        let size = 0;
        try {
          for await (const chunk of result.stream) {
            const bytes = Buffer.from(chunk);
            size += bytes.length;
            if (size > limit) throw new Error("object_journal_object_too_large");
            chunks.push(bytes);
          }
        } finally {
          result.stream.destroy();
        }
        read?.throwIfAborted();
        return { bytes: Buffer.concat(chunks), etag: result.etag };
      }, read);
    }, readPriority, signal);
  }
  async function head(signal?: AbortSignal): Promise<ObjectJournalWriter | null> {
    let result;
    try {
      result = await getBytes(headKey, 64 * 1024, signal);
    } catch (error) {
      if (isMissing(error)) return null;
      throw error;
    }
    if (!result.etag) throw new Error("object_journal_missing_etag");
    const value = headSchema.parse(JSON.parse(result.bytes.toString("utf8")));
    if (value.roots) {
      let end = 0,
        level = MAX_INDEX_LEVEL + 1;
      if (value.root) throw new Error("object_journal_invalid_head");
      for (const root of value.roots) {
        validateReference(root);
        if (root.start !== end || root.level >= level)
          throw new Error("object_journal_invalid_head");
        end = root.end;
        level = root.level;
      }
      if (end !== value.end) throw new Error("object_journal_invalid_head");
    } else if ((value.root?.end ?? 0) !== value.end || (value.root && value.root.start !== 0)) {
      throw new Error("object_journal_invalid_head");
    }
    return { head: value, etag: result.etag };
  }
  function sourcePrefixProof(snapshot: ObjectJournalHead) {
    // Leave strict head v1 unchanged for older readers/rollback. Ownership and
    // close/claim state may change without changing this immutable byte prefix.
    const fingerprint = sha256(Buffer.from(JSON.stringify({
      end: snapshot.end, sequence: snapshot.sequence, sourceCursor: snapshot.sourceCursor,
      root: snapshot.root, roots: snapshot.roots ?? null,
    })));
    return {
      // Existing scoped object cleanup also removes these unreferenced metadata objects.
      key: `${objectPrefix}source-prefix-proof-v1-${fingerprint}`,
      bytes: Buffer.from(JSON.stringify({ schema: "paperclip.run-log.source-prefix-proof.v1", fingerprint })),
    };
  }
  async function hasVerifiedSourcePrefix(snapshot: ObjectJournalHead, signal?: AbortSignal) {
    const proof = sourcePrefixProof(snapshot);
    let stored: Awaited<ReturnType<typeof getBytes>>;
    try { stored = await getBytes(proof.key, 4096, signal); }
    catch (error) { if (isMissing(error)) return false; throw error; }
    if (!stored.bytes.equals(proof.bytes)) throw new ObjectJournalSourcePrefixProofInvalidError();
    return true;
  }
  async function recordVerifiedSourcePrefix(writer: ObjectJournalWriter, sourceCursor: number, signal?: AbortSignal) {
    const callerSignal = signal;
    signal = signal
      ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
      : AbortSignal.timeout(15_000);
    signal.throwIfAborted();
    if (!Number.isSafeInteger(sourceCursor) || sourceCursor < writer.head.sourceCursor)
      throw new Error("object_journal_invalid_source_prefix_cursor");
    const current = await head(signal);
    if (!current) throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
    if (current.etag !== writer.etag || current.head.closed || current.head.deleted)
      throw new ObjectJournalConflictError();
    // One-time verification can repair legacy max-cursor metadata, never payload bytes.
    if (sourceCursor !== writer.head.sourceCursor)
      writer = await publish({ ...writer.head, sourceCursor }, writer.etag, signal);
    const proof = sourcePrefixProof(writer.head);
    try {
      await retryObjectOperation(() => putConditional({
        objectKey: proof.key, body: proof.bytes, contentLength: proof.bytes.length,
        contentType: "application/json", ifNoneMatch: "*", signal,
      }), signal);
    } catch (error) {
      callerSignal?.throwIfAborted();
      if (error instanceof ObjectJournalConflictError) throw error;
      // A lost reply or an already published identical immutable proof is safe.
      try {
        if (await hasVerifiedSourcePrefix(writer.head, signal)) return writer;
      } catch (readError) {
        callerSignal?.throwIfAborted();
        if (readError instanceof ObjectJournalSourcePrefixProofInvalidError ||
          readError instanceof ObjectJournalConflictError) throw readError;
        if (signal.aborted || isRetryable(readError))
          throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
        throw readError;
      }
      // Only this immutable proof publication is optional. The ownership read
      // and any cursor/head CAS above retain their fatal failure boundary.
      if (signal.aborted || isRetryable(error))
        throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
      throw error;
    }
    return writer;
  }
  async function publish(value: ObjectJournalHead, etag: string | null, signal?: AbortSignal) {
    value = headSchema.parse(value);
    signal = signal
      ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
      : AbortSignal.timeout(15_000);
    const bytes = Buffer.from(JSON.stringify(value));
    return retryObjectOperation(async () => {
      try {
        const result = await putConditional({
          objectKey: headKey,
          body: bytes,
          contentLength: bytes.length,
          contentType: "application/json",
          signal,
          ...(etag ? { ifMatch: etag } : { ifNoneMatch: "*" as const }),
        });
        return { head: value, etag: result.etag };
      } catch (error) {
        if (signal.aborted) throw error;
        const observed = await head(signal);
        if (observed && JSON.stringify(observed.head) === JSON.stringify(value)) return observed;
        if (isObjectPreconditionFailure(error) || (observed?.etag ?? null) !== etag)
          throw new ObjectJournalConflictError();
        throw error;
      }
    }, signal);
  }
  async function claim(input: {
    assertAuthority: () => Promise<void>;
    signal?: AbortSignal;
  }) {
    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<void>;
    signal?: AbortSignal;
  }) {
    if (current?.head.deleted) throw new ObjectJournalConflictError();
    await input.assertAuthority();
    await qualifyConditionalWrites(provider, prefix, input.signal);
    await input.assertAuthority();
    const value: ObjectJournalHead = {
      v: 1,
      owner: randomUUID(),
      generation: (current?.head.generation ?? 0) + 1,
      end: current?.head.end ?? 0,
      sequence: current?.head.sequence ?? 0,
      sourceCursor: current?.head.sourceCursor ?? 0,
      root: current?.head.root ?? null,
      ...(current?.head.roots ? { roots: current.head.roots } : {}),
      closed: false,
      deleted: false,
      cleanupAfter: 0,
    };
    return publish(value, current?.etag ?? null, input.signal);
  }
  function validateReference(ref: Reference) {
    if (
      !ref.key.startsWith(objectPrefix) ||
      ref.key !== `${objectPrefix}${ref.sha256}` ||
      ref.end <= ref.start
    )
      throw new Error("object_journal_invalid_reference");
  }
  async function load(ref: Reference, signal?: AbortSignal) {
    validateReference(ref);
    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 = runSharedStorageWork(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,
    start: number,
    end: number,
    signal?: AbortSignal,
  ): Promise<Reference> {
    if (bytes.length > MAX_OBJECT_BYTES) throw new Error("object_journal_segment_too_large");
    signal = signal
      ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
      : AbortSignal.timeout(15_000);
    const digest = sha256(bytes),
      key = `${objectPrefix}${digest}`;
    const body = options.encode ? await options.encode(bytes) : bytes;
    try {
      await retryObjectOperation(
        () =>
          putConditional({
            objectKey: key,
            body,
            contentLength: body.length,
            contentType: "application/octet-stream",
            ifNoneMatch: "*",
            signal,
          }),
        signal,
      );
    } catch (error) {
      const ref = { key, sha256: digest, start, end };
      try {
        await load(ref, signal);
      } catch {
        throw error;
      }
    }
    return { key, start, end, sha256: digest };
  }
  async function loadNode(ref: Reference, signal?: AbortSignal): Promise<Node> {
    const node = nodeSchema.parse(JSON.parse((await load(ref, signal)).toString("utf8")));
    let next = ref.start;
    for (const child of node.children) {
      validateReference(child);
      if (child.start !== next) throw new Error("object_journal_noncontiguous_index");
      next = child.end;
    }
    if (next !== ref.end) throw new Error("object_journal_invalid_index_range");
    return node;
  }
  const saveNode = (node: Node, signal?: AbortSignal) =>
    save(
      Buffer.from(JSON.stringify(node)),
      node.children[0]!.start,
      node.children.at(-1)!.end,
      signal,
    );
  async function readRange(
    snapshot: ObjectJournalHead,
    start: number,
    limit: number,
    signal?: AbortSignal,
  ): Promise<Buffer> {
    if (snapshot.deleted) throw new Error("object_journal_deleted");
    if (
      !Number.isSafeInteger(start) ||
      start < 0 ||
      !Number.isSafeInteger(limit) ||
      limit < 0 ||
      limit > MAX_OBJECT_BYTES
    )
      throw new Error("invalid_object_journal_range");
    if ((!snapshot.root && !snapshot.roots?.length) || limit === 0 || start >= snapshot.end)
      return Buffer.alloc(0);
    const end = Math.min(snapshot.end, start + limit),
      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 limited(() => load(ref, readSignal));
      if (bytes.length !== ref.end - ref.start)
        throw new Error("object_journal_invalid_segment_length");
      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): Promise<void> {
      const node = await limited(() => loadNode(ref, readSignal));
      if (expectedLevel !== undefined && node.level !== expectedLevel)
        throw new Error("object_journal_invalid_index_level");
      await Promise.all(node.children.filter(overlaps).map((child) =>
        node.level > 0 ? visit(child, node.level - 1) : readPayload(child)));
    }
    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;
  }
  async function append(
    writer: ObjectJournalWriter,
    bytes: Buffer,
    signal?: AbortSignal,
    sequence = writer.head.sequence,
    sourceCursor = writer.head.sourceCursor,
  ) {
    signal = signal
      ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
      : AbortSignal.timeout(15_000);
    if (writer.head.closed || writer.head.deleted) throw new ObjectJournalConflictError();
    if (bytes.length === 0) return writer;
    if (
      bytes.length > 8 * 1024 * 1024 ||
      !Number.isSafeInteger(writer.head.end + bytes.length) ||
      !Number.isSafeInteger(sequence) ||
      sequence < writer.head.sequence ||
      !Number.isSafeInteger(sourceCursor) ||
      sourceCursor < writer.head.sourceCursor
    )
      throw new Error("object_journal_invalid_batch");
    const roots: Root[] = writer.head.roots
      ? [...writer.head.roots]
      : writer.head.root
        ? [{ ...writer.head.root, level: (await loadNode(writer.head.root, signal)).level + 1 }]
        : [];
    let end = writer.head.end;
    for (let start = 0; start < bytes.length; start += 1024 * 1024) {
      const part = bytes.subarray(start, start + 1024 * 1024);
      let root: Root = { ...(await save(part, end, end + part.length, signal)), level: 0 };
      end = root.end;
      while (roots.at(-1)?.level === root.level) {
        const left = roots.pop()!;
        const { level: leftLevel, ...leftRef } = left;
        const { level, ...rightRef } = root;
        if (level >= MAX_INDEX_LEVEL || leftLevel !== level)
          throw new Error("object_journal_index_capacity_exceeded");
        root = {
          ...(await saveNode({ level, children: [leftRef, rightRef] }, signal)),
          level: level + 1,
        };
      }
      roots.push(root);
    }
    return publish(
      { ...writer.head, end, root: null, roots, sequence, sourceCursor },
      writer.etag,
      signal,
    );
  }

  async function close(writer: ObjectJournalWriter, signal?: AbortSignal) {
    return publish({ ...writer.head, closed: true }, writer.etag, signal);
  }
  async function remove(signal?: AbortSignal) {
    let tombstone: ObjectJournalWriter | undefined;
    for (let attempt = 0; attempt < 8; attempt++) {
      const current = await head(signal);
      if (current?.head.deleted) {
        tombstone = current;
        break;
      }
      const value: ObjectJournalHead = current
        ? {
            ...current.head,
            deleted: true,
            closed: true,
            cleanupAfter: current.head.closed ? 0 : Date.now() + 30_000,
          }
        : {
            v: 1,
            sourceCursor: 0,
            owner: randomUUID(),
            generation: 0,
            end: 0,
            sequence: 0,
            root: null,
            deleted: true,
            closed: true,
            cleanupAfter: 0,
          };
      try {
        tombstone = await publish(value, current?.etag ?? null, signal);
        break;
      } catch (error) {
        if (!(error instanceof ObjectJournalConflictError)) throw error;
      }
    }
    if (!tombstone) throw new ObjectJournalConflictError();
    if (tombstone.head.cleanupAfter > Date.now()) throw new Error("object_journal_cleanup_pending");

    let continuationToken: string | undefined;
    do {
      const deadline = () =>
        signal
          ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
          : AbortSignal.timeout(15_000);
      const page = await objectIO(
        () => provider.listObjects!({ prefix: objectPrefix, continuationToken, signal: deadline() }),
        "background",
        signal,
      );
      for (const key of page.keys) {
        if (!key.startsWith(objectPrefix)) throw new Error("object_journal_invalid_listing");
        await objectIO(
          () => provider.deleteObject({ objectKey: key, signal: deadline() }),
          "background",
          signal,
        );
      }
      continuationToken = page.continuationToken;
    } while (continuationToken);
  }
  return { head, claim, append, close, readRange, remove, hasVerifiedSourcePrefix, recordVerifiedSourcePrefix };
}
export type ObjectByteJournal = ReturnType<typeof createObjectByteJournal>;