Layout F: fixes from #61's review

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

@@ -194,7 +194,18 @@ export class KeyedLayout implements RecordStore {
   /** Every pack: the manifest's and the registry's (reclaim, forks). */
   async allPacks(): Promise<PackInfo[]> {
     const m = await this.manifests.latest()
-    return [...m.packs, ...(await this.registry.read()).packs]
+    return [...m.packs, ...(await this.#registered(m)).packs]
+  }
+
+  /**
+   * The registry's objects, and its packs `m` doesn't list: a reclaim lists the packs
+   * it will rewrite in the manifest, then takes them out of the registry, so for a
+   * moment a pack can be in both.
+   */
+  async #registered(m: Manifest): Promise<{ keys: string[]; packs: PackInfo[] }> {
+    const r = await this.registry.read()
+    const listed = new Set(m.packs.map((p) => p.id))
+    return { keys: r.keys, packs: r.packs.filter((p) => !listed.has(p.id)) }
   }
 
   async #find(m: Manifest, entries: readonly RecordEntry[]): Promise<(string | undefined)[]> {
@@ -321,14 +332,28 @@ export class KeyedLayout implements RecordStore {
           held ? (JSON.parse(await held.text()) as { token: string; at: number }) : null,
         )
         .catch(() => null)
-      if (v?.token === token) return () => this.store.delete(key)
+      if (v?.token === token) return () => this.#release(key, token)
       // A lease older than a job can run is a job that died holding it.
       if (!v || Date.now() - v.at < MAINTAIN_LEASE_MS) return null
-      await this.store.delete(key)
+      await this.#release(key, v.token)
     }
     return null
   }
 
