From d2e0dca7a4594954633d69c3e537a57f03d498f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C4=BDubom=C3=ADr=20Jesze?= Date: Sat, 17 Oct 2026 15:36:50 +0200 Subject: [PATCH] feat(grabber): a site remembers a programme's details between runs A source that wants a request per programme for its details makes a nightly refetch expensive. Today is fetched again every run, and with it every detail call for programmes already asked about. homepi carries the same forty lines in three sites to avoid it, and webtv.sk went from 9,643 requests a cold day to 1,543. Every context a site is handed now has `remember(key, { until, miss? }, task)`, which answers with what an earlier run kept under the key, or with the task's answer, kept until `until`. A failure `miss` matches is kept as `null`, so "no details" is asked once; any other is thrown and not kept, so the next run asks again. A kept "none" is an answer only to an ask with a `miss` of its own, which is what lets the type leave `null` out when there is none. Two asks of one key while the first is on its way are one task, each reading what it came to with its own `miss`. A parse waiting on another's lends its slot meanwhile, as it does for `paced`; a request cannot lend its queue slot, so it asks for itself. Every caller is handed a copy. Values are kept in a state group of their own, `remembered`, beside the site's `state`, and what has passed its `until` is dropped at the end of every run, whether or not anything asked. `--refresh` reads nothing kept and still keeps. An `until` already past keeps a value for this run only, and says so once. A value must be plain, as a channel's data must, since it comes back from JSON. A site state read that fails is no longer kept as the group, so the next ask tries again. --- docs/api.md | 6 + docs/site-config.md | 67 ++++++- src/cli/try.ts | 5 + src/core/plain.ts | 71 +++++++ src/grabber/channels.ts | 60 +----- src/grabber/context.ts | 29 +++ src/grabber/remember.ts | 289 +++++++++++++++++++++++++++ src/grabber/site-run.ts | 15 +- src/grabber/state.ts | 13 +- src/grabber/types.ts | 59 ++++++ test/grabber.test.ts | 428 +++++++++++++++++++++++++++++++++++++++- test/readme-examples.ts | 41 ++++ 12 files changed, 1017 insertions(+), 66 deletions(-) create mode 100644 src/core/plain.ts create mode 100644 src/grabber/remember.ts diff --git a/docs/api.md b/docs/api.md index 6f1692c..e8e08ad 100644 --- a/docs/api.md +++ b/docs/api.md @@ -687,6 +687,12 @@ leaves it free to keep a group however it likes, an append-log replayed on read included, so long as `readState` answers with what the last `writeState` was given. +The package keeps four for a site: `channels`, a list it fetched; `validators`, +what each url last answered with; `state`, the site's own bag; and +`remembered`, what a context's +[`remember`](./site-config.md#a-programmes-details-asked-once) kept, each +value with when it is forgotten. `StateKey` names them. + `FsCacheDriver` puts each group in `//.json`, beside the channel directories of the same site; a prune only ever looks at the day files, so state outlives every listing a site has. `SqliteCacheDriver` keeps a row per diff --git a/docs/site-config.md b/docs/site-config.md index 33f1b3a..15db351 100644 --- a/docs/site-config.md +++ b/docs/site-config.md @@ -305,6 +305,7 @@ What a `parseDay` call is given: | `programme` | `(start, title, options?) => ProgrammeBuilder` | A [builder bound to this channel-day](#building-programmes). | | `http` | `KyInstance` | The site's client — the very instance `request` was handed. See [a parse that needs another request](#a-parse-that-needs-another-request). | | `paced` | `(task) => Promise` | Run a request through the site's queue, so its `concurrency`, `rateLimit` and backoff apply to it. | +| `remember` | `(key, { until, miss? }, task) => Promise` | What an earlier run kept under `key`, or `task`'s answer, kept until `until` — see [a programme's details, asked once](#a-programmes-details-asked-once). | | `state` | `Map` | What this site [remembers between runs](#remembering-something-between-runs). | | `signal` | `AbortSignal \| undefined` | Already applied to `http`; here for work that does not go through it. | | `log`, `warn` | `(message, fields?) => void` | Somewhere to [say something](#saying-something), under the site's name. | @@ -424,6 +425,68 @@ Nothing else in the package reads it, and it is stored [beside the listings](./api.md#what-a-site-remembers-between-runs) — `state.json` in the site's cache directory. +### A programme's details, asked once + +A source that wants a request per programme for its details makes a nightly +refetch expensive: today is fetched again every run, and with it every detail +call for programmes already asked about. `remember` asks once and keeps the +answer for the runs after: + +```ts +async parseDay({ payload, http, paced, remember, programme }) { + return Promise.all(payload.items.map(async (item) => { + const details = await remember( + `detail:${item.id}|${item.start}`, + { + // Asked about while its day can be fetched again, and no longer. + until: Date.parse(item.end) + 86_400_000, + // "This one has no details" is an answer, and worth keeping. + miss: (error) => error instanceof HTTPError && error.response.status === 404, + }, + async () => { + const full = await paced(({ signal }) => + http.get(`detail/${item.id}`, { signal }).json()); + + return { desc: full.description, cast: full.cast }; // only what the parse uses + }, + ); + + const p = programme(new Date(item.start), item.title); + return details ? p.desc(details.desc) : p; + })); +} +``` + +It hands back what an earlier run kept under the key, or the task's answer, +kept until `until` — a `Date` or epoch milliseconds, always given, since how +long something is worth keeping is the site's to say. Only an answer is kept: a +failure `miss` matches is kept as `null`, and any other is thrown and kept by +nobody, so the next run asks again. A kept "none" is an answer only to an ask +with a `miss` of its own; one without asks again, and its type has no `null` +in it. + +Two asks of one key while the first is still on its way are one task — a +broadcast past midnight is in two days' listings — and each reads what that +task came to with its own `miss`. A parse waiting on another's lets its slot go +meanwhile, as it does for `paced`. A request cannot let its queue slot go, so +one asks for itself instead. + +Each caller is handed a copy, so changing what it was given changes nothing +kept. Like `state`, a value must be plain; a `Date` or a `Map` is refused +rather than coming back as a string or `{}` on the next run. An `until` already +past keeps the value for this run only, and says so once — usually an `until` +in seconds where milliseconds were meant. + +It is kept in a group of its own beside `state`, written whole when anything in +it changes, and what has passed its `until` is dropped at the end of every +run. So keep only what the parse uses, and pick keys that change when what they +are about does — a programme moved to another start is worth asking about +again. `--refresh` reads nothing an earlier run kept, as it reads no cached +listings, and still keeps what it is answered. Every context has it: a +request, a parse, and a pass over a whole document. A mark that lasts — "this +channel has no details, ask again in a week" — is a value remembered for a +week. + ### A parse that needs another request Plenty of sources put the schedule in one response and the description of each @@ -772,8 +835,8 @@ channel-day, its `grabbedAt` and `programmeCount`, and absent where it holds nothing — since a stale channel-day is usually there and merely old enough to ask about again. -And every mode gets `timeZone`, `dayZone` and [`at`](#times-of-day), as a parse -does. Under `'days'` and `'both'` there is no one day to assume, so `at` takes +And every mode gets `timeZone`, `dayZone`, [`at`](#times-of-day) and +[`remember`](#a-programmes-details-asked-once), as a parse does. Under `'days'` and `'both'` there is no one day to assume, so `at` takes the day as well — from the first day's midnight to the end of the last is the stretch a range API wants, on the clock of the site's days, where `from` and `to` are only UTC midnights: diff --git a/src/cli/try.ts b/src/cli/try.ts index bf348de..6659e57 100644 --- a/src/cli/try.ts +++ b/src/cli/try.ts @@ -36,6 +36,7 @@ import { kyWith, resolveChannels, siteHttp } from '../grabber/channels.js'; import { parseContext, requestContext, streamContext } from '../grabber/context.js'; import { planRequests } from '../grabber/planner.js'; import { resolveSite } from '../grabber/site.js'; +import { Memory } from '../grabber/remember.js'; import { SiteStateHandle } from '../grabber/state.js'; import type { AnySiteConfig, GrabberChannel, ParsedProgramme } from '../grabber/types.js'; import { ProgrammeBuilder } from '../xmltv/builder.js'; @@ -405,6 +406,10 @@ export async function tryChannelDay( task({ ...(options.signal ? { signal: options.signal } : {}) }), timeZone: config.timeZone, dayZone: resolved.dayZone, + // Over a store that keeps nothing, as `state` is: remembered for as long + // as this command, so a parse asking twice sees the first answer, and by + // no run after it. + memory: new Memory(state, site.site, now), }; const began = Date.now(); diff --git a/src/core/plain.ts b/src/core/plain.ts new file mode 100644 index 0000000..af569a1 --- /dev/null +++ b/src/core/plain.ts @@ -0,0 +1,71 @@ +/** + * Whether a value comes back from JSON as it went in. + * + * What the cache keeps is JSON, so anything handed to it to keep — a channel's + * `data`, a remembered value — has to be a value JSON gives back unchanged, or a + * run that wrote it and the run that reads it would see different things, and a + * site would find out only on the second. + */ + +/** + * The first thing in a value that is not plain, named with where it is — or + * nothing, when all of it is. + * + * Plain is what JSON gives back as it was given: strings, finite numbers, + * booleans, `null`, arrays and plain objects. A property holding `undefined` is + * fine, since leaving it out is all JSON does with it and reading it gives the + * same. Anything else comes back as something else — a `Date` a string, a `Map` + * an empty object, an instance of a class without its methods, `NaN` a `null` — + * or not at all, as a function does. + */ +export function unplain( + value: unknown, + at: string, + ancestors = new Set(), +): string | undefined { + if (value === null || typeof value === 'string' || typeof value === 'boolean') { + return undefined; + } + + if (typeof value === 'number') { + return Number.isFinite(value) ? undefined : `${String(value)} at ${at}`; + } + + if (typeof value !== 'object') { + return `${typeof value === 'undefined' ? 'undefined' : `a ${typeof value}`} at ${at}`; + } + + // The objects above this one, not every one seen: the same object twice in + // a list is fine — JSON writes it twice — and only one inside itself is not. + if (ancestors.has(value)) { + return `a circular reference at ${at}`; + } + + const prototype: unknown = Object.getPrototypeOf(value); + + if (!Array.isArray(value) && prototype !== Object.prototype && prototype !== null) { + const name = (value as { constructor?: { name?: string } }).constructor?.name; + + return `${name ? `an instance of ${name}` : 'an object of a class'} at ${at}`; + } + + ancestors.add(value); + + try { + for (const [key, item] of Object.entries(value)) { + const where = Array.isArray(value) ? `${at}[${key}]` : `${at}.${key}`; + // `undefined` as a property is left out, which reads the same; in an + // array it becomes `null`, which does not. + const found = + item === undefined && !Array.isArray(value) ? undefined : unplain(item, where, ancestors); + + if (found !== undefined) { + return found; + } + } + + return undefined; + } finally { + ancestors.delete(value); + } +} diff --git a/src/grabber/channels.ts b/src/grabber/channels.ts index c4f39dd..54e64f6 100644 --- a/src/grabber/channels.ts +++ b/src/grabber/channels.ts @@ -8,6 +8,7 @@ import type { ChannelCoverage } from '../merge/types.js'; import type { CacheStore } from '../cache/types.js'; import { localTimeZone } from '../core/days.js'; import { errorMessage, unanswered } from '../core/error.js'; +import { unplain } from '../core/plain.js'; import type { Emit, Says } from '../core/events.js'; import type { SiteSession } from './session.js'; import type { @@ -186,65 +187,6 @@ export interface ResolveChannelsOptions { paced?: (fetch: () => Promise) => Promise; } -/** - * The first thing in a channel's `data` that is not a plain value, named with - * where it is — or nothing, when all of it is. - * - * Plain is what JSON gives back as it was given: strings, finite numbers, - * booleans, `null`, arrays and plain objects. A property holding `undefined` is - * fine, since leaving it out is all JSON does with it and reading it gives the - * same. Anything else comes back as something else — a `Date` a string, a `Map` - * an empty object, an instance of a class without its methods, `NaN` a `null` — - * or not at all, as a function does. - */ -function unplain(value: unknown, at: string, ancestors = new Set()): string | undefined { - if (value === null || typeof value === 'string' || typeof value === 'boolean') { - return undefined; - } - - if (typeof value === 'number') { - return Number.isFinite(value) ? undefined : `${String(value)} at ${at}`; - } - - if (typeof value !== 'object') { - return `${typeof value === 'undefined' ? 'undefined' : `a ${typeof value}`} at ${at}`; - } - - // The objects above this one, not every one seen: the same object twice in - // a list is fine — JSON writes it twice — and only one inside itself is not. - if (ancestors.has(value)) { - return `a circular reference at ${at}`; - } - - const prototype: unknown = Object.getPrototypeOf(value); - - if (!Array.isArray(value) && prototype !== Object.prototype && prototype !== null) { - const name = (value as { constructor?: { name?: string } }).constructor?.name; - - return `${name ? `an instance of ${name}` : 'an object of a class'} at ${at}`; - } - - ancestors.add(value); - - try { - for (const [key, item] of Object.entries(value)) { - const where = Array.isArray(value) ? `${at}[${key}]` : `${at}.${key}`; - // `undefined` as a property is left out, which reads the same; in an - // array it becomes `null`, which does not. - const found = - item === undefined && !Array.isArray(value) ? undefined : unplain(item, where, ancestors); - - if (found !== undefined) { - return found; - } - } - - return undefined; - } finally { - ancestors.delete(value); - } -} - /** * Refuse a channel list whose `data` holds anything but plain values. * diff --git a/src/grabber/context.ts b/src/grabber/context.ts index 27a45b7..13e44c3 100644 --- a/src/grabber/context.ts +++ b/src/grabber/context.ts @@ -15,6 +15,7 @@ import { dayToDate, wallClock } from '../core/days.js'; import type { Says } from '../core/events.js'; import { ProgrammeBuilder } from '../xmltv/builder.js'; +import type { Away, Memory } from './remember.js'; import type { Request } from './planner.js'; import type { ResolvedBatching } from './planner.js'; import type { @@ -24,6 +25,8 @@ import type { GrabberChannel, PacedRequest, ParseContext, + Remember, + RememberOptions, RequestContextFor, SiteState, StreamContext, @@ -42,6 +45,16 @@ export interface ContextDeps { timeZone: string; /** The zone the site's days are on: the run's, or the one its `dayZone` names. */ dayZone: string; + /** What the site remembers between runs, for a context's `remember`. */ + memory: Memory; + /** + * How this context waits on something another one started — an ask of + * `remember`'s — without holding a slot meanwhile: a parse lends its + * `localWork` slot, as `paced` does, and a stream holds none. Absent where + * the context holds one it cannot lend, a request its site's queue slot, so + * it asks for itself rather than wait. + */ + away?: Away; } /** What every request context carries, whichever shape the rest of it takes. */ @@ -59,6 +72,7 @@ export function baseContext(request: Request, deps: ContextDeps): BaseRequestCon ...(deps.signal ? { signal: deps.signal } : {}), timeZone: deps.timeZone, dayZone: deps.dayZone, + remember: remembering(deps), }; } @@ -76,6 +90,20 @@ function clock(deps: ContextDeps, assumed?: string): (time: string, day?: string }; } +/** + * A context's `remember`: the site's memory, and how this context waits on an + * ask another one started. + * + * The same on every context, whatever days it covers, since how long a value + * is worth keeping is the site's to say — a programme's details for a while + * after it ends, a mark that a channel has none for a week. The options are + * the memory's to check, which knows whose site it is and says so. + */ +function remembering(deps: ContextDeps): Remember { + return ((key: string, options: RememberOptions, task: () => unknown) => + deps.memory.remember(key, options, task, deps.away)) as Remember; +} + /** * The days a request covers, as a context that batches them says them. * @@ -179,6 +207,7 @@ export function parseContext( timeZone: deps.timeZone, dayZone: deps.dayZone, at: clock(deps, day), + remember: remembering(deps), programme: (start, title, options) => new ProgrammeBuilder({ channel: channel.xmltvId, diff --git a/src/grabber/remember.ts b/src/grabber/remember.ts new file mode 100644 index 0000000..f6ea35c --- /dev/null +++ b/src/grabber/remember.ts @@ -0,0 +1,289 @@ +/** + * What a site remembers between runs about one thing at a time — a + * programme's details, say — and for how long. + * + * Kept in a state group of its own rather than in the site's bag. A group is + * one blob, rewritten whole when anything in it changes: a few thousand + * programmes' details beside a token would rewrite the token's file every run + * for the details' sake. Each value carries when it is forgotten, and what has + * passed that is dropped at the end of every run, so the group holds about a + * window's worth however long a site has been running. + */ + +import type { Says } from '../core/events.js'; +import { unplain } from '../core/plain.js'; +import { StateKey, type SiteStateHandle, type TrackedMap } from './state.js'; + +/** + * One remembered value, and the instant it is forgotten. + * + * `missed` marks a `null` that a failure `miss` matched stands for — "this has + * no details" — which is an answer only to an ask that says such failures are + * answers. One that does not is asked again. + */ +interface Kept { + until: number; + value: unknown; + missed?: true; +} + +/** + * What one ask came to: an answer, or the failure it ended in. Shared by every + * ask of the key while it was on its way, each of which reads it with its own + * `miss`. + */ +type Outcome = { value: unknown } | { error: unknown; refused?: true }; + +/** + * How a caller waits on an ask somebody else started without holding a slot + * meanwhile — a parse's `localWork` slot, which the ask it is waiting on may + * need back before it can finish. + */ +export type Away = (wait: () => Promise) => Promise; + +/** One site's remembered values for the length of a run. */ +export class Memory { + readonly #handle: SiteStateHandle; + readonly #site: string; + readonly #now: number; + readonly #refresh: boolean; + readonly #warn: Says['warn'] | undefined; + #kept: Promise> | undefined; + /** Asks under way, so two of one key at once are one task. */ + readonly #asking = new Map>(); + /** Answered by this run — what `--refresh` still lets an ask read. */ + readonly #fresh = new Set(); + /** + * Answers given an `until` this run has already passed: good for the run, + * and kept by nobody after it. + */ + readonly #passing = new Map(); + #warnedPast = false; + + /** + * This site's memory, in the state its run holds, as of the run's `now`. + * + * Made up front and read the first time it is asked, so a site that never + * remembers anything reads nothing more than it did until the run ends — see + * {@link prune}. Under `refresh` nothing kept by an earlier run is read: as + * everywhere else `--refresh` reaches, the cache's reading is turned off and + * its writing is not. + */ + constructor( + handle: SiteStateHandle, + site: string, + now: Date, + options: { refresh?: boolean; warn?: Says['warn'] } = {}, + ) { + this.#handle = handle; + this.#site = site; + this.#now = now.getTime(); + this.#refresh = options.refresh === true; + this.#warn = options.warn; + } + + /** + * What the site kept under `key`, or the task's answer, kept until the + * options' `until` — a context's `remember`. + * + * `away` is how this caller waits on an ask another one started; a caller + * without one holds something it cannot let go of meanwhile — a request, its + * site's queue slot — and runs a task of its own instead of waiting. + */ + async remember( + key: string, + options: unknown, + task: () => unknown, + away?: Away, + ): Promise { + const { until, miss } = this.#read(key, options); + const kept = await this.#open(); + const held = + this.#passing.get(key) ?? + (this.#refresh && !this.#fresh.has(key) ? undefined : kept.get(key)); + + // A "none" is an answer only to an ask that takes such failures for one. + if (held !== undefined && (held.missed !== true || miss !== undefined)) { + return structuredClone(held.value); + } + + const asking = this.#asking.get(key); + let outcome: Outcome; + + if (asking !== undefined && away !== undefined) { + outcome = await away(() => asking); + } else { + const own = this.#ask(kept, key, until, miss, task); + + if (asking === undefined) { + this.#asking.set(key, own); + void own.then(() => { + if (this.#asking.get(key) === own) { + this.#asking.delete(key); + } + }); + } + + outcome = await own; + } + + if ('value' in outcome) { + return structuredClone(outcome.value); + } + + // Each caller's own `miss` on what the ask came to, so the answer is the + // one its types promise: a caller without one gets the failure, whoever + // asked first and however they took it. + if (outcome.refused !== true && miss?.(outcome.error) === true) { + return null; + } + + throw outcome.error; + } + + /** + * Drop what has passed its `until` — at the end of a run, whether or not + * anything asked, so a site that stopped remembering stops keeping too. + * + * A group that cannot be read has nothing to drop: what failed is a read, + * and the run it ends is not the place to say so. + */ + async prune(): Promise { + await this.#open().catch(() => undefined); + } + + /** The options, checked: an `until` that is a time, and a `miss` that is a function. */ + #read( + key: string, + options: unknown, + ): { until: number; miss: ((error: unknown) => boolean) | undefined } { + const { until, miss } = (typeof options === 'object' && options !== null ? options : {}) as { + until?: unknown; + miss?: unknown; + }; + const at = + typeof until === 'number' ? until : until instanceof Date ? until.getTime() : Number.NaN; + + if (!Number.isFinite(at)) { + throw new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)}) needs an until, a Date or epoch ` + + `milliseconds (got ${until === undefined ? 'none' : JSON.stringify(until)})`, + ); + } + + if (miss !== undefined && typeof miss !== 'function') { + throw new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)})'s miss must be a function`, + ); + } + + return { until: at, miss: miss as ((error: unknown) => boolean) | undefined }; + } + + /** + * The group, without what was to be forgotten by the run's `now`. + * + * Dropped as it is read, which does two things: the group is saved without + * them, and within a run everything held is good. A read that fails is not + * remembered, so the next ask tries again. + */ + #open(): Promise> { + this.#kept ??= this.#handle.bag(StateKey.REMEMBERED).then( + (kept) => { + for (const [key, entry] of kept) { + // A pair that is not one is forgotten with the rest: whatever wrote + // it, it is no answer to anything. + if (typeof entry?.until !== 'number' || !(entry.until > this.#now)) { + kept.delete(key); + } + } + + return kept; + }, + (error: unknown) => { + this.#kept = undefined; + throw error; + }, + ); + + return this.#kept; + } + + /** + * Ask, and keep the answer by the asker's own options — a failure its + * `miss` matches as `null`, marked as standing for one. What it hands back + * is the outcome itself, the failure included, for every caller to read with + * its own `miss`. + */ + async #ask( + kept: TrackedMap, + key: string, + until: number, + miss: ((error: unknown) => boolean) | undefined, + task: () => unknown, + ): Promise { + let value: unknown; + + try { + value = await task(); + } catch (error) { + // An answer that says "none" — a 404 for a programme with no details — + // is as much worth keeping as one with something in it. Anything else + // is a failure, kept by nobody, so the next run asks again. + if (miss?.(error) === true) { + this.#keep(kept, key, { until, value: null, missed: true }); + } + + return { error }; + } + + // Refused rather than kept as whatever JSON makes of it: this run would be + // handed a `Date` and the next one a string, and a site would find out + // only on the run after the one it was tried on. + const found = value === undefined ? undefined : unplain(value, 'value'); + + if (found !== undefined) { + return { + error: new TypeError( + `Site "${this.#site}": remember(${JSON.stringify(key)}) was answered with ${found}. ` + + `A remembered value must be plain — strings, finite numbers, booleans, null, ` + + `arrays and plain objects — since it is stored as JSON and read back by the next run`, + ), + refused: true, + }; + } + + // A copy, so what the task handed back can be changed by whoever was + // handed it without changing what is kept. + const entry: Kept = { until, value: structuredClone(value) }; + + this.#keep(kept, key, entry); + + return { value: entry.value }; + } + + #keep(kept: TrackedMap, key: string, entry: Kept): void { + if (entry.until > this.#now) { + kept.set(key, entry); + this.#fresh.add(key); + + return; + } + + // Already past: kept for this run, which is all it would have been good + // for, rather than written to be dropped by the next one. Said once, + // since an `until` in seconds where milliseconds were meant puts every + // value here, and a run asking everything again with nothing said would + // be a puzzle. + this.#passing.set(key, entry); + + if (!this.#warnedPast) { + this.#warnedPast = true; + this.#warn?.( + `remember(${JSON.stringify(key)}) was given an until already past ` + + `(${new Date(entry.until).toISOString()}), so it is kept for this run only — ` + + `an until in seconds rather than milliseconds?`, + ); + } + } +} diff --git a/src/grabber/site-run.ts b/src/grabber/site-run.ts index 47838b6..1e63ca2 100644 --- a/src/grabber/site-run.ts +++ b/src/grabber/site-run.ts @@ -29,6 +29,7 @@ import { resolveChannels } from './channels.js'; import { parseContext, requestContext, streamContext, type ContextDeps } from './context.js'; import { LocalSlot } from './local-slot.js'; import { planRequests, type Pair, type Request } from './planner.js'; +import { Memory } from './remember.js'; import { forget, isUnchanged, @@ -186,6 +187,8 @@ export class SiteRun { /** Filled in by {@link run}, since each of them has to be waited for. */ #siteState: SiteState = new Map(); + /** What the site remembers between runs — opened the first time it asks. */ + readonly #memory: Memory; #validators: TrackedMap = new TrackedMap(); #channels: GrabberChannel[] = []; @@ -222,6 +225,10 @@ export class SiteRun { }); this.#http = this.#session.pacing.http; this.#state = this.#session.state; + this.#memory = new Memory(this.#state, this.#site, run.now, { + refresh: this.#resolved.staleness.refetchAll, + warn: this.#says.warn, + }); /** * What this bounds is how many responses the site has in hand at once: @@ -301,7 +308,9 @@ export class SiteRun { // Beside `dispose`, and for the same reason: a site that threw, or was // cancelled part way, has as much to hand back as one that finished — the // channel list it fetched, and whatever its own code remembered. Only the - // groups that changed are written. + // groups that changed are written. What the site remembered is pruned + // first, so a site that no longer asks stops keeping what it kept. + await this.#memory.prune(); await this.#state.save(); // For a failed site too, where every count is nought — which is the truth, // and reads correctly beside the `site:failed` that says why. @@ -508,6 +517,7 @@ export class SiteRun { ...(signal ? { signal } : {}), timeZone: this.#run.timeZone, dayZone: this.#resolved.dayZone, + memory: this.#memory, }; } @@ -640,6 +650,7 @@ export class SiteRun { // the request being parsed did — ahead of the planned ones, so a // channel-day in hand is finished rather than joined by another. paced: (task) => slot.away(() => this.#sending(task, { priority: 1 })), + away: (wait) => slot.away(wait), }), ); @@ -901,6 +912,8 @@ export class SiteRun { // As a parse's is, and ahead of the planned ones for the same // reason: a pass already under way is finished rather than joined. paced: (task) => this.#sending(task, { priority: 1 }), + // A pass holds no slot while it runs, so it waits as it is. + away: (wait) => wait(), }), )) { // Between emissions, which is as often as this has anything to say: a diff --git a/src/grabber/state.ts b/src/grabber/state.ts index d14fb81..beb9690 100644 --- a/src/grabber/state.ts +++ b/src/grabber/state.ts @@ -31,6 +31,8 @@ export const StateKey = { VALIDATORS: 'validators', /** Whatever the site itself put there. */ SITE: 'state', + /** What a site's `remember` kept, each value with when it is forgotten. */ + REMEMBERED: 'remembered', } as const; /** @@ -448,7 +450,16 @@ export class SiteStateHandle { let pending = this.#groups.get(key); if (pending === undefined) { - pending = this.#store.getState(this.#site, key).then(make); + const reading = this.#store.getState(this.#site, key).then(make); + + // A read that failed is not remembered as the group: a store that was + // briefly out of descriptors is asked again by whoever asks next. + reading.catch(() => { + if (this.#groups.get(key) === reading) { + this.#groups.delete(key); + } + }); + pending = reading; this.#groups.set(key, pending); } diff --git a/src/grabber/types.ts b/src/grabber/types.ts index ec266ce..40a1a9e 100644 --- a/src/grabber/types.ts +++ b/src/grabber/types.ts @@ -169,6 +169,39 @@ interface SiteZones { dayZone: string; } +/** How long a remembered value is kept, and which failures are an answer. */ +export interface RememberOptions { + /** + * When it is forgotten: a `Date`, or epoch milliseconds. For a programme's + * details, a while after it ends — once its day has left the window nothing + * asks for them again. + */ + until: Date | number; + /** + * A failure that is an answer — a 404 for a programme with no details — kept + * as `null`, so the next run does not ask again. Any other is thrown and kept + * by nobody, and the next run does ask. + */ + miss?: (error: unknown) => boolean; +} + +/** + * What a site kept under `key` on an earlier run, or `task`'s answer, kept for + * the next — see {@link ParseContext.remember}. + * + * `null` is in the answer exactly when the ask says some failure is one — has + * a `miss`, or options typed as possibly having one. A "none" kept for an ask + * with a `miss` is never handed to one without: it asks for itself. + */ +export interface Remember { + ( + key: string, + options: RememberOptions & { miss?: undefined }, + task: () => T | Promise, + ): Promise; + (key: string, options: RememberOptions, task: () => T | Promise): Promise; +} + /** * What every context carries, whichever shape the rest of it takes: the * channel-days it is for, the site's client, its state, and the run's signal. @@ -196,6 +229,8 @@ export interface BaseRequestContext extends Says, SiteZones { signal?: AbortSignal; /** What this site remembers between runs — see {@link SiteState}. */ state: SiteState; + /** Something kept between runs, with a time to forget it — see {@link ParseContext.remember}. */ + remember: Remember; } /** @@ -510,6 +545,30 @@ export interface ParseContext extends Says, SiteZones { * morning. Anything else throws `XmltvDateError`. */ at(time: string, day?: string): Date; + /** + * What the site kept under `key` on an earlier run, or `task`'s answer, kept + * for the next — a programme's details, fetched once rather than every time + * its day is fetched again. + * + * ```ts + * const details = await remember( + * `detail:${item.id}|${item.start}`, + * { + * until: new Date(item.stop).getTime() + 86_400_000, + * miss: (error) => error instanceof HTTPError && error.response.status === 404, + * }, + * async () => trim(await paced(({ signal }) => http.get(`detail/${item.id}`, { signal }).json())), + * ); + * ``` + * + * Kept until `until`, and only an answer is kept: a failure `miss` does not + * match is thrown, and the next run asks again. Two asks of one key at once + * are one task, each reading what it came to with its own `miss`, and each + * handed a copy. What is kept must be plain — it is stored as JSON — so keep only what + * the parse uses: the site's memory is one group, written whole when it + * changes. + */ + remember: Remember; /** * What {@link SiteConfig.request} handed back, whole and unchanged. A request * covering several channel-days is parsed once for each of them, and every diff --git a/test/grabber.test.ts b/test/grabber.test.ts index e0b4f0d..8e9ec8a 100644 --- a/test/grabber.test.ts +++ b/test/grabber.test.ts @@ -2,7 +2,7 @@ import { createServer } from 'node:http'; import type { IncomingMessage, ServerResponse } from 'node:http'; import type { AddressInfo } from 'node:net'; import { HTTPError, NetworkError } from 'ky'; -import { describe, expect, it, vi } from 'vitest'; +import { describe, expect, expectTypeOf, it, vi } from 'vitest'; import type { CacheEntryMeta, CacheStore, @@ -10,6 +10,7 @@ import type { StateEntry, StoredStateMeta, } from '../src/cache/types.js'; +import { CacheManager, MemoryCacheDriver } from '../src/cache/main.js'; import type { XmltvProgramme } from '../src/xmltv/types.js'; import { fellShort, @@ -30,6 +31,8 @@ import type { ChannelsContext, GrabberChannel, GrabSummary, + Remember, + RememberOptions, SiteConfig, StreamedChannelDay, StreamSiteConfig, @@ -1382,9 +1385,10 @@ describe('grab', () => { await grab([config], { cache, now: NOW }); // A site that never touches its bag costs no write — the group is read - // once per site run and left as it was. + // once per site run and left as it was. So is what it remembers, read at + // the end to drop what has passed, and finding nothing to drop. expect(cache.stateWrites).toEqual([]); - expect(cache.stateReads).toEqual(['example.com|state']); + expect(cache.stateReads).toEqual(['example.com|state', 'example.com|remembered']); }); it('is empty for a store that remembers nothing, and still works', async () => { @@ -3694,6 +3698,424 @@ describe('a stream that says nothing has changed', () => { }); }); +describe('what a site remembers between runs', () => { + const LATER = new Date(NOW.getTime() + 3_600_000); + const UNTIL = NOW.getTime() + 86_400_000; + + /** A site asking for one programme's details through `remember`, saying how often it asked. */ + function detailed( + asked: string[], + answer: () => unknown = () => ({ desc: 'Details' }), + options: Partial = {}, + ): SiteConfig { + return makeConfig({ + async parseDay({ day, remember }) { + const details = await remember(`detail:${day}`, { until: UNTIL, ...options }, () => { + asked.push(day); + + return answer(); + }); + + return [ + { + ...programme(`${day}T06:00:00.000Z`), + ...(details === null ? {} : { desc: [{ value: JSON.stringify(details) }] }), + }, + ]; + }, + }); + } + + it('asks once, and the next run is handed what it kept', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + + await grab([detailed(asked)], { cache, now: NOW }); + // Today is fetched again every run, and parsed again with it. + await grab([detailed(asked)], { cache, now: LATER }); + + expect(asked).toEqual([TODAY]); + expect( + cache.get({ site: 'example.com', channelId: 'one.example', day: TODAY })?.programmes, + ).toMatchObject([{ desc: [{ value: '{"desc":"Details"}' }] }]); + }); + + it('keeps it in a group of its own, leaving the site’s state alone', async () => { + const cache = new MemoryCache(); + + await grab([detailed([])], { cache, now: NOW }); + + expect(cache.stateWrites).toEqual(['example.com|remembered']); + expect(cache.state.get('example.com|remembered')?.data).toEqual([ + [`detail:${TODAY}`, { until: UNTIL, value: { desc: 'Details' } }], + ]); + }); + + it('keeps a failure `miss` matches as null, and asks again after one it does not', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const missing = Object.assign(new Error('404'), { status: 404 }); + const miss = (error: unknown) => (error as { status?: number }).status === 404; + + await grab([detailed(asked, () => Promise.reject(missing), { miss })], { cache, now: NOW }); + await grab([detailed(asked, () => Promise.reject(missing), { miss })], { cache, now: LATER }); + + // Answered "none" once, and remembered as that. + expect(asked).toEqual([TODAY]); + + const failing = new MemoryCache(); + const tried: string[] = []; + const down = () => Promise.reject(new Error('the source is down')); + + const first = await grab([detailed(tried, down, { miss })], { cache: failing, now: NOW }); + await grab([detailed(tried, down, { miss })], { cache: failing, now: LATER }); + + expect(first.failed).toBe(1); + expect(tried).toEqual([TODAY, TODAY]); + }); + + it('forgets a value once its until has passed', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const until = NOW.getTime() + 60_000; + + await grab([detailed(asked, undefined, { until })], { cache, now: NOW }); + await grab([detailed(asked, undefined, { until })], { cache, now: LATER }); + + expect(asked).toEqual([TODAY, TODAY]); + }); + + it('drops what has passed from what it stores, though nothing is asked', async () => { + const cache = new MemoryCache(); + + cache.seedState('example.com', 'remembered', [ + ['old', { until: NOW.getTime() - 1, value: 'gone' }], + ['current', { until: UNTIL, value: 'kept' }], + ]); + + // A site that no longer remembers anything. + await grab([makeConfig()], { cache, now: NOW }); + + expect(cache.state.get('example.com|remembered')?.data).toEqual([ + ['current', { until: UNTIL, value: 'kept' }], + ]); + }); + + it('asks once for two asks of one key at the same time', async () => { + const cache = new MemoryCache(); + let asked = 0; + const answers: unknown[] = []; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + // A broadcast past midnight is in both days' listings, and both + // ask about it while the first answer is still on its way. + const ask = () => + remember('shared', { until: UNTIL }, async () => { + asked++; + await new Promise((resolve) => setTimeout(resolve, 20)); + + return 'once'; + }); + + answers.push(...(await Promise.all([ask(), ask()]))); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW }, + ); + + expect(asked).toBe(1); + expect(answers).toEqual(['once', 'once']); + }); + + it('refuses a value JSON would bring back as something else', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab([detailed([], () => new Date(NOW))], { cache, now: NOW, reporter: report.reporter }); + + expect(errorMessage(report.of('entry:failed')[0]?.error)).toContain( + 'remember("detail:2026-07-17") was answered with an instance of Date at value', + ); + }); + + it('refuses to keep anything without an until', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab([detailed([], undefined, { until: undefined as unknown as number })], { + cache, + now: NOW, + reporter: report.reporter, + }); + + expect(errorMessage(report.of('entry:failed')[0]?.error)).toContain( + 'remember("detail:2026-07-17") needs an until', + ); + }); + + it('keeps what it was answered, not what a caller made of it afterwards', async () => { + // A store that keeps state as JSON, as every real one does. + const cache = new CacheManager({ driver: new MemoryCacheDriver() }); + const seen: unknown[] = []; + const site = () => + makeConfig({ + async parseDay({ day, remember }) { + // An answer the site holds on to, and a copy it was handed: changing + // either afterwards changes neither what is kept nor the next answer. + const answer = { cast: ['Ann'], year: 2001 }; + const ask = () => remember('details', { until: UNTIL }, () => answer); + const first = await ask(); + + answer.year = 1999; + first.cast.push('changed by the parse'); + seen.push(await ask()); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER }); + + expect(seen).toEqual([ + { cast: ['Ann'], year: 2001 }, + { cast: ['Ann'], year: 2001 }, + ]); + }); + + it('reads nothing kept by an earlier run under --refresh, and still keeps', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const site = () => + makeConfig({ + async parseDay({ day, remember }) { + for (const _ of [1, 2]) { + await remember('details', { until: UNTIL }, () => { + asked.push(day); + + return 'fresh'; + }); + } + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER, staleness: { refetchAll: true } }); + + // Asked again under --refresh, once: what this run asked it may read. + expect(asked).toEqual([TODAY, TODAY]); + }); + + it('hands a kept "none" only to an ask that takes a failure for one', async () => { + const cache = new MemoryCache(); + const missing = Object.assign(new Error('404'), { status: 404 }); + const miss = (error: unknown) => (error as { status?: number }).status === 404; + let asked = 0; + const answers: unknown[] = []; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + const task = () => { + asked++; + + return Promise.reject(missing); + }; + + // Both at once: the second joins the first, and reads its failure + // with its own options. + const both = await Promise.allSettled([ + remember('details', { until: UNTIL, miss }, task), + remember('details', { until: UNTIL }, task), + ]); + + answers.push(...both.map((one) => (one.status === 'fulfilled' ? one.value : 'thrown'))); + // And after: kept as "none", which this ask does not take for an answer. + answers.push(await remember('details', { until: UNTIL }, () => 'its own')); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW }, + ); + + expect(answers).toEqual([null, 'thrown', 'its own']); + expect(asked).toBe(1); + }); + + it('keeps a value whose until has already passed for this run only, and says so', async () => { + const cache = new MemoryCache(); + const report = collect(); + let asked = 0; + + await grab( + [ + makeConfig({ + async parseDay({ day, remember }) { + for (const key of ['one', 'one', 'two']) { + // In seconds, where milliseconds were meant. + await remember(key, { until: UNTIL / 1000 }, () => ++asked); + } + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter }, + ); + + expect(asked).toBe(2); + expect(cache.stateWrites).toEqual([]); + expect( + report.of('site:warning').filter(({ message }) => message.includes('already past')), + ).toHaveLength(1); + }); + + it('names the site and the key when the options are not ones', async () => { + const cache = new MemoryCache(); + const report = collect(); + + await grab( + [ + makeConfig({ + channels: [channel('one.example'), channel('two.example')], + async parseDay({ channel: { xmltvId }, day, remember }) { + await (xmltvId === 'one.example' + ? remember('a', { until: '2026-07-18' as unknown as number }, () => 1) + : ( + remember as (key: string, options: unknown, task: () => number) => Promise + )('b', undefined, () => 1)); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter }, + ); + + expect(report.of('entry:failed').map(({ error }) => errorMessage(error))).toEqual([ + 'Site "example.com": remember("a") needs an until, a Date or epoch milliseconds (got "2026-07-18")', + 'Site "example.com": remember("b") needs an until, a Date or epoch milliseconds (got none)', + ]); + }); + + it('asks again after a read of what it kept failed, and finishes the run', async () => { + const cache = new MemoryCache(); + const report = collect(); + const read = cache.getState.bind(cache); + let failures = 1; + + cache.getState = async (site, key) => { + if (key === 'remembered' && failures-- > 0) { + throw Object.assign(new Error('EMFILE: too many open files'), { code: 'EMFILE' }); + } + + return read(site, key); + }; + + const summary = await grab( + [ + makeConfig({ + channels: [channel('one.example'), channel('two.example')], + async parseDay({ day, remember }) { + await remember('details', { until: UNTIL }, () => 'answer'); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, reporter: report.reporter, localConcurrency: 1 }, + ); + + // The first channel-day met the failed read; the second asked again. + expect(summary).toMatchObject({ fetched: 1, failed: 1, sitesFailed: 0 }); + expect(report.of('site:done')).toHaveLength(1); + expect(report.of('site:failed')).toHaveLength(0); + }); + + it('lets a parse waiting on another parse’s ask give up its slot meanwhile', async () => { + // One local slot, and one request covering two days whose listings both + // name a broadcast past midnight: the second day's parse waits on the + // first's ask, whose detail request needs the slot back to finish. + const cache = new MemoryCache(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), 5000); + let asked = 0; + + try { + const summary = await grab( + [ + makeDaysConfig({ + days: 2, + async parseDay({ day, remember, paced }) { + await remember('past midnight', { until: UNTIL }, () => + paced(async () => { + asked++; + await new Promise((resolve) => setTimeout(resolve, 20)); + + return 'details'; + }), + ); + + return [programme(`${day}T06:00:00.000Z`)]; + }, + }), + ], + { cache, now: NOW, localConcurrency: 1, signal: controller.signal }, + ); + + expect(controller.signal.aborted).toBe(false); + expect(summary.fetched).toBe(2); + expect(asked).toBe(1); + } finally { + clearTimeout(timer); + } + }); + + it('types an answer as possibly null only where a failure may be one', () => { + // Typechecked, never run: what each way of asking is promised. + const typed = (remember: Remember, held: RememberOptions) => { + expectTypeOf(remember('k', { until: 1 }, () => 1)).toEqualTypeOf>(); + expectTypeOf(remember('k', { until: 1, miss: () => true }, () => 1)).toEqualTypeOf< + Promise + >(); + expectTypeOf(remember('k', held, () => 1)).toEqualTypeOf>(); + }; + + expect(typed).toBeTypeOf('function'); + }); + + it('is there for a request too', async () => { + const cache = new MemoryCache(); + const asked: string[] = []; + const site = () => + makeConfig({ + async request({ day, remember }) { + return remember(`token:${day}`, { until: UNTIL }, () => { + asked.push(day); + + return 'token'; + }); + }, + }); + + await grab([site()], { cache, now: NOW }); + await grab([site()], { cache, now: LATER }); + + expect(asked).toEqual([TODAY]); + }); +}); + describe('what a run reports', () => { const types = (events: EpgEvent[]): string[] => events.map((event) => event.type); diff --git a/test/readme-examples.ts b/test/readme-examples.ts index d98c03d..c0a0282 100644 --- a/test/readme-examples.ts +++ b/test/readme-examples.ts @@ -508,6 +508,47 @@ export const remembering = defineSiteConfig({ parseDay: () => [], }); +// --- docs/site-config.md: A programme's details, asked once ----------------- +interface Detail { + description: string; + cast: string[]; +} + +export const askedOnce = defineSiteConfig<{ + items: { id: string; start: string; end: string; title: string }[]; +}>({ + site: 'example.tv', + channels: [], + async request({ channel, day, http }) { + return http.get(`epg/${channel.siteId}/${day}`).json(); + }, + async parseDay({ payload, http, paced, remember, programme }) { + return Promise.all( + payload.items.map(async (item) => { + const details = await remember( + `detail:${item.id}|${item.start}`, + { + // Asked about while its day can be fetched again, and no longer. + until: Date.parse(item.end) + 86_400_000, + // "This one has no details" is an answer, and worth keeping. + miss: (error) => error instanceof HTTPError && error.response.status === 404, + }, + async () => { + const full = await paced(({ signal }) => + http.get(`detail/${item.id}`, { signal }).json(), + ); + + return { desc: full.description, cast: full.cast }; // only what the parse uses + }, + ); + + const p = programme(new Date(item.start), item.title); + return details ? p.desc(details.desc) : p; + }), + ); + }, +}); + // The mode decides which caps are accepted, and the shape of the context. export const wrongCap = defineSiteConfig({ site: 'example.tv',