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..61b71e0 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,27 @@ 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 { + // 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() + } + + 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 +1266,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,