From ec599e649442988cbd41d5bf79cded53d91b0511 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Thu, 15 Oct 2026 20:27:46 +0200 Subject: [PATCH 1/3] feat(enrichment): the metadata cache and the engine that fills it Step C of the RFC-011 plan. No surface reads the cache yet; every container answers as it does today. - Settings under `WAVEFLOW_METADATA_*`, off unless a provider is named. Naming `lastfm` without its key, an unknown name or a name twice refuses to boot. Scrobbling's key does not switch this on. - `metadata_artist` and `metadata_artist_similar`, keyed on the provider and the `artist.canonical_name` form of the name. - The engine: a fill per provider and name, in the background, spaced to the configured rate and collapsed when concurrent. A hit stands for the TTL, a miss for the miss TTL, and an unreached lookup writes nothing. The writer gate is taken for the row write only. - The outbound client moves out of `scrobblers/` into `outbound`, which both use. - RFC-011 gains a Revised line and corrections to decisions 4, 6 and 10; `operations.md` says how to switch enrichment on. Signed-off-by: InstaZDLL --- CLAUDE.md | 2 +- Cargo.toml | 4 + docs/operations.md | 41 ++ docs/rfcs/RFC-011-metadata-enrichment.md | 49 +- .../20261015000000_metadata_artist.sql | 53 ++ src/config.rs | 159 ++++++ src/database.rs | 130 +++++ src/enrichment.rs | 486 ++++++++++++++++++ src/lib.rs | 14 +- src/outbound.rs | 47 ++ src/scrobblers/mod.rs | 46 +- tests/enrichment.rs | 435 ++++++++++++++++ 12 files changed, 1422 insertions(+), 44 deletions(-) create mode 100644 migrations-v2/20261015000000_metadata_artist.sql create mode 100644 src/enrichment.rs create mode 100644 src/outbound.rs create mode 100644 tests/enrichment.rs diff --git a/CLAUDE.md b/CLAUDE.md index d9dede4d..e42c6c40 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,7 +44,7 @@ bun run build # webapp first, then cargo — the order Tests need no service container: `test_app()` builds a `Config` over a `TempDir` and the suite drives the same router `main` serves. FFmpeg and ffprobe must be on `PATH`. -The integration suite is seventeen targets under `tests/`, one per subject — `auth`, `scanner`, `identity`, `catalog`, `subsonic_contract`, `subsonic_browse`, `subsonic_fields`, `subsonic_methods`, `native_api`, `sync`, `media`, `service`, `cli`, `scrobbling`, `uploads`, `canvas`, `migrations` — nearly all of them sharing `tests/support/mod.rs`. `migrations` is the exception and deliberately so: every other target starts from `test_app()`, which builds its database from nothing, and that is exactly the path an upgrade does not take. It seeds a database at each **released** schema, boots the server on it, and reads the rows back through the services — and it is where anything else that needs a database *behind* this binary belongs, such as proving that a key change reads without migrating and copies the pair aside before it does. `autotests = false`, so each is a target only because `Cargo.toml` declares it: a new file under `tests/` is compiled by nothing until a `[[test]]` entry names it, and the count above is worth checking against that list rather than trusted. `support/mod.rs` is the exception that proves it — shared module code, declared nowhere, a target never. File a new test by the surface it exercises, and check that by what it calls rather than by what its name says. +The integration suite is eighteen targets under `tests/`, one per subject — `auth`, `scanner`, `identity`, `catalog`, `subsonic_contract`, `subsonic_browse`, `subsonic_fields`, `subsonic_methods`, `native_api`, `sync`, `media`, `service`, `cli`, `scrobbling`, `enrichment`, `uploads`, `canvas`, `migrations` — nearly all of them sharing `tests/support/mod.rs`. `migrations` is the exception and deliberately so: every other target starts from `test_app()`, which builds its database from nothing, and that is exactly the path an upgrade does not take. It seeds a database at each **released** schema, boots the server on it, and reads the rows back through the services — and it is where anything else that needs a database *behind* this binary belongs, such as proving that a key change reads without migrating and copies the pair aside before it does. `autotests = false`, so each is a target only because `Cargo.toml` declares it: a new file under `tests/` is compiled by nothing until a `[[test]]` entry names it, and the count above is worth checking against that list rather than trusted. `support/mod.rs` is the exception that proves it — shared module code, declared nowhere, a target never. File a new test by the surface it exercises, and check that by what it calls rather than by what its name says. `cargo build` works without a client build — `build.rs` embeds a placeholder page when `webapp/dist` is absent. A real `vite build` overwrites it. diff --git a/Cargo.toml b/Cargo.toml index e53e6e93..9c99c9a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -154,3 +154,7 @@ path = "tests/canvas.rs" [[test]] name = "scrobbling" path = "tests/scrobbling.rs" + +[[test]] +name = "enrichment" +path = "tests/enrichment.rs" diff --git a/docs/operations.md b/docs/operations.md index 278010ee..e3e15231 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -169,6 +169,47 @@ Every tunable is a field on `Config` in [`src/config.rs`](../src/config.rs), with its environment variable documented on it. That file is the authority; this page is not a copy of it. +## Switching metadata enrichment on + +**Off by default, and an upgrade leaves it off.** A server that has not been +told otherwise makes no request to any metadata provider. +[RFC-011](rfcs/RFC-011-metadata-enrichment.md) carries the reasoning. + +```bash +WAVEFLOW_METADATA_PROVIDERS=lastfm +WAVEFLOW_METADATA_LASTFM_KEY=… # your own Last.fm API application's key +``` + +| Variable | Default | What it does | +| --- | --- | --- | +| `WAVEFLOW_METADATA_PROVIDERS` | empty | The switch: an ordered, comma-separated list. Empty is off. Today the only provider is `lastfm` | +| `WAVEFLOW_METADATA_LASTFM_KEY` | — | Required when `lastfm` is listed; the server refuses to start without it. No key ships in the binary | +| `WAVEFLOW_METADATA_IMAGES` | on | `off` keeps the lookups and refuses artist image downloads | +| `WAVEFLOW_METADATA_TTL_DAYS` | 30 | How long an answer with content is kept before it is asked again | +| `WAVEFLOW_METADATA_MISS_TTL_DAYS` | 7 | How long "the provider knows nothing about this artist" is kept | +| `WAVEFLOW_METADATA_RATE_PER_SEC` | 4 | Requests per second, per provider | +| `WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS` | 10 | How long one request may take before it counts as unanswered | + +**Scrobbling's Last.fm key does not switch this on.** An application you +registered for submitting listens is a different consent from lookups about your +whole catalogue. You may place the same key in both variables; each one switches +on only its own feature. + +**What the server starts contacting.** `ws.audioscrobbler.com`, over https, and +only when somebody opens an artist. Nothing is fetched at scan time. Each +request carries the artist's name and your key, and no redirect is followed. +Requests are spaced to the rate above, and simultaneous visits to one artist +share one request. + +**What it stores.** In `waveflow.db`, for each provider and each artist name +asked about: the biography, the ranked list of similar artists by name, whether +the provider knew the artist, and when to ask again. Nothing in it belongs to an +account, and nothing records who asked. A failed request stores nothing, so the +next visit asks again. + +**What it logs.** The provider and a verdict, such as `answered HTTP 503` — +never the request URL, which carries your key. + ## Probes | Path | Answers | diff --git a/docs/rfcs/RFC-011-metadata-enrichment.md b/docs/rfcs/RFC-011-metadata-enrichment.md index 16d05a56..8d472a0c 100644 --- a/docs/rfcs/RFC-011-metadata-enrichment.md +++ b/docs/rfcs/RFC-011-metadata-enrichment.md @@ -1,10 +1,18 @@ # RFC-011 — Server-side metadata enrichment - **Status**: Proposed -- **Implemented by**: nothing yet. The *Status* field above never flips in this - project; this line is the one that says what is running, and today it says - the answer is still the empty container. +- **Implemented by**: in progress, in the order of + [the implementation plan](../metadata-enrichment-plan.md): + InstaZDLL/WaveFlow#855 (the core's clients accept a client, an address and a + key), #270 (the pin moved onto them), and the pull request closing #265 (the + cache and the engine that fills it). No surface answers from the cache yet, + so every container is still the empty one. The *Status* field above never + flips in this project; this line is the one that says what is running. - **Date**: 2026-10-14 +- **Revised**: 2026-10-15, while implementing the cache (#265). Decisions 4, 6 + and 10 rested on attachment points the code does not hold; each correction + below is marked and says what the first version asserted, rather than + erasing it. - **Authors**: the WaveFlow project - **Depends on**: [RFC-002](RFC-002-waveflow-server-v2.md), [RFC-010](RFC-010-external-scrobbling.md) @@ -170,6 +178,14 @@ The cache is therefore keyed on the **canonical name**, as the desktop's own - `metadata_album` — the same shape for album notes, keyed on the provider and the canonical `(artist, title)` pair that `album_match.rs` already computes. +**Revised on 2026-10-15.** The migration that shipped +(`20261015000000_metadata_artist.sql`) differs from this list in three places. +The biography is two columns, `bio_summary` and `bio_full`, because that is how +the providers return it. `image_url` is not there: portraits add their columns +with the fetch that fills them. And `metadata_album` was not created — the +open question on album notes below decides whether it exists, and +`album_match.rs` was not even in the pinned core when this was written. + ### One row per provider, merged on read **The provider belongs in the key, not in a column**, and the first draft of @@ -204,6 +220,16 @@ matching a provider's ranked candidate list back to the name asked for is `select_by_name`. Neither is re-implemented here; both have a desktop issue behind them (#342) recording what naive equality dropped. +**Revised on 2026-10-15: the key is not `normalize_name`.** The paragraph above +asserted it was. Every `artist` row already stores a `canonical_name`, computed +by `waveflow_core::scanner::canonical_name`, and the two functions disagree: +`normalize_name` also folds diacritics and turns `&` into `and`, so "Björk" is +`björk` in the column and `bjork` through it. The cache is keyed on the +column's form, which lets the read path join +`metadata_artist.name_canonical = artist.canonical_name` with nothing +recomputed. `normalize_name` and `select_by_name` keep the job they were written +for — matching a provider's candidates — where a provider returns candidates. + **Nothing in these tables is account data**, which is what makes an instance-wide cache correct rather than merely convenient. Anything that ever becomes account-specific — a per-person override, a hidden artist — belongs in @@ -267,6 +293,13 @@ for someone to reload the page. The mechanism already exists and decision 8 explains why it may now carry an artist: RFC-007's library event stream, whose decision 6 declined an artist event for a reason this RFC removes. +**Revised on 2026-10-15: only one client is told.** The paragraph above implied +every client is. The desktop is, because it mirrors the library through that +feed. The web client is not: the only event stream it reads is a scan's +progress. It re-asks instead, a bounded number of times while the answer says +the lookup is pending, and then stops — which is the web client's step of the +plan, not a property this mechanism gives it. + ## Decision 7 — similar artists resolve to artists the caller can see, and to nothing else `artist.getSimilar` returns names. Subsonic's `similarArtist` entries carry an @@ -415,6 +448,16 @@ them empty. - `getArtistInfo` / `getArtistInfo2` — `biography`, `similarArtist` entries (decision 7), and `smallImageUrl` / `mediumImageUrl` / `largeImageUrl` pointing at this server's own artwork route (decision 8). + + **Revised on 2026-10-15: the three image URLs stay absent.** The bullet above + asserted a route they could point at. Neither artwork route serves a URL a + client loads as-is: `media::artwork` wants the native bearer, and + `getCoverArt` wants Subsonic credentials in the query string, which would put + a member's token into a URL that clients cache and log. A client finds the + portrait through `coverArt`, which every artist element already carries. The + compatibility campaign measures whether a client in the matrix reads + `largeImageUrl` at all; a sealed artwork ticket can be added then, while a URL + scheme cannot be withdrawn once clients have cached it. - `getSimilarSongs` / `getSimilarSongs2` — tracks by the neighbours the caller can see. - `getAlbumInfo` / `getAlbumInfo2` — `notes` beside the `musicBrainzId` they diff --git a/migrations-v2/20261015000000_metadata_artist.sql b/migrations-v2/20261015000000_metadata_artist.sql new file mode 100644 index 00000000..3da05afb --- /dev/null +++ b/migrations-v2/20261015000000_metadata_artist.sql @@ -0,0 +1,53 @@ +-- What the metadata providers said about an artist — RFC-011 decisions 4 and 5. +-- +-- **Instance-wide, and nothing in it is account data.** A biography is the same +-- text whoever asks, and `artist` rows are per library, so keying on +-- `artist.id` would fetch the same answer once per library. The key is the +-- name instead, in the form `artist.canonical_name` already stores +-- (`waveflow_core::scanner::canonical_name`), so the read path joins on that +-- column with nothing recomputed. RFC-011 first named +-- `name_match::normalize_name` here; the two disagree — "Björk" is `björk` in +-- the column and `bjork` through it — and its erratum says so. +-- +-- **The provider is part of the key.** Each provider keeps its own row, its +-- own outcome and its own clock, and the fields are merged at read time in the +-- operator's order. One row per name would have to pick a winner when written, +-- and would let one provider's "never heard of them" silence another that has +-- a biography. +-- +-- `provider` carries no CHECK: the vocabulary is the server's `Provider` enum, +-- and widening a CHECK in SQLite means rebuilding the table. `outcome` does, +-- because its two words are decision 5's and do not grow — the third outcome, +-- unreached, is the absence of a row. +-- +-- Album notes (`metadata_album`) wait for the RFC's open question on them. +-- Portraits add their columns with the fetch that fills them. +CREATE TABLE metadata_artist ( + provider TEXT NOT NULL, + name_canonical TEXT NOT NULL, + outcome TEXT NOT NULL CHECK (outcome IN ('hit', 'miss')), + bio_summary TEXT, + bio_full TEXT, + fetched_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + PRIMARY KEY (provider, name_canonical) +) STRICT, WITHOUT ROWID; + +-- The ranked neighbours, as the provider gave them. Kept by name and resolved +-- to local artists only when read, through the caller's own libraries +-- (decision 7): an id here would name a row in somebody's library. +-- +-- Rewritten whole with its parent row, in the same transaction, so a list is +-- never half one answer and half the next. The cascade only matters to a row +-- deleted outright; a refresh updates the parent in place. +CREATE TABLE metadata_artist_similar ( + provider TEXT NOT NULL, + name_canonical TEXT NOT NULL, + rank INTEGER NOT NULL, + similar_name TEXT NOT NULL, + similar_canonical TEXT NOT NULL, + score REAL NOT NULL, + PRIMARY KEY (provider, name_canonical, rank), + FOREIGN KEY (provider, name_canonical) + REFERENCES metadata_artist(provider, name_canonical) ON DELETE CASCADE +) STRICT, WITHOUT ROWID; diff --git a/src/config.rs b/src/config.rs index 08b90814..0f3b3d0a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -189,6 +189,75 @@ pub struct ScrobbleDestination { pub fingerprint: String, } +/// Metadata enrichment, as RFC-011 decision 3 lists its settings. +/// +/// **Off unless `providers` names one**, and that is the only switch: an empty +/// list makes no outbound request of any kind, and nothing else here can +/// contradict it. Separate from scrobbling on purpose — an operator who placed +/// a Last.fm application for submissions has not thereby agreed to lookups +/// about the whole catalogue, so `lastfm_key` is its own setting even when it +/// holds the same value. +#[derive(Clone)] +pub struct MetadataSettings { + /// `WAVEFLOW_METADATA_PROVIDERS`: an ordered, comma-separated list, empty + /// by default. The order is the order fields are merged in when read. + pub providers: Vec, + /// `WAVEFLOW_METADATA_IMAGES`, on by default, `off` to keep the lookups and + /// refuse decision 8's image fetch. A narrowing only: it switches nothing + /// on while `providers` is empty. + pub images: bool, + /// `WAVEFLOW_METADATA_LASTFM_KEY`, the operator's own application key. + /// Required when `providers` names `lastfm`: no key ships in this binary. + pub lastfm_key: Option, + /// `WAVEFLOW_METADATA_TTL_DAYS`: how long an answer with content stands. + pub ttl: Duration, + /// `WAVEFLOW_METADATA_MISS_TTL_DAYS`: how long "the provider knows nothing" + /// stands. Shorter than `ttl`, and a setting of its own, because it is the + /// answer most likely to change. + pub miss_ttl: Duration, + /// `WAVEFLOW_METADATA_RATE_PER_SEC`: requests per second, per provider. + pub rate_per_sec: u32, + /// `WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS`: how long one request to a + /// provider may take before it counts as unreached. + pub request_timeout: Duration, +} + +impl MetadataSettings { + /// Enrichment switched off — what a server that is merely upgraded runs. + pub fn off() -> Self { + Self { + providers: Vec::new(), + images: true, + lastfm_key: None, + ttl: Duration::from_secs(30 * 24 * 60 * 60), + miss_ttl: Duration::from_secs(7 * 24 * 60 * 60), + rate_per_sec: 4, + request_timeout: Duration::from_secs(10), + } + } +} + +/// Written by hand so the key never reaches a log: a `Config` is printed at +/// startup under `RUST_LOG=debug`, and a Last.fm key travels in a query string +/// where anyone holding it can spend the operator's quota. +impl std::fmt::Debug for MetadataSettings { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("MetadataSettings") + .field("providers", &self.providers) + .field("images", &self.images) + .field( + "lastfm_key", + &self.lastfm_key.as_ref().map(|_| "[redacted]"), + ) + .field("ttl", &self.ttl) + .field("miss_ttl", &self.miss_ttl) + .field("rate_per_sec", &self.rate_per_sec) + .field("request_timeout", &self.request_timeout) + .finish() + } +} + #[derive(Clone)] pub struct Config { pub bind_addr: SocketAddr, @@ -352,6 +421,8 @@ pub struct Config { /// flag says "this address is mine", not "send my members' tokens in clear /// wherever I point you". pub outbound_allow_plaintext: bool, + /// Metadata enrichment — RFC-011. See [`MetadataSettings`]. + pub metadata: MetadataSettings, pub allowed_origins: Vec, /// How the catalogue decides which row a scanned file belongs to. /// @@ -398,6 +469,7 @@ impl std::fmt::Debug for Config { .field("destinations", &self.destinations) .field("lastfm", &self.lastfm) .field("outbound_allow_plaintext", &self.outbound_allow_plaintext) + .field("metadata", &self.metadata) .field("allowed_origins", &self.allowed_origins) .field("pid", &self.pid) .finish() @@ -590,6 +662,8 @@ impl Config { ), }; + let metadata = parse_metadata_settings()?; + let allowed_origins = std::env::var("WAVEFLOW_ALLOWED_ORIGINS") .unwrap_or_default() .split(',') @@ -664,6 +738,7 @@ impl Config { destinations, lastfm, outbound_allow_plaintext, + metadata, allowed_origins, pid, }) @@ -766,6 +841,10 @@ impl Config { // two tests here stand on. lastfm: None, outbound_allow_plaintext: true, + // Off, as production defaults: the suite never reaches a real + // provider, and the targets that exercise enrichment build their + // own engine against a listener of their own. + metadata: MetadataSettings::off(), allowed_origins: Vec::new(), // The real defaults, so the whole test suite exercises the specs // production runs under rather than a simplified stand-in. @@ -972,6 +1051,66 @@ fn normalize_public_url(value: &str) -> anyhow::Result { Ok(parsed.origin().ascii_serialization()) } +/// Reads RFC-011 decision 3's settings. +fn parse_metadata_settings() -> anyhow::Result { + let defaults = MetadataSettings::off(); + let lastfm_key = read_trimmed_env("WAVEFLOW_METADATA_LASTFM_KEY"); + let providers = parse_metadata_providers( + &std::env::var("WAVEFLOW_METADATA_PROVIDERS").unwrap_or_default(), + lastfm_key.is_some(), + )?; + let days = |name: &str, default: Duration| -> anyhow::Result { + let days: u64 = parse_positive_env(name, default.as_secs() / (24 * 60 * 60))?; + days.checked_mul(24 * 60 * 60) + .map(Duration::from_secs) + .ok_or_else(|| anyhow::anyhow!("invalid {name}: too large")) + }; + Ok(MetadataSettings { + providers, + images: parse_bool_env("WAVEFLOW_METADATA_IMAGES", defaults.images)?, + lastfm_key, + ttl: days("WAVEFLOW_METADATA_TTL_DAYS", defaults.ttl)?, + miss_ttl: days("WAVEFLOW_METADATA_MISS_TTL_DAYS", defaults.miss_ttl)?, + rate_per_sec: parse_positive_env("WAVEFLOW_METADATA_RATE_PER_SEC", defaults.rate_per_sec)?, + request_timeout: Duration::from_secs(parse_positive_env( + "WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS", + defaults.request_timeout.as_secs(), + )?), + }) +} + +/// `WAVEFLOW_METADATA_PROVIDERS`, in the operator's order. +/// +/// A provider named without its key, an unknown name and a name given twice +/// all refuse to boot. Each is a typo, and a server that started with +/// enrichment quietly narrower than its operator wrote is the silent failure +/// these settings must not have — the operator learns it here, not from an +/// artist page that never fills. +fn parse_metadata_providers( + list: &str, + has_lastfm_key: bool, +) -> anyhow::Result> { + let mut providers = Vec::new(); + for name in list + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + { + let provider = crate::enrichment::Provider::parse(name) + .ok_or_else(|| anyhow::anyhow!("invalid WAVEFLOW_METADATA_PROVIDERS entry {name:?}"))?; + if providers.contains(&provider) { + anyhow::bail!("WAVEFLOW_METADATA_PROVIDERS names {name:?} twice"); + } + if provider == crate::enrichment::Provider::LastFm && !has_lastfm_key { + anyhow::bail!( + "WAVEFLOW_METADATA_PROVIDERS names lastfm without WAVEFLOW_METADATA_LASTFM_KEY" + ); + } + providers.push(provider); + } + Ok(providers) +} + fn parse_env(name: &str, default: &str) -> anyhow::Result where T: std::str::FromStr, @@ -1072,6 +1211,7 @@ fn default_pid_specs() -> crate::pid::PidSpecs { #[cfg(test)] mod tests { use super::normalize_public_url; + use super::parse_metadata_providers; use super::DEFAULT_DB_ACQUIRE_TIMEOUT_MS; use super::{parse_destinations, DEFAULT_SCROBBLE_DESTINATION}; use super::{validate_canvas, CanvasLimits}; @@ -1089,6 +1229,25 @@ mod tests { } } + /// RFC-011 decision 3: empty is off, and every typo refuses to boot rather + /// than starting enrichment narrower than the operator wrote. + #[test] + fn metadata_providers_are_read_strictly() { + use crate::enrichment::Provider; + + assert_eq!(parse_metadata_providers("", false).unwrap(), []); + assert_eq!(parse_metadata_providers(" , ", false).unwrap(), []); + assert_eq!( + parse_metadata_providers(" lastfm ", true).unwrap(), + [Provider::LastFm] + ); + // No key ships in the binary: naming the provider without one is a + // misconfiguration, not a quiet off. + assert!(parse_metadata_providers("lastfm", false).is_err()); + assert!(parse_metadata_providers("lastfm,lastfm", true).is_err()); + assert!(parse_metadata_providers("last.fm", true).is_err()); + } + #[test] fn upload_limits_that_contradict_each_other_are_refused() { assert!(validate_uploads(&workable()).is_ok()); diff --git a/src/database.rs b/src/database.rs index d60c5667..d0557d46 100644 --- a/src/database.rs +++ b/src/database.rs @@ -1900,6 +1900,136 @@ async fn insert_audit( Ok(()) } +/// RFC-011's cache, which [`crate::enrichment`] fills and reads. +/// +/// No user id and no tenancy anywhere in it: what a provider says about a name +/// is the same whoever asks, and nothing here is account data (decision 4). +/// Tenancy applies where a cached name meets the catalogue, which is the +/// reader's query, not these. +impl Database { + /// Every provider's row for one canonical name, in no particular order. + /// + /// A row from a provider this build does not know — written by a later + /// version the instance was downgraded from — is passed over rather than + /// refused: it is a fact about a provider nobody configured here. + pub async fn metadata_artist_rows( + &self, + name_canonical: &str, + ) -> Result, sqlx::Error> { + let rows = sqlx::query( + "SELECT provider, outcome, bio_summary, bio_full, fetched_at, expires_at \ + FROM metadata_artist WHERE name_canonical = ?", + ) + .bind(name_canonical) + .fetch_all(&self.pool) + .await?; + let mut cached = Vec::with_capacity(rows.len()); + for row in rows { + let provider: String = row.try_get("provider")?; + let Some(provider) = crate::enrichment::Provider::parse(&provider) else { + continue; + }; + let outcome: String = row.try_get("outcome")?; + let outcome = crate::enrichment::Outcome::parse(&outcome).ok_or_else(|| { + sqlx::Error::Decode(format!("unknown enrichment outcome {outcome:?}").into()) + })?; + cached.push(crate::enrichment::CachedArtist { + provider, + outcome, + bio_summary: row.try_get("bio_summary")?, + bio_full: row.try_get("bio_full")?, + fetched_at: row.try_get("fetched_at")?, + expires_at: row.try_get("expires_at")?, + }); + } + Ok(cached) + } + + /// One provider's ranked neighbours for one canonical name, best first. + pub async fn metadata_artist_similar( + &self, + provider: crate::enrichment::Provider, + name_canonical: &str, + ) -> Result, sqlx::Error> { + let rows = sqlx::query( + "SELECT similar_name, similar_canonical, score FROM metadata_artist_similar \ + WHERE provider = ? AND name_canonical = ? ORDER BY rank", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .fetch_all(&self.pool) + .await?; + rows.into_iter() + .map(|row| { + Ok(crate::enrichment::CachedSimilar { + name: row.try_get("similar_name")?, + name_canonical: row.try_get("similar_canonical")?, + score: row.try_get("score")?, + }) + }) + .collect() + } + + /// Records what a provider answered: its row and its whole neighbour list, + /// in one transaction. + /// + /// The writer gate is taken here and nowhere earlier — the request that + /// produced `answer` is over before this is called, so no network I/O ever + /// runs under it. + pub(crate) async fn write_metadata_artist( + &self, + provider: crate::enrichment::Provider, + name_canonical: &str, + answer: &crate::enrichment::ArtistAnswer, + fetched_at: i64, + expires_at: i64, + ) -> Result<(), sqlx::Error> { + let _writer = self.writer_guard().await; + let mut tx = self.pool.begin().await?; + sqlx::query( + "INSERT INTO metadata_artist \ + (provider, name_canonical, outcome, bio_summary, bio_full, fetched_at, expires_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?) \ + ON CONFLICT (provider, name_canonical) DO UPDATE SET \ + outcome = excluded.outcome, bio_summary = excluded.bio_summary, \ + bio_full = excluded.bio_full, fetched_at = excluded.fetched_at, \ + expires_at = excluded.expires_at", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .bind(answer.outcome().as_str()) + .bind(answer.bio_summary.as_deref()) + .bind(answer.bio_full.as_deref()) + .bind(fetched_at) + .bind(expires_at) + .execute(&mut *tx) + .await?; + sqlx::query( + "DELETE FROM metadata_artist_similar WHERE provider = ? AND name_canonical = ?", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .execute(&mut *tx) + .await?; + for (rank, similar) in answer.similar.iter().enumerate() { + sqlx::query( + "INSERT INTO metadata_artist_similar \ + (provider, name_canonical, rank, similar_name, similar_canonical, score) \ + VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .bind(rank as i64) + .bind(&similar.name) + .bind(&similar.name_canonical) + .bind(similar.score) + .execute(&mut *tx) + .await?; + } + tx.commit().await + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/enrichment.rs b/src/enrichment.rs new file mode 100644 index 00000000..19409487 --- /dev/null +++ b/src/enrichment.rs @@ -0,0 +1,486 @@ +//! Metadata enrichment — RFC-011: the cache, and the engine that fills it. +//! +//! **Demand-driven and never on the answering path** (decision 6). A reader asks +//! [`Enrichment::artist`] for what the cache holds; for every configured +//! provider with nothing fresh, that call schedules a fill in the background +//! and returns at once. The first visit to an artist shows nothing; the next +//! one shows what came back. +//! +//! **Three outcomes** (decision 5), written on the provider's own row: a *hit* +//! stands for the long TTL, a *miss* — the provider answered and knows nothing +//! — for the short one, and *unreached* writes nothing at all, so the next +//! visit asks again. A network blip must never become "this artist has no +//! biography" for a month. +//! +//! **Bounded per provider**: requests are spaced to the configured rate, and +//! concurrent first visits for one name collapse onto a single fill. The +//! writer gate is taken for the row write only, after the requests are over. +//! +//! **Nothing about a request is logged but the provider and a verdict.** A +//! Last.fm request carries the operator's key in its query string; neither the +//! URL nor a response body is ever written to a log (decision 9). + +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; + +use dashmap::DashMap; +use waveflow_core::metadata::{lastfm::LastfmClient, LookupError}; + +use crate::{config::MetadataSettings, database::Database}; + +/// Last.fm's API root. A constant, not a setting (decision 9): there is no +/// operator field, so there is no URL to validate. +const LASTFM_BASE_URL: &str = "https://ws.audioscrobbler.com/2.0/"; + +/// How many neighbours are asked for and kept. Subsonic's `getArtistInfo` +/// defaults to twenty; asking for more leaves room for the ones a caller cannot +/// see, which decision 7 drops at read time. +const SIMILAR_LIMIT: u32 = 50; + +/// A provider an operator can name in `WAVEFLOW_METADATA_PROVIDERS`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum Provider { + LastFm, +} + +impl Provider { + pub fn parse(value: &str) -> Option { + match value { + "lastfm" => Some(Self::LastFm), + _ => None, + } + } + + /// The name in the setting, in the cache's `provider` column, and in logs. + pub fn as_str(self) -> &'static str { + match self { + Self::LastFm => "lastfm", + } + } +} + +/// What a provider said, when it said something. Unreached has no variant: it +/// is the absence of a row. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + Hit, + Miss, +} + +impl Outcome { + pub fn parse(value: &str) -> Option { + match value { + "hit" => Some(Self::Hit), + "miss" => Some(Self::Miss), + _ => None, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::Hit => "hit", + Self::Miss => "miss", + } + } +} + +/// One provider's cached row for one name. +#[derive(Debug, Clone)] +pub struct CachedArtist { + pub provider: Provider, + pub outcome: Outcome, + pub bio_summary: Option, + pub bio_full: Option, + pub fetched_at: i64, + pub expires_at: i64, +} + +/// One neighbour in a provider's cached list. +#[derive(Debug, Clone)] +pub struct CachedSimilar { + pub name: String, + pub name_canonical: String, + pub score: f64, +} + +/// What one fill brought back from one provider, before it is written. +#[derive(Debug, Default)] +pub(crate) struct ArtistAnswer { + pub(crate) bio_summary: Option, + pub(crate) bio_full: Option, + pub(crate) similar: Vec, +} + +impl ArtistAnswer { + /// A hit carries something; an answer with nothing in it is a miss. + pub(crate) fn outcome(&self) -> Outcome { + if self.bio_summary.is_some() || self.bio_full.is_some() || !self.similar.is_empty() { + Outcome::Hit + } else { + Outcome::Miss + } + } +} + +/// Where each provider is reached. [`Endpoints::default`] is production; the +/// test suite points these at listeners of its own, which is the only reason +/// the type exists. +#[derive(Debug, Clone)] +pub struct Endpoints { + pub lastfm: String, +} + +impl Default for Endpoints { + fn default() -> Self { + Self { + lastfm: LASTFM_BASE_URL.to_owned(), + } + } +} + +/// The engine, or its absence. +/// +/// Cheap to clone; every clone drives the same engine. Disabled when no +/// provider is configured, and then every call returns at once having read and +/// sent nothing. +#[derive(Clone)] +pub struct Enrichment { + engine: Option>, +} + +struct Engine { + db: Database, + /// In the operator's order, which is the order fields are merged in. + sources: Vec, + ttl_ms: i64, + miss_ttl_ms: i64, + request_timeout: Duration, + /// The fills in flight, by provider and canonical name. A visit that finds + /// its key here schedules nothing: the fill already running answers it. + in_flight: DashMap<(Provider, String), ()>, + /// How many fills are running, for [`Enrichment::idle`]. + running: tokio::sync::watch::Sender, +} + +struct Source { + provider: Provider, + client: Client, + pace: Pace, +} + +enum Client { + LastFm { client: LastfmClient, key: String }, +} + +impl Enrichment { + /// Enrichment switched off. + pub fn disabled() -> Self { + Self { engine: None } + } + + /// The engine `settings` describe, against the providers' real addresses. + pub fn new(db: Database, settings: &MetadataSettings) -> anyhow::Result { + Self::with_endpoints(db, settings, Endpoints::default()) + } + + /// [`Self::new`], with the providers reached at `endpoints`. + /// + /// Only the configured providers are built. A key placed with no provider + /// naming it builds nothing — the provider list is the switch, and a key + /// is not consent (decision 3). + pub fn with_endpoints( + db: Database, + settings: &MetadataSettings, + endpoints: Endpoints, + ) -> anyhow::Result { + let mut sources = Vec::new(); + for &provider in &settings.providers { + let http = crate::outbound::client(settings.request_timeout) + .map_err(|_| anyhow::anyhow!("the outbound client could not be built"))?; + let client = match provider { + Provider::LastFm => { + let key = settings.lastfm_key.clone().ok_or_else(|| { + anyhow::anyhow!("metadata provider lastfm is configured without a key") + })?; + Client::LastFm { + client: LastfmClient::with_http(http, endpoints.lastfm.clone())?, + key, + } + } + }; + sources.push(Source { + provider, + client, + pace: Pace::new(settings.rate_per_sec), + }); + } + if sources.is_empty() { + return Ok(Self::disabled()); + } + Ok(Self { + engine: Some(Arc::new(Engine { + db, + sources, + ttl_ms: duration_ms(settings.ttl), + miss_ttl_ms: duration_ms(settings.miss_ttl), + request_timeout: settings.request_timeout, + in_flight: DashMap::new(), + running: tokio::sync::watch::channel(0).0, + })), + }) + } + + pub fn is_enabled(&self) -> bool { + self.engine.is_some() + } + + /// What the cache holds about the artist called `name`, one row per + /// configured provider that has one, in the operator's order — and a fill + /// scheduled for every configured provider whose row is missing or stale. + /// + /// Stale rows are returned, not withheld (decision 5): a slightly old + /// biography beats none, and serving it keeps a provider outage invisible. + /// Nothing here waits on a provider. + pub async fn artist(&self, name: &str) -> Result, sqlx::Error> { + let Some(engine) = &self.engine else { + return Ok(Vec::new()); + }; + let canonical = waveflow_core::scanner::canonical_name(name); + if canonical.is_empty() { + return Ok(Vec::new()); + } + let rows = engine.db.metadata_artist_rows(&canonical).await?; + let now = crate::authentication::now_ms(); + let mut ordered = Vec::new(); + for (index, source) in engine.sources.iter().enumerate() { + let row = rows.iter().find(|row| row.provider == source.provider); + if row.is_none_or(|row| row.expires_at <= now) { + Arc::clone(engine).schedule(index, name, &canonical); + } + ordered.extend(row.cloned()); + } + Ok(ordered) + } + + /// Waits until no fill is running. + /// + /// For the test suite, which has to know a fill is over before it can say + /// what the fill wrote — or that it wrote nothing. + #[doc(hidden)] + pub async fn idle(&self) { + let Some(engine) = &self.engine else { + return; + }; + let mut running = engine.running.subscribe(); + // The sender lives as long as the engine, which `self` holds. + let _ = running.wait_for(|count| *count == 0).await; + } +} + +impl Engine { + /// Starts a fill for one provider and one name, unless one is running. + fn schedule(self: Arc, index: usize, name: &str, canonical: &str) { + let key = (self.sources[index].provider, canonical.to_owned()); + if self.in_flight.insert(key.clone(), ()).is_some() { + return; + } + self.running.send_modify(|count| *count += 1); + // Released however the fill ends — a panic, or a task dropped before + // its first poll included, which is why it is built here rather than + // inside the task. A key left behind would stop that name from ever + // being asked again. + let flight = Flight { engine: self, key }; + let name = name.to_owned(); + tokio::spawn(async move { + flight.engine.fill(index, &name, &flight.key.1).await; + drop(flight); + }); + } + + async fn fill(&self, index: usize, name: &str, canonical: &str) { + let source = &self.sources[index]; + let answer = match &source.client { + Client::LastFm { client, key } => self.ask_lastfm(source, client, key, name).await, + }; + let answer = match answer { + Ok(answer) => answer, + Err(verdict) => { + tracing::warn!( + provider = source.provider.as_str(), + verdict = %verdict, + "artist lookup unreached; nothing cached" + ); + return; + } + }; + let fetched_at = crate::authentication::now_ms(); + let outcome = answer.outcome(); + let ttl = match outcome { + Outcome::Hit => self.ttl_ms, + Outcome::Miss => self.miss_ttl_ms, + }; + if let Err(error) = self + .db + .write_metadata_artist( + source.provider, + canonical, + &answer, + fetched_at, + fetched_at.saturating_add(ttl), + ) + .await + { + tracing::warn!( + provider = source.provider.as_str(), + error = %error, + "artist lookup answered but could not be cached" + ); + return; + } + tracing::debug!( + provider = source.provider.as_str(), + outcome = outcome.as_str(), + "artist lookup cached" + ); + } + + /// `artist.getInfo` and `artist.getSimilar`. Both must be reached for the + /// answer to count: half an answer written as a whole one would cache the + /// missing half as "nothing" for the TTL. + async fn ask_lastfm( + &self, + source: &Source, + client: &LastfmClient, + key: &str, + name: &str, + ) -> Result { + source.pace.wait().await; + let info = self.bounded(client.artist_get_info(name, key)).await?; + source.pace.wait().await; + let similar = self + .bounded(client.artist_get_similar(name, key, SIMILAR_LIMIT)) + .await?; + let own = waveflow_core::scanner::canonical_name(name); + let similar = similar + .into_iter() + .filter_map(|neighbour| { + let name_canonical = waveflow_core::scanner::canonical_name(&neighbour.name); + (!name_canonical.is_empty() && name_canonical != own).then(|| CachedSimilar { + name: neighbour.name, + name_canonical, + score: f64::from(neighbour.match_score), + }) + }) + .collect(); + let (bio_summary, bio_full) = info + .map(|info| (info.bio_summary, info.bio_full)) + .unwrap_or_default(); + Ok(ArtistAnswer { + bio_summary, + bio_full, + similar, + }) + } + + /// One request, under the request timeout; anything but an answer is a + /// verdict. + async fn bounded( + &self, + request: impl std::future::Future>, + ) -> Result { + match tokio::time::timeout(self.request_timeout, request).await { + Ok(Ok(answer)) => Ok(answer), + Ok(Err(error)) => Err(Verdict::of(&error)), + Err(_) => Err(Verdict::Timeout), + } + } +} + +/// Releases a fill's key and count when the fill ends. +struct Flight { + engine: Arc, + key: (Provider, String), +} + +impl Drop for Flight { + fn drop(&mut self) { + self.engine.in_flight.remove(&self.key); + self.engine.running.send_modify(|count| *count -= 1); + } +} + +/// Why a lookup was unreached, in words that carry nothing from the request or +/// the response — no URL, no body, no key. +#[derive(Debug)] +enum Verdict { + Timeout, + Transport, + Refused { code: i64 }, + Status { status: u16 }, + TooLarge, + Undecodable, +} + +impl Verdict { + fn of(error: &LookupError) -> Self { + match error { + LookupError::Http(error) if error.is_timeout() => Self::Timeout, + LookupError::Http(_) => Self::Transport, + LookupError::Refused { code } => Self::Refused { code: *code }, + LookupError::Status { status } => Self::Status { status: *status }, + LookupError::TooLarge { .. } => Self::TooLarge, + LookupError::Decode(_) => Self::Undecodable, + } + } +} + +impl std::fmt::Display for Verdict { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Timeout => formatter.write_str("timed out"), + Self::Transport => formatter.write_str("transport failure"), + Self::Refused { code } => write!(formatter, "refused with code {code}"), + Self::Status { status } => write!(formatter, "answered HTTP {status}"), + Self::TooLarge => formatter.write_str("answer over the size ceiling"), + Self::Undecodable => formatter.write_str("answer not understood"), + } + } +} + +/// Spaces one provider's requests to its rate: a token bucket that holds one +/// token, so there is never a burst to spend. +/// +/// The reservation is taken under the lock and the wait happens outside it, so +/// callers queue in order without one sleeping caller holding up the rest. +struct Pace { + interval: Duration, + next: std::sync::Mutex, +} + +impl Pace { + fn new(rate_per_sec: u32) -> Self { + Self { + interval: Duration::from_secs(1) / rate_per_sec.max(1), + next: std::sync::Mutex::new(Instant::now()), + } + } + + async fn wait(&self) { + let slot = { + let mut next = self + .next + .lock() + .unwrap_or_else(|poison| poison.into_inner()); + let slot = (*next).max(Instant::now()); + *next = slot + self.interval; + slot + }; + tokio::time::sleep_until(slot.into()).await; + } +} + +fn duration_ms(duration: Duration) -> i64 { + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX) +} diff --git a/src/lib.rs b/src/lib.rs index 896c856f..b853c549 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,9 +6,11 @@ pub mod catalog; pub mod cli; pub mod config; pub mod database; +pub mod enrichment; pub mod lyrics; pub mod media; pub mod oauth; +pub mod outbound; pub mod pid; pub mod playlist_file; pub mod scanner; @@ -76,6 +78,9 @@ pub struct AppState { /// The body ceiling the canvas route carries, derived from /// `WAVEFLOW_CANVAS_MAX_BYTES` so the two cannot disagree. pub canvas_body_limit: usize, + /// Metadata enrichment — RFC-011. Disabled unless the operator named a + /// provider. + pub enrichment: enrichment::Enrichment, } /// Roughly one offer's worth of JSON — a sixty-four character hash, a size, a @@ -596,8 +601,9 @@ pub async fn initialize(config: &Config) -> anyhow::Result { scanner.clone(), config, ); - // The only outbound surface this server has, and it is switched on by the - // operator naming a destination rather than by anybody linking an account. + // One of the two outbound surfaces this server has — metadata enrichment, + // below, is the other — and it is switched on by the operator naming a + // destination rather than by anybody linking an account. // Nothing leaves until both are true: an adapter with no authorisation // behind it is never handed a listen. RFC-010 decision 11. for destination in &config.destinations { @@ -673,6 +679,9 @@ pub async fn initialize(config: &Config) -> anyhow::Result { // the background task already runs leaves a window, short and sufficient, // in which listens leave for yesterday's destination. RFC-010 decision 10. services.reconcile_scrobble_links().await?; + // Off unless the operator named a provider, and then it still makes no + // request until somebody asks about an artist (RFC-011 decisions 3 and 6). + let enrichment = enrichment::Enrichment::new(db.clone(), &config.metadata)?; Ok(AppState { db, auth, @@ -691,6 +700,7 @@ pub async fn initialize(config: &Config) -> anyhow::Result { // Refused at startup if it does not fit, so this conversion cannot be // the place that decides what a ceiling falls back to. canvas_body_limit: usize::try_from(config.canvas.max_bytes).unwrap_or(usize::MAX), + enrichment, }) } diff --git a/src/outbound.rs b/src/outbound.rs new file mode 100644 index 00000000..2f9ac95a --- /dev/null +++ b/src/outbound.rs @@ -0,0 +1,47 @@ +//! The HTTP client every outbound request in this server is made with. +//! +//! Two subsystems call out: scrobbling (RFC-010), to destinations an operator +//! names, and metadata enrichment (RFC-011), to providers whose addresses are +//! constants. They share this client so the rules below are written once and +//! cannot drift apart — RFC-011 decision 9 adopts RFC-010 decision 10 rather +//! than restating it, and a second builder would be the place it quietly +//! stopped being true. + +use std::time::Duration; + +/// How much of the budget a connection may spend before the rest of the request +/// has had any of it. +/// +/// A third, so a destination that accepts slowly still leaves room for the +/// request itself rather than consuming the whole deadline in the handshake. +const CONNECT_SHARE: u32 = 3; + +/// The client. +/// +/// Each property below is RFC-010 decision 10 written down: +/// +/// - **No redirect is followed.** `reqwest` follows up to ten by default, and +/// the tenth can be anywhere. A destination that answers `302` to an internal +/// address would otherwise have this server fetch it and hand back what it +/// found, which is the shape of every request-forgery there is. +/// - **No proxy is read from the environment.** An operator's `HTTP_PROXY`, +/// set for something else entirely, must not silently become the route every +/// request takes. +/// - **Only the connection is bounded here.** The request as a whole is bounded +/// once, by the caller's own `tokio::time::timeout`, and deliberately not a +/// second time by this client. +/// +/// That last one was two equal deadlines racing until a review noticed. Both +/// ended at `Ambiguous`, so the scrobbling queue looked the same either way — +/// but only the drain's own deadline records the link as stalled, and that +/// record is what stops one silent destination from spending an entry per row. +/// Which of the two fired was a coin toss, so which protection applied was a +/// coin toss. +pub fn client(timeout: Duration) -> Result { + reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(timeout / CONNECT_SHARE) + .no_proxy() + .user_agent(concat!("WaveFlowServer/", env!("CARGO_PKG_VERSION"))) + .build() +} diff --git a/src/scrobblers/mod.rs b/src/scrobblers/mod.rs index 41045e35..fe70ab5c 100644 --- a/src/scrobblers/mod.rs +++ b/src/scrobblers/mod.rs @@ -5,10 +5,11 @@ //! That line is what decision 6 exists to draw, and putting an adapter beside //! the drain would rub it out within a release. //! -//! **This is the only place in the server that makes an outbound request.** -//! Decision 10 calls that the real risk of the whole RFC — a server that calls -//! a URL is a server somebody can make call a URL — so the rules live here, -//! once, rather than in each adapter. +//! Decision 10 calls outbound requests the real risk of the whole RFC — a +//! server that calls a URL is a server somebody can make call a URL — so the +//! rules live once, in [`crate::outbound`], rather than in each adapter. That +//! module is shared with metadata enrichment (RFC-011), the server's only other +//! outbound surface. use std::time::Duration; @@ -118,41 +119,10 @@ pub fn destination_fingerprint(url: &url::Url) -> String { )) } -/// How much of the budget a connection may spend before the rest of the request -/// has had any of it. -/// -/// A third, so a destination that accepts slowly still leaves room for the -/// submission itself rather than consuming the whole deadline in the handshake. -const CONNECT_SHARE: u32 = 3; - -/// The client every adapter shares. -/// -/// Each property below is decision 10 written down: -/// -/// - **No redirect is followed.** `reqwest` follows up to ten by default, and -/// the tenth can be anywhere. A destination that answers `302` to an internal -/// address would otherwise have this server fetch it and hand back what it -/// found, which is the shape of every request-forgery there is. -/// - **No proxy is read from the environment.** An operator's `HTTP_PROXY`, -/// set for something else entirely, must not silently become the route every -/// listen takes. -/// - **Only the connection is bounded here.** The request as a whole is bounded -/// once, by the drain's own `tokio::time::timeout`, and deliberately not a -/// second time by this client. -/// -/// That last one was two equal deadlines racing until a review noticed. Both -/// ended at `Ambiguous`, so the queue looked the same either way — but only the -/// drain's own deadline records the link as stalled, and that record is what -/// stops one silent destination from spending an entry per row. Which of the two -/// fired was a coin toss, so which protection applied was a coin toss. +/// The client every adapter shares: [`crate::outbound::client`], which holds +/// decision 10's rules and is shared with metadata enrichment. pub fn outbound_client(timeout: Duration) -> Result { - reqwest::Client::builder() - .redirect(reqwest::redirect::Policy::none()) - .connect_timeout(timeout / CONNECT_SHARE) - .no_proxy() - .user_agent(concat!("WaveFlowServer/", env!("CARGO_PKG_VERSION"))) - .build() - .map_err(|_| OutboundError::Unbuildable) + crate::outbound::client(timeout).map_err(|_| OutboundError::Unbuildable) } /// Checks a destination an operator configured, before it is ever called. diff --git a/tests/enrichment.rs b/tests/enrichment.rs new file mode 100644 index 00000000..1c682eb7 --- /dev/null +++ b/tests/enrichment.rs @@ -0,0 +1,435 @@ +//! Metadata enrichment's cache and engine — RFC-011 decisions 3, 5, 6 and 9. +//! +//! Every test points the engine at a fake Last.fm on a loopback listener of its +//! own; the suite never reaches a real provider. The fake counts what it is +//! asked, which is what most of these assertions are about: enrichment's +//! failures are requests made that should not have been, or answers cached that +//! were never given. + +// Not every target uses every fixture, and a shared module is not dead +// code for being partly unused here. +#[allow(dead_code)] +mod support; + +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Duration, +}; + +use axum::{ + extract::{Query, State}, + http::{header, StatusCode}, + response::{IntoResponse, Response}, + routing::get, + Router, +}; +use waveflow_server::{ + config::MetadataSettings, + enrichment::{Endpoints, Enrichment, Outcome, Provider}, +}; + +#[derive(Clone, Copy)] +enum Mode { + /// A biography and two neighbours. + Hit, + /// Last.fm's code 6 to both methods: it knows nothing of this artist. + Miss, + /// A failure status with a body that is not Last.fm's. + Unavailable, + /// A `302` to `/followed`, which answers like `Hit`. + Redirect, + /// A well-formed answer three mebibytes long. + Oversized, +} + +#[derive(Clone)] +struct Fake { + mode: Mode, + delay: Duration, + info: Arc, + similar: Arc, + followed: Arc, +} + +struct Listener { + fake: Fake, + base_url: String, +} + +async fn fake_lastfm(mode: Mode, delay: Duration) -> Listener { + let fake = Fake { + mode, + delay, + info: Arc::default(), + similar: Arc::default(), + followed: Arc::default(), + }; + let router = Router::new() + .route("/2.0/", get(answer)) + .route("/followed", get(followed)) + .with_state(fake.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + Listener { + fake, + base_url: format!("http://{address}/2.0/"), + } +} + +async fn answer( + State(fake): State, + Query(query): Query>, +) -> Response { + let method = query.get("method").map(String::as_str).unwrap_or_default(); + match method { + "artist.getinfo" => fake.info.fetch_add(1, Ordering::SeqCst), + "artist.getsimilar" => fake.similar.fetch_add(1, Ordering::SeqCst), + _ => return StatusCode::BAD_REQUEST.into_response(), + }; + tokio::time::sleep(fake.delay).await; + match fake.mode { + Mode::Hit => found(method), + Mode::Miss => { + json(r#"{"error":6,"message":"The artist you supplied could not be found"}"#.to_owned()) + } + Mode::Unavailable => (StatusCode::SERVICE_UNAVAILABLE, "down").into_response(), + Mode::Redirect => (StatusCode::FOUND, [(header::LOCATION, "/followed")]).into_response(), + Mode::Oversized => json(format!( + r#"{{"artist":{{"name":"Björk","bio":{{"summary":"{}"}}}}}}"#, + "a".repeat(3 * 1024 * 1024) + )), + } +} + +async fn followed(State(fake): State) -> Response { + fake.followed.fetch_add(1, Ordering::SeqCst); + found("artist.getinfo") +} + +fn found(method: &str) -> Response { + json(if method == "artist.getinfo" { + r#"{"artist":{"name":"Björk","bio":{"summary":"Icelandic singer.","content":"Icelandic singer, at length."}}}"#.to_owned() + } else { + r#"{"similarartists":{"artist":[ + {"name":"Sigur Rós","match":"0.9","image":[]}, + {"name":"Múm","match":"0.75","image":[]} + ]}}"# + .to_owned() + }) +} + +fn json(body: String) -> Response { + ([(header::CONTENT_TYPE, "application/json")], body).into_response() +} + +/// A key made up for the run, so no credential sits in the source and a log +/// search for it cannot match anything but this run's own output. +fn run_key() -> String { + uuid::Uuid::new_v4().simple().to_string() +} + +fn settings(providers: Vec, key: &str) -> MetadataSettings { + MetadataSettings { + providers, + lastfm_key: Some(key.to_owned()), + // Fast enough that pacing never decides a test. + rate_per_sec: 1_000, + request_timeout: Duration::from_secs(5), + ..MetadataSettings::off() + } +} + +fn engine( + state: &waveflow_server::AppState, + settings: &MetadataSettings, + listener: &Listener, +) -> Enrichment { + Enrichment::with_endpoints( + state.db.clone(), + settings, + Endpoints { + lastfm: listener.base_url.clone(), + }, + ) + .unwrap() +} + +fn canonical(name: &str) -> String { + waveflow_core::scanner::canonical_name(name) +} + +/// Decision 3: the provider list is the switch. A key placed without a +/// provider naming it — the shape an operator who set up scrobbling first +/// leaves behind — is not consent, and asks nothing. +#[tokio::test] +async fn an_empty_provider_list_asks_nothing_even_with_a_key() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Hit, Duration::ZERO).await; + let enrichment = engine(&state, &settings(Vec::new(), &run_key()), &listener); + + assert!(!enrichment.is_enabled()); + assert!(enrichment.artist("Björk").await.unwrap().is_empty()); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 0); + assert_eq!(listener.fake.similar.load(Ordering::SeqCst), 0); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// The ordinary path, and decision 6's cost stated as a test: the first visit +/// answers from an empty cache, the second has what came back. +#[tokio::test] +async fn a_hit_is_cached_with_its_biography_and_ranked_neighbours() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Hit, Duration::ZERO).await; + let settings = settings(vec![Provider::LastFm], &run_key()); + let enrichment = engine(&state, &settings, &listener); + + assert!(enrichment.artist("Björk").await.unwrap().is_empty()); + enrichment.idle().await; + let rows = enrichment.artist("Björk").await.unwrap(); + + assert_eq!(rows.len(), 1); + let row = &rows[0]; + assert_eq!(row.provider, Provider::LastFm); + assert_eq!(row.outcome, Outcome::Hit); + assert_eq!(row.bio_summary.as_deref(), Some("Icelandic singer.")); + assert_eq!( + row.bio_full.as_deref(), + Some("Icelandic singer, at length.") + ); + assert_eq!( + row.expires_at - row.fetched_at, + settings.ttl.as_millis() as i64 + ); + let similar = state + .db + .metadata_artist_similar(Provider::LastFm, &canonical("Björk")) + .await + .unwrap(); + let names: Vec<_> = similar.iter().map(|s| s.name.as_str()).collect(); + assert_eq!(names, ["Sigur Rós", "Múm"]); + assert_eq!(similar[0].name_canonical, canonical("Sigur Rós")); + // A fresh row is served, not asked again. + enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); +} + +/// Decision 5: "the provider knows nothing" is an answer, cached — for the +/// short window, because it is the answer most likely to change. +#[tokio::test] +async fn a_provider_that_knows_nothing_is_cached_as_a_miss_for_the_miss_ttl() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Miss, Duration::ZERO).await; + let settings = settings(vec![Provider::LastFm], &run_key()); + let enrichment = engine(&state, &settings, &listener); + + enrichment.artist("Nobody Knows").await.unwrap(); + enrichment.idle().await; + + let rows = state + .db + .metadata_artist_rows(&canonical("Nobody Knows")) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].outcome, Outcome::Miss); + assert_eq!(rows[0].bio_summary, None); + assert_eq!( + rows[0].expires_at - rows[0].fetched_at, + settings.miss_ttl.as_millis() as i64, + "a miss stands for the miss TTL, not the hit TTL" + ); +} + +/// Decision 5's third outcome: a provider that could not be asked writes no +/// row, so the next visit asks again instead of reading the outage as an +/// artist without a biography. +#[tokio::test] +async fn a_provider_that_cannot_be_reached_writes_no_row() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Unavailable, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 2); +} + +/// Decision 6: ten clients opening the same page are one request. +#[tokio::test] +async fn ten_concurrent_first_visits_make_one_request() { + let (_temp, _config, state) = support::test_app().await; + // Slow enough that every visit lands while the first fill is still out. + let listener = fake_lastfm(Mode::Hit, Duration::from_millis(300)).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + let visits = (0..10).map(|_| enrichment.artist("Björk")); + for answer in futures_util::future::join_all(visits).await { + assert!(answer.unwrap().is_empty()); + } + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert_eq!(listener.fake.similar.load(Ordering::SeqCst), 1); +} + +/// Decision 9: no redirect is followed. Followed, this one would have been +/// cached as a hit from wherever the provider pointed. +#[tokio::test] +async fn a_redirect_from_the_provider_is_not_followed() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Redirect, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert_eq!(listener.fake.followed.load(Ordering::SeqCst), 0); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// Decision 9: a response is bounded. Read whole, this one would have been +/// cached as a three-mebibyte biography. +#[tokio::test] +async fn an_answer_over_the_ceiling_is_refused() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Oversized, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// A writer for the log, kept so the test can read it back. +#[derive(Clone, Default)] +struct Captured(Arc>>); + +impl std::io::Write for Captured { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Captured { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// The log of this whole test binary, at every level. +/// +/// Global rather than scoped to one test: a subscriber installed with +/// `set_default` covers one thread, and tracing caches each callsite's interest +/// for the whole process — with the other tests running beside it, a scoped +/// capture kept two lines out of hundreds. Every test here writes into this +/// one; the assertion below looks for its own run's key, which no other test +/// knows. +fn captured_log() -> &'static Captured { + static LOG: std::sync::OnceLock = std::sync::OnceLock::new(); + LOG.get_or_init(|| { + let captured = Captured::default(); + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_writer(captured.clone()) + .with_ansi(false) + .init(); + captured + }) +} + +/// Decision 9 and `CLAUDE.md`: the key travels in Last.fm's query string, and +/// must reach no log line — not on success, not on a refusal, not on a +/// connection that never opened — at any level, the HTTP stack's own included. +#[tokio::test] +async fn the_key_never_reaches_the_log() { + let captured = captured_log(); + + let (_temp, _config, state) = support::test_app().await; + let key = run_key(); + let settings = settings(vec![Provider::LastFm], &key); + for mode in [Mode::Hit, Mode::Unavailable] { + let listener = fake_lastfm(mode, Duration::ZERO).await; + let enrichment = engine(&state, &settings, &listener); + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + } + // Nothing listens on port 1 of loopback: the connection itself fails. + let closed = Enrichment::with_endpoints( + state.db.clone(), + &settings, + Endpoints { + lastfm: "http://127.0.0.1:1/2.0/".to_owned(), + }, + ) + .unwrap(); + closed.artist("Sigur Rós").await.unwrap(); + closed.idle().await; + + let log = String::from_utf8(captured.0.lock().unwrap().clone()).unwrap(); + // The capture works: both failures were logged, by provider and verdict. + // A closed port is refused on Linux and left to the connect timeout on + // Windows, so the second verdict is one of two. + assert!(log.contains("answered HTTP 503"), "{log}"); + assert!( + log.contains("verdict=transport failure") || log.contains("verdict=timed out"), + "{log}" + ); + assert!(!log.contains(&key), "the key reached the log:\n{log}"); +} From e4e3d945fcd0b470cdbb387d6ff3af160f8e77ca Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Thu, 15 Oct 2026 20:56:33 +0200 Subject: [PATCH 2/3] fix(enrichment): cap the fills in flight, and log a failure of the test's own Pacing refuses no request, so nothing bounded how many fills could queue: a page opening two thousand artists, or an outage that writes no row, would grow the queue for as long as visits continued. Beyond `MAX_IN_FLIGHT` (64) nothing is scheduled, and the next visit asks again. `the_key_never_reaches_the_log` asked about one artist twice, so the second engine found a fresh row and sent nothing; the 503 it checked for was only there when another test in the binary had logged one. Each mode now asks about its own artist, and the test passes run alone. Signed-off-by: InstaZDLL --- src/enrichment.rs | 16 +++++++++++++++- tests/enrichment.rs | 36 ++++++++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/src/enrichment.rs b/src/enrichment.rs index 19409487..d442e374 100644 --- a/src/enrichment.rs +++ b/src/enrichment.rs @@ -39,6 +39,16 @@ const LASTFM_BASE_URL: &str = "https://ws.audioscrobbler.com/2.0/"; /// see, which decision 7 drops at read time. const SIMILAR_LIMIT: u32 = 50; +/// How many fills may be waiting or running at once, across providers. +/// +/// Pacing reserves a slot for every request and refuses none, so without this +/// a page that opens two thousand artists would queue two thousand fills — +/// a quarter of an hour at the default rate — and an outage, which writes no +/// row, would let every further visit add more. Beyond the ceiling nothing is +/// scheduled; the visit after asks again, which is what an unreached lookup +/// does anyway. +pub const MAX_IN_FLIGHT: usize = 64; + /// A provider an operator can name in `WAVEFLOW_METADATA_PROVIDERS`. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Provider { @@ -280,9 +290,13 @@ impl Enrichment { } impl Engine { - /// Starts a fill for one provider and one name, unless one is running. + /// Starts a fill for one provider and one name, unless one is running or + /// [`MAX_IN_FLIGHT`] are. fn schedule(self: Arc, index: usize, name: &str, canonical: &str) { let key = (self.sources[index].provider, canonical.to_owned()); + if self.in_flight.len() >= MAX_IN_FLIGHT { + return; + } if self.in_flight.insert(key.clone(), ()).is_some() { return; } diff --git a/tests/enrichment.rs b/tests/enrichment.rs index 1c682eb7..0988764c 100644 --- a/tests/enrichment.rs +++ b/tests/enrichment.rs @@ -301,6 +301,34 @@ async fn ten_concurrent_first_visits_make_one_request() { assert_eq!(listener.fake.similar.load(Ordering::SeqCst), 1); } +/// Pacing refuses no request, so the ceiling on fills in flight is what bounds +/// the queue: one artist past it is not asked, and its next visit will be. +#[tokio::test] +async fn no_more_than_the_ceiling_of_fills_is_ever_queued() { + let (_temp, _config, state) = support::test_app().await; + // Slow enough that every fill is still out when the last visit lands. + let listener = fake_lastfm(Mode::Hit, Duration::from_secs(1)).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + let names: Vec = (0..=waveflow_server::enrichment::MAX_IN_FLIGHT) + .map(|index| format!("Artist {index}")) + .collect(); + let visits = names.iter().map(|name| enrichment.artist(name)); + for answer in futures_util::future::join_all(visits).await { + answer.unwrap(); + } + enrichment.idle().await; + + assert_eq!( + listener.fake.info.load(Ordering::SeqCst), + waveflow_server::enrichment::MAX_IN_FLIGHT + ); +} + /// Decision 9: no redirect is followed. Followed, this one would have been /// cached as a hit from wherever the provider pointed. #[tokio::test] @@ -404,11 +432,15 @@ async fn the_key_never_reaches_the_log() { let (_temp, _config, state) = support::test_app().await; let key = run_key(); let settings = settings(vec![Provider::LastFm], &key); - for mode in [Mode::Hit, Mode::Unavailable] { + // One artist per mode: asked about the same one, the second engine would + // find the first one's fresh row and send nothing — and the 503 checked + // below would then only be found if another test happened to log one. + for (mode, artist) in [(Mode::Hit, "Björk"), (Mode::Unavailable, "Múm")] { let listener = fake_lastfm(mode, Duration::ZERO).await; let enrichment = engine(&state, &settings, &listener); - enrichment.artist("Björk").await.unwrap(); + enrichment.artist(artist).await.unwrap(); enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1, "{artist}"); } // Nothing listens on port 1 of loopback: the connection itself fails. let closed = Enrichment::with_endpoints( From b68d908b0ae44254ea5683ff6d4bb0b83d1f7f15 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Thu, 15 Oct 2026 21:08:49 +0200 Subject: [PATCH 3/3] fix(enrichment): reserve a place under the ceiling in one step Reading the map's length and then inserting let two visits on two threads both see one place left and both take it. The place is now reserved on the fills counter under its own lock, before the key is claimed, and given back when the key turns out to be in flight already. Signed-off-by: InstaZDLL --- src/enrichment.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/src/enrichment.rs b/src/enrichment.rs index d442e374..7f2b99eb 100644 --- a/src/enrichment.rs +++ b/src/enrichment.rs @@ -170,7 +170,8 @@ struct Engine { /// The fills in flight, by provider and canonical name. A visit that finds /// its key here schedules nothing: the fill already running answers it. in_flight: DashMap<(Provider, String), ()>, - /// How many fills are running, for [`Enrichment::idle`]. + /// How many fills are waiting or running: the count [`MAX_IN_FLIGHT`] + /// bounds, and the one [`Enrichment::idle`] waits on. running: tokio::sync::watch::Sender, } @@ -294,13 +295,23 @@ impl Engine { /// [`MAX_IN_FLIGHT`] are. fn schedule(self: Arc, index: usize, name: &str, canonical: &str) { let key = (self.sources[index].provider, canonical.to_owned()); - if self.in_flight.len() >= MAX_IN_FLIGHT { + // A place is reserved before the key is claimed, in one step under the + // counter's lock: a length read then an insert would let two visits on + // two threads both see one place left and both take it. + let reserved = self.running.send_if_modified(|count| { + let free = *count < MAX_IN_FLIGHT; + if free { + *count += 1; + } + free + }); + if !reserved { return; } if self.in_flight.insert(key.clone(), ()).is_some() { + self.running.send_modify(|count| *count -= 1); return; } - self.running.send_modify(|count| *count += 1); // Released however the fill ends — a panic, or a task dropped before // its first poll included, which is why it is built here rather than // inside the task. A key left behind would stop that name from ever