Use current time for reply backoff

Base remote reply scrape completion, failure, and 429 backoff
timestamps on the worker clock at the point where processing finishes
instead of the job start time.  This keeps retry windows from becoming
immediately claimable after long-running scrapes.

#447 (comment)

Assisted-by: Codex:gpt-5.5

Author: dahlia

src/federation/replies-worker.test.ts

@@ -364,6 +364,52 @@ describe("remote replies scrape worker", () => {
     expect(origin?.nextRequestAt.getTime()).toBe(now.getTime() + 120_000);
   });
 
+  it("bases 429 backoff on the actual failure time", async () => {
+    expect.assertions(3);
+    const { jobId, postIri, repliesIri } = await seedPostWithScrapeJob();
+    const now = new Date("2026-04-25T00:00:00.000Z");
+    const failureTime = new Date("2026-04-25T00:10:00.000Z");
+    const error = new Error("rate limited") as Error & {
+      response: Response;
+    };
+    error.response = new Response(null, {
+      status: 429,
+      headers: { "Retry-After": "120" },
+    });
+
+    await processDueRemoteReplyScrapeJobs({
+      clock: () => failureTime,
+      documentLoader: async (url): Promise<RemoteDocument> => {
+        if (url === repliesIri) {
+          return {
+            contextUrl: null,
+            document: collection(repliesIri, [
+              reply({
+                id: "https://remote.test/@replyer/posts/1",
+                replyTarget: postIri,
+              }),
+            ]),
+            documentUrl: url,
+          };
+        }
+        if (url === "https://remote.test/@replyer") throw error;
+        throw new Error(`Unexpected fetch: ${url}`);
+      },
+      now,
+      sleep: async () => undefined,
+    });
+
+    const job = await db.query.remoteReplyScrapeJobs.findFirst({
+      where: eq(remoteReplyScrapeJobs.id, jobId),
+    });
+    const origin = await db.query.remoteReplyScrapeOrigins.findFirst();
+    expect(job?.status).toBe("pending");
+    expect(job?.nextAttemptAt.getTime()).toBe(failureTime.getTime() + 120_000);
+    expect(origin?.nextRequestAt.getTime()).toBe(
+      failureTime.getTime() + 120_000,
+    );
+  });
+
   it("records per-request timestamps for throttled origin request fields", async () => {
     expect.assertions(3);
     const { postIri, repliesIri } = await seedPostWithScrapeJob();
@@ -397,6 +443,6 @@ describe("remote replies scrape worker", () => {
     expect(origin?.nextRequestAt.toISOString()).toBe(
       "2026-04-25T00:00:12.000Z",
     );
-    expect(origin?.updated.toISOString()).toBe("2026-04-25T00:00:02.000Z");
+    expect(origin?.updated.toISOString()).toBe("2026-04-25T00:00:03.000Z");
   });
 });

src/federation/replies-worker.ts

@@ -196,13 +196,13 @@ async function processRemoteReplyScrapeJob(
   job: RemoteReplyScrapeJob,
   options: ProcessRemoteReplyScrapeJobsOptions,
 ): Promise<number> {
-  const now = options.now ?? new Date();
+  const clock = options.clock ?? (() => options.now ?? new Date());
   const post = await db.query.posts.findFirst({
     where: eq(posts.id, job.postId),
   });
 
   if (post == null) {
-    await failJob(job, "Post not found", now);
+    await failJob(job, "Post not found", clock());
     return 0;
   }
 
@@ -213,7 +213,7 @@ async function processRemoteReplyScrapeJob(
       documentLoader,
       intervalSeconds:
         options.intervalSeconds ?? REMOTE_REPLIES_SCRAPE_INTERVAL_SECONDS,
-      clock: options.clock ?? (() => new Date()),
+      clock,
       sleep: options.sleep ?? sleep,
     });
     let lastFetchError: unknown;
@@ -275,24 +275,25 @@ async function processRemoteReplyScrapeJob(
 
     await updatePostStats(db, { id: job.postId });
     await updateScrapedRepliesCount(job.postId);
-    await completeJob(job, fetchedItems, now);
+    await completeJob(job, fetchedItems, clock());
     return fetchedItems;
   } catch (error) {
     if (getErrorStatus(error) === 429) {
+      const failedAt = clock();
       await backOffJob(
         job,
-        retryAfterSeconds(error) ??
+        retryAfterSeconds(error, failedAt) ??
           options.backoffSeconds ??
           REMOTE_REPLIES_SCRAPE_BACKOFF_SECONDS,
         error,
-        now,
+        failedAt,
       );
       return 0;
     }
     await failJob(
       job,
       error instanceof Error ? error.message : String(error),
-      now,
+      clock(),
     );
     return 0;
   }
@@ -458,7 +459,7 @@ function getErrorStatus(error: unknown): number | null {
   return error.response.status;
 }
 
-function retryAfterSeconds(error: unknown): number | null {
+function retryAfterSeconds(error: unknown, now = new Date()): number | null {
   if (
     error == null ||
     typeof error !== "object" ||
@@ -476,7 +477,7 @@ function retryAfterSeconds(error: unknown): number | null {
 
   const date = Date.parse(retryAfter);
   if (Number.isNaN(date)) return null;
-  return Math.max(0, Math.ceil((date - Date.now()) / 1000));
+  return Math.max(0, Math.ceil((date - now.getTime()) / 1000));
 }
 
 function sleep(milliseconds: number): Promise<void> {