From 48cde467c74fb1d448e4253f3263f3e1fd06d41a Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Mon, 9 Nov 2026 22:52:29 +0100 Subject: [PATCH 1/3] fix(settings): commit and publish updates atomically (RUST-BUG-184) --- Cargo.lock | 1 + crates/emulebb-core/src/app_api.rs | 102 ++++++++--- crates/emulebb-core/src/lib.rs | 3 + crates/emulebb-core/src/profile_state.rs | 169 ++++++++++------- crates/emulebb-core/src/tests/core_config.rs | 179 +++++++++++++++++++ crates/emulebb-metadata/src/profile_store.rs | 161 ++++++++++++++++- crates/emulebb-rest/Cargo.toml | 1 + crates/emulebb-rest/src/handlers/app.rs | 8 + crates/emulebb-rest/src/tests/route_app.rs | 143 +++++++++++++++ 9 files changed, 680 insertions(+), 87 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8342fe4d..002e7b49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -825,6 +825,7 @@ dependencies = [ "futures-util", "http-body-util", "percent-encoding", + "rusqlite", "serde", "serde_json", "serde_urlencoded", diff --git a/crates/emulebb-core/src/app_api.rs b/crates/emulebb-core/src/app_api.rs index ca65015f..8b4ea7f7 100644 --- a/crates/emulebb-core/src/app_api.rs +++ b/crates/emulebb-core/src/app_api.rs @@ -1,6 +1,22 @@ use super::*; +#[derive(Debug)] +struct CommittedSettingsRefreshFailure; + +impl std::fmt::Display for CommittedSettingsRefreshFailure { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("settings committed but runtime server-plan refresh failed") + } +} + +impl std::error::Error for CommittedSettingsRefreshFailure {} + impl EmulebbCore { + /// Identify a saved settings update whose subsequent runtime refresh failed. + pub fn settings_update_was_committed(error: &anyhow::Error) -> bool { + error.is::() + } + pub fn diagnostic_writer_health(&self) -> emulebb_diagnostics::DiagWriterHealth { emulebb_diagnostics::writer_health() } @@ -40,22 +56,22 @@ impl EmulebbCore { } pub async fn app_settings(&self) -> Result { + let _settings_order = self.settings_update_order.lock().await; profile_state::load_app_settings(&self.metadata_store) } pub async fn update_app_settings(&self, mut request: AppSettingsUpdate) -> Result { validate_app_settings_update_paths(&mut request)?; - let core_update = request.core.take(); - if core_update.is_none() { - profile_state::persist_app_settings_update(&self.metadata_store, request)?; - return self.app_settings().await; - } - if !app_settings_update_is_empty(&request) { - profile_state::persist_app_settings_update(&self.metadata_store, request)?; - } - self.update_core_settings(core_update.expect("checked as Some")) - .await?; - self.app_settings().await + // WHY: admit at most one owned completion task. Cancellation while queued + // writes nothing; cancellation after commit cannot abandon publication. + let settings_order = Arc::clone(&self.settings_update_order).lock_owned().await; + let core = self.clone(); + tokio::spawn(async move { + let _settings_order = settings_order; + core.update_app_settings_ordered(request).await + }) + .await + .context("settings completion task failed; inspect canonical settings")? } pub async fn update_core_settings(&self, request: CoreSettingsUpdate) -> Result { @@ -63,39 +79,55 @@ impl EmulebbCore { !core_settings_update_is_empty(&request), "settings.core PATCH requires at least one core setting" ); - let core_settings = { - let mut state = self.state.lock().await; - let mut core_settings = state.core_settings.clone(); - apply_core_settings_update(&mut core_settings, request)?; - profile_state::persist_core_settings(&self.metadata_store, &core_settings)?; - state.core_settings = core_settings.clone(); - core_settings - }; + Ok(self + .update_app_settings(AppSettingsUpdate { + core: Some(request), + ..AppSettingsUpdate::default() + }) + .await? + .core) + } + + async fn update_app_settings_ordered(&self, request: AppSettingsUpdate) -> Result { + let core_changed = request.core.is_some(); + let committed = profile_state::persist_app_settings_update(&self.metadata_store, request)?; + if core_changed { + self.state.lock().await.core_settings = committed.core.clone(); + self.publish_committed_core_settings(&committed.core) + .await?; + } + Ok(committed) + } + + async fn publish_committed_core_settings(&self, core_settings: &CoreSettings) -> Result<()> { self.ed2k_transfers .apply_upload_queue_policy(&ed2k_upload_queue_policy_from_core_settings( self.ed2k_network .as_ref() .map(|network| &network.ed2k.upload_queue), - &core_settings, + core_settings, )) .await; self.ed2k_transfers .apply_download_limit(ed2k_download_limit_bytes_per_sec_from_core_settings( - &core_settings, + core_settings, )) .await; // Apply the global connection budget + per-file source caps live, like // the download limit (eMule GetMaxConnections / GetMaxConperFive / // GetConfiguredMaxSourcesPerFile core setting changes take effect at once). self.ed2k_transfers.apply_download_coordinator_config( - ed2k_download_coordinator_config_from_core_settings(&core_settings), + ed2k_download_coordinator_config_from_core_settings(core_settings), ); // Apply the credit-system toggle live (eMule thePrefs.GetCreditSystem()): // when off, upload scoring uses the neutral 1.0 credit ratio for everyone. self.ed2k_transfers .set_credit_system_enabled(core_settings.credit_system); - self.refresh_ed2k_server_connect_plan().await?; - Ok(core_settings) + if let Err(error) = self.refresh_ed2k_server_connect_plan().await { + tracing::error!(%error, "settings committed but runtime server-plan refresh failed"); + return Err(error).context(CommittedSettingsRefreshFailure); + } + Ok(()) } pub async fn status(&self) -> Status { @@ -146,6 +178,28 @@ impl EmulebbCore { } } +#[cfg(test)] +mod settings_error_tests { + use super::*; + + #[test] + fn committed_refresh_classification_uses_the_type_not_the_message() { + let same_message = anyhow::anyhow!(CommittedSettingsRefreshFailure.to_string()); + assert!(!EmulebbCore::settings_update_was_committed(&same_message)); + let committed = Err::<(), _>(anyhow::anyhow!("synthetic runtime refresh error")) + .context(CommittedSettingsRefreshFailure) + .unwrap_err(); + assert!(EmulebbCore::settings_update_was_committed(&committed)); + assert_eq!( + committed.to_string(), + CommittedSettingsRefreshFailure.to_string() + ); + assert!(EmulebbCore::settings_update_was_committed( + &committed.context("outer context") + )); + } +} + fn validate_app_settings_update_paths(request: &mut AppSettingsUpdate) -> Result<()> { if let Some(daemon) = &mut request.daemon && let Some(emulebb_settings::NullableUpdate::Value(path)) = &mut daemon.incoming_dir diff --git a/crates/emulebb-core/src/lib.rs b/crates/emulebb-core/src/lib.rs index 3f0437c8..0afa1b32 100644 --- a/crates/emulebb-core/src/lib.rs +++ b/crates/emulebb-core/src/lib.rs @@ -526,6 +526,8 @@ pub struct EmulebbCore { search_queue: Arc>, transfer_events: broadcast::Sender, next_transfer_event_id: Arc, + /// Orders settings commit and runtime publication, including canceled waiters. + settings_update_order: Arc>, state: Arc>, } @@ -701,6 +703,7 @@ impl EmulebbCore { search_queue: Arc::new(parking_lot::Mutex::new(SearchQueue::new())), transfer_events, next_transfer_event_id: Arc::new(AtomicU64::new(1)), + settings_update_order: Arc::new(Mutex::new(())), state: Arc::new(Mutex::new(core_state)), }) } diff --git a/crates/emulebb-core/src/profile_state.rs b/crates/emulebb-core/src/profile_state.rs index 93a6ba6b..f6f02fe2 100644 --- a/crates/emulebb-core/src/profile_state.rs +++ b/crates/emulebb-core/src/profile_state.rs @@ -1,6 +1,6 @@ use std::collections::{BTreeMap, HashMap, HashSet}; -use anyhow::Result; +use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use emulebb_ed2k::long_path::normal_path_display; use emulebb_metadata::{MetadataCategory, MetadataFriend, MetadataServer, MetadataStore}; @@ -9,12 +9,13 @@ use emulebb_settings::{ SECTION_KAD, SECTION_NAT, SECTION_VPN_GUARD, app_settings_update_is_empty, apply_daemon_settings_update, apply_ed2k_settings_update, apply_ip_filter_settings_update, apply_kad_settings_update, apply_nat_settings_update, apply_vpn_guard_settings_update, - core_settings_from_values, core_settings_to_values, section_settings_from_values, - section_settings_to_values, validate_ed2k_settings, + core_settings_from_values, core_settings_to_values, core_settings_update_is_empty, + section_settings_from_values, section_settings_to_values, validate_ed2k_settings, }; use crate::{ - Category, CoreSettings, CoreState, Friend, ServerInfo, SharedDirectoryRoot, default_categories, + Category, CoreState, Friend, ServerInfo, SharedDirectoryRoot, apply_core_settings_update, + default_categories, }; pub(crate) fn load_core_state( @@ -105,9 +106,10 @@ pub(crate) fn has_persisted_core_settings(metadata: &MetadataStore) -> Result Result<()> { let entries = core_settings_to_values(core_settings)?; metadata.replace_settings_section( @@ -119,20 +121,50 @@ pub(crate) fn persist_core_settings( Ok(()) } +const APP_SETTINGS_SECTIONS: [&str; 7] = [ + SECTION_CORE, + SECTION_DAEMON, + SECTION_ED2K, + SECTION_KAD, + SECTION_NAT, + SECTION_VPN_GUARD, + SECTION_IP_FILTER, +]; + pub(crate) fn load_app_settings(metadata: &MetadataStore) -> Result { - let mut daemon = load_settings_section(metadata, SECTION_DAEMON)?; + let sections = APP_SETTINGS_SECTIONS + .iter() + .map(|section| { + Ok(( + (*section).to_owned(), + metadata.load_settings_section(section)?, + )) + }) + .collect::>>()?; + app_settings_from_sections(§ions) +} + +fn app_settings_from_sections( + sections: &BTreeMap>, +) -> Result { + let mut daemon = decode_settings_section(sections, SECTION_DAEMON)?; normalize_daemon_settings_paths(&mut daemon); - let ed2k = load_settings_section(metadata, SECTION_ED2K)?; + let ed2k = decode_settings_section(sections, SECTION_ED2K)?; validate_ed2k_settings(&ed2k)?; - let nat = load_settings_section(metadata, SECTION_NAT)?; + let rows = sections + .get(SECTION_CORE) + .context("core settings snapshot is missing")?; Ok(AppSettings { - core: load_core_settings(metadata)?, + core: core_settings_from_values( + rows.iter() + .map(|(key, value)| (key.as_str(), value.as_str())), + )?, daemon, ed2k, - kad: load_settings_section(metadata, SECTION_KAD)?, - nat, - vpn_guard: load_settings_section(metadata, SECTION_VPN_GUARD)?, - ip_filter: load_settings_section(metadata, SECTION_IP_FILTER)?, + kad: decode_settings_section(sections, SECTION_KAD)?, + nat: decode_settings_section(sections, SECTION_NAT)?, + vpn_guard: decode_settings_section(sections, SECTION_VPN_GUARD)?, + ip_filter: decode_settings_section(sections, SECTION_IP_FILTER)?, }) } @@ -144,56 +176,85 @@ pub(crate) fn persist_app_settings_update( !app_settings_update_is_empty(&update), "settings PATCH requires at least one settings section" ); - let ed2k_candidate = if let Some(ed2k_update) = update.ed2k.as_ref() { - let mut settings = load_settings_section(metadata, SECTION_ED2K)?; - apply_ed2k_settings_update(&mut settings, ed2k_update.clone())?; - Some(settings) - } else { - None - }; + let changed = [ + (SECTION_CORE, update.core.is_some()), + (SECTION_DAEMON, update.daemon.is_some()), + (SECTION_ED2K, update.ed2k.is_some()), + (SECTION_KAD, update.kad.is_some()), + (SECTION_NAT, update.nat.is_some()), + (SECTION_VPN_GUARD, update.vpn_guard.is_some()), + (SECTION_IP_FILTER, update.ip_filter.is_some()), + ]; + metadata.update_settings_sections(&APP_SETTINGS_SECTIONS, move |sections| { + let candidate = apply_app_settings_update(app_settings_from_sections(sections)?, update)?; + // No SQL writes occur until the whole candidate validates and serializes. + for (section, changed) in changed { + if !changed { + continue; + } + let entries = match section { + SECTION_CORE => core_settings_to_values(&candidate.core)? + .into_iter() + .map(|(key, value)| (key.to_owned(), value)) + .collect(), + SECTION_DAEMON => section_settings_to_values(&candidate.daemon)?, + SECTION_ED2K => section_settings_to_values(&candidate.ed2k)?, + SECTION_KAD => section_settings_to_values(&candidate.kad)?, + SECTION_NAT => section_settings_to_values(&candidate.nat)?, + SECTION_VPN_GUARD => section_settings_to_values(&candidate.vpn_guard)?, + SECTION_IP_FILTER => section_settings_to_values(&candidate.ip_filter)?, + _ => unreachable!("selected application settings section"), + }; + sections.insert(section.to_owned(), entries); + } + Ok(candidate) + }) +} + +fn apply_app_settings_update( + mut settings: AppSettings, + update: AppSettingsUpdate, +) -> Result { + if let Some(update) = update.core { + anyhow::ensure!( + !core_settings_update_is_empty(&update), + "settings.core PATCH requires at least one core setting" + ); + apply_core_settings_update(&mut settings.core, update)?; + } if let Some(update) = update.daemon { - let mut settings = load_settings_section(metadata, SECTION_DAEMON)?; - apply_daemon_settings_update(&mut settings, update); - persist_settings_section(metadata, SECTION_DAEMON, &settings)?; + apply_daemon_settings_update(&mut settings.daemon, update); + normalize_daemon_settings_paths(&mut settings.daemon); } - if let Some(settings) = ed2k_candidate { - persist_settings_section(metadata, SECTION_ED2K, &settings)?; + if let Some(update) = update.ed2k { + apply_ed2k_settings_update(&mut settings.ed2k, update)?; } if let Some(update) = update.kad { - let mut settings = load_settings_section(metadata, SECTION_KAD)?; - apply_kad_settings_update(&mut settings, update); - persist_settings_section(metadata, SECTION_KAD, &settings)?; + apply_kad_settings_update(&mut settings.kad, update); } if let Some(update) = update.nat { - let mut settings = load_settings_section(metadata, SECTION_NAT)?; - apply_nat_settings_update(&mut settings, update); - persist_settings_section(metadata, SECTION_NAT, &settings)?; + apply_nat_settings_update(&mut settings.nat, update); } if let Some(update) = update.vpn_guard { - let mut settings = load_settings_section(metadata, SECTION_VPN_GUARD)?; - apply_vpn_guard_settings_update(&mut settings, update); - persist_settings_section(metadata, SECTION_VPN_GUARD, &settings)?; + apply_vpn_guard_settings_update(&mut settings.vpn_guard, update); } if let Some(update) = update.ip_filter { - let mut settings = load_settings_section(metadata, SECTION_IP_FILTER)?; - apply_ip_filter_settings_update(&mut settings, update); - persist_settings_section(metadata, SECTION_IP_FILTER, &settings)?; + apply_ip_filter_settings_update(&mut settings.ip_filter, update); } - load_app_settings(metadata) + validate_ed2k_settings(&settings.ed2k)?; + Ok(settings) } -pub(crate) fn load_core_settings(metadata: &MetadataStore) -> Result { - let rows = metadata.load_settings_section(SECTION_CORE)?; - Ok(core_settings_from_values(rows.iter().map( - |(key, value_json)| (key.as_str(), value_json.as_str()), - ))?) -} - -fn load_settings_section(metadata: &MetadataStore, section: &str) -> Result +fn decode_settings_section( + sections: &BTreeMap>, + section: &str, +) -> Result where T: Default + serde::de::DeserializeOwned, { - let rows = metadata.load_settings_section(section)?; + let rows = sections + .get(section) + .with_context(|| format!("settings snapshot is missing {section}"))?; Ok(section_settings_from_values( section, rows.iter() @@ -201,20 +262,6 @@ where )?) } -fn persist_settings_section(metadata: &MetadataStore, section: &str, settings: &T) -> Result<()> -where - T: serde::Serialize, -{ - let entries = section_settings_to_values(settings)?; - metadata.replace_settings_section( - section, - entries - .iter() - .map(|(key, value_json)| (key.as_str(), value_json.as_str())), - )?; - Ok(()) -} - pub(crate) fn persist_category(metadata: &MetadataStore, category: &Category) -> Result<()> { metadata.upsert_category(&MetadataCategory { id: category.id, diff --git a/crates/emulebb-core/src/tests/core_config.rs b/crates/emulebb-core/src/tests/core_config.rs index 87fb87a9..d709781e 100644 --- a/crates/emulebb-core/src/tests/core_config.rs +++ b/crates/emulebb-core/src/tests/core_config.rs @@ -1,5 +1,184 @@ use super::*; +fn atomic_settings_patch(value: serde_json::Value) -> AppSettingsUpdate { + serde_json::from_value(value).unwrap() +} + +async fn wait_for_atomic_settings_commit(core: &EmulebbCore, key: &str, value: &str) { + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + if core + .metadata_store + .load_setting_json("core", key) + .unwrap() + .as_deref() + == Some(value) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("settings transaction did not reach its commit barrier"); +} + +#[tokio::test] +async fn invalid_direct_settings_candidate_changes_neither_sql_nor_runtime() { + let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap(); + let before = core.app_settings().await.unwrap(); + let core_rows = core.metadata_store.load_settings_section("core").unwrap(); + let nat_rows = core.metadata_store.load_settings_section("nat").unwrap(); + let error = core + .update_app_settings(atomic_settings_patch(serde_json::json!({ + "nat": {"enabled": !before.nat.enabled}, "core": {"uploadLimitKiBps": 0}, + }))) + .await + .unwrap_err(); + assert!(error.to_string().contains("uploadLimitKiBps")); + assert_eq!(core.app_settings().await.unwrap(), before); + assert_eq!(core.core_settings().await, before.core); + assert_eq!( + core.metadata_store.load_settings_section("core").unwrap(), + core_rows + ); + assert_eq!( + core.metadata_store.load_settings_section("nat").unwrap(), + nat_rows + ); +} + +#[tokio::test] +async fn concurrent_settings_patches_preserve_disjoint_fields_and_omitted_sections() { + let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap(); + let before = core.app_settings().await.unwrap(); + let lease = if before.nat.lease_duration_secs == 7200 { + 3600 + } else { + 7200 + }; + let first = atomic_settings_patch(serde_json::json!({"nat": {"enabled": !before.nat.enabled}})); + let second = atomic_settings_patch(serde_json::json!({"nat": {"leaseDurationSecs": lease}})); + let (first, second) = tokio::join!( + core.update_app_settings(first), + core.update_app_settings(second) + ); + first.unwrap(); + second.unwrap(); + let after = core.app_settings().await.unwrap(); + assert_eq!(after.nat.enabled, !before.nat.enabled); + assert_eq!(after.nat.lease_duration_secs, lease); + assert_eq!(after.nat.backend_order, before.nat.backend_order); + assert_eq!(after.core, before.core); + assert_eq!(after.daemon, before.daemon); + assert_eq!(after.ed2k, before.ed2k); + assert_eq!(after.kad, before.kad); + assert_eq!(after.vpn_guard, before.vpn_guard); + assert_eq!(after.ip_filter, before.ip_filter); +} + +#[tokio::test] +async fn canceled_settings_waiter_cannot_abandon_committed_runtime_publication() { + let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap(); + // Hold the existing state lock so the owned task commits, then waits before + // publishing any core state or awaited driver policy. + let state_barrier = core.state.lock().await; + let updater = core.clone(); + let waiter = tokio::spawn(async move { + updater + .update_app_settings(atomic_settings_patch(serde_json::json!({ + "core": {"uploadLimitKiBps": 1024, "downloadLimitKiBps": 2048}, + }))) + .await + }); + wait_for_atomic_settings_commit(&core, "uploadLimitKiBps", "1024").await; + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + drop(state_barrier); + // The ordered read waits for the detached completion task, including its + // runtime effects, rather than merely observing the committed SQL rows. + let committed = tokio::time::timeout(std::time::Duration::from_secs(2), core.app_settings()) + .await + .unwrap() + .unwrap(); + assert_eq!(core.core_settings().await, committed.core); + assert_eq!( + core.ed2k_transfers + .upload_queue_policy_snapshot() + .await + .upload_limit_bytes_per_sec, + 1024 * 1024 + ); + assert_eq!( + core.ed2k_transfers.download_limit_bytes_per_sec().await, + 2048 * 1024 + ); +} + +#[tokio::test] +async fn canceled_queued_settings_request_writes_nothing() { + let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap(); + let before = core.app_settings().await.unwrap(); + let order = Arc::clone(&core.settings_update_order).lock_owned().await; + let updater = core.clone(); + let (entered, waiting) = tokio::sync::oneshot::channel(); + let waiter = tokio::spawn(async move { + entered.send(()).unwrap(); + updater + .update_app_settings(atomic_settings_patch(serde_json::json!({ + "core": {"uploadLimitKiBps": 1024}, + }))) + .await + }); + waiting.await.unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + drop(order); + assert_eq!(core.app_settings().await.unwrap(), before); + assert!( + core.metadata_store + .load_settings_section("core") + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn concurrent_core_requests_publish_in_committed_order_after_a_blocked_update() { + let core = EmulebbCore::new_in_memory("test", FileIndex::in_memory().unwrap()).unwrap(); + let state_barrier = core.state.lock().await; + let first_core = core.clone(); + let first = tokio::spawn(async move { + first_core + .update_core_settings(CoreSettingsUpdate { + upload_limit_ki_bps: Some(1024), + ..CoreSettingsUpdate::default() + }) + .await + }); + wait_for_atomic_settings_commit(&core, "uploadLimitKiBps", "1024").await; + let second_core = core.clone(); + let second = tokio::spawn(async move { + second_core + .update_core_settings(CoreSettingsUpdate { + download_limit_ki_bps: Some(2048), + ..CoreSettingsUpdate::default() + }) + .await + }); + drop(state_barrier); + first.await.unwrap().unwrap(); + second.await.unwrap().unwrap(); + let committed = core.app_settings().await.unwrap(); + assert_eq!(committed.core.upload_limit_ki_bps, 1024); + assert_eq!(committed.core.download_limit_ki_bps, 2048); + assert_eq!(core.core_settings().await, committed.core); + assert_eq!( + core.ed2k_transfers.download_limit_bytes_per_sec().await, + 2048 * 1024 + ); +} + #[test] fn upload_queue_policy_uses_core_settings_for_slot_and_queue_limits() { let mut core_settings = default_core_settings(); diff --git a/crates/emulebb-metadata/src/profile_store.rs b/crates/emulebb-metadata/src/profile_store.rs index f4506a29..042bd723 100644 --- a/crates/emulebb-metadata/src/profile_store.rs +++ b/crates/emulebb-metadata/src/profile_store.rs @@ -1,7 +1,7 @@ -use std::path::Path; +use std::{collections::BTreeMap, path::Path}; use anyhow::{Result, ensure}; -use rusqlite::{OptionalExtension, params}; +use rusqlite::{OptionalExtension, TransactionBehavior, params}; use crate::{ profile_model::{MetadataCategory, MetadataFriend, MetadataServer}, @@ -113,6 +113,64 @@ impl super::MetadataStore { Ok(()) } + /// Validate and replace selected settings sections in one serialized transaction. + /// + /// The callback receives a coherent snapshot and must perform only synchronous + /// candidate computation, without calling back into this store. Unselected + /// and unchanged sections are preserved. A callback or write failure rolls + /// back every section; success is returned only after durable commit. + pub fn update_settings_sections( + &self, + sections: &[&str], + update: impl FnOnce(&mut BTreeMap>) -> Result, + ) -> Result { + ensure!(!sections.is_empty(), "settings sections must not be empty"); + let mut conn = self.connection()?; + // WHY: read/modify/write must serialize across independent connections, + // not only clones sharing this store's connection mutex. + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let mut rows = BTreeMap::new(); + { + let mut select = + tx.prepare("SELECT key, value_json FROM settings WHERE section = ?1 ORDER BY key")?; + for section in sections { + ensure!( + !section.trim().is_empty(), + "settings section must not be empty" + ); + let entries = select + .query_map([section], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::>>()?; + ensure!( + rows.insert((*section).to_owned(), entries).is_none(), + "duplicate settings section" + ); + } + } + let previous = rows.clone(); + let result = update(&mut rows)?; + ensure!( + rows.keys().eq(previous.keys()), + "settings update changed selected section names" + ); + let now = unix_ms(); + for (section, entries) in &rows { + if previous.get(section) == Some(entries) { + continue; + } + tx.execute("DELETE FROM settings WHERE section = ?1", [section])?; + for (key, value_json) in entries { + ensure!(!key.trim().is_empty(), "settings key must not be empty"); + tx.execute( + "INSERT INTO settings(section, key, value_json, updated_at_ms) VALUES (?1, ?2, ?3, ?4)", + params![section, key, value_json, now], + )?; + } + } + tx.commit()?; + Ok(result) + } + pub fn load_kad_bootstrap_endpoints(&self) -> Result> { let conn = self.connection()?; let mut stmt = conn.prepare( @@ -521,6 +579,105 @@ mod tests { use super::*; + #[test] + fn settings_transaction_rolls_back_an_earlier_section_after_later_write_failure() { + for fault in ["invalid-json", "empty-key", "duplicate-key"] { + let store = super::super::MetadataStore::in_memory().unwrap(); + store.put_setting_json("alpha", "value", "1").unwrap(); + store.put_setting_json("omega", "value", "2").unwrap(); + let result = store.update_settings_sections(&["alpha", "omega"], |sections| { + sections.insert("alpha".into(), vec![("value".into(), "3".into())]); + let invalid = match fault { + "invalid-json" => vec![("value".into(), "invalid-json".into())], + "empty-key" => vec![(String::new(), "4".into())], + _ => vec![("value".into(), "4".into()), ("value".into(), "5".into())], + }; + sections.insert("omega".into(), invalid); + Ok(()) + }); + assert!(result.is_err(), "accepted injected {fault} failure"); + assert_eq!( + store + .load_setting_json("alpha", "value") + .unwrap() + .as_deref(), + Some("1") + ); + assert_eq!( + store + .load_setting_json("omega", "value") + .unwrap() + .as_deref(), + Some("2") + ); + } + } + + #[test] + fn candidate_rejection_and_section_escape_leave_all_rows_unchanged() { + let store = super::super::MetadataStore::in_memory().unwrap(); + store.put_setting_json("selected", "value", "1").unwrap(); + let rejected: Result<()> = store.update_settings_sections(&["selected"], |sections| { + sections.insert("selected".into(), vec![("value".into(), "2".into())]); + anyhow::bail!("candidate rejected before writes") + }); + assert!(rejected.is_err()); + let escaped = store.update_settings_sections(&["selected"], |sections| { + sections.insert("unselected".into(), vec![("value".into(), "3".into())]); + Ok(()) + }); + assert!(escaped.is_err()); + assert_eq!( + store + .load_setting_json("selected", "value") + .unwrap() + .as_deref(), + Some("1") + ); + assert!( + store + .load_settings_section("unselected") + .unwrap() + .is_empty() + ); + } + + #[test] + fn independent_connections_serialize_settings_read_modify_write() { + let path = unique_metadata_path("atomic-settings-connections"); + let store = super::super::MetadataStore::open(&path).unwrap(); + store.put_setting_json("counter", "value", "0").unwrap(); + let start = std::sync::Arc::new(std::sync::Barrier::new(5)); + let mut workers = Vec::new(); + for _ in 0..4 { + let store = super::super::MetadataStore::open(&path).unwrap(); + let start = std::sync::Arc::clone(&start); + workers.push(std::thread::spawn(move || { + start.wait(); + store + .update_settings_sections(&["counter"], |sections| { + let entries = sections.get_mut("counter").unwrap(); + let value = entries[0].1.parse::().unwrap(); + std::thread::sleep(std::time::Duration::from_millis(5)); + entries[0].1 = (value + 1).to_string(); + Ok(()) + }) + .unwrap(); + })); + } + start.wait(); + for worker in workers { + worker.join().unwrap(); + } + assert_eq!( + store + .load_setting_json("counter", "value") + .unwrap() + .as_deref(), + Some("4") + ); + } + #[test] fn profile_state_roundtrips() { let store = super::super::MetadataStore::in_memory().unwrap(); diff --git a/crates/emulebb-rest/Cargo.toml b/crates/emulebb-rest/Cargo.toml index 8a633521..27e18283 100644 --- a/crates/emulebb-rest/Cargo.toml +++ b/crates/emulebb-rest/Cargo.toml @@ -30,6 +30,7 @@ tower.workspace = true tower-http.workspace = true [dev-dependencies] +rusqlite.workspace = true tokio = { workspace = true, features = ["macros", "rt"] } emulebb-ed2k = { path = "../emulebb-ed2k" } emulebb-index = { path = "../emulebb-index" } diff --git a/crates/emulebb-rest/src/handlers/app.rs b/crates/emulebb-rest/src/handlers/app.rs index 7f07730b..7577361e 100644 --- a/crates/emulebb-rest/src/handlers/app.rs +++ b/crates/emulebb-rest/src/handlers/app.rs @@ -231,6 +231,14 @@ pub(crate) async fn update_settings( }; match state.core.update_app_settings(request).await { Ok(settings) => api_ok(settings).into_response(), + Err(error) if emulebb_core::EmulebbCore::settings_update_was_committed(&error) => { + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "INTERNAL_ERROR", + error.to_string(), + ) + .into_response() + } Err(error) => { api_error(StatusCode::BAD_REQUEST, "BAD_REQUEST", error.to_string()).into_response() } diff --git a/crates/emulebb-rest/src/tests/route_app.rs b/crates/emulebb-rest/src/tests/route_app.rs index f1318ea6..600768fc 100644 --- a/crates/emulebb-rest/src/tests/route_app.rs +++ b/crates/emulebb-rest/src/tests/route_app.rs @@ -1,5 +1,148 @@ use crate::rest_test_support::*; +fn atomic_settings_router() -> ( + Router, + Arc, + std::path::PathBuf, + rusqlite::Connection, +) { + let root = unique_test_dir("atomic-settings"); + let database = root.join("metadata.sqlite"); + let core = Arc::new( + EmulebbCore::new( + "test", + FileIndex::open(&database).unwrap(), + root.join("transfers"), + ) + .unwrap(), + ); + let app = router( + Arc::clone(&core), + RestServerSettings { + api_key: "secret".to_owned(), + web_root_dir: None, + }, + ); + let connection = rusqlite::Connection::open(&database).unwrap(); + (app, core, database, connection) +} + +async fn atomic_settings_response(app: &Router, method: &str, body: Value) -> (StatusCode, Value) { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri("/api/v1/app/settings") + .header("X-API-Key", "secret") + .header("Content-Type", "application/json") + .body(if method == "GET" { + Body::empty() + } else { + Body::from(body.to_string()) + }) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) +} + +fn atomic_settings_rows(connection: &rusqlite::Connection) -> Vec<(String, String, String, i64)> { + connection + .prepare( + "SELECT section, key, value_json, updated_at_ms FROM settings ORDER BY section, key", + ) + .unwrap() + .query_map([], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) + }) + .unwrap() + .collect::>>() + .unwrap() +} + +#[tokio::test] +async fn settings_http_middleware_rejects_invalid_core_before_any_setting_changes() { + let (app, core, _, connection) = atomic_settings_router(); + let before = core.app_settings().await.unwrap(); + let rows = atomic_settings_rows(&connection); + let (status, response) = atomic_settings_response( + &app, + "PATCH", + json!({ + "nat": {"enabled": !before.nat.enabled}, "core": {"uploadLimitKiBps": 0}, + }), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(response["error"]["code"], "INVALID_ARGUMENT"); + assert_eq!(core.app_settings().await.unwrap(), before); + assert_eq!(core.core_settings().await, before.core); + assert_eq!(atomic_settings_rows(&connection), rows); +} + +#[tokio::test] +async fn valid_settings_http_patch_rolls_back_every_section_when_later_sql_write_fails() { + let (app, core, database, connection) = atomic_settings_router(); + let (_, before) = atomic_settings_response(&app, "GET", Value::Null).await; + let rows = atomic_settings_rows(&connection); + let old_core = core.core_settings().await; + let reload = !before["data"]["daemon"]["initialSharedDirectoryReload"] + .as_bool() + .unwrap(); + let nat = !before["data"]["nat"]["enabled"].as_bool().unwrap(); + let request = json!({ + "core": {"uploadLimitKiBps": 1024}, + "daemon": {"initialSharedDirectoryReload": reload}, "nat": {"enabled": nat}, + }); + // This request passes route validation. NAT is written after the earlier + // core/daemon sections, so a real SQLite failure proves transaction rollback. + connection.execute_batch( + "CREATE TRIGGER synthetic_settings_failure BEFORE INSERT ON settings + WHEN NEW.section = 'nat' BEGIN SELECT RAISE(ABORT, 'synthetic later settings failure'); END;", + ).unwrap(); + let (status, response) = atomic_settings_response(&app, "PATCH", request.clone()).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!( + response["error"]["message"] + .as_str() + .unwrap() + .contains("synthetic later settings failure") + ); + assert_eq!(atomic_settings_rows(&connection), rows); + assert_eq!(core.core_settings().await, old_core); + let (_, after) = atomic_settings_response(&app, "GET", Value::Null).await; + assert_eq!(after["data"], before["data"]); + + connection + .execute_batch("DROP TRIGGER synthetic_settings_failure") + .unwrap(); + let (status, committed) = atomic_settings_response(&app, "PATCH", request).await; + assert_eq!(status, StatusCode::OK); + let mut expected = before["data"].clone(); + expected["core"]["uploadLimitKiBps"] = json!(1024); + expected["daemon"]["initialSharedDirectoryReload"] = json!(reload); + expected["nat"]["enabled"] = json!(nat); + assert_eq!(committed["data"], expected); + assert_eq!(core.core_settings().await.upload_limit_ki_bps, 1024); + drop(app); + drop(core); + drop(connection); + let restored = EmulebbCore::new( + "test", + FileIndex::open(database).unwrap(), + unique_test_dir("atomic-settings-restart"), + ) + .unwrap() + .app_settings() + .await + .unwrap(); + assert_eq!(serde_json::to_value(restored).unwrap(), expected); +} + #[tokio::test] async fn event_stream_closes_without_replay_when_shutdown_is_already_requested() { let core = From be297b55cd721fc33ff208cd34030641c07b0dc5 Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Tue, 10 Nov 2026 15:16:53 +0100 Subject: [PATCH 2/3] test(settings): admit routed rollback fixture (RUST-BUG-184) Use a valid daemon hostname lookup update and prove identical HTTP admission before injecting the later SQLite write failure. Preserve transaction, runtime, and restart assertions without depending on internal SQL error text. Refs #299 --- crates/emulebb-rest/src/tests/route_app.rs | 41 ++++++++++++++++------ 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/crates/emulebb-rest/src/tests/route_app.rs b/crates/emulebb-rest/src/tests/route_app.rs index 600768fc..32d815fb 100644 --- a/crates/emulebb-rest/src/tests/route_app.rs +++ b/crates/emulebb-rest/src/tests/route_app.rs @@ -90,27 +90,48 @@ async fn valid_settings_http_patch_rolls_back_every_section_when_later_sql_write let (_, before) = atomic_settings_response(&app, "GET", Value::Null).await; let rows = atomic_settings_rows(&connection); let old_core = core.core_settings().await; - let reload = !before["data"]["daemon"]["initialSharedDirectoryReload"] - .as_bool() + let cache_ttl = before["data"]["daemon"]["hostnameLookup"]["cacheTtlSecs"] + .as_u64() + .unwrap() + .checked_add(1) .unwrap(); let nat = !before["data"]["nat"]["enabled"].as_bool().unwrap(); let request = json!({ "core": {"uploadLimitKiBps": 1024}, - "daemon": {"initialSharedDirectoryReload": reload}, "nat": {"enabled": nat}, + "daemon": {"hostnameLookup": {"cacheTtlSecs": cache_ttl}}, "nat": {"enabled": nat}, }); - // This request passes route validation. NAT is written after the earlier - // core/daemon sections, so a real SQLite failure proves transaction rollback. + assert!( + crate::route_body_metadata::validate_json_body_fields( + "PATCH", + "/api/v1/app/settings", + request.to_string().as_bytes(), + ) + .is_ok(), + "rollback fixture must pass the current request-body validator: {request}", + ); + // WHY: a rejected fixture never reaches SQLite. Prove identical HTTP + // admission separately, then isolate the later-section trigger as the fault. + let (control_app, _, _, _) = atomic_settings_router(); + let (control_status, control_response) = + atomic_settings_response(&control_app, "PATCH", request.clone()).await; + assert_eq!( + control_status, + StatusCode::OK, + "identical request without the trigger: {control_response}", + ); + // NAT is written after core/daemon; all their rows must roll back together. connection.execute_batch( "CREATE TRIGGER synthetic_settings_failure BEFORE INSERT ON settings WHEN NEW.section = 'nat' BEGIN SELECT RAISE(ABORT, 'synthetic later settings failure'); END;", ).unwrap(); let (status, response) = atomic_settings_response(&app, "PATCH", request.clone()).await; - assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(status, StatusCode::BAD_REQUEST, "{response}"); + assert_eq!(response["error"]["code"], "INVALID_ARGUMENT", "{response}"); assert!( response["error"]["message"] .as_str() - .unwrap() - .contains("synthetic later settings failure") + .is_some_and(|message| !message.trim().is_empty()), + "injected failure must retain the public error envelope: {response}", ); assert_eq!(atomic_settings_rows(&connection), rows); assert_eq!(core.core_settings().await, old_core); @@ -121,10 +142,10 @@ async fn valid_settings_http_patch_rolls_back_every_section_when_later_sql_write .execute_batch("DROP TRIGGER synthetic_settings_failure") .unwrap(); let (status, committed) = atomic_settings_response(&app, "PATCH", request).await; - assert_eq!(status, StatusCode::OK); + assert_eq!(status, StatusCode::OK, "{committed}"); let mut expected = before["data"].clone(); expected["core"]["uploadLimitKiBps"] = json!(1024); - expected["daemon"]["initialSharedDirectoryReload"] = json!(reload); + expected["daemon"]["hostnameLookup"]["cacheTtlSecs"] = json!(cache_ttl); expected["nat"]["enabled"] = json!(nat); assert_eq!(committed["data"], expected); assert_eq!(core.core_settings().await.upload_limit_ki_bps, 1024); From 2d08dddc6580a8b43c682e882ab06d95a5cb81fb Mon Sep 17 00:00:00 2001 From: Threepwood-7 Date: Wed, 11 Nov 2026 14:26:54 +0100 Subject: [PATCH 3/3] test(settings): keep private error tests after production items (RUST-BUG-184) --- crates/emulebb-core/src/app_api.rs | 44 +++++++++++++++--------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/crates/emulebb-core/src/app_api.rs b/crates/emulebb-core/src/app_api.rs index 8b4ea7f7..ac666c67 100644 --- a/crates/emulebb-core/src/app_api.rs +++ b/crates/emulebb-core/src/app_api.rs @@ -178,28 +178,6 @@ impl EmulebbCore { } } -#[cfg(test)] -mod settings_error_tests { - use super::*; - - #[test] - fn committed_refresh_classification_uses_the_type_not_the_message() { - let same_message = anyhow::anyhow!(CommittedSettingsRefreshFailure.to_string()); - assert!(!EmulebbCore::settings_update_was_committed(&same_message)); - let committed = Err::<(), _>(anyhow::anyhow!("synthetic runtime refresh error")) - .context(CommittedSettingsRefreshFailure) - .unwrap_err(); - assert!(EmulebbCore::settings_update_was_committed(&committed)); - assert_eq!( - committed.to_string(), - CommittedSettingsRefreshFailure.to_string() - ); - assert!(EmulebbCore::settings_update_was_committed( - &committed.context("outer context") - )); - } -} - fn validate_app_settings_update_paths(request: &mut AppSettingsUpdate) -> Result<()> { if let Some(daemon) = &mut request.daemon && let Some(emulebb_settings::NullableUpdate::Value(path)) = &mut daemon.incoming_dir @@ -241,3 +219,25 @@ fn validate_app_settings_update_paths(request: &mut AppSettingsUpdate) -> Result } Ok(()) } + +#[cfg(test)] +mod settings_error_tests { + use super::*; + + #[test] + fn committed_refresh_classification_uses_the_type_not_the_message() { + let same_message = anyhow::anyhow!(CommittedSettingsRefreshFailure.to_string()); + assert!(!EmulebbCore::settings_update_was_committed(&same_message)); + let committed = Err::<(), _>(anyhow::anyhow!("synthetic runtime refresh error")) + .context(CommittedSettingsRefreshFailure) + .unwrap_err(); + assert!(EmulebbCore::settings_update_was_committed(&committed)); + assert_eq!( + committed.to_string(), + CommittedSettingsRefreshFailure.to_string() + ); + assert!(EmulebbCore::settings_update_was_committed( + &committed.context("outer context") + )); + } +}