Layout F: reads in fewer round trips

packages/protocol/src/repo/build.ts

@@ -47,6 +47,7 @@ import {
   tracksTypes,
   typeFileRefs,
 } from './file-sets.js'
+import { leafChunks } from './layouts/index.js'
 import {
   bodyOfRecord,
   dropRecordBody,
@@ -200,8 +201,7 @@ async function copyRecords(
   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)
+    for (const at of leafChunks(entries)) {
       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?.()

packages/protocol/src/repo/fsck.ts

@@ -23,7 +23,7 @@ import { mergeTree } from '../tree/merge.js'
 import { fileTree, recordTree, type RecordEntry } from '../tree/node.js'
 import { diffTrees, leaves, newNodes } from '../tree/read.js'
 import { verifyTree } from '../tree/verify.js'
-import type { RecordSet } from './layouts/index.js'
+import { leafChunks, type RecordSet } from './layouts/index.js'
 import {
   type PublicKeyInfo,
   readCollectionInfo,
@@ -36,18 +36,15 @@ 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
+ * untrusted). Layout F reads them a few MB at a time (leafChunks): 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)
-  }
+  for (const at of leafChunks(leaf.entries)) await repo.linesAt(leaf, at, set)
 }
 
 export interface FsckOptions {

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

@@ -50,10 +50,18 @@ export const mergeRead = (inputs: number) =>
     MERGE_READ_MIN,
     Math.min(MERGE_READ, Math.floor(MERGE_READ_BUDGET / Math.max(1, inputs))),
   )
-/** Blocks of one pack a read takes in one request, at most (neighbours, gaps of up to two). */
-const READ_BLOCKS = 16
-/** Layers an index read works through at once, newest first. */
-const READ_LAYERS = 4
+/**
+ * Blocks of one pack a read takes in one request, at most (neighbours, gaps of up to
+ * two): a leaf of large records in a request or two, about 1 MB compressed each.
+ */
+const READ_BLOCKS = 64
+/**
+ * Layers an index read works through at once, newest first, of those whose keys
+ * overlap it: the ~10 runs a collection usually has in one round of reads.
+ */
+const READ_LAYERS = 16
+/** Pages of one layer a leaf's read has in flight (a typed leaf's range is a few). */
+const READ_PAGES = 8
 
 export const newId = () => crypto.randomUUID().replace(/-/g, '')
 
@@ -686,12 +694,12 @@ export class LayerReader {
     }
     const lo = pageAt(dir, maybe[0]!)
     const hi = Math.max(lo, pageAt(dir, maybe.at(-1)!))
-    // One page ahead: the next page's read overlaps this one's scan.
-    let ahead = this.#page(layer.id, dir, lo)
+    // READ_PAGES in flight, taken in order: a range of a few pages in one round trip.
+    const ahead: Promise<[string, string][]>[] = []
+    let next = lo
     for (let p = lo; p <= hi; p++) {
-      const page = await ahead
-      if (p < hi) ahead = this.#page(layer.id, dir, p + 1)
-      keep(page)
+      while (next <= hi && next < p + READ_PAGES) ahead.push(this.#page(layer.id, dir, next++))
+      keep(await ahead.shift()!)
     }
     return out
   }
