fix(activitypub,safe-fetch,webfinger): cap outbox scan and dedupe timeout signal

.changeset/activitypub-outbox-scan-cap.md

@@ -0,0 +1,13 @@
+---
+"@dwk/activitypub": patch
+"@dwk/safe-fetch": patch
+"@dwk/webfinger": patch
+---
+
+Cap the number of batches scanned per client-list page — both the outbox
+owner-post merge into a Mastodon timeline and the inbox notifications scan —
+so a like/announce-dominated outbox or a plain-post-dominated inbox can no
+longer force a near-full-table scan per request. Also de-duplicate the
+cancellable timeout-signal helper: `@dwk/safe-fetch` now exports
+`createTimeoutSignal`, reused by `@dwk/activitypub` and `@dwk/webfinger`
+instead of each carrying its own copy.

packages/activitypub/src/delivery.ts

@@ -12,10 +12,13 @@
  * primitive; this module wires it to the network.
  */
 
-import { assertPublicUrl, SsrfError } from "@dwk/safe-fetch";
+import {
+  assertPublicUrl,
+  createTimeoutSignal,
+  SsrfError,
+} from "@dwk/safe-fetch";
 
 import { signRequest, type SignerKey } from "./signature.js";
-import { createTimeoutSignal } from "./timeout.js";
 
 /** Machine-readable cause of a blocked delivery target. */
 export type BlockedReason =