+  /**
+   * Delete the lease if `token` still holds it: a job that ran past its lease, which
+   * another then took, mustn't release the new holder's. The store has no conditional
+   * delete, so a takeover between this read and the delete still loses it; the window
+   * is one request, not the rest of a job.
+   */
+  async #release(key: string, token: string): Promise<void> {
+    const held = await this.store
+      .get(key)
+      .then(async (o) => (o ? (JSON.parse(await o.text()) as { token: string }).token : null))
+      .catch(() => null)
+    if (held === token) await this.store.delete(key)
+  }
+
   /**
    * Merge the newest runs, while each older one is no larger than all the newer
    * ones together, when there are LAYER_FANOUT of them or more. A merged run
@@ -478,6 +503,8 @@ export class KeyedLayout implements RecordStore {
     unlisted?: string[]
     newLayers?: LayerInfo[]
     newPacks?: PackInfo[]
+    /** Registry packs to list in the manifest too (reclaim's, until their rewrite swaps them out). */
+    relist?: PackInfo[]
     /** New layers keep their sequence (a merged run); else they're the newest (re-pointed records). */
     keepSeq?: boolean
   }): Promise<boolean> {
@@ -504,7 +531,8 @@ export class KeyedLayout implements RecordStore {
       // A pack merge's layer is the newest: its places win over the old ones.
       const seq = topSeq(m) + 1
       for (const l of newLayers) m.layers.push(o.keepSeq ? l : { ...l, seq })
-      m.packs.push(...listed)
+      const have = new Set(m.packs.map((p) => p.id))
+      m.packs.push(...listed, ...(o.relist ?? []).filter((p) => !have.has(p.id)))
       for (const id of oldLayers)
         m.retired.push(
           { key: layerKey(id), at: now },
@@ -726,7 +754,7 @@ export class KeyedLayout implements RecordStore {
   async reclaimPlan(rangeEntries = RECLAIM_RANGE_ENTRIES): Promise<ReclaimPlan> {
     const m = await this.manifests.latest()
     const ranges = await this.#cutRanges(m.layers, rangeEntries)
-    const registry = await this.registry.read()
+    const registry = await this.#registered(m)
     const plan: ReclaimPlan = {
       id: newId(),
       layers: m.layers,
@@ -840,26 +868,32 @@ export class KeyedLayout implements RecordStore {
     // One run across the ranges, at the level its whole size puts it.
     const level = Math.max(plan.level, levelFor(stats.entriesKept))
     for (const f of files) f.level = level
+    const registered: PackInfo[] = []
+    for (const key of plan.registry) {
+      const o = await this.store.get(key)
+      if (o) registered.push(...(JSON.parse(await o.text()) as PackInfo[]))
+    }
     const m = await this.manifests.latest()
     const listed = new Set(m.packs.map((p) => p.id))
+    const rewriting = new Set(rewrite)
     const swapped = await this.#swap({
       oldLayers: plan.layers.map((l) => l.id),
       oldPacks: drop.filter((id) => listed.has(id)),
       unlisted: drop.filter((id) => !listed.has(id)),
       newLayers: files,
+      // Registry packs being rewritten move to the manifest: listed somewhere until
+      // their rewrite swaps them out (else a reclaim planned meanwhile drops their
+      // live records, and a fork or a mirror misses them).
+      relist: registered.filter((p) => rewriting.has(p.id)),
       keepSeq: true,
     })
     if (swapped) {
-      // The registry's objects as one, without the dropped packs or the ones being rewritten
-      // (their rewrite lists what it writes).
+      // The registry's objects as one, without the dropped packs or the ones now listed.
       const gone = new Set([...drop, ...rewrite])
-      const kept: PackInfo[] = []
-      for (const key of plan.registry) {
-        const o = await this.store.get(key)
-        if (o)
-          kept.push(...(JSON.parse(await o.text()) as PackInfo[]).filter((p) => !gone.has(p.id)))
-      }
-      await this.registry.replace(plan.registry, kept)
+      await this.registry.replace(
+        plan.registry,
+        registered.filter((p) => !gone.has(p.id)),
+      )
     }
     await this.#dropPlan(plan)
     await this.manifests.settle()
@@ -883,7 +917,7 @@ export class KeyedLayout implements RecordStore {
    */
   async reclaimRewrite(packIds: readonly string[]): Promise<Record<string, number>> {
     const m = await this.manifests.latest()
-    // Packs the manifest lists must still be there; the others reclaim took out of the registry.
+    // Packs the manifest lists must still be there; the others (an older reclaim's) are in neither.
     const listed = new Set(m.packs.map((p) => p.id))
     const retired = new Set(m.retired.map((r) => r.key))
     const ids = packIds.filter((id) => !retired.has(packKey(id)))
@@ -944,7 +978,7 @@ export class KeyedLayout implements RecordStore {
     if (!(from instanceof HomeStore) || !(to instanceof HomeStore))
       throw new Error('forkInto: F needs both collections in their own prefixes')
     const m = await this.manifests.latest()
-    const registered = (await this.registry.read()).packs
+    const registered = (await this.#registered(m)).packs
     // Each file's home: where the parent read it from (its own parent's), else the parent.
     const mine = await from.inherited()
     const inherited: Inherited = { packs: {}, layers: {} }
@@ -988,7 +1022,7 @@ export class KeyedLayout implements RecordStore {
     const home = this.store
     if (!(home instanceof HomeStore)) throw new Error('mirrorFiles: layout F only')
     const m = await this.manifests.latest()
-    const registry = await this.registry.read()
+    const registry = await this.#registered(m)
     const inh = await home.inherited()
     const at = (h: string | undefined, key: string) => `runs/${h ?? home.home}/${key}`
     const files: { key: string; bytes?: number; optional?: boolean }[] = []
@@ -1018,6 +1052,26 @@ export class KeyedLayout implements RecordStore {
     }
   }
 
+  /**
+   * F: of the files a mirror copied that `mirrorFiles` no longer lists, which it may
+   * delete (`drop`) and which to forget without deleting (`forget`). Another
+   * collection's (a fork's parent's, at its own keys) are forgotten: that collection
+   * may be mirrored at the same location. This set's own stay while a live fork reads
+   * them, as here: that fork's mirror there may list them. The rest are dropped.
+   */
+  async mirrorDroppable(keys: readonly string[]): Promise<{ drop: string[]; forget: string[] }> {
+    const home = this.store
+    if (!(home instanceof HomeStore)) throw new Error('mirrorDroppable: layout F only')
+    const own = `runs/${home.home}/`
+    const held = keys.some((k) => k.startsWith(own)) ? await home.forkHeld() : new Set<string>()
+    const drop: string[] = []
+    const forget: string[] = []
+    for (const k of keys)
+      if (!k.startsWith(own)) forget.push(k)
+      else if (!held.has(fileId(k.slice(own.length)) ?? '')) drop.push(k)
+    return { drop, forget }
+  }
+
   /**
    * F: delete what a deleted collection wrote, but for files a live fork still
    * reads (they go when that fork does); then the same for each collection it

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

@@ -408,6 +408,18 @@ interface LayerDir {
   off: number[]
 }
 
+/** The last page whose first key is ≤ `key` (0 if none is), by binary search. */
+function pageAt(dir: LayerDir, key: string): number {
+  let lo = 0
+  let hi = dir.first.length - 1
+  while (lo < hi) {
+    const mid = (lo + hi + 1) >> 1
+    if (compareUtf8(dir.first[mid]!, key) <= 0) lo = mid
+    else hi = mid - 1
+  }
+  return lo
+}
+
 export class LayerWriter {
   readonly id = newId()
   #pages: Promise<Uint8Array>[] = []
@@ -622,13 +634,7 @@ export class LayerReader {
     const pages = new Set<number>()
     for (const id of maybe) {
       // Pages whose range can hold id's keys: from the last page starting at or before id.
-      let lo = 0
-      let hi = dir.first.length - 1
-      while (lo < hi) {
-        const mid = (lo + hi + 1) >> 1
-        if (compareUtf8(dir.first[mid]!, id) <= 0) lo = mid
-        else hi = mid - 1
-      }
+      const lo = pageAt(dir, id)
       pages.add(lo)
       for (
         let p = lo + 1;
@@ -649,10 +655,8 @@ export class LayerReader {
     if (compareUtf8(layer.max, from) < 0 || compareUtf8(layer.min, to) > 0) return []
     const dir = await this.#dir(layer.id)
     // The last page whose first key is ≤ from, through the last whose first key is ≤ to.
-    let lo = 0
-    while (lo + 1 < dir.first.length && compareUtf8(dir.first[lo + 1]!, from) <= 0) lo++
-    let hi = lo
-    while (hi + 1 < dir.first.length && compareUtf8(dir.first[hi + 1]!, to) <= 0) hi++
+    const lo = pageAt(dir, from)
+    const hi = Math.max(lo, pageAt(dir, to))
     const out: [string, string][] = []
     const pages = await Promise.all(
       Array.from({ length: hi - lo + 1 }, (_, j) => this.#page(layer.id, dir, lo + j)),
@@ -678,10 +682,7 @@ export class LayerReader {
     if (to !== undefined && compareUtf8(layer.min, to) >= 0) return
     const dir = await this.#dir(layer.id)
     const off = dir.off
-    let start = 0
-    if (from !== undefined)
-      while (start + 1 < dir.first.length && compareUtf8(dir.first[start + 1]!, from) <= 0) start++
-    for (let p = start; p < dir.first.length;) {
+    for (let p = from === undefined ? 0 : pageAt(dir, from); p < dir.first.length;) {
       if (to !== undefined && compareUtf8(dir.first[p]!, to) >= 0) return
       let e = p
       while (e + 1 < dir.first.length && off[e + 2]! - off[p]! <= MERGE_READ) e++
@@ -1079,6 +1080,17 @@ export class ManifestStore {
    */
   async appendRuns(runs: PendingRun[]): Promise<void> {
     if (!runs.length) return
+    // A retry (assembly appended them, then stopped before saving its tree) names them
+    // once only: a second object, folded in after the first's layers merged away, would
+    // bring them back as the newest run, of files swept since. A tree's runs go in one
+    // object, all or none, so they were named if the index lists any of their layers
+    // (pending ones included) or retired one, or its files are gone (swept: a unit's
+    // layer files are deleted no other way).
+    const m = await this.latest()
+    const known = new Set([...m.layers.map((l) => layerKey(l.id)), ...m.retired.map((r) => r.key)])
+    if (runs.some((r) => r.layers.some((l) => known.has(layerKey(l.id))))) return
+    const first = runs.find((r) => r.layers.length)?.layers[0]
+    if (first && !(await this.store.head(layerDirKey(first.id)))) return
     await this.#putPending({ runs })
   }
 

packages/protocol/src/repo/repo.ts

@@ -159,6 +159,8 @@ export interface RepoOptions {
 
 /** A stored node that is a delta: `{"b":<base hash>,"p":[tuples put],"d":[keys removed]}`. */
 const DELTA_PREFIX = '{"b":"'
+/** Whether a stored node's JSON is a delta (node deltas), not the node itself. */
+export const isNodeDelta = (stored: string) => stored.startsWith(DELTA_PREFIX)
 /** A delta larger than this share of the full node is stored full instead (a new keyframe). */
 const DELTA_MAX_SHARE = 0.25
 
@@ -401,20 +403,32 @@ export class Repo {
     const lines = await this.#resolve(
       await this.#inSet(set, (r) => r.leafLines(leaf.hash, leaf.entries)),
     )
-    // A packed layout's records come from its packs, not a body the leaf hashes: on an
-    // untrusted location (a mirror read by a third party) each is checked as a body's is.
-    if (!this.trusted)
-      lines.forEach((l, i) => {
-        const e = leaf.entries[i]!
-        if (sha256Hex(l) !== e.hash)
-          throw new IntegrityError(`Leaf ${leaf.hash}: record ${i} fails its hash`)
-        if (utf8ByteLength(l) !== e.size || !l.startsWith(`{"id":${JSON.stringify(e.key)},"type":`))
-          throw new IntegrityError(`Leaf ${leaf.hash}: record ${i} isn't its entry's record`)
-      })
+    this.#checkPacked(leaf, lines)
     this.lru.set(this.#ck(key), { lines, approxBytes: lines.reduce((n, l) => n + l.length * 2, 0) })
     return lines
   }
 
+  /**
+   * A packed layout's records come from its packs, not a body the leaf hashes: on an
+   * untrusted location (a mirror read by a third party) each line read, resolved, is
+   * checked as a body's is, against its entry (`at[j]`, else the j-th).
+   */
+  #checkPacked(
+    leaf: { hash: string; entries: readonly RecordEntry[] },
+    lines: readonly string[],
+    at?: readonly number[],
+  ): void {
+    if (this.trusted) return
+    lines.forEach((l, j) => {
+      const i = at ? at[j]! : j
+      const e = leaf.entries[i]!
+      if (sha256Hex(l) !== e.hash)
+        throw new IntegrityError(`Leaf ${leaf.hash}: record ${i} fails its hash`)
+      if (utf8ByteLength(l) !== e.size || !l.startsWith(`{"id":${JSON.stringify(e.key)},"type":`))
+        throw new IntegrityError(`Leaf ${leaf.hash}: record ${i} isn't its entry's record`)
+    })
+  }
+
   /**
    * Some of a leaf's records (by entry index), resolved. A packed layout reads only
    * those; leaf bodies are read whole (or from memory) and picked from.
@@ -428,7 +442,11 @@ export class Repo {
       const lines = await this.bodyLines(leaf)
       return at.map((i) => lines[i]!)
     }
-    return this.#resolve(await this.#inSet(set, (r) => r.linesAt(leaf.hash, leaf.entries, at)))
+    const lines = await this.#resolve(
+      await this.#inSet(set, (r) => r.linesAt(leaf.hash, leaf.entries, at)),
+    )
+    this.#checkPacked(leaf, lines, at)
+    return lines
   }
 
   /** One entry's record line (resolved): a changed record's old body, in a packed layout. */
@@ -438,8 +456,10 @@ export class Repo {
     set?: RecordSet,
   ): Promise<string> {
     if (!this.records) return (await this.bodyLines(leaf))[i]!
-    const line = await this.#inSet(set, (r) => r.entryLine(leaf.hash, leaf.entries, i))
-    return line.startsWith(REF_PREFIX) ? this.#outOfLine(line) : line
+    const stored = await this.#inSet(set, (r) => r.entryLine(leaf.hash, leaf.entries, i))
+    const line = stored.startsWith(REF_PREFIX) ? await this.#outOfLine(stored) : stored
+    this.#checkPacked(leaf, [line], [i])
+    return line
   }
 
   async #bodyLines(
@@ -863,7 +883,10 @@ export class RepoSink<E> implements TreeSink<E> {
   }
 
   leaf(desc: NodeDesc, json: string, entries: readonly E[], spilled: number): void {
+    // The base is the leaf being rewritten, for this leaf alone: a later one (after
+    // reused nodes, or appended) would only be diffed against an unrelated node.
     const base = this.#base
+    this.#base = undefined
     this.#run(keys.node(desc.hash), () => this.repo.putNode(desc.hash, json, base))
     if (!this.opts.bodyOf) return
     if (this.repo.records) {

packages/protocol/src/repo/sync.ts

@@ -40,7 +40,14 @@ import {
   versionHash,
 } from '../format.js'
 import { gunzipText, gzip, splitLines } from './gzip.js'
-import { IntegrityError, keys, OUT_OF_LINE_BYTES, outOfLineHash, type Repo } from './repo.js'
+import {
+  IntegrityError,
+  isNodeDelta,
+  keys,
+  OUT_OF_LINE_BYTES,
+  outOfLineHash,
+  type Repo,
+} from './repo.js'
 
 /** Which sets a sync moves: `public`, or `all` (public and private). */
 export type SyncSets = 'public' | 'all'
@@ -193,15 +200,14 @@ export async function* treeObjects(
   for await (const n of newNodes(source as NodeSource<unknown>, tree.base, tree.target, {
     after,
   })) {
-    // A receiver checks each node against its hash: one stored as a delta (node
-    // deltas) goes in full. A mirror copies it as stored, with the node it's against.
-    const node = {
-      key: keys.node(n.hash),
-      bytes:
-        opts.bodies && repo.nodeDeltas
-          ? await gzip(await repo.nodeJson(n.hash))
-          : await raw(repo, keys.node(n.hash)),
-    }
+    // A node stored as a delta (node deltas) goes in full: a receiver checks each node
+    // against its hash, and a mirror may not hold the node a delta is against (another
+    // version's, a fork's parent's, or a private leaf a public mirror never gets). The
+    // cost: a mirror stores rewritten leaves in full, a delta being under a quarter of one.
+    let bytes = await raw(repo, keys.node(n.hash))
+    if (n.level === 0 && isNodeDelta(await gunzipText(bytes)))
+      bytes = await gzip(await repo.nodeJson(n.hash))
+    const node = { key: keys.node(n.hash), bytes }
     if (n.level !== 0) {
       yield node
       continue