Record layout F: records in packs per collection and set

packages/protocol/src/index.ts

@@ -6,6 +6,28 @@
 export * from './format.js'
 export * from './repo/types.js'
 export * from './repo/repo.js'
+// Record layouts: today's (leaf bodies) and F (packs per collection and set), record-storage-layout.md.
+export {
+  headConfig,
+  HomeStore,
+  type Inherited,
+  type HeldRun,
+  KeyedLayout,
+  LAYOUTS,
+  type LayerInfo,
+  type LayoutName,
+  layoutConfig,
+  type LiveSet,
+  type MergePlan,
+  type PackInfo,
+  MissingRecordError,
+  type ReclaimPlan,
+  recordKey,
+  type RecordSet,
+  type RecordSets,
+  setHome,
+} from './repo/layouts/index.js'
+export { LiveHashes, markRange, recordRoots } from './repo/layouts/reclaim.js'
 export {
   fsck,
   type FsckCursor,

packages/protocol/src/repo/build.ts

@@ -28,6 +28,7 @@ import {
   getEntry,
   isEmptySet,
   iterate,
+  leaves,
   makeRoot,
   mergeTree,
   type PrivateSetObject,
@@ -166,16 +167,51 @@ async function* overlay(
   }
 }
 
-/** Every entry of a tree (with bodies) as upserts. */
+/** Every entry of a tree (with bodies) as upserts, read from the tree's set. */
 async function* treeAsUpserts(
   repo: Repo,
   root: string | null,
+  set: SetName,
 ): AsyncGenerator<Change<RecordEntry>> {
-  for await (const e of iterate(new RepoSource(recordTree, repo), root, { payloads: true })) {
+  for await (const e of iterate(new RepoSource(recordTree, repo, { set }), root, {
+    payloads: true,
+  })) {
     yield { key: e.key, entry: e }
   }
 }
 
+/** Records one copy writes as a pending run of the set's index before starting another. */
+const COPY_BATCH = 100_000
+
+/**
+ * Layout F: a type's tree moving to the other set takes its records along, copied
+ * as stored into that set's packs, since a set's files hold only its own records
+ * (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) {
+  if (!repo.records || root === null) return
+  const source = new RepoSource(recordTree, repo, { set: from })
+  let writer = repo.records[to].writer()
+  let n = 0
+  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]!))
+      n += at.length
+    }
+    if (n >= COPY_BATCH) {
+      await writer.finish()
+      writer = repo.records[to].writer()
+      n = 0
+    }
+  }
+  await writer.finish()
+}
+
 const summary = (t: TreeSummary | undefined): TreeSummary =>
   t ? { root: t.root, count: t.count, bytes: t.bytes } : { root: null, count: 0, bytes: 0 }
 
@@ -187,8 +223,15 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     ? await repo.privateSet(baseRoot.private)
     : emptySet()
 
-  const source = new RepoSource(recordTree, repo)
-  const sink = new RepoSink<RecordEntry>(repo, { bodyOf: bodyOfRecord })
+  // 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' }),
+  }
+  const sinks = {
+    public: new RepoSink<RecordEntry>(repo, { bodyOf: bodyOfRecord, set: 'public' }),
+    private: new RepoSink<RecordEntry>(repo, { bodyOf: bodyOfRecord, set: 'private' }),
+  }
   const builderOpts = {
     payloadBytes: recordPayloadBytes,
     dropPayload: dropRecordBody,
@@ -208,8 +251,9 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     baseTree: string | null,
     changes: AsyncIterable<Change<RecordEntry>>,
   ) => {
-    const result = await mergeTree(source, sink, baseTree, changes, {
+    const result = await mergeTree(sources[set], sinks[set], baseTree, changes, {
       ...builderOpts,
+      oldPayloads: !refs.noOldRefs.has(set),
       onChange: (before, after) => {
         refs.record(set, before, after, type)
         recordsChanged = true
@@ -229,6 +273,9 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     public: base?.publicRefsRoot ?? null,
     private: base?.privateRefsRoot ?? null,
   }
+  // A set whose base has no file references: old records can't hold any.
+  for (const set of ['public', 'private'] as const)
+    if (refsRoots[set] === null) refs.noOldRefs.add(set)
   const tracks: Record<SetName, boolean> = {
     public: await tracksTypes(repo, refsRoots.public),
     private: await tracksTypes(repo, refsRoots.private),
@@ -242,7 +289,7 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       }
       return
     }
-    for await (const e of treeAsUpserts(repo, tree)) {
+    for await (const e of treeAsUpserts(repo, tree, from)) {
       refs.record(from, e.entry, null, slug)
       if (to) refs.record(to, null, e.entry, slug)
     }
@@ -256,7 +303,7 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       }
     }),
   )
-  sink.written.push(...input.types.map((t) => `schemas/${t.schemaHash}.json`))
+  sinks.public.written.push(...input.types.map((t) => `schemas/${t.schemaHash}.json`))
 
   for (const t of input.types) {
     const pubBase = basePublic.types[t.slug]
@@ -303,16 +350,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')
         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), privChanges ?? asAsync([])),
+          overlay(treeAsUpserts(repo, pubRoot, 'public'), privChanges ?? asAsync([])),
         )
-        for await (const e of treeAsUpserts(repo, pubRoot))
+        for await (const e of treeAsUpserts(repo, pubRoot, 'public'))
           refs.record('public', e.entry, null, t.slug)
       } else {
         priv = privChanges
@@ -327,6 +375,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')
       const before = { ...stats }
       pub = await merge('public', t.slug, privRoot, pubChanges ?? asAsync([]))
       Object.assign(stats, before)
@@ -361,7 +410,8 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       if (entry.count > 0) recordsChanged = true
     }
   }
-  await sink.flush()
+  await sinks.public.flush()
+  await sinks.private.flush()
 
   // Record ids are unique per type across both sets (docs/protocol-v2.md §4). The
   // base holds that, so only ids entering a set can break it: probe the other set
@@ -376,9 +426,10 @@ 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) {
-        for await (const d of diffTrees(source, before, after)) {
+        // Nodes alone: either set's source reads them.
+        for await (const d of diffTrees(sources.public, before, after)) {
           if (d.before || !d.after) continue
-          if (await getEntry(source, other, d.key)) {
+          if (await getEntry(sources.public, other, d.key)) {
             throw new Error(
               `Record ${JSON.stringify(d.key)} of type ${t.slug} is in both the public and private sets`,
             )
@@ -397,8 +448,11 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
       const before = basePublic.types[t.slug] ?? basePrivate.types[t.slug]
       if (!before || before.schema === t.schemaHash) continue
       if (schemaRelaxes(await repo.schema(before.schema), t.schema)) continue
-      for (const tree of [newPublic.types[t.slug], newPrivate.types[t.slug]]) {
-        for await (const e of iterate(source, tree?.root ?? null, { payloads: true })) {
+      for (const [set, tree] of [
+        ['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 })) {
           const errs = input.validate(t.schema, (JSON.parse(e.body!) as { data: unknown }).data)
           if (errs) {
             errorCount++
@@ -452,6 +506,6 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     stats,
     schemaChanged,
     recordsChanged,
-    written: sink.written,
+    written: [...sinks.public.written, ...sinks.private.written],
   }
 }

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

@@ -62,6 +62,11 @@ const typedKey = (type: string, hash: string) => `${type}/${hash}`
  * `resolve`, which must run before `refs` is used.
  */
 export class FileRefDelta {
+  /**
+   * Sets whose base holds no file reference at all: a changed record's old version
+   * can't take one away, so it isn't read (packed layouts skip fetching it).
+   */
+  readonly noOldRefs = new Set<SetName>()
   /** Per set: file hash → change in references. */
   readonly refs: Record<SetName, Map<string, number>> = { public: new Map(), private: new Map() }
   /** Per set: "<type>/<file hash>" → change in that type's references. */
@@ -98,7 +103,7 @@ export class FileRefDelta {
 
   /** Feed one record change of a type in a set (mergeTree's onChange). */
   record(set: SetName, before: RecordEntry | null, after: RecordEntry | null, type: string): void {
-    if (before) this.#body(set, type, before.body, -1)
+    if (before && !this.noOldRefs.has(set)) this.#body(set, type, before.body, -1)
     if (after) this.#body(set, type, after.body, +1)
   }
 
@@ -304,7 +309,8 @@ export async function rebuildFileRefs(
     const set = sets[name]
     const delta = new FileRefDelta()
     for (const [slug, t] of Object.entries(set.types)) {
-      for await (const e of iterate(new RepoSource(recordTree, repo), t.root, { payloads: true })) {
+      const source = new RepoSource(recordTree, repo, { set: name })
+      for await (const e of iterate(source, t.root, { payloads: true })) {
         delta.record(name, null, e, slug)
       }
     }

packages/protocol/src/repo/fsck.ts

@@ -23,6 +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 {
   type PublicKeyInfo,
   readCollectionInfo,
@@ -31,7 +32,7 @@ import {
   verifyLog,
   verifyLogEntries,
 } from './log.js'
-import { keys, openRepo, type Repo, RepoSource } from './repo.js'
+import { keys, type Repo, RepoSource } from './repo.js'
 
 export interface FsckOptions {
   /** Version hashes to check; with a collection id and none given, the log's. */
@@ -63,7 +64,8 @@ export interface FsckReport {
 export async function fsck(given: Repo, opts: FsckOptions = {}): Promise<FsckReport> {
   // Storage itself: no shared cache or memory of earlier reads, and every object
   // hash-checked as if from a location we don't operate.
-  const repo = openRepo(given.blobs, { trusted: false })
+  // Untrusted, so every object is checked against its hash; in the layout it has.
+  const repo = given.reopen({ trusted: false })
   const max = opts.maxErrors ?? 100
   const report: FsckReport = {
     ok: true,
@@ -121,7 +123,7 @@ export async function fsck(given: Repo, opts: FsckOptions = {}): Promise<FsckRep
   const seenFiles = new Set<string>()
   const seenSchemas = new Set<string>()
 
-  const checkSet = async (where: string, set: SetObject) => {
+  const checkSet = async (where: string, set: SetObject, rs: RecordSet) => {
     for (const [slug, t] of Object.entries(set.types)) {
       if (!seenSchemas.has(t.schema)) {
         seenSchemas.add(t.schema)
@@ -147,7 +149,7 @@ 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[] }),
+          repo.bodyLines({ hash: leaf.hash, entries: node.entries as RecordEntry[] }, rs),
         )
       }
     }
@@ -204,15 +206,15 @@ export async function fsck(given: Repo, opts: FsckOptions = {}): Promise<FsckRep
     if (!root) continue
     const pubErr = checkSetObject(root.public)
     if (pubErr) fail(`${where} public ${pubErr}`)
-    else await checkSet(`${where} public`, root.public)
+    else await checkSet(`${where} public`, root.public, 'public')
     if (root.private) {
       // A public-only copy (a mirror of public sets) lacks the private set: not an error.
       if (!(await repo.blobs.head(keys.privateSet(root.private)))) continue
       const priv = await attempt(`${where} private set`, () => repo.privateSet(root.private!))
       if (!priv) continue
       const privErr = checkSetObject(priv, true)
       if (privErr) fail(`${where} private ${privErr}`)
-      else await checkSet(`${where} private`, priv)
+      else await checkSet(`${where} private`, priv, 'private')
     }
   }
   return report
@@ -254,7 +256,8 @@ export async function fsckStep(
   given: Repo,
   opts: FsckStepOptions,
 ): Promise<{ report: FsckReport; cursor: FsckCursor | null }> {
-  const repo = openRepo(given.blobs, { trusted: false })
+  // Untrusted, so every object is checked against its hash; in the layout it has.
+  const repo = given.reopen({ trusted: false })
   const max = opts.maxErrors ?? 100
   const budget = opts.changeBudget ?? 1_000_000
   const report: FsckReport = {
@@ -347,7 +350,7 @@ export async function fsckStep(
     if (n !== size || hasher.hex() !== hash) fail(`file ${hash} fails its hash or size`)
   }
 
-  const checkSet = async (where: string, set: SetObject, base: SetObject | null) => {
+  const checkSet = async (where: string, set: SetObject, base: SetObject | null, rs: RecordSet) => {
     let changes = 0
     for (const [slug, t] of Object.entries(set.types)) {
       const was = base?.types[slug]
@@ -369,7 +372,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[] }),
+          repo.bodyLines({ hash: n.hash, entries: node.entries as RecordEntry[] }, rs),
         )
       }
     }
@@ -418,11 +421,17 @@ export async function fsckStep(
     if (now) {
       const pubErr = checkSetObject(now.root.public)
       if (pubErr) fail(`${where} public ${pubErr}`)
-      else spent += await checkSet(`${where} public`, now.root.public, before?.root.public ?? null)
+      else
+        spent += await checkSet(
+          `${where} public`,
+          now.root.public,
+          before?.root.public ?? null,
+          'public',
+        )
       if (now.priv) {
         const privErr = checkSetObject(now.priv, true)
         if (privErr) fail(`${where} private ${privErr}`)
-        else spent += await checkSet(`${where} private`, now.priv, before?.priv ?? null)
+        else spent += await checkSet(`${where} private`, now.priv, before?.priv ?? null, 'private')
       }
     }
     cursor = { seq, log, versionHash: entry.versionHash }

packages/protocol/src/repo/gzip.ts

@@ -35,6 +35,7 @@ interface Gunzip extends AsyncIterable<Uint8Array> {
   end(chunk?: Uint8Array): void
   destroy(err?: Error): void
   once(event: 'drain' | 'close', listener: () => void): void
+  off(event: 'drain' | 'close', listener: () => void): void
   readonly destroyed: boolean
 }
 const zlib = (
@@ -129,9 +130,16 @@ export async function* gunzipStream(input: AsyncIterable<Uint8Array>): AsyncGene
       for await (const c of input) {
         if (stream.destroyed) return
         if (!stream.write(c))
+          // Whichever comes first; the other listener is removed, or one is left
+          // behind every time the stream pushes back (a leak across a long body).
           await new Promise<void>((resolve) => {
-            stream.once('drain', resolve)
-            stream.once('close', resolve)
+            const done = () => {
+              stream.off('drain', done)
+              stream.off('close', done)
+              resolve()
+            }
+            stream.once('drain', done)
+            stream.once('close', done)
           })
       }
       stream.end()