builder: trees in slabs; too big by what a replay holds; pooled pages given back

web/lib/city.ts

@@ -2,6 +2,7 @@
 import { bundle, mineOf, type Payload } from './engine/bundle.js';
 import { discover, type Discovery, NotFound, RateLimited, type Target } from './engine/discover.js';
 import { GitError } from './engine/git.js';
+import { collectGarbage } from './engine/memory.js';
 import { RECORD_VERSION, type RepoRecord } from './engine/replay.js';
 import { replayRepo } from './engine/repo.js';
 import { HttpError } from './http.js';
@@ -169,10 +170,12 @@ export async function startBuild(target: Target, token: string, opts: { whole?:
 // Fluid compute runs concurrent requests in one process, 2 GB between them. Big replays side by side can outgrow
 // that together though each fits alone (every client sends a city's biggest repos first, four at a time), so
 // they take turns; small repos, most of any city, never wait. A request that waits too long is told the builder
-// is busy, and the page asks again.
+// is busy, and the page asks again. Each turn starts with a thorough collection: what the last big replay left is
+// garbage until V8 gets to it, and its pooled pages stay resident until V8 is told to give them back.
 const HEAVY_KB = 16 * 1024;
 const TURN_WAIT_MS = 150_000;      // the function has 300 s: this leaves a replay that waited the whole time 130
 const heavy = new Turns();
+const mb = (n: number) => (n / 1e6).toFixed(0);
 
 export async function buildRepo(ticketToken: string, name: string) {
   const t0 = Date.now();
@@ -187,6 +190,12 @@ export async function buildRepo(ticketToken: string, name: string) {
   if (big && !(await heavy.take(TURN_WAIT_MS))) throw new HttpError(503, 'the builder is busy with other big repos', { status: 'busy' });
   try {
     if (big && Date.now() - t0 > 2000 && await exists(path)) return { cached: true };   // replayed for another build meanwhile
+    if (big) {
+      const a = process.memoryUsage();
+      collectGarbage(true);
+      const b = process.memoryUsage();
+      console.log(`${repo}: big replay's turn after ${Date.now() - t0} ms; collected: live ${mb(a.heapUsed + a.external)} -> ${mb(b.heapUsed + b.external)} MB, rss ${mb(a.rss)} -> ${mb(b.rss)} MB`);
+    }
     const { record, stats } = await replayRepo({
       repo, head, headTime, start: t.s, timeoutMs: Math.min(240_000, 280_000 - (Date.now() - t0)), maxPackBytes: 400e6,
       token: process.env.GITHUB_TOKEN, log: m => console.log(m),
@@ -196,7 +205,7 @@ export async function buildRepo(ticketToken: string, name: string) {
   } catch (e) {
     if (e instanceof GitError) {
       const m = process.memoryUsage();
-      console.log(`${repo}: ${e.code}: ${e.message}; live ${((m.heapUsed + m.external) / 1e6).toFixed(0)} MB, rss ${(m.rss / 1e6).toFixed(0)} MB, ${heavy.waiting} waiting`);
+      console.log(`${repo}: ${e.code}: ${e.message}; live ${mb(m.heapUsed + m.external)} MB, rss ${mb(m.rss)} MB, ${heavy.waiting} waiting`);
       if (e.code === 'busy') throw new HttpError(503, 'the builder is short of memory right now', { status: 'busy' });
       const reason = e.code === 'too_big' ? 'too big for the hosted builder' : e.code === 'timeout' ? 'took too long to replay'
         : e.code === 'not_found' ? 'not readable' : `fetch failed (${e.message.slice(0, 120)})`;

web/lib/engine/memory.ts

@@ -0,0 +1,81 @@
+// The replay guard: stops a replay that is out of time, or whose memory would take the instance down. Fluid compute
+// runs concurrent requests in one process, 2 GB between them, and past that the platform kills the instance with
+// every request on it.
+import v8 from 'node:v8';
+import vm from 'node:vm';
+import { GitError } from './git.js';
+
+export const MAX_LIVE = 1.2e9;  // heap + buffers one replay may add to what was live when it began
+export const MAX_RSS = 1.9e9;   // resident size, kept pages included: past it the platform is about to kill the instance
+const MEM_EVERY_MS = 100;       // reading memory is a syscall, and the guard runs every 1024 objects, commit and pkt-line
+const GC_EVERY_MS = 2000;       // a full collection is not free: at most one per two seconds per replay
+
+export interface Memory { live: number; rss: number }    // heap + buffers; resident size
+
+export function readMemory(): Memory {
+  const m = process.memoryUsage();
+  return { live: m.heapUsed + m.external, rss: m.rss };
+}
+
+// V8's collector, which Node hides unless started with --expose-gc; a context made after the flag is set has it.
+// A normal collection runs twice: buffers the first pass frees still count as external memory until the next one
+// settles the sweep. The thorough one is V8's last resort, which also gives back the heap pages V8 keeps pooled:
+// after google/dawn, two normal collections left 1,002 MB resident for 57 MB live; the last resort, 427 MB.
+let gc: NodeJS.GCFunction | null | undefined;
+export function collectGarbage(thorough = false) {
+  if (gc === undefined) {
+    try {
+      gc = typeof globalThis.gc === 'function' ? globalThis.gc : (v8.setFlagsFromString('--expose-gc'), vm.runInNewContext('gc') as NodeJS.GCFunction);
+    } catch { gc = null; }
+  }
+  if (!gc) return;
+  if (thorough) gc({ type: 'major', execution: 'sync', flavor: 'last-resort' });
+  else { gc(); gc(); }
+}
+
+export interface GuardOptions {
+  deadline: number;             // epoch ms: past it the replay stops with timeout
+  maxLive?: number;
+  maxRss?: number;
+  read?: () => Memory;          // tests
+  collect?: (thorough: boolean) => void;
+  now?: () => number;
+}
+
+const mb = (n: number) => Math.round(n / 1e6);
+
+/** The check a replay runs as it goes; `reserve` is memory the step under way is about to add (a pack's copy). */
+export function replayGuard(o: GuardOptions): (reserve?: number) => void {
+  const read = o.read ?? readMemory, collect = o.collect ?? collectGarbage, now = o.now ?? Date.now;
+  const maxLive = o.maxLive ?? MAX_LIVE, maxRss = o.maxRss ?? MAX_RSS;
+  // Too big is about the repo, so it counts what this replay holds: live memory beyond what was live when it began.
+  // The process's whole figure failed repos on an instance still busy with others. Busy is about the instance: its
+  // resident size, with what the step under way is about to add.
+  const base = read().live;
+  let readAt = -Infinity, gcAt = -Infinity, thoroughAt = -Infinity;
+  return (reserve = 0) => {
+    const t = now();
+    if (t > o.deadline) throw new GitError('replay ran out of time', 'timeout');
+    if (t - readAt < MEM_EVERY_MS) return;
+    readAt = t;
+    let m = read();
+    const full = () => m.rss + reserve > maxRss;
+    const busy = () => new GitError(`the hosted builder is short of memory right now (${mb(m.rss)} MB resident${reserve ? `, ${mb(reserve)} MB to come` : ''})`, 'busy');
+    if (m.live - base <= maxLive && !full()) return;
+    // Garbage counts as live until V8 collects it, and V8 collects lazily while memory is plentiful: collect before
+    // believing a figure, at most once per GC_EVERY_MS (over again a moment later, it is new garbage). A full
+    // instance gets the thorough collection, the one that gives pooled pages back; still full a moment after one,
+    // it is full.
+    const tight = full();
+    if (t - (tight ? thoroughAt : gcAt) < GC_EVERY_MS) {
+      if (tight) throw busy();
+      return;
+    }
+    gcAt = t;
+    if (tight) thoroughAt = t;
+    collect(tight);
+    m = read();
+    if (m.live - base > maxLive) throw new GitError(`replay needs more memory than the hosted builder has (${mb(m.live - base)} MB of its own)`, 'too_big');
+    if (full()) throw busy();
+  };
+}

web/lib/engine/pack.ts

@@ -21,6 +21,7 @@ interface Loc { pack: Buffer; start: number; type: number; base: string | null }
 const TYPE_NAME = ['', 'commit', 'tree', 'blob', 'tag'];
 const OFS_DELTA = 6, REF_DELTA = 7;
 const BINARY_PROBE = 8000;    // git's buffer_is_binary(): a NUL in the first 8000 bytes
+export const TREE_SLAB = 1 << 20;    // kept trees are copied into shared slabs of this size
 
 export class ObjectDB {
   commits = new Map<string, Commit>();
@@ -32,6 +33,8 @@ export class ObjectDB {
   guard: (() => void) | null = null;
   private locs = new Map<string, Loc>();      // where each object lives, to read one again later
   private keepLocs = true;
+  private slab: Buffer | null = null;
+  private slabUsed = 0;
 
   /** Forget object locations (and with them the pack buffers) once nothing more needs reading. */
   releaseReads() { this.locs.clear(); this.keepLocs = false; }
@@ -112,14 +115,27 @@ export class ObjectDB {
     return base ? applyDelta(base, raw) : null;
   }
 
+  /** A tree to keep for the replay, copied out of the buffer it was built in when that buffer is bigger. An inflated
+   *  object is a view of zlib's 16 KB output buffer and a small delta result a slice of Node's shared 8 KB pool, so
+   *  a 300-byte tree kept as it came held 16 KB: anomalyco/opencode's window held 129 MB of trees in 684 MB. */
+  private hold(data: Buffer): Buffer {
+    if (data.byteOffset === 0 && data.length === data.buffer.byteLength) return data;
+    if (data.length > TREE_SLAB / 16) return Buffer.from(data);
+    if (!this.slab || this.slabUsed + data.length > this.slab.length) { this.slab = Buffer.allocUnsafeSlow(TREE_SLAB); this.slabUsed = 0; }
+    const out = this.slab.subarray(this.slabUsed, this.slabUsed + data.length);
+    data.copy(out);
+    this.slabUsed += data.length;
+    return out;
+  }
+
   private process(type: number, data: Buffer): string {
     const h = createHash('sha1');
     h.update(`${TYPE_NAME[type]} ${data.length}\0`);
     h.update(data);
     const sha = h.digest('hex');
     if ((++this.objects & 1023) === 0) this.guard?.();
     if (type === 1) this.commits.set(sha, parseCommit(sha, data));
-    else if (type === 2) this.trees.set(sha, data);
+    else if (type === 2) this.trees.set(sha, this.hold(data));
     else if (type === 3) this.blobs.set(sha, blobInfo(data));
     return sha;
   }

web/lib/engine/repo.ts

@@ -1,10 +1,9 @@
 // Fetch + replay one repository. Fetches are shallow and incremental: the window's commits first, then
 // whatever the shallow edge cut off that the replay still needs (the commit the window starts from, the
 // rest of a first-parent chain cut at a merge of an old branch, the parents of side-branch commits).
-import v8 from 'node:v8';
-import vm from 'node:vm';
 import { type FetchOptions, type FetchResult, fetchPack, GitError, httpTransport, type Transport } from './git.js';
 import { makeKeep } from './langs.js';
+import { replayGuard } from './memory.js';
 import { ObjectDB } from './pack.js';
 import { MissingObject, type RepoRecord, Replayer } from './replay.js';
 
@@ -17,7 +16,7 @@ export interface ReplayOptions {
   transport?: Transport;        // tests inject a local upload-pack
   maxPackBytes?: number;
   timeoutMs?: number;
-  maxLiveBytes?: number;        // stop with too_big when the heap and buffers hold more than this after a collection
+  maxLiveBytes?: number;        // stop with too_big when this replay holds more than this after a collection
   maxRssBytes?: number;         // stop with busy past this resident size (the instance's memory is shared)
   blobLimit?: number;           // blobs at least this big are fetched only when the replay needs them (0 = all)
   log?: (msg: string) => void;
@@ -30,53 +29,20 @@ const MAX_SIDE_PARENTS = 400;
 const MAX_SIDE_ROUNDS = 12;     // merged branches inside merged branches
 const BLOB_LIMIT = 128 * 1024;
 const BLOBS_PER_FETCH = 4000;
-const MAX_LIVE = 1.2e9;         // heap + buffers one replay may hold: functions get 2 GB, and the allocator keeps freed pages
-const MAX_RSS = 1.9e9;          // resident size, kept pages included: past it the platform is about to kill the instance
-const MEM_EVERY_MS = 100;       // reading memory is a syscall, and the guard runs every 1024 objects and every commit
-const GC_EVERY_MS = 2000;      // two full collections are not free: at most one pair per two seconds
-
-// V8's collector, which Node hides unless started with --expose-gc; a context made after the flag is set has it.
-// Twice: buffers the first pass frees still count as external memory until the next one settles the sweep.
-let gc: (() => void) | null | undefined;
-function collectGarbage() {
-  if (gc === undefined) {
-    try {
-      gc = typeof globalThis.gc === 'function' ? globalThis.gc : (v8.setFlagsFromString('--expose-gc'), vm.runInNewContext('gc') as () => void);
-    } catch { gc = null; }
-  }
-  gc?.();
-  gc?.();
-}
 
 export async function replayRepo(o: ReplayOptions): Promise<{ record: RepoRecord; stats: ReplayStats }> {
   const t0 = Date.now();
   const tr = o.transport ?? httpTransport(`https://github.com/${o.repo}`, o.token);
   const maxBytes = o.maxPackBytes ?? 350e6;
   const deadline = t0 + (o.timeoutMs ?? 200_000);
   const db = new ObjectDB();
-  // Indexing a pack and replaying it are synchronous and can run for minutes on a big repo, so the deadline
-  // is checked inside them too, not only between fetches: past it, the platform kills the function at 300 s
-  // (or the instance runs out of memory) and the visitor gets a bare 500 after a five-minute wait.
+  // Indexing a pack and replaying it are synchronous and can run for minutes on a big repo, so the deadline and
+  // memory are checked inside them too, and while a pack downloads, not only between fetches: past either, the
+  // platform kills the function at 300 s or the instance with every request on it, and the visitor gets a bare 500.
   // Memory is judged by what is live, not what is resident: RSS keeps the pages the allocator freed but did not
   // return, so a warm instance that had replayed one big repo stayed "full" and failed every replay after it
-  // (vercel/bridge, a 1 MB repo, in 0.5 s). Garbage counts as live until V8 collects it, and V8 collects lazily
-  // while memory is plentiful, so the guard collects before it believes a number over the limit.
-  const maxLive = o.maxLiveBytes ?? MAX_LIVE, maxRss = o.maxRssBytes ?? MAX_RSS;
-  let memAt = 0, gcAt = 0;
-  const guard = () => {
-    const now = Date.now();
-    if (now > deadline) throw new GitError('replay ran out of time', 'timeout');
-    if (now - memAt < MEM_EVERY_MS) return;
-    memAt = now;
-    let m = process.memoryUsage();
-    if (m.heapUsed + m.external <= maxLive && m.rss <= maxRss) return;
-    if (now - gcAt < GC_EVERY_MS) return;              // collected a moment ago: what is over now is new garbage
-    gcAt = now;
-    collectGarbage();
-    m = process.memoryUsage();
-    if (m.heapUsed + m.external > maxLive) throw new GitError('replay needs more memory than the hosted builder has', 'too_big');
-    if (m.rss > maxRss) throw new GitError('the hosted builder is short of memory right now', 'busy');
-  };
+  // (vercel/bridge, a 1 MB repo, in 0.5 s). memory.ts has the rest.
+  const guard = replayGuard({ deadline, maxLive: o.maxLiveBytes, maxRss: o.maxRssBytes });
   db.guard = guard;
   const shallow = new Set<string>();
   const limit = o.blobLimit ?? BLOB_LIMIT;
@@ -90,12 +56,14 @@ export async function replayRepo(o: ReplayOptions): Promise<{ record: RepoRecord
     const budget = Math.max(1, maxBytes - db.packBytes);
     const t = Date.now();
     let r: FetchResult;
+    // the pack's bytes so far will be copied once more when it is whole: the guard keeps room for that copy
+    const onProgress = (bytes: number) => guard(bytes);
     try {
-      r = await fetchPack(tr, { ...f, shallow: [...shallow], maxBytes: budget, timeoutMs: left, filter: unfiltered ? undefined : filter });
+      r = await fetchPack(tr, { ...f, shallow: [...shallow], maxBytes: budget, timeoutMs: left, filter: unfiltered ? undefined : filter, onProgress });
     } catch (e) {
       if (!f.haves?.length || !(e instanceof GitError) || e.code === 'empty' || e.code === 'too_big' || e.code === 'timeout' || e.code === 'busy') throw e;
       fetches++;                                    // negotiation trouble: ask again without haves
-      r = await fetchPack(tr, { ...f, haves: [], shallow: [...shallow], maxBytes: budget, timeoutMs: deadline - Date.now(), filter: unfiltered ? undefined : filter });
+      r = await fetchPack(tr, { ...f, haves: [], shallow: [...shallow], maxBytes: budget, timeoutMs: deadline - Date.now(), filter: unfiltered ? undefined : filter, onProgress });
     }
     db.addPack(r.pack);
     for (const s of r.shallow) shallow.add(s);

web/test/engine.test.ts

@@ -1,9 +1,11 @@
 // The hosted engine must draw the same city as the Python pipeline: replay synthetic repos through a
 // real `git upload-pack` (protocol v2, shallow) and compare against commit_city.extract.replay.
 import assert from 'node:assert/strict';
+import { randomBytes } from 'node:crypto';
 import path from 'node:path';
 import { test } from 'node:test';
 import { GitError } from '../lib/engine/git';
+import { collectGarbage } from '../lib/engine/memory';
 import { replayRepo } from '../lib/engine/repo';
 import type { RepoRecord } from '../lib/engine/replay';
 import { localTransport, pythonReplay, Repo, tmpdir } from './helpers';
@@ -237,12 +239,28 @@ test('a branch merged into a main that was quiet before the window keeps all its
 
 test('a replay that outgrows its memory stops with too_big instead of taking the instance down', async () => {
   const root = tmpdir();
-  const r = scenario(path.join(root, 'repo'));
-  // any real process holds over 1 byte, even after a collection: the guard inside the replay loop must fire after the fetch succeeded
-  await assert.rejects(replayRepo({ repo: 'test/repo', head: r.head(), start: 0, transport: localTransport(r.bare()), maxLiveBytes: 1 }),
+  const r = new Repo(path.join(root, 'repo'));
+  r.write('data.bin', 0, randomBytes(4e6)).write('a.txt', 3);
+  r.commit(T0);
+  const head = r.head(), transport = localTransport(r.bare());
+  // its own 4 MB pack, all blobs in it, against a 1 MB limit. Collected first, so garbage from before (the clone
+  // above, earlier tests) is not in the starting point the replay is measured from, to be freed under it
+  collectGarbage();
+  await assert.rejects(replayRepo({ repo: 'test/repo', head, start: 0, transport, blobLimit: 0, maxLiveBytes: 1e6 }),
     (e: unknown) => e instanceof GitError && e.code === 'too_big' && /memory/.test(e.message));
 });
 
+test('memory another request holds does not make a replay too big', async () => {
+  const root = tmpdir();
+  const r = scenario(path.join(root, 'repo'));
+  // google/dawn's first fetch read 1,196 MB live in production, where it was then refused as too big, and 484 MB
+  // replayed alone: the rest was other requests'. 200 MB held throughout, against a 100 MB limit
+  const other = Array.from({ length: 50 }, () => Buffer.alloc(4e6, 1));
+  const { record } = await replayRepo({ repo: 'test/repo', head: r.head(), start: 0, transport: localTransport(r.bare()), maxLiveBytes: 100e6 });
+  assert.ok(record.commits.length > 0);
+  assert.equal(other.length, 50);
+});
+
 test('a full instance makes a replay busy, to be asked again, not too big', async () => {
   const root = tmpdir();
   const r = scenario(path.join(root, 'repo'));