perf(pds): stream getRepo to fix OOM on large repos (#131)

* perf(pds): stream getRepo response to fix OOM on large repos

Route getRepo through the DO's fetch() handler instead of RPC, enabling
a streaming Response. Lazily iterates SQLite rows and pipes them through
writeCarStream, dropping memory usage from O(repo_size) to O(single_block).

* ci: add changeset for streaming getRepo

Author: ascorbic

.changeset/real-teams-lead.md

@@ -0,0 +1,5 @@
+---
+"@getcirrus/pds": patch
+---
+
+Stream getRepo response to fix OOM on large repos

packages/pds/src/account-do.ts

@@ -4,8 +4,10 @@ import {
 	WriteOpAction,
 	BlockMap,
 	blocksToCarFile,
+	writeCarStream,
 	readCarWithRoot,
 	getRecords,
+	type CarBlock,
 	type RecordCreateOp,
 	type RecordUpdateOp,
 	type RecordDeleteOp,
@@ -738,31 +740,50 @@ export class AccountDurableObject extends DurableObject<PDSEnv> {
 	}
 
 	/**
-	 * RPC method: Export repo as CAR
+	 * Handle streaming getRepo via fetch (not RPC, to enable streaming response).
 	 */
-	async rpcGetRepoCar(): Promise<Uint8Array> {
+	private async handleGetRepo(): Promise<Response> {
 		const storage = await this.getStorage();
 		const root = await storage.getRoot();
 
 		if (!root) {
-			throw new Error("No repository root found");
+			return Response.json(
+				{ error: "RepoNotFound", message: "No repository root found" },
+				{ status: 404 },
+			);
 		}
 
-		// Get all blocks from SQLite storage
-		const rows = this.ctx.storage.sql
-			.exec("SELECT cid, bytes FROM blocks")
-			.toArray();
-
-		// Build BlockMap
-		const blocks = new BlockMap();
-		for (const row of rows) {
-			const cid = CID.parse(row.cid as string);
-			const bytes = new Uint8Array(row.bytes as ArrayBuffer);
-			blocks.set(cid, bytes);
+		// Lazily iterate SQLite rows — the cursor is already lazy,
+		// only .toArray() would materialize everything in memory.
+		const cursor = this.ctx.storage.sql.exec(
+			"SELECT cid, bytes FROM blocks",
+		);
+
+		async function* blocks(): AsyncGenerator<CarBlock> {
+			for (const row of cursor) {
+				yield {
+					cid: CID.parse(row.cid as string),
+					bytes: new Uint8Array(row.bytes as ArrayBuffer),
+				};
+			}
 		}
 
-		// Use the official CAR builder
-		return blocksToCarFile(root, blocks);
+		const carIter = writeCarStream(root, blocks())[Symbol.asyncIterator]();
+
+		const stream = new ReadableStream<Uint8Array>({
+			async pull(controller) {
+				const { value, done } = await carIter.next();
+				if (done) {
+					controller.close();
+				} else {
+					controller.enqueue(value);
+				}
+			},
+		});
+
+		return new Response(stream, {
+			headers: { "Content-Type": "application/vnd.ipld.car" },
+		});
 	}
 
 	/**
@@ -1541,15 +1562,18 @@ export class AccountDurableObject extends DurableObject<PDSEnv> {
 	}
 
 	/**
-	 * HTTP fetch handler for WebSocket upgrades.
-	 * This is used instead of RPC to avoid WebSocket serialization errors.
+	 * HTTP fetch handler for WebSocket upgrades and streaming responses.
+	 * Used instead of RPC when the response can't be serialized (WebSocket)
+	 * or when streaming is needed to avoid buffering large payloads (getRepo).
 	 */
 	override async fetch(request: Request): Promise<Response> {
-		// Only handle WebSocket upgrades via fetch
 		const url = new URL(request.url);
 		if (url.pathname === "/xrpc/com.atproto.sync.subscribeRepos") {
 			return this.handleFirehoseUpgrade(request);
 		}
+		if (url.pathname === "/xrpc/com.atproto.sync.getRepo") {
+			return this.handleGetRepo();
+		}
 
 		// All other requests should use RPC methods, not fetch
 		return new Response("Method not allowed", { status: 405 });

packages/pds/src/xrpc/sync.ts

@@ -38,15 +38,10 @@ export async function getRepo(
 		);
 	}
 
-	const carBytes = await accountDO.rpcGetRepoCar();
-
-	return new Response(carBytes, {
-		status: 200,
-		headers: {
-			"Content-Type": "application/vnd.ipld.car",
-			"Content-Length": carBytes.length.toString(),
-		},
-	});
+	// Stream through the DO's fetch handler to avoid buffering the entire CAR
+	return accountDO.fetch(
+		new Request("https://do/xrpc/com.atproto.sync.getRepo"),
+	);
 }
 
 export async function getRepoStatus(

packages/pds/test/xrpc.test.ts

@@ -1140,6 +1140,71 @@ describe("XRPC Endpoints", () => {
 			expect(blocks.length).toBeGreaterThan(0);
 		});
 
+		it("should stream getRepo response without Content-Length", async () => {
+			const { CarReader } = await import("@ipld/car");
+
+			// Ensure repo has content
+			await worker.fetch(
+				new Request("http://pds.test/xrpc/com.atproto.repo.createRecord", {
+					method: "POST",
+					headers: {
+						"Content-Type": "application/json",
+						Authorization: `Bearer ${env.AUTH_TOKEN}`,
+					},
+					body: JSON.stringify({
+						repo: env.DID,
+						collection: "app.bsky.feed.post",
+						rkey: "stream-test",
+						record: {
+							$type: "app.bsky.feed.post",
+							text: "Streaming test",
+							createdAt: new Date().toISOString(),
+						},
+					}),
+				}),
+				env,
+			);
+
+			const response = await worker.fetch(
+				new Request(
+					`http://pds.test/xrpc/com.atproto.sync.getRepo?did=${env.DID}`,
+				),
+				env,
+			);
+			expect(response.status).toBe(200);
+
+			// Streaming: no Content-Length header
+			expect(response.headers.get("Content-Length")).toBeNull();
+
+			// Body is a ReadableStream, not a fixed buffer
+			expect(response.body).toBeInstanceOf(ReadableStream);
+
+			// Read incrementally to verify chunked delivery
+			const reader = response.body!.getReader();
+			const chunks: Uint8Array[] = [];
+			for (;;) {
+				const { value, done } = await reader.read();
+				if (done) break;
+				chunks.push(value);
+			}
+
+			// Should arrive in multiple chunks (header + blocks)
+			expect(chunks.length).toBeGreaterThan(1);
+
+			// Reassembled bytes are a valid CAR
+			const totalLength = chunks.reduce((n, c) => n + c.byteLength, 0);
+			const carBytes = new Uint8Array(totalLength);
+			let offset = 0;
+			for (const chunk of chunks) {
+				carBytes.set(chunk, offset);
+				offset += chunk.byteLength;
+			}
+
+			const car = await CarReader.fromBytes(carBytes);
+			const roots = await car.getRoots();
+			expect(roots).toHaveLength(1);
+		});
+
 		it("should export CAR file for empty repo", async () => {
 			const { CarReader } = await import("@ipld/car");