Layout F: index keys lead with the type; reads and merges bounded

packages/protocol/src/index.ts

@@ -21,13 +21,17 @@ export {
   type MergePlan,
   type PackInfo,
   MissingRecordError,
+  RECLAIM_MAX_RUNS,
   type ReclaimPlan,
+  recordIdKey,
   recordKey,
   type RecordSet,
   type RecordSets,
   setHome,
+  TooManyRunsError,
+  type TypedLeaf,
 } from './repo/layouts/index.js'
-export { LiveHashes, markRange, recordRoots } from './repo/layouts/reclaim.js'
+export { LiveHashes, markRange, recordRoots, type TypedRoot } from './repo/layouts/reclaim.js'
 export {
   fsck,
   type FsckCursor,

packages/protocol/src/repo/build.ts

@@ -167,13 +167,14 @@ async function* overlay(
   }
 }
 
-/** Every entry of a tree (with bodies) as upserts, read from the tree's set. */
+/** Every entry of a type's tree (with bodies) as upserts, read from the tree's set. */
 async function* treeAsUpserts(
   repo: Repo,
   root: string | null,
   set: SetName,
+  type: string,
 ): AsyncGenerator<Change<RecordEntry>> {
-  for await (const e of iterate(new RepoSource(recordTree, repo, { set }), root, {
+  for await (const e of iterate(new RepoSource(recordTree, repo, { set, type }), root, {
     payloads: true,
   })) {
     yield { key: e.key, entry: e }
@@ -186,17 +187,23 @@ async function* treeAsUpserts(
  * (a public mirror copies them as they are). O(type size), as a rebuild is; the
  * old set's copies go when its reclaim finds them unreferenced.
  */
-async function copyRecords(repo: Repo, root: string | null, from: SetName, to: SetName) {
+async function copyRecords(
+  repo: Repo,
+  root: string | null,
+  type: string,
+  from: SetName,
+  to: SetName,
+) {
   if (!repo.records || root === null) return
-  const source = new RepoSource(recordTree, repo, { set: from })
+  const source = new RepoSource(recordTree, repo, { set: from, type })
   const writer = repo.records[to].writer()
   for await (const leaf of leaves(source, root)) {
     const node = await source.node(leaf.hash)
     const entries = (node.kind === 'leaf' ? node.entries : []) as RecordEntry[]
     for (let i = 0; i < entries.length; i += 256) {
       const at = Array.from({ length: Math.min(256, entries.length - i) }, (_, j) => i + j)
-      const lines = await repo.records[from].linesAt(leaf.hash, entries, at)
-      at.forEach((k, j) => writer.add(entries[k]!, lines[j]!))
+      const lines = await repo.records[from].linesAt({ hash: leaf.hash, type, entries }, at)
+      at.forEach((k, j) => writer.add(type, entries[k]!, lines[j]!))
       await writer.drain?.()
     }
   }
@@ -214,11 +221,11 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     ? await repo.privateSet(baseRoot.private)
     : emptySet()
 
-  // A source and a sink a set: layout F reads and writes each set's records apart.
-  const sources = {
-    public: new RepoSource(recordTree, repo, { set: 'public' }),
-    private: new RepoSource(recordTree, repo, { set: 'private' }),
-  }
+  // A source a set and type, and a sink a set (told each tree's type): layout F reads
+  // and writes each set's records apart, and finds them by type.
+  const source = (set: SetName, type: string) => new RepoSource(recordTree, repo, { set, type })
+  // Nodes alone: any source reads them.
+  const nodes = new RepoSource(recordTree, repo)
   const sinks = {
     public: new RepoSink<RecordEntry>(repo, { bodyOf: bodyOfRecord, set: 'public' }),
     private: new RepoSink<RecordEntry>(repo, { bodyOf: bodyOfRecord, set: 'private' }),
@@ -242,7 +249,8 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     baseTree: string | null,
     changes: AsyncIterable<Change<RecordEntry>>,
   ) => {
-    const result = await mergeTree(sources[set], sinks[set], baseTree, changes, {
+    sinks[set].type = type
+    const result = await mergeTree(source(set, type), sinks[set], baseTree, changes, {
       ...builderOpts,
       oldPayloads: !refs.noOldRefs.has(set),
       onChange: (before, after) => {
@@ -280,7 +288,7 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       }
       return
     }
-    for await (const e of treeAsUpserts(repo, tree, from)) {
+    for await (const e of treeAsUpserts(repo, tree, from, slug)) {
       refs.record(from, e.entry, null, slug)
       if (to) refs.record(to, null, e.entry, slug)
     }
@@ -341,17 +349,17 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
         // The tree moves as it is (summary() drops the old schema hash), with its
         // file references; this push's private changes then merge into it.
         await moveRefs('public', 'private', t.slug, pubRoot)
-        await copyRecords(repo, pubRoot, 'public', 'private')
+        await copyRecords(repo, pubRoot, t.slug, 'public', 'private')
         priv = privChanges ? await merge('private', t.slug, pubRoot, privChanges) : summary(pubBase)
       } else if (pubRoot) {
         // Records in both sets: the union is a merge, O(public part).
         priv = await merge(
           'private',
           t.slug,
           privRoot,
-          overlay(treeAsUpserts(repo, pubRoot, 'public'), privChanges ?? asAsync([])),
+          overlay(treeAsUpserts(repo, pubRoot, 'public', t.slug), privChanges ?? asAsync([])),
         )
-        for await (const e of treeAsUpserts(repo, pubRoot, 'public'))
+        for await (const e of treeAsUpserts(repo, pubRoot, 'public', t.slug))
           refs.record('public', e.entry, null, t.slug)
       } else {
         priv = privChanges
@@ -366,7 +374,7 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       // merge into it (trees are canonical, so this is the tree a rebuild makes).
       // Stats count every record of the new public tree as added, as a rebuild did.
       await moveRefs('private', 'public', t.slug, privRoot)
-      await copyRecords(repo, privRoot, 'private', 'public')
+      await copyRecords(repo, privRoot, t.slug, 'private', 'public')
       const before = { ...stats }
       pub = await merge('public', t.slug, privRoot, pubChanges ?? asAsync([]))
       Object.assign(stats, before)
@@ -417,10 +425,9 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
         [basePublic.types[t.slug]?.root ?? null, pub, priv],
         [basePrivate.types[t.slug]?.root ?? null, priv, pub],
       ] as const) {
-        // Nodes alone: either set's source reads them.
-        for await (const d of diffTrees(sources.public, before, after)) {
+        for await (const d of diffTrees(nodes, before, after)) {
           if (d.before || !d.after) continue
-          if (await getEntry(sources.public, other, d.key)) {
+          if (await getEntry(nodes, other, d.key)) {
             throw new Error(
               `Record ${JSON.stringify(d.key)} of type ${t.slug} is in both the public and private sets`,
             )
@@ -443,7 +450,9 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
         ['public', newPublic.types[t.slug]],
         ['private', newPrivate.types[t.slug]],
       ] as const) {
-        for await (const e of iterate(sources[set], tree?.root ?? null, { payloads: true })) {
+        for await (const e of iterate(source(set, t.slug), tree?.root ?? null, {
+          payloads: true,
+        })) {
           const errs = input.validate(t.schema, (JSON.parse(e.body!) as { data: unknown }).data)
           if (errs) {
             errorCount++

packages/protocol/src/repo/file-sets.ts

@@ -309,7 +309,7 @@ export async function rebuildFileRefs(
     const set = sets[name]
     const delta = new FileRefDelta()
     for (const [slug, t] of Object.entries(set.types)) {
-      const source = new RepoSource(recordTree, repo, { set: name })
+      const source = new RepoSource(recordTree, repo, { set: name, type: slug })
       for await (const e of iterate(source, t.root, { payloads: true })) {
         delta.record(name, null, e, slug)
       }

packages/protocol/src/repo/fsck.ts

@@ -32,7 +32,23 @@ import {
   verifyLog,
   verifyLogEntries,
 } from './log.js'
-import { keys, type Repo, RepoSource } from './repo.js'
+import { keys, type Leaf, type Repo, RepoSource } from './repo.js'
+
+/**
+ * Read a leaf's records, each checked against its entry (the repository is opened
+ * untrusted). Layout F reads them a few hundred at a time: a leaf of large
+ * records is never held whole.
+ */
+async function checkLeaf(repo: Repo, leaf: Leaf, set: RecordSet): Promise<void> {
+  if (!repo.records) {
+    await repo.bodyLines(leaf, set)
+    return
+  }
+  for (let i = 0; i < leaf.entries.length; i += 256) {
+    const at = Array.from({ length: Math.min(256, leaf.entries.length - i) }, (_, j) => i + j)
+    await repo.linesAt(leaf, at, set)
+  }
+}
 
 export interface FsckOptions {
   /** Version hashes to check; with a collection id and none given, the log's. */
@@ -149,7 +165,11 @@ export async function fsck(given: Repo, opts: FsckOptions = {}): Promise<FsckRep
         const node = await attempt(`leaf ${leaf.hash}`, () => records.node(leaf.hash))
         if (!node || node.kind !== 'leaf') continue
         await attempt(`body of ${leaf.hash}`, () =>
-          repo.bodyLines({ hash: leaf.hash, entries: node.entries as RecordEntry[] }, rs),
+          checkLeaf(
+            repo,
+            { hash: leaf.hash, type: slug, entries: node.entries as RecordEntry[] },
+            rs,
+          ),
         )
       }
     }
@@ -372,7 +392,7 @@ export async function fsckStep(
         const node = await attempt(`leaf ${n.hash}`, () => records.node(n.hash))
         if (!node || node.kind !== 'leaf') continue
         await attempt(`body of ${n.hash}`, () =>
-          repo.bodyLines({ hash: n.hash, entries: node.entries as RecordEntry[] }, rs),
+          checkLeaf(repo, { hash: n.hash, type: slug, entries: node.entries as RecordEntry[] }, rs),
         )
       }
     }

packages/protocol/src/repo/layouts/index.ts

@@ -11,7 +11,9 @@ export {
   layoutConfig,
   type MergePlan,
   MissingRecordError,
+  RECLAIM_MAX_RUNS,
   type ReclaimPlan,
+  TooManyRunsError,
 } from './keyed.js'
 export { headConfig, type LayerInfo, type PackInfo, pendingConfig } from './kit.js'
 export * from './types.js'