From d1a82e4d89af8acb0e789feeca6e938c1f594b0f Mon Sep 17 00:00:00 2001 From: zakius <2071503+zakius@users.noreply.github.com> Date: Sat, 12 Sep 2026 22:03:46 +0200 Subject: [PATCH] Fold a mid-run refresh into the run instead of re-enqueueing it The follow-up refresh #813 promises never ran. runPendingRefresh called enqueueSource from inside the processor, where BullMQ still holds ParseSource- as active -- so enqueueSource took its own active branch and wrote the pending marker straight back instead of queueing anything. The marker ping-ponged and the content still waited for the next poll, which for a verified websub source is up to a day: the exact symptom the issue is about. The unit test could not see it, stubbing enqueueSource out entirely. The id cannot be free until the processor returns, so the run answers the requests itself: take the marker, parse once more with the flags they asked for. A deferral or a failure leaves the marker for the next run, as before. This also drops a dead branch -- parseSource only rethrows HttpDeferredError, so the non-deferred catch never ran. Also: the re-registration path stored its fresh activation token after mailing it, so a failed write mailed a token the row never took and put the address back in the dead end #810 fixes. Co-Authored-By: Claude Opus 5 (1M context) --- src/features/auth/routes/register.ts | 13 ++++-- src/features/feeds/source-enqueue.ts | 7 ++-- src/features/jobs/__tests__/main.test.ts | 28 +++++++++---- src/features/jobs/main.ts | 52 ++++++++---------------- 4 files changed, 51 insertions(+), 49 deletions(-) diff --git a/src/features/auth/routes/register.ts b/src/features/auth/routes/register.ts index 08577d3a..00ddc593 100644 --- a/src/features/auth/routes/register.ts +++ b/src/features/auth/routes/register.ts @@ -148,15 +148,20 @@ export function createRegisterRoute() { 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(); diff --git a/src/features/feeds/source-enqueue.ts b/src/features/feeds/source-enqueue.ts index abd7da69..e016198c 100644 --- a/src/features/feeds/source-enqueue.ts +++ b/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; } diff --git a/src/features/jobs/__tests__/main.test.ts b/src/features/jobs/__tests__/main.test.ts index d9829ac6..8f96fee9 100644 --- a/src/features/jobs/__tests__/main.test.ts +++ b/src/features/jobs/__tests__/main.test.ts @@ -386,19 +386,27 @@ test("moves deferred validated jobs with their BullMQ token", async () => { }); // 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) | undefined; const createWorker: MainWorkerFactory = (value, options) => { processor = value; return noopWorkerFactory(value, options); }; + const parsed: Array<[boolean | undefined, string | undefined]> = []; const followedUp: Parameters[] = []; + 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, @@ -410,8 +418,12 @@ test("folds requests that arrived mid-run into one follow-up refresh", async () 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 @@ test("folds requests that arrived mid-run into one follow-up refresh", async () 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 diff --git a/src/features/jobs/main.ts b/src/features/jobs/main.ts index 2612df48..e8ebdf91 100644 --- a/src/features/jobs/main.ts +++ b/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; }