diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index 40d5b2b..22af20c 100644 --- a/docs/rfcs/RFC-007-library-event-stream.md +++ b/docs/rfcs/RFC-007-library-event-stream.md @@ -3,7 +3,10 @@ - **Statut** : Proposed - **Implémentée par** : [#145](https://github.com/InstaZDLL/waveflow-server/pull/145) (le flux et son filigrane), [#152](https://github.com/InstaZDLL/waveflow-server/pull/152) - (l'appareil d'origine). Le champ *Statut* ci-dessus ne bascule jamais dans ce + (l'appareil d'origine), [#159](https://github.com/InstaZDLL/waveflow-server/pull/159) + (les événements d'album, décision 6), + [#166](https://github.com/InstaZDLL/waveflow-server/pull/166) (la rétention, + décision 7). Le champ *Statut* ci-dessus ne bascule jamais dans ce projet : c'est cette ligne qui dit ce qui tourne, et elle se vérifie — une PR se lit, un mot de statut ne s'audite pas. - **Date** : 2026-08-25 @@ -235,9 +238,7 @@ chose l'écrira, et ce jour-là ce sera cette ligne qu'il faudra relire. ## Décision 7 — deux bornes, parce qu'une seule se trompe toujours -> **Décidée le 2026-08-30, pas encore construite.** Rien ne purge aujourd'hui ; -> `library.events_purged_through` existe et vaut 0. Cette décision dit ce que la -> purge fera, et la ligne *Implémentée par* de cet en-tête dira quand. +> **Construite.** Voir la ligne *Implémentée par* de l'en-tête. **Un âge et un plancher, tenus ensemble.** Ni l'un ni l'autre ne suffit, et les deux échouent dans des directions opposées. @@ -324,6 +325,6 @@ Plus rien de cette RFC n'est ouvert. Les trois questions qu'elle portait ont pour les deux. - ~~La forme exacte de l'acquittement.~~ Décision 8. -Les décisions 7 et 8 sont **décidées et pas encore construites**. La ligne -*Implémentée par* de l'en-tête ne les nommera que lorsqu'elles le seront ; c'est -elle qui dit ce qui tourne, pas cette section. +La décision 7 est construite ; la **décision 8 est décidée et pas encore +construite**. La ligne *Implémentée par* de l'en-tête ne nomme que ce qui +tourne, et c'est elle qu'il faut lire — pas cette section. diff --git a/src/config.rs b/src/config.rs index e2f9612..e7cead9 100644 --- a/src/config.rs +++ b/src/config.rs @@ -36,6 +36,21 @@ pub struct UploadLimits { pub session_ttl: Duration, } +/// How long a library's change feed keeps what it has written. +/// +/// The floor wins over the age: a library below it keeps everything, however +/// old. RFC-007 decision 7. +#[derive(Debug, Clone, Copy)] +pub struct LibraryEventRetention { + /// Whole days. Only what is **strictly** older is cut, so an event exactly + /// this old survives — the same exclusive bound `stream_ticket::verify` + /// uses, and the cautious direction: keeping one event too many breaks + /// nobody, cutting one too many sends somebody back to the snapshot. + pub days: u32, + /// The fewest events a library keeps whatever their age. + pub min_events: i64, +} + /// What the server accepts when a member attaches a loop to a track. /// /// Apart from [`UploadLimits`] rather than folded into it: the two magazines @@ -104,6 +119,16 @@ pub struct Config { /// the matching MIME map, so a video in that directory would oblige the two /// lists to agree forever. pub canvas_dir: PathBuf, + /// How long a library's change feed keeps what it has written. + /// + /// `WAVEFLOW_LIBRARY_EVENT_RETENTION_DAYS`, + /// `WAVEFLOW_LIBRARY_EVENT_RETENTION_MIN`. + /// + /// Two bounds because either alone fails in the opposite direction: an age + /// alone lets a library that rescans daily grow without limit, and a count + /// alone cuts the head off a quiet one whose ten thousand events cover two + /// years. RFC-007 decision 7. + pub library_event_retention: LibraryEventRetention, /// What the server accepts when a member attaches a loop to a track. /// /// `WAVEFLOW_CANVAS_MAX_BYTES`, `WAVEFLOW_CANVAS_MAX_DURATION_SECS`, @@ -152,6 +177,7 @@ impl std::fmt::Debug for Config { .field("transcode_global_limit", &self.transcode_global_limit) .field("transcode_per_user_limit", &self.transcode_per_user_limit) .field("uploads", &self.uploads) + .field("library_event_retention", &self.library_event_retention) .field("canvas_dir", &self.canvas_dir) .field("canvas", &self.canvas) .field("allowed_origins", &self.allowed_origins) @@ -236,6 +262,14 @@ impl Config { )?, }; validate_canvas(&canvas)?; + // Both refuse zero and negatives at startup rather than falling back: + // every fallback for a bound is wrong, and the operator is turned away + // where they can see why. No ceiling — an enormous value means "purge + // nothing", which is safe and legible. + let library_event_retention = LibraryEventRetention { + days: parse_positive_env("WAVEFLOW_LIBRARY_EVENT_RETENTION_DAYS", 30u32)?, + min_events: parse_positive_env("WAVEFLOW_LIBRARY_EVENT_RETENTION_MIN", 10_000i64)?, + }; if transcode_per_user_limit > transcode_global_limit { anyhow::bail!( "WAVEFLOW_TRANSCODE_PER_USER_LIMIT cannot exceed WAVEFLOW_TRANSCODE_GLOBAL_LIMIT" @@ -305,6 +339,7 @@ impl Config { transcode_global_limit, transcode_per_user_limit, uploads, + library_event_retention, canvas_dir, canvas, allowed_origins, @@ -347,6 +382,13 @@ impl Config { sessions_per_user: 2, session_ttl: Duration::from_secs(3600), }, + // Small enough that a test can write past the floor without + // writing ten thousand rows, and shaped like production rather + // than unlimited. + library_event_retention: LibraryEventRetention { + days: 30, + min_events: 4, + }, canvas_dir: canvas_dir_for_tests, // Same reasoning as the upload limits above: small enough that a // test can reach every bound, shaped like production rather than diff --git a/src/main.rs b/src/main.rs index c3b8d4b..c749b18 100644 --- a/src/main.rs +++ b/src/main.rs @@ -36,6 +36,7 @@ async fn serve(config: Config, state: waveflow_server::AppState) -> anyhow::Resu // reclaim it until somebody offered another file. state.services.spawn_upload_sweeper(); state.services.spawn_canvas_sweeper(); + state.services.spawn_library_event_purge(); state.db.spawn_authorization_pruning(); let router = waveflow_server::app(&config, state); let listener = tokio::net::TcpListener::bind(bind_addr) diff --git a/src/services/library_events.rs b/src/services/library_events.rs index bc029f4..8eba965 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -6,7 +6,159 @@ use super::*; use crate::sync::MAX_SYNC_LIMIT; +/// How often the feed is trimmed. +/// +/// Retention is measured in days, so a pass a day is the coarsest interval that +/// still honours it: the oldest surviving event is never more than a day past +/// the bound. The other sweepers in this crate have the same shape. +const PURGE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60); + +/// What one pass of the retention purge removed. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct EventPurge { + /// Rows cut, across every library. + pub events_removed: u64, + /// Libraries that lost at least one. + pub libraries_trimmed: usize, +} + impl DomainServices { + /// Trims every library's feed to what RFC-007 decision 7 says it keeps. + /// + /// Boot first, then a pass a day. The other sweepers in this crate have the + /// same shape, and for the same reason: a server that has been down should + /// not wait a full interval to catch up on what it owes. + pub fn spawn_library_event_purge(&self) { + let services = self.clone(); + tokio::spawn(async move { + services.purge_now().await; + let mut ticker = tokio::time::interval(PURGE_INTERVAL); + ticker.tick().await; + loop { + ticker.tick().await; + services.purge_now().await; + } + }); + } + + async fn purge_now(&self) { + match self.purge_library_events(now_ms()).await { + Ok(purged) if purged == EventPurge::default() => {} + Ok(purged) => tracing::info!( + events = purged.events_removed, + libraries = purged.libraries_trimmed, + "library event feeds trimmed" + ), + Err(error) => tracing::warn!(%error, "could not trim the library event feeds"), + } + } + + /// One pass. Public so a test can run it rather than wait a day for it. + /// + /// Two bounds, and the floor wins. An age alone lets a library that rescans + /// daily grow without limit; a count alone cuts the head off a quiet one + /// whose ten thousand events cover two years. + /// + /// `now_ms` is a parameter for the same reason `sweep_expired_sessions` + /// takes one: the bound is exclusive, so a caller that cannot name the + /// instant cannot place an event *on* it — it can only put one a few + /// milliseconds either side and hope. A test written that way passes or + /// fails on how long the lines between it and here took to run. + pub async fn purge_library_events(&self, now_ms: i64) -> Result { + let retention = self.library_event_retention; + // Whole days in milliseconds, and the multiplication is checked: a + // configured value large enough to overflow means "keep everything", + // which is what a cutoff of zero produces anyway. + let cutoff = i64::from(retention.days) + .checked_mul(24 * 60 * 60 * 1000) + .and_then(|window| now_ms.checked_sub(window)); + let Some(cutoff) = cutoff else { + return Ok(EventPurge::default()); + }; + + let libraries: Vec = sqlx::query_scalar("SELECT id FROM library") + .fetch_all(self.db.pool()) + .await?; + let mut purged = EventPurge::default(); + for library_id in libraries { + // One transaction per library rather than one for all of them: the + // writer gate is process-wide, and holding it across every feed on + // a server with fifty libraries would stall every other mutation + // for the whole pass. + let removed = self.purge_one_library(&library_id, cutoff).await?; + if removed > 0 { + purged.events_removed += removed; + purged.libraries_trimmed += 1; + } + } + Ok(purged) + } + + /// Cuts one library's feed and moves its watermark with it. + /// + /// The delete and the watermark are one transaction, and that is the whole + /// of what makes the expiry answer honest. Written separately, there is a + /// window where the watermark claims less than has gone — and a client + /// reading into it is handed a catch-up that looks complete while silently + /// skipping the gap, which is the failure decision 4 exists to refuse. + async fn purge_one_library(&self, library_id: &str, cutoff: i64) -> Result { + let _writer = self.db.writer_guard().await; + let mut tx = self.db.pool().begin().await?; + + // What is eligible, measured before it is gone. `changes()` after the + // delete would give the count and not the highest cursor, and reading + // the maximum back afterwards would read a table the delete has already + // emptied of exactly the rows in question. + // + // The floor is expressed as the newest cursor that may be cut: skip the + // `min_events` newest rows and take the one after them. With fewer rows + // than the floor this subquery is NULL, `cursor <= NULL` is never true, + // and the library keeps everything however old — which is the rule. + let eligible = sqlx::query( + "SELECT COUNT(*) AS n, MAX(cursor) AS highest FROM library_event \ + WHERE library_id=? AND changed_at < ? AND cursor <= ( \ + SELECT cursor FROM library_event WHERE library_id=? \ + ORDER BY cursor DESC LIMIT 1 OFFSET ?)", + ) + .bind(library_id) + .bind(cutoff) + .bind(library_id) + .bind(self.library_event_retention.min_events) + .fetch_one(&mut *tx) + .await?; + let removed: i64 = eligible.try_get("n")?; + let highest: Option = eligible.try_get("highest")?; + let (Some(highest), true) = (highest, removed > 0) else { + return Ok(0); + }; + + sqlx::query( + "DELETE FROM library_event \ + WHERE library_id=? AND changed_at < ? AND cursor <= ( \ + SELECT cursor FROM library_event WHERE library_id=? \ + ORDER BY cursor DESC LIMIT 1 OFFSET ?)", + ) + .bind(library_id) + .bind(cutoff) + .bind(library_id) + .bind(self.library_event_retention.min_events) + .execute(&mut *tx) + .await?; + + // `MAX` never decreases: a pass that cut an older tail must not lower a + // watermark an earlier one raised, and two passes racing must not + // either. + sqlx::query( + "UPDATE library SET events_purged_through=MAX(events_purged_through, ?) WHERE id=?", + ) + .bind(highest) + .bind(library_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(u64::try_from(removed).unwrap_or(0)) + } + /// One page of a library's changes, for a caller entitled to that library. /// /// A caller who is not a member gets `NotFound`, not `Forbidden`: a feed diff --git a/src/services/mod.rs b/src/services/mod.rs index 4c1c8c7..f8ea146 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -822,6 +822,7 @@ pub struct DomainServices { sync: SyncService, scanner: crate::scanner::ScanManager, uploads: crate::config::UploadLimits, + library_event_retention: crate::config::LibraryEventRetention, /// One lock per open upload session. /// /// A fragment is a file write and a row update that have to agree, and the @@ -914,6 +915,7 @@ impl DomainServices { sync, scanner, uploads: config.uploads, + library_event_retention: config.library_event_retention, upload_locks: Arc::new(dashmap::DashMap::new()), canvas: config.canvas, canvas_dir: config.canvas_dir.clone(), diff --git a/tests/catalog.rs b/tests/catalog.rs index 2e31b40..9a0caeb 100644 --- a/tests/catalog.rs +++ b/tests/catalog.rs @@ -1758,3 +1758,164 @@ async fn an_album_that_is_merely_retagged_announces_itself_once() { assert_eq!(announced.len(), 1, "the retag is announced: {announced:?}"); assert_eq!(announced[0].action, "upsert"); } + +/// RFC-007 decision 7: an age and a floor, the floor winning, and the age bound +/// exclusive. Nothing purged before this, so the feed's expiry answer was +/// correct code that had never once run. +#[tokio::test] +async fn retention_cuts_by_age_never_below_the_floor_and_never_at_the_bound() { + let temp = tempfile::tempdir().unwrap(); + let mut config = waveflow_server::Config::for_data_dir(temp.path().join("data")); + // Tuned before `initialize`: `DomainServices` copies these out of `Config` + // when it is built, so raising a bound on the returned one changes nothing. + config.library_event_retention.min_events = 2; + config.library_event_retention.days = 30; + let state = waveflow_server::initialize(&config).await.unwrap(); + + let hash = security::hash_password("correct horse battery staple").unwrap(); + let owner = state + .db + .create_account("retention", &hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let music = config.data_dir.join("retention-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Retention", + &std::fs::canonicalize(&music).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + // Five events through the ordinary path. No album on the inputs, so each + // track writes one event and nothing else — an album upsert emits its own + // since #159, and a count this test reasons about must not include them. + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 5, false).await.unwrap(); + for index in 0..5usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let cursors: Vec = + sqlx::query_scalar("SELECT cursor FROM library_event WHERE library_id=? ORDER BY cursor") + .bind(library.to_string()) + .fetch_all(state.db.pool()) + .await + .unwrap(); + assert_eq!(cursors.len(), 5, "one event per track and no album events"); + + // Backdated in SQL because the ages are what this test is about: two far + // past the bound, one exactly on it, two inside. + // + // `now` is handed to the purge rather than read by it. The bound is + // exclusive, so "exactly thirty days old" is a single instant: read the + // clock twice and the boundary event lands a millisecond on the wrong side, + // and this test passes or fails on how long the lines above it took. + let day = 24 * 60 * 60 * 1000i64; + let now = now_ms(); + for (cursor, age) in cursors + .iter() + .zip([40 * day, 35 * day, 30 * day, 10 * day, 0]) + { + sqlx::query("UPDATE library_event SET changed_at=? WHERE cursor=?") + .bind(now - age) + .bind(cursor) + .execute(state.db.pool()) + .await + .unwrap(); + } + + let purged = state.services.purge_library_events(now).await.unwrap(); + assert_eq!(purged.libraries_trimmed, 1); + // Two go. The thirty-day one sits exactly on the bound and the bound is + // exclusive — keeping one event too many breaks nobody, cutting one too + // many sends somebody back to the snapshot. + assert_eq!(purged.events_removed, 2, "only what is strictly older"); + + // The watermark moved with the delete, in the same transaction. Written + // separately there is a window where it claims less than has gone, and a + // client reading into it gets a catch-up that looks complete while skipping + // the gap. + let watermark: i64 = sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!(watermark, cursors[1], "the highest cursor actually cut"); + assert_eq!( + state + .services + .library_changes(owner, library, watermark, 500) + .await + .unwrap() + .events + .len(), + 3, + "the bound and everything after it stay" + ); + assert!( + matches!( + state + .services + .library_changes(owner, library, watermark - 1, 500) + .await, + Err(ServiceError::Conflict) + ), + "a cursor below the watermark has missed events" + ); + + // And the floor wins over the age. Everything left is now ancient, and the + // floor is two: exactly one may go. + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * day) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let purged = state.services.purge_library_events(now).await.unwrap(); + assert_eq!( + purged.events_removed, 1, + "three rows, a floor of two: one may go however old they all are" + ); + + // A second pass changes nothing: the floor is reached and stays reached. + assert_eq!( + state.services.purge_library_events(now).await.unwrap(), + Default::default(), + "a library at its floor is not trimmed again" + ); + let watermark: i64 = sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!( + state + .services + .library_changes(owner, library, watermark, 500) + .await + .unwrap() + .events + .len(), + 2, + "the floor kept a usable tail rather than an empty feed" + ); +}