Layout F: packs never span types (#68, which merged into upload-packs, onto main)

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

@@ -540,28 +540,36 @@ export class KeyedLayout implements RecordStore {
     })
   }
 
-  /** Merge eight small packs of a tier into one, re-pointing their records in a new layer. */
+  /**
+   * Merge eight small packs of a tier and type into one, re-pointing their records in
+   * a new layer. A merge takes one type's packs, so its output holds one type and a
+   * type's records gather in fewer packs (a leaf read of one type at three years of
+   * Crossref dailies, simulated: 72 GETs at the median where mixing types took 255).
+   * Packs from before types were recorded merge among themselves.
+   */
   async #mergePacks(): Promise<boolean> {
     const m = await this.manifests.latest()
-    const tiers = new Map<number, PackInfo[]>()
+    const queues = new Map<string, { tier: number; packs: PackInfo[] }>()
     for (const p of m.packs) {
       if (KeyedLayout.final(p)) continue
-      tiers.set(p.tier, [...(tiers.get(p.tier) ?? []), p])
+      const q = `${p.tier}\u0000${p.type ?? ''}`
+      const queue = queues.get(q) ?? { tier: p.tier, packs: [] }
+      queue.packs.push(p)
+      queues.set(q, queue)
     }
-    // The tier with the most packs: the lowest that has enough never runs out while
-    // commits land, so higher tiers never merged (round 14 at 20,000 collections:
-    // 718 tier-1 packs listed and rising).
-    const [tier, most] = [...tiers].reduce<[number, number]>(
-      (best, [t, ps]) => (ps.length > best[1] ? [t, ps.length] : best),
-      [-1, 0],
-    )
-    if (most < FANOUT) return false
-    // As many of the tier's packs as fit MERGE_PACKS_BYTES stored (at least FANOUT): many
+    // The queue with the most packs: the lowest tier that has enough never runs out
+    // while commits land, so higher tiers never merged (round 14 at 20,000
+    // collections: 718 tier-1 packs listed and rising).
+    let queue: { tier: number; packs: PackInfo[] } | undefined
+    for (const q of queues.values()) if (q.packs.length > (queue?.packs.length ?? 0)) queue = q
+    if (!queue || queue.packs.length < FANOUT) return false
+    const tier = queue.tier
+    // As many of the queue's packs as fit MERGE_PACKS_BYTES stored (at least FANOUT): many
     // collections' tiny commits each write a tiny pack, and eight at a time fell behind
     // (6,639 waiting at 15,000 collections, every commit reading them all).
     const inputs: PackInfo[] = []
     let stored = 0
