Layout F: convert a collection from today's layout

packages/protocol/src/index.ts

@@ -32,6 +32,13 @@ export {
   type TypedLeaf,
 } from './repo/layouts/index.js'
 export { LiveHashes, markRange, recordRoots, type TypedRoot } from './repo/layouts/reclaim.js'
+export {
+  type ConvertBudget,
+  type ConvertProgress,
+  type ConvertTree,
+  convertTree,
+  versionTrees,
+} from './repo/convert.js'
 export {
   fsck,
   type FsckCursor,

packages/protocol/src/repo/convert.ts

@@ -0,0 +1,126 @@
+import { type RecordEntry, recordTree } from '../tree/node.js'
+import { newNodes } from '../tree/read.js'
+/**
+ * Converting a collection from today's layout to layout F (planning: layout-f-pr.md,
+ * "PR 2"). A version's record trees are walked against the version before's; each
+ * leaf it adds is read from its body (today's layout) and the records F's index
+ * doesn't hold yet are written to the set's packs, as stored (an out-of-line
+ * record stays a pointer). Nodes don't change, so neither do roots or version
+ * hashes. A record is written once however many versions hold it, and a walk that
+ * runs again (a retried job) writes nothing twice.
+ */
+import { compareUtf8 } from '../utf8.js'
+import { KeyedLayout, type RecordSet } from './layouts/index.js'
+import type { Repo } from './repo.js'
+import { RepoSource } from './repo.js'
+
+/** One record tree of a version: a set's tree of a type, and the version before's. */
+export interface ConvertTree {
+  set: RecordSet
+  type: string
+  base: string | null
+  root: string | null
+}
+
+export interface ConvertBudget {
+  /** Leaves to read at most before stopping (one at least). */
+  leaves: number
+  /** Stop past this time (ms since the epoch). */
+  until: number
+}
+
+export interface ConvertProgress {
+  /** The last key of the last leaf done when the budget ran out; null when the tree is done. */
+  after: string | null
+  leaves: number
+  records: number
+  /** Records the index already held (an earlier version's, or a job run again). */
+  skipped: number
+}
+
+/**
+ * Write the records of `tree`'s new leaves (from `from`, today's layout) into
+ * `to`'s set (layout F), from `after` on, within `budget`.
+ */
+export async function convertTree(
+  from: Repo,
+  to: Repo,
+  tree: ConvertTree,
+  after: string | null,
+  budget: ConvertBudget,
+): Promise<ConvertProgress> {
+  const store = to.records?.[tree.set]
+  if (!(store instanceof KeyedLayout)) throw new Error('convertTree: the target is not on layout F')
+  if (from.records) throw new Error("convertTree: the source is not on today's layout")
+  const out: ConvertProgress = { after: null, leaves: 0, records: 0, skipped: 0 }
+  if (tree.root === null) return out
+  const source = new RepoSource(recordTree, from)
+  const writer = store.writer()
+  try {
+    for await (const n of newNodes(source, tree.base, tree.root, { after })) {
+      if (n.level !== 0) continue
+      const node = await from.decoded(recordTree, n.hash)
+      if (node.kind !== 'leaf') continue
+      const entries = node.entries as RecordEntry[]
+      const missing = new Set(await store.missing(tree.type, entries))
+      out.skipped += entries.length - missing.size
+      if (missing.size) {
+        // The stored body, streamed (an out-of-line record's pointer stays a pointer).
+        let i = 0
+        for await (const line of from.bodyLineStream({ hash: n.hash, entries }, undefined, {
+          pointers: true,
+        })) {
+          if (missing.has(i)) {
+            writer.add(tree.type, entries[i]!, line)
+            out.records++
+          }
+          if (++i % 1024 === 0) await writer.drain?.()
+        }
+        if (i !== entries.length)
+          throw new Error(
+            `convertTree: leaf ${n.hash} has ${i} lines for ${entries.length} entries`,
+          )
+        await writer.drain?.()
+      }
+      out.leaves++
+      if (out.leaves >= budget.leaves || Date.now() >= budget.until) {
+        out.after = n.lastKey
+        break
+      }
+    }
+  } finally {
+    await writer.finish()
+  }
+  // A leaf past the cursor's last key can't come before it.
+  if (out.after !== null && after !== null && compareUtf8(out.after, after) <= 0)
+    throw new Error('convertTree: the walk went backwards')
+  return out
+}
+
+/**
+ * A version's record trees in a fixed order (public then private, types by slug),
+ * each with the same set's tree of the type in the version before (`prev`, null
+ * for the first): what `convertTree` walks.
+ */
+export async function versionTrees(
+  repo: Repo,
+  version: string,
+  prev: string | null,
+): Promise<ConvertTree[]> {
+  const setsOf = async (hash: string | null) => {
+    if (hash === null)
+      return { public: {}, private: {} } as Record<RecordSet, Record<string, string | null>>
+    const root = await repo.root(hash)
+    const priv = root.private ? await repo.privateSet(root.private) : null
+    const roots = (types: Record<string, { root: string | null }> | undefined) =>
+      Object.fromEntries(Object.entries(types ?? {}).map(([slug, t]) => [slug, t.root]))
+    return { public: roots(root.public.types), private: roots(priv?.types) }
+  }
+  const now = await setsOf(version)
+  const before = await setsOf(prev)
+  const out: ConvertTree[] = []
+  for (const set of ['public', 'private'] as const)
+    for (const type of Object.keys(now[set]).sort(compareUtf8))
+      out.push({ set, type, base: before[set][type] ?? null, root: now[set][type] ?? null })
+  return out
+}

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

@@ -238,6 +238,17 @@ export class KeyedLayout implements RecordStore {
     return keys.map((k) => found.get(k))
   }
 
+  /**
+   * Which of `entries` (of type `type`) the index doesn't hold, by index into
+   * `entries`, from the newest manifest: a conversion writes only these, so a job
+   * that runs again writes nothing twice.
+   */
+  async missing(type: string, entries: readonly RecordEntry[]): Promise<number[]> {
+    if (!entries.length) return []
+    const found = await this.#find(await this.manifests.latest(), type, entries)
+    return found.flatMap((x, i) => (x === undefined ? [i] : []))
+  }
+
   async leafLines(leaf: TypedLeaf): Promise<string[]> {
     if (!leaf.entries.length) return []
     try {

packages/server/src/api/admin.ts

@@ -17,6 +17,8 @@
  *   POST    /api/admin/fsck                      stewards: {collection, fileBytes?} checks its repository
  *   GET     /api/admin/fsck?collection=          stewards: the last check's report
  *   POST    /api/admin/usage/rebuild             stewards: {day} recomputes them from the usage log
+ *   POST    /api/admin/convert                   stewards: {collection} converts it to layout F
+ *   GET     /api/admin/convert?collection=       stewards: the conversion's state
  *
  * A steward is a KF Auth user whose role is 'admin', read fresh on each check.
  */
@@ -27,6 +29,7 @@ import type { AppEnv } from '../app.js'
 import { usageFor } from '../billing/usage.js'
 import { chunks } from '../db/chunks.js'
 import * as schema from '../db/schema.js'
+import { conversionState, startConversion } from '../layouts/convert.js'
 import { forgetDenylist } from '../lib/limits.js'
 import { jsonError } from './access.js'
 
@@ -252,6 +255,27 @@ export function adminRoutes() {
     return c.json({ ok: true, queued: true }, 202)
   })
 
+  app.post('/api/admin/convert', async (c) => {
+    const denied = await stewardOnly(c)
+    if (denied) return denied
+    const b = (await c.req.json().catch(() => null)) as { collection?: unknown } | null
+    const col = await collectionBySlugs(c, b?.collection)
+    if (!col) return jsonError(c, 404, 'Collection not found (send {"collection": "owner/slug"})')
+    if (col.recordLayout !== 'today')
+      return jsonError(c, 409, `Collection is already on layout ${col.recordLayout}`)
+    return c.json(await startConversion(c.var.ports, col.id), 202)
+  })
+
+  app.get('/api/admin/convert', async (c) => {
+    const denied = await stewardOnly(c)
+    if (denied) return denied
+    const col = await collectionBySlugs(c, c.req.query('collection'))
+    if (!col) return jsonError(c, 404, 'Collection not found (?collection=owner/slug)')
+    const state = await conversionState(c.var.ports, col.id)
+    if (!state) return jsonError(c, 404, 'Not converted')
+    return c.json({ ...state, recordLayout: col.recordLayout })
+  })
+
   app.get('/api/admin/fsck', async (c) => {
     const denied = await stewardOnly(c)
     if (denied) return denied

packages/server/src/handlers.ts

@@ -30,6 +30,7 @@ import './files/files.js'
 import { reconcileDue } from './billing/reconcile.js'
 import { cleanupTick } from './cleanup/runs.js'
 import './cleanup/layout-reclaim.js'
+import './layouts/convert.js'
 import * as schema from './db/schema.js'
 import './push/compact.js'
 import './push/parallel.js'
@@ -66,9 +67,17 @@ registerJob('version.published', async (job, ports) => {
   }
 })
 
-/** A job's record set, and the collection's layout F store for it (null: not on F). */
+/**
+ * A job's record set, and the collection's layout F store for it (null: not on F).
+ * `layout: 'f'` opens it on F whatever the collection's layout: a conversion's
+ * merges run while the collection is still on today's (layouts/convert.ts).
+ */
 async function setLayout(ports: Ports, job: JobMessage): Promise<KeyedLayout | null> {
-  const repo = await ports.stores.forCollection(String(job.collectionId))
+  const id = String(job.collectionId)
+  const repo =
+    job.layout === 'f'
+      ? await ports.stores.forCollectionAs(id, 'f')
+      : await ports.stores.forCollection(id)
   const records = repo.records?.[job.set === 'private' ? 'private' : 'public']
   return records instanceof KeyedLayout ? records : null
 }
@@ -86,6 +95,7 @@ registerJob('layout.maintain', async (job, ports) => {
           set: String(job.set ?? 'public'),
           plan: plan.id,
           range: i,
+          ...(job.layout === 'f' ? { layout: 'f' } : {}),
         })),
       ),
   })
@@ -96,6 +106,7 @@ registerJob('layout.maintain', async (job, ports) => {
       type: 'layout.maintain',
       collectionId: String(job.collectionId),
       set: String(job.set ?? 'public'),
+      ...(job.layout === 'f' ? { layout: 'f' } : {}),
     })
 })
 

packages/server/src/jobs.ts

@@ -55,6 +55,8 @@ export const BULK_JOBS: ReadonlySet<string> = new Set([
   // Its reclaim, a key range a job (cleanup/layout-reclaim.ts), and merges too large for one job.
   'layout.reclaim',
   'layout.merge',
+  // Converting a collection from today's layout to F (layouts/convert.ts).
+  'layout.convert',
 ])
 
 export const isBulk = (type: string) => BULK_JOBS.has(type)