Push any size as one version: paged run indexes, compaction pieces, byte-sized units, streamed leaves, parallel schema and privacy changes, stalled-commit recovery, a bulk Worker

docs/protocol-v2.md

@@ -850,9 +850,22 @@ already stores the file.
   the client uploads the files and commits the same session again. A polling client sees `status`
   `open`, with the refusal in `error`.
 - Committing a session that has already committed answers 201 with the same body.
+- A server MAY let a session whose commit failed on the server's side (`error.statusCode` 500 or 503) be committed again from the same uploads; underlay.org does until shortly before it
+  deletes the session's uploads, about a day after it would have expired. A failure the client can
+  fix only by pushing again (422, 409) answers 409.
 - A client that holds the base SHOULD compute the new version's hash itself and compare it with
   `hash`.
 
+**One version from any number of uploads** (informative). A session has no size limit beyond the
+server's storage: a client can upload a whole corpus into one session over many requests, and
+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
+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.
+
 **Clients without a copy** (informative). A client that keeps no copy of the collection reads the
 base's manifest (Section 11.3.6), compares each of its records with the manifest by (type, id),
 hash and set, uploads the records that are new, changed or moved between sets, and deletes the
@@ -896,10 +909,15 @@ 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` 16 MiB, `batch_lines` 10,000, `session_idle_seconds` 3,600,
-`open_sessions` 20 and `file_bytes` 32 MiB; caps log pages at 1,000 entries and manifest pages at
-25,000 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 schema change revalidates more than 100,000.
+`open_bytes` 8 MiB, `batch_bytes` 8 MiB, `batch_lines` 10,000, `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
+schema change revalidates more than 100,000. A schema change that can't reject a record (dropping
+`required` names, editing annotations) isn't revalidated; any other is revalidated in parallel
+jobs, so a type of any size can change its schema. A privacy change that would rewrite the whole
+type (records already in both sets, or private records pushed as the type turns public) is refused
+past about 2 GiB: push it on its own first.
 
 ## 13. Security considerations
 

packages/protocol/src/index.ts

@@ -15,7 +15,17 @@ export {
   type FsckStepOptions,
 } from './repo/fsck.js'
 export * from './repo/log.js'