@@ -1207,7 +1215,13 @@ export class LayeredIndex {
   async values(m: Manifest, keys: Iterable<string>): Promise<Map<string, string>> {
     const sorted = [...new Set(keys)].sort(compareUtf8)
     const out = new Map<string, string>()
-    const layers = m.layers.filter(this.filter).sort((a, b) => b.seq - a.seq)
+    if (!sorted.length) return out
+    // Only layers whose keys overlap the read's (from the manifest: no reads).
+    const lo = sorted[0]!
+    const hi = sorted.at(-1)!
+    const layers = m.layers
+      .filter((l) => this.filter(l) && compareUtf8(l.min, hi) <= 0 && compareUtf8(l.max, lo) >= 0)
+      .sort((a, b) => b.seq - a.seq)
     for (let i = 0; i < layers.length && out.size < sorted.length; i += READ_LAYERS) {
       const left = out.size ? sorted.filter((k) => !out.has(k)) : sorted
       const got = await Promise.all(

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

@@ -85,5 +85,33 @@ export type HeldRun = PendingRun
 export const recordKey = (type: string, id: string, hash: string) =>
   `${type}\u0000${id}\u0000${hash}`
 
+/** Record bytes a reader of a leaf takes at once (layout F): a leaf of small records in one read. */
+export const LEAF_CHUNK_BYTES = 4 * 1024 * 1024
+
+/**
+ * A leaf's entries as runs of indexes of at most LEAF_CHUNK_BYTES of records (one at
+ * least): each run is one index lookup, so a leaf of small records is read in one,
+ * and one of large records is never held whole.
+ */
+export function leafChunks(
+  entries: readonly RecordEntry[],
+  maxBytes = LEAF_CHUNK_BYTES,
+): number[][] {
+  const out: number[][] = []
+  let run: number[] = []
+  let bytes = 0
+  entries.forEach((e, i) => {
+    if (run.length && bytes + e.size > maxBytes) {
+      out.push(run)
+      run = []
+      bytes = 0
+    }
+    run.push(i)
+    bytes += e.size
+  })
+  if (run.length) out.push(run)
+  return out
+}
+
 /** The key a record's versions share (`recordKey` without the hash): what a filter holds. */
 export const recordIdKey = (type: string, id: string) => `${type}\u0000${id}`

packages/protocol/src/repo/repo.ts

@@ -45,6 +45,7 @@ import { compareUtf8 } from '../utf8.js'
 import { chunkLines, gunzipChunks, gunzipStream, gunzipText, gzip, splitLines } from './gzip.js'
 import {
   type HeldRun,
+  leafChunks,
   type LayoutName,
   MissingRecordError,
   type Placed,
@@ -497,11 +498,8 @@ export class Repo {
     // The memory cache holds lines resolved.
     const hit = opts.pointers && this.trusted ? undefined : this.lru.get(this.#ck(key))
     if (this.records && !hit) {
-      // A packed layout: 256 records at a time, so a leaf of large records is never held whole.
-      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)
-        yield* await this.linesAt(leaf, at, opts.set)
-      }
+      // A packed layout: a few MB of records at a time, so a leaf of large records is never held whole.
+      for (const at of leafChunks(leaf.entries)) yield* await this.linesAt(leaf, at, opts.set)
       return
     }
     if (hit || !this.trusted) {

packages/protocol/src/repo/sync.ts

@@ -40,6 +40,7 @@ import {
   versionHash,
 } from '../format.js'
 import { gunzipText, gzip, splitLines } from './gzip.js'
+import { leafChunks } from './layouts/index.js'
 import {
   IntegrityError,
   isNodeDelta,
@@ -231,15 +232,13 @@ export async function* treeObjects(
         yield { ...node, resumeKey: n.lastKey }
         continue
       }
-      // Built a member per 256 records, so the leaf is never held raw: as stored, a
+      // Built a member per few MB of records, so the leaf is never held raw: as stored, a
       // body is gzip members, and only its compressed size is held (as a stored one is).
       const store = repo.records[tree.set]
       const leafOf = { hash: n.hash, type: tree.slug!, entries }
       const members: Uint8Array[] = []
-      for (let i = 0; i < entries.length; i += 256) {
-        const at = Array.from({ length: Math.min(256, entries.length - i) }, (_, j) => i + j)
+      for (const at of leafChunks(entries))
         members.push(await gzip((await store.linesAt(leafOf, at)).join('\n') + '\n'))
-      }
       yield { key: keys.body(n.hash), bytes: concatBytes(members), resumeKey: n.lastKey }
       continue
     }

packages/protocol/test/repo/layouts.test.ts

@@ -988,7 +988,8 @@ describe("layout F's reads and merges hold a bounded amount", () => {
     expect(await layout.leafLines({ hash: 'leaf', type: 'T', entries })).toEqual(
       entries.map((_, i) => `line ${i}`),
     )
-    expect(n.most).toBeLessThanOrEqual(4)
+    // Sixteen at a time: never all forty.
+    expect(n.most).toBeLessThanOrEqual(16)
   })
 
   it('plans a reclaim only once merges bring the runs down', async () => {