Layout F: uploads write records straight into packs; runs carry their places

docs/protocol-v2.md

@@ -778,23 +778,25 @@ DELETE <collection>/push/<sid>            abandon the session
 
 **Opening a session.** The request body is a JSON object. Every member is OPTIONAL.
 
-| Member                 | Meaning                                                                                                                                                                                    |
-| ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| `base`                 | The semver of the version the changes are against. If it is not the collection's head, the server MUST answer 409 with `currentVersion`. `null` or absent: the head at the time of opening |
-| `schemas`              | The complete type set, slug → schema (Section 5). It replaces the base's type set; a type omitted is removed with its records. Absent: the base's type set                                 |
-| `metadata`             | Replaces the base's metadata. An object or `null`                                                                                                                                          |
-| `metadata_patch`       | An object whose top-level members are merged into the base's metadata. Ignored if `metadata` is present. With neither, the base's metadata is kept                                         |
-| `files`                | `{"add": [fileHash, …], "remove": [fileHash, …]}`: files to declare or remove in addition to those records reference (Section 9)                                                           |
-| `message`              | A string recorded in the log entry                                                                                                                                                         |
-| `app_id`, `actor_id`   | Strings identifying the publishing application and actor                                                                                                                                   |
-| `strip_unknown_fields` | A boolean; see **Records**                                                                                                                                                                 |
-
-The response is 200 with `{"session_id", "base", "needed_files", "expires_at", "limits"}`:
+| Member                 | Meaning                                                                                                                                                                                                  |
+| ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `base`                 | The semver of the version the changes are against. If it is not the collection's head, the server MUST answer 409 with `currentVersion`. `null` or absent: the head at the time of opening               |
+| `schemas`              | The complete type set, slug → schema (Section 5). It replaces the base's type set; a type omitted is removed with its records. Absent: the base's type set                                               |
+| `metadata`             | Replaces the base's metadata. An object or `null`                                                                                                                                                        |
+| `metadata_patch`       | An object whose top-level members are merged into the base's metadata. Ignored if `metadata` is present. With neither, the base's metadata is kept                                                       |
+| `files`                | `{"add": [fileHash, …], "remove": [fileHash, …]}`: files to declare or remove in addition to those records reference (Section 9)                                                                         |
+| `message`              | A string recorded in the log entry                                                                                                                                                                       |
+| `app_id`, `actor_id`   | Strings identifying the publishing application and actor                                                                                                                                                 |
+| `strip_unknown_fields` | A boolean; see **Records**                                                                                                                                                                               |
+| `in_key_order`         | A boolean: the client sends each records request's lines in key order (type, then id); see **Records**. A server MAY decline it (the response's `in_key_order` says), and MAY then allow larger requests |
+
+The response is 200 with `{"session_id", "base", "needed_files", "expires_at", "in_key_order", "limits"}`:
 
 - `base` is the semver the session is against, or `null` if the collection has no versions;
 - `needed_files` lists the declared files the server does not hold for the collection; the client
   MUST upload them before committing;
 - `expires_at` is when the session expires; it is extended by each records or deletes request;
+- `in_key_order` is whether the server took the session as key-ordered;
 - `limits` are the server's limits, by which a client MUST size its requests:
 
 | Limit                  | Meaning                                                           |
@@ -819,6 +821,10 @@ type's schema (Section 5.1), and its type MUST be in the session's type set.
   identified by its 1-based `line` number among the request's non-empty lines, and MUST store
   nothing from the request. A server MAY list only the first failures, with `totalErrors` giving
   their number.
+- In a session the server took as `in_key_order`, each request's lines MUST come in key order:
+  by type, then id, as UTF-8 bytes (equal keys allowed; the later wins). A line before the one
+  ahead of it fails the request with 422, and nothing from it is stored. Order across requests is
+  the client's to keep; a server MAY report how many requests overlapped earlier ones' key ranges.
 
 **Deletes.** `POST …/deletes` takes NDJSON lines `{"type", "id"}` and answers 200 with
 `{"received": n}`. Each line MUST satisfy the syntax, duplicate key, unsafe integer and lone surrogate rules, and its `id` and `type` the record id and type slug rules (Section 3); the type MUST be in the session's type set. Deleting a (type, id) the base does
@@ -861,7 +867,10 @@ server's storage: a client can upload a whole corpus into one session over many
 days, and commit it as one version. A server keeps the uploads in storage rather than memory, and
 a background commit's status MAY carry `progress` (underlay.org: `runs`, `run_bytes`,
 `runs_compacting`, `phase` of `compacting`, `planning`, `units` or `assembling`, `units` done of
