feat(api): trim the library feed, so its expiry answer is finally exercised

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.

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

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)

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<EventPurge, ServiceError> {
+        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<String> = 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<u64, ServiceError> {
+        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<i64> = 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

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(),