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 c34ddda..aa25ea5 100644 --- a/packages/pds/src/xrpc/repo.ts +++ b/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, }); 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 0963a3d..a9caf34 100644 --- a/packages/pds/test/xrpc.test.ts +++ b/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( 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 |