Traffic: keep the last-read settings instead of the defaults when the settings cannot be read

crates/server/src/admin/traffic.rs

@@ -232,7 +232,7 @@ pub async fn put_settings(
         if let Err(e) = write_settings(&state.pool, &next).await {
             return super::db_error(e);
         }
-        state.traffic.recorder.set_enabled(next.enabled);
+        state.traffic.apply(&next);
         tracing::info!(
             target: "forge::audit",
             settings = ?next,

crates/server/src/traffic/maxmind.rs

@@ -250,7 +250,19 @@ pub async fn update(state: &AppState) -> anyhow::Result<()> {
         anyhow::Ok(())
     }
     .await;
-    let (mut recorded, _) = read_state(&state.pool).await.unwrap_or_default();
+    // Recorded over the state read again; nothing is written over a state
+    // that cannot be read, which would lose the account ID.
+    let mut recorded = match read_state(&state.pool).await {
+        Ok((recorded, _)) => recorded,
+        Err(e) => {
+            forge_db::observe_error(&e);
+            tracing::warn!(error = %e, "cannot record the MaxMind download: the state cannot be read");
+            if let Err(e) = &result {
+                tracing::warn!(error = %format!("{e:#}"), "cannot download the MaxMind databases");
+            }
+            return result;
+        }
+    };
     match &result {
         Ok(()) => {
             recorded.last_success = Some(Utc::now());

crates/server/src/traffic/mod.rs

@@ -45,6 +45,9 @@ pub const GEO_REFRESH_INTERVAL: Duration = Duration::from_secs(600);
 /// How long the last flush on shutdown may take.
 const SHUTDOWN_FLUSH: Duration = Duration::from_secs(10);
 
+/// How long of it the settings may take to read.
+const SHUTDOWN_READ: Duration = Duration::from_secs(3);
+
 /// The network `ip` is counted under, as text: the address itself for IPv4,
 /// `<prefix>/64` for IPv6.
 pub fn network_label(ip: IpAddr) -> String {
@@ -73,24 +76,66 @@ pub async fn write_settings(pool: &PgPool, settings: &TrafficSettings) -> sqlx::
     forge_db::authz_dao::set_setting(pool, SETTINGS_KEY, &json).await
 }
 
-/// The stored settings, applied to this instance's recorder; the ones
-/// applied before when they cannot be read.
-async fn follow_settings(
-    pool: &PgPool,
-    traffic: &Traffic,
-    last: &TrafficSettings,
-) -> TrafficSettings {
+/// The stored settings, applied to this instance ([`Traffic::apply`]); the
+/// ones applied before when they cannot be read, `None` when none were. The
+/// defaults never stand in for them: they record client addresses an
+/// administrator may have turned off, and their retention may be shorter.
+async fn follow_settings(pool: &PgPool, traffic: &Traffic) -> Option<TrafficSettings> {
     match read_settings(pool).await {
         Ok(settings) => {
-            traffic.recorder.set_enabled(settings.enabled);
-            settings
+            traffic.apply(&settings);
+            Some(settings)
         }
         Err(e) => {
             forge_db::observe_error(&e);
-            tracing::warn!(error = %e, "cannot read the traffic settings; keeping the ones in force");
-            last.clone()
+            let applied = traffic.applied();
+            if applied.is_some() {
+                tracing::warn!(error = %e, "cannot read the traffic settings; keeping the ones in force");
+            } else {
+                tracing::warn!(
+                    error = %e,
+                    "cannot read the traffic settings; until they are read, client addresses \
+                     are dropped and nothing is pruned"
+                );
+            }
+            applied
+        }
+    }
+}
+
+/// What a flush writes with: `settings`, or before any were read, the
+/// defaults without client addresses.
+fn flush_settings(settings: Option<&TrafficSettings>) -> TrafficSettings {
+    settings.cloned().unwrap_or(TrafficSettings {
+        client_addresses: false,
+        ..TrafficSettings::default()
+    })
+}
+
+/// Flush what was counted with `settings` ([`flush_settings`]) and, when
+/// `prune` and the settings were read, delete the files past their retention.
+/// Returns whether all of it went well.
+async fn flush_and_prune(
+    traffic: &Traffic,
+    settings: Option<&TrafficSettings>,
+    prune: bool,
+    now: DateTime<Utc>,
+) -> bool {
+    let mut ok = traffic.flush(&flush_settings(settings), now).await.failures == 0;
+    let Some(settings) = settings.filter(|_| prune) else {
+        return ok;
+    };
+    match forge_traffic::layout::prune(&*traffic.objects, settings, now).await {
+        Ok(0) => {}
+        Ok(deleted) => {
+            tracing::info!(deleted, "deleted expired traffic statistics");
+        }
+        Err(e) => {
+            ok = false;
+            tracing::warn!(error = %e, "cannot delete expired traffic statistics");
         }
     }
+    ok
 }
 
 /// Load the geolocation databases from the bucket when they changed.
@@ -116,8 +161,7 @@ fn until_next_flush(now: DateTime<Utc>) -> Duration {
 pub fn spawn_worker(state: AppState) -> JoinHandle<()> {
     tokio::spawn(async move {
         let traffic = state.traffic.clone();
-        let mut settings =
-            follow_settings(&state.pool, &traffic, &TrafficSettings::default()).await;
+        let mut settings = follow_settings(&state.pool, &traffic).await;
         refresh_geo(&traffic).await;
         let mut next_geo = tokio::time::Instant::now() + GEO_REFRESH_INTERVAL;
         // Instances started together prune and download at different times.
@@ -127,27 +171,16 @@ pub fn spawn_worker(state: AppState) -> JoinHandle<()> {
             tokio::time::sleep(until_next_flush(Utc::now())).await;
             let run = forge_telemetry::workers::Run::start(WORKER);
             let ok = async {
-                settings = follow_settings(&state.pool, &traffic, &settings).await;
+                settings = follow_settings(&state.pool, &traffic).await;
                 let mut ok = true;
                 if tokio::time::Instant::now() >= next_geo {
                     next_geo = tokio::time::Instant::now() + GEO_REFRESH_INTERVAL;
                     ok &= refresh_geo(&traffic).await;
                 }
-                ok &= traffic.flush(&settings, Utc::now()).await.failures == 0;
-                if tokio::time::Instant::now() >= next_hourly {
+                let hourly = tokio::time::Instant::now() >= next_hourly;
+                ok &= flush_and_prune(&traffic, settings.as_ref(), hourly, Utc::now()).await;
+                if hourly {
                     next_hourly = tokio::time::Instant::now() + HOURLY;
-                    match forge_traffic::layout::prune(&*traffic.objects, &settings, Utc::now())
-                        .await
-                    {
-                        Ok(0) => {}
-                        Ok(deleted) => {
-                            tracing::info!(deleted, "deleted expired traffic statistics");
-                        }
-                        Err(e) => {
-                            ok = false;
-                            tracing::warn!(error = %e, "cannot delete expired traffic statistics");
-                        }
-                    }
                     if let Err(e) = maxmind::update_if_due(&state).await {
                         ok = false;
                         tracing::warn!(
@@ -169,8 +202,14 @@ pub fn spawn_worker(state: AppState) -> JoinHandle<()> {
 /// exits: at most [`SHUTDOWN_FLUSH`].
 pub async fn flush_on_shutdown(traffic: &Traffic, pool: &PgPool) {
     let flush = async {
-        let settings = read_settings(pool).await.unwrap_or_default();
-        traffic.flush(&settings, Utc::now()).await
+        // Bounded, so that a database that does not answer leaves the time
+        // to write with the settings applied before.
+        let settings = tokio::time::timeout(SHUTDOWN_READ, follow_settings(pool, traffic))
+            .await
+            .unwrap_or_else(|_| traffic.applied());
+        traffic
+            .flush(&flush_settings(settings.as_ref()), Utc::now())
+            .await
     };
     match tokio::time::timeout(SHUTDOWN_FLUSH, flush).await {
         Ok(report) if report.failures > 0 => {
@@ -187,6 +226,101 @@ pub async fn flush_on_shutdown(traffic: &Traffic, pool: &PgPool) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::Arc;
+
+    use forge_traffic::{Event, MemoryObjects, Noted};
+
+    /// A pool whose every read fails: nothing listens on port 1.
+    fn unreachable_pool() -> PgPool {
+        sqlx::postgres::PgPoolOptions::new()
+            .acquire_timeout(Duration::from_millis(200))
+            .connect_lazy("postgres://forge@127.0.0.1:1/forge")
+            .unwrap()
+    }
+
+    fn traffic() -> Traffic {
+        Traffic::new(Arc::new(MemoryObjects::default()), "a".into())
+    }
+
+    fn request(traffic: &Traffic, at: DateTime<Utc>) {
+        traffic.recorder.record(Event {
+            at,
+            bytes: 10,
+            noted: Noted {
+                client: "192.0.2.1".parse().ok(),
+                network: Some("192.0.2.1".into()),
+                repository: Some("npm".into()),
+            },
+        });
+    }
+
+    /// What `traffic`'s files of `day` hold.
+    async fn overview(traffic: &Traffic, day: DateTime<Utc>) -> forge_traffic::query::Overview {
+        let day = day.date_naive();
+        forge_traffic::query::overview(
+            &*traffic.objects,
+            &traffic.cache,
+            day,
+            day,
+            &Default::default(),
+        )
+        .await
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn addresses_turned_off_stay_off_when_the_settings_cannot_be_read() {
+        let traffic = traffic();
+        let pool = unreachable_pool();
+        let stored = TrafficSettings {
+            client_addresses: false,
+            retention_days: 365,
+            ..TrafficSettings::default()
+        };
+        // What the worker read while the database answered.
+        traffic.apply(&stored);
+
+        // The worker's next read fails: it keeps them.
+        assert_eq!(follow_settings(&pool, &traffic).await, Some(stored));
+
+        // So does the flush on shutdown.
+        let now = Utc::now();
+        request(&traffic, now);
+        flush_on_shutdown(&traffic, &pool).await;
+        let written = overview(&traffic, now).await;
+        assert_eq!(written.totals.requests, 1, "the request is counted");
+        assert_eq!(written.totals.clients, 0, "{:?}", written.clients);
+        assert!(written.clients.is_empty(), "{:?}", written.clients);
+    }
+
+    #[tokio::test]
+    async fn nothing_is_pruned_and_no_address_written_before_the_settings_are_read() {
+        let traffic = traffic();
+        let pool = unreachable_pool();
+        let now = Utc::now();
+        // Past the default retention, within the one an administrator set.
+        let old =
+            forge_traffic::layout::day_key((now - chrono::Duration::days(30)).timestamp(), "b");
+        traffic.objects.put(&old, vec![1]).await.unwrap();
+
+        let settings = follow_settings(&pool, &traffic).await;
+        assert_eq!(settings, None, "no defaults stand in for the settings");
+        request(&traffic, now);
+        flush_and_prune(&traffic, settings.as_ref(), true, now).await;
+        assert!(
+            traffic.objects.get(&old).await.unwrap().is_some(),
+            "pruned before the settings were read"
+        );
+        let written = overview(&traffic, now).await;
+        assert_eq!(written.totals.requests, 1, "the request is counted");
+        assert!(written.clients.is_empty(), "{:?}", written.clients);
+
+        // Read, the settings prune.
+        let read = TrafficSettings::default();
+        traffic.apply(&read);
+        flush_and_prune(&traffic, Some(&read), true, now).await;
+        assert!(traffic.objects.get(&old).await.unwrap().is_none());
+    }
 
     #[test]
     fn ipv6_clients_are_counted_by_their_64() {

crates/server/tests/traffic.rs

@@ -379,3 +379,28 @@ async fn maxmind_databases_are_downloaded_and_locate_clients() {
     assert_eq!(view["country"], Value::Null);
     assert_eq!(view["maxmind"]["license_key_set"], false);
 }
+
+#[tokio::test]
+async fn a_download_recorded_over_a_state_it_cannot_read_keeps_the_account_id() {
+    let srv = server(None).await;
+    // A state this version cannot read, with the account ID in it.
+    let unreadable = r#"{"account_id":"1234","last_success":"not a time"}"#;
+    forge_db::authz_dao::set_setting(&srv.pool, "traffic_maxmind", unreadable)
+        .await
+        .unwrap();
+    assert!(forge_server::traffic::maxmind::read_state(&srv.pool)
+        .await
+        .is_err());
+
+    assert!(forge_server::traffic::maxmind::update(&srv.state)
+        .await
+        .is_err());
+    assert_eq!(
+        forge_db::authz_dao::get_setting(&srv.pool, "traffic_maxmind")
+            .await
+            .unwrap()
+            .as_deref(),
+        Some(unreadable),
+        "nothing is written over the state"
+    );
+}

crates/traffic/src/lib.rs

@@ -65,6 +65,10 @@ pub struct Traffic {
     /// of the worker and the last flush on shutdown adds to what the worker
     /// wrote instead of replacing it.
     flusher: Arc<tokio::sync::Mutex<Flusher>>,
+    /// The settings last read and applied ([`Self::apply`]); `None` until a
+    /// read succeeded. Shared, so that a restarted worker and the last flush
+    /// on shutdown keep to them when the settings cannot be read.
+    applied: Arc<std::sync::Mutex<Option<TrafficSettings>>>,
 }
 
 impl Traffic {
@@ -81,9 +85,26 @@ impl Traffic {
             objects,
             instance,
             cache: query::FileCache::new(),
+            applied: Arc::default(),
         }
     }
 
+    /// Apply the settings just read: the recorder follows
+    /// [`TrafficSettings::enabled`], and they are kept as [`Self::applied`].
+    pub fn apply(&self, settings: &TrafficSettings) {
+        self.recorder.set_enabled(settings.enabled);
+        *self.applied.lock().unwrap_or_else(|e| e.into_inner()) = Some(settings.clone());
+    }
+
+    /// The settings last applied ([`Self::apply`]); `None` before the first
+    /// read succeeded, when no defaults may stand in for them.
+    pub fn applied(&self) -> Option<TrafficSettings> {
+        self.applied
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .clone()
+    }
+
     /// Write what was counted since the last flush.
     pub async fn flush(
         &self,

docs/changes/2.2.0/789.fixed.md

@@ -0,0 +1,9 @@
+When the traffic settings could not be read, for example while PostgreSQL
+failed over, the traffic statistics fell back to the defaults: an instance
+starting during the outage recorded client addresses an administrator had
+turned off, and its hourly pruning deleted the days past the default
+retention rather than the one set; an instance stopping wrote its last
+minute's client addresses the same way. An instance now keeps the settings
+it last read, and until it has read them once it writes no client address
+and prunes nothing. A failed read of the MaxMind state after a download
+no longer overwrites the stored account ID.