fix

src/federation.ts

@@ -25,9 +25,11 @@ import { actorUrl, base64Decode, encoder, hostFromBaseUrl, parseAcctFromActor }
 
 const ACTIVITY_HEADERS = "application/activity+json, application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"";
 const DELIVERY_BATCH_SIZE = 20;
+const DELIVERY_CONCURRENCY = 4;
 const DELIVERY_MAX_ATTEMPTS = 8;
 const DELIVERY_LEASE_MS = 60_000;
 const DELIVERY_MAX_BACKOFF_SECONDS = 60 * 60;
+const REMOTE_FETCH_TIMEOUT_MS = 10_000;
 
 export async function resolveRemoteActor(env: Env, actorId: string, opts: { force?: boolean } = {}): Promise<ActorCache | null> {
   if (!actorId) return null;
@@ -42,6 +44,7 @@ export async function fetchRemoteActor(actorId: string): Promise<RemoteActor | n
   try {
     const response = await fetch(actorId, {
       headers: { accept: ACTIVITY_HEADERS },
+      signal: AbortSignal.timeout(REMOTE_FETCH_TIMEOUT_MS),
       cf: { cacheTtl: 60 }
     });
     if (!response.ok) return null;
@@ -151,7 +154,8 @@ export async function sendSignedActivity(env: Env, user: User, inboxUrl: string,
         signature: headerValue,
         "user-agent": `toot-worker (+https://${hostFromBaseUrl(env)})`
       },
-      body
+      body,
+      signal: AbortSignal.timeout(REMOTE_FETCH_TIMEOUT_MS)
     });
     const text = response.ok ? "" : await response.text().catch(() => "");
     if (!response.ok) console.warn("signed-delivery", inboxUrl, response.status, text.slice(0, 200));
@@ -169,9 +173,18 @@ export async function deliverToInboxes(env: Env, user: User, inboxes: Iterable<s
 export async function processOutgoingDeliveries(env: Env): Promise<void> {
   const now = new Date().toISOString();
   const deliveries = await listDueOutgoingDeliveries(env, now, DELIVERY_BATCH_SIZE);
-  for (const delivery of deliveries) {
-    await processOutgoingDelivery(env, delivery);
-  }
+  await runWithConcurrency(deliveries, DELIVERY_CONCURRENCY, (delivery) => processOutgoingDelivery(env, delivery));
+}
+
+async function runWithConcurrency<T>(items: T[], concurrency: number, worker: (item: T) => Promise<void>): Promise<void> {
+  let cursor = 0;
+  const runners = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
+    while (cursor < items.length) {
+      const item = items[cursor++];
+      await worker(item);
+    }
+  });
+  await Promise.all(runners);
 }
 
 async function processOutgoingDelivery(env: Env, delivery: OutgoingDelivery): Promise<void> {
@@ -218,17 +231,12 @@ function nextDeliveryAttemptAt(attempts: number): string {
 
 export async function gatherFollowerInboxes(env: Env, userId: string): Promise<string[]> {
   const rows = await env.DB.prepare(
-    "SELECT inbox FROM follows WHERE local_user_id = ? AND accepted = 1"
+    `SELECT COALESCE(ac.shared_inbox, f.inbox) AS inbox
+     FROM follows f
+     LEFT JOIN actor_cache ac ON ac.inbox = f.inbox
+     WHERE f.local_user_id = ? AND f.accepted = 1`
   ).bind(userId).all<{ inbox: string }>();
-  const inboxes = new Set<string>();
-  for (const row of rows.results) {
-    if (row.inbox) {
-      const actor = await getActorFromCache(env, row.inbox);
-      const shared = (await getActorByKeyId(env, row.inbox))?.shared_inbox;
-      inboxes.add(actor?.shared_inbox ?? shared ?? row.inbox);
-    }
-  }
-  return [...inboxes];
+  return [...new Set(rows.results.map((row) => row.inbox).filter(Boolean))];
 }
 
 export async function resolveDeliveryInboxes(env: Env, actorIds: Iterable<string>): Promise<string[]> {