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.