From 3e83b62df39fb1fc8cb3df8c00a98b4ba0cc9e8c Mon Sep 17 00:00:00 2001 From: Nate Moore Date: Fri, 6 Nov 2026 01:30:45 -0500 Subject: [PATCH] Stop paging on repeated or empty cursors, with a page cap Every listRecords and backlink loop ended only when the server stopped sending a cursor, so a PDS or Constellation that repeated one kept the caller looping until the tab ran out of memory. paginate() stops at an empty page, a missing or already-seen cursor, or 100 pages, and listRecords yields each record key once. --- packages/sdk/src/paging.ts | 23 +++++++++++ packages/sdk/src/records.ts | 22 +++++----- packages/sdk/src/social/util.ts | 22 +++++----- packages/sdk/tests/paging.test.ts | 69 +++++++++++++++++++++++++++++++ 4 files changed, 116 insertions(+), 20 deletions(-) create mode 100644 packages/sdk/src/paging.ts create mode 100644 packages/sdk/tests/paging.test.ts diff --git a/packages/sdk/src/paging.ts b/packages/sdk/src/paging.ts new file mode 100644 index 0000000..586dae1 --- /dev/null +++ b/packages/sdk/src/paging.ts @@ -0,0 +1,23 @@ +/** How many pages a listing reads before it stops. At 100 records a page, that is 10,000 records. */ +export const MAX_PAGES = 100; + +export interface Page { + items: readonly T[]; + cursor?: string | null; +} + +/** + * Yields every item across pages. Stops at an empty page, a missing cursor, a cursor the server already + * sent, or after `maxPages`, so a server that repeats or cycles its cursor cannot keep the caller looping. + */ +export async function* paginate(fetchPage: (cursor: string | undefined) => Promise>, maxPages = MAX_PAGES): AsyncGenerator { + const seen = new Set(); + let cursor: string | undefined; + for (let page = 0; page < maxPages; page++) { + const { items, cursor: next } = await fetchPage(cursor); + yield* items; + if (items.length === 0 || !next || seen.has(next)) return; + seen.add(next); + cursor = next; + } +} diff --git a/packages/sdk/src/records.ts b/packages/sdk/src/records.ts index 623c4bd..729f52e 100644 --- a/packages/sdk/src/records.ts +++ b/packages/sdk/src/records.ts @@ -3,6 +3,7 @@ import type { Client } from '@atcute/client'; import type { Cid, Did, Nsid, ResourceUri } from '@atcute/lexicons'; import { isCanonicalResourceUri, parseCanonicalResourceUri } from '@atcute/lexicons'; import { SdkError } from './errors.ts'; +import { paginate } from './paging.ts'; /** * A record read from a PDS. `uri`, `did`, and `rkey` come from the request, never from the response: @@ -25,19 +26,20 @@ export async function getRecord(client: Client, did: Did, collection: Nsid, rkey return { ok: true, record: { uri: recordUri(did, collection, rkey), did, rkey, cid: res.data.cid, value: res.data.value } }; } -/** Every record in `did`'s `collection`, skipping any whose URI names another repo or collection. */ +/** Every record in `did`'s `collection` once, skipping any whose URI names another repo or collection. */ export async function* listRecords(client: Client, did: Did, collection: Nsid): AsyncGenerator { - let cursor: string | undefined; - do { + const items = paginate(async (cursor) => { const res = await client.get('com.atproto.repo.listRecords', { params: { repo: did, collection, limit: 100, cursor } }); if (!res.ok) throw new SdkError(`listRecords ${collection} on ${did} failed: ${res.data.error}`, 'ERR_NIB_RECORD_INVALID'); - for (const item of res.data.records) { - const rkey = rkeyOf(item.uri, did, collection); - if (rkey === undefined) continue; - yield { uri: recordUri(did, collection, rkey), did, rkey, cid: item.cid, value: item.value }; - } - cursor = res.data.records.length > 0 ? res.data.cursor : undefined; - } while (cursor); + return { items: res.data.records, cursor: res.data.cursor }; + }); + const seen = new Set(); + for await (const item of items) { + const rkey = rkeyOf(item.uri, did, collection); + if (rkey === undefined || seen.has(rkey)) continue; + seen.add(rkey); + yield { uri: recordUri(did, collection, rkey), did, rkey, cid: item.cid, value: item.value }; + } } /** The record key of `uri` when it names a record in `did`'s `collection`, else undefined. */ diff --git a/packages/sdk/src/social/util.ts b/packages/sdk/src/social/util.ts index 7afa838..fbd23bd 100644 --- a/packages/sdk/src/social/util.ts +++ b/packages/sdk/src/social/util.ts @@ -2,6 +2,7 @@ import type { Did } from '@atcute/lexicons'; import { isCanonicalResourceUri, parseCanonicalResourceUri } from '@atcute/lexicons'; import { SdkError } from '../errors.ts'; import { getBacklinks, type Backlink } from '../microcosm.ts'; +import { paginate } from '../paging.ts'; import type { Session } from '../session.ts'; /** Parses an AT-URI the session must own, in the given collection. */ @@ -18,16 +19,17 @@ export function parseOwnUri(session: Session, uri: string, collection: string): * fails because Constellation is unset or down; a later page failing just stops the pagination. */ export async function allBacklinks(subject: string, source: string, cap = 1000): Promise { - const first = await getBacklinks({ subject, source, limit: 100 }); - if (!first) return undefined; const collection = source.slice(0, source.indexOf(':')); - const out = first.records.filter((link) => link.collection === collection); - let cursor = first.cursor; - while (cursor && out.length < cap) { - const page = await getBacklinks({ subject, source, limit: 100, cursor }); - if (!page) break; - out.push(...page.records.filter((link) => link.collection === collection)); - cursor = page.cursor; + let reached = false; + const links = paginate(async (cursor) => { + const page = await getBacklinks({ subject, source, limit: 100, ...(cursor ? { cursor } : {}) }); + if (page) reached = true; + return page ? { items: page.records, cursor: page.cursor } : { items: [] }; + }); + const out: Backlink[] = []; + for await (const link of links) { + if (link.collection === collection) out.push(link); + if (out.length >= cap) break; } - return out; + return reached ? out : undefined; } diff --git a/packages/sdk/tests/paging.test.ts b/packages/sdk/tests/paging.test.ts new file mode 100644 index 0000000..f62b16b --- /dev/null +++ b/packages/sdk/tests/paging.test.ts @@ -0,0 +1,69 @@ +import assert from 'node:assert/strict'; +import { afterEach, test } from 'node:test'; +import type { FetchHandler } from '@atcute/client'; +import { configure, createProject, likers, listProjects, resolveRepo, TangledError } from '../src/index.ts'; +import { createFakePds, fakeNetwork } from './fake-pds.ts'; + +const pds = createFakePds({ url: 'https://looping-pds.test' }); +const alice = pds.createAccount('alice'); +configure(fakeNetwork([pds])); + +afterEach(() => { + delete pds.tamper.cursor; + pds.resetCounts(); + configure(fakeNetwork([pds])); +}); + +const files = new Map([['a.txt', new TextEncoder().encode('a')]]); +await createProject(pds.session(alice), { name: 'one', files }); +await createProject(pds.session(alice), { name: 'two', files }); + +test('listProjects stops when the PDS repeats its cursor', { timeout: 5000 }, async () => { + pds.tamper.cursor = () => 'zzzz'; + assert.deepEqual((await listProjects(alice)).map((p) => p.record.name), ['two', 'one']); + assert.equal(pds.count('com.atproto.repo.listRecords'), 2); +}); + +test('listProjects stops when the PDS cycles between two cursors', { timeout: 5000 }, async () => { + let n = 0; + pds.tamper.cursor = () => (n++ % 2 === 0 ? 'zzzz' : 'zzzy'); + assert.deepEqual((await listProjects(alice)).map((p) => p.record.name), ['two', 'one']); + assert.equal(pds.count('com.atproto.repo.listRecords'), 3); +}); + +test('listProjects reads at most 100 pages from a PDS that always sends a new cursor', { timeout: 5000 }, async () => { + let n = 0; + pds.tamper.cursor = () => `zz${n++}`; + assert.deepEqual((await listProjects(alice)).map((p) => p.record.name), ['two', 'one']); + assert.equal(pds.count('com.atproto.repo.listRecords'), 100); +}); + +test('resolveRepo stops scanning legacy records when the cursor repeats', { timeout: 5000 }, async () => { + await pds.session(alice).client.post('com.atproto.repo.createRecord', { + input: { repo: alice, collection: 'sh.tangled.repo', record: { $type: 'sh.tangled.repo', knot: 'knot.test', name: 'other', createdAt: '2026-01-01T00:00:00Z' } }, + }); + pds.tamper.cursor = () => 'zzzz'; + await assert.rejects(resolveRepo(`${alice}/missing`), (e: unknown) => e instanceof TangledError && e.code === 'repo-not-found'); +}); + +test('backlink pagination stops on an empty page that still carries a cursor', { timeout: 5000 }, async () => { + let calls = 0; + const constellation: FetchHandler = async () => { + calls++; + return Response.json({ total: 10, records: calls === 1 ? [{ did: 'did:plc:bob', collection: 'run.nibling.feed.like', rkey: '3m2abcdefghij' }] : [], cursor: 'again' }); + }; + configure({ ...fakeNetwork([pds], { 'https://constellation.test': constellation }), constellation: 'https://constellation.test' }); + assert.deepEqual(await likers('at://did:plc:alice/run.nibling.project/3m2abcdefghij'), ['did:plc:bob']); + assert.equal(calls, 2); +}); + +test('backlink pagination stops when Constellation repeats a cursor', { timeout: 5000 }, async () => { + let calls = 0; + const constellation: FetchHandler = async () => { + calls++; + return Response.json({ total: 10, records: [{ did: `did:plc:liker${calls}`, collection: 'run.nibling.feed.like', rkey: '3m2abcdefghij' }], cursor: 'again' }); + }; + configure({ ...fakeNetwork([pds], { 'https://constellation.test': constellation }), constellation: 'https://constellation.test' }); + assert.deepEqual(await likers('at://did:plc:alice/run.nibling.project/3m2abcdefghij'), ['did:plc:liker1', 'did:plc:liker2']); + assert.equal(calls, 2); +}); -- knotmirror 0.0.1