Merge pull request #1256 from sij411/fix/backfill-follow

Fix `@fedify/backfill` not follow collection pages during conversation backfill

Author: dahlia

CHANGES.md

@@ -8,6 +8,16 @@ Version 2.3.12
 
 To be released.
 
+### @fedify/backfill
+
+ -  Fixed conversation backfill skipping posts in paginated context and replies
+    collections.  Backfill now follows the first and subsequent pages while
+    respecting traversal limits, and retains posts already found if a page
+    cannot be loaded.  [[#1248], [#1256] by Jiwon Kwon\]
+
+[#1248]: https://github.com/fedify-dev/fedify/issues/1248
+[#1256]: https://github.com/fedify-dev/fedify/pull/1256
+
 
 Version 2.3.11
 --------------

changes.d/backfill/follow-collection-pages.md

@@ -0,0 +1,9 @@
+---
+links:
+  '#1248': https://github.com/fedify-dev/fedify/issues/1248
+  '#1256': https://github.com/fedify-dev/fedify/pull/1256
+---
+ -  Fixed conversation backfill skipping posts in paginated context and replies
+    collections.  Backfill now follows the first and subsequent pages while
+    respecting traversal limits, and retains posts already found if a page
+    cannot be loaded.  [[#1248], [#1256] by Jiwon Kwon]

packages/backfill/README.md

@@ -65,6 +65,14 @@ for await (
 The seed object itself is not yielded.  If it appears in the discovered
 collection, it is skipped by ID.
 
+Context and `replies` collections support inline items and pagination through
+`first` and successive `next` links.  A collection page used as the starting
+collection also follows `next`.  Pages are loaded as needed, and traversal
+stops on repeated page links.  Pagination does not increase reply-tree depth.
+If a page cannot be loaded, items already yielded are retained and later
+configured strategies can continue unless cancellation or traversal limits
+stop them.
+
 Configured strategies run in order.  They share `maxItems`, `maxRequests`,
 abort state, and object ID deduplication; if two strategies discover the same
 object, the first strategy keeps its `BackfillItem` metadata.
@@ -133,7 +141,7 @@ All configured strategies share the same traversal controls:
  -  `maxItems` limits the number of yielded objects.  Skipped duplicates do
     not count.
  -  `maxRequests` limits calls to `documentLoader`.  Embedded objects and
-    collections do not count.
+    collections, including embedded pages, do not count.
  -  `maxDepth` limits reply-tree traversal and defaults to 10.  It does not
     limit context collection items.
  -  `interval` adds a delay between loader requests.  Its callback receives

packages/backfill/src/backfill.test.ts

@@ -1,7 +1,15 @@
 import { deepStrictEqual, ok, rejects, strictEqual } from "node:assert/strict";
 import test, { describe } from "node:test";
 import { backfill, type BackfillContext, MaxRequestsExceeded } from "./mod.ts";
-import { Announce, Collection, Create, Note } from "@fedify/vocab";
+import {
+  Announce,
+  Collection,
+  CollectionPage,
+  Create,
+  Note,
+  OrderedCollection,
+  OrderedCollectionPage,
+} from "@fedify/vocab";
 
 async function collect(
   context: BackfillContext,
@@ -1772,3 +1780,251 @@ describe("backfill", () => {
     deepStrictEqual(iterations, [0, 1]);
   });
 });
+
+describe("collection pagination", () => {
+  const iri = (name: string) => new URL(`https://example.com/${name}`);
+
+  for (const ordered of [false, true]) {
+    for (const replies of [false, true]) {
+      for (const embedded of [false, true]) {
+        test(`${ordered ? "ordered" : "unordered"} ${replies ? "replies" : "context"} with ${embedded ? "embedded" : "linked"} pages`, async () => {
+          const CollectionType = ordered ? OrderedCollection : Collection;
+          const PageType = ordered ? OrderedCollectionPage : CollectionPage;
+          const a = new Note({ id: iri("a") });
+          const b = new Note({ id: iri("b") });
+          const second = new PageType({ id: iri("page-2"), items: [b] });
+          const first = new PageType({
+            id: iri("page-1"),
+            items: [a],
+            next: embedded ? second : second.id,
+          });
+          const collection = new CollectionType({
+            id: iri("thread"),
+            first: embedded ? first : first.id,
+          });
+          const seed = new Note({
+            id: iri("seed"),
+            contexts: replies ? [] : [iri("thread")],
+            replies: replies ? collection : null,
+          });
+          const documents = new Map<string, Collection>([
+            [iri("thread").href, collection],
+            [iri("page-1").href, first],
+            [iri("page-2").href, second],
+          ]);
+          const calls: string[] = [];
+          const context: BackfillContext = {
+            documentLoader: (url) => {
+              calls.push(url.href);
+              return Promise.resolve(documents.get(url.href) ?? null);
+            },
+          };
+          const items = await collect(context, seed, {
+            strategies: replies ? ["reply-tree"] : ["context-auto"],
+            maxDepth: 1,
+            maxRequests: embedded ? (replies ? 0 : 1) : 3,
+          });
+          deepStrictEqual(items.map((item) => item.id?.href), [
+            a.id?.href,
+            b.id?.href,
+          ]);
+          deepStrictEqual(
+            items.map((item) => item.depth),
+            replies ? [1, 1] : [0, 0],
+          );
+          strictEqual(
+            calls.length,
+            embedded ? (replies ? 0 : 1) : (replies ? 2 : 3),
+          );
+        });
+      }
+    }
+  }
+
+  test("starting page follows next and stops before loading a repeated page", async () => {
+    const first = new OrderedCollectionPage({
+      id: iri("page-1"),
+      items: [new Note({ id: iri("a") })],
+      next: iri("page-2"),
+    });
+    const second = new OrderedCollectionPage({
+      id: iri("page-2"),
+      items: [new Note({ id: iri("b") })],
+      next: first.id,
+    });
+    const calls: string[] = [];
+    const items = await collect({
+      documentLoader: (url) => {
+        calls.push(url.href);
+        return Promise.resolve(url.href === first.id?.href ? first : second);
+      },
+    }, new Note({ contexts: [first.id!] }));
+    deepStrictEqual(items.map((item) => item.id?.href), [
+      iri("a").href,
+      iri("b").href,
+    ]);
+    deepStrictEqual(calls, [first.id?.href, second.id?.href]);
+  });
+
+  for (const limit of ["maxItems", "maxRequests"] as const) {
+    test(`${limit} stops between pages`, async () => {
+      const first = new CollectionPage({
+        id: iri("page-1"),
+        items: [new Note({ id: iri("a") })],
+        next: iri("page-2"),
+      });
+      const calls: string[] = [];
+      const items = await collect(
+        {
+          documentLoader: (url) => {
+            calls.push(url.href);
+            return Promise.resolve(first);
+          },
+        },
+        new Note({
+          contexts: [first.id!],
+          replies: new Collection({
+            items: [new Note({ id: iri("fallback") })],
+          }),
+        }),
+        { [limit]: 1, strategies: ["context-auto", "reply-tree"] },
+      );
+      deepStrictEqual(items.map((item) => item.id?.href), [iri("a").href]);
+      deepStrictEqual(calls, [first.id?.href]);
+    });
+  }
+
+  for (const failure of ["missing", "throw", "invalid"] as const) {
+    test(`${failure} page retains items and allows the next strategy`, async () => {
+      const seed = new Note({
+        contexts: [iri("thread")],
+        replies: new Collection({ items: [new Note({ id: iri("reply") })] }),
+      });
+      const first = new CollectionPage({
+        items: [new Note({ id: iri("a") })],
+        next: iri("broken"),
+      });
+      const items = await collect(
+        {
+          documentLoader: (url) => {
+            if (url.href === iri("thread").href) {
+              return Promise.resolve(new Collection({ first }));
+            }
+            if (failure === "throw") throw new Error("page failed");
+            return Promise.resolve(failure === "missing" ? null : new Note({}));
+          },
+        },
+        seed,
+        { strategies: ["context-auto", "reply-tree"] },
+      );
+      deepStrictEqual(items.map((item) => item.id?.href), [
+        iri("a").href,
+        iri("reply").href,
+      ]);
+    });
+  }
+
+  test("inline items precede page items", async () => {
+    const items = await collect({
+      documentLoader: () =>
+        Promise.resolve(
+          new Collection({
+            items: [new Note({ id: iri("inline") })],
+            first: new CollectionPage({
+              items: [new Note({ id: iri("paged") })],
+            }),
+          }),
+        ),
+    }, new Note({ contexts: [iri("thread")] }));
+    deepStrictEqual(items.map((item) => item.id?.href), [
+      iri("inline").href,
+      iri("paged").href,
+    ]);
+  });
+
+  test("cancellation during a page load prevents later strategies", async () => {
+    const controller = new AbortController();
+    const reason = new Error("cancelled");
+    const seed = new Note({
+      contexts: [iri("thread")],
+      replies: new Collection({ items: [new Note({ id: iri("fallback") })] }),
+    });
+    await rejects(
+      collect(
+        {
+          documentLoader: (url, options) => {
+            strictEqual(options?.signal, controller.signal);
+            if (url.href === iri("thread").href) {
+              return Promise.resolve(new Collection({ first: iri("page") }));
+            }
+            controller.abort(reason);
+            throw reason;
+          },
+        },
+        seed,
+        {
+          signal: controller.signal,
+          strategies: ["context-auto", "reply-tree"],
+        },
+      ),
+      (error) => error === reason,
+    );
+  });
+});
+
+test("page loads share the cache and interval across strategies", async () => {
+  const iri = (name: string) => new URL(`https://example.com/${name}`);
+  const collection = new Collection({ first: iri("page") });
+  const page = new CollectionPage({ items: [new Note({ id: iri("a") })] });
+  const seed = new Note({
+    contexts: [iri("thread")],
+    replies: new Collection({ first: iri("page") }),
+  });
+  const calls: string[] = [];
+  const intervals: number[] = [];
+  const items = await collect(
+    {
+      documentLoader: (url) => {
+        calls.push(url.href);
+        return Promise.resolve(
+          url.href === iri("thread").href ? collection : page,
+        );
+      },
+    },
+    seed,
+    {
+      strategies: ["context-auto", "reply-tree"],
+      maxRequests: 2,
+      interval: (count) => {
+        intervals.push(count);
+        return { milliseconds: 0 };
+      },
+    },
+  );
+  deepStrictEqual(items.map((item) => item.id?.href), [iri("a").href]);
+  deepStrictEqual(calls, [iri("thread").href, iri("page").href]);
+  deepStrictEqual(intervals, [0, 1]);
+});
+
+test("embedded pages without IDs follow next without loader calls", async () => {
+  const a = new Note({ id: new URL("https://example.com/a") });
+  const b = new Note({ id: new URL("https://example.com/b") });
+  const seed = new Note({
+    replies: new Collection({
+      first: new CollectionPage({
+        items: [a],
+        next: new CollectionPage({ items: [b] }),
+      }),
+    }),
+  });
+  const items = await collect(
+    {
+      documentLoader: () => {
+        throw new Error("embedded pages must not be fetched");
+      },
+    },
+    seed,
+    { strategies: ["reply-tree"], maxRequests: 0, maxDepth: 1 },
+  );
+  deepStrictEqual(items.map((item) => item.id?.href), [a.id?.href, b.id?.href]);
+});

packages/backfill/src/backfill.ts

@@ -517,8 +517,8 @@ async function* getCollectionItems(
   budget: RequestBudget,
   skipIds?: ReadonlySet<string>,
 ): AsyncIterable<APObject | Link> {
-  yield* collection.getItems({
-    documentLoader: async (url) => {
+  const itemOptions = {
+    documentLoader: async (url: string) => {
       return await loadCollectionItemDocument(
         context,
         url,
@@ -527,8 +527,61 @@ async function* getCollectionItems(
         skipIds,
       );
     },
-    crossOrigin: "trust",
-  });
+    crossOrigin: "trust" as const,
+  };
+  const pageOptions = {
+    documentLoader: async (url: string) => {
+      const object = await loadObject(
+        context,
+        new URL(url),
+        options,
+        budget,
+        true,
+      );
+      if (object == null) throw new Error(`Collection page not found: ${url}`);
+      return {
+        contextUrl: null,
+        documentUrl: url,
+        document: await object.toJsonLd(),
+      };
+    },
+    crossOrigin: "trust" as const,
+  };
+  const visitedIds = new Set<string>();
+  const visitedPages = new WeakSet<BackfillCollection>();
+  let current: BackfillCollection | null = collection;
+  while (current != null) {
+    budget.signal?.throwIfAborted();
+    if (visitedPages.has(current)) return;
+    if (current.id != null) {
+      if (visitedIds.has(current.id.href)) return;
+      visitedIds.add(current.id.href);
+    }
+    visitedPages.add(current);
+    yield* current.getItems(itemOptions);
+
+    const page: CollectionPage | OrderedCollectionPage | null =
+      current instanceof CollectionPage ||
+        current instanceof OrderedCollectionPage
+        ? current
+        : null;
+    const pageId: URL | null = page == null ? current.firstId : page.nextId;
+    if (pageId != null && visitedIds.has(pageId.href)) return;
+    try {
+      budget.signal?.throwIfAborted();
+      current = page == null
+        ? await current.getFirst(pageOptions)
+        : await page.getNext(pageOptions);
+    } catch (error) {
+      if (error instanceof MaxRequestsExceeded) throw error;
+      budget.signal?.throwIfAborted();
+      return;
+    }
+    // Remember the requested IRI too, in case the page has a different ID.
+    if (pageId != null && pageId.href !== current?.id?.href) {
+      visitedIds.add(pageId.href);
+    }
+  }
 }
 
 async function getCreateActivityObject(