From c69b043a32e82720940b100ed6ba22a4f34be342 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 18:25:42 +0200 Subject: [PATCH 1/2] feat(api): trim the library feed, so its expiry answer is finally exercised MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RFC-007 decision 7, built. Until now nothing purged: `library.events_purged_through` sat at 0 by construction, which meant the feed's whole expiry path — the watermark read, the `Conflict`, the comment explaining why a derived floor would be wrong — was correct code that had never once run. Two bounds, and the floor wins. Thirty days by default, never fewer than ten thousand events per library, both refused at startup if zero or negative rather than quietly defaulted. The age bound is exclusive: an event exactly thirty days old lives. **The delete and the watermark are one transaction.** 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. `MAX` on the update, so a pass cutting an older tail cannot lower a watermark an earlier one raised. 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 length of the pass. Three rules removed to watch the test fall: make the bound inclusive and the event sitting on it is cut; drop the floor and three rows go where one should; stop moving the watermark and the feed hands back a tail it should have refused. ## The test was flaky, and it said so on the second run It read the clock, then let the purge read it again — and with an exclusive bound, "exactly thirty days old" is a single instant. The milliseconds between the two reads put the boundary event on the wrong side, so it passed once and failed the next time for nothing that had changed. `purge_library_events` takes `now_ms` now, the way `sweep_expired_sessions` already did. A caller that cannot name the instant cannot place an event on it — it can only aim near it and hope, which is not a test of a boundary. Five consecutive runs green, and it was caught because the suite was run again rather than once. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- docs/rfcs/RFC-007-library-event-stream.md | 10 +- src/config.rs | 42 ++++++ src/main.rs | 1 + src/services/library_events.rs | 152 ++++++++++++++++++++ src/services/mod.rs | 2 + tests/catalog.rs | 161 ++++++++++++++++++++++ 6 files changed, 362 insertions(+), 6 deletions(-) diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index 40d5b2b..6fbe734 100644 --- a/docs/rfcs/RFC-007-library-event-stream.md +++ b/docs/rfcs/RFC-007-library-event-stream.md @@ -235,9 +235,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 +322,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" + ); +} From a61aa893aa9567dcb80b2eb71d5b784ba1b1136d Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 18:26:14 +0200 Subject: [PATCH 2/2] docs(rfc): name the pull requests that built decisions 6 and 7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `Implémentée par` line stopped at #152 and had missed #159 since this morning — the album events, decision 6 — while the line right below it claims to be the one thing in this document that can be checked. A line that says what runs is worth exactly as much as its last update. #166 joins it for decision 7. Decision 8 stays absent, because it is decided and not built, and that is the distinction the line exists to carry. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- docs/rfcs/RFC-007-library-event-stream.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index 6fbe734..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