fix(pds): page listRecords without repeating or stalling

.changeset/list-records-pagination.md

@@ -0,0 +1,7 @@
+---
+"@getcirrus/pds": minor
+---
+
+Fix `com.atproto.repo.listRecords` pagination. Following the cursor returned the record at each page boundary up to three times, and with `reverse=true` or `limit=2` the cursor never advanced, so clients that paged until the cursor ran out looped forever.
+
+Records are now listed newest first by default and oldest first with `reverse=true`, matching the reference PDS (the order was previously the other way round). The cursor is now the bare rkey of the last record, as in the reference PDS; cursors in the old `collection/rkey` form are still accepted. `limit` is clamped to 1–100, and listing a collection with no records no longer scans the rest of the repository.

packages/pds/src/repo/engine.ts

@@ -14,6 +14,7 @@ import { CID, asCid, isBlobRef } from "@atproto/lex-data";
 import { now as tidNow } from "@atcute/tid";
 import { jsonToLex } from "@atproto/lex-json";
 import { SqliteRepoStorage } from "./storage";
+import { listCollectionLeaves } from "./list-records";
 import type { Sequencer, SeqEvent, CommitData } from "./sequencer";
 import { RecordAlreadyExistsError, type ValidationStatus } from "../validation";
 
@@ -192,7 +193,8 @@ export class RepoEngine {
 	}
 
 	/**
-	 * List records in a collection.
+	 * List records in a collection, newest first unless `reverse` is set.
+	 * The cursor is the rkey of the last record returned.
 	 */
 	async listRecords(
 		collection: string,
@@ -206,35 +208,27 @@ export class RepoEngine {
 		cursor?: string;
 	}> {
 		const repo = await this.getRepo();
-		const records = [];
-		const startFrom = opts.cursor || `${collection}/`;
-
-		for await (const record of repo.walkRecords(startFrom)) {
-			if (record.collection !== collection) {
-				if (records.length > 0) break;
-				continue;
-			}
+		// Fetch one extra leaf to tell whether there's another page.
+		const leaves = await listCollectionLeaves(repo.data, collection, {
+			...opts,
+			limit: opts.limit + 1,
+		});
+		const hasMore = leaves.length > opts.limit;
+		const page = hasMore ? leaves.slice(0, opts.limit) : leaves;
 
+		const records = [];
+		let rkey = "";
+		for (const leaf of page) {
+			rkey = leaf.key.slice(collection.length + 1);
+			const record = await repo.storage.readRecord(leaf.value);
 			records.push({
-				uri: `at://${repo.did}/${record.collection}/${record.rkey}`,
-				cid: record.cid.toString(),
-				value: serializeRecord(record.record),
+				uri: `at://${repo.did}/${collection}/${rkey}`,
+				cid: leaf.value.toString(),
+				value: serializeRecord(record),
 			});
-
-			if (records.length >= opts.limit + 1) break;
-		}
-
-		if (opts.reverse) {
-			records.reverse();
 		}
 
-		const hasMore = records.length > opts.limit;
-		const results = hasMore ? records.slice(0, opts.limit) : records;
-		const cursor = hasMore
-			? `${collection}/${results[results.length - 1]?.uri.split("/").pop() ?? ""}`
-			: undefined;
-
-		return { records: results, cursor };
+		return { records, cursor: hasMore ? rkey : undefined };
 	}
 
 	/**

packages/pds/src/repo/list-records.ts

@@ -0,0 +1,80 @@
+import type { Leaf, MST } from "@atproto/repo";
+
+/**
+ * List one page of a collection's MST leaves.
+ *
+ * Leaves come back newest first by default, as in the reference PDS, and
+ * oldest first with `reverse`. The cursor is the rkey of the last leaf of the
+ * previous page and is excluded from this one.
+ *
+ * MST.walkFrom() can't be used here: it yields its start key twice when that
+ * key exists, and it only walks forwards.
+ */
+export async function listCollectionLeaves(
+	mst: MST,
+	collection: string,
+	opts: { limit: number; cursor?: string; reverse?: boolean },
+): Promise<Leaf[]> {
+	const prefix = `${collection}/`;
+	// Cursors used to be `collection/rkey`; accept those for clients that
+	// are mid-pagination across an upgrade.
+	const rkey = opts.cursor?.startsWith(prefix)
+		? opts.cursor.slice(prefix.length)
+		: opts.cursor;
+
+	// Exclusive bounds. Record keys only use ASCII below DEL, so `prefix\x7f`
+	// sorts after every key in the collection.
+	let gt = prefix;
+	let lt = `${prefix}\x7f`;
+	if (rkey) {
+		if (opts.reverse) gt = `${prefix}${rkey}`;
+		else lt = `${prefix}${rkey}`;
+	}
+
+	const leaves: Leaf[] = [];
+	if (opts.limit < 1) return leaves;
+	for await (const leaf of walkRange(mst, gt, lt, !opts.reverse)) {
+		leaves.push(leaf);
+		if (leaves.length >= opts.limit) break;
+	}
+	return leaves;
+}
+
+/**
+ * Yield the leaves with keys strictly between `gt` and `lt`, in key order or,
+ * when `descending`, in reverse. A subtree holds only keys between the leaves
+ * either side of it, so subtrees outside the range are never loaded.
+ */
+async function* walkRange(
+	node: MST,
+	gt: string,
+	lt: string,
+	descending: boolean,
+): AsyncGenerator<Leaf> {
+	const entries = await node.getEntries();
+	if (descending) {
+		for (let i = entries.length - 1; i >= 0; i--) {
+			const entry = entries[i]!;
+			if (entry.isLeaf()) {
+				if (entry.key <= gt) return;
+				if (entry.key < lt) yield entry;
+			} else {
+				const prev = entries[i - 1];
+				if (prev?.isLeaf() && prev.key >= lt) continue;
+				yield* walkRange(entry, gt, lt, descending);
+			}
+		}
+	} else {
+		for (let i = 0; i < entries.length; i++) {
+			const entry = entries[i]!;
+			if (entry.isLeaf()) {
+				if (entry.key >= lt) return;
+				if (entry.key > gt) yield entry;
+			} else {
+				const next = entries[i + 1];
+				if (next?.isLeaf() && next.key <= gt) continue;
+				yield* walkRange(entry, gt, lt, descending);
+			}
+		}
+	}
+}

packages/pds/src/xrpc/repo.ts

@@ -266,12 +266,15 @@ export async function listRecords(
 		);
 	}
 
-	const limit = Math.min(limitStr ? Number.parseInt(limitStr, 10) : 50, 100);
+	const limit = Math.min(
+		Math.max(limitStr ? Number.parseInt(limitStr, 10) || 50 : 50, 1),
+		100,
+	);
 	const reverse = reverseStr === "true";
 
 	const result = await accountDO.repo().listRecords(collection, {
 		limit,
-		cursor,
+		cursor: cursor || undefined,
 		reverse,
 	});
 

packages/pds/test/list-records.test.ts

@@ -0,0 +1,114 @@
+import { describe, it, expect, beforeAll } from "vitest";
+import { MST, MemoryBlockstore, cidForRecord } from "@atproto/repo";
+import { listCollectionLeaves } from "../src/repo/list-records";
+
+const COLLECTION = "app.bsky.graph.follow";
+const COUNT = 340;
+
+describe("listCollectionLeaves", () => {
+	let mst: MST;
+	let rkeys: string[];
+
+	beforeAll(async () => {
+		const cid = await cidForRecord({ test: true });
+		mst = await MST.create(new MemoryBlockstore());
+		rkeys = [];
+		for (let i = 0; i < COUNT; i++) {
+			const rkey = `3m${(1e10 + i * 7919).toString(36)}`;
+			rkeys.push(rkey);
+			mst = await mst.add(`${COLLECTION}/${rkey}`, cid);
+		}
+		rkeys.sort();
+		// Neighbours that sort just before and just after the collection.
+		for (const key of [
+			"app.bsky.feed.post/3maaaaaaaaaaa",
+			"app.bsky.graph.follow.ext/3maaaaaaaaaaa",
+			"app.bsky.graph.followa/3maaaaaaaaaaa",
+			"app.bsky.graph.listitem/3maaaaaaaaaaa",
+		]) {
+			mst = await mst.add(key, cid);
+		}
+		// Make sure the cases below cross subtree boundaries.
+		expect(await mst.getLayer()).toBeGreaterThan(1);
+	});
+
+	async function pageThrough(limit: number, reverse: boolean) {
+		const seen: string[] = [];
+		let cursor: string | undefined;
+		for (let pages = 0; pages <= COUNT; pages++) {
+			const leaves = await listCollectionLeaves(mst, COLLECTION, {
+				limit,
+				cursor,
+				reverse,
+			});
+			seen.push(...leaves.map((l) => l.key.slice(COLLECTION.length + 1)));
+			if (leaves.length < limit) return seen;
+			cursor = seen[seen.length - 1];
+		}
+		throw new Error("pagination did not terminate");
+	}
+
+	it("pages newest first with no duplicates or gaps", async () => {
+		const expected = [...rkeys].reverse();
+		for (const limit of [1, 2, 3, 7, 50, 100, COUNT, COUNT + 1]) {
+			expect(await pageThrough(limit, false)).toEqual(expected);
+		}
+	});
+
+	it("pages oldest first with reverse", async () => {
+		for (const limit of [1, 2, 3, 7, 50, 100, COUNT, COUNT + 1]) {
+			expect(await pageThrough(limit, true)).toEqual(rkeys);
+		}
+	});
+
+	it("excludes the cursor key and starts after a missing one", async () => {
+		const cursor = rkeys[100]!;
+		const before = await listCollectionLeaves(mst, COLLECTION, {
+			limit: 2,
+			cursor,
+		});
+		expect(before.map((l) => l.key)).toEqual([
+			`${COLLECTION}/${rkeys[99]}`,
+			`${COLLECTION}/${rkeys[98]}`,
+		]);
+
+		const after = await listCollectionLeaves(mst, COLLECTION, {
+			limit: 2,
+			cursor: `${cursor}0`,
+			reverse: true,
+		});
+		expect(after.map((l) => l.key)).toEqual([
+			`${COLLECTION}/${rkeys[101]}`,
+			`${COLLECTION}/${rkeys[102]}`,
+		]);
+	});
+
+	it("accepts the old collection/rkey cursor format", async () => {
+		const leaves = await listCollectionLeaves(mst, COLLECTION, {
+			limit: 1,
+			cursor: `${COLLECTION}/${rkeys[10]}`,
+			reverse: true,
+		});
+		expect(leaves.map((l) => l.key)).toEqual([`${COLLECTION}/${rkeys[11]}`]);
+	});
+
+	it("stays inside the collection", async () => {
+		const first = await listCollectionLeaves(mst, COLLECTION, {
+			limit: 1,
+			cursor: rkeys[0],
+		});
+		expect(first).toEqual([]);
+
+		const last = await listCollectionLeaves(mst, COLLECTION, {
+			limit: 1,
+			cursor: rkeys[COUNT - 1],
+			reverse: true,
+		});
+		expect(last).toEqual([]);
+
+		const missing = await listCollectionLeaves(mst, "app.bsky.graph.block", {
+			limit: 10,
+		});
+		expect(missing).toEqual([]);
+	});
+});

packages/pds/test/xrpc.test.ts

@@ -583,6 +583,87 @@ describe("XRPC Endpoints", () => {
 			});
 		});
 
