fix

src/federation.ts

@@ -6,18 +6,28 @@ import {
 } from "./crypto";
 import {
   actorCacheStale,
+  claimOutgoingDelivery,
   deleteActorFromCache,
+  enqueueOutgoingDeliveries,
   ensureActorLocalId,
   getActorByKeyId,
   getActorFromCache,
+  getUserById,
+  listDueOutgoingDeliveries,
+  markOutgoingDeliveryDelivered,
+  markOutgoingDeliveryFailed,
   recordNotification,
   upsertActorCache
 } from "./db";
-import type { ActorCache, Json, RemoteActor, Status, User } from "./types";
+import type { ActorCache, Json, OutgoingDelivery, RemoteActor, Status, User } from "./types";
 import { SIGNATURE_MAX_SKEW_MS } from "./types";
 import { actorUrl, base64Decode, encoder, hostFromBaseUrl, parseAcctFromActor } from "./util";
 
 const ACTIVITY_HEADERS = "application/activity+json, application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"";
+const DELIVERY_BATCH_SIZE = 20;
+const DELIVERY_MAX_ATTEMPTS = 8;
+const DELIVERY_LEASE_MS = 60_000;
+const DELIVERY_MAX_BACKOFF_SECONDS = 60 * 60;
 
 export async function resolveRemoteActor(env: Env, actorId: string, opts: { force?: boolean } = {}): Promise<ActorCache | null> {
   if (!actorId) return null;
@@ -153,11 +163,57 @@ export async function sendSignedActivity(env: Env, user: User, inboxUrl: string,
 }
 
 export async function deliverToInboxes(env: Env, user: User, inboxes: Iterable<string>, activity: Json): Promise<void> {
-  const unique = new Set<string>();
-  for (const inbox of inboxes) {
-    if (inbox) unique.add(inbox);
+  await enqueueOutgoingDeliveries(env, user.id, inboxes, activity);
+}
+
+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 Promise.allSettled([...unique].map((inbox) => sendSignedActivity(env, user, inbox, activity)));
+}
+
+async function processOutgoingDelivery(env: Env, delivery: OutgoingDelivery): Promise<void> {
+  const now = new Date();
+  const nowIso = now.toISOString();
+  const lockedUntil = new Date(now.getTime() + DELIVERY_LEASE_MS).toISOString();
+  const claimed = await claimOutgoingDelivery(env, delivery.id, nowIso, lockedUntil);
+  if (!claimed) return;
+
+  let activity: Json;
+  try {
+    activity = JSON.parse(delivery.activity_json) as Json;
+  } catch {
+    await markOutgoingDeliveryFailed(env, delivery.id, DELIVERY_MAX_ATTEMPTS, null, "invalid_activity_json");
+    return;
+  }
+
+  const user = await getUserById(env, delivery.user_id);
+  if (!user) {
+    await markOutgoingDeliveryFailed(env, delivery.id, DELIVERY_MAX_ATTEMPTS, null, "delivery_user_missing");
+    return;
+  }
+
+  const result = await sendSignedActivity(env, user, delivery.inbox, activity).catch((error) => ({
+    ok: false,
+    status: 0,
+    text: String(error)
+  }));
+  if (result.ok) {
+    await markOutgoingDeliveryDelivered(env, delivery.id);
+    return;
+  }
+
+  const attempts = delivery.attempts + 1;
+  const nextAttemptAt = attempts >= DELIVERY_MAX_ATTEMPTS ? null : nextDeliveryAttemptAt(attempts);
+  const error = result.status ? `${result.status} ${result.text}` : result.text;
+  await markOutgoingDeliveryFailed(env, delivery.id, attempts, nextAttemptAt, error);
+}
+
+function nextDeliveryAttemptAt(attempts: number): string {
+  const delaySeconds = Math.min(DELIVERY_MAX_BACKOFF_SECONDS, 60 * (2 ** Math.max(0, attempts - 1)));
+  return new Date(Date.now() + delaySeconds * 1000).toISOString();
 }
 
 export async function gatherFollowerInboxes(env: Env, userId: string): Promise<string[]> {