diff --git a/server/envSchema.ts b/server/envSchema.ts
index 60f077972..b3f83db7e 100644
--- a/server/envSchema.ts
+++ b/server/envSchema.ts
@@ -194,6 +194,15 @@ export const envSchema = z.object({
 
 	// ── Worker ───────────────────────────────────────────────────────────
 	WORKER: booleanish.describe('Set to true when running as a standalone worker process'),
+	WORKER_MAX_OLD_SPACE_MB: z.coerce
+		.number()
+		.int()
+		.positive()
+		.optional()
+		.describe(
+			'Heap ceiling (MB) for each worker thread. Unset means Node derives it from host memory, ' +
+				'which makes OOM behaviour differ between machines. Set it to make failures reproducible.',
+		),
 	DEFAULT_QUEUE_TASK_PRIORITY: z.coerce
 		.number()
 		.int()
diff --git a/server/underlay/__tests__/client.test.ts b/server/underlay/__tests__/client.test.ts
new file mode 100644
index 000000000..f7cc59648
--- /dev/null
+++ b/server/underlay/__tests__/client.test.ts
@@ -0,0 +1,312 @@
+import type { UnderlayPushPayload } from '../mapping';
+
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { UnderlayClient } from '../client';
+
+/**
+ * Async commit. Underlay finalizes a large commit in the background: it answers 202 and records the
+ * outcome on the session, which the client polls. These tests pin that handshake — including the
+ * fallback to a synchronous 201, so the two services can be deployed in either order.
+ */
+
+const BASE = 'https://underlay.test/api';
+const COLLECTION = `${BASE}/collections/org/coll`;
+
+const makeClient = () =>
+	new UnderlayClient({
+		apiKey: 'key',
+		owner: 'org',
+		slug: 'coll',
+		baseUrl: BASE,
+		pollIntervalMs: 1,
+		pollTimeoutMs: 500,
+	});
+
+const json = (body: unknown, status = 200) =>
+	new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/json' } });
+
+/** A minimal push: one record, no files. */
+const payload = (): UnderlayPushPayload => ({
+	records: [{ id: 'r1', type: 'Pub', data: { slug: 'p1' } }],
+	files: [],
+	fileHashes: [],
+	schemas: {},
+	manifest: [{ id: 'r1', type: 'Pub', hash: 'h1' }],
+});
+
+/**
+ * Routes fetches by URL so a test only has to describe the handful of responses it cares about.
+ * `sessionStates` is consumed one poll at a time, so a test can script `committing → committed`.
+ */
+const stubFetch = (opts: {
+	commit: Response | (() => Response);
+	sessionStates?: unknown[];
+	onPut?: (url: string) => void;
+}) => {
+	const calls: string[] = [];
+	const states = [...(opts.sessionStates ?? [])];
+	const fetchMock = vi.fn(async (input: any, init?: any) => {
+		const url = String(input);
+		calls.push(`${init?.method ?? 'GET'} ${url}`);
+
+		// Re-read of the current head, done when push() recovers by re-negotiating.
+		if (url.endsWith('/versions/latest')) {
+			return json({ semver: '1.0.0' });
+		}
+		if (url.endsWith('/versions/negotiate')) {
+			// The server already has the records; these tests are about the commit handshake.
+			return json({ session_id: 's1', needed_records: [], needed_files: [] });
+		}
+		if (url.includes('/versions/negotiate/s1/records')) {
+			return json({ ok: true });
+		}
+		if (url.includes('/versions/negotiate/s1/commit')) {
+			return typeof opts.commit === 'function' ? opts.commit() : opts.commit;
+		}
+		if (url.includes('/files/sha256:')) {
+			opts.onPut?.(url);
+			return json({ status: 'exists' });
+		}
+		// Session poll.
+		if (url.endsWith('/versions/negotiate/s1')) {
+			return json(states.shift() ?? { status: 'committing' });
+		}
+		throw new Error(`unexpected fetch: ${url}`);
+	});
+	vi.stubGlobal('fetch', fetchMock);
+	return { calls, fetchMock };
+};
+
+afterEach(() => {
+	vi.unstubAllGlobals();
+});
+
+describe('underlay/client — async commit', () => {
+	it('requests an async commit and polls the session until it is committed', async () => {
+		const { calls } = stubFetch({
+			commit: json({ session_id: 's1', status: 'committing' }, 202),
+			sessionStates: [
+				{ status: 'committing', result: null, error: null },
+				{
+					status: 'committed',
+					result: { semver: '1.1.0', hash: 'abc', recordCount: 1, fileCount: 0 },
+				},
+			],
+		});
+
+		const result = await makeClient().push(payload(), '1.0.0', 'msg');
+
+		expect(result).toEqual({
+			status: 'committed',
+			semver: '1.1.0',
+			hash: 'abc',
+			recordCount: 1,
+			fileCount: 0,
+		});
+		// The commit must opt in to async, or the request hangs open past the 60s client timeout.
+		expect(calls.some((c) => c.includes('/commit?async=true'))).toBe(true);
+		// It kept polling through the non-terminal state rather than giving up on the first read.
+		expect(calls.filter((c) => c === `GET ${COLLECTION}/versions/negotiate/s1`)).toHaveLength(
+			2,
+		);
+	});
+
+	it('still accepts a synchronous 201 from an Underlay that predates async commit', async () => {
+		const { calls } = stubFetch({
+			commit: json({ semver: '2.0.0', hash: 'def', recordCount: 1, fileCount: 0 }, 201),
+		});
+
+		const result = await makeClient().push(payload(), '1.0.0', 'msg');
+
+		expect(result).toEqual({
+			status: 'committed',
+			semver: '2.0.0',
+			hash: 'def',
+			recordCount: 1,
+			fileCount: 0,
+		});
+		// No polling — the version came back inline.
+		expect(calls.some((c) => c === `GET ${COLLECTION}/versions/negotiate/s1`)).toBe(false);
+	});
+
+	it('surfaces the server error when the async finalize fails', async () => {
+		stubFetch({
+			commit: json({ session_id: 's1', status: 'committing' }, 202),
+			sessionStates: [
+				{ status: 'failed', error: { statusCode: 400, error: 'Manifest incomplete' } },
+			],
+		});
+
+		await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(
+			/Manifest incomplete/,
+		);
+	});
+
+	it('fails with a clear message when the session expires mid-commit', async () => {
+		stubFetch({
+			commit: json({ session_id: 's1', status: 'committing' }, 202),
+			sessionStates: [{ status: 'expired' }],
+		});
+
+		await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(/expired/i);
+	});
+
+	it('gives up with an actionable message if the commit never reaches a terminal state', async () => {
+		stubFetch({
+			commit: json({ session_id: 's1', status: 'committing' }, 202),
+			sessionStates: [], // always 'committing'
+		});
+
+		await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(
+			/did not finish within/,
+		);
+	});
+
+	it('uploads files the failed commit asked for, then re-negotiates and succeeds', async () => {
+		const uploaded: string[] = [];
+		let commitCount = 0;
+		// First commit fails wanting a file; after we upload it, the retried push commits.
+		const { calls } = stubFetch({
+			commit: () => {
+				commitCount += 1;
+				return json({ session_id: 's1', status: 'committing' }, 202);
+			},
+			sessionStates: [
+				{ status: 'failed', error: { filesNeeded: ['sha256:f1'] } },
+				{
+					status: 'committed',
+					result: { semver: '1.2.0', hash: 'ghi', recordCount: 1, fileCount: 1 },
+				},
+			],
+			onPut: (url) => uploaded.push(url),
+		});
+
+		const withFile: UnderlayPushPayload = {
+			...payload(),
+			files: [{ hash: 'f1', contentType: 'text/html', bytes: Buffer.from('x') }],
+			fileHashes: ['f1'],
+		};
+
+		const result = await makeClient().push(withFile, '1.0.0', 'msg');
+
+		expect(result.status).toBe('committed');
+		expect(uploaded.some((u) => u.includes('sha256:f1'))).toBe(true);
+		// The failed session can't be re-committed, so recovery means a second negotiate.
+		expect(calls.filter((c) => c === `POST ${COLLECTION}/versions/negotiate`)).toHaveLength(2);
+		expect(commitCount).toBe(2);
+	});
+
+	it('does not retry forever when the needed file cannot be produced', async () => {
+		stubFetch({
+			commit: json({ session_id: 's1', status: 'committing' }, 202),
+			sessionStates: [{ status: 'failed', error: { filesNeeded: ['sha256:missing'] } }],
+		});
+
+		await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(
+			/can no longer produce/,
+		);
+	});
+});
+
+describe('underlay/client — authenticated reads', () => {
+	it('sends credentials when reading the latest version', async () => {
+		const seen: { url: string; auth: boolean }[] = [];
+		vi.stubGlobal(
+			'fetch',
+			vi.fn(async (input: any, init?: any) => {
+				const headers = new Headers(init?.headers);
+				seen.push({ url: String(input), auth: headers.has('Authorization') });
+				if (String(input).endsWith('/versions/latest')) {
+					return json({ semver: '3.0.0' });
+				}
+				return json({}, 404);
+			}),
+		);
+
+		const base = await makeClient().getBaseVersion();
+		expect(base).toBe('3.0.0');
+
+		// Anonymous, this 404s on a private collection and is misread as "no versions yet" — which
+		// then pushes base_version: null and dies on a 409 conflict.
+		const call = seen.find((c) => c.url.endsWith('/versions/latest'));
+		expect(call?.auth).toBe(true);
+	});
+
+	it('sends credentials when checking whether the collection exists', async () => {
+		const seen: { url: string; auth: boolean }[] = [];
+		vi.stubGlobal(
+			'fetch',
+			vi.fn(async (input: any, init?: any) => {
+				const headers = new Headers(init?.headers);
+				seen.push({ url: String(input), auth: headers.has('Authorization') });
+				return json({ slug: 'coll' });
+			}),
+		);
+
+		await makeClient().ensureCollection();
+		expect(seen[0]?.auth).toBe(true);
+		// It existed, so nothing was created.
+		expect(seen.some((c) => c.url.endsWith('/collections'))).toBe(false);
+	});
+});
+
+describe('underlay/client — poll resilience and missing-file diagnosis', () => {
+	it('keeps polling when a session read fails, instead of failing a commit that is still running', async () => {
+		let polls = 0;
+		vi.stubGlobal(
+			'fetch',
+			vi.fn(async (input: any) => {
+				const url = String(input);
+				if (url.endsWith('/versions/latest')) return json({ semver: '1.0.0' });
+				if (url.endsWith('/versions/negotiate')) {
+					return json({ session_id: 's1', needed_records: [], needed_files: [] });
+				}
+				if (url.includes('/commit')) return json({ session_id: 's1' }, 202);
+				if (url.endsWith('/versions/negotiate/s1')) {
+					polls += 1;
+					// A 200 whose body is not JSON: `json()` throws, which reaches the poll loop's
+					// catch directly. (Throwing from fetch instead would be absorbed by `request`'s
+					// own retry, so it would never exercise this path.)
+					if (polls <= 3) {
+						return new Response('<html>502 upstream</html>', {
+							status: 200,
+							headers: { 'Content-Type': 'text/html' },
+						});
+					}
+					return json({
+						status: 'committed',
+						result: { semver: '1.1.0', hash: 'h', recordCount: 1, fileCount: 0 },
+					});
+				}
+				throw new Error(`unexpected ${url}`);
+			}),
+		);
+
+		const result = await makeClient().push(payload(), '1.0.0', 'msg');
+		expect(result).toMatchObject({ status: 'committed', semver: '1.1.0' });
+		expect(polls).toBeGreaterThan(3);
+	});
+
+	it('reports which files could not be produced rather than a generic commit failure', async () => {
+		vi.stubGlobal(
+			'fetch',
+			vi.fn(async (input: any) => {
+				const url = String(input);
+				if (url.endsWith('/versions/latest')) return json({ semver: '1.0.0' });
+				if (url.endsWith('/versions/negotiate')) {
+					return json({ session_id: 's1', needed_records: [], needed_files: [] });
+				}
+				// Synchronous 422 naming a file the push cannot regenerate.
+				if (url.includes('/commit')) {
+					return json({ error: 'missing', filesNeeded: ['sha256:gone'] }, 422);
+				}
+				throw new Error(`unexpected ${url}`);
+			}),
+		);
+
+		await expect(makeClient().push(payload(), '1.0.0', 'msg')).rejects.toThrow(
+			/can no longer produce/,
+		);
+	});
+});
diff --git a/server/underlay/__tests__/incremental.test.ts b/server/underlay/__tests__/incremental.test.ts
index 3ef16f0f9..2db97bb9f 100644
--- a/server/underlay/__tests__/incremental.test.ts
+++ b/server/underlay/__tests__/incremental.test.ts
@@ -1,4 +1,10 @@
-import type { CommunityInput, PubInput, PushOptions, UnderlayRecord } from '../mapping';
+import type {
+	CommunityInput,
+	PubInput,
+	PushOptions,
+	UnderlayFile,
+	UnderlayRecord,
+} from '../mapping';
 
 import { describe, expect, it, vi } from 'vitest';
 
@@ -269,8 +275,11 @@ describe('underlay/incremental — buildIncrementalPush', () => {
 		expect(record).not.toBeNull();
 		expect(hashRecord(record!).hash).toBe(releaseHash); // producing it reproduces the requested hash
 		expect(file?.hash).toBe(fileHash);
-		// Both resolves hit the same pub → hydrated exactly once (memoized).
-		expect(mapPub).toHaveBeenCalledTimes(1);
+		// Twice, not once: RECORDS are memoized per pub, FILES deliberately are not. Caching the
+		// bytes that re-mapping yields would reaccumulate the whole collection in memory on a
+		// resumed push, which is exactly what streaming exists to prevent — so the file resolve
+		// re-maps rather than reading a retained byte cache.
+		expect(mapPub).toHaveBeenCalledTimes(2);
 		expect(mapPub).toHaveBeenCalledWith(expect.objectContaining({ id: 'p1' }));
 
 		// An unknown hash resolves to null rather than throwing.
@@ -423,3 +432,357 @@ describe('underlay/incremental — immutable asset cache', () => {
 		await expect(result.payload.resolveFileByHash?.(declaredHash)).rejects.toThrow(/mismatch/);
 	});
 });
+
+describe('underlay/incremental — streaming file upload', () => {
+	const pubs = [makePub('p1', 1), makePub('p2', 1)];
+	const pubUpdatedAt = { p1: '2026-01-05T00:00:00.000Z', p2: '2026-01-06T00:00:00.000Z' };
+
+	const buildStreaming = async (uploaded: UnderlayFile[]) =>
+		buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async (file) => {
+				uploaded.push(file);
+			},
+		});
+
+	it('uploads every file during mapping and retains no bytes', async () => {
+		const uploaded: UnderlayFile[] = [];
+		const result = await buildStreaming(uploaded);
+
+		// Both pubs' content files went up as they were mapped...
+		expect(uploaded).toHaveLength(2);
+		expect(uploaded.every((f) => f.bytes.length > 0)).toBe(true);
+		// ...and nothing is held afterwards, which is the whole point.
+		expect(result.payload.files).toHaveLength(0);
+		// The hashes are still declared, so negotiate sees an identical file set.
+		expect(result.payload.fileHashes).toEqual(uploaded.map((f) => f.hash).sort());
+	});
+
+	it('commits exactly what a non-streaming push would (same signature and manifest)', async () => {
+		const buffered = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+		});
+		const streamed = await buildStreaming([]);
+
+		// Streaming is a transport change, not a content change — the version must be identical, or
+		// the no-op guard would fire spuriously on the next push.
+		expect(streamed.signature).toBe(buffered.signature);
+		expect(streamed.payload.manifest).toEqual(buffered.payload.manifest);
+		expect(streamed.payload.fileHashes).toEqual(
+			buffered.payload.files.map((f) => f.hash).sort(),
+		);
+	});
+
+	it('uploads a file shared by several pubs only once', async () => {
+		const shared = Buffer.from('shared-bytes');
+		const sharedHash = hashBytes(shared);
+		const uploaded: UnderlayFile[] = [];
+
+		await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			// Both pubs reference the identical file.
+			mapPub: async (pub) => ({
+				records: [{ id: pub.id, type: 'Pub', data: { slug: pub.slug } }],
+				files: [{ hash: sharedHash, contentType: 'text/html', bytes: shared }],
+			}),
+			uploadFile: async (file) => {
+				uploaded.push(file);
+			},
+		});
+
+		expect(uploaded).toHaveLength(1);
+		expect(uploaded[0].hash).toBe(sharedHash);
+	});
+
+	it('keeps bytes for a scope image it could not otherwise reproduce', async () => {
+		// A branding image NOT on assets.pubpub.org is never entered into the asset cache, and a
+		// scope file has no owning pub to re-map — so dropping its bytes after upload would leave a
+		// later needed_files request for it unrecoverable.
+		const bytes = Buffer.from('external-logo');
+		const hash = hashBytes(bytes);
+		const uploaded: UnderlayFile[] = [];
+
+		const result = await buildIncrementalPush({
+			community: { ...community, avatar: 'https://example.com/logo.png' },
+			collections: [],
+			pubs: [],
+			pubUpdatedAt: {},
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			fetchAsset: async () => bytes,
+			assetCache: { preloaded: new Map(), learned: new Map(), byHash: new Map() },
+			uploadFile: async (file) => {
+				uploaded.push(file);
+			},
+		});
+
+		// It was uploaded during mapping like any other file …
+		expect(uploaded.map((f) => f.hash)).toContain(hash);
+		// … but its bytes are retained, because nothing else could produce them again.
+		expect(result.payload.files.some((f) => f.hash === hash)).toBe(true);
+		expect(result.payload.fileHashes).toContain(hash);
+	});
+
+	it('drops bytes for a scope image that can be re-fetched from its immutable URL', async () => {
+		// The assets.pubpub.org counterpart: the localizer records url→hash, so resolveFileByHash can
+		// re-fetch it and the bytes need not be held.
+		const bytes = Buffer.from('cacheable-logo');
+		const hash = hashBytes(bytes);
+
+		const result = await buildIncrementalPush({
+			community: { ...community, avatar: 'https://assets.pubpub.org/logo.png' },
+			collections: [],
+			pubs: [],
+			pubUpdatedAt: {},
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			fetchAsset: async () => bytes,
+			assetCache: { preloaded: new Map(), learned: new Map(), byHash: new Map() },
+			uploadFile: async () => {},
+		});
+
+		expect(result.payload.files.some((f) => f.hash === hash)).toBe(false);
+		expect(result.payload.fileHashes).toContain(hash);
+		const recovered = await result.payload.resolveFileByHash?.(hash);
+		expect(recovered?.bytes.equals(bytes)).toBe(true);
+	});
+
+	it('can still regenerate a streamed file the server unexpectedly asks for', async () => {
+		const uploaded: UnderlayFile[] = [];
+		const result = await buildStreaming(uploaded);
+
+		// Bytes were dropped after upload, but the pub that produced them is still resolvable — so a
+		// server that GC'd the blob (or never received it) can be served without failing the push.
+		const wanted = uploaded[0].hash;
+		const resolved = await result.payload.resolveFileByHash?.(wanted);
+		expect(resolved?.hash).toBe(wanted);
+		expect(hashBytes(resolved!.bytes)).toBe(wanted);
+	});
+});
+
+describe('underlay/incremental — upload concurrency and checkpointing', () => {
+	const manyPubs = Array.from({ length: 40 }, (_, i) => makePub(`p${i}`, 1));
+	const manyUpdatedAt = Object.fromEntries(
+		manyPubs.map((p) => [p.id, '2026-01-05T00:00:00.000Z']),
+	);
+
+	it('overlaps uploads instead of running them one at a time', async () => {
+		let inFlight = 0;
+		let peak = 0;
+		await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs: manyPubs,
+			pubUpdatedAt: manyUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async () => {
+				inFlight += 1;
+				peak = Math.max(peak, inFlight);
+				await new Promise((r) => setTimeout(r, 5));
+				inFlight -= 1;
+			},
+		});
+		// Serial uploads would never exceed one in flight; that was the hours-long behaviour.
+		expect(peak).toBeGreaterThan(1);
+	});
+
+	it('never leaves an upload in flight once the push is assembled', async () => {
+		let settled = 0;
+		let started = 0;
+		const result = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs: manyPubs,
+			pubUpdatedAt: manyUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async () => {
+				started += 1;
+				await new Promise((r) => setTimeout(r, 3));
+				settled += 1;
+			},
+		});
+		// The manifest declares these hashes as already present, so every byte must have landed.
+		expect(settled).toBe(started);
+		expect(started).toBe(result.payload.fileHashes!.length);
+	});
+
+	it('fails the push when an upload fails, rather than committing a version missing files', async () => {
+		await expect(
+			buildIncrementalPush({
+				community,
+				collections: [],
+				pubs: manyPubs,
+				pubUpdatedAt: manyUpdatedAt,
+				options: OPTIONS,
+				cacheEntries: [],
+				mapPub: makeMapPub({}),
+				uploadFile: async (file) => {
+					if (file.hash.length > 0) throw new Error('upload exploded');
+				},
+			}),
+		).rejects.toThrow(/upload exploded/);
+	});
+
+	it('checkpoints only pubs whose bytes have actually landed', async () => {
+		const landed = new Set<string>();
+		const checkpointed: string[] = [];
+		await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs: manyPubs,
+			pubUpdatedAt: manyUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async (file) => {
+				await new Promise((r) => setTimeout(r, 2));
+				landed.add(file.hash);
+			},
+			flushCacheEntries: async (entries) => {
+				for (const e of entries) {
+					// Every file this pub declares must already be on the server at checkpoint time.
+					for (const h of e.fileHashes) expect(landed.has(h)).toBe(true);
+					checkpointed.push(e.pubId);
+				}
+			},
+		});
+		expect(checkpointed.sort()).toEqual(manyPubs.map((p) => p.id).sort());
+	});
+
+	it('lets a resumed push skip pubs checkpointed by a previous failed attempt', async () => {
+		const firstAttemptEntries: CachedPubEntry[] = [];
+		await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs: manyPubs,
+			pubUpdatedAt: manyUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async () => {},
+			flushCacheEntries: async (entries) => {
+				firstAttemptEntries.push(...entries);
+			},
+		});
+
+		// Retry using only what the (failed) first attempt managed to checkpoint.
+		const retryCalls: Record<string, number> = {};
+		const uploads: string[] = [];
+		const retry = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs: manyPubs,
+			pubUpdatedAt: manyUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: firstAttemptEntries,
+			mapPub: makeMapPub(retryCalls),
+			uploadFile: async (f) => {
+				uploads.push(f.hash);
+			},
+		});
+
+		expect(retry.stats.cacheHits).toBe(manyPubs.length);
+		expect(retryCalls).toEqual({}); // nothing re-rendered
+		expect(uploads).toEqual([]); // nothing re-uploaded
+	});
+});
+
+describe('underlay/incremental — resumed push re-supplies records without hoarding bytes', () => {
+	const pubs = [makePub('p1', 1), makePub('p2', 1)];
+	const pubUpdatedAt = { p1: '2026-01-05T00:00:00.000Z', p2: '2026-01-06T00:00:00.000Z' };
+
+	/**
+	 * A checkpointed pub has had its FILES uploaded, but records are only sent after negotiate — so a
+	 * push that died mid-mapping leaves cache entries whose records the server has never seen. The
+	 * next push must be able to produce them on demand, without retaining the file bytes that
+	 * re-mapping also yields.
+	 */
+	const checkpointedEntriesFrom = async () => {
+		const first = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: [],
+			mapPub: makeMapPub({}),
+			uploadFile: async () => {},
+			flushCacheEntries: async () => {},
+		});
+		return first.cacheUpserts;
+	};
+
+	it('produces records for a checkpointed pub the server never received', async () => {
+		const entries = await checkpointedEntriesFrom();
+		const calls: Record<string, number> = {};
+		const resumed = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: entries,
+			mapPub: makeMapPub(calls),
+			uploadFile: async () => {},
+		});
+
+		expect(resumed.stats.cacheHits).toBe(2);
+		expect(calls).toEqual({}); // nothing re-mapped during assembly
+
+		// The server asks for a cache-hit pub's record; it must be resolvable.
+		const wanted = Object.values(entries[0].recordHashes)[0].hash;
+		const record = await resumed.payload.resolveRecordByHash?.(wanted);
+		expect(record).toBeTruthy();
+	});
+
+	it('memoizes records but never retains file bytes between resolutions', async () => {
+		const entries = await checkpointedEntriesFrom();
+		const calls: Record<string, number> = {};
+		const resumed = await buildIncrementalPush({
+			community,
+			collections: [],
+			pubs,
+			pubUpdatedAt,
+			options: OPTIONS,
+			cacheEntries: entries,
+			mapPub: makeMapPub(calls),
+			uploadFile: async () => {},
+		});
+
+		const hashes = Object.values(entries[0].recordHashes).map((r) => r.hash);
+		await resumed.payload.resolveRecordByHash?.(hashes[0]);
+		await resumed.payload.resolveRecordByHash?.(hashes[1] ?? hashes[0]);
+		// Records are memoized: one re-map serves every record of that pub.
+		expect(calls.p1).toBe(1);
+
+		// Files are NOT memoized — resolving one re-maps rather than reading a retained byte cache.
+		// That re-map is the observable proof the bytes were dropped.
+		const fileHash = entries[0].fileHashes[0];
+		const file = await resumed.payload.resolveFileByHash?.(fileHash);
+		expect(file?.hash).toBe(fileHash);
+		expect(calls.p1).toBe(2);
+	});
+});
diff --git a/server/underlay/__tests__/pushLogAdoption.test.ts b/server/underlay/__tests__/pushLogAdoption.test.ts
new file mode 100644
index 000000000..540f85d6f
--- /dev/null
+++ b/server/underlay/__tests__/pushLogAdoption.test.ts
@@ -0,0 +1,50 @@
+import { describe, expect, it, vi } from 'vitest';
+
+/**
+ * A `running` push log can outlive the task that created it (killed worker, redelivered AMQP
+ * message). The next task adopts that row rather than creating a duplicate — and must take
+ * ownership of it, because the queue finalizes crashed pushes by looking the row up via
+ * workerTaskId. An adopted row still naming the dead task is invisible to that lookup and stays
+ * `running` forever, which is the bug the finalization fix exists to prevent.
+ */
+describe('underlayPushLog — adopting a stale running row', () => {
+	it('re-points an adopted row at the task that is actually running it', async () => {
+		const updates: Record<string, unknown>[] = [];
+		const existing = {
+			id: 'log-1',
+			workerTaskId: 'dead-task',
+			update: vi.fn(async (patch: Record<string, unknown>) => {
+				updates.push(patch);
+				Object.assign(existing, patch);
+			}),
+		};
+
+		// Mirrors the adoption branch of beginPushLog.
+		const adopt = async (row: typeof existing, workerTaskId: string | null) => {
+			if (workerTaskId && row.workerTaskId !== workerTaskId) {
+				await row.update({ workerTaskId });
+			}
+			return row;
+		};
+
+		await adopt(existing, 'live-task');
+		expect(updates).toEqual([{ workerTaskId: 'live-task' }]);
+		expect(existing.workerTaskId).toBe('live-task');
+	});
+
+	it('does not rewrite the row when the same task re-adopts it', async () => {
+		const existing = {
+			id: 'log-1',
+			workerTaskId: 'same-task',
+			update: vi.fn(async (_patch: Record<string, unknown>) => {}),
+		};
+		const adopt = async (row: typeof existing, workerTaskId: string | null) => {
+			if (workerTaskId && row.workerTaskId !== workerTaskId) {
+				await row.update({ workerTaskId });
+			}
+			return row;
+		};
+		await adopt(existing, 'same-task');
+		expect(existing.update).not.toHaveBeenCalled();
+	});
+});
diff --git a/server/underlay/__tests__/pushLogWarnings.test.ts b/server/underlay/__tests__/pushLogWarnings.test.ts
new file mode 100644
index 000000000..766b7f59d
--- /dev/null
+++ b/server/underlay/__tests__/pushLogWarnings.test.ts
@@ -0,0 +1,58 @@
+import { describe, expect, it } from 'vitest';
+
+/**
+ * Warnings are one-per-skipped-asset, and a large community can produce tens of thousands. Storing
+ * them all put megabytes of JSONB in a single row, which the history endpoint then returns fifty of
+ * at once and the settings UI renders in full. The row keeps a bounded sample; the true total has to
+ * survive on the message so the number an admin sees is still honest.
+ */
+const MAX_STORED_WARNINGS = 100;
+
+/** Mirrors the truncation in finishPushLog. */
+const truncate = (warnings: { reason: string }[], baseMessage: string | null) => {
+	const truncated = warnings.length > MAX_STORED_WARNINGS;
+	return {
+		warnings: truncated ? warnings.slice(0, MAX_STORED_WARNINGS) : warnings,
+		message: truncated
+			? `${baseMessage ? `${baseMessage} — ` : ''}${warnings.length} assets skipped (showing first ${MAX_STORED_WARNINGS})`
+			: baseMessage,
+	};
+};
+
+const makeWarnings = (n: number) =>
+	Array.from({ length: n }, (_, i) => ({ reason: `failed ${i}` }));
+
+describe('underlayPushLog — warning truncation', () => {
+	it('stores every warning when the push produced few', () => {
+		const result = truncate(makeWarnings(7), 'Pushed version v1.0.0');
+		expect(result.warnings).toHaveLength(7);
+		// Nothing was dropped, so the message must not claim otherwise.
+		expect(result.message).toBe('Pushed version v1.0.0');
+	});
+
+	it('caps the stored sample but preserves the true total in the message', () => {
+		const result = truncate(makeWarnings(33818), 'Pushed version v1.0.0');
+		expect(result.warnings).toHaveLength(MAX_STORED_WARNINGS);
+		expect(result.message).toContain('33818 assets skipped');
+		expect(result.message).toContain('showing first 100');
+		expect(result.message).toContain('Pushed version v1.0.0');
+	});
+
+	it('keeps the row small enough to serve fifty of them in one response', () => {
+		const capped = truncate(makeWarnings(33818), null);
+		// The uncapped array was ~6MB; a history page returns up to 50 logs at once.
+		const bytes = JSON.stringify(capped.warnings).length;
+		expect(bytes * 50).toBeLessThan(1_000_000);
+	});
+
+	it('does not lose the total when there is no base message', () => {
+		const result = truncate(makeWarnings(500), null);
+		expect(result.message).toBe('500 assets skipped (showing first 100)');
+	});
+
+	it('does not truncate exactly at the boundary', () => {
+		expect(truncate(makeWarnings(100), null).warnings).toHaveLength(100);
+		expect(truncate(makeWarnings(100), null).message).toBeNull();
+		expect(truncate(makeWarnings(101), null).warnings).toHaveLength(100);
+	});
+});
diff --git a/server/underlay/client.ts b/server/underlay/client.ts
index 60b2ec0c4..92e7540a4 100644
--- a/server/underlay/client.ts
+++ b/server/underlay/client.ts
@@ -19,6 +19,19 @@ const REQUEST_TIMEOUT_MS = 60_000;
 const RECORD_BATCH_SIZE = 10_000;
 const MAX_RETRIES = 4;
 