+		describe("listRecords pagination", () => {
+			const collection = "com.example.pagination";
+			const rkeys: string[] = [];
+
+			async function list(params: string) {
+				const response = await worker.fetch(
+					new Request(
+						`http://pds.test/xrpc/com.atproto.repo.listRecords?repo=${env.DID}&collection=${collection}&${params}`,
+					),
+					env,
+				);
+				expect(response.status).toBe(200);
+				return (await response.json()) as {
+					records: Array<{ uri: string }>;
+					cursor?: string;
+				};
+			}
+
+			async function pageThrough(params: string) {
+				const seen: string[] = [];
+				let cursor: string | undefined;
+				for (let pages = 0; pages < 20; pages++) {
+					const data = await list(
+						cursor ? `${params}&cursor=${cursor}` : params,
+					);
+					seen.push(...data.records.map((r) => r.uri.split("/").pop()!));
+					if (!data.cursor) return seen;
+					cursor = data.cursor;
+				}
+				throw new Error("pagination did not terminate");
+			}
+
+			it("creates records to page through", async () => {
+				for (let i = 0; i < 12; i++) rkeys.push(genTid());
+				const response = await worker.fetch(
+					new Request("http://pds.test/xrpc/com.atproto.repo.applyWrites", {
+						method: "POST",
+						headers: {
+							"Content-Type": "application/json",
+							Authorization: `Bearer ${env.AUTH_TOKEN}`,
+						},
+						body: JSON.stringify({
+							repo: env.DID,
+							writes: rkeys.map((rkey, i) => ({
+								$type: "com.atproto.repo.applyWrites#create",
+								collection,
+								rkey,
+								value: { $type: collection, index: i },
+							})),
+						}),
+					}),
+					env,
+				);
+				expect(response.status).toBe(200);
+				rkeys.sort();
+			});
+
+			it("pages newest first by default", async () => {
+				expect(await pageThrough("limit=5")).toEqual([...rkeys].reverse());
+			});
+
+			it("pages oldest first with reverse=true", async () => {
+				expect(await pageThrough("limit=5&reverse=true")).toEqual(rkeys);
+			});
+
+			it("returns a bare rkey cursor only when there are more records", async () => {
+				const first = await list("limit=5");
+				expect(first.cursor).toBe(rkeys[rkeys.length - 5]);
+
+				const all = await list("limit=12");
+				expect(all.records).toHaveLength(12);
+				expect(all.cursor).toBeUndefined();
+			});
+
+			it("clamps out-of-range limits", async () => {
+				expect((await list("limit=0")).records).toHaveLength(12);
+				expect((await list("limit=-3")).records).toHaveLength(1);
+				expect((await list("limit=abc")).records).toHaveLength(12);
+			});
+		});
+
 		it("should delete a record", async () => {
 			const rkey = genTid();
 			await worker.fetch(

plans/todo/endpoint-implementation.md

@@ -30,7 +30,7 @@ This document tracks the implementation status of all AT Protocol XRPC endpoints
 | `getRecord`        | ✅ Complete | With CID and value                             |
 | `importRepo`       | ✅ Complete | CAR file import with validation, blob tracking |
 | `listMissingBlobs` | ✅ Complete | Lists blobs referenced but not imported        |
-| `listRecords`      | ✅ Complete | Pagination, cursor, reverse                    |
+| `listRecords`      | ✅ Complete | Newest first, rkey cursor, reverse (#251)      |
 | `putRecord`        | ✅ Complete | Create or update with validation               |
 | `uploadBlob`       | ✅ Complete | 5MB limit, R2 storage, tracks imports          |