fix(graph): stop repeated enumeration cursors

.changeset/fix-enumeration-pagination.md

@@ -0,0 +1,5 @@
+---
+"@atproto/bsky": patch
+---
+
+Stop follows and list-member enumeration at terminal pages and refill results removed by filtering.

packages/bsky/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@atproto/bsky",
-  "version": "0.0.270",
+  "version": "0.0.271",
   "license": "MIT",
   "description": "Reference implementation of app.bsky App View (Bluesky API)",
   "keywords": [

packages/bsky/src/api/app/bsky/graph/getFollows.ts

@@ -16,7 +16,7 @@ import {
   createPipeline,
 } from '../../../../pipeline.js'
 import type { Views } from '../../../../views/index.js'
-import { clearlyBadCursor, resHeaders } from '../../../util.js'
+import { clearlyBadCursor, fillPage, resHeaders } from '../../../util.js'
 
 export default function (server: Server, ctx: AppContext) {
   const getFollows = createPipeline(skeleton, hydration, noBlocks, presentation)
@@ -34,7 +34,13 @@ export default function (server: Server, ctx: AppContext) {
       })
 
       // @TODO ensure canViewTakedowns gets threaded through and applied properly
-      const result = await getFollows({ ...params, hydrateCtx }, ctx)
+      const result = await fillPage({
+        cursor: params.cursor,
+        limit: params.limit,
+        fetch: ({ cursor, limit }) =>
+          getFollows({ ...params, cursor, limit, hydrateCtx }, ctx),
+        items: (r) => r.follows,
+      })
 
       return {
         encoding: 'application/json',

packages/bsky/src/api/app/bsky/graph/getList.ts

@@ -19,7 +19,7 @@ import {
 import type { ListItemInfo } from '../../../../proto/bsky_pb.js'
 import { uriToDid as didFromUri } from '../../../../util/uris.js'
 import type { Views } from '../../../../views/index.js'
-import { clearlyBadCursor, resHeaders } from '../../../util.js'
+import { clearlyBadCursor, fillPage, resHeaders } from '../../../util.js'
 
 export default function (server: Server, ctx: AppContext) {
   const getList = createPipeline(skeleton, hydration, noBlocks, presentation)
@@ -36,7 +36,13 @@ export default function (server: Server, ctx: AppContext) {
         includeTakedowns,
         skipViewerBlocks,
       })
-      const result = await getList({ ...params, hydrateCtx }, ctx)
+      const result = await fillPage({
+        cursor: params.cursor,
+        limit: params.limit,
+        fetch: ({ cursor, limit }) =>
+          getList({ ...params, cursor, limit, hydrateCtx }, ctx),
+        items: (r) => r.items,
+      })
       return {
         encoding: 'application/json',
         body: result,

packages/bsky/src/api/util.test.ts

@@ -0,0 +1,40 @@
+import { describe, expect, it, vi } from 'vitest'
+import { fillPage } from './util.js'
+
+describe('fillPage', () => {
+  it('returns a terminal short page without refilling', async () => {
+    const fetch = vi
+      .fn()
+      .mockResolvedValue({ items: [1], cursor: undefined, metadata: 'first' })
+
+    await expect(
+      fillPage({ cursor: undefined, limit: 3, fetch, items: (r) => r.items }),
+    ).resolves.toEqual({ items: [1], cursor: undefined, metadata: 'first' })
+    expect(fetch).toHaveBeenCalledOnce()
+    expect(fetch).toHaveBeenCalledWith({ cursor: undefined, limit: 3 })
+  })
+
+  it('fills across filtered and empty pages', async () => {
+    const fetch = vi
+      .fn()
+      .mockResolvedValueOnce({ items: [1], cursor: 'a', metadata: 'first' })
+      .mockResolvedValueOnce({ items: [], cursor: 'b' })
+      .mockResolvedValueOnce({ items: [2, 3], cursor: 'c' })
+
+    await expect(
+      fillPage({ cursor: 'start', limit: 3, fetch, items: (r) => r.items }),
+    ).resolves.toEqual({ items: [1, 2, 3], cursor: 'c', metadata: 'first' })
+    expect(fetch).toHaveBeenNthCalledWith(1, { cursor: 'start', limit: 3 })
+    expect(fetch).toHaveBeenNthCalledWith(2, { cursor: 'a', limit: 2 })
+    expect(fetch).toHaveBeenNthCalledWith(3, { cursor: 'b', limit: 2 })
+  })
+
+  it('stops on a repeated cursor', async () => {
+    const fetch = vi.fn().mockResolvedValue({ items: [], cursor: 'a' })
+
+    await expect(
+      fillPage({ cursor: undefined, limit: 1, fetch, items: (r) => r.items }),
+    ).resolves.toEqual({ items: [], cursor: undefined })
+    expect(fetch).toHaveBeenCalledTimes(2)
+  })
+})

packages/bsky/src/api/util.ts

@@ -29,6 +29,51 @@ export const clearlyBadCursor = (cursor?: string) => {
   return !!cursor?.includes('::')
 }
 
+const DEFAULT_FILL_PAGE_MAX_REQUESTS = 10
+
+type PageFetch<R extends { cursor?: string }> = (params: {
+  cursor?: string
+  limit: number
+}) => Promise<R>
+
+export const fillPage = async <
+  F extends PageFetch<{ cursor?: string }>,
+  T,
+>(opts: {
+  cursor?: string
+  limit: number
+  maxRequests?: number
+  fetch: F
+  items: (result: Awaited<ReturnType<F>>) => T[]
+}): Promise<Awaited<ReturnType<F>>> => {
+  const maxRequests = opts.maxRequests ?? DEFAULT_FILL_PAGE_MAX_REQUESTS
+  const result = (await opts.fetch({
+    cursor: opts.cursor,
+    limit: opts.limit,
+  })) as Awaited<ReturnType<F>>
+  const items = opts.items(result)
+  let cursor = result.cursor
+  for (
+    let requests = 1;
+    requests < maxRequests && cursor && items.length < opts.limit;
+    requests++
+  ) {
+    const previousCursor = cursor
+    const page = (await opts.fetch({
+      cursor,
+      limit: opts.limit - items.length,
+    })) as Awaited<ReturnType<F>>
+    items.push(...opts.items(page))
+    cursor = page.cursor
+    if (cursor === previousCursor) {
+      cursor = undefined
+      break
+    }
+    if (!cursor) break
+  }
+  return { ...result, cursor } as Awaited<ReturnType<F>>
+}
+
 // @TEMPORARY backdoor to force search v2 via a request header, gated by the
 // BSKY_SEARCH_V2_OVERRIDE_HEADER env var. Remove once search v2 is fully rolled out.
 const SEARCH_V2_OVERRIDE_HEADER = 'x-bsky-search-v2-override'

packages/bsky/src/data-plane/server/db/pagination.ts

@@ -36,6 +36,19 @@ export abstract class GenericKeyset<R, LR extends KeysetLabeledResult> {
     if (!result) return
     return this.pack(this.labelResult(result))
   }
+  page<Result extends R>(
+    results: Result[],
+    limit: number,
+  ): { items: Result[]; cursor?: string } {
+    const items = results.slice(0, limit)
+    return {
+      items,
+      cursor:
+        results.length > limit && items.length
+          ? this.packFromResult(items[items.length - 1])
+          : undefined,
+    }
+  }
   pack(labeled?: LR): string | undefined {
     if (!labeled) return
     const cursor = this.labeledResultToCursor(labeled)

packages/bsky/src/data-plane/server/routes/follows.ts

@@ -109,21 +109,21 @@ export default (db: Database): Partial<ServiceImpl<typeof Service>> => ({
 
     const keyset = new TimeCidKeyset(ref('follow.sortAt'), ref('follow.cid'))
     followsReq = paginate(followsReq, {
-      limit,
+      limit: limit + 1,
       cursor,
       keyset,
       tryIndex: true,
     })
 
-    const follows = await followsReq.execute()
+    const page = keyset.page(await followsReq.execute(), limit)
 
     return {
-      follows: follows.map((f) => ({
+      follows: page.items.map((f) => ({
         uri: f.uri,
         actorDid: f.creatorDid,
         subjectDid: f.subjectDid,
       })),
-      cursor: keyset.packFromResult(follows),
+      cursor: page.cursor,
     }
   },
 

packages/bsky/src/data-plane/server/routes/lists.ts

@@ -41,19 +41,19 @@ export default (db: Database): Partial<ServiceImpl<typeof Service>> => ({
     )
 
     builder = paginate(builder, {
-      limit,
+      limit: limit + 1,
       cursor,
       keyset,
       tryIndex: true,
     })
 
-    const listItems = await builder.execute()
+    const page = keyset.page(await builder.execute(), limit)
     return {
-      listitems: listItems.map((item) => ({
+      listitems: page.items.map((item) => ({
         uri: item.uri,
         did: item.subjectDid,
       })),
-      cursor: keyset.packFromResult(listItems),
+      cursor: page.cursor,
     }
   },
 

packages/bsky/tests/data-plane/__snapshots__/indexing.test.ts.snap

@@ -579,7 +579,6 @@ exports[`indexing > indexRepo > updates indexes when records change. 1`] = `
     ],
   },
   {
-    "cursor": "0000000000000__bafycid",
     "follows": [
       {
         "associated": {

packages/bsky/tests/views/__snapshots__/block-lists.test.ts.snap

@@ -560,7 +560,6 @@ exports[`pds views with blocking from block lists > returns lists associated wit
 
 exports[`pds views with blocking from block lists > returns the contents of a list 1`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "items": [
     {
       "subject": {

packages/bsky/tests/views/__snapshots__/follows.test.ts.snap

@@ -423,7 +423,6 @@ exports[`pds follow views > fetches followers 5`] = `
 
 exports[`pds follow views > fetches follows 1`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "follows": [
     {
       "associated": {
@@ -544,7 +543,6 @@ exports[`pds follow views > fetches follows 1`] = `
 
 exports[`pds follow views > fetches follows 2`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "follows": [
     {
       "associated": {
@@ -619,7 +617,6 @@ exports[`pds follow views > fetches follows 2`] = `
 
 exports[`pds follow views > fetches follows 3`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "follows": [
     {
       "associated": {
@@ -671,7 +668,6 @@ exports[`pds follow views > fetches follows 3`] = `
 
 exports[`pds follow views > fetches follows 4`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "follows": [
     {
       "associated": {
@@ -769,7 +765,6 @@ exports[`pds follow views > fetches follows 4`] = `
 
 exports[`pds follow views > fetches follows 5`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "follows": [
     {
       "associated": {

packages/bsky/tests/views/__snapshots__/mute-lists.test.ts.snap

@@ -580,7 +580,6 @@ exports[`bsky views with mutes from mute lists > returns lists associated with a
 
 exports[`bsky views with mutes from mute lists > returns the contents of a list 1`] = `
 {
-  "cursor": "0000000000000__bafycid",
   "items": [
     {
       "subject": {

packages/bsky/tests/views/follows.test.ts

@@ -319,6 +319,82 @@ describe('pds follow views', () => {
     expect(results(paginatedAll)).toEqual(results([full.data]))
   })
 
+  it('only returns a follows cursor when another raw row exists', async () => {
+    const terminal = await network.bsky.ctx.dataplane.getFollows({
+      actorDid: alice,
+      limit: 4,
+    })
+    const nonterminal = await network.bsky.ctx.dataplane.getFollows({
+      actorDid: alice,
+      limit: 2,
+    })
+
+    expect(terminal.follows).toHaveLength(4)
+    expect(terminal.cursor).toBe('')
+    expect(nonterminal.follows).toHaveLength(2)
+    expect(nonterminal.cursor).not.toBe('')
+  })
+
+  it('fills a page after filtering follows', async () => {
+    const subject = await sc.createAccount('follows-fill-subject', {
+      handle: 'follows-fill-sub.test',
+      email: 'follows-fill-subject@example.com',
+      password: 'hunter2',
+    })
+    const olderVisible = await sc.createAccount('follows-fill-visible-older', {
+      handle: 'follows-fill-old.test',
+      email: 'follows-fill-visible-older@example.com',
+      password: 'hunter2',
+    })
+    const newerVisible = await sc.createAccount('follows-fill-visible-newer', {
+      handle: 'follows-fill-new.test',
+      email: 'follows-fill-visible-newer@example.com',
+      password: 'hunter2',
+    })
+    const takenDown = await sc.createAccount('follows-fill-taken-down', {
+      handle: 'follows-fill-down.test',
+      email: 'follows-fill-taken-down@example.com',
+      password: 'hunter2',
+    })
+    const blocked = await sc.createAccount('follows-fill-blocked', {
+      handle: 'follows-fill-block.test',
+      email: 'follows-fill-blocked@example.com',
+      password: 'hunter2',
+    })
+
+    await sc.follow(subject.did, olderVisible.did, {
+      createdAt: '2025-02-01T00:00:00.000Z',
+    })
+    await sc.follow(subject.did, newerVisible.did, {
+      createdAt: '2025-02-02T00:00:00.000Z',
+    })
+    await sc.follow(subject.did, takenDown.did, {
+      createdAt: '2025-02-03T00:00:00.000Z',
+    })
+    await sc.follow(subject.did, blocked.did, {
+      createdAt: '2025-02-04T00:00:00.000Z',
+    })
+    await sc.block(subject.did, blocked.did)
+    await network.processAll()
+    await network.bsky.ctx.dataplane.takedownActor({ did: takenDown.did })
+
+    const res = await agent.api.app.bsky.graph.getFollows(
+      { actor: subject.did, limit: 2 },
+      {
+        headers: await network.serviceHeaders(
+          subject.did,
+          ids.AppBskyGraphGetFollows,
+        ),
+      },
+    )
+
+    expect(res.data.follows.map((follow) => follow.did)).toEqual([
+      newerVisible.did,
+      olderVisible.did,
+    ])
+    expect(res.data.cursor).toBeUndefined()
+  })
+
   it('fetches follows unauthed', async () => {
     const { data: authed } = await agent.api.app.bsky.graph.getFollows(
       { actor: sc.dids.alice },