-    for (const p of tiers.get(tier)!) {
+    for (const p of queue.packs) {
       if (
         inputs.length >= FANOUT &&
         (stored + p.bytes > MERGE_PACKS_BYTES || inputs.length >= layoutConfig.mergeInputs)

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

@@ -82,6 +82,17 @@ export interface PackInfo {
   /** Stored (compressed) bytes, and raw. */
   bytes: number
   raw: number
+  /**
+   * The record type the pack holds: a pack closes where the type changes, so no pack
+   * spans types and merges take one type's packs. Absent on packs written before.
+   */
+  type?: string
+}
+
+/** The record type a key leads with (recordKey); '' for a key without one. */
+const typeOf = (key: string) => {
+  const i = key.indexOf('\u0000')
+  return i < 0 ? '' : key.slice(0, i)
 }
 
 /**
@@ -101,6 +112,7 @@ export class PackWriter {
   #packRaw = 0
   #records = 0
   #lastKey: string | null = null
+  #type = ''
   readonly done: PackInfo[] = []
   readonly #writing = new Set<Promise<void>>()
   #error: unknown = null
@@ -110,9 +122,15 @@ export class PackWriter {
     readonly opts: { tier?: number; hooks?: PackHooks } = {},
   ) {}
 
-  /** Add a record line under `key`; returns its place. A key below the last starts a new pack. */
+  /**
+   * Add a record line under `key`; returns its place. A key below the last, or of
+   * another type, starts a new pack.
+   */
   append(key: string, line: string): string {
-    if (this.#lastKey !== null && compareUtf8(key, this.#lastKey) < 0) this.#close()
+    const type = typeOf(key)
+    if (this.#lastKey !== null && (type !== this.#type || compareUtf8(key, this.#lastKey) < 0))
+      this.#close()
+    this.#type = type
     this.#lastKey = key
     if (!this.#id) this.#id = newId()
     this.#lines.push(`${key}\t${line}`)
@@ -151,7 +169,13 @@ export class PackWriter {
     this.#seal()
     this.#lastKey = null
     if (!this.#blocks.length) return
-    const job = { id: this.#id, blocks: this.#blocks, records: this.#records, raw: this.#packRaw }
+    const job = {
+      id: this.#id,
+      blocks: this.#blocks,
+      records: this.#records,
+      raw: this.#packRaw,
+      type: this.#type,
+    }
     this.#id = ''
     this.#blocks = []
     this.#records = 0
@@ -164,19 +188,26 @@ export class PackWriter {
     this.#writing.add(p)
   }
 
-  async #write(job: { id: string; blocks: Promise<Uint8Array>[]; records: number; raw: number }) {
+  async #write(job: {
+    id: string
+    blocks: Promise<Uint8Array>[]
+    records: number
+    raw: number
+    type: string
+  }) {
     const parts = await Promise.all(job.blocks)
     const offsets = [0]
     for (const p of parts) offsets.push(offsets.at(-1)! + p.byteLength)
     await this.opts.hooks?.before?.(job.id)
     await putJoined(this.store, packKey(job.id), parts, { ifAbsent: true })
     await this.store.put(packDirKey(job.id), JSON.stringify(offsets), { ifAbsent: true })
-    const pack = {
+    const pack: PackInfo = {
       id: job.id,
       tier: this.opts.tier ?? 0,
       records: job.records,
       bytes: offsets.at(-1)!,
       raw: job.raw,
+      ...(job.type ? { type: job.type } : {}),
     }
     this.done.push(pack)
     await this.opts.hooks?.after?.(pack)

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

@@ -392,6 +392,58 @@ describe('pack merges', () => {
   }, 30_000)
 })
 
+describe('packs never span types', () => {
+  it('close where the type changes, and say which type they hold', async () => {
+    const layout = fLayout()
+    const w = layout.uploadWriter()
+    for (const type of ['A', 'B', 'C'])
+      for (let i = 0; i < 5; i++)
+        w.append({ type, id: ids[i]!, hash: `h${i}`, line: `${type}${i}` })
+    // Packs are reported as their writes finish, in any order.
+    const packs = await w.finish()
+    expect(packs.map((p) => `${p.type}: ${p.records}`).sort()).toEqual(['A: 5', 'B: 5', 'C: 5'])
+  })
+
+  it("merge one type's packs at a time, keeping every record readable", async () => {
+    const saved = { ...layoutConfig }
+    layoutConfig.graceMs = 0
+    // Eight packs a merge: eight of one type make a pack of 32 records, where eight of
+    // both types would make two of 16 and tier up half-full.
+    layoutConfig.mergeInputs = 8
+    try {
+      const layout = fLayout()
+      // Tiny commits, each of two types: every commit writes a pack of each.
+      const types = ['A', 'B']
+      const rec = (i: number) => ({ key: ids[i]!, hash: `h${i}`, size: 10 })
+      for (let c = 0; c < 16; c++) {
+        const w = layout.writer()
+        for (const type of types)
+          for (let i = c * 4; i < c * 4 + 4; i++) w.add(type, rec(i), `${type} ${i}`)
+        await w.finish()
+      }
+      const before = (await layout.manifests.latest()).packs
+      expect(before).toHaveLength(32)
+      expect(new Set(before.map((p) => p.type))).toEqual(new Set(types))
+      while (await layout.maintain());
+      const after = (await layout.manifests.latest()).packs
+      expect(after.map((p) => `${p.type} tier ${p.tier}: ${p.records}`).sort()).toEqual([
+        'A tier 1: 32',
+        'A tier 1: 32',
+        'B tier 1: 32',
+        'B tier 1: 32',
+      ])
+      for (const type of types) {
+        const entries = Array.from({ length: 64 }, (_, i) => rec(i))
+        expect(await layout.leafLines({ hash: 'leaf', type, entries })).toEqual(
+          entries.map((_, i) => `${type} ${i}`),
+        )
+      }
+    } finally {
+      Object.assign(layoutConfig, saved)
+    }
+  })
+})
+
 describe('layout F', () => {
   const rec = (i: number, v = 1) => ({ key: ids[i]!, hash: `h${v}.${i}`, size: 10 })
   const line = (i: number, v = 1) => `line ${v}.${i}`

packages/server/drizzle/0027_push_pack_types.sql

@@ -0,0 +1 @@
+ALTER TABLE `push_packs` ADD `record_type` text;