From 552a59947513d6afaecdaeb6806f7d42b7b8a617 Mon Sep 17 00:00:00 2001 From: maashkinoozhe Date: Sun, 25 Oct 2026 09:02:03 +0100 Subject: [PATCH] Harvester pauses user refreshes after AniList stops (BL-0051) Co-Authored-By: Claude Opus 5.5 --- discovery/harvest.js | 38 +++++++++++--- test/discovery-harvest.test.js | 91 ++++++++++++++++++++++++++++++++-- 2 files changed, 116 insertions(+), 13 deletions(-) diff --git a/discovery/harvest.js b/discovery/harvest.js index 6213e56..e860106 100644 --- a/discovery/harvest.js +++ b/discovery/harvest.js @@ -55,7 +55,8 @@ async function runHarvest({ kind, targets, cache, client, budget }) { // With nobody to harvest for, even the global list is not worth a request. if (targets.length) await fetchInto('fresh', 'global', () => client.fresh()); } catch (error) { - if (!(error instanceof StopRun)) throw error; + // The caller records how many requests a local fault had already spent. + if (!(error instanceof StopRun)) throw Object.assign(error, { requests }); return { requests, budget, status: error.status, stopReason: error.stopReason }; } return { requests, budget, status: 'success', stopReason: null }; @@ -74,12 +75,20 @@ function shouldRunNightly({ now, harvestAt, lastNightlyDate, enabled }) { const httpError = (status, message, extra = {}) => Object.assign(new Error(message), { status }, extra); +// These reasons mean AniList pushed back: no further calls until a nightly or admin run. +const BLOCKING_REASONS = ['rate-limited', 'forbidden', 'challenge']; +const blockingReason = record => record?.status === 'stopped' && BLOCKING_REASONS.includes(record.stopReason) + ? record.stopReason : null; + class Harvester { constructor({ cache, client, store, targetsFor, onDone, now = Date.now }) { Object.assign(this, { cache, client, store, targetsFor, onDone, now }); this.queue = []; this.active = null; this.lastAccepted = new Map(); + // The pause outlives a restart through the persisted harvest record. + try { this.blockedReason = blockingReason(store.load?.()?.harvest); } + catch { this.blockedReason = null; } } nextAllowedAt(user) { const last = this.lastAccepted.get(user); @@ -88,16 +97,21 @@ class Harvester { } jobs() { return [this.active, ...this.queue].filter(Boolean); } request({ kind, user }) { - if (!(kind in BUDGETS)) throw httpError(400, `Unknown harvest kind: ${kind}`); + if (!Object.hasOwn(BUDGETS, kind)) throw httpError(400, `Unknown harvest kind: ${kind}`); if (kind === 'user') { if (!user) throw httpError(400, 'A user refresh needs a user'); if (this.jobs().some(job => job.kind === 'user' && job.user === user)) throw httpError(409, 'A refresh for this user is already queued or running'); + if (this.blockedReason) + throw httpError(503, `AniList paused the harvest: ${this.blockedReason}`, { blockedReason: this.blockedReason }); const nextAllowedAt = this.nextAllowedAt(user); if (nextAllowedAt) throw httpError(429, 'Refresh is cooling down', { nextAllowedAt }); this.lastAccepted.set(user, this.now()); } else if (this.jobs().length) { throw httpError(409, 'A harvest is already queued or running'); + } else { + // An accepted nightly or admin run is the one allowed retry after a pause. + this.blockedReason = null; } const job = { id: crypto.randomBytes(8).toString('hex'), kind, ...(kind === 'user' ? { user } : {}) }; const position = this.jobs().length; @@ -107,17 +121,23 @@ class Harvester { } state(userName) { const nextAllowedAt = this.nextAllowedAt(userName); - const cooldown = nextAllowedAt ? { nextAllowedAt } : {}; + const extra = { ...(nextAllowedAt ? { nextAllowedAt } : {}), ...(this.blockedReason ? { blockedReason: this.blockedReason } : {}) }; const index = this.jobs().findIndex(job => job.kind === 'user' && job.user === userName); - if (index === 0 && this.active) return { state: 'running', ...cooldown }; - if (index >= 0) return { state: 'queued', position: index, ...cooldown }; - return { state: 'idle', ...cooldown }; + if (index === 0 && this.active) return { state: 'running', ...extra }; + if (index >= 0) return { state: 'queued', position: index, ...extra }; + return { state: 'idle', ...extra }; } current() { if (!this.active) return null; const { id, kind, user, startedAt } = this.active; return { id, kind, ...(user ? { user } : {}), startedAt }; } + // Queued refreshes are dropped unrun; their users get their cooldown back for after the pause. + pause(reason) { + this.blockedReason = reason; + for (const job of this.queue) if (job.kind === 'user') this.lastAccepted.delete(job.user); + this.queue = this.queue.filter(job => job.kind !== 'user'); + } // Runs synchronously up to the first await, so a job is active before request() returns. async drain() { while (this.queue.length) { @@ -135,13 +155,15 @@ class Harvester { await this.store.patch({ harvest: { ...base, status: 'running', startedAt: job.startedAt, requests: 0, budget } }); let result; try { - const targets = this.targetsFor(job.kind, job.user); + const targets = await this.targetsFor(job.kind, job.user); result = await runHarvest({ kind: job.kind, targets, cache: this.cache, client: this.client, budget }); } catch (error) { // A local fault (cache write, bad target) must not leave the record "running" until restart. console.error(`Discovery harvest ${job.id} failed:`, error); - result = { requests: 0, budget, status: 'stopped', stopReason: 'error' }; + result = { requests: error?.requests ?? 0, budget, status: 'stopped', stopReason: 'error' }; } + const blocked = blockingReason(result); + if (blocked) this.pause(blocked); await this.store.patch({ harvest: { ...base, status: result.status, startedAt: job.startedAt, finishedAt: new Date(this.now()).toISOString(), requests: result.requests, budget, stopReason: result.stopReason } }); try { await this.onDone(job, result); } diff --git a/test/discovery-harvest.test.js b/test/discovery-harvest.test.js index afa6aa7..41dddd3 100644 --- a/test/discovery-harvest.test.js +++ b/test/discovery-harvest.test.js @@ -148,7 +148,7 @@ test('shouldRunNightly', () => { }); // A run that waits until the test releases it, so queue state can be observed. -function gatedHarvester({ onDone = async () => {}, now }) { +function gatedHarvester({ onDone = async () => {}, now, load }) { const gates = []; const patches = []; const done = []; @@ -156,7 +156,7 @@ function gatedHarvester({ onDone = async () => {}, now }) { calls: [], async fresh() { this.calls.push('fresh'); - await new Promise((resolve) => gates.push(resolve)); + await new Promise((resolve, reject) => gates.push({ resolve, reject })); return []; }, }; @@ -165,17 +165,19 @@ function gatedHarvester({ onDone = async () => {}, now }) { const harvester = new Harvester({ cache, client, - store: { patch(value) { patches.push(structuredClone(value)); } }, + store: { patch(value) { patches.push(structuredClone(value)); }, ...(load ? { load } : {}) }, targetsFor: (kind, user) => [{ user: user || 'all', anchors: [], mature: false }], onDone: async (job, result) => { done.push({ job, result }); await onDone(job, result); }, now, }); - const release = async () => { + // Lets the waiting request finish, or fail with `error`. + const release = async (error) => { for (let i = 0; !gates.length; i++) { if (i > 1000) throw new Error('no harvest request is waiting'); await new Promise(setImmediate); } - gates.shift()(); + const gate = gates.shift(); + if (error) gate.reject(error); else gate.resolve(); await new Promise(setImmediate); }; return { harvester, patches, done, release, client }; @@ -259,3 +261,82 @@ test('onDone failure does not stall the queue', async (t) => { assert.equal(h.harvester.current(), null); assert.ok(errors.length >= 2); }); + +test('AniList stop pauses user refreshes until an admin or nightly run', async () => { + let clock = Date.parse('2026-10-25T10:00:00Z'); + const h = gatedHarvester({ now: () => clock }); + const { harvester, client } = h; + harvester.request({ kind: 'user', user: 'ann' }); + harvester.request({ kind: 'user', user: 'bob' }); + assert.equal(harvester.state('bob').state, 'queued'); + + // (a) 429 while ann runs: bob's queued refresh is dropped without touching AniList. + await h.release(new AniListStop('rate-limited')); + await settle(); + assert.equal(client.calls.length, 1); + assert.equal(harvester.current(), null); + assert.deepEqual(harvester.state('bob'), { state: 'idle', blockedReason: 'rate-limited' }); + assert.deepEqual(h.patches.map((p) => [p.harvest.user, p.harvest.status, p.harvest.stopReason ?? null]), + [['ann', 'running', null], ['ann', 'stopped', 'rate-limited']]); + assert.deepEqual(h.done.map((d) => d.job.user), ['ann']); + + // (b) a new refresh is refused while paused, even for a user outside any cooldown. + clock += 2 * REFRESH_COOLDOWN_MS; + assert.throws(() => harvester.request({ kind: 'user', user: 'cid' }), (error) => { + assert.equal(error.status, 503); + assert.equal(error.blockedReason, 'rate-limited'); + assert.match(error.message, /AniList paused the harvest: rate-limited/); + return true; + }); + assert.equal(client.calls.length, 1); + + // (c) an accepted admin run is the allowed retry and lifts the pause. + harvester.request({ kind: 'admin' }); + assert.equal(harvester.state('cid').blockedReason, undefined); + await h.release(); + await settle(); + assert.equal(client.calls.length, 2); + assert.equal(harvester.request({ kind: 'user', user: 'cid' }).position, 0); + await h.release(); + await settle(); +}); + +test('pause survives a restart through the stored harvest record', () => { + const load = () => ({ harvest: { status: 'stopped', stopReason: 'forbidden' } }); + const { harvester } = gatedHarvester({ now: () => 0, load }); + assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }), + (error) => error.status === 503 && error.blockedReason === 'forbidden'); + assert.equal(harvester.state('ann').blockedReason, 'forbidden'); + + const budgetStop = gatedHarvester({ now: () => 0, load: () => ({ harvest: { status: 'stopped', stopReason: 'budget' } }) }); + assert.deepEqual(budgetStop.harvester.state('ann'), { state: 'idle' }); + const broken = gatedHarvester({ now: () => 0, load: () => { throw new Error('corrupt'); } }); + assert.deepEqual(broken.harvester.state('ann'), { state: 'idle' }); +}); + +test('unknown kinds are refused', () => { + const { harvester } = gatedHarvester({ now: () => 0 }); + for (const kind of ['constructor', 'toString', 'other']) + assert.throws(() => harvester.request({ kind }), (error) => error.status === 400); +}); + +test('local fault keeps the real request count and async targets work', async (t) => { + const original = console.error; + console.error = () => {}; + t.after(() => { console.error = original; }); + let puts = 0; + const cache = { isFresh: () => false, get: () => null, put() { if (++puts === 2) throw new Error('disk full'); } }; + const patches = []; + const done = []; + const harvester = new Harvester({ + cache, client: fakeClient(), store: { patch(value) { patches.push(value); } }, + targetsFor: async () => [{ user: 'ann', anchors: [anchor(1)], mature: false }], + onDone: async (job, result) => { done.push(result); }, + }); + harvester.request({ kind: 'admin' }); + await settle(); + assert.deepEqual(patches.at(-1).harvest.status, 'stopped'); + assert.equal(patches.at(-1).harvest.stopReason, 'error'); + assert.equal(patches.at(-1).harvest.requests, 2); + assert.equal(done.length, 1); +}); \ No newline at end of file