From fe438947f183dc37aeb26b69d887e840fd81f4d0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 13:35:36 +0000 Subject: [PATCH] Traffic: never fall back to the default settings after a failed read The worker seeded its settings with TrafficSettings::default() and the shutdown flush used read_settings().unwrap_or_default(): with the database unreachable, an instance recorded client addresses an administrator had turned off and pruned with the default retention. Traffic now keeps the settings last applied. The worker and the shutdown flush fall back to them; before one read succeeded, the flush drops client addresses and nothing is pruned. maxmind::update writes nothing over a state it could not read again, which wiped account_id. Fixes #789 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012e6DJbe2bmzbuPSvHMxpUd --- crates/server/src/admin/traffic.rs | 2 +- crates/server/src/traffic/maxmind.rs | 14 +- crates/server/src/traffic/mod.rs | 194 ++++++++++++++++++++++----- crates/server/tests/traffic.rs | 25 ++++ crates/traffic/src/lib.rs | 21 +++ docs/changes/2.2.0/789.fixed.md | 9 ++ 6 files changed, 233 insertions(+), 32 deletions(-) create mode 100644 docs/changes/2.2.0/789.fixed.md diff --git a/crates/server/src/admin/traffic.rs b/crates/server/src/admin/traffic.rs index c0dfb95b..5012b50a 100644 --- a/crates/server/src/admin/traffic.rs +++ b/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, diff --git a/crates/server/src/traffic/maxmind.rs b/crates/server/src/traffic/maxmind.rs index 9589c56c..27360c02 100644 --- a/crates/server/src/traffic/maxmind.rs +++ b/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()); diff --git a/crates/server/src/traffic/mod.rs b/crates/server/src/traffic/mod.rs index 496a55fd..30b0c6a7 100644 --- a/crates/server/src/traffic/mod.rs +++ b/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, /// `/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 { 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, +) -> 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) -> 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) { + 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) -> 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() { diff --git a/crates/server/tests/traffic.rs b/crates/server/tests/traffic.rs index 020f2c44..7d853f3e 100644 --- a/crates/server/tests/traffic.rs +++ b/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" + ); +} diff --git a/crates/traffic/src/lib.rs b/crates/traffic/src/lib.rs index 908c6522..8e0881ac 100644 --- a/crates/traffic/src/lib.rs +++ b/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>, + /// 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>>, } 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 { + self.applied + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + /// Write what was counted since the last flush. pub async fn flush( &self, diff --git a/docs/changes/2.2.0/789.fixed.md b/docs/changes/2.2.0/789.fixed.md new file mode 100644 index 00000000..3282224c --- /dev/null +++ b/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.