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; }