+/**
+ * Async-commit polling. Underlay validates every record, folds the version digests and writes
+ * version_records during a commit — minutes of work on a large collection, far past
+ * REQUEST_TIMEOUT_MS. Holding the request open would abort and then *retry* that work, so we ask
+ * for an async commit and poll the session instead.
+ */
+const COMMIT_POLL_INITIAL_MS = 2_000;
+const COMMIT_POLL_MAX_MS = 15_000;
+/** Generous: the worker's own task timeout (4h) is the real backstop. */
+const COMMIT_POLL_TIMEOUT_MS = 60 * 60 * 1000;
+/** How often to log that a commit is still running, so a long push isn't silent in the worker logs. */
+const COMMIT_LOG_INTERVAL_MS = 30_000;
+
 export type PushClientOptions = {
 	apiKey: string;
 	owner: string;
@@ -27,6 +40,9 @@ export type PushClientOptions = {
 	/** Identifies the pushing app + actor in the commit metadata. */
 	appId?: string;
 	actorId?: string;
+	/** Async-commit poll timing. Overridable so tests don't wait real seconds. */
+	pollIntervalMs?: number;
+	pollTimeoutMs?: number;
 };
 
 export type PushResult =
@@ -40,6 +56,13 @@ export class UnderlayPushError extends Error {
 		message: string,
 		public readonly statusCode?: number,
 		public readonly detail?: unknown,
+		/**
+		 * The push can recover by re-negotiating from scratch. Set when a commit failed for a cause
+		 * we've since fixed (missing files, now uploaded) but whose session can no longer be
+		 * re-committed — Underlay only accepts a commit on an `open` session, and a failed async
+		 * finalize leaves it `failed`.
+		 */
+		public readonly retriable = false,
 	) {
 		super(message);
 		this.name = 'UnderlayPushError';
@@ -205,6 +228,8 @@ export class UnderlayClient {
 	private readonly slug: string;
 	private readonly appId: string;
 	private readonly actorId: string;
+	private readonly pollIntervalMs: number;
+	private readonly pollTimeoutMs: number;
 
 	constructor(options: PushClientOptions) {
 		this.baseUrl = (options.baseUrl ?? DEFAULT_BASE_URL).replace(/\/$/, '');
@@ -213,6 +238,8 @@ export class UnderlayClient {
 		this.slug = options.slug;
 		this.appId = options.appId ?? 'pubpub';
 		this.actorId = options.actorId ?? 'pubpub:push-to-underlay';
+		this.pollIntervalMs = options.pollIntervalMs ?? COMMIT_POLL_INITIAL_MS;
+		this.pollTimeoutMs = options.pollTimeoutMs ?? COMMIT_POLL_TIMEOUT_MS;
 	}
 
 	private collectionPath() {
@@ -281,15 +308,19 @@ export class UnderlayClient {
 		}
 	}
 
-	/** Returns the latest version semver, or null if the collection has no versions yet. */
+	/**
+	 * Returns the latest version semver, or null if the collection has no versions yet.
+	 *
+	 * Sent authenticated: Underlay resolves this route through the caller's access, so an anonymous
+	 * request against a PRIVATE collection 404s — which is indistinguishable here from "no versions
+	 * yet". That would push `base_version: null`, the server would reject the mismatch with a 409,
+	 * and the conflict retry would re-read the same 404 and fail again. Every push after the first
+	 * would die on a misleading version conflict the moment a collection was made private.
+	 */
 	async getBaseVersion(): Promise<string | null> {
-		const response = await this.request(
-			`${this.collectionPath()}/versions/latest`,
-			{
-				method: 'GET',
-			},
-			{ auth: false },
-		);
+		const response = await this.request(`${this.collectionPath()}/versions/latest`, {
+			method: 'GET',
+		});
 		if (response.status === 404) {
 			return null;
 		}
@@ -419,11 +450,9 @@ export class UnderlayClient {
 
 	/** Ensure the collection exists, creating it under the org if missing. */
 	async ensureCollection(): Promise<void> {
-		const response = await this.request(
-			this.collectionPath(),
-			{ method: 'GET' },
-			{ auth: false },
-		);
+		// Authenticated for the same reason as getBaseVersion: an anonymous probe of a private
+		// collection 404s, sending us down the create path for something that already exists.
+		const response = await this.request(this.collectionPath(), { method: 'GET' });
 		if (response.ok) {
 			return;
 		}
@@ -444,6 +473,15 @@ export class UnderlayClient {
 		}
 	}
 
+	/**
+	 * Upload one file's bytes, content-addressed by hash. Public so a push can stream files up as it
+	 * maps them instead of accumulating every byte in memory until commit; the endpoint is idempotent
+	 * (an already-present hash is a cheap no-op), so re-uploading across attempts is harmless.
+	 */
+	async putFile(file: UnderlayFile): Promise<void> {
+		return this.uploadFile(file);
+	}
+
 	private async uploadFile(file: UnderlayFile): Promise<void> {
 		// Send the file's real content type — Underlay reports back whatever it received, so this is
 		// what downstream consumers see. Falls back to octet-stream only when we truly don't know.
@@ -539,7 +577,9 @@ export class UnderlayClient {
 						actor_id: this.actorId,
 						schemas: payload.schemas,
 						manifest,
-						files: payload.files.map((f) => f.hash),
+						// Every referenced hash — which on a streaming push is a superset of the
+						// bytes still held in `files` (those were uploaded and dropped during mapping).
+						files: payload.fileHashes ?? payload.files.map((f) => f.hash),
 						...(metadata ? { metadata } : {}),
 					}),
 				},
@@ -600,8 +640,10 @@ export class UnderlayClient {
 		try {
 			return await negotiateOnce(baseVersion);
 		} catch (err) {
-			if (err instanceof UnderlayPushError && err.statusCode === 409) {
-				// Someone pushed while we were diffing — re-fetch and retry once.
+			// 409: someone pushed while we were diffing. `retriable`: the commit was rejected for a
+			// cause we've since fixed (missing files, now uploaded) on a session that can no longer
+			// be re-committed. Both are resolved by re-negotiating against the current head, once.
+			if (err instanceof UnderlayPushError && (err.statusCode === 409 || err.retriable)) {
 				const freshBase = await this.getBaseVersion();
 				return negotiateOnce(freshBase);
 			}
@@ -609,15 +651,160 @@ export class UnderlayClient {
 		}
 	}
 
+	/**
+	 * Upload any files named in a `filesNeeded` rejection. Returns the hashes it could not produce.
+	 */
+	private async uploadMissingFiles(
+		filesNeeded: string[],
+		filesByHash: Map<string, UnderlayFile>,
+		resolveFileByHash?: (hash: string) => Promise<UnderlayFile | null>,
+	): Promise<string[]> {
+		const unresolved: string[] = [];
+		for (const ref of filesNeeded) {
+			const hash = ref.replace(/^sha256:/, '');
+			let file = filesByHash.get(hash);
+			if (!file && resolveFileByHash) {
+				// biome-ignore lint/performance/noAwaitInLoops: bounded retry
+				file = (await resolveFileByHash(hash)) ?? undefined;
+			}
+			if (!file) {
+				unresolved.push(hash);
+				continue;
+			}
+			await this.uploadFile(file);
+		}
+		return unresolved;
+	}
+
+	/**
+	 * Poll a session whose commit was accepted asynchronously, until it reports a terminal status.
+	 *
+	 * A transient failure to *read* the session is not a failed commit — the finalize is running
+	 * server-side regardless — so read errors are swallowed and retried until the deadline.
+	 */
+	private async awaitAsyncCommit(
+		sessionId: string,
+		filesByHash: Map<string, UnderlayFile>,
+		resolveFileByHash?: (hash: string) => Promise<UnderlayFile | null>,
+	): Promise<PushResult> {
+		const startedAt = Date.now();
+		const deadline = startedAt + this.pollTimeoutMs;
+		let delay = this.pollIntervalMs;
+		let lastLoggedAt = startedAt;
+
+		while (Date.now() < deadline) {
+			// biome-ignore lint/performance/noAwaitInLoops: polling is inherently sequential
+			await sleep(delay);
+			delay = Math.min(COMMIT_POLL_MAX_MS, Math.round(delay * 1.5));
+
+			// The server keeps finalizing regardless of whether we can read the session, so a failed
+			// poll must not fail the push. `request` throws once its own retries are exhausted, and
+			// a malformed body throws from `json` — both are transient from here, so both keep
+			// polling. If reads never recover, the deadline below produces the actionable timeout.
+			let session: {
+				status: 'open' | 'committing' | 'committed' | 'failed' | 'expired';
+				result?: {
+					semver: string;
+					hash: string;
+					recordCount: number;
+					fileCount: number;
+				} | null;
+				error?: unknown;
+			};
+			try {
+				const response = await this.request(
+					`${this.collectionPath()}/versions/negotiate/${sessionId}`,
+					{ method: 'GET' },
+				);
+				if (!response.ok) {
+					continue;
+				}
+				session = await this.json<typeof session>(response);
+			} catch {
+				continue;
+			}
+
+			if (session.status === 'committed') {
+				if (!session.result?.semver) {
+					throw new UnderlayPushError(
+						'Underlay reported the commit as committed but returned no version',
+						undefined,
+						session,
+					);
+				}
+				return { status: 'committed', ...session.result };
+			}
+
+			if (session.status === 'failed') {
+				const detail = session.error as
+					| { error?: string; filesNeeded?: string[] }
+					| undefined;
+				// Same recovery the synchronous 422 path gets: upload what the server is missing and
+				// push again. The failed session can't be re-committed, so this re-negotiates.
+				if (detail?.filesNeeded && detail.filesNeeded.length > 0) {
+					const unresolved = await this.uploadMissingFiles(
+						detail.filesNeeded,
+						filesByHash,
+						resolveFileByHash,
+					);
+					if (unresolved.length === 0) {
+						throw new UnderlayPushError(
+							'Commit rejected for missing files; they have been uploaded, retrying',
+							422,
+							detail,
+							true,
+						);
+					}
+					throw new UnderlayPushError(
+						`Commit rejected: the Underlay server needs ${unresolved.length} file(s) we can no longer produce. ` +
+							'Check the [underlay] warnings in the worker logs for skipped assets.',
+						422,
+						detail,
+					);
+				}
+				throw new UnderlayPushError(
+					detail?.error ?? 'Underlay reported the commit as failed',
+					undefined,
+					session.error,
+				);
+			}
+
+			if (session.status === 'expired') {
+				throw new UnderlayPushError(
+					'The negotiate session expired before the commit finished',
+					undefined,
+					session,
+				);
+			}
+
+			const now = Date.now();
+			if (now - lastLoggedAt >= COMMIT_LOG_INTERVAL_MS) {
+				lastLoggedAt = now;
+				console.info(
+					`[underlay] Commit still finalizing (${Math.round((now - startedAt) / 1000)}s elapsed)…`,
+				);
+			}
+		}
+
+		throw new UnderlayPushError(
+			`The commit did not finish within ${Math.round(this.pollTimeoutMs / 1000)}s. ` +
+				'It may still complete server-side; check the collection before re-pushing.',
+		);
+	}
+
 	private async commit(
 		sessionId: string,
 		filesByHash: Map<string, UnderlayFile>,
 		resolveFileByHash?: (hash: string) => Promise<UnderlayFile | null>,
 	): Promise<PushResult> {
+		// Ask for an async finalize. An Underlay that predates async commit ignores the query param
+		// and answers 201 with the version inline, which the 2xx path below handles unchanged — so
+		// this is safe against either side being deployed first.
 		const doCommit = () =>
-			this.request(`${this.collectionPath()}/versions/negotiate/${sessionId}/commit`, {
-				method: 'POST',
-			});
+			this.request(
+				`${this.collectionPath()}/versions/negotiate/${sessionId}/commit?async=true`,
+				{ method: 'POST' },
+			);
 
 		let response = await doCommit();
 
@@ -629,17 +816,20 @@ export class UnderlayClient {
 				extraFields?: string[];
 			}>(response);
 			if (body.filesNeeded && body.filesNeeded.length > 0) {
-				for (const ref of body.filesNeeded) {
-					const hash = ref.replace(/^sha256:/, '');
-					let file = filesByHash.get(hash);
-					if (!file && resolveFileByHash) {
-						// biome-ignore lint/performance/noAwaitInLoops: bounded retry
-						file = (await resolveFileByHash(hash)) ?? undefined;
-					}
-					if (file) {
-						// biome-ignore lint/performance/noAwaitInLoops: bounded retry
-						await this.uploadFile(file);
-					}
+				const unresolved = await this.uploadMissingFiles(
+					body.filesNeeded,
+					filesByHash,
+					resolveFileByHash,
+				);
+				// Retrying the commit when we know we could not supply everything just trades a
+				// specific diagnosis for a generic "Commit failed". Same message the async path gives.
+				if (unresolved.length > 0) {
+					throw new UnderlayPushError(
+						`Commit rejected: the Underlay server needs ${unresolved.length} file(s) we can no longer produce. ` +
+							'Check the [underlay] warnings in the worker logs for skipped assets.',
+						422,
+						body,
+					);
 				}
 				response = await doCommit();
 			} else {
@@ -656,6 +846,11 @@ export class UnderlayClient {
 			throw new UnderlayPushError('Commit failed', response.status, detail);
 		}
 
+		// 202: the server is finalizing in the background; the outcome lands on the session.
+		if (response.status === 202) {
+			return this.awaitAsyncCommit(sessionId, filesByHash, resolveFileByHash);
+		}
+
 		const committed = await this.json<{
 			semver: string;
 			hash: string;
diff --git a/server/underlay/incremental.ts b/server/underlay/incremental.ts
index aa42c313f..e06635520 100644
--- a/server/underlay/incremental.ts
+++ b/server/underlay/incremental.ts
@@ -76,6 +76,28 @@ export type IncrementalPushInput = {
 	 * cached file the server later needs is fetched lazily by hash via `resolveFileByHash`.
 	 */
 	assetCache?: AssetCacheContext;
+	/**
+	 * Streaming file upload. When set, each file's bytes are uploaded as soon as the pub that
+	 * produced them is mapped, and then dropped — so peak memory is one pub's files, not the whole
+	 * collection's. Without it, every file (release HTML, PDF/EPUB exports, images) is retained
+	 * until commit, which is what makes a large community's push unbounded in memory.
+	 *
+	 * Safe to do before the version exists: files are content-addressed, the endpoint is idempotent
+	 * by hash, and a push that later fails just leaves unreferenced blobs for GC. The hash is still
+	 * declared in the manifest, and `resolveFileByHash` can always regenerate the bytes if the
+	 * server asks for them anyway.
+	 */
+	uploadFile?: (file: UnderlayFile) => Promise<void>;
+	/**
+	 * Checkpoint callback: persist cache entries for pubs whose records are computed AND whose files
+	 * have landed on the server. Called periodically during mapping, not just at the end.
+	 *
+	 * This is what makes a retry cheap. A push that dies partway (timeout, crash, restart) otherwise
+	 * throws away everything it did, because the cache was only written after a successful commit —
+	 * so the next attempt re-renders and re-uploads content the server already has. With
+	 * checkpointing, those pubs come back as cache hits and are skipped entirely.
+	 */
+	flushCacheEntries?: (entries: CachedPubEntry[]) => Promise<void>;
 };
 
 export type IncrementalPushResult = {
@@ -162,6 +184,72 @@ const recordHashesFrom = (
 	return out;
 };
 
+/**
+ * How many file uploads may be in flight at once.
+ *
+ * A push is dominated by upload latency, not bandwidth or CPU: each PUT is a round trip that spends
+ * almost all of its time waiting. Uploading serially made a large community's push take hours (a
+ * measured ~390ms per file × tens of thousands of files), enough to exceed the worker's own task
+ * timeout. Overlapping them removes that wall without raising peak memory much — at most this many
+ * files' bytes are resident at once.
+ */
+const UPLOAD_CONCURRENCY = 12;
+
+/** How many mapped pubs to accumulate before checkpointing their cache entries. */
+const CHECKPOINT_EVERY_PUBS = 200;
+
+/**
+ * Bounded-concurrency uploader. `submit` returns as soon as a slot is free (so the caller keeps
+ * mapping while uploads are in flight) and `drain` waits for everything to land.
+ *
+ * The first failure is latched and re-thrown from the next `submit`/`drain`, so a broken upload
+ * fails the push promptly instead of letting it continue and commit a version whose files never
+ * arrived.
+ */
+const createUploadPool = (
+	upload: (file: UnderlayFile) => Promise<void>,
+	concurrency: number = UPLOAD_CONCURRENCY,
+) => {
+	const inFlight = new Set<Promise<void>>();
+	let failure: unknown = null;
+
+	const start = (file: UnderlayFile) => {
+		const task: Promise<void> = upload(file)
+			.catch((err) => {
+				failure ??= err;
+			})
+			.finally(() => {
+				inFlight.delete(task);
+			});
+		inFlight.add(task);
+	};
+
+	return {
+		submit: async (file: UnderlayFile): Promise<void> => {
+			if (failure) {
+				throw failure;
+			}
+			while (inFlight.size >= concurrency) {
+				// biome-ignore lint/performance/noAwaitInLoops: waiting for a free slot is the point
+				await Promise.race(inFlight);
+				if (failure) {
+					throw failure;
+				}
+			}
+			start(file);
+		},
+		drain: async (): Promise<void> => {
+			while (inFlight.size > 0) {
+				// biome-ignore lint/performance/noAwaitInLoops: draining is inherently sequential
+				await Promise.race(inFlight);
+			}
+			if (failure) {
+				throw failure;
+			}
+		},
+	};
+};
+
 export const buildIncrementalPush = async (
 	input: IncrementalPushInput,
 ): Promise<IncrementalPushResult> => {
@@ -178,8 +266,24 @@ export const buildIncrementalPush = async (
 		fetchAsset,
 		onAssetWarning,
 		assetCache,
+		uploadFile,
+		flushCacheEntries,
 	} = input;
 
+	const streaming = Boolean(uploadFile);
+	// Hashes already pushed to the server this run, so a file shared across pubs (a reused image, an
+	// identical export) uploads once rather than per referencing pub. Marked on submit, not on
+	// completion, so a duplicate submitted while the first is still in flight is not sent twice.
+	const uploadedHashes = new Set<string>();
+	const pool = uploadFile ? createUploadPool(uploadFile) : null;
+	const streamFile = async (file: UnderlayFile): Promise<void> => {
+		if (!pool || uploadedHashes.has(file.hash)) {
+			return;
+		}
+		uploadedHashes.add(file.hash);
+		await pool.submit(file);
+	};
+
 	const optionsSig = optionsSignature(options);
 	const entryByPubId = new Map(cacheEntries.map((e) => [e.pubId, e]));
 	const pubById = new Map(pubs.map((p) => [p.id, p]));
@@ -200,6 +304,7 @@ export const buildIncrementalPush = async (
 	const freshFilesByHash = new Map<string, UnderlayFile>();
 	const allFileHashes = new Set<string>();
 	const cacheUpserts: CachedPubEntry[] = [];
+	const pendingCheckpoint: CachedPubEntry[] = [];
 	let cacheHits = 0;
 	let cacheMisses = 0;
 
@@ -232,6 +337,26 @@ export const buildIncrementalPush = async (
 	freshRecords.push(...scopeRecords);
 	manifest.push(...buildManifest(scopeRecords));
 
+	// Scope images (community branding, collection + author avatars) are collected synchronously by
+	// `addScopeFile` because the mapper's addFile contract is sync. Upload them now so they aren't
+	// held for the rest of the push.
+	//
+	// Bytes are dropped only when they can be produced again. A scope file has no owning pub for
+	// `hydratePubRecords` to re-map, so its ONLY recovery path is re-fetching the source URL via
+	// `assetCache.byHash` — which the localizer populates for assets.pubpub.org URLs alone. Dropping
+	// an externally-hosted branding image would make a later `needed_files` request for it
+	// unrecoverable, so those stay in memory. That set is small (branding images not on
+	// assets.pubpub.org are the exception), so this costs little and removes the sharp edge.
+	if (streaming) {
+		for (const [hash, file] of [...freshFilesByHash]) {
+			// biome-ignore lint/performance/noAwaitInLoops: sequential to bound memory
+			await streamFile(file);
+			if (assetCache?.byHash.has(hash)) {
+				freshFilesByHash.delete(hash);
+			}
+		}
+	}
+
 	for (const pub of pubs) {
 		const entry = entryByPubId.get(pub.id);
 		if (
@@ -255,7 +380,12 @@ export const buildIncrementalPush = async (
 		freshRecords.push(...records);
 		manifest.push(...buildManifest(records));
 		for (const file of files) {
-			if (!freshFilesByHash.has(file.hash)) {
+			if (streaming) {
+				// Upload now and drop the bytes. `pubIdByHash` still maps the hash back to this pub,
+				// so resolveFileByHash can regenerate it if the server turns out to need it.
+				// biome-ignore lint/performance/noAwaitInLoops: sequential to bound memory
+				await streamFile(file);
+			} else if (!freshFilesByHash.has(file.hash)) {
 				freshFilesByHash.set(file.hash, file);
 			}
 			allFileHashes.add(file.hash);
@@ -264,7 +394,7 @@ export const buildIncrementalPush = async (
 		for (const record of records) {
 			pubIdByHash.set(hashRecord(record).hash, pub.id);
 		}
-		cacheUpserts.push({
+		const cacheEntry: CachedPubEntry = {
 			pubId: pub.id,
 			recordHashes: recordHashesFrom(records),
 			fileHashes: files.map((f) => f.hash),
@@ -272,7 +402,24 @@ export const buildIncrementalPush = async (
 			pubUpdatedAt: toIso(pubUpdatedAt[pub.id] ?? new Date(0).toISOString()),
 			optionsSignature: optionsSig,
 			facetsSignature: pubFacetsSignature[pub.id] ?? '',
-		});
+		};
+		cacheUpserts.push(cacheEntry);
+		pendingCheckpoint.push(cacheEntry);
+
+		// Drain before checkpointing: a pub may only be recorded as cached once its bytes have
+		// actually landed, not merely been queued. Submitting is asynchronous, so without this
+		// barrier a crash could leave a pub marked cached whose files never reached the server.
+		if (flushCacheEntries && pendingCheckpoint.length >= CHECKPOINT_EVERY_PUBS) {
+			await pool?.drain();
+			await flushCacheEntries(pendingCheckpoint.splice(0, pendingCheckpoint.length));
+		}
+	}
+
+	// Everything still queued must land before the manifest is declared: negotiate announces these
+	// hashes as present, so an unfinished upload would make the server ask for a file mid-commit.
+	await pool?.drain();
+	if (flushCacheEntries && pendingCheckpoint.length > 0) {
+		await flushCacheEntries(pendingCheckpoint.splice(0, pendingCheckpoint.length));
 	}
 
 	// Stable ordering (by type, then id) for a tidy manifest; the signature is order-independent.
@@ -290,36 +437,33 @@ export const buildIncrementalPush = async (
 	}
 
 	// ── Lazy re-hydration: produce a needed record/file for a cache-hit pub on demand. ──────────
-	const hydratedByPubId = new Map<
-		string,
-		{ records: Map<string, UnderlayRecord>; files: Map<string, UnderlayFile> }
-	>();
-	const hydratePub = async (pubId: string) => {
-		const cached = hydratedByPubId.get(pubId);
+	//
+	// Only RECORDS are memoized. Re-mapping a pub also produces its file bytes, and holding those
+	// would defeat the streaming this module exists to do: a resumed push can legitimately need to
+	// re-produce records for every pub it checkpointed (see below), so a cache that retained bytes
+	// would reaccumulate the entire collection in memory.
+	//
+	// Why a resumed push needs this at all: a checkpointed pub has had its files uploaded, but its
+	// records are only ever sent AFTER negotiate. A push that dies mid-mapping therefore leaves pubs
+	// marked cached whose records the server has never seen, and the next push gets them all back in
+	// `needed_records`. Records are metadata-sized (the HTML and exports live in files), so memoizing
+	// them is bounded; bytes are not.
+	const hydratedRecordsByPubId = new Map<string, Map<string, UnderlayRecord>>();
+	const hydratePubRecords = async (pubId: string): Promise<Map<string, UnderlayRecord>> => {
+		const cached = hydratedRecordsByPubId.get(pubId);
 		if (cached) {
 			return cached;
 		}
-		const pub = pubById.get(pubId);
-		if (!pub) {
-			const empty = {
-				records: new Map<string, UnderlayRecord>(),
-				files: new Map<string, UnderlayFile>(),
-			};
-			hydratedByPubId.set(pubId, empty);
-			return empty;
-		}
-		const { records, files } = await mapPub(pub);
 		const recordMap = new Map<string, UnderlayRecord>();
-		for (const record of records) {
-			recordMap.set(hashRecord(record).hash, record);
-		}
-		const fileMap = new Map<string, UnderlayFile>();
-		for (const file of files) {
-			fileMap.set(file.hash, file);
+		const pub = pubById.get(pubId);
+		if (pub) {
+			const { records } = await mapPub(pub);
+			for (const record of records) {
+				recordMap.set(hashRecord(record).hash, record);
+			}
 		}
-		const result = { records: recordMap, files: fileMap };
-		hydratedByPubId.set(pubId, result);
-		return result;
+		hydratedRecordsByPubId.set(pubId, recordMap);
+		return recordMap;
 	};
 
 	const resolveRecordByHash = async (hash: string): Promise<UnderlayRecord | null> => {
@@ -327,7 +471,7 @@ export const buildIncrementalPush = async (
 		if (!pubId) {
 			return null;
 		}
-		return (await hydratePub(pubId)).records.get(hash) ?? null;
+		return (await hydratePubRecords(pubId)).get(hash) ?? null;
 	};
 	const resolveFileByHash = async (hash: string): Promise<UnderlayFile | null> => {
 		// A cache-resolved file (e.g. a scope image) has no bytes in memory — fetch them from the
@@ -354,12 +498,22 @@ export const buildIncrementalPush = async (
 		if (!pubId) {
 			return null;
 		}
-		return (await hydratePub(pubId)).files.get(hash) ?? null;
+		const pub = pubById.get(pubId);
+		if (!pub) {
+			return null;
+		}
+		// Deliberately NOT memoized: retaining these bytes is precisely what makes a large push run
+		// out of memory. Underlay keeps uploaded blobs even when a push never commits, so this path
+		// is rare — worth re-mapping a pub for, not worth holding every pub's bytes against.
+		const { files } = await mapPub(pub);
+		return files.find((f) => f.hash === hash) ?? null;
 	};
 
 	const payload: UnderlayPushPayload = {
 		records: freshRecords,
+		// Streaming pushes have already uploaded (and dropped) every byte; only the hashes remain.
 		files: [...freshFilesByHash.values()].sort((a, b) => a.hash.localeCompare(b.hash)),
+		fileHashes: [...allFileHashes].sort(),
 		schemas,
 		manifest,
 		resolveRecordByHash,
diff --git a/server/underlay/mapping.ts b/server/underlay/mapping.ts
index ab0cbe83e..ed78b79e1 100644
--- a/server/underlay/mapping.ts
+++ b/server/underlay/mapping.ts
@@ -60,8 +60,17 @@ export type ManifestEntry = { id: string; type: string; hash: string; private?:
 export type UnderlayPushPayload = {
 	records: UnderlayRecord[];
 	schemas: Record<string, JsonSchema>;
-	/** Deduplicated by hash. */
+	/**
+	 * Deduplicated by hash. Empty when the push streams files (see `fileHashes`): bytes are uploaded
+	 * as each pub is mapped and then dropped, so nothing proportional to the collection is retained.
+	 */
 	files: UnderlayFile[];
+	/**
+	 * Every file hash this version references, whether or not its bytes are still in `files`. The
+	 * streaming path uploads bytes during mapping and keeps only hashes, so this — not `files` — is
+	 * what the negotiate call must declare. Falls back to the hashes of `files` when omitted.
+	 */
+	fileHashes?: string[];
 	/**
 	 * Precomputed full manifest. Set by the incremental push path, where the manifest spans both
 	 * freshly-mapped records and records reused from the push cache (whose data is not in `records`).
diff --git a/server/underlayPushEntry/queries.ts b/server/underlayPushEntry/queries.ts
index 5988041e7..6d05a2377 100644
--- a/server/underlayPushEntry/queries.ts
+++ b/server/underlayPushEntry/queries.ts
@@ -20,44 +20,62 @@ export const getPushCacheEntries = async (
 	}));
 };
 
+/**
+ * Upsert cache entries for pubs that have been fully mapped AND whose files are already uploaded.
+ *
+ * Split out of `applyPushCache` so it can also be called DURING a push, not just after a successful
+ * commit. Entries are pure functions of the pub's inputs (updatedAt, latest release, options,
+ * facets), so writing one early is not a claim that the push succeeded — only that this pub's
+ * records and file hashes have been computed and its bytes handed to Underlay. That makes a
+ * retry after a failed or timed-out push resume instead of restarting: the pub becomes a cache hit,
+ * skipping both its re-render and its re-upload. If the server later turns out to lack one of those
+ * files, negotiate reports it in `needed_files` and `resolveFileByHash` re-maps the pub on demand.
+ */
+export const upsertPushCacheEntries = async (
+	underlayIntegrationId: string,
+	upserts: CachedPubEntry[],
+): Promise<void> => {
+	if (upserts.length === 0) {
+		return;
+	}
+	await UnderlayPushEntry.bulkCreate(
+		upserts.map((entry) => ({
+			underlayIntegrationId,
+			pubId: entry.pubId,
+			recordHashes: entry.recordHashes,
+			fileHashes: entry.fileHashes,
+			latestReleaseHistoryKey: entry.latestReleaseHistoryKey,
+			pubUpdatedAt: new Date(entry.pubUpdatedAt),
+			optionsSignature: entry.optionsSignature,
+			facetsSignature: entry.facetsSignature,
+		})),
+		{
+			// Explicit ON CONFLICT target. Without this, Sequelize derives the conflict keys
+			// from the model's unique indexes, whose fields sequelize-typescript stores as
+			// objects — which crashes quoteIdentifier ("s.replace is not a function").
+			conflictAttributes: ['underlayIntegrationId', 'pubId'],
+			updateOnDuplicate: [
+				'recordHashes',
+				'fileHashes',
+				'latestReleaseHistoryKey',
+				'pubUpdatedAt',
+				'optionsSignature',
+				'facetsSignature',
+			],
+		},
+	);
+};
+
 /**
  * Persist the push cache after a SUCCESSFUL commit: upsert the changed pubs' entries and drop
- * entries for pubs no longer present. Call this only on success so a failed push never poisons the
- * cache.
+ * entries for pubs no longer present. The delete pass is what must wait for success.
  */
 export const applyPushCache = async (
 	underlayIntegrationId: string,
 	upserts: CachedPubEntry[],
 	presentPubIds: string[],
 ): Promise<void> => {
-	if (upserts.length > 0) {
-		await UnderlayPushEntry.bulkCreate(
-			upserts.map((entry) => ({
-				underlayIntegrationId,
-				pubId: entry.pubId,
-				recordHashes: entry.recordHashes,
-				fileHashes: entry.fileHashes,
-				latestReleaseHistoryKey: entry.latestReleaseHistoryKey,
-				pubUpdatedAt: new Date(entry.pubUpdatedAt),
-				optionsSignature: entry.optionsSignature,
-				facetsSignature: entry.facetsSignature,
-			})),
-			{
-				// Explicit ON CONFLICT target. Without this, Sequelize derives the conflict keys
-				// from the model's unique indexes, whose fields sequelize-typescript stores as
-				// objects — which crashes quoteIdentifier ("s.replace is not a function").
-				conflictAttributes: ['underlayIntegrationId', 'pubId'],
-				updateOnDuplicate: [
-					'recordHashes',
-					'fileHashes',
-					'latestReleaseHistoryKey',
-					'pubUpdatedAt',
-					'optionsSignature',
-					'facetsSignature',
-				],
-			},
-		);
-	}
+	await upsertPushCacheEntries(underlayIntegrationId, upserts);
 
 	// Retention: remove cache rows for pubs that no longer exist in this push.
 	await UnderlayPushEntry.destroy({
diff --git a/server/underlayPushLog/queries.ts b/server/underlayPushLog/queries.ts
index ec766f74a..8c94bbabe 100644
--- a/server/underlayPushLog/queries.ts
+++ b/server/underlayPushLog/queries.ts
@@ -22,6 +22,17 @@ export type PushLogView = {
 /** Entries older than this are pruned when a new push begins. */
 const RETENTION_DAYS = 90;
 
+/**
+ * Cap on how many individual warnings a push log stores.
+ *
+ * Warnings are one-per-skipped-asset and a large community can skip tens of thousands (every legacy
+ * .epub whose object predates public-read ACLs, for instance). Persisting all of them put megabytes
+ * of JSONB in a single row — which `getPushHistory` then returns fifty rows of in one response, and
+ * the settings UI renders in full into a popover. The stored sample is for diagnosis; the true total
+ * is preserved separately on the log's message and on the integration's status text.
+ */
+const MAX_STORED_WARNINGS = 100;
+
 /**
  * A `running` log older than this is treated as stale (its worker died without finalizing), so it
  * neither blocks new pushes nor shows as "in progress" forever.
@@ -76,7 +87,12 @@ export const beginPushLog = async (
 
 	const existing = await findRunningRow(communityId);
 	if (existing) {
-		if (workerTaskId && !existing.workerTaskId) {
+		// Re-point the row at whichever task is actually running it, even if it already names one.
+		// A `running` row can outlive its task (killed worker, redelivered message), and the next
+		// task adopts it — so keeping the dead task's id would strand the row: the queue finalizes
+		// crashed pushes via `failPushLogForWorkerTask`, which looks the row up BY workerTaskId and
+		// would silently match nothing, leaving it `running` forever.
+		if (workerTaskId && existing.workerTaskId !== workerTaskId) {
 			await existing.update({ workerTaskId });
 		}
 		return existing;
@@ -106,15 +122,24 @@ export const finishPushLog = async (
 	if (!row) {
 		return;
 	}
+	// Truncate defensively here rather than at the call site, so no caller can put an unbounded
+	// array into the row. The count is folded into the message so nothing is silently lost.
+	const allWarnings = result.warnings ?? [];
+	const truncated = allWarnings.length > MAX_STORED_WARNINGS;
+	const baseMessage = result.message ?? null;
+	const message = truncated
+		? `${baseMessage ? `${baseMessage} — ` : ''}${allWarnings.length} assets skipped (showing first ${MAX_STORED_WARNINGS})`
+		: baseMessage;
+
 	await row.update({
 		status: result.status,
 		finishedAt: new Date(),
 		semver: result.semver ?? null,
 		recordCount: result.recordCount ?? null,
 		fileCount: result.fileCount ?? null,
-		message: result.message ?? null,
+		message,
 		error: result.error ?? null,
-		warnings: result.warnings ?? [],
+		warnings: truncated ? allWarnings.slice(0, MAX_STORED_WARNINGS) : allWarnings,
 	});
 };
 
@@ -150,3 +175,26 @@ export const getPushState = async (
 /** True if a fresh push is already running for this community (concurrency guard). */
 export const hasRunningPush = async (communityId: string): Promise<UnderlayPushLog | null> =>
 	findRunningRow(communityId);
+
+/**
+ * Finalize a `running` log whose worker died before it could finalize itself.
+ *
+ * The task finalizes its own log from a try/catch, which covers every failure it can observe — but
+ * not the process dying underneath it (OOM, container replacement, the queue's watchdog terminating
+ * the thread). In those cases the queue records the error on the WorkerTask and the log is left at
+ * `running` forever, showing an in-progress push in the history that will never resolve. The queue
+ * calls this from its worker-error path to close that gap.
+ *
+ * Keyed on workerTaskId and scoped to `running`, so it can never overwrite a log the task already
+ * finalized (the ordinary case, where the task's own catch wrote a specific error first).
+ */
+export const failPushLogForWorkerTask = async (
+	workerTaskId: string,
+	error: string,
+): Promise<void> => {
+	const row = await UnderlayPushLog.findOne({ where: { workerTaskId, status: 'running' } });
+	if (!row) {
+		return;
+	}
+	await row.update({ status: 'error', finishedAt: new Date(), error });
+};
diff --git a/workers/queue.ts b/workers/queue.ts
index 27ed81a21..04e6d8eed 100644
--- a/workers/queue.ts
+++ b/workers/queue.ts
@@ -7,6 +7,7 @@ import { Worker } from 'worker_threads';
 
 import { env } from 'server/env';
 import { WorkerTask } from 'server/models';
+import { failPushLogForWorkerTask } from 'server/underlayPushLog/queries';
 import { expect } from 'utils/assert';
 import { createCachePurgeDebouncer } from 'utils/caching/createCachePurgeDebouncer';
 import { getAppCommit, isProd } from 'utils/environment';
@@ -61,8 +62,20 @@ const processTask = (channel) => async (message) => {
 	const startTime = Date.now();
 	console.log(`Beginning ${taskData.id} (load ${currentWorkerThreads}/${maxWorkerThreads})`);
 
+	// Without an explicit limit a worker thread's heap ceiling is derived from the host's memory, so
+	// the same task OOMs at different points on different machines and the failure is not
+	// reproducible locally. Setting WORKER_MAX_OLD_SPACE_MB pins it. Left unset by default so this
+	// change alters nothing until someone chooses a value.
+	// Must be positive: a zero or negative limit makes the Worker constructor throw, which would
+	// break every task type, not just this one. The schema rejects such values at boot; this guards
+	// the case where the constructor is reached anyway.
+	const configuredHeapMb = Number(env.WORKER_MAX_OLD_SPACE_MB);
+	const maxOldSpaceMb =
+		Number.isFinite(configuredHeapMb) && configuredHeapMb > 0 ? configuredHeapMb : undefined;
+
 	const worker = new Worker(path.join(__dirname, 'initWorker.js'), {
 		workerData: taskData,
+		...(maxOldSpaceMb ? { resourceLimits: { maxOldGenerationSizeMb: maxOldSpaceMb } } : {}),
 	});
 
 	const onWorkerFinished = async (updatedTaskData) => {
@@ -119,9 +132,24 @@ const processTask = (channel) => async (message) => {
 		if (env.NODE_ENV === 'production') {
 			Sentry.captureException(error);
 		}
+		const message = error.message ? error.message : error;
+
+		// A task that throws finalizes its own bookkeeping, but a task whose *process* dies (OOM,
+		// container replacement, the watchdog below terminating the thread) never gets the chance.
+		// pushToUnderlay keeps a user-visible push log that would otherwise show `running` forever,
+		// so close it out here. Best-effort: never let this stop the WorkerTask from being updated
+		// and the message acked, or a failed task would be redelivered indefinitely.
+		if (taskData.type === 'pushToUnderlay') {
+			try {
+				await failPushLogForWorkerTask(taskData.id, String(message));
+			} catch (err) {
+				console.error(`Failed to finalize underlay push log for ${taskData.id}:`, err);
+			}
+		}
+
 		await onWorkerFinished({
 			isProcessing: false,
-			error: error.message ? error.message : error,
+			error: message,
 			output: null,
 		});
 	};
diff --git a/workers/tasks/pushToUnderlay.ts b/workers/tasks/pushToUnderlay.ts
index 840e4234a..a164a08b9 100644
--- a/workers/tasks/pushToUnderlay.ts
+++ b/workers/tasks/pushToUnderlay.ts
@@ -41,7 +41,11 @@ import {
 	getUnderlayIntegrationWithKey,
 	recordPushResult,
 } from '../../server/underlayIntegration/queries';
-import { applyPushCache, getPushCacheEntries } from '../../server/underlayPushEntry/queries';
+import {
+	applyPushCache,
+	getPushCacheEntries,
+	upsertPushCacheEntries,
+} from '../../server/underlayPushEntry/queries';
 import { beginPushLog, finishPushLog } from '../../server/underlayPushLog/queries';
 import { getReleaseHtml } from './communityExport';
 
@@ -114,55 +118,16 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 
 		const collections = await Collection.findAll({ where: { communityId } });
 
-		// Cheap hydration: attributions, collectionPubs, edges, and release metadata — but NOT the
-		// ProseMirror docs. Docs are loaded lazily per pub only when that pub must be (re)rendered.
-		const pubs = await Pub.findAll({
-			where: { communityId },
-			include: [
-				{
-					model: Release,
-					as: 'releases',
-					separate: true,
-					order: [['historyKey', 'ASC']],
-				},
-				{
-					model: PubAttribution,
-					as: 'attributions',
-					include: [includeUserModel({ as: 'user' })],
-				},
-				{ model: CollectionPub, as: 'collectionPubs' },
-				{ model: Export, as: 'exports' },
-				{
-					model: PubEdge,
-					as: 'outboundEdges',
-					include: [{ model: ExternalPublication, as: 'externalPublication' }],
-				},
-			],
-			order: [['createdAt', 'ASC']],
-		});
-
-		const facets = await fetchFacetsForScopeIds({ pub: pubs.map((p) => p.id) }, [
-			...RENDER_FACET_NAMES,
-		]);
-
 		// Release → docId and pub → releaseIds, so docs can be fetched lazily by pub.
 		const docIdByReleaseId = new Map<string, string>();
 		const releaseIdsByPubId = new Map<string, string[]>();
 		const attributionsByPubId = new Map<string, unknown[]>();
-		for (const pub of pubs) {
-			attributionsByPubId.set(
-				pub.id,
-				(pub.attributions ?? []).map((a) => a.toJSON()),
-			);
-			const releaseIds: string[] = [];
-			for (const release of pub.releases ?? []) {
-				releaseIds.push(release.id);
-				if (release.docId) {
-					docIdByReleaseId.set(release.id, release.docId);
-				}
-			}
-			releaseIdsByPubId.set(pub.id, releaseIds);
-		}
+		// pubId → the pub's fully-resolved facet cascade, retained for the lazy render below. Plain
+		// values, not model instances: a few small objects per pub.
+		const resolvedFacetsByPubId = new Map<string, unknown>();
+		const pubInputs: PubInput[] = [];
+		const pubUpdatedAt: Record<string, Date> = {};
+		const pubFacetsSignature: Record<string, string> = {};
 
 		// Lazy doc loading: fetch a pub's release docs once, on first render of that pub.
 		const docByReleaseId = new Map<string, DocJson>();
@@ -251,10 +216,23 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 			return undefined;
 		};
 
-		const pubInputs: PubInput[] = pubs.map((pub) => {
+		// Hydrate in chunks. A community's pubs are loaded as Sequelize model instances, which cost
+		// several times what the plain PubInput we derive from them does; loading all of them at once
+		// put peak memory in proportion to the community. Chunking keeps only CHUNK pubs' instances
+		// live at a time — the derived plain objects are all we retain.
+		const PUB_HYDRATION_CHUNK = 250;
+		const pubIdRows = (await Pub.findAll({
+			where: { communityId },
+			attributes: ['id'],
+			order: [['createdAt', 'ASC']],
+			raw: true,
+		})) as unknown as { id: string }[];
+		const orderedPubIds = pubIdRows.map((row) => row.id);
+
+		const toPubInput = (pub: Pub): PubInput => {
 			// License comes from the already-resolved facet cascade (no extra query); enrich the raw
 			// kind with the SPDX id + canonical URI via the shared license table.
-			const pubFacets = facets.pub[pub.id] as
+			const pubFacets = resolvedFacetsByPubId.get(pub.id) as
 				| { License?: { value?: { kind?: string } } }
 				| undefined;
 			const licenseKind = pubFacets?.License?.value?.kind;
@@ -335,22 +313,78 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 						historyKey: e.historyKey,
 					})),
 			};
-		});
+		};
 
-		const pubUpdatedAt: Record<string, Date> = {};
-		for (const pub of pubs) {
-			pubUpdatedAt[pub.id] = pub.updatedAt;
-		}
+		for (let i = 0; i < orderedPubIds.length; i += PUB_HYDRATION_CHUNK) {
+			const chunkIds = orderedPubIds.slice(i, i + PUB_HYDRATION_CHUNK);
+
+			// `separate: true` on every hasMany. Without it Sequelize emits ONE query joining
+			// attributions × collectionPubs × exports × outboundEdges, whose row count is the product
+			// of those four per pub — each row repeating the pub's full payload (description,
+			// htmlDescription, metadata, …). On a large community with long author lists and several
+			// export formats that product is enormous, and materializing it is what exhausted the
+			// worker heap. Separate queries make the cost additive instead of multiplicative.
+			// biome-ignore lint/performance/noAwaitInLoops: chunks are sequential to bound memory
+			const chunkPubs = await Pub.findAll({
+				where: { id: chunkIds },
+				include: [
+					{
+						model: Release,
+						as: 'releases',
+						separate: true,
+						order: [['historyKey', 'ASC']],
+					},
+					{
+						model: PubAttribution,
+						as: 'attributions',
+						separate: true,
+						include: [includeUserModel({ as: 'user' })],
+					},
+					{ model: CollectionPub, as: 'collectionPubs', separate: true },
+					{ model: Export, as: 'exports', separate: true },
+					{
+						model: PubEdge,
+						as: 'outboundEdges',
+						separate: true,
+						include: [{ model: ExternalPublication, as: 'externalPublication' }],
+					},
+				],
+			});
 
-		// Facet change signal. `fetchFacetsForScopeIds({ pub })` already resolved the full
-		// community→collection→pub cascade for us (no extra query), so `facets.pub[pubId]` is the exact
-		// resolved facet stack that feeds getReleaseHtml. Hashing that value gives cascade-correct
-		// invalidation for free: a community facet edit changes every pub's resolved value; a collection
-		// edit changes only that collection's pubs; a pub edit changes only that pub. It's value-based,
-		// so a no-op facet edit that doesn't change the effective value won't force a needless re-render.
-		const pubFacetsSignature: Record<string, string> = {};
-		for (const pub of pubs) {
-			pubFacetsSignature[pub.id] = computeFacetsSignature(facets.pub[pub.id]);
+			// Facet change signal. `fetchFacetsForScopeIds({ pub })` resolves the full
+			// community→collection→pub cascade, so this is the exact resolved facet stack that feeds
+			// getReleaseHtml. Hashing that value gives cascade-correct invalidation for free: a
+			// community facet edit changes every pub's resolved value; a collection edit changes only
+			// that collection's pubs; a pub edit changes only that pub. It's value-based, so a no-op
+			// facet edit that doesn't change the effective value won't force a needless re-render.
+			const chunkFacets = await fetchFacetsForScopeIds({ pub: chunkIds }, [
+				...RENDER_FACET_NAMES,
+			]);
+
+			// Preserve the global createdAt ordering: `where: { id: [...] }` does not guarantee it.
+			const byId = new Map(chunkPubs.map((pub) => [pub.id, pub]));
+			for (const pubId of chunkIds) {
+				const pub = byId.get(pubId);
+				if (!pub) {
+					continue;
+				}
+				resolvedFacetsByPubId.set(pubId, chunkFacets.pub[pubId]);
+				pubFacetsSignature[pubId] = computeFacetsSignature(chunkFacets.pub[pubId]);
+				pubUpdatedAt[pubId] = pub.updatedAt;
+				attributionsByPubId.set(
+					pubId,
+					(pub.attributions ?? []).map((a) => a.toJSON()),
+				);
+				const releaseIds: string[] = [];
+				for (const release of pub.releases ?? []) {
+					releaseIds.push(release.id);
+					if (release.docId) {
+						docIdByReleaseId.set(release.id, release.docId);
+					}
+				}
+				releaseIdsByPubId.set(pubId, releaseIds);
+				pubInputs.push(toPubInput(pub));
+			}
 		}
 
 		const renderReleaseHtml = async ({
@@ -369,7 +403,7 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 				return null;
 			}
 			// The legacy renderer's facet/metadata types are broad; cast at this boundary only.
-			const pubFacets = facets.pub[pub.id] as any;
+			const pubFacets = resolvedFacetsByPubId.get(pub.id) as any;
 			const metadata = {
 				attributions: attributionsByPubId.get(pub.id) ?? [],
 				licenseKind: pubFacets?.License?.value?.kind,
@@ -491,6 +525,14 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 		// the no-op signature below so a readme-only edit isn't skipped as "no changes".
 		const pushMetadata = integration.readme ? { readme: integration.readme } : undefined;
 
+		// The collection must exist before any file can be uploaded into it, and the streaming
+		// uploader below runs during mapping — so this moves ahead of the build (it used to sit just
+		// before negotiate). Creating it early is harmless: it was going to be created either way,
+		// and a push that fails afterwards just leaves an empty collection with no versions.
+		await client.ensureCollection();
+
+		let uploadedFileCount = 0;
+		let checkpointedPubs = 0;
 		const cacheEntries = await getPushCacheEntries(integration.id);
 		const incremental = await buildIncrementalPush({
 			community: communityInput,
@@ -505,6 +547,23 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 			fetchAsset,
 			onAssetWarning: collectAssetWarning,
 			assetCache,
+			// Stream each pub's files up as it is mapped rather than holding every release's HTML
+			// and every PDF/EPUB export in memory until commit. Peak file memory becomes one pub's
+			// worth, regardless of how large the community is.
+			uploadFile: (file) => {
+				uploadedFileCount += 1;
+				return client.putFile(file);
+			},
+			// Checkpoint progress so a push that dies partway (worker timeout, restart, crash) is
+			// resumable: the pubs already mapped and uploaded come back as cache hits next attempt
+			// instead of being re-rendered and re-uploaded from scratch.
+			flushCacheEntries: async (entries) => {
+				await upsertPushCacheEntries(integration.id, entries);
+				checkpointedPubs += entries.length;
+				console.info(
+					`[underlay] Checkpointed ${checkpointedPubs} pub(s); ${uploadedFileCount} file(s) uploaded so far.`,
+				);
+			},
 		});
 
 		// Client-side no-op guard: identical content since the last push → skip entirely.
@@ -530,10 +589,9 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 		}
 
 		console.info(
-			`[underlay] Mapped ${incremental.stats.totalPubs} pub(s): ${incremental.stats.cacheHits} cache hit(s), ${incremental.stats.cacheMisses} re-mapped. Negotiating…`,
+			`[underlay] Mapped ${incremental.stats.totalPubs} pub(s): ${incremental.stats.cacheHits} cache hit(s), ${incremental.stats.cacheMisses} re-mapped, ${uploadedFileCount} file(s) streamed. Negotiating…`,
 		);
 
-		await client.ensureCollection();
 		const baseVersion = await client.getBaseVersion();
 		const result = await client.push(
 			incremental.payload,
@@ -544,10 +602,15 @@ export const pushToUnderlayTask = async (input: PushToUnderlayInput) => {
 
 		const warnings: AssetWarning[] = [...assetWarnings.values()];
 		if (warnings.length > 0) {
+			// Log a sample, not the whole set: a large community can skip tens of thousands of
+			// assets, and joining them all produced a single multi-megabyte log line.
+			const LOGGED = 50;
+			const shown = warnings.slice(0, LOGGED);
+			const more = warnings.length - shown.length;
 			console.warn(
-				`[underlay] Push completed with ${warnings.length} skipped asset(s):\n${warnings
+				`[underlay] Push completed with ${warnings.length} skipped asset(s):\n${shown
 					.map((w) => `  - ${w.assetUrl} (pub ${w.pubId}): ${w.reason}`)
-					.join('\n')}`,
+					.join('\n')}${more > 0 ? `\n  … and ${more} more` : ''}`,
 			);
 		}