Fold a mid-run refresh into the run instead of re-enqueueing it

src/features/auth/routes/register.ts

@@ -30,7 +30,7 @@
 async function validateCaptcha(
   token: string | undefined,
   secret: string,
   fetcher: typeof globalThis.fetch,
 ) {
   if (!token) return false;
   try {
@@ -148,15 +148,20 @@
                 const activationTokenExpiresAt = new Date(
                   Date.now() + activationLifetimeMs,
                 );
-                await mailSender.sendActivationEmail(
-                  existing.email,
-                  activationToken,
-                );
+                // Stored before it is sent, unlike a first registration
+                // where a failed write leaves nothing to be locked out of:
+                // here the mail would land on a token the row never took,
+                // putting the address back in the dead end it just asked to
+                // leave -- one throttle slot poorer.
                 await usersDataService.refreshActivationToken(
                   existing.id,
                   activationToken,
                   activationTokenExpiresAt,
                 );
+                await mailSender.sendActivationEmail(
+                  existing.email,
+                  activationToken,
+                );
               }
             } else {
               const resetToken = randomUUID();

src/features/feeds/source-enqueue.ts

@@ -92,9 +92,10 @@ export class SourceEnqueuer {
     if (holder && state === "active") {
       // The worker is mid-run holding this id, so BullMQ dedupes the add
       // away -- and the caller (a hub, a user's refresh button) has already
-      // been answered. Record the request instead: when the run finishes,
-      // the worker takes everything that accumulated and enqueues one
-      // follow-up refresh for the source (#813).
+      // been answered. Record the request instead: the run takes everything
+      // that accumulated when it ends and parses the source once more itself
+      // (#813). It cannot come back through here to do that -- the id is
+      // still active until its processor returns.
       await this.mergePendingRefresh(source.id, skipCache);
       return;
     }

src/features/jobs/__tests__/main.test.ts

@@ -386,19 +386,27 @@
 });
 
 // A request that arrived while the job held the source's id was recorded
-// rather than queued (the add would have been deduped away); the end of the
-// run is where it is collected and answered with one follow-up refresh.
-test("folds requests that arrived mid-run into one follow-up refresh", async () => {
+// rather than queued (the add would have been deduped away); the run folds it
+// in itself at the end. It cannot hand it back to the enqueuer -- the id is
+// still active until this processor returns, so that add would be deduped
+// against this very job and the refresh would wait for the next poll.
+test("folds requests that arrived mid-run into one more parse", async () => {
   let processor: ((job: MainWorkerJob) => Promise<void>) | undefined;
   const createWorker: MainWorkerFactory = (value, options) => {
     processor = value;
     return noopWorkerFactory(value, options);
   };
+  const parsed: Array<[boolean | undefined, string | undefined]> = [];
   const followedUp: Parameters<SourceEnqueuer["enqueueSource"]>[] = [];
+  let pending: { skipCache: boolean } | null = { skipCache: true };
   const worker = await createMainWorker(
     config,
     { async add() {}, async addBulk() {} },
-    idleParser,
+    {
+      async parseSource(input) {
+        parsed.push([input.skipCache, input.trigger]);
+      },
+    },
     idleFaviconRefresher,
     idleSources,
     idleSources,
@@ -407,11 +415,15 @@
     createWorker,
     idleHubPoster,
     {
       async enqueueSource(source, trigger, skipCache) {
         followedUp.push([source, trigger, skipCache]);
       },
+      // getdel: the take is the consume, so a second call sees nothing.
       async takePendingRefresh(sourceId) {
-        return sourceId === source.id ? { skipCache: true } : null;
+        if (sourceId !== source.id) return null;
+        const taken = pending;
+        pending = null;
+        return taken;
       },
     },
   );
@@ -424,9 +436,11 @@
     name: JobName.ParseSource,
   });
 
-  expect(followedUp).toEqual([
-    [{ id: source.id, url: source.url }, "manual", true],
+  expect(parsed).toEqual([
+    [undefined, "poll"],
+    [true, "manual"],
   ]);
+  expect(followedUp).toEqual([]);
 });
 
 // A deferral keeps the job (and its id) alive for another run, so the

src/features/jobs/main.ts

@@ -82,25 +82,6 @@ export class MainWorker {
     await this.worker?.close();
   }
 
-  // Everything that arrived while this job held the source's id -- pushes
-  // the hub was already answered for, refresh clicks still spinning -- folds
-  // into exactly one follow-up refresh, so the content those requests were
-  // about never waits for the next poll (#813). The job's id is free by the
-  // time this runs: completion or a non-deferred failure has erased it, so
-  // the follow-up add cannot be deduped away.
-  private readonly runPendingRefresh = async (source: {
-    id: number;
-    url: string;
-  }) => {
-    const pending = await sourceEnqueuer.takePendingRefresh(source.id);
-    if (!pending) return;
-    await sourceEnqueuer.enqueueSource(
-      { id: source.id, url: source.url },
-      "manual",
-      pending.skipCache,
-    );
-  };
-
   private async gatherParseSourceJobs() {
     const sources = await sourcesDataService.getSourcesToProcess();
 
@@ -169,25 +150,26 @@ export class MainWorker {
           if (!source) {
             throw new Error(`Source with ID ${input.data.id} not found`);
           }
-          try {
+          const parse = async (
+            skipCache: boolean | undefined,
+            trigger: "manual" | "poll" | "websub-push",
+          ) =>
             await feedParser.parseSource({
               ...source,
-              ...(input.data.skipCache === undefined
-                ? {}
-                : { skipCache: input.data.skipCache }),
-              trigger: input.data.trigger ?? "poll",
+              ...(skipCache === undefined ? {} : { skipCache }),
+              trigger,
             });
-          } catch (cause) {
-            // A deferral keeps this job's id -- it runs again later, and a
-            // request that arrives meanwhile reaches it through the queued
-            // payload. Anything else ends the job for good, so it owes the
-            // same follow-up a success owes.
-            if (!isHttpDeferredError(cause)) {
-              await this.runPendingRefresh(source);
-            }
-            throw cause;
-          }
-          await this.runPendingRefresh(source);
+          await parse(input.data.skipCache, input.data.trigger ?? "poll");
+          // Requests that landed mid-run were recorded on a marker rather
+          // than queued, because BullMQ would have deduped their add against
+          // this job (#813). Handing them back to the enqueuer would hit that
+          // same dedupe -- the id stays active until this processor returns --
+          // so the run folds them in itself and parses once more, right here,
+          // the way the newest request asked. Anything arriving during that
+          // pass stays on the marker: a deferral, the next poll or the next
+          // request all reach it.
+          const pending = await sourceEnqueuer.takePendingRefresh(source.id);
+          if (pending) await parse(pending.skipCache, "manual");
           break;
         }