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..ac666c67 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 { @@ -187,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") + )); + } +} 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..32d815fb 100644 --- a/crates/emulebb-rest/src/tests/route_app.rs +++ b/crates/emulebb-rest/src/tests/route_app.rs @@ -1,5 +1,169 @@ 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 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": {"hostnameLookup": {"cacheTtlSecs": cache_ttl}}, "nat": {"enabled": nat}, + }); + 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, "{response}"); + assert_eq!(response["error"]["code"], "INVALID_ARGUMENT", "{response}"); + assert!( + response["error"]["message"] + .as_str() + .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); + 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, "{committed}"); + let mut expected = before["data"].clone(); + expected["core"]["uploadLimitKiBps"] = json!(1024); + 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); + 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 =