From 21f480daecc1bbea1e5770aabb8840d9dc57266f Mon Sep 17 00:00:00 2001 From: Travis Rich Date: Tue, 20 Oct 2026 13:33:03 -0400 Subject: [PATCH 1/2] Layout F: a commit's writer writes a run every 20,000 entries A commit's index entries were held until it finished, about 0.33 KB a record: a 204k-record push ran the bulk Worker out of memory on staging. The writer now writes a self-contained run (closed packs, an index layer) whenever drain finds writerRunEntries held, named as it goes or, held, returned for the commit to name. copyRecords drains the same way. memory.ts takes --layout. --- packages/protocol/src/repo/build.ts | 13 +------ packages/protocol/src/repo/layouts/keyed.ts | 42 ++++++++++++++++----- packages/protocol/src/repo/layouts/types.ts | 12 +++--- packages/protocol/src/repo/repo.ts | 8 ++-- packages/protocol/test/repo/layouts.test.ts | 40 +++++++++++++++++++- packages/server/scripts/memory.ts | 10 ++++- 6 files changed, 93 insertions(+), 32 deletions(-) diff --git a/packages/protocol/src/repo/build.ts b/packages/protocol/src/repo/build.ts index d2ceeb7..5208cb2 100644 --- a/packages/protocol/src/repo/build.ts +++ b/packages/protocol/src/repo/build.ts @@ -180,9 +180,6 @@ async function* treeAsUpserts( } } -/** Records one copy writes as a pending run of the set's index before starting another. */ -const COPY_BATCH = 100_000 - /** * Layout F: a type's tree moving to the other set takes its records along, copied * as stored into that set's packs, since a set's files hold only its own records @@ -192,8 +189,7 @@ const COPY_BATCH = 100_000 async function copyRecords(repo: Repo, root: string | null, from: SetName, to: SetName) { if (!repo.records || root === null) return const source = new RepoSource(recordTree, repo, { set: from }) - let writer = repo.records[to].writer() - let n = 0 + const writer = repo.records[to].writer() for await (const leaf of leaves(source, root)) { const node = await source.node(leaf.hash) const entries = (node.kind === 'leaf' ? node.entries : []) as RecordEntry[] @@ -201,12 +197,7 @@ async function copyRecords(repo: Repo, root: string | null, from: SetName, to: S const at = Array.from({ length: Math.min(256, entries.length - i) }, (_, j) => i + j) const lines = await repo.records[from].linesAt(leaf.hash, entries, at) at.forEach((k, j) => writer.add(entries[k]!, lines[j]!)) - n += at.length - } - if (n >= COPY_BATCH) { - await writer.finish() - writer = repo.records[to].writer() - n = 0 + await writer.drain?.() } } await writer.finish() diff --git a/packages/protocol/src/repo/layouts/keyed.ts b/packages/protocol/src/repo/layouts/keyed.ts index ff47313..d6db77e 100644 --- a/packages/protocol/src/repo/layouts/keyed.ts +++ b/packages/protocol/src/repo/layouts/keyed.ts @@ -127,6 +127,8 @@ export const layoutConfig = { * one job ~10,460). */ maintainCalls: 6_000, + /** Index entries a commit's writer holds before writing them out as a run (tests set it low). */ + writerRunEntries: 20_000, /** * Index runs, or packs, one merge takes at most: each is two reads, so a pass's * calls stay well inside what `maintainCalls` leaves (tests set it low). @@ -268,8 +270,8 @@ export class KeyedLayout implements RecordStore { return (await this.leafLines(leafHash, [entries[i]!]))[0]! } - writer(): RecordWriter { - return new KeyedWriter(this) + writer(opts: { hold?: boolean } = {}): RecordWriter { + return new KeyedWriter(this, opts) } /** A commit's runs its units held back (RecordWriter.finish), as one pending object. */ @@ -1215,10 +1217,15 @@ export class HomeStore extends PrefixedStore { } class KeyedWriter implements RecordWriter { - readonly #packs: PackWriter + #packs: PackWriter #entries: [string, string][] = [] + /** Runs written and held back from the manifest (`hold`), in order. */ + readonly #held: PendingRun[] = [] - constructor(readonly layout: KeyedLayout) { + constructor( + readonly layout: KeyedLayout, + readonly opts: { hold?: boolean } = {}, + ) { this.#packs = new PackWriter(layout.store) } @@ -1229,9 +1236,25 @@ class KeyedWriter implements RecordWriter { return at } - async finish(opts: { hold?: boolean } = {}): Promise { + /** + * 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 + * out of memory on staging, 2026-10-20). + */ + async drain(): Promise { + if (this.#entries.length >= layoutConfig.writerRunEntries) await this.#run() + } + + async finish(): Promise { + await this.#run() + return this.#held.splice(0) + } + + /** The entries held as an index run, with the packs they're in (closed: a run stands alone). */ + async #run(): Promise { const packs = await this.#packs.finish() - if (!this.#entries.length) return null + this.#packs = new PackWriter(this.layout.store) + if (!this.#entries.length) return this.#entries.sort((a, b) => compareUtf8(a[0], b[0])) const lw = new RunWriter(this.layout.store, 0) for (const [k, v] of this.#entries) lw.add(k, v) @@ -1241,10 +1264,9 @@ class KeyedWriter implements RecordWriter { // (a failed update) is one reclaim finds with nothing live, and drops. await this.layout.registry.add(packs.filter(KeyedLayout.final)) const run = { layers: files, packs: packs.filter((p) => !KeyedLayout.final(p)) } - // Held: the caller names it later, with the rest of its commit (appendRuns). - if (opts.hold) return run - await this.layout.manifests.append(run.layers, run.packs) + // Held: the caller names them later, with the rest of its commit (appendRuns). + if (this.opts.hold) this.#held.push(run) + else await this.layout.manifests.append(run.layers, run.packs) // Merges run after the commit, as their own job (the server's layout.maintain). - return null } } diff --git a/packages/protocol/src/repo/layouts/types.ts b/packages/protocol/src/repo/layouts/types.ts index 8fc847b..428e139 100644 --- a/packages/protocol/src/repo/layouts/types.ts +++ b/packages/protocol/src/repo/layouts/types.ts @@ -44,8 +44,8 @@ export interface RecordStore { entries: readonly RecordEntry[], at: readonly number[], ): Promise - /** A writer for one commit unit or build. */ - writer(): RecordWriter + /** A writer for one commit unit or build; with `hold`, its runs are returned, not named. */ + writer(opts?: { hold?: boolean }): RecordWriter /** * Merges due after a write (layers, packs), run in the writer's own job as a commit's * tail. True when a budget stopped it with work perhaps left: queue another. @@ -62,11 +62,13 @@ export interface RecordStore { export interface RecordWriter { /** Store one record line (a spilled or new entry): its place. */ add(entry: RecordEntry, line: string): string + /** Write a run out if enough is held (called as a build goes): memory stays bounded. */ + drain?(): Promise /** - * Write what's held: packs, the index layer, then the manifest. With `hold`, the - * manifest isn't told: the run is returned, for the commit to name with its others. + * Write what's held: packs, the index layer, then the manifest. A writer made with + * `hold` doesn't tell the manifest: its runs are returned, for the commit to name. */ - finish(opts?: { hold?: boolean }): Promise + finish(): Promise } /** A run a commit unit wrote and held back from the manifest (layout F). */ diff --git a/packages/protocol/src/repo/repo.ts b/packages/protocol/src/repo/repo.ts index c702865..7ab6680 100644 --- a/packages/protocol/src/repo/repo.ts +++ b/packages/protocol/src/repo/repo.ts @@ -831,7 +831,9 @@ export class RepoSink implements TreeSink { get #writer(): RecordWriter | null { if (!this.repo.records) return null if (!this.opts.set) throw new Error('RepoSink: layout F writes records to a set; pass `set`') - return (this.#records ??= this.repo.records[this.opts.set].writer()) + return (this.#records ??= this.repo.records[this.opts.set].writer({ + hold: this.opts.holdRecords ?? false, + })) } rewriting(base: string): void { @@ -910,6 +912,7 @@ export class RepoSink implements TreeSink { } async drain(): Promise { + await this.#records?.drain?.() const limit = this.opts.concurrency ?? 16 const bytes = this.opts.pendingBytes ?? 16 * 1024 * 1024 while (this.#pending.size >= limit || (this.#pending.size > 0 && this.#pendingBytes > bytes)) { @@ -939,8 +942,7 @@ export class RepoSink implements TreeSink { while (this.#pending.size > 0) await Promise.race(this.#pending) if (this.#error) throw this.#error if (this.#records) { - const run = await this.#records.finish({ hold: this.opts.holdRecords ?? false }) - if (run) this.held.push(run) + this.held.push(...(await this.#records.finish())) this.#records = null } } diff --git a/packages/protocol/test/repo/layouts.test.ts b/packages/protocol/test/repo/layouts.test.ts index c7cae33..65bc67f 100644 --- a/packages/protocol/test/repo/layouts.test.ts +++ b/packages/protocol/test/repo/layouts.test.ts @@ -557,9 +557,10 @@ describe("a commit's held runs", () => { const root = memoryStore() const layout = fLayout(root) const pending = async () => (await root.list('runs/c1/manifest/f/pending/')).keys - const w = layout.writer() + const w = layout.writer({ hold: true }) for (let i = 0; i < 100; i++) w.add(rec(i), line(i)) - const run = (await w.finish({ hold: true }))! + const [run] = await w.finish() + if (!run) throw new Error('no run') await layout.appendRuns([run]) // Before the first copy is folded in. await layout.appendRuns([run]) @@ -595,6 +596,41 @@ describe("a commit's held runs", () => { }) }) +describe("a commit's writer", () => { + it('writes a run out each time a run of entries is held, not at the end', async () => { + // A commit's entries held to its end grew with the push: 204k records ran a + // 128 MB isolate out of memory (staging, 2026-10-20). + const saved = layoutConfig.writerRunEntries + layoutConfig.writerRunEntries = 100 + try { + const root = memoryStore() + const layout = fLayout(root) + const keys = Array.from({ length: 350 }, (_, i) => `k${String(i).padStart(4, '0')}`) + const rec = (i: number) => ({ key: keys[i]!, hash: `h${i}`, size: 10 }) + const held = layout.writer({ hold: true }) + for (let i = 0; i < 350; i++) { + held.add(rec(i), `line ${i}`) + await held.drain?.() + } + const runs = await held.finish() + // Three full runs as it went, the rest at the end; each stands alone. + expect(runs.map((r) => r.layers.length > 0)).toEqual([true, true, true, true]) + await layout.appendRuns(runs) + const all = Array.from({ length: 350 }, (_, i) => i) + expect(await layout.leafLines('leaf', all.map(rec))).toEqual(all.map((i) => `line ${i}`)) + // Not held: each run is named as it's written. + const w = fLayout(memoryStore()).writer() + for (let i = 0; i < 350; i++) { + w.add(rec(i), `line ${i}`) + await w.drain?.() + } + expect(await w.finish()).toEqual([]) + } finally { + layoutConfig.writerRunEntries = saved + } + }) +}) + describe('the maintain lease', () => { it('is left alone when another job took it over', async () => { // A job that ran past its lease, which another job then took: releasing it at diff --git a/packages/server/scripts/memory.ts b/packages/server/scripts/memory.ts index bac6e01..9c5d22a 100644 --- a/packages/server/scripts/memory.ts +++ b/packages/server/scripts/memory.ts @@ -43,6 +43,7 @@ * the unit's peak, where it was (which base leaf, how far in), and * what its sink held then * --seed N data seed (default 1) + * --layout L the collection's record layout: today (default) or f * --read-ms N then read the last version whole through records.ndjson and * records.ndjson.gz, as a client taking a chunk every N ms (find-doi * over a range, 2026-10-18), sampling at every chunk (default 0: no read) @@ -122,6 +123,7 @@ const LEAF_MB = str('leaf-mb', '').split(',').filter(Boolean).map(Number) const REPUSH = args.includes('--repush') const PUSHES = opt('pushes', 2) const CACHE = str('cache', 'none') +const LAYOUT = str('layout', 'today') as 'today' | 'f' const TIMER_MS = opt('timer-ms', 0) const READ_MS = opt('read-ms', 0) const PIECES = args.includes('--pieces') @@ -779,7 +781,13 @@ await db.insert(schema.user).values({ id: 'u1', name: 'u1', email: 'u1@example.o await db.insert(schema.member).values({ organizationId: 'org1', userId: 'u1', role: 'owner' }) const [collection] = await db .insert(schema.collections) - .values({ organizationId: 'org1', slug: 'works', name: 'works', privateSalt: newSalt() }) + .values({ + organizationId: 'org1', + slug: 'works', + name: 'works', + privateSalt: newSalt(), + recordLayout: LAYOUT, + }) .returning() await db.insert(schema.placements).values({ collectionId: collection!.id, From 1702dd18abe1d0fac3fd48638c995f0d5846a2cf Mon Sep 17 00:00:00 2001 From: Travis Rich Date: Tue, 20 Oct 2026 13:41:50 -0400 Subject: [PATCH 2/2] Layout F: a commit's writer waits on its pack writes at each drain Closed packs wrote with no limit: with large records a run's worth could be compressing at once. --- packages/protocol/src/repo/layouts/keyed.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/protocol/src/repo/layouts/keyed.ts b/packages/protocol/src/repo/layouts/keyed.ts index d6db77e..61b71e0 100644 --- a/packages/protocol/src/repo/layouts/keyed.ts +++ b/packages/protocol/src/repo/layouts/keyed.ts @@ -1242,6 +1242,8 @@ class KeyedWriter implements RecordWriter { * out of memory on staging, 2026-10-20). */ async drain(): Promise { + // Packs write as they fill: without waiting, large records piled up compressing. + await this.#packs.ready() if (this.#entries.length >= layoutConfig.writerRunEntries) await this.#run() }