-export { gunzip, gunzipMembers, gunzipText, gzip, isGzip, splitLines } from './repo/gzip.js'
+export { schemaRelaxes } from './schema-relax.js'
+export {
+  chunkLines,
+  gunzip,
+  gunzipChunks,
+  gunzipMembers,
+  gunzipText,
+  gzip,
+  isGzip,
+  splitLines,
+} from './repo/gzip.js'
 export { Lru } from './repo/lru.js'
 export {
   type BuildInput,

packages/protocol/src/repo/build.ts

@@ -12,7 +12,8 @@
  * Moving a type between sets, or removing one, moves or drops its file
  * references by their per-type counts (file-sets.ts): O(files it references).
  * On a base whose count trees predate those counts, it reads the type's bodies.
- *   3. types whose schema changed are revalidated (O(type size));
+ *   3. types whose schema changed are revalidated (O(type size)), unless the
+ *      change can't make a valid record invalid (schemaRelaxes);
  *   4. file sets from reference-count deltas (file-sets.ts);
  *   5. the private set object and the root.
  *
@@ -36,6 +37,7 @@ import {
   type TreeSummary,
   type VersionRoot,
 } from '../format.js'
+import { schemaRelaxes } from '../schema-relax.js'
 import {
   applyFileSet,
   FileRefDelta,
@@ -82,13 +84,18 @@ export interface BuildInput {
   /**
    * Trees already merged elsewhere (a parallel commit), per type, with the file
    * reference deltas and change counts that built them. The types' change
-   * sources are then ignored. Only when no type changes privacy or schema and
-   * none is removed.
+   * sources are then ignored, and so is `validate`: a parallel commit
+   * revalidates in its own jobs. A type in `moves` changed privacy by moving its
+   * whole tree to the other set (the serial build's plain moves); its trees
+   * were merged into the moved tree, and its file references move here. For a
+   * type made public, `stats` leave out its public changes: every record of the
+   * new public tree counts as added, as in the serial build.
    */
   prebuilt?: {
     trees: Record<string, { public: TreeSummary; private: TreeSummary }>
     refs: FileRefDelta
     stats: { added: number; removed: number; updated: number }
+    moves?: Record<string, 'toPrivate' | 'toPublic'>
   }
 }
 
@@ -258,6 +265,15 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
     const prebuilt = input.prebuilt?.trees[t.slug]
     if (input.prebuilt) {
       if (!prebuilt) throw new Error(`Parallel commit has no trees for type ${t.slug}`)
+      const move = input.prebuilt.moves?.[t.slug]
+      if (move === 'toPrivate') {
+        await moveRefs('public', 'private', t.slug, pubBase?.root ?? null)
+        recordsChanged = true
+      } else if (move === 'toPublic') {
+        await moveRefs('private', 'public', t.slug, privBase?.root ?? null)
+        stats.added += prebuilt.public.count
+        recordsChanged = true
+      }
       if (isPrivateSchema(t.schema)) {
         newPrivate.types[t.slug] = { schema: t.schemaHash, ...prebuilt.private }
       } else {
@@ -376,10 +392,11 @@ export async function buildVersion(repo: Repo, input: BuildInput): Promise<Build
   // against the old schema). O(type size), inherent to a schema change.
   const errors: { recordId: string; type: string; errors: string[] }[] = []
   let errorCount = 0
-  if (input.validate) {
+  if (input.validate && !input.prebuilt) {
     for (const t of input.types) {
       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 })) {
           const errs = input.validate(t.schema, (JSON.parse(e.body!) as { data: unknown }).data)

packages/protocol/src/repo/gzip.ts

@@ -27,6 +27,8 @@ export const gzip = (data: Uint8Array | string) =>
 
 interface Zlib {
   gunzip(bytes: Uint8Array, done: (err: Error | null, out: Uint8Array) => void): void
+  /** A streaming gunzip (a Node Transform): written once, read as it decompresses. */
+  createGunzip?(): { end(chunk: Uint8Array): void } & AsyncIterable<Uint8Array>
 }
 const zlib = (
   globalThis as { process?: { getBuiltinModule?: (id: 'node:zlib') => Zlib | undefined } }
@@ -86,6 +88,38 @@ export async function gunzipMembers(
 
 export const gunzipText = async (bytes: Uint8Array) => dec.decode(await gunzip(bytes))
 
+/**
+ * A gzip object's decompressed bytes (any number of members) a chunk at a time,
+ * as zlib produces them, so a large body never sits whole in memory. Runtimes
+ * without a streaming zlib get it in one chunk.
+ */
+export async function* gunzipChunks(bytes: Uint8Array): AsyncGenerator<Uint8Array> {
+  const stream = zlib?.createGunzip?.()
+  if (!stream || typeof stream[Symbol.asyncIterator] !== 'function') {
+    yield await gunzip(bytes)
+    return
+  }
+  stream.end(bytes)
+  for await (const chunk of stream) yield chunk
+}
+
+/** NDJSON lines from chunks of UTF-8 bytes (a trailing newline ends the last line). */
+export async function* chunkLines(chunks: AsyncIterable<Uint8Array>): AsyncGenerator<string> {
+  const decoder = new TextDecoder()
+  let rest = ''
+  for await (const chunk of chunks) {
+    const text = rest + decoder.decode(chunk, { stream: true })
+    let start = 0
+    for (let nl = text.indexOf('\n'); nl !== -1; nl = text.indexOf('\n', start)) {
+      yield text.slice(start, nl)
+      start = nl + 1
+    }
+    rest = text.slice(start)
+  }
+  rest += decoder.decode()
+  if (rest.length > 0) yield rest
+}
+
 export const isGzip = (bytes: Uint8Array) =>
   bytes.length >= 2 && bytes[0] === 0x1f && bytes[1] === 0x8b
 

packages/protocol/src/repo/repo.ts

@@ -41,7 +41,7 @@ import {
   versionDigest,
   versionHash,
 } from '../format.js'
-import { gunzipText, gzip, splitLines } from './gzip.js'
+import { chunkLines, gunzipChunks, gunzipText, gzip, splitLines } from './gzip.js'
 import { Lru } from './lru.js'
 import { type Store, type Cache, noCache } from './types.js'
 
@@ -144,8 +144,23 @@ export class Repo {
     return `${this.scope}/${key}`
   }
 
+  /** Reads in progress, by what they read: a second caller waits for the first. */
+  readonly #pending = new Map<string, Promise<unknown>>()
+
+  #shared<T>(key: string, read: () => Promise<T>): Promise<T> {
+    const running = this.#pending.get(key)
+    if (running) return running as Promise<T>
+    const p = read().finally(() => this.#pending.delete(key))
+    this.#pending.set(key, p)
+    return p
+  }
+
   /** An immutable object's bytes: shared cache, then the bucket. `verify` runs before caching. */
-  async #immutable(
+  #immutable(key: string, verify?: (bytes: Uint8Array) => Promise<void>) {
+    return this.#shared(`bytes:${key}`, () => this.#fetch(key, verify))
+  }
+
+  async #fetch(
     key: string,
     verify?: (bytes: Uint8Array) => Promise<void>,
   ): Promise<Uint8Array | null> {
@@ -165,6 +180,10 @@ export class Repo {
     const key = keys.node(hash)
     const hit = this.lru.get(this.#ck(key))
     if (typeof hit === 'string') return hit
+    return this.#shared(`node:${key}`, () => this.#nodeJson(hash, key))
+  }
+
+  async #nodeJson(hash: string, key: string): Promise<string> {
     const bytes = await this.#immutable(key, async (b) => {
       if (sha256Hex(await gunzipText(b)) !== hash)
         throw new IntegrityError(`Node ${hash} fails its hash`)
@@ -203,6 +222,14 @@ export class Repo {
     const key = keys.body(leaf.hash)
     const hit = this.lru.get(this.#ck(key))
     if (hit) return (hit as { lines: string[] }).lines
+    // Two readers of one leaf at once decompress it once.
+    return this.#shared(`lines:${key}`, () => this.#bodyLines(leaf, key))
+  }
+
+  async #bodyLines(
+    leaf: { hash: string; entries: readonly RecordEntry[] },
+    key: string,
+  ): Promise<string[]> {
     const check = async (b: Uint8Array) => {
       const lines = await this.#resolve(splitLines(await gunzipText(b)))
       if (lines.length !== leaf.entries.length)
@@ -228,6 +255,36 @@ export class Repo {
     return lines
   }
 
+  /**
+   * A leaf's body lines one at a time, out-of-line records resolved, without
+   * holding the body decompressed: for rewriting leaves, where one leaf of large
+   * records can be tens of MB. Nothing is put in the memory cache. Untrusted
+   * locations read through bodyLines, which checks every line before use.
+   */
+  async *bodyLineStream(leaf: {
+    hash: string
+    entries: readonly RecordEntry[]
+  }): AsyncGenerator<string> {
+    const key = keys.body(leaf.hash)
+    const hit = this.lru.get(this.#ck(key))
+    if (hit || !this.trusted) {
+      yield* hit ? (hit as { lines: string[] }).lines : await this.bodyLines(leaf)
+      return
+    }
+    const bytes = await this.#immutable(key)
+    if (!bytes) throw new Error(`Missing body for leaf ${leaf.hash}`)
+    let n = 0
+    for await (const line of chunkLines(gunzipChunks(bytes))) {
+      if (++n > leaf.entries.length) break
+      yield line.startsWith(REF_PREFIX) ? await this.#outOfLine(line) : line
+    }
+    if (n !== leaf.entries.length) {
+      throw new IntegrityError(
+        `Body of ${leaf.hash} has ${n} lines for ${leaf.entries.length} entries`,
+      )
+    }
+  }
+
   /**
    * A leaf body's stored bytes (gzip members), as written: for serving a run of
    * leaves without decoding them. Pointer lines (out-of-line records) stay
@@ -370,6 +427,21 @@ export class RepoSource<E> implements NodeSource<E> {
     const lines = await this.repo.bodyLines({ hash, entries })
     return entries.map((e, i) => ({ ...e, body: lines[i]! })) as E[]
   }
+
+  /** A leaf's entries with bodies, one at a time (Repo.bodyLineStream). */
+  async *leafStream(hash: string): AsyncGenerator<E> {
+    const node = await this.node(hash)
+    if (node.kind !== 'leaf') throw new Error(`Node ${hash} is not a leaf`)
+    if (this.spec !== (recordTree as TreeSpec<unknown>)) {
+      yield* node.entries
+      return
+    }
+    const entries = node.entries as RecordEntry[]
+    let i = 0
+    for await (const body of this.repo.bodyLineStream({ hash, entries })) {
+      yield { ...entries[i++]!, body } as E
+    }
+  }
 }
 
 /**

packages/protocol/src/schema-relax.ts

@@ -0,0 +1,82 @@
+/**
+ * Is a schema change one that can't make a valid record invalid? Then a commit
+ * needn't revalidate the type's records, which is O(type size).
+ *
+ * `after` relaxes `before` when the two are the same schema except for:
+ *
+ *   - `required` lists that lose names, where loosening can only accept more
+ *     (not under `not`, `oneOf`, `if`, `definitions` or `$defs`: there a looser
+ *     subschema can reject a record, or be reached from anywhere by `$ref`);
+ *   - annotations: `title`, `description`, `$comment`;
+ *   - the root's `private` flag, which says which set records go in, not what
+ *     they hold.
+ *
+ * A schema with a `$ref` gets only the annotation and `private` allowances: a
+ * reference can reach a subschema from under `not`. Everything else must be
+ * equal (JCS). False means "revalidate", never "invalid".
+ */
+import { jcs } from './jcs.js'
+
+const ANNOTATIONS = new Set(['title', 'description', '$comment'])
+/** Keywords whose value is a map of name → subschema, each in a monotone position. */
+const SCHEMA_MAPS = new Set(['properties', 'patternProperties'])
+/** Keywords whose value is one subschema in a monotone position. */
+const SCHEMA_ONE = new Set([
+  'items',
+  'additionalItems',
+  'additionalProperties',
+  'contains',
+  'propertyNames',
+  'then',
+  'else',
+])
+/** Keywords whose value is a list of subschemas in monotone positions. */
+const SCHEMA_LIST = new Set(['allOf', 'anyOf', 'items'])
+
+const isObject = (v: unknown): v is Record<string, unknown> =>
+  v !== null && typeof v === 'object' && !Array.isArray(v)
+const same = (a: unknown, b: unknown) =>
+  a === undefined || b === undefined ? a === b : jcs(a) === jcs(b)
+
+function hasRef(v: unknown): boolean {
+  if (Array.isArray(v)) return v.some(hasRef)
+  if (!isObject(v)) return false
+  return Object.entries(v).some(([k, x]) => k === '$ref' || hasRef(x))
+}
+
+export function schemaRelaxes(before: unknown, after: unknown): boolean {
+  return relaxes(before, after, true, !hasRef(before) && !hasRef(after))
+}
+
+function relaxes(a: unknown, b: unknown, root: boolean, loosen: boolean): boolean {
+  if (!isObject(a) || !isObject(b)) return same(a, b)
+  for (const k of new Set([...Object.keys(a), ...Object.keys(b)])) {
+    if (ANNOTATIONS.has(k) || (root && k === 'private')) continue
+    const x = a[k]
+    const y = b[k]
+    if (k === 'required' && loosen) {
+      const was = x === undefined ? [] : x
+      const now = y === undefined ? [] : y
+      if (!Array.isArray(was) || !Array.isArray(now)) return false
+      if (!now.every((n) => was.includes(n))) return false
+      continue
+    }
+    if (SCHEMA_MAPS.has(k) && isObject(x) && isObject(y)) {
+      if (!same(Object.keys(x).sort(), Object.keys(y).sort())) return false
+      for (const name of Object.keys(x)) if (!relaxes(x[name], y[name], false, loosen)) return false
+      continue
+    }
+    if (SCHEMA_LIST.has(k) && Array.isArray(x) && Array.isArray(y)) {
+      if (x.length !== y.length) return false
+      for (let i = 0; i < x.length; i++) if (!relaxes(x[i], y[i], false, loosen)) return false
+      continue
+    }
+    if (SCHEMA_ONE.has(k) && isObject(x) && isObject(y)) {
+      if (!relaxes(x, y, false, loosen)) return false
+      continue
+    }
+    // Everything else, including `not`, `oneOf`, `if` and definitions, must be equal.
+    if (!same(x, y)) return false
+  }
+  return true
+}