From 1ecc9f24f268d59cfa3bea8be4a44b5f2a227e46 Mon Sep 17 00:00:00 2001 From: Arjun Krishna Date: Mon, 5 Oct 2026 06:50:29 +0530 Subject: [PATCH 1/2] fix(pds): clamp listRecords limit to the lexicon range listRecords passed limit straight through, capped only at 100. limit=0 or a negative value returned an empty page whose cursor pointed back at the start of the collection, so clients that follow the cursor looped forever. A non-numeric limit became NaN, which disabled the limit and returned every record in the collection in one response. Clamp it to 1-100 and fall back to the default of 50 when it is missing or not a number, as spaces.ts already does for listSpaces. --- packages/pds/src/xrpc/repo.ts | 5 +++- packages/pds/test/xrpc.test.ts | 50 ++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/packages/pds/src/xrpc/repo.ts b/packages/pds/src/xrpc/repo.ts index c34ddda..14e864c 100644 --- a/packages/pds/src/xrpc/repo.ts +++ b/packages/pds/src/xrpc/repo.ts @@ -266,7 +266,10 @@ 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, { diff --git a/packages/pds/test/xrpc.test.ts b/packages/pds/test/xrpc.test.ts index 0963a3d..403dcd2 100644 --- a/packages/pds/test/xrpc.test.ts +++ b/packages/pds/test/xrpc.test.ts @@ -583,6 +583,56 @@ 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; + }; + } + + 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("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( From 19fa5cca76655aac74c16247bb3bb80703e55fd1 Mon Sep 17 00:00:00 2001 From: Arjun Krishna Date: Mon, 5 Oct 2026 06:50:43 +0530 Subject: [PATCH 2/2] fix(pds): page listRecords without repeating or stalling listRecords started each page with repo.walkRecords(cursor). The cursor was the last record of the previous page, and MST.walkFrom yields its start key twice when that key exists, so the boundary record came back up to three times across pages. With limit=2 a page was just that record twice, so the cursor never moved. reverse=true walked forwards and reversed the page afterwards, which made the cursor the smallest key of the page, so following it returned the same page forever. Walk the collection's MST range directly instead, in either direction, excluding the cursor key and skipping subtrees outside the range. Only the records on the page are read. Records are listed newest first by default and oldest first with reverse=true, matching the reference PDS; the order used to be the other way round. The cursor is the bare rkey of the last record, as in the reference PDS, and cursors in the old collection/rkey form are still accepted. Listing a collection with no records no longer reads every record after it in the repo. Closes #251 --- .changeset/list-records-pagination.md | 7 ++ packages/pds/src/repo/engine.ts | 44 +++++----- packages/pds/src/repo/list-records.ts | 80 +++++++++++++++++ packages/pds/src/xrpc/repo.ts | 2 +- packages/pds/test/list-records.test.ts | 114 +++++++++++++++++++++++++ packages/pds/test/xrpc.test.ts | 31 +++++++ plans/todo/endpoint-implementation.md | 2 +- 7 files changed, 253 insertions(+), 27 deletions(-) create mode 100644 .changeset/list-records-pagination.md create mode 100644 packages/pds/src/repo/list-records.ts create mode 100644 packages/pds/test/list-records.test.ts diff --git a/.changeset/list-records-pagination.md b/.changeset/list-records-pagination.md new file mode 100644 index 0000000..e281176 --- /dev/null +++ b/.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. diff --git a/packages/pds/src/repo/engine.ts b/packages/pds/src/repo/engine.ts index 960c187..5d00f2a 100644 --- a/packages/pds/src/repo/engine.ts +++ b/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 }; } /** diff --git a/packages/pds/src/repo/list-records.ts b/packages/pds/src/repo/list-records.ts new file mode 100644 index 0000000..77aa2c0 --- /dev/null +++ b/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 { + 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 { + 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); + } + } + } +} diff --git a/packages/pds/src/xrpc/repo.ts b/packages/pds/src/xrpc/repo.ts index 14e864c..aa25ea5 100644 --- a/packages/pds/src/xrpc/repo.ts +++ b/packages/pds/src/xrpc/repo.ts @@ -274,7 +274,7 @@ export async function listRecords( const result = await accountDO.repo().listRecords(collection, { limit, - cursor, + cursor: cursor || undefined, reverse, }); diff --git a/packages/pds/test/list-records.test.ts b/packages/pds/test/list-records.test.ts new file mode 100644 index 0000000..1a8740e --- /dev/null +++ b/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([]); + }); +}); diff --git a/packages/pds/test/xrpc.test.ts b/packages/pds/test/xrpc.test.ts index 403dcd2..a9caf34 100644 --- a/packages/pds/test/xrpc.test.ts +++ b/packages/pds/test/xrpc.test.ts @@ -601,6 +601,20 @@ describe("XRPC Endpoints", () => { }; } + 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( @@ -626,6 +640,23 @@ describe("XRPC Endpoints", () => { 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); diff --git a/plans/todo/endpoint-implementation.md b/plans/todo/endpoint-implementation.md index 59b9706..af2e66b 100644 --- a/plans/todo/endpoint-implementation.md +++ b/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 |