From 62ec69bc0d0026ad57df1b3c7a2e9eec6501bbc9 Mon Sep 17 00:00:00 2001 From: maashkinoozhe Date: Sun, 25 Oct 2026 08:58:48 +0100 Subject: [PATCH] Single-flight discovery harvester with budgets and stop reasons (BL-0051) Co-Authored-By: Claude Opus 5.5 --- discovery/harvest.js | 152 +++++++++++++++++++ test/discovery-harvest.test.js | 261 +++++++++++++++++++++++++++++++++ 2 files changed, 413 insertions(+) create mode 100644 discovery/harvest.js create mode 100644 test/discovery-harvest.test.js diff --git a/discovery/harvest.js b/discovery/harvest.js new file mode 100644 index 0000000..6213e56 --- /dev/null +++ b/discovery/harvest.js @@ -0,0 +1,152 @@ +'use strict'; + +const crypto = require('node:crypto'); +const { AniListStop, AniListSkip } = require('./anilist'); +const { topTagsFor } = require('./anchors'); + +const BUDGETS = { nightly: 300, admin: 300, user: 60 }; +const REFRESH_COOLDOWN_MS = 3600000; +const MAX_CONSECUTIVE_FAILURES = 5; +const CREATOR_MIN_WEIGHT = 3; + +// Thrown out of a run to unwind every stage at once. +class StopRun extends Error { + constructor(status, stopReason) { super(stopReason); this.status = status; this.stopReason = stopReason; } +} + +async function runHarvest({ kind, targets, cache, client, budget }) { + let requests = 0, failures = 0; + const seen = new Set(); + // One request: skipped when already done this run or still fresh in the cache, otherwise awaited in turn. + const fetchInto = async (cacheKind, key, call) => { + const id = `${cacheKind}\n${key}`; + if (seen.has(id)) return; + seen.add(id); + if (cache.isFresh(cacheKind, key)) return; + if (requests >= budget) throw new StopRun('success', 'budget'); + requests++; + let data; + try { data = await call(); } + catch (error) { + if (error instanceof AniListStop) throw new StopRun('stopped', error.reason); + if (!(error instanceof AniListSkip)) throw error; + if (++failures >= MAX_CONSECUTIVE_FAILURES) throw new StopRun('stopped', 'upstream-errors'); + return; + } + failures = 0; + // A null title is not a manga format: nothing worth keeping. + if (data !== null) cache.put(cacheKind, key, data); + }; + + const anchors = targets.flatMap(target => target.anchors); + const recordOf = id => cache.get('title', id)?.data || null; + try { + for (const { anilistId } of anchors) await fetchInto('title', anilistId, () => client.title(anilistId)); + for (const { anilistId } of anchors) await fetchInto('recs', anilistId, () => client.recs(anilistId)); + for (const anchor of anchors) { + if (anchor.weight < CREATOR_MIN_WEIGHT) continue; + for (const { staffId } of recordOf(anchor.anilistId)?.authors || []) + await fetchInto('creator', staffId, () => client.creator(staffId)); + } + for (const target of targets) { + for (const tag of topTagsFor(target.anchors, recordOf, { mature: target.mature })) + await fetchInto('tagTop', `${tag.kind}:${tag.name}`, () => client.tagTop(tag)); + } + // 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; + return { requests, budget, status: error.status, stopReason: error.stopReason }; + } + return { requests, budget, status: 'success', stopReason: null }; +} + +const localDate = date => [date.getFullYear(), String(date.getMonth() + 1).padStart(2, '0'), + String(date.getDate()).padStart(2, '0')].join('-'); + +function shouldRunNightly({ now, harvestAt, lastNightlyDate, enabled }) { + if (!enabled) return false; + const match = /^(\d{2}):(\d{2})$/.exec(harvestAt || ''); + if (!match) return false; + if (now.getHours() * 60 + now.getMinutes() < Number(match[1]) * 60 + Number(match[2])) return false; + return localDate(now) !== lastNightlyDate; +} + +const httpError = (status, message, extra = {}) => Object.assign(new Error(message), { status }, extra); + +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(); + } + nextAllowedAt(user) { + const last = this.lastAccepted.get(user); + if (last === undefined || this.now() - last >= REFRESH_COOLDOWN_MS) return null; + return new Date(last + REFRESH_COOLDOWN_MS).toISOString(); + } + jobs() { return [this.active, ...this.queue].filter(Boolean); } + request({ kind, user }) { + if (!(kind in BUDGETS)) 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'); + 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'); + } + const job = { id: crypto.randomBytes(8).toString('hex'), kind, ...(kind === 'user' ? { user } : {}) }; + const position = this.jobs().length; + this.queue.push(job); + if (!this.active) this.drain(); + return { harvestId: job.id, position }; + } + state(userName) { + const nextAllowedAt = this.nextAllowedAt(userName); + const cooldown = nextAllowedAt ? { nextAllowedAt } : {}; + 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 }; + } + current() { + if (!this.active) return null; + const { id, kind, user, startedAt } = this.active; + return { id, kind, ...(user ? { user } : {}), startedAt }; + } + // Runs synchronously up to the first await, so a job is active before request() returns. + async drain() { + while (this.queue.length) { + const job = this.queue.shift(); + job.startedAt = new Date(this.now()).toISOString(); + this.active = job; + try { await this.run(job); } + catch (error) { console.error(`Discovery harvest ${job.id} failed:`, error); } + finally { this.active = null; } + } + } + async run(job) { + const budget = BUDGETS[job.kind]; + const base = { id: job.id, kind: job.kind, user: job.user ?? null }; + 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); + 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' }; + } + 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); } + catch (error) { console.error(`Discovery harvest ${job.id} follow-up failed:`, error); } + } +} + +module.exports = { BUDGETS, REFRESH_COOLDOWN_MS, MAX_CONSECUTIVE_FAILURES, runHarvest, shouldRunNightly, Harvester }; diff --git a/test/discovery-harvest.test.js b/test/discovery-harvest.test.js new file mode 100644 index 0000000..afa6aa7 --- /dev/null +++ b/test/discovery-harvest.test.js @@ -0,0 +1,261 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { DiscoveryCache } = require('../discovery/cache'); +const { AniListStop, AniListSkip } = require('../discovery/anilist'); +const { + BUDGETS, REFRESH_COOLDOWN_MS, MAX_CONSECUTIVE_FAILURES, runHarvest, shouldRunNightly, Harvester, +} = require('../discovery/harvest'); + +const tempCache = (t) => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'discovery-harvest-')); + t.after(() => fs.rmSync(dir, { recursive: true, force: true })); + return new DiscoveryCache(dir); +}; + +const record = (id, { tags = [], authors = [] } = {}) => ({ anilistId: id, title: `T${id}`, tags, authors }); + +// Fake client: answers from fixture maps, logs every call, and throws scripted errors by call number. +function fakeClient({ titles = {}, errors = {} } = {}) { + const calls = []; + const answer = (name, key, value) => { + calls.push(`${name}:${key}`); + const error = errors[calls.length]; + if (error) throw error; + return value; + }; + return { + calls, + async title(id) { return answer('title', id, id in titles ? titles[id] : record(id)); }, + async recs(id) { return answer('recs', id, [{ anilistId: id * 100 }]); }, + async creator(staffId) { return answer('creator', staffId, [{ anilistId: staffId * 10 }]); }, + async tagTop({ name, kind }) { return answer('tagTop', `${kind}:${name}`, [{ anilistId: 1 }]); }, + async fresh() { return answer('fresh', 'global', [{ anilistId: 2 }]); }, + }; +} + +const anchor = (anilistId, weight = 1) => ({ anilistId, weight, source: 'owned', title: null }); + +test('constants', () => { + assert.deepEqual(BUDGETS, { nightly: 300, admin: 300, user: 60 }); + assert.equal(REFRESH_COOLDOWN_MS, 3600000); + assert.equal(MAX_CONSECUTIVE_FAILURES, 5); +}); + +test('plan order and dedupe', async (t) => { + const cache = tempCache(t); + const titles = { + 1: record(1, { tags: [{ name: 'Drama', kind: 'genre', rank: 100, isAdult: false }], authors: [{ staffId: 9, name: 'A', role: 'Story' }] }), + 2: record(2, { tags: [{ name: 'Drama', kind: 'genre', rank: 100, isAdult: false }] }), + 3: record(3, { tags: [{ name: 'Gore', kind: 'tag', rank: 80, isAdult: false }], authors: [{ staffId: 8, name: 'B', role: 'Art' }] }), + }; + const client = fakeClient({ titles }); + const targets = [ + { user: 'ann', anchors: [anchor(1, 5), anchor(2, 1)], mature: false }, + { user: 'bob', anchors: [anchor(1, 3), anchor(3, 1)], mature: false }, + ]; + const result = await runHarvest({ kind: 'admin', targets, cache, client, budget: 300 }); + assert.deepEqual(client.calls, [ + 'title:1', 'title:2', 'title:3', + 'recs:1', 'recs:2', 'recs:3', + 'creator:9', + 'tagTop:genre:Drama', 'tagTop:tag:Gore', + 'fresh:global', + ]); + assert.deepEqual(result, { requests: 10, budget: 300, status: 'success', stopReason: null }); + assert.equal(cache.get('title', 1).data.anilistId, 1); + assert.deepEqual(cache.get('recs', 2).data, [{ anilistId: 200 }]); + assert.deepEqual(cache.get('creator', 9).data, [{ anilistId: 90 }]); + assert.deepEqual(cache.get('tagTop', 'genre:Drama').data, [{ anilistId: 1 }]); + assert.deepEqual(cache.get('fresh', 'global').data, [{ anilistId: 2 }]); +}); + +test('non-manga titles are counted but not cached', async (t) => { + const cache = tempCache(t); + const client = fakeClient({ titles: { 4: null } }); + const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [anchor(4, 5)], mature: false }], cache, client, budget: 60 }); + assert.equal(cache.get('title', 4), null); + assert.deepEqual(client.calls, ['title:4', 'recs:4', 'fresh:global']); + assert.equal(result.requests, 3); +}); + +test('fresh cache entries are skipped', async (t) => { + const cache = tempCache(t); + cache.put('title', 1, record(1)); + cache.put('recs', 1, []); + cache.put('fresh', 'global', []); + const client = fakeClient(); + const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [anchor(1), anchor(2)], mature: false }], cache, client, budget: 60 }); + assert.deepEqual(client.calls, ['title:2', 'recs:2']); + assert.equal(result.requests, 2); + assert.equal(result.stopReason, null); +}); + +test('budget stops with reason budget', async (t) => { + const cache = tempCache(t); + const client = fakeClient(); + const targets = [{ user: 'ann', anchors: [anchor(1), anchor(2), anchor(3)], mature: false }]; + const result = await runHarvest({ kind: 'user', targets, cache, client, budget: 3 }); + assert.deepEqual(result, { requests: 3, budget: 3, status: 'success', stopReason: 'budget' }); + assert.equal(client.calls.length, 3); +}); + +test('429 stops at once', async (t) => { + const cache = tempCache(t); + const client = fakeClient({ errors: { 2: new AniListStop('rate-limited') } }); + const targets = [{ user: 'ann', anchors: [anchor(1), anchor(2), anchor(3)], mature: false }]; + const result = await runHarvest({ kind: 'admin', targets, cache, client, budget: 300 }); + assert.deepEqual(result, { requests: 2, budget: 300, status: 'stopped', stopReason: 'rate-limited' }); + assert.equal(client.calls.length, 2); + assert.equal(cache.get('title', 2), null); +}); + +test('network error stops with upstream-errors after 5', async (t) => { + const skip = () => new AniListSkip('timeout'); + const targets = [{ user: 'ann', anchors: [1, 2, 3, 4, 5, 6, 7, 8].map((id) => anchor(id)), mature: false }]; + + const client = fakeClient({ errors: { 1: skip(), 2: skip(), 3: skip(), 4: skip(), 5: skip() } }); + const result = await runHarvest({ kind: 'admin', targets, cache: tempCache(t), client, budget: 300 }); + assert.deepEqual(result, { requests: 5, budget: 300, status: 'stopped', stopReason: 'upstream-errors' }); + assert.equal(client.calls.length, 5); + + // A success in the middle resets the counter, so the run gets past five failures in total. + const mixed = fakeClient({ errors: { 1: skip(), 2: skip(), 3: skip(), 4: skip(), 6: skip(), 7: skip(), 8: skip() } }); + const reset = await runHarvest({ kind: 'admin', targets, cache: tempCache(t), client: mixed, budget: 300 }); + assert.equal(reset.status, 'success'); + assert.equal(reset.stopReason, null); + assert.ok(mixed.calls.length > 8); +}); + +test('plan for anchorless user only fetches fresh', async (t) => { + const client = fakeClient(); + const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [], mature: false }], cache: tempCache(t), client, budget: 60 }); + assert.deepEqual(client.calls, ['fresh:global']); + assert.deepEqual(result, { requests: 1, budget: 60, status: 'success', stopReason: null }); +}); + +test('shouldRunNightly', () => { + const at = (h, m) => new Date(2026, 9, 25, h, m); + assert.equal(shouldRunNightly({ now: at(3, 29), harvestAt: '03:30', lastNightlyDate: null, enabled: true }), false); + assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: null, enabled: true }), true); + assert.equal(shouldRunNightly({ now: at(23, 0), harvestAt: '03:30', lastNightlyDate: '2026-10-24', enabled: true }), true); + assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: '2026-10-25', enabled: true }), false); + assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: null, enabled: false }), false); +}); + +// A run that waits until the test releases it, so queue state can be observed. +function gatedHarvester({ onDone = async () => {}, now }) { + const gates = []; + const patches = []; + const done = []; + const client = { + calls: [], + async fresh() { + this.calls.push('fresh'); + await new Promise((resolve) => gates.push(resolve)); + return []; + }, + }; + // Nothing stays cached, so every job makes its one fresh call and waits at the gate. + const cache = { isFresh: () => false, get: () => null, put() {} }; + const harvester = new Harvester({ + cache, + client, + store: { patch(value) { patches.push(structuredClone(value)); } }, + 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 () => { + for (let i = 0; !gates.length; i++) { + if (i > 1000) throw new Error('no harvest request is waiting'); + await new Promise(setImmediate); + } + gates.shift()(); + await new Promise(setImmediate); + }; + return { harvester, patches, done, release, client }; +} + +const settle = async () => { for (let i = 0; i < 20; i++) await new Promise(setImmediate); }; + +test('harvester single-flight, positions, 409 and 429 cooldown', async (t) => { + let clock = Date.parse('2026-10-25T10:00:00Z'); + const h = gatedHarvester({ now: () => clock }); + const { harvester } = h; + + assert.deepEqual(harvester.state('ann'), { state: 'idle' }); + assert.equal(harvester.current(), null); + + const first = harvester.request({ kind: 'user', user: 'ann' }); + assert.equal(first.position, 0); + assert.equal(typeof first.harvestId, 'string'); + assert.equal(harvester.state('ann').state, 'running'); + assert.deepEqual(harvester.current(), { id: first.harvestId, kind: 'user', user: 'ann', startedAt: '2026-10-25T10:00:00.000Z' }); + + assert.throws(() => harvester.request({ kind: 'admin' }), (error) => error.status === 409); + assert.throws(() => harvester.request({ kind: 'nightly' }), (error) => error.status === 409); + assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }), (error) => error.status === 409); + + const second = harvester.request({ kind: 'user', user: 'bob' }); + assert.equal(second.position, 1); + assert.deepEqual(harvester.state('bob'), { state: 'queued', position: 1, nextAllowedAt: '2026-10-25T11:00:00.000Z' }); + assert.throws(() => harvester.request({ kind: 'user', user: 'bob' }), (error) => error.status === 409); + + await h.release(); + await settle(); + assert.equal(harvester.state('bob').state, 'running'); + assert.equal(harvester.state('ann').state, 'idle'); + + clock += 30 * 60000; + assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }), (error) => { + assert.equal(error.status, 429); + assert.equal(error.nextAllowedAt, '2026-10-25T11:00:00.000Z'); + return true; + }); + assert.deepEqual(harvester.state('ann'), { state: 'idle', nextAllowedAt: '2026-10-25T11:00:00.000Z' }); + + await h.release(); + await settle(); + assert.equal(harvester.current(), null); + + // The persisted records: running first, then the final one, for each job in order. + assert.deepEqual(h.patches.map((p) => [p.harvest.user, p.harvest.status]), + [['ann', 'running'], ['ann', 'success'], ['bob', 'running'], ['bob', 'success']]); + assert.deepEqual(h.patches[0].harvest, { id: first.harvestId, kind: 'user', user: 'ann', status: 'running', + startedAt: '2026-10-25T10:00:00.000Z', requests: 0, budget: 60 }); + assert.equal(h.patches[1].harvest.requests, 1); + assert.equal(h.patches[1].harvest.stopReason, null); + assert.equal(typeof h.patches[1].harvest.finishedAt, 'string'); + assert.deepEqual(h.done.map((d) => d.job.user), ['ann', 'bob']); + + clock += 30 * 60000; + const admin = harvester.request({ kind: 'admin' }); + assert.equal(admin.position, 0); + const ann = harvester.request({ kind: 'user', user: 'ann' }); + assert.equal(ann.position, 1); + await h.release(); + await h.release(); + await settle(); + assert.equal(h.patches.at(-4).harvest.budget, 300); +}); + +test('onDone failure does not stall the queue', async (t) => { + const errors = []; + const original = console.error; + console.error = (...args) => errors.push(args); + t.after(() => { console.error = original; }); + const h = gatedHarvester({ now: () => Date.parse('2026-10-25T10:00:00Z'), onDone: async () => { throw new Error('score failed'); } }); + h.harvester.request({ kind: 'nightly' }); + h.harvester.request({ kind: 'user', user: 'ann' }); + await h.release(); + await h.release(); + await settle(); + assert.deepEqual(h.done.map((d) => d.job.kind), ['nightly', 'user']); + assert.equal(h.harvester.current(), null); + assert.ok(errors.length >= 2); +});