@@ -88,7 +91,7 @@ export async function deliverActivity(
   const signed = await signRequest(inboxUrl, body, signer, { now });
 
   let response: Response;
-  const timeout = createTimeoutSignal(timeoutMs);
+  const timeout = createTimeoutSignal(undefined, timeoutMs);
   try {
     response = await fetchImpl(inboxUrl, {
       method: "POST",

packages/activitypub/src/discovery.ts

@@ -9,10 +9,10 @@
  * @see spec/fediverse-interop.md §2.4
  */
 
+import { createTimeoutSignal } from "@dwk/safe-fetch";
 import { resolveHandle } from "@dwk/webfinger";
 
 import { assertPublicHttpsTarget } from "./delivery.js";
-import { createTimeoutSignal } from "./timeout.js";
 
 /** Whether a string looks like a handle rather than an IRI. */
 export function isHandleShaped(value: string): boolean {
@@ -112,7 +112,7 @@ export async function fetchActorGuarded(
   fetchImpl: typeof fetch,
 ): Promise<ResolvedActor | null> {
   let response: Response;
-  const timeout = createTimeoutSignal(10_000);
+  const timeout = createTimeoutSignal(undefined, 10_000);
   try {
     response = await guardedFetch(fetchImpl)(iri, {
       headers: { accept: "application/activity+json" },

packages/activitypub/src/mastodon-api.test.ts

@@ -420,6 +420,53 @@ describe("buildMastodonBackend", () => {
     ]);
   });
 
+  it("caps the outbox scan instead of exhausting a like-dominated table", async () => {
+    const config = freshConfig();
+    const timestamp = Date.now();
+    // limit:1 => BATCH = max(1*4, 40) = 40; MAX_OUTBOX_SCAN_BATCHES = 25, so
+    // the scan gives up after 1000 rows. Bury a real owner post behind 1000
+    // newer non-post rows so it falls just past that cap.
+    const buriedPost = {
+      id: `${config.iris.outbox}/buried-owner-post`,
+      type: "Create",
+      actor: config.iris.id,
+      object: {
+        id: `${config.iris.outbox}/buried-owner-post/object`,
+        type: "Note",
+        content: "buried owner post",
+      },
+    };
+    const stub = testEnv.ACTOR.get(testEnv.ACTOR.idFromName(config.iris.id));
+    await runInDurableObject(stub, async (_instance, state) => {
+      state.storage.sql.exec(
+        `INSERT INTO outbox (id, json, published_at) VALUES (?, ?, ?)`,
+        buriedPost.id,
+        JSON.stringify(buriedPost),
+        timestamp,
+      );
+      for (let i = 0; i < 1000; i++) {
+        const like = {
+          id: `${config.iris.outbox}/scan-cap-like-${i}`,
+          type: "Like",
+          actor: config.iris.id,
+          object: `https://remote.example/objects/${i}`,
+        };
+        state.storage.sql.exec(
+          `INSERT INTO outbox (id, json, published_at) VALUES (?, ?, ?)`,
+          like.id,
+          JSON.stringify(like),
+          timestamp + i + 1,
+        );
+      }
+    });
+    const backend = buildMastodonBackend({ config, actor: testEnv.ACTOR });
+
+    const page = await backend.timeline({ limit: 1 });
+    expect(
+      page.entries.some((entry) => entry.activity["id"] === buriedPost.id),
+    ).toBe(false);
+  });
+
   it("serves cached actor profile fields without an outbound request", async () => {
     const config = freshConfig();
     const stub = testEnv.ACTOR.get(testEnv.ACTOR.idFromName(config.iris.id));
@@ -510,6 +557,43 @@ describe("buildMastodonBackend", () => {
     );
   });
 
+  it("caps the inbox scan instead of exhausting a plain-post-dominated table", async () => {
+    const config = freshConfig();
+    const timestamp = Date.now();
+    // limit:1 => BATCH = max(1*4, 40) = 40; MAX_SCAN_BATCHES = 25, so the
+    // scan gives up after 1000 rows. Bury a real mention behind 1000 newer
+    // plain (non-notification) rows so it falls just past that cap.
+    const buriedMention = createMention(config);
+    const stub = testEnv.ACTOR.get(testEnv.ACTOR.idFromName(config.iris.id));
+    await runInDurableObject(stub, async (_instance, state) => {
+      state.storage.sql.exec(
+        `INSERT INTO inbox (id, json, received_at, object_type)
+           VALUES (?, ?, ?, 'Note')`,
+        buriedMention["id"] as string,
+        JSON.stringify(buriedMention),
+        timestamp,
+      );
+      for (let i = 0; i < 1000; i++) {
+        const note = createNote(config);
+        state.storage.sql.exec(
+          `INSERT INTO inbox (id, json, received_at, object_type)
+             VALUES (?, ?, ?, 'Note')`,
+          note["id"] as string,
+          JSON.stringify(note),
+          timestamp + i + 1,
+        );
+      }
+    });
+    const backend = buildMastodonBackend({ config, actor: testEnv.ACTOR });
+
+    const page = await backend.notifications({ limit: 1 });
+    expect(
+      page.entries.some(
+        (entry) => entry.activity["id"] === buriedMention["id"],
+      ),
+    ).toBe(false);
+  });
+
   it("timeline() maxId cursor translates to max_received_at/tie_seq and excludes newer rows", async () => {
     const config = freshConfig();
     const first = createNote(config);

packages/activitypub/src/object.ts

@@ -71,6 +71,16 @@ const VOTE_SWEEP_MS = 10 * 60_000;
 const ACTOR_PROFILE_DEBOUNCE_MS = 1_000;
 /** Actor documents are optional display metadata; keep their cache well below a DO SQLite cell. */
 const ACTOR_PROFILE_MAX_BODY_BYTES = 128 * 1024;
+/**
+ * Hard cap on batches scanned per client-list page in `#serveClientList`,
+ * both for the inbox (notifications: favourite/reblog/mention) and for the
+ * outbox owner-post merge into the timeline. Without it, an inbox or outbox
+ * dominated by unwanted activity types (plain Create/Update rows for
+ * notifications; Like/Announce/etc. for the timeline merge) forces a
+ * near-full-table scan per request; past this cap the page simply returns
+ * fewer than `limit` matches rather than exhausting the table.
+ */
+const MAX_SCAN_BATCHES = 25;
 /**
  * Cardinality cap on the pending-metrics table: at most this many distinct
  * `(event, fields)` keys accumulate between drains. Delivery fields include
@@ -1373,7 +1383,16 @@ export class ActivityPubObject extends DurableObject<ActivityPubEnv> {
     // Gating it on `matches.length > 0` instead (as a prior version did) lets
     // a zero-match first batch re-issue the exact same query forever.
     let isFirstBatch = true;
-    while (matches.length < limit && !exhausted) {
+    // Bounded for the same reason as the outbox merge below: a notifications
+    // page (favourite/reblog/mention) over an inbox dominated by plain
+    // Create/Update rows would otherwise scan the whole table.
+    let inboxBatches = 0;
+    while (
+      matches.length < limit &&
+      !exhausted &&
+      inboxBatches < MAX_SCAN_BATCHES
+    ) {
+      inboxBatches++;
       let batchWhere = where;
       const batchParams = [...params];
       if (cursorReceivedAt !== null && !isFirstBatch) {
@@ -1448,7 +1467,17 @@ export class ActivityPubObject extends DurableObject<ActivityPubEnv> {
       let outboxCursorSeq = tieSeq !== null ? Number(tieSeq) : null;
       let outboxExhausted = false;
       let isFirstOutboxBatch = true;
-      while (outboxMatches.length < limit && !outboxExhausted) {
+      // An owner outbox dominated by non-post activities (Like/Announce/etc.)
+      // would otherwise force a near-full-table scan per timeline request;
+      // cap the number of batches so a sparse outbox degrades to "found
+      // fewer than `limit` owner posts this page" instead of an unbounded scan.
+      let outboxBatches = 0;
+      while (
+        outboxMatches.length < limit &&
+        !outboxExhausted &&
+        outboxBatches < MAX_SCAN_BATCHES
+      ) {
+        outboxBatches++;
         let outboxWhere = initialOutbox.where;
         const outboxParams = [...initialOutbox.params];
         if (outboxCursorReceivedAt !== null && !isFirstOutboxBatch) {

packages/host-meta/tsconfig.json

@@ -7,7 +7,11 @@
     // rootDir-containment check. This config is `noEmit`, so rootDir is moot.
     "paths": {
       "@dwk/log": ["../log/src/index.ts"],
-      "@dwk/webfinger": ["../webfinger/src/index.ts"]
+      "@dwk/webfinger": ["../webfinger/src/index.ts"],
+      // `@dwk/webfinger`'s own source (path-mapped above) imports
+      // `@dwk/safe-fetch` for its shared timeout-signal helper, and that
+      // import resolves under THIS program's paths too.
+      "@dwk/safe-fetch": ["../safe-fetch/src/index.ts"]
     },
     "types": ["@cloudflare/workers-types"],
     "noEmit": true

packages/remotestorage/tsconfig.json

@@ -12,7 +12,11 @@
       // `@dwk/store`'s `Store` interface references `@dwk/rdf`'s `StoredQuad`
       // (the `readQuads`/`writeQuads` RDF tier we don't call); map it so
       // `typecheck` resolves the transitive type before the workspace is built.
-      "@dwk/rdf": ["../rdf/src/index.ts"]
+      "@dwk/rdf": ["../rdf/src/index.ts"],
+      // `@dwk/webfinger`'s own source (path-mapped above) imports
+      // `@dwk/safe-fetch` for its shared timeout-signal helper, and that
+      // import resolves under THIS program's paths too.
+      "@dwk/safe-fetch": ["../safe-fetch/src/index.ts"]
     },
     "types": [
       "@cloudflare/workers-types",

packages/safe-fetch/src/index.ts

@@ -15,6 +15,7 @@ export {
   isPrivateOrReservedHost,
   assertPublicUrl,
   safeFetch,
+  createTimeoutSignal,
   SsrfError,
   DEFAULT_MAX_REDIRECTS,
   DEFAULT_TIMEOUT_MS,

packages/safe-fetch/src/safe-fetch.ts

@@ -38,11 +38,14 @@ const DEFAULT_ALLOWED_SCHEMES = ["http:", "https:"] as const;
 const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
 
 /**
- * Create a request signal that is cancelled by either the caller or an
- * overall timeout. Unlike `AbortSignal.timeout()`, its timer can be released
- * as soon as the request completes.
+ * Create a request signal that is cancelled by either an optional caller
+ * signal or an overall timeout. Unlike `AbortSignal.timeout()`, its timer can
+ * be released as soon as the request completes, so it doesn't keep a Durable
+ * Object's event loop alive after a successful call. The shared primitive for
+ * every `@dwk` package that bounds an outbound request with a cancellable
+ * timeout, rather than each re-deriving its own copy.
  */
-function createTimeoutSignal(
+export function createTimeoutSignal(
   callerSignal: AbortSignal | null | undefined,
   timeoutMs: number,
 ): { readonly signal: AbortSignal; readonly cancel: () => void } {

packages/webfinger/package.json

@@ -41,6 +41,7 @@
     "access": "public"
   },
   "dependencies": {
-    "@dwk/log": "workspace:*"
+    "@dwk/log": "workspace:*",
+    "@dwk/safe-fetch": "workspace:*"
   }
 }

packages/webfinger/src/lookup.ts

@@ -14,6 +14,8 @@
  * @see spec/fediverse-interop.md §2.4 (community discovery)
  */
 
+import { createTimeoutSignal } from "@dwk/safe-fetch";
+
 /** A parsed `user@host` handle. `user` keeps its case; `host` is lowercased. */
 export interface ParsedHandle {
   readonly user: string;
@@ -113,19 +115,6 @@ export interface ResolveHandleOptions {
   readonly timeoutMs?: number;
 }
 
-/** Make a timeout signal whose timer is released when the lookup completes. */
-function createTimeoutSignal(timeoutMs: number): {
-  readonly signal: AbortSignal;
-  readonly cancel: () => void;
-} {
-  const controller = new AbortController();
-  const timeout = setTimeout(() => controller.abort(), timeoutMs);
-  return {
-    signal: controller.signal,
-    cancel: () => clearTimeout(timeout),
-  };
-}
-
 /**
  * Resolve a handle to its ActivityPub actor IRI: parse, query
  * `/.well-known/webfinger` on the handle's host, and select the `self` actor
@@ -139,7 +128,7 @@ export async function resolveHandle(
   const parsed = parseHandle(handle);
   if (!parsed) return null;
   let response: Response;
-  const timeout = createTimeoutSignal(options.timeoutMs ?? 10_000);
+  const timeout = createTimeoutSignal(undefined, options.timeoutMs ?? 10_000);
   try {
     response = await options.fetch(webfingerQueryUrl(parsed), {
       headers: { accept: "application/jrd+json, application/json" },

packages/webfinger/tsconfig.json

@@ -6,7 +6,8 @@
     // typecheck ahead of build), which would otherwise trip the
     // rootDir-containment check. This config is `noEmit`, so rootDir is moot.
     "paths": {
-      "@dwk/log": ["../log/src/index.ts"]
+      "@dwk/log": ["../log/src/index.ts"],
+      "@dwk/safe-fetch": ["../safe-fetch/src/index.ts"]
     },
     "types": ["@cloudflare/workers-types"],
     "noEmit": true