-`total`, and `progress_at`). An open session expires only after `session_idle_seconds` without an
+`total`, and `progress_at`). underlay.org's status also carries `in_key_order` and
+`overlapping_uploads`, the records requests whose key range overlapped an earlier one's: a corpus
+sent in key order, with none overlapping, is stored with each record next to its neighbours, and
+reads back with the fewest requests. An open session expires only after `session_idle_seconds` without an
 upload, so a client that stops can pick the same session up again by its id. Each version rewrites
 the leaves its changes land in, so one large version costs far less than the same records in many
 small ones.
@@ -909,7 +918,8 @@ The reference implementation defines these in `packages/protocol/src/constants.t
 test vectors repeat them.
 
 Server limits (Section 11.4) are not protocol constants. Informative: underlay.org advertises
-`open_bytes` 8 MiB, `batch_bytes` 8 MiB, `batch_lines` 10,000, `session_idle_seconds` 43,200,
+`open_bytes` 8 MiB, `batch_bytes` 8 MiB (64 MiB in an `in_key_order` session), `batch_lines`
+10,000 (50,000 in one), `session_idle_seconds` 43,200,
 `open_sessions` 20 and `file_bytes` 32 MiB; caps log pages at 1,000 entries, manifest pages at
 25,000 records and diff pages at 5,000 records or 4 MiB of records; stores records over 64 KiB
 out of line; and commits in the background when a session uploaded more than 100,000 records or a

packages/protocol/src/index.ts

@@ -20,11 +20,16 @@ export {
   type LiveSet,
   type MergePlan,
   type PackInfo,
+  type PackHooks,
+  type PackRecord,
+  type UploadWriter,
+  type Placed,
   MissingRecordError,
   RECLAIM_MAX_RUNS,
   type ReclaimPlan,
   recordIdKey,
   recordKey,
+  UPLOAD_PACK_FINAL_RAW,
   type RecordSet,
   type RecordSets,
   setHome,

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

@@ -38,6 +38,7 @@ import {
   type SetObject,
   type TreeSummary,
 } from '../format.js'
+import type { Placed } from './layouts/types.js'
 import { outOfLineHash, type Repo, RepoSink, RepoSource } from './repo.js'
 
 export type SetName = 'public' | 'private'
@@ -103,8 +104,16 @@ 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.noOldRefs.has(set)) this.#body(set, type, before.body, -1)
-    if (after) this.#body(set, type, after.body, +1)
+    if (before && !this.noOldRefs.has(set)) this.#entry(set, type, before, -1)
+    if (after) this.#entry(set, type, after, +1)
+  }
+
+  #entry(set: SetName, type: string, e: RecordEntry, by: number) {
+    // Written into a pack at upload: its references were read from it then.
+    const p = e as Placed
+    if (p.body === undefined && p.packed !== undefined) {
+      for (const h of p.fileRefs ?? []) this.add(set, type, h, by)
+    } else this.#body(set, type, e.body, by)
   }
 
   /**

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

@@ -14,8 +14,9 @@ export {
   RECLAIM_MAX_RUNS,
   type ReclaimPlan,
   TooManyRunsError,
+  UPLOAD_PACK_FINAL_RAW,
 } from './keyed.js'
-export { headConfig, type LayerInfo, type PackInfo, pendingConfig } from './kit.js'
+export { headConfig, type LayerInfo, type PackHooks, type PackInfo, pendingConfig } from './kit.js'
 export * from './types.js'
 
 /** A set's home under the collection's: `runs/<collection>/<set>/` holds its files. */

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

@@ -38,6 +38,7 @@ import {
   newId,
   type PackInfo,
   PackReader,
+  type PackHooks,
   PackWriter,
   packDirKey,
   packKey,
@@ -51,10 +52,12 @@ import {
 import { type KeyRange, LiveHashes } from './reclaim.js'
 import {
   type LiveSet,
+  type PackRecord,
   recordKey,
   type RecordStore,
   type RecordWriter,
   type TypedLeaf,
+  type UploadWriter,
 } from './types.js'
 
 /** Index entries a reclaim range job works on: its live set is 8 bytes an entry. */
@@ -130,6 +133,16 @@ const MAX_RUNS = 10
 /** Packs under this many stored bytes, times 8 per tier, are merged eight at a time. */
 const SMALL_PACK = 4 * 1024 * 1024
 const MAX_PACK_TIER = 3
+/**
+ * An upload's pack this large (raw) is listed with the packs no merge takes: a bulk
+ * load's batches (8 MB each) are written once, where they stay. Merging them was
+ * the external sort again, one pack at a time on the set's one maintain lane.
+ * Smaller ones (a small push's) are merged like a commit's.
+ */
+export const UPLOAD_PACK_FINAL_RAW = 1024 * 1024
+/** Commits whose uploads' packs a manifest remembers listing (`Manifest.adopted`). */
+const ADOPTED_KEPT = 1000
+
 /** How long a replaced object stays readable (readers holding an older manifest). */
 export const layoutConfig = {
   graceMs: 60_000,
@@ -277,6 +290,83 @@ export class KeyedLayout implements RecordStore {
     return new KeyedWriter(this, opts)
   }
 
+  /**
+   * A writer of an upload's records into packs of its own, outside the index and the
+   * manifest until a commit names them (adoptPacks). Their places go in the push's
+   * runs, so a commit indexes the records without moving them. Records arriving in
+   * key order (an upload sent sorted) stream straight into whole packs; one below the
+   * last starts a new pack.
+   */
+  uploadWriter(hooks: PackHooks = {}): UploadWriter {
+    const pw = new PackWriter(this.store, { hooks })
+    return {
+      append: (r) => pw.append(recordKey(r.type, r.id, r.hash), r.line),
+      ready: () => pw.ready(),
+      finish: () => pw.finish(),
+    }
+  }
+
+  /** An upload's records into packs, sorted first: their places, in the order given. */
+  async writePacks(
+    records: readonly PackRecord[],
+    hooks: PackHooks = {},
+  ): Promise<{ places: string[]; packs: PackInfo[] }> {
+    const w = this.uploadWriter(hooks)
+    const keys = records.map((r) => recordKey(r.type, r.id, r.hash))
+    const order = keys.map((_, i) => i).sort((a, b) => compareUtf8(keys[a]!, keys[b]!))
+    const places: string[] = Array.from({ length: records.length })
+    let n = 0
+    for (const i of order) {
+      places[i] = w.append(records[i]!)
+      if (++n % 1024 === 0) await w.ready()
+    }
+    return { places, packs: await w.finish() }
+  }
+
+  /**
+   * List packs uploads wrote, once the commit's index names their records (after its
+   * runs are appended: a merge re-points a pack's records in a newer layer, which a
+   * commit's layer folded in later would otherwise override with the old places).
+   * Large ones go in the registry, small ones in the manifest for merges. Once only
+   * a `name`: a retry finds it in `adopted` and lists nothing.
+   */
+  async adoptPacks(name: string, packs: readonly PackInfo[]): Promise<void> {
+    if (!packs.length) return
+    if ((await this.manifests.latest()).adopted?.includes(name)) return
+    const large = (p: PackInfo) => p.raw >= UPLOAD_PACK_FINAL_RAW || KeyedLayout.final(p)
+    // A fixed key: written once even if this stops before the manifest says so.
+    await this.registry.add(packs.filter(large), `upload-${name}`)
+    const small = packs.filter((p) => !large(p))
+    await this.manifests.update((m) => {
+      if (m.adopted?.includes(name)) return null
+      const have = new Set(m.packs.map((p) => p.id))
+      m.packs.push(...small.filter((p) => !have.has(p.id)))
+      m.adopted = [...(m.adopted ?? []), name].slice(-ADOPTED_KEPT)
+      return m
+    })
+  }
+
+  /**
+   * Delete packs an upload wrote for commit `name`, but none the collection lists:
+   * a pack `adoptPacks` named (in the manifest, its retired files, or the registry
+   * object it wrote) goes only by reclaim. The ids deleted.
+   */
+  async dropPacks(name: string, ids: readonly string[]): Promise<string[]> {
+    const m = await this.manifests.latest()
+    const listed = new Set([...m.packs.map((p) => p.id), ...m.retired.map((r) => r.key)])
+    const registered = await this.store.get(this.registry.keyFor(`upload-${name}`))
+    if (registered)
+      for (const p of JSON.parse(await registered.text()) as PackInfo[]) listed.add(p.id)
+    const gone = ids.filter((id) => !listed.has(id) && !listed.has(packKey(id)))
+    for (let i = 0; i < gone.length; i += 16)
+      await Promise.all(
+        gone
+          .slice(i, i + 16)
+          .flatMap((id) => [this.store.delete(packKey(id)), this.store.delete(packDirKey(id))]),
+      )
+    return gone
+  }
+
   /** A commit's runs its units held back (RecordWriter.finish), as one pending object. */
   appendRuns(runs: PendingRun[]): Promise<void> {
     return this.manifests.appendRuns(runs)
@@ -1271,6 +1361,12 @@ class KeyedWriter implements RecordWriter {
     return at
   }
 
+  addPlaced(type: string, entry: RecordEntry, place: string): string {
+    // Counted with `add`'s entries: `drain` writes a run out at `writerRunEntries` either way.
+    this.#entries.push([recordKey(type, entry.key, entry.hash), place])
+    return place
+  }
+
   /**
    * Write a run out once `writerRunEntries` index entries are held: a commit's
    * entries held to its end grew with the push (204k records, a 128 MB isolate

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

@@ -76,6 +76,15 @@ export interface PackInfo {
   raw: number
 }
 
+/**
+ * Called around each pack's write. An upload records a pack before it's written, so
+ * cleanup can find it if the upload dies, and its stats once it is.
+ */
+export interface PackHooks {
+  before?(id: string): Promise<void>
+  after?(pack: PackInfo): Promise<void>
+}
+
 export class PackWriter {
   #id = ''
   #blocks: Promise<Uint8Array>[] = []
@@ -90,7 +99,7 @@ export class PackWriter {
 
   constructor(
     readonly store: Store,
-    readonly opts: { tier?: number } = {},
+    readonly opts: { tier?: number; hooks?: PackHooks } = {},
   ) {}
 
   /** Add a record line under `key`; returns its place. A key below the last starts a new pack. */
@@ -151,15 +160,18 @@ export class PackWriter {
     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 })
-    this.done.push({
+    const pack = {
       id: job.id,
       tier: this.opts.tier ?? 0,
       records: job.records,
       bytes: offsets.at(-1)!,
       raw: job.raw,
-    })
+    }
+    this.done.push(pack)
+    await this.opts.hooks?.after?.(pack)
   }
 
   /** Close the current pack and wait for every write. */
@@ -796,6 +808,11 @@ export interface Manifest {
    * delete failed must never be folded again once its layers are merged away.
    */
   folded?: string[]
+  /**
+   * The latest commits whose uploads' packs were listed (KeyedLayout.adoptPacks): a
+   * retry lists none again, not even one a merge has since taken and swept.
+   */
+  adopted?: string[]
 }
 
 const empty = (): Manifest => ({ seq: 0, layers: [], packs: [], retired: [], stats: {} })
@@ -850,14 +867,22 @@ export class PackRegistry {
     return `manifest/${this.name}/packs/`
   }
 
-  /** List `packs` in a new object; its key (null for none). */
-  async add(packs: readonly PackInfo[]): Promise<string | null> {
+  /**
+   * List `packs` in a new object; its key (null for none). `name` makes the key
+   * fixed, so listing the same packs again (a retried commit) writes nothing.
+   */
+  async add(packs: readonly PackInfo[], name: string = newId()): Promise<string | null> {
     if (!packs.length) return null
-    const key = `${this.#prefix}${newId()}.json`
+    const key = this.keyFor(name)
     await this.store.put(key, JSON.stringify(packs), { ifAbsent: true })
     return key
   }
 
+  /** The key of the registry object `name`. */
+  keyFor(name: string): string {
+    return `${this.#prefix}${name}.json`
+  }
+
   async read(): Promise<{ keys: string[]; packs: PackInfo[] }> {
     const keys: string[] = []
     let cursor: string | undefined
@@ -866,11 +891,13 @@ export class PackRegistry {
       keys.push(...page.keys)
       cursor = page.cursor
     } while (cursor)
-    const packs: PackInfo[] = []
+    // A pack listed twice (a retried commit naming an upload's packs after a reclaim
+    // replaced their object) is one pack.
+    const packs = new Map<string, PackInfo>()
     for (let i = 0; i < keys.length; i += 16)
       for (const o of await Promise.all(keys.slice(i, i + 16).map((k) => this.store.get(k))))
-        if (o) packs.push(...(JSON.parse(await o.text()) as PackInfo[]))
-    return { keys, packs }
+        if (o) for (const p of JSON.parse(await o.text()) as PackInfo[]) packs.set(p.id, p)
+    return { keys, packs: [...packs.values()] }
   }
 
   /** Replace the objects `keys` with one listing `packs`. */