Harvester pauses user refreshes after AniList stops (BL-0051)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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 && this.active) return { state: 'running', ...extra };
-    if (index >= 0) return { state: 'queued', position: index, ...cooldown };
+    if (index >= 0) return { state: 'queued', position: index, ...extra };
-    return { state: 'idle', ...cooldown };
+    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); }

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