From 588e69ec0467d2eb49661abb18e8949bcd3dba5d Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Mon, 13 Jul 2026 16:07:30 -0500 Subject: [PATCH] Fix: Restore StormGuard rates when disabled --- docs-es/v2.0/stormguard-es.md | 2 + docs/v2.0/stormguard.md | 2 + src/rust/lqos_bakery/src/commands.rs | 54 ++ src/rust/lqos_bakery/src/lib.rs | 762 +++++++++++++++--- src/rust/lqos_network_devices/src/lib.rs | 23 + src/rust/lqos_network_devices/src/tests.rs | 30 +- src/rust/lqos_queue_tracker/src/lib.rs | 3 +- .../src/queue_structure/mod.rs | 2 +- .../queing_structure_json_monitor.rs | 72 +- src/rust/lqos_stormguard/README.md | 1 + .../lqos_stormguard/src/adaptive_actions.rs | 208 +++-- src/rust/lqos_stormguard/src/config.rs | 2 + src/rust/lqos_stormguard/src/lib.rs | 258 +++++- .../lqos_stormguard/src/queue_structure.rs | 2 + src/rust/lqos_stormguard/src/site_state.rs | 625 ++++++++++++-- 15 files changed, 1742 insertions(+), 304 deletions(-) diff --git a/docs-es/v2.0/stormguard-es.md b/docs-es/v2.0/stormguard-es.md index 7ffbbe20..1e288637 100644 --- a/docs-es/v2.0/stormguard-es.md +++ b/docs-es/v2.0/stormguard-es.md @@ -41,6 +41,8 @@ minimum_upload_percentage = 0.5 Si está probando, comience con `dry_run = true`. +Al deshabilitar StormGuard, o al volver a `dry_run = true` después de usarlo en modo activo, las colas administradas recuperan sus tasas garantizadas y límites máximos configurados, y se eliminan los ajustes adaptativos persistidos por StormGuard. Los ajustes administrados por el operador no se modifican. + ## UI y depuración - WebUI (Node Manager) incluye una pestaña dedicada de StormGuard además de las vistas de estado y depuración. diff --git a/docs/v2.0/stormguard.md b/docs/v2.0/stormguard.md index 2a285867..0a95c721 100644 --- a/docs/v2.0/stormguard.md +++ b/docs/v2.0/stormguard.md @@ -41,6 +41,8 @@ minimum_upload_percentage = 0.5 If you are testing, start with `dry_run = true` so you can observe decisions before allowing live limit changes. +Disabling StormGuard, or changing an active deployment back to `dry_run = true`, restores its managed queues to their configured rates and ceilings and removes StormGuard's persisted adaptive overrides. Operator-managed overrides are not changed. + ## UI and Debugging - WebUI provides a dedicated StormGuard dashboard tab plus status and debug views. diff --git a/src/rust/lqos_bakery/src/commands.rs b/src/rust/lqos_bakery/src/commands.rs index 14230ab1..0aa824cf 100644 --- a/src/rust/lqos_bakery/src/commands.rs +++ b/src/rust/lqos_bakery/src/commands.rs @@ -115,6 +115,19 @@ pub enum ExecutionMode { LiveUpdate, } +/// One HTB class that StormGuard must restore before leaving live mode. +#[derive(Debug, Clone, Allocative)] +pub struct StormGuardRestoreAdjustment { + /// Network interface containing the class. + pub interface_name: String, + /// Fully qualified HTB class identifier. + pub class_id: TcHandle, + /// Planned guaranteed class rate in Mbps. + pub planned_rate: u64, + /// Planned class ceiling in Mbps. + pub planned_ceil: u64, +} + /// List of commands that the Bakery system can handle. #[derive(Debug, Clone, Allocative)] pub enum BakeryCommands { @@ -255,6 +268,8 @@ pub enum BakeryCommands { }, /// Change a specific HTB class rate on-the-fly; optionally dry-run. StormGuardAdjustment { + /// Bakery shaping-tree generation used to resolve this class handle. + tree_generation: u64, /// If true, log the tc command instead of executing it. dry_run: bool, /// Network interface name (e.g., `eth0`) containing the class. @@ -263,6 +278,45 @@ pub enum BakeryCommands { class_id: String, /// New class ceiling rate in Mbps (the handler sets ceil and rate-1). new_rate: u64, + /// Original guaranteed class rate in Mbps. + planned_rate: u64, + /// Original class ceiling in Mbps. + planned_ceil: u64, + /// Optional synchronous completion reply. + #[allocative(skip)] + reply: Option>>, + }, + /// Restore planned HTB rates and clear retained StormGuard replay state after success. + ResetStormGuardAdjustments { + /// Bakery shaping-tree generation used to resolve the supplied class handles. + tree_generation: u64, + /// Complete set of StormGuard-managed HTB classes to restore. + adjustments: Vec, + /// Restore the supplied adjustments when Bakery has no in-process ownership cache. + restore_untracked: bool, + /// Optional synchronous completion reply. + #[allocative(skip)] + reply: Option>>, + }, + /// Discard retained StormGuard class ownership after a shaping-tree rebuild. + DiscardStormGuardAdjustments { + /// Bakery shaping-tree generation whose cached ownership may be discarded. + tree_generation: u64, + /// Optional synchronous completion reply. + #[allocative(skip)] + reply: Option>>, + }, + /// Apply a StormGuard circuit SQM update immediately and acknowledge the live result. + StormGuardCircuitAdjustment { + /// Bakery shaping-tree generation used to resolve this circuit. + tree_generation: u64, + /// Stable circuit hash in Bakery state. + circuit_hash: i64, + /// Requested per-circuit SQM override, or `None` to restore the configured default. + sqm_override: Option, + /// Optional synchronous completion reply. + #[allocative(skip)] + reply: Option>>, }, /// Runtime TreeGuard request to virtualize or restore a non-top-level site without a full reload. TreeGuardSetNodeVirtual { diff --git a/src/rust/lqos_bakery/src/lib.rs b/src/rust/lqos_bakery/src/lib.rs index 3771bbdf..ed5cd18c 100644 --- a/src/rust/lqos_bakery/src/lib.rs +++ b/src/rust/lqos_bakery/src/lib.rs @@ -58,7 +58,7 @@ pub use commands::{ BakeryCommands, RuntimeNodeOperationAction as BakeryRuntimeNodeOperationAction, RuntimeNodeOperationFailureReason as BakeryRuntimeNodeOperationFailureReason, RuntimeNodeOperationSnapshot as BakeryRuntimeNodeOperationSnapshot, - RuntimeNodeOperationStatus as BakeryRuntimeNodeOperationStatus, + RuntimeNodeOperationStatus as BakeryRuntimeNodeOperationStatus, StormGuardRestoreAdjustment, }; use lqos_bus::{ BusRequest, BusResponse, InsightLicenseSummary, LibreqosBusClient, TcHandle, UrgentSeverity, @@ -248,6 +248,22 @@ struct StormguardOverrideKey { class: TcHandle, } +#[derive(Clone, Copy, Debug)] +struct StormguardOverrideValue { + planned_rate: u64, + planned_ceil: u64, +} + +struct StormguardAdjustmentRequest { + tree_generation: u64, + dry_run: bool, + interface_name: String, + class_id: String, + new_rate: u64, + planned_rate: u64, + planned_ceil: u64, +} + #[derive(Clone, Debug)] struct VirtualizedSiteQdiscHandles { down: Option, @@ -1483,6 +1499,10 @@ fn circuits_with_pending_migration_targets( pub static ACTIVE_CIRCUITS: AtomicUsize = AtomicUsize::new(0); /// True while Bakery is applying a full reload batch to `tc`. static FULL_RELOAD_IN_PROGRESS: AtomicBool = AtomicBool::new(false); +/// Monotonic count of committed Bakery shaping-tree changes. +static STORMGUARD_TREE_GENERATION: AtomicU64 = AtomicU64::new(0); +/// Latest generation produced by a successful full tree rebuild. +static STORMGUARD_TREE_REBUILD_GENERATION: AtomicU64 = AtomicU64::new(0); #[cfg(test)] pub(crate) fn test_state_lock() -> &'static std::sync::Mutex<()> { static LOCK: OnceLock> = OnceLock::new(); @@ -3705,6 +3725,16 @@ pub fn full_reload_in_progress() -> bool { FULL_RELOAD_IN_PROGRESS.load(Ordering::Relaxed) } +/// Returns the Bakery shaping-tree generation observed by StormGuard. +pub fn stormguard_tree_generation() -> u64 { + STORMGUARD_TREE_GENERATION.load(Ordering::Acquire) +} + +/// Returns the latest generation that replaced class handles with a full tree rebuild. +pub fn stormguard_tree_rebuild_generation() -> u64 { + STORMGUARD_TREE_REBUILD_GENERATION.load(Ordering::Acquire) +} + #[derive(Debug, Default, Clone, Copy)] struct MappedLimitStats { enforced_limit: Option, @@ -4040,8 +4070,9 @@ fn bakery_main(rx: Receiver, tx: Sender) { .ok() .map(|config| QdiscHandleState::load(&config)) .unwrap_or_default(); - // Persist latest StormGuard ceilings keyed by interface + class so we can replay after rebuilds. - let mut stormguard_overrides: HashMap = HashMap::new(); + // Retain each StormGuard-owned class's original plan until disable/reset reconciliation. + let mut stormguard_overrides: HashMap = + HashMap::new(); let mut virtualized_sites: HashMap = HashMap::new(); let mut runtime_node_operations: HashMap = HashMap::new(); let mut next_runtime_operation_id: u64 = 1; @@ -5088,78 +5119,92 @@ fn bakery_main(rx: Receiver, tx: Sender) { ); } BakeryCommands::StormGuardAdjustment { + tree_generation, dry_run, interface_name, class_id, new_rate, + planned_rate, + planned_ceil, + reply, } => { - let has_mq_run = MQ_CREATED.load(Relaxed); - if !has_mq_run { - debug!("StormGuardAdjustment received before MQ setup, skipping."); - continue; + let result = apply_stormguard_adjustment( + StormguardAdjustmentRequest { + tree_generation, + dry_run, + interface_name, + class_id, + new_rate, + planned_rate, + planned_ceil, + }, + &mut stormguard_overrides, + ); + if let Some(reply) = reply { + let _ = reply.send(result); } - let Ok(config) = lqos_config::load_config() else { - error!("Failed to load configuration, skipping StormGuardAdjustment."); - continue; - }; - let Ok(tc_handle) = TcHandle::from_string(&class_id) else { - warn!( - "StormGuardAdjustment has invalid class_id [{}], skipping.", - class_id - ); - continue; - }; - if !dry_run { - let key = StormguardOverrideKey { - interface: interface_name.to_string(), - class: tc_handle, - }; - stormguard_overrides.insert(key, new_rate); - } - if let Some(reason) = live_tree_mutation_blocker_for_config(&config) { - info!( - "Skipping StormGuard live class change for {} {} because {}.", - interface_name, class_id, reason - ); - continue; - } - let normalized_class = tc_handle.as_tc_string(); - // Build the HTB command - let args = vec![ - "class".to_string(), - "replace".to_string(), - "dev".to_string(), - interface_name.to_string(), - "classid".to_string(), - normalized_class.clone(), - "htb".to_string(), - "rate".to_string(), - format!("{}mbit", new_rate.saturating_sub(1)), - "ceil".to_string(), - format!("{}mbit", new_rate), - ]; - if dry_run { - info!("DRY RUN: /sbin/tc {}", args.join(" ")); + } + BakeryCommands::ResetStormGuardAdjustments { + tree_generation, + adjustments, + restore_untracked, + reply, + } => { + let current_generation = stormguard_tree_generation(); + let result = if tree_generation != current_generation { + Err(format!( + "StormGuard reset used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}" + )) + } else if batch.is_some() { + Err("Bakery queue rebuild batch is still open".to_string()) } else { - let output = std::process::Command::new("/sbin/tc").args(&args).output(); - match output { - Err(e) => { - warn!("Failed to run tc command: {}", e); - } - Ok(out) => { - if !out.status.success() { - warn!( - "tc command failed: {}", - String::from_utf8_lossy(&out.stderr) - ); - } else { - debug!( - "tc command succeeded: {}", - String::from_utf8_lossy(&out.stdout) - ); - } - } - } + reset_stormguard_adjustments( + &adjustments, + restore_untracked, + &mut stormguard_overrides, + ) + }; + if let Some(reply) = reply { + let _ = reply.send(result); + } + } + BakeryCommands::DiscardStormGuardAdjustments { + tree_generation, + reply, + } => { + let current_generation = stormguard_tree_generation(); + let result = if tree_generation != current_generation { + Err(format!( + "StormGuard discard used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}" + )) + } else if batch.is_some() { + Err("Bakery queue rebuild batch is still open".to_string()) + } else { + stormguard_overrides.clear(); + Ok(()) + }; + if let Some(reply) = reply { + let _ = reply.send(result); + } + } + BakeryCommands::StormGuardCircuitAdjustment { + tree_generation, + circuit_hash, + sqm_override, + reply, + } => { + let current_generation = stormguard_tree_generation(); + let result = if tree_generation != current_generation { + Err(format!( + "StormGuard circuit adjustment used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}" + )) + } else if batch.is_some() { + Err("Bakery queue rebuild batch is still open".to_string()) + } else { + apply_stormguard_circuit_adjustment(circuit_hash, sqm_override, &mut circuits) + }; + if let Some(reply) = reply { + let _ = reply.send(result); } } BakeryCommands::TreeGuardSetNodeVirtual { @@ -5201,7 +5246,7 @@ fn handle_commit_batch( qdisc_handles: &mut QdiscHandleState, tx: &Sender, migrations: &mut HashMap, - stormguard_overrides: &HashMap, + stormguard_overrides: &HashMap, virtualized_sites: &mut HashMap, runtime_node_operations: &mut HashMap, ) { @@ -5271,7 +5316,6 @@ fn handle_commit_batch( &config, new_batch, resolved_mq_layout, - stormguard_overrides, virtualized_sites, runtime_node_operations, summary, @@ -5318,7 +5362,6 @@ fn handle_commit_batch( &config, new_batch, resolved_mq_layout, - stormguard_overrides, virtualized_sites, runtime_node_operations, summary, @@ -5364,7 +5407,6 @@ fn handle_commit_batch( &config, new_batch, resolved_mq_layout, - stormguard_overrides, virtualized_sites, runtime_node_operations, summary, @@ -5414,7 +5456,6 @@ fn handle_commit_batch( &config, new_batch, resolved_mq_layout, - stormguard_overrides, virtualized_sites, runtime_node_operations, summary.clone(), @@ -5483,7 +5524,40 @@ fn handle_commit_batch( &config, new_batch, resolved_mq_layout, - stormguard_overrides, + virtualized_sites, + runtime_node_operations, + summary, + ); + return; + } + + if stormguard_site_changes_intersect_ownership(&site_change_mode, &config, stormguard_overrides) + { + let summary = "Bakery full reload triggered to keep StormGuard class ownership synchronized with a changed shaping plan.".to_string(); + let (new_batch, mapped_limit_stats) = filter_batch_by_mapped_circuit_limit( + raw_batch.clone(), + &baseline_circuits, + effective_limit, + ); + log_mapped_limit_decision( + "StormGuard-consistent rebuild", + mapped_limit, + mapped_limit_stats, + ); + if mapped_limit_stats.dropped_mapped > 0 { + maybe_emit_mapped_circuit_limit_urgent(&mapped_limit_stats); + } + announce_full_reload(&summary); + full_reload( + batch, + sites, + circuits, + live_circuits, + mq_layout, + qdisc_handles, + &config, + new_batch, + resolved_mq_layout, virtualized_sites, runtime_node_operations, summary, @@ -5536,7 +5610,6 @@ fn handle_commit_batch( &config, raw_batch.clone(), resolved_mq_layout.clone(), - stormguard_overrides, virtualized_sites, runtime_node_operations, summary, @@ -5544,6 +5617,12 @@ fn handle_commit_batch( return; // Skip the rest of this CommitBatch processing } } + if stormguard_live_enabled(&config) { + // Publish the generation only after every live speed command is in this same channel. + // Previously queued StormGuard work retains the old generation and will be rejected; + // work created for the new generation is queued behind the new site plan. + STORMGUARD_TREE_GENERATION.fetch_add(1, Ordering::Release); + } } // Now we can process circuit changes incrementally @@ -5830,7 +5909,6 @@ fn handle_commit_batch( &config, raw_batch.clone(), resolved_mq_layout.clone(), - stormguard_overrides, virtualized_sites, runtime_node_operations, summary, @@ -6179,6 +6257,29 @@ fn handle_tick( execute_and_record_live_change(&commands, "pruning lazy queues"); } +fn stormguard_site_changes_intersect_ownership( + site_change_mode: &SiteDiffResult, + config: &Arc, + overrides: &HashMap, +) -> bool { + let SiteDiffResult::SpeedChanges { changes } = site_change_mode else { + return false; + }; + + changes.iter().any(|change| { + let Some((down_class, up_class)) = site_class_handles(change) else { + return false; + }; + overrides.contains_key(&StormguardOverrideKey { + interface: config.isp_interface(), + class: down_class, + }) || overrides.contains_key(&StormguardOverrideKey { + interface: config.internet_interface(), + class: up_class, + }) + }) +} + fn handle_change_site_speed_live( site_hash: i64, download_bandwidth_min: f32, @@ -6295,7 +6396,14 @@ fn handle_change_site_speed_live( ), ], ]; - execute_and_record_live_change(&commands, "changing site speed live"); + let result = execute_and_record_live_change(&commands, "changing site speed live"); + if !result.ok { + mark_reload_required(format!( + "Bakery live site speed update failed: {}", + summarize_apply_result("changing site speed live", &result) + )); + return; + } // Update the site speeds in the site map - create a new Arc with updated values let new_site = Arc::new(BakeryCommands::AddSite { site_hash, @@ -10913,7 +11021,6 @@ fn full_reload( config: &Arc, new_batch: Vec>, resolved_mq_layout: Option, - stormguard_overrides: &HashMap, virtualized_sites: &mut HashMap, runtime_node_operations: &mut HashMap, trigger_summary: String, @@ -11045,7 +11152,13 @@ fn full_reload( "A successful Bakery full reload re-established baseline state; incremental topology mutations can resume.", ); FIRST_COMMIT_APPLIED.store(true, Ordering::Relaxed); - apply_stormguard_overrides(stormguard_overrides, config); + let rebuild_generation = STORMGUARD_TREE_GENERATION + .load(Ordering::Relaxed) + .wrapping_add(1); + STORMGUARD_TREE_REBUILD_GENERATION.store(rebuild_generation, Ordering::Release); + STORMGUARD_TREE_GENERATION.store(rebuild_generation, Ordering::Release); + // StormGuard replays persisted adjustments after the queue tracker publishes the rebuilt + // structure. Replaying cached class handles here could target reassigned handles. } else { *sites = previous_sites; *circuits = previous_circuits; @@ -11185,56 +11298,483 @@ fn process_batch( } } -fn apply_stormguard_overrides( - overrides: &HashMap, - config: &Arc, +fn stormguard_live_enabled(config: &Config) -> bool { + config + .stormguard + .as_ref() + .is_some_and(|stormguard| stormguard.enabled && !stormguard.dry_run) +} + +fn record_stormguard_override( + overrides: &mut HashMap, + key: StormguardOverrideKey, + planned_rate: u64, + planned_ceil: u64, ) { - if config.queues.queue_mode.is_observe() { - push_bakery_event( - "stormguard_override_replay_skipped", - "info", - "Skipping StormGuard HTB override replay because queue_mode is observe.".to_string(), + overrides.entry(key).or_insert(StormguardOverrideValue { + planned_rate, + planned_ceil, + }); +} + +fn apply_stormguard_adjustment( + request: StormguardAdjustmentRequest, + overrides: &mut HashMap, +) -> Result<(), String> { + let current_generation = stormguard_tree_generation(); + if request.tree_generation != current_generation { + return Err(format!( + "StormGuard class change used shaping-tree generation {}, but Bakery is at generation {current_generation}", + request.tree_generation + )); + } + if !MQ_CREATED.load(Relaxed) { + return Err("StormGuard class change received before MQ setup".to_string()); + } + let config = lqos_config::load_config().map_err(|error| { + format!("failed to load configuration for StormGuard class change: {error}") + })?; + let tc_handle = TcHandle::from_string(&request.class_id) + .map_err(|error| format!("invalid StormGuard class ID {}: {error}", request.class_id))?; + if !request.dry_run && !stormguard_live_enabled(&config) { + return Err("StormGuard is disabled or in dry-run mode".to_string()); + } + if let Some(reason) = live_tree_mutation_blocker_for_config(&config) { + return Err(format!( + "StormGuard live class change is blocked because {reason}" + )); + } + + let args = stormguard_htb_command( + &request.interface_name, + tc_handle, + request.new_rate.saturating_sub(1), + request.new_rate, + ); + if request.dry_run { + info!("DRY RUN: /sbin/tc {}", args.join(" ")); + return Ok(()); + } + + let output = std::process::Command::new("/sbin/tc") + .args(&args) + .output() + .map_err(|error| format!("failed to run StormGuard tc command: {error}"))?; + if !output.status.success() { + return Err(format!( + "StormGuard tc command failed: {}", + String::from_utf8_lossy(&output.stderr) + )); + } + record_stormguard_override( + overrides, + StormguardOverrideKey { + interface: request.interface_name, + class: tc_handle, + }, + request.planned_rate, + request.planned_ceil, + ); + debug!( + "StormGuard tc command succeeded: {}", + String::from_utf8_lossy(&output.stdout) + ); + Ok(()) +} + +fn stormguard_htb_command( + interface_name: &str, + class_id: TcHandle, + rate: u64, + ceil: u64, +) -> Vec { + vec![ + "class".to_string(), + "replace".to_string(), + "dev".to_string(), + interface_name.to_string(), + "classid".to_string(), + class_id.as_tc_string(), + "htb".to_string(), + "rate".to_string(), + format!("{}mbit", rate), + "ceil".to_string(), + format!("{}mbit", ceil), + ] +} + +fn stormguard_restorations( + adjustments: &[StormGuardRestoreAdjustment], + restore_untracked: bool, + overrides: &HashMap, +) -> Vec { + let mut restorations: HashMap = + HashMap::new(); + if restore_untracked { + for adjustment in adjustments { + restorations.insert( + StormguardOverrideKey { + interface: adjustment.interface_name.clone(), + class: adjustment.class_id, + }, + adjustment.clone(), + ); + } + } + for (key, value) in overrides { + restorations.insert( + key.clone(), + StormGuardRestoreAdjustment { + interface_name: key.interface.clone(), + class_id: key.class, + planned_rate: value.planned_rate, + planned_ceil: value.planned_ceil, + }, ); - return; } - if overrides.is_empty() { - return; + restorations.into_values().collect() +} + +fn reset_stormguard_adjustments( + adjustments: &[StormGuardRestoreAdjustment], + restore_untracked: bool, + overrides: &mut HashMap, +) -> Result<(), String> { + if overrides.is_empty() && (!restore_untracked || adjustments.is_empty()) { + overrides.clear(); + debug!("Cleared inactive Bakery StormGuard replay state."); + return Ok(()); } + if !MQ_CREATED.load(Relaxed) { + return Err("Bakery MQ setup has not completed".to_string()); + } + let config = lqos_config::load_config() + .map_err(|error| format!("failed to load configuration: {error}"))?; + if let Some(reason) = live_tree_mutation_blocker_for_config(&config) { + return Err(format!("live queue mutation is blocked because {reason}")); + } + + let restorations = stormguard_restorations(adjustments, restore_untracked, overrides); + invalidate_live_tc_snapshots(); + let mut live_classes: HashMap> = HashMap::new(); let mut commands = Vec::new(); - for (key, rate) in overrides.iter() { - commands.push(vec![ - "class".to_string(), - "replace".to_string(), - "dev".to_string(), - key.interface.clone(), - "classid".to_string(), - key.class.as_tc_string(), - "htb".to_string(), - "rate".to_string(), - format!("{}mbit", rate.saturating_sub(1)), - "ceil".to_string(), - format!("{}mbit", rate), - ]); + for restoration in restorations { + if !live_classes.contains_key(&restoration.interface_name) { + let snapshot = read_live_class_snapshot(&restoration.interface_name)?; + live_classes.insert( + restoration.interface_name.clone(), + snapshot.into_keys().collect(), + ); + } + if live_classes + .get(&restoration.interface_name) + .is_some_and(|classes| classes.contains(&restoration.class_id)) + { + commands.push(stormguard_htb_command( + &restoration.interface_name, + restoration.class_id, + restoration.planned_rate, + restoration.planned_ceil, + )); + } else { + debug!( + "StormGuard restore skipped absent class {} on {}.", + restoration.class_id.as_tc_string(), + restoration.interface_name + ); + } } - let result = execute_in_memory(&commands, "replaying StormGuard overrides"); + if !commands.is_empty() { + let result = execute_in_memory(&commands, "restoring StormGuard planned rates"); + if !result.ok { + return Err(summarize_apply_result( + "restoring StormGuard planned rates", + &result, + )); + } + } + + overrides.clear(); + debug!("Restored planned StormGuard rates and cleared retained adjustments."); + Ok(()) +} + +fn apply_stormguard_circuit_adjustment( + circuit_hash: i64, + sqm_override: Option, + circuits: &mut HashMap>, +) -> Result { + let Some(existing) = circuits.get(&circuit_hash) else { + return Ok(false); + }; + let command = circuit_with_sqm_override(existing.as_ref(), sqm_override)?; + if !MQ_CREATED.load(Relaxed) { + return Err("Bakery MQ setup has not completed".to_string()); + } + let config = lqos_config::load_config() + .map_err(|error| format!("failed to load configuration: {error}"))?; + if let Some(reason) = live_tree_mutation_blocker_for_config(&config) { + return Err(format!("live queue mutation is blocked because {reason}")); + } + let commands = add_commands_for_circuit(&command, &config, ExecutionMode::Builder) + .filter(|commands| !commands.is_empty()) + .ok_or_else(|| "StormGuard circuit adjustment produced no live tc commands".to_string())?; + let result = execute_in_memory(&commands, "applying StormGuard circuit SQM adjustment"); if !result.ok { - push_bakery_event( - "stormguard_override_replay_failed", - "error", - summarize_apply_result("replaying StormGuard overrides", &result), - ); + return Err(summarize_apply_result( + "applying StormGuard circuit SQM adjustment", + &result, + )); } + circuits.insert(circuit_hash, Arc::new(command)); + Ok(true) +} + +fn circuit_with_sqm_override( + existing: &BakeryCommands, + sqm_override: Option, +) -> Result { + let mut command = existing.clone(); + let BakeryCommands::AddCircuit { + sqm_override: current, + .. + } = &mut command + else { + return Err("existing Bakery circuit state is not AddCircuit".to_string()); + }; + *current = sqm_override; + Ok(command) } #[cfg(test)] mod tests { use super::*; - use lqos_config::Config; + use lqos_config::{Config, StormguardConfig}; use std::sync::Mutex; use std::time::{SystemTime, UNIX_EPOCH}; static TC_BYPASS_SETTER_CALLS: Mutex> = Mutex::new(Vec::new()); + #[test] + fn stormguard_live_requires_enabled_non_dry_run_config() { + let mut config = Config::default(); + assert!(!stormguard_live_enabled(&config)); + + config.stormguard = Some(StormguardConfig { + enabled: true, + dry_run: true, + ..StormguardConfig::default() + }); + assert!(!stormguard_live_enabled(&config)); + + if let Some(stormguard) = &mut config.stormguard { + stormguard.dry_run = false; + } + assert!(stormguard_live_enabled(&config)); + } + + #[test] + fn stormguard_restore_command_uses_planned_rate() -> anyhow::Result<()> { + let command = stormguard_htb_command("eth0", TcHandle::from_string("1:2")?, 4, 10); + assert_eq!( + command, + [ + "class", "replace", "dev", "eth0", "classid", "0x1:0x2", "htb", "rate", "4mbit", + "ceil", "10mbit" + ] + ); + Ok(()) + } + + #[test] + fn stormguard_override_retains_original_rate_pair() -> anyhow::Result<()> { + let key = StormguardOverrideKey { + interface: "eth0".to_string(), + class: TcHandle::from_string("1:2")?, + }; + let mut overrides = HashMap::new(); + record_stormguard_override(&mut overrides, key.clone(), 4, 10); + record_stormguard_override(&mut overrides, key.clone(), 99, 100); + let Some(value) = overrides.get(&key) else { + anyhow::bail!("missing StormGuard override"); + }; + assert_eq!(value.planned_rate, 4); + assert_eq!(value.planned_ceil, 10); + Ok(()) + } + + #[test] + fn stormguard_rejects_stale_tree_generation_before_live_mutation() { + let current_generation = stormguard_tree_generation(); + let result = apply_stormguard_adjustment( + StormguardAdjustmentRequest { + tree_generation: current_generation.wrapping_add(1), + dry_run: false, + interface_name: "eth0".to_string(), + class_id: "1:2".to_string(), + new_rate: 5, + planned_rate: 10, + planned_ceil: 10, + }, + &mut HashMap::new(), + ); + + let error = result.expect_err("stale generation must be rejected"); + assert!(error.contains("generation")); + } + + #[test] + fn stormguard_circuit_adjustment_reports_missing_bakery_state() { + let result = + apply_stormguard_circuit_adjustment(42, Some("cake".to_string()), &mut HashMap::new()); + assert_eq!(result, Ok(false)); + } + + #[test] + fn stormguard_rebuild_gate_only_matches_changed_owned_site() { + let config = Arc::new(Config { + bridge: Some(lqos_config::BridgeConfig { + use_xdp_bridge: false, + to_internet: "internet0".to_string(), + to_network: "isp0".to_string(), + mtu: None, + }), + ..Config::default() + }); + let changed_site = mk_add_site(42, 0x1_0000, 0x2_0000, 0x10); + let change_mode = SiteDiffResult::SpeedChanges { + changes: vec![changed_site.as_ref().clone()], + }; + let (down_class, up_class) = + site_class_handles(changed_site.as_ref()).expect("site class handles"); + let mut overrides = HashMap::new(); + overrides.insert( + StormguardOverrideKey { + interface: config.isp_interface(), + class: down_class, + }, + StormguardOverrideValue { + planned_rate: 10, + planned_ceil: 100, + }, + ); + + assert!(stormguard_site_changes_intersect_ownership( + &change_mode, + &config, + &overrides + )); + + overrides.clear(); + overrides.insert( + StormguardOverrideKey { + interface: config.internet_interface(), + class: up_class, + }, + StormguardOverrideValue { + planned_rate: 10, + planned_ceil: 100, + }, + ); + assert!(stormguard_site_changes_intersect_ownership( + &change_mode, + &config, + &overrides + )); + + overrides.clear(); + overrides.insert( + StormguardOverrideKey { + interface: config.isp_interface(), + class: TcHandle::from_u32(0x9_0009), + }, + StormguardOverrideValue { + planned_rate: 10, + planned_ceil: 100, + }, + ); + assert!(!stormguard_site_changes_intersect_ownership( + &change_mode, + &config, + &overrides + )); + } + + #[test] + fn stormguard_reset_prefers_successful_bakery_ownership() -> anyhow::Result<()> { + let class = TcHandle::from_string("1:2")?; + let supplied = [ + StormGuardRestoreAdjustment { + interface_name: "eth0".to_string(), + class_id: class, + planned_rate: 99, + planned_ceil: 100, + }, + StormGuardRestoreAdjustment { + interface_name: "eth0".to_string(), + class_id: TcHandle::from_string("1:3")?, + planned_rate: 7, + planned_ceil: 20, + }, + ]; + let key = StormguardOverrideKey { + interface: "eth0".to_string(), + class, + }; + let mut overrides = HashMap::new(); + record_stormguard_override(&mut overrides, key, 4, 10); + + let restorations = stormguard_restorations(&supplied, true, &overrides); + assert_eq!(restorations.len(), 2); + let cached = restorations + .iter() + .find(|adjustment| adjustment.class_id == class) + .ok_or_else(|| anyhow::anyhow!("missing cached restoration"))?; + assert_eq!(cached.planned_rate, 4); + assert_eq!(cached.planned_ceil, 10); + + let startup = stormguard_restorations(&supplied, true, &HashMap::new()); + assert_eq!(startup.len(), 2); + assert!(stormguard_restorations(&supplied, false, &HashMap::new()).is_empty()); + Ok(()) + } + + #[test] + fn stormguard_circuit_sqm_change_preserves_qdisc_handles() -> anyhow::Result<()> { + let existing = BakeryCommands::AddCircuit { + circuit_hash: 42, + circuit_name: Some("Circuit A".to_string()), + site_name: Some("Site A".to_string()), + parent_class_id: TcHandle::from_string("1:1")?, + up_parent_class_id: TcHandle::from_string("2:1")?, + class_minor: 3, + download_bandwidth_min: 5.0, + upload_bandwidth_min: 2.0, + download_bandwidth_max: 10.0, + upload_bandwidth_max: 4.0, + class_major: 1, + up_class_major: 2, + down_qdisc_handle: Some(0x9000), + up_qdisc_handle: Some(0x9001), + ip_addresses: "192.0.2.1".to_string(), + sqm_override: Some("cake".to_string()), + }; + let updated = circuit_with_sqm_override(&existing, None).map_err(anyhow::Error::msg)?; + let BakeryCommands::AddCircuit { + down_qdisc_handle, + up_qdisc_handle, + sqm_override, + .. + } = updated + else { + anyhow::bail!("expected AddCircuit"); + }; + assert_eq!(down_qdisc_handle, Some(0x9000)); + assert_eq!(up_qdisc_handle, Some(0x9001)); + assert_eq!(sqm_override, None); + Ok(()) + } + fn recording_tc_bypass_setter(enabled: bool) -> anyhow::Result<()> { TC_BYPASS_SETTER_CALLS .lock() diff --git a/src/rust/lqos_network_devices/src/lib.rs b/src/rust/lqos_network_devices/src/lib.rs index 6dac496b..06e8faea 100644 --- a/src/rust/lqos_network_devices/src/lib.rs +++ b/src/rust/lqos_network_devices/src/lib.rs @@ -23,6 +23,7 @@ use anyhow::{Context, Result}; use lqos_config::{ Config, ConfigShapedDevices, NetworkJson, TopologyShapingInputsFile, }; +use lqos_topology_compile::TopologyImportFile; use std::sync::Arc; use tracing::debug; @@ -152,6 +153,16 @@ fn load_shaped_devices_for_config(config: &lqos_config::Config) -> Result Result { + if lqos_config::integration_ingress_enabled(config) { + return Ok(TopologyImportFile::load(config)? + .map(|topology_import| topology_import.into_imported_bundle().shaped_devices) + .unwrap_or_default()); + } + + ConfigShapedDevices::load_for_config(config).context("Unable to load ShapedDevices.csv") +} + /// Loads shaped-device configuration from disk. /// /// When integration-backed topology ingress is enabled, this prefers ready runtime shaping inputs, @@ -161,6 +172,18 @@ pub fn load_shaped_devices() -> Result { Ok(load_shaped_devices_with_source()?.shaped) } +/// Loads the operator or integration source shaped-device rows without runtime overrides. +/// +/// Integration-ingress mode reads `topology_import.json`; manual mode reads +/// `ShapedDevices.csv`. This deliberately bypasses ready runtime shaping inputs so callers can +/// detect source settings hidden by an effective adaptive override. +/// +/// Side effects: reads the configured source file from disk. +pub fn load_source_shaped_devices() -> Result { + let config = lqos_config::load_config().context("Unable to load /etc/lqos.conf")?; + load_source_shaped_devices_for_config(config.as_ref()) +} + pub(crate) fn load_shaped_devices_with_source() -> Result { let config = lqos_config::load_config().context("Unable to load /etc/lqos.conf")?; load_shaped_devices_for_config(config.as_ref()) diff --git a/src/rust/lqos_network_devices/src/tests.rs b/src/rust/lqos_network_devices/src/tests.rs index 3c1f5df0..480ccbac 100644 --- a/src/rust/lqos_network_devices/src/tests.rs +++ b/src/rust/lqos_network_devices/src/tests.rs @@ -1,6 +1,7 @@ use crate::{ DynamicCircuit, ShapedDevicesCatalog, load_shaped_devices_for_config, - resolve_parent_node_reference, runtime_inputs, with_network_json_write, + load_source_shaped_devices_for_config, resolve_parent_node_reference, runtime_inputs, + with_network_json_write, }; use lqos_config::{ CircuitAnchorsFile, Config, ConfigShapedDevices, NetworkJsonNode, ShapedDevice, @@ -367,7 +368,7 @@ fn runtime_inputs_leave_runtime_fallback_circuits_unparented() { } #[test] -fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() { +fn source_loader_bypasses_ready_runtime_inputs() { let _guard = TEST_LOCK.lock(); let lqos_directory = unique_temp_dir("lqos-network-devices-runtime-empty"); @@ -384,10 +385,24 @@ fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() { "csv-circuit", "192.0.2.10/32", ); + let runtime_inputs = TopologyShapingInputsFile { + circuits: vec![TopologyShapingCircuitInput { + circuit_id: "runtime-circuit".to_string(), + circuit_name: "Runtime Circuit".to_string(), + effective_parent_node_name: "Runtime Parent".to_string(), + effective_parent_node_id: "runtime-parent".to_string(), + devices: vec![TopologyShapingDeviceInput { + device_id: "runtime-device".to_string(), + ..TopologyShapingDeviceInput::default() + }], + ..TopologyShapingCircuitInput::default() + }], + shaping_generation: "shape-1".to_string(), + ..TopologyShapingInputsFile::default() + }; std::fs::write( &runtime_path, - serde_json::to_string_pretty(&TopologyShapingInputsFile::default()) - .expect("empty shaping inputs should encode"), + serde_json::to_string_pretty(&runtime_inputs).expect("runtime shaping inputs should encode"), ) .expect("runtime shaping inputs should write"); let mut import_devices = ConfigShapedDevices::default(); @@ -432,7 +447,12 @@ fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() { let loaded = load_shaped_devices_for_config(&config) .expect("preferred shaped-device source should load"); assert_eq!(loaded.shaped.devices.len(), 1); - assert_eq!(loaded.shaped.devices[0].circuit_id, "import-circuit"); + assert_eq!(loaded.shaped.devices[0].circuit_id, "runtime-circuit"); + + let source = + load_source_shaped_devices_for_config(&config).expect("raw integration source should load"); + assert_eq!(source.devices.len(), 1); + assert_eq!(source.devices[0].circuit_id, "import-circuit"); } #[test] diff --git a/src/rust/lqos_queue_tracker/src/lib.rs b/src/rust/lqos_queue_tracker/src/lib.rs index a75734da..fb194a54 100644 --- a/src/rust/lqos_queue_tracker/src/lib.rs +++ b/src/rust/lqos_queue_tracker/src/lib.rs @@ -23,7 +23,8 @@ pub use interval::set_queue_refresh_interval; #[doc(hidden)] pub use queue_structure::{ EFFECTIVE_CIRCUIT_RATES, EFFECTIVE_NODE_RATES, QUEUE_STRUCTURE, - QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueNode, QueueStructure, spawn_queue_structure_monitor, + QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueNode, QueueStructure, reload_queue_structure, + spawn_queue_structure_monitor, }; pub use queue_types::deserialize_tc_tree; // Exported for the benchmarker pub use tracking::spawn_queue_monitor; diff --git a/src/rust/lqos_queue_tracker/src/queue_structure/mod.rs b/src/rust/lqos_queue_tracker/src/queue_structure/mod.rs index 65008b7e..3e5ad62f 100644 --- a/src/rust/lqos_queue_tracker/src/queue_structure/mod.rs +++ b/src/rust/lqos_queue_tracker/src/queue_structure/mod.rs @@ -1,12 +1,12 @@ mod queing_structure_json_monitor; mod queue_network; mod queue_node; -pub use queing_structure_json_monitor::spawn_queue_structure_monitor; #[doc(hidden)] pub use queing_structure_json_monitor::{ EFFECTIVE_CIRCUIT_RATES, EFFECTIVE_NODE_RATES, QUEUE_STRUCTURE, QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueStructure, }; +pub use queing_structure_json_monitor::{reload_queue_structure, spawn_queue_structure_monitor}; use queue_network::QueueNetwork; #[doc(hidden)] pub use queue_node::QueueNode; diff --git a/src/rust/lqos_queue_tracker/src/queue_structure/queing_structure_json_monitor.rs b/src/rust/lqos_queue_tracker/src/queue_structure/queing_structure_json_monitor.rs index f23f59de..b59b25e9 100644 --- a/src/rust/lqos_queue_tracker/src/queue_structure/queing_structure_json_monitor.rs +++ b/src/rust/lqos_queue_tracker/src/queue_structure/queing_structure_json_monitor.rs @@ -1,5 +1,6 @@ use crate::queue_structure::{ - queue_network::QueueNetwork, queue_node::QueueNode, read_queueing_structure, + QueueStructureError, queue_network::QueueNetwork, queue_node::QueueNode, + read_queueing_structure, }; use arc_swap::ArcSwap; use lqos_utils::file_watcher::FileWatcher; @@ -26,17 +27,14 @@ static INITIAL_EFFECTIVE_RATES: Lazy = Lazy::new(|| { /// /// This contains only named queue nodes that map cleanly back to authored network-tree /// entries. Circuit/device rows and generated placeholder nodes are intentionally excluded. -pub static EFFECTIVE_NODE_RATES: Lazy>> = Lazy::new(|| { - ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.nodes.clone())) -}); +pub static EFFECTIVE_NODE_RATES: Lazy>> = + Lazy::new(|| ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.nodes.clone()))); /// Global effective circuit-rate overlay derived from `queuingStructure.json`. /// /// This contains the currently programmed circuit queue rates keyed by normalized circuit ID. -pub static EFFECTIVE_CIRCUIT_RATES: Lazy>> = Lazy::new(|| { - ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.circuits.clone())) -}); -/// Set to true when the queue structure changes. This is here rather than in StormGuard -/// to avoid circular dependencies. +pub static EFFECTIVE_CIRCUIT_RATES: Lazy>> = + Lazy::new(|| ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.circuits.clone()))); +/// Set when StormGuard should reconsider a non-live queue-plan snapshot. pub static QUEUE_STRUCTURE_CHANGED_STORMGUARD: AtomicBool = AtomicBool::new(false); #[allow(missing_docs)] @@ -122,33 +120,39 @@ pub fn spawn_queue_structure_monitor() -> anyhow::Result<()> { Ok(()) } -fn update_queue_structure() { +/// Reloads the queue structure from its generated JSON file. +/// +/// This is exposed for consumers that must synchronize an in-memory snapshot with a +/// successfully applied shaping-tree generation instead of waiting for the file watcher. +pub fn reload_queue_structure() -> Result<(), QueueStructureError> { debug!("queueingStructure.json reload requested"); - match read_queueing_structure() { - Ok(queues) => { - let effective_rates = build_effective_rates(&queues); - let new_queue_structure = QueueStructure { - maybe_queues: Some(queues), - }; - QUEUE_STRUCTURE.store(Arc::new(new_queue_structure)); - EFFECTIVE_NODE_RATES.store(Arc::new(effective_rates.nodes)); - EFFECTIVE_CIRCUIT_RATES.store(Arc::new(effective_rates.circuits)); - QUEUE_STRUCTURE_CHANGED_STORMGUARD.store(true, std::sync::atomic::Ordering::Relaxed); - } - Err(err) => { - if QUEUE_STRUCTURE.load().maybe_queues.is_some() { - warn!( - "Failed to reload queuingStructure.json ({err:?}); preserving last-known-good snapshot" - ); - } else { - warn!( - "Failed to load queuingStructure.json ({err:?}); leaving queue structure unavailable" - ); - QUEUE_STRUCTURE.store(Arc::new(QueueStructure { maybe_queues: None })); - EFFECTIVE_NODE_RATES.store(Arc::new(HashMap::new())); - EFFECTIVE_CIRCUIT_RATES.store(Arc::new(HashMap::new())); - } + let queues = read_queueing_structure()?; + let effective_rates = build_effective_rates(&queues); + let new_queue_structure = QueueStructure { + maybe_queues: Some(queues), + }; + QUEUE_STRUCTURE.store(Arc::new(new_queue_structure)); + EFFECTIVE_NODE_RATES.store(Arc::new(effective_rates.nodes)); + EFFECTIVE_CIRCUIT_RATES.store(Arc::new(effective_rates.circuits)); + Ok(()) +} + +fn update_queue_structure() { + if let Err(err) = reload_queue_structure() { + if QUEUE_STRUCTURE.load().maybe_queues.is_some() { + warn!( + "Failed to reload queuingStructure.json ({err:?}); preserving last-known-good snapshot" + ); + } else { + warn!( + "Failed to load queuingStructure.json ({err:?}); leaving queue structure unavailable" + ); + QUEUE_STRUCTURE.store(Arc::new(QueueStructure { maybe_queues: None })); + EFFECTIVE_NODE_RATES.store(Arc::new(HashMap::new())); + EFFECTIVE_CIRCUIT_RATES.store(Arc::new(HashMap::new())); } + } else { + QUEUE_STRUCTURE_CHANGED_STORMGUARD.store(true, std::sync::atomic::Ordering::Relaxed); } } diff --git a/src/rust/lqos_stormguard/README.md b/src/rust/lqos_stormguard/README.md index a1d09cf6..1b8fb4b0 100644 --- a/src/rust/lqos_stormguard/README.md +++ b/src/rust/lqos_stormguard/README.md @@ -68,6 +68,7 @@ active_ping_timeout_seconds = 1.0 You can list as many sites as you want in `targets`, or turn on `all_sites` and carve out exceptions with `exclude_sites`. `dry_run` is the recommended starting point while tuning the thresholds for a network. +Disabling StormGuard, or switching a live deployment back to `dry_run`, restores its managed queues to their planned rates and clears StormGuard-owned adaptive overrides. ## How it works diff --git a/src/rust/lqos_stormguard/src/adaptive_actions.rs b/src/rust/lqos_stormguard/src/adaptive_actions.rs index 21ab8125..4c4addbe 100644 --- a/src/rust/lqos_stormguard/src/adaptive_actions.rs +++ b/src/rust/lqos_stormguard/src/adaptive_actions.rs @@ -1,11 +1,13 @@ -use anyhow::{Result, anyhow}; +use anyhow::{Context, Result, anyhow}; use lqos_bakery::BakeryCommands; use lqos_config::ShapedDevice; use lqos_overrides::{OverrideFile, OverrideLayer, OverrideStore}; -use lqos_queue_tracker::QUEUE_STRUCTURE; use lqos_utils::hash_to_i64; use std::collections::{BTreeMap, HashMap, HashSet}; +use std::time::{Duration, Instant}; +use tracing::warn; +#[derive(Debug)] pub enum CircuitFallbackOutcome { Applied { persisted: bool }, Cleared { persisted: bool }, @@ -25,6 +27,31 @@ pub struct PersistedCircuitFallback { pub devices: Vec, } +/// Waits asynchronously for an acknowledged Bakery operation. +/// +/// Side effects: yields to the Tokio runtime while polling the synchronous reply channel. +pub(crate) async fn wait_for_bakery_reply( + reply_receiver: std::sync::mpsc::Receiver>, + operation: &str, +) -> Result { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + match reply_receiver.try_recv() { + Ok(Ok(value)) => return Ok(value), + Ok(Err(error)) => return Err(anyhow!(error)), + Err(std::sync::mpsc::TryRecvError::Disconnected) => { + return Err(anyhow!("Bakery closed the {operation} reply channel")); + } + Err(std::sync::mpsc::TryRecvError::Empty) if Instant::now() >= deadline => { + return Err(anyhow!("timed out waiting for Bakery to {operation}")); + } + Err(std::sync::mpsc::TryRecvError::Empty) => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + } +} + pub fn apply_site_override_updates(updates: &[SiteOverrideUpdate]) -> Result { if updates.is_empty() { return Ok(false); @@ -64,11 +91,19 @@ pub fn apply_site_override_updates(updates: &[SiteOverrideUpdate]) -> Result Result<()> { + OverrideStore::save_layer(OverrideLayer::Stormguard, &OverrideFile::default()) +} + +pub async fn apply_circuit_fallback( circuit_id: &str, sqm_override: &str, persist: bool, dry_run: bool, + tree_generation: u64, bakery_sender: crossbeam_channel::Sender, ) -> Result { let devices = load_devices_for_circuit(circuit_id)?; @@ -88,18 +123,57 @@ pub fn apply_circuit_fallback( }); } + let reply_receiver = apply_circuit_sqm_override_live( + circuit_id, + Some(sqm_override), + tree_generation, + bakery_sender.clone(), + )?; + if !wait_for_bakery_reply(reply_receiver, "apply StormGuard circuit fallback").await? { + return Ok(CircuitFallbackOutcome::Skipped { + reason: "Circuit is absent from Bakery state.".to_string(), + }); + } let persisted = if persist { - set_devices_sqm_override(&devices, sqm_override)? + match set_devices_sqm_override(&devices, sqm_override) { + Ok(persisted) => persisted, + Err(persistence_error) => { + let rollback = apply_circuit_sqm_override_live( + circuit_id, + None, + tree_generation, + bakery_sender, + ); + let rollback = match rollback { + Ok(rollback_receiver) => wait_for_bakery_reply( + rollback_receiver, + "roll back unpersisted StormGuard circuit fallback", + ) + .await + .map(|_| ()), + Err(error) => Err(error), + }; + if let Err(rollback_error) = rollback { + warn!( + "StormGuard could not persist or roll back circuit fallback for {circuit_id}; retaining in-memory ownership: persistence={persistence_error}; rollback={rollback_error}" + ); + return Ok(CircuitFallbackOutcome::Applied { persisted: false }); + } + return Err(persistence_error).context( + "failed to persist StormGuard circuit fallback; live change was rolled back", + ); + } + } } else { false }; - apply_circuit_sqm_override_live(circuit_id, &devices, Some(sqm_override), bakery_sender)?; Ok(CircuitFallbackOutcome::Applied { persisted }) } -pub fn clear_circuit_fallback( +pub async fn clear_circuit_fallback( circuit_id: &str, dry_run: bool, + tree_generation: u64, bakery_sender: crossbeam_channel::Sender, ) -> Result { let fallback = load_persisted_circuit_fallbacks()? @@ -129,8 +203,15 @@ pub fn clear_circuit_fallback( .iter() .map(|d| d.device_id.clone()) .collect(); + let reply_receiver = + apply_circuit_sqm_override_live(circuit_id, None, tree_generation, bakery_sender)?; + if !wait_for_bakery_reply(reply_receiver, "clear StormGuard circuit fallback").await? { + let persisted = clear_device_overrides(&device_ids)?; + return Ok(CircuitFallbackOutcome::Skipped { + reason: format!("Circuit is absent from Bakery state; persisted={persisted}"), + }); + } let persisted = clear_device_overrides(&device_ids)?; - apply_circuit_sqm_override_live(circuit_id, &fallback.devices, None, bakery_sender)?; Ok(CircuitFallbackOutcome::Cleared { persisted }) } @@ -169,6 +250,12 @@ fn load_devices_for_circuit(circuit_id: &str) -> Result> { fn conflicting_sqm_owner(devices: &[ShapedDevice]) -> Result> { let device_ids: HashSet<&str> = devices.iter().map(|d| d.device_id.as_str()).collect(); + let source_devices = lqos_network_devices::load_source_shaped_devices()?.devices; + if has_source_sqm_override(&device_ids, &source_devices) { + return Ok(Some( + "The circuit already has a source-configured SQM override.".to_string(), + )); + } let operator = OverrideStore::load_layer(OverrideLayer::Operator)?; if has_sqm_override_for_device_ids(&operator, &device_ids) { @@ -187,6 +274,16 @@ fn conflicting_sqm_owner(devices: &[ShapedDevice]) -> Result> { Ok(None) } +fn has_source_sqm_override(device_ids: &HashSet<&str>, source_devices: &[ShapedDevice]) -> bool { + source_devices.iter().any(|device| { + device_ids.contains(device.device_id.as_str()) + && device + .sqm_override + .as_deref() + .is_some_and(|token| !token.trim().is_empty()) + }) +} + fn has_sqm_override_for_device_ids(overrides: &OverrideFile, device_ids: &HashSet<&str>) -> bool { overrides.circuit_adjustments().iter().any(|adj| { matches!( @@ -253,89 +350,19 @@ fn clear_device_overrides(device_ids: &[String]) -> Result { fn apply_circuit_sqm_override_live( circuit_id: &str, - devices: &[ShapedDevice], sqm_override: Option<&str>, + tree_generation: u64, bakery_sender: crossbeam_channel::Sender, -) -> Result<()> { - let snapshot = QUEUE_STRUCTURE.load(); - let Some(queues) = snapshot.maybe_queues.as_ref() else { - return Err(anyhow!("queueingStructure.json not loaded")); - }; - - let mut stack = Vec::new(); - for queue in queues.iter() { - stack.push(queue); - } - - let mut found = None; - while let Some(node) = stack.pop() { - if node.circuit_id.as_deref() == Some(circuit_id) && node.device_id.is_none() { - found = Some(node); - break; - } - - for child in node.children.iter() { - stack.push(child); - } - for circuit in node.circuits.iter() { - stack.push(circuit); - } - for device in node.devices.iter() { - stack.push(device); - } - } - - let Some(node) = found else { - return Err(anyhow!( - "circuit not found in queue structure: {circuit_id}" - )); - }; - - let class_minor = u16::try_from(node.class_minor) - .map_err(|_| anyhow!("class_minor too large: {}", node.class_minor))?; - let class_major = u16::try_from(node.class_major) - .map_err(|_| anyhow!("class_major too large: {}", node.class_major))?; - let up_class_major = u16::try_from(node.up_class_major) - .map_err(|_| anyhow!("up_class_major too large: {}", node.up_class_major))?; - - bakery_sender.send(BakeryCommands::AddCircuit { +) -> Result>> { + let (reply, reply_receiver) = std::sync::mpsc::channel(); + bakery_sender.send(BakeryCommands::StormGuardCircuitAdjustment { + tree_generation, circuit_hash: hash_to_i64(circuit_id), - circuit_name: node - .circuit_name - .clone() - .or_else(|| Some(circuit_id.to_string())), - site_name: node.parent_node.clone(), - parent_class_id: node.parent_class_id, - up_parent_class_id: node.up_parent_class_id, - class_minor, - download_bandwidth_min: node.download_bandwidth_mbps_min as f32, - upload_bandwidth_min: node.upload_bandwidth_mbps_min as f32, - download_bandwidth_max: node.download_bandwidth_mbps as f32, - upload_bandwidth_max: node.upload_bandwidth_mbps as f32, - class_major, - up_class_major, - down_qdisc_handle: None, - up_qdisc_handle: None, - ip_addresses: ip_list(devices), - sqm_override: sqm_override.map(|value| value.to_string()), + sqm_override: sqm_override.map(str::to_string), + reply: Some(reply), })?; - Ok(()) -} - -fn ip_list(devices: &[ShapedDevice]) -> String { - let mut ips = Vec::new(); - for dev in devices { - for (ip, prefix) in dev.ipv4.iter() { - ips.push(format!("{ip}/{prefix}")); - } - for (ip, prefix) in dev.ipv6.iter() { - ips.push(format!("{ip}/{prefix}")); - } - } - ips.sort(); - ips.dedup(); - ips.join(",") + Ok(reply_receiver) } fn group_circuit_fallbacks( @@ -513,4 +540,15 @@ mod tests { ); assert!(!grouped.contains_key("c2")); } + + #[test] + fn source_sqm_change_is_visible_while_effective_device_has_stormguard_token() { + let mut effective = sample_device("dev1", "c1"); + effective.sqm_override = Some("cake".to_string()); + let mut source = sample_device("dev1", "c1"); + source.sqm_override = Some("fq_codel".to_string()); + let device_ids = HashSet::from([effective.device_id.as_str()]); + + assert!(has_source_sqm_override(&device_ids, &[source])); + } } diff --git a/src/rust/lqos_stormguard/src/config.rs b/src/rust/lqos_stormguard/src/config.rs index 6e897728..84db6bd6 100644 --- a/src/rust/lqos_stormguard/src/config.rs +++ b/src/rust/lqos_stormguard/src/config.rs @@ -26,6 +26,8 @@ pub struct WatchingSite { pub struct WatchingSiteDependency { pub name: String, pub class_id: TcHandle, + pub original_min_download_mbps: u64, + pub original_min_upload_mbps: u64, pub original_max_download_mbps: u64, pub original_max_upload_mbps: u64, } diff --git a/src/rust/lqos_stormguard/src/lib.rs b/src/rust/lqos_stormguard/src/lib.rs index cd29ec18..11ee9397 100644 --- a/src/rust/lqos_stormguard/src/lib.rs +++ b/src/rust/lqos_stormguard/src/lib.rs @@ -15,8 +15,9 @@ use lqos_config::NetworkJsonTransport; use lqos_probe::ProbeClient; use lqos_queue_tracker::QUEUE_STRUCTURE_CHANGED_STORMGUARD; use parking_lot::Mutex; +use std::collections::HashSet; use std::time::Duration; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; mod active_ping; mod adaptive_actions; @@ -28,6 +29,55 @@ mod site_state; const READING_ACCUMULATOR_SIZE: usize = 15; const MOVING_AVERAGE_BUFFER_SIZE: usize = 15; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum StormguardRuntimeMode { + Disabled, + DryRun, + Live, +} + +fn runtime_mode(enabled: bool, dry_run: bool) -> StormguardRuntimeMode { + if !enabled { + StormguardRuntimeMode::Disabled + } else if dry_run { + StormguardRuntimeMode::DryRun + } else { + StormguardRuntimeMode::Live + } +} + +fn requested_runtime_mode() -> Option { + lqos_config::load_config().ok().map(|config| { + config + .stormguard + .as_ref() + .map(|stormguard| runtime_mode(stormguard.enabled, stormguard.dry_run)) + .unwrap_or(StormguardRuntimeMode::Disabled) + }) +} + +fn active_runtime_mode(config: Option<&config::StormguardConfig>) -> StormguardRuntimeMode { + config + .map(|config| runtime_mode(true, config.dry_run)) + .unwrap_or(StormguardRuntimeMode::Disabled) +} + +fn requires_live_reset( + active_mode: StormguardRuntimeMode, + requested_mode: StormguardRuntimeMode, +) -> bool { + active_mode == StormguardRuntimeMode::Live && requested_mode != StormguardRuntimeMode::Live +} + +fn rebuild_occurred_since(observed: u64, current: u64, latest_rebuild: u64) -> bool { + observed < latest_rebuild && latest_rebuild <= current +} + +fn clear_published_state() { + STORMGUARD_STATS.lock().clear(); + STORMGUARD_DEBUG.lock().clear(); +} + /// Globally accessible stormguard statistics pub static STORMGUARD_STATS: Mutex> = Mutex::new(Vec::new()); @@ -50,6 +100,10 @@ pub async fn start_stormguard( let mut log_sender: Option> = None; let mut site_state_tracker: Option = None; let mut active_ping = active_ping::ActivePingManager::new(probe_client); + let mut inactive_reconciled = false; + let mut live_reset_pending = false; + let mut observed_tree_generation = lqos_bakery::stormguard_tree_generation(); + let mut retained_circuit_fallbacks = HashSet::new(); // Main Cycle - use tokio interval instead of blocking TimerFd let mut interval = tokio::time::interval(Duration::from_secs(1)); @@ -58,25 +112,168 @@ pub async fn start_stormguard( loop { interval.tick().await; - // Check if queue structure has changed or if we need initial configuration - let queue_structure_changed = - QUEUE_STRUCTURE_CHANGED_STORMGUARD.swap(false, std::sync::atomic::Ordering::Relaxed); + let Some(requested_mode) = requested_runtime_mode() else { + warn!("StormGuard could not read the current runtime mode; retaining existing state."); + continue; + }; + let active_mode = active_runtime_mode(config.as_ref()); + let dry_run_plan_changed = QUEUE_STRUCTURE_CHANGED_STORMGUARD + .swap(false, std::sync::atomic::Ordering::Relaxed) + && active_mode == StormguardRuntimeMode::DryRun + && requested_mode == StormguardRuntimeMode::DryRun; + let current_tree_generation = lqos_bakery::stormguard_tree_generation(); + let bakery_tree_changed = current_tree_generation != observed_tree_generation; + let bakery_tree_rebuilt = rebuild_occurred_since( + observed_tree_generation, + current_tree_generation, + lqos_bakery::stormguard_tree_rebuild_generation(), + ); + if bakery_tree_changed { + if let Err(error) = lqos_queue_tracker::reload_queue_structure() { + warn!( + "StormGuard could not load the queue structure for Bakery generation {current_tree_generation}; retrying: {error}" + ); + continue; + } + if lqos_bakery::stormguard_tree_generation() != current_tree_generation { + debug!( + "Bakery tree changed again while StormGuard refreshed its queue snapshot; retrying." + ); + continue; + } + if bakery_tree_rebuilt + && let Err(error) = + site_state::discard_bakery_adjustments(current_tree_generation, bakery.clone()) + .await + { + warn!("StormGuard rebuild reconciliation will retry: {error}"); + continue; + } + inactive_reconciled = false; + } + if active_mode != StormguardRuntimeMode::Live { + observed_tree_generation = current_tree_generation; + } + if active_mode == StormguardRuntimeMode::Live && bakery_tree_changed { + if bakery_tree_rebuilt { + retained_circuit_fallbacks.clear(); + } else if let Some(tracker) = &site_state_tracker { + retained_circuit_fallbacks.extend(tracker.active_circuit_fallbacks()); + } + observed_tree_generation = current_tree_generation; + config = None; + site_state_tracker = None; + log_sender = None; + live_reset_pending = false; + clear_published_state(); + } + if !bakery_tree_changed && requires_live_reset(active_mode, requested_mode) { + live_reset_pending = true; + } + if live_reset_pending { + let Some(tracker) = &mut site_state_tracker else { + warn!("StormGuard reset is pending without an active tracker; retrying."); + continue; + }; + info!("StormGuard is restoring planned queue rates before reconfiguration."); + match tracker.reset_adjustments(bakery.clone()).await { + Ok(()) => { + config = None; + site_state_tracker = None; + log_sender = None; + inactive_reconciled = true; + live_reset_pending = false; + clear_published_state(); + } + Err(error) => { + warn!("StormGuard could not restore planned queue rates; retrying: {error}"); + continue; + } + } + } else if active_mode != requested_mode && active_mode != StormguardRuntimeMode::Disabled { + config = None; + site_state_tracker = None; + log_sender = None; + clear_published_state(); + } - if config.is_none() || queue_structure_changed { + if requested_mode != StormguardRuntimeMode::Live + && !retained_circuit_fallbacks.is_empty() + { + if let Err(error) = site_state::clear_retained_circuit_fallbacks( + &retained_circuit_fallbacks, + current_tree_generation, + bakery.clone(), + ) + .await + { + warn!("StormGuard retained circuit cleanup will retry: {error}"); + continue; + } + retained_circuit_fallbacks.clear(); + } + + if requested_mode != StormguardRuntimeMode::Live && !inactive_reconciled { + match site_state::reconcile_inactive_state(bakery.clone()).await { + Ok(()) => inactive_reconciled = true, + Err(error) => { + warn!("StormGuard inactive-state cleanup will retry: {error}"); + continue; + } + } + } else if requested_mode == StormguardRuntimeMode::Live { + inactive_reconciled = false; + } + + if requested_mode == StormguardRuntimeMode::Disabled { + config = None; + site_state_tracker = None; + log_sender = None; + clear_published_state(); + active_ping.reconfigure(None); + continue; + } + + if config.is_none() || bakery_tree_changed || dry_run_plan_changed { + let configuration_generation = lqos_bakery::stormguard_tree_generation(); // Try to (re)configure StormGuard match config::configure() { Ok(new_config) => { + if lqos_bakery::stormguard_tree_generation() != configuration_generation { + debug!( + "Bakery tree changed while StormGuard was configuring; retrying with the new generation." + ); + config = None; + site_state_tracker = None; + continue; + } if new_config.is_empty() { debug!("No StormGuard sites found in queue structure yet"); config = None; + site_state_tracker = None; + clear_published_state(); } else { info!("StormGuard configuration loaded successfully"); // Initialize or reinitialize everything if log_sender.is_none() { log_sender = datalog::start_datalog(&new_config).ok(); } - let mut tracker = site_state::SiteStateTracker::from_config(&new_config); - tracker.replay_persisted_adjustments(&new_config, bakery.clone()); + let mut tracker = site_state::SiteStateTracker::from_config( + &new_config, + configuration_generation, + ); + if let Err(error) = tracker + .replay_persisted_adjustments(&new_config, bakery.clone()) + .await + { + warn!("StormGuard persisted adjustment replay will retry: {error}"); + config = None; + site_state_tracker = None; + continue; + } + tracker.retain_active_circuit_fallbacks(std::mem::take( + &mut retained_circuit_fallbacks, + )); site_state_tracker = Some(tracker); config = Some(new_config); } @@ -84,6 +281,8 @@ pub async fn start_stormguard( Err(e) => { debug!("StormGuard configuration not ready: {}", e); config = None; + site_state_tracker = None; + clear_published_state(); } } } @@ -113,8 +312,51 @@ pub async fn start_stormguard( if !recommendations.is_empty() && let Some(sender) = &log_sender { - tracker.apply_recommendations(recommendations, cfg, sender.clone(), bakery.clone()); + tracker + .apply_recommendations(recommendations, cfg, sender.clone(), bakery.clone()) + .await; } } } } + +#[cfg(test)] +mod tests { + use super::{ + StormguardRuntimeMode, rebuild_occurred_since, requires_live_reset, runtime_mode, + }; + + #[test] + fn runtime_mode_tracks_disabled_dry_run_and_live_states() { + assert_eq!(runtime_mode(false, false), StormguardRuntimeMode::Disabled); + assert_eq!(runtime_mode(false, true), StormguardRuntimeMode::Disabled); + assert_eq!(runtime_mode(true, true), StormguardRuntimeMode::DryRun); + assert_eq!(runtime_mode(true, false), StormguardRuntimeMode::Live); + } + + #[test] + fn live_mode_resets_only_when_leaving_live_operation() { + assert!(requires_live_reset( + StormguardRuntimeMode::Live, + StormguardRuntimeMode::Disabled, + )); + assert!(requires_live_reset( + StormguardRuntimeMode::Live, + StormguardRuntimeMode::DryRun, + )); + assert!(!requires_live_reset( + StormguardRuntimeMode::Live, + StormguardRuntimeMode::Live, + )); + assert!(!requires_live_reset( + StormguardRuntimeMode::DryRun, + StormguardRuntimeMode::Disabled, + )); + } + + #[test] + fn incremental_generation_preserves_bakery_ownership() { + assert!(!rebuild_occurred_since(4, 5, 4)); + assert!(rebuild_occurred_since(4, 6, 5)); + } +} diff --git a/src/rust/lqos_stormguard/src/queue_structure.rs b/src/rust/lqos_stormguard/src/queue_structure.rs index 6f8b8430..8a9d61e3 100644 --- a/src/rust/lqos_stormguard/src/queue_structure.rs +++ b/src/rust/lqos_stormguard/src/queue_structure.rs @@ -46,6 +46,8 @@ pub fn find_queue_dependents(parent_name: &str) -> Result, active_circuit_fallbacks: HashSet, } struct CircuitQueueRecommendationContext<'a> { + tree_generation: u64, active_circuit_fallbacks: &'a mut HashSet, site: &'a mut SiteState, config: &'a StormguardConfig, @@ -45,8 +50,266 @@ struct CircuitQueueRecommendationContext<'a> { bakery_sender: Sender, } +struct HtbChangeRequest { + tree_generation: u64, + dry_run: bool, + interface_name: String, + class_handle: TcHandle, + new_rate: u64, + planned_rate: u64, + planned_ceil: u64, +} + +async fn reset_bakery_adjustments( + tree_generation: u64, + adjustments: Vec, + restore_untracked: bool, + bakery_sender: Sender, +) -> anyhow::Result<()> { + let (reply_sender, reply_receiver) = std::sync::mpsc::channel(); + bakery_sender + .send(BakeryCommands::ResetStormGuardAdjustments { + tree_generation, + adjustments, + restore_untracked, + reply: Some(reply_sender), + }) + .context("failed to send StormGuard reset command to Bakery")?; + + wait_for_bakery_reply(reply_receiver, "reset StormGuard adjustments").await +} + +/// Discards class ownership cached against the previous shaping tree. +/// +/// Side effects: clears Bakery's in-process StormGuard adjustment cache after acknowledgement. +pub(crate) async fn discard_bakery_adjustments( + tree_generation: u64, + bakery_sender: Sender, +) -> anyhow::Result<()> { + let (reply_sender, reply_receiver) = std::sync::mpsc::channel(); + bakery_sender + .send(BakeryCommands::DiscardStormGuardAdjustments { + tree_generation, + reply: Some(reply_sender), + }) + .context("failed to send StormGuard discard command to Bakery")?; + wait_for_bakery_reply(reply_receiver, "discard stale StormGuard ownership").await +} + +/// Clears circuit fallbacks retained across an incremental tracker reload. +/// +/// Side effects: applies acknowledged Bakery circuit updates and removes any matching persisted +/// StormGuard device overrides for each retained circuit. +pub(crate) async fn clear_retained_circuit_fallbacks( + circuit_ids: &HashSet, + tree_generation: u64, + bakery_sender: Sender, +) -> anyhow::Result<()> { + for circuit_id in circuit_ids { + match clear_circuit_fallback( + circuit_id, + false, + tree_generation, + bakery_sender.clone(), + ) + .await? + { + CircuitFallbackOutcome::Cleared { .. } | CircuitFallbackOutcome::Skipped { .. } => {} + outcome => { + return Err(anyhow!( + "unexpected retained circuit cleanup outcome for {circuit_id}: {outcome:?}" + )); + } + } + } + Ok(()) +} + +/// Clears stale StormGuard replay and persistence state while StormGuard is inactive. +/// +/// Side effects: asks Bakery to discard retained HTB replay entries, then empties the persisted +/// StormGuard override layer. The persisted layer is unchanged if Bakery cannot acknowledge. +pub async fn reconcile_inactive_state(bakery_sender: Sender) -> anyhow::Result<()> { + let tree_generation = lqos_bakery::stormguard_tree_generation(); + let override_snapshot = OverrideStore::load_layer(OverrideLayer::Stormguard)?; + let adjustments = persisted_restore_adjustments(&override_snapshot)?; + if lqos_bakery::stormguard_tree_generation() != tree_generation { + return Err(anyhow!( + "Bakery shaping tree changed while preparing inactive StormGuard reconciliation" + )); + } + reset_bakery_adjustments(tree_generation, adjustments, true, bakery_sender.clone()).await?; + let circuit_ids: Vec = load_persisted_circuit_fallbacks()?.into_keys().collect(); + for circuit_id in circuit_ids { + let outcome = + clear_circuit_fallback(&circuit_id, false, tree_generation, bakery_sender.clone()) + .await; + let outcome = match outcome { + Ok(outcome) => outcome, + Err(error) => { + restore_override_snapshot( + &override_snapshot, + "after inactive circuit cleanup failed", + )?; + return Err(error); + } + }; + match outcome { + CircuitFallbackOutcome::Cleared { .. } | CircuitFallbackOutcome::Skipped { .. } => {} + outcome => { + restore_override_snapshot( + &override_snapshot, + "after unexpected inactive circuit cleanup outcome", + )?; + return Err(anyhow!( + "unexpected inactive reset outcome for circuit {circuit_id}: {outcome:?}" + )); + } + } + } + ensure_generation_or_restore_overrides( + tree_generation, + &override_snapshot, + "before inactive StormGuard persistence cleanup", + )?; + clear_stormguard_override_layer().context("failed to clear inactive StormGuard overrides")?; + ensure_generation_or_restore_overrides( + tree_generation, + &override_snapshot, + "during inactive StormGuard persistence cleanup", + )?; + Ok(()) +} + +fn ensure_generation_or_restore_overrides( + expected_generation: u64, + snapshot: &OverrideFile, + context: &str, +) -> anyhow::Result<()> { + if lqos_bakery::stormguard_tree_generation() == expected_generation { + return Ok(()); + } + restore_override_snapshot(snapshot, &format!("after tree changed {context}"))?; + Err(anyhow!("Bakery shaping tree changed {context}")) +} + +fn restore_override_snapshot(snapshot: &OverrideFile, context: &str) -> anyhow::Result<()> { + OverrideStore::save_layer(OverrideLayer::Stormguard, snapshot) + .with_context(|| format!("failed to restore StormGuard ownership {context}")) +} + +fn persisted_restore_adjustments( + overrides: &OverrideFile, +) -> anyhow::Result> { + let site_adjustments: Vec<_> = overrides + .network_adjustments() + .iter() + .filter_map(|adjustment| match adjustment { + NetworkAdjustment::AdjustSiteSpeed { + site_name, + download_bandwidth_mbps, + upload_bandwidth_mbps, + .. + } => Some(( + site_name.clone(), + *download_bandwidth_mbps, + *upload_bandwidth_mbps, + )), + _ => None, + }) + .collect(); + if site_adjustments.is_empty() { + return Ok(Vec::new()); + } + + let config = lqos_config::load_config()?; + let queue_snapshot = QUEUE_STRUCTURE.load(); + let queues = queue_snapshot + .maybe_queues + .as_ref() + .ok_or_else(|| anyhow!("queue structure is unavailable for inactive StormGuard reset"))?; + let mut adjustments: HashMap<(String, TcHandle), (u64, u64)> = HashMap::new(); + + for (site_name, download_target, upload_target) in site_adjustments { + let Some(queue) = queues + .iter() + .find(|queue| queue.name.as_deref() == Some(site_name.as_str())) + else { + warn!( + "Skipping stale persisted StormGuard site {site_name}: its queue no longer exists" + ); + continue; + }; + let dependencies = crate::queue_structure::find_queue_dependents(&site_name)?; + for (direction, target) in [ + (RecommendationDirection::Download, download_target), + (RecommendationDirection::Upload, upload_target), + ] { + let Some(target) = target else { + continue; + }; + let interface = match direction { + RecommendationDirection::Download => config.isp_interface(), + RecommendationDirection::Upload => config.internet_interface(), + }; + let planned_rate = match direction { + RecommendationDirection::Download => queue.download_bandwidth_mbps_min, + RecommendationDirection::Upload => queue.upload_bandwidth_mbps_min, + }; + let planned_ceil = match direction { + RecommendationDirection::Download => queue.download_bandwidth_mbps, + RecommendationDirection::Upload => queue.upload_bandwidth_mbps, + }; + adjustments.insert( + (interface.clone(), queue.class_id), + (planned_rate, planned_ceil), + ); + for dependency in &dependencies { + let (dependency_rate, dependency_ceil) = match direction { + RecommendationDirection::Download => ( + dependency.original_min_download_mbps, + dependency.original_max_download_mbps, + ), + RecommendationDirection::Upload => ( + dependency.original_min_upload_mbps, + dependency.original_max_upload_mbps, + ), + }; + if dependency_ceil >= target.max(0.0) as u64 { + adjustments.insert( + (interface.clone(), dependency.class_id), + (dependency_rate, dependency_ceil), + ); + } + } + } + } + + Ok(adjustments + .into_iter() + .map( + |((interface_name, class_id), (planned_rate, planned_ceil))| { + StormGuardRestoreAdjustment { + interface_name, + class_id, + planned_rate, + planned_ceil, + } + }, + ) + .collect()) +} + impl SiteStateTracker { - pub fn from_config(config: &StormguardConfig) -> Self { + pub(crate) fn active_circuit_fallbacks(&self) -> HashSet { + self.active_circuit_fallbacks.clone() + } + + pub(crate) fn retain_active_circuit_fallbacks(&mut self, circuit_ids: HashSet) { + self.active_circuit_fallbacks.extend(circuit_ids); + } + + pub fn from_config(config: &StormguardConfig, tree_generation: u64) -> Self { let mut sites = HashMap::new(); for (name, site) in &config.sites { let delay_probe = matches!( @@ -97,47 +360,50 @@ impl SiteStateTracker { ); } Self { + tree_generation, sites, active_circuit_fallbacks: HashSet::new(), } } - pub fn replay_persisted_adjustments( + pub async fn replay_persisted_adjustments( &mut self, config: &StormguardConfig, bakery_sender: Sender, - ) { + ) -> anyhow::Result<()> { if config.dry_run { - return; + return Ok(()); } let Some(queues) = &QUEUE_STRUCTURE.load().maybe_queues else { info!("No queue structure - cannot replay StormGuard adjustments"); - return; + return Ok(()); }; - match load_persisted_circuit_fallbacks() { - Ok(fallbacks) => { - for (circuit_id, fallback) in fallbacks { + let fallbacks = load_persisted_circuit_fallbacks()?; + for (circuit_id, fallback) in fallbacks { + match apply_circuit_fallback( + &circuit_id, + &fallback.sqm_override, + false, + false, + self.tree_generation, + bakery_sender.clone(), + ) + .await? + { + CircuitFallbackOutcome::Applied { .. } => { self.active_circuit_fallbacks.insert(circuit_id.clone()); - if let Err(e) = apply_circuit_fallback( - &circuit_id, - &fallback.sqm_override, - false, - false, - bakery_sender.clone(), - ) { - warn!( - "Failed to replay persisted StormGuard circuit fallback for {}: {}", - circuit_id, e - ); - } + } + CircuitFallbackOutcome::Skipped { reason } => warn!( + "Skipped persisted StormGuard circuit fallback for {circuit_id}: {reason}" + ), + outcome => { + return Err(anyhow!( + "unexpected persisted circuit replay outcome for {circuit_id}: {outcome:?}" + )); } } - Err(e) => warn!( - "Failed to load persisted StormGuard circuit fallbacks: {}", - e - ), } for (name, site) in &self.sites { @@ -167,23 +433,101 @@ impl SiteStateTracker { "Replaying persisted StormGuard {} override for {}: {} -> {}", direction, name, planned_rate, current_rate ); - Self::apply_dependents( - &site.config, - direction, - current_rate, - config, - &interface_name, - bakery_sender.clone(), - ); - Self::apply_htb_change( + for (class_id, planned_rate, planned_ceil, rate) in + Self::dependent_adjustments(&site.config, direction, current_rate) + { + Self::apply_htb_change_confirmed( + config, + &interface_name, + class_id, + rate, + (planned_rate, planned_ceil), + self.tree_generation, + bakery_sender.clone(), + ) + .await?; + } + Self::apply_htb_change_confirmed( config, &interface_name, queue.class_id, current_rate, + (Self::queue_minimum_rate(queue, direction), planned_rate), + self.tree_generation, bakery_sender.clone(), - ); + ) + .await?; } } + Ok(()) + } + + /// Restores StormGuard-managed classes and circuit fallbacks to their planned state. + /// + /// Side effects: applies live Bakery commands and empties the persisted StormGuard override + /// layer. Persisted and replay state is cleared only after the live HTB reset succeeds. + pub async fn reset_adjustments( + &mut self, + bakery_sender: Sender, + ) -> anyhow::Result<()> { + let override_snapshot = OverrideStore::load_layer(OverrideLayer::Stormguard)?; + reset_bakery_adjustments( + self.tree_generation, + Vec::new(), + false, + bakery_sender.clone(), + ) + .await?; + + let active_circuit_fallbacks: Vec = + self.active_circuit_fallbacks.iter().cloned().collect(); + for circuit_id in active_circuit_fallbacks { + match clear_circuit_fallback( + &circuit_id, + false, + self.tree_generation, + bakery_sender.clone(), + ) + .await + { + Ok(CircuitFallbackOutcome::Cleared { .. }) => {} + Ok(CircuitFallbackOutcome::Skipped { reason }) => { + warn!( + "StormGuard circuit fallback for {circuit_id} no longer owns the live SQM setting: {reason}" + ); + } + Ok(outcome) => { + restore_override_snapshot( + &override_snapshot, + "after unexpected circuit cleanup outcome", + )?; + return Err(anyhow!( + "unexpected reset outcome for StormGuard circuit fallback {circuit_id}: {outcome:?}" + )); + } + Err(error) => { + restore_override_snapshot(&override_snapshot, "after circuit cleanup failed")?; + return Err(error).with_context(|| { + format!("failed to clear StormGuard circuit fallback {circuit_id}") + }); + } + } + } + + ensure_generation_or_restore_overrides( + self.tree_generation, + &override_snapshot, + "before StormGuard persistence cleanup", + )?; + clear_stormguard_override_layer() + .context("failed to clear persisted StormGuard adjustments")?; + ensure_generation_or_restore_overrides( + self.tree_generation, + &override_snapshot, + "during StormGuard persistence cleanup", + )?; + self.active_circuit_fallbacks.clear(); + Ok(()) } pub fn read_new_tick_data( @@ -443,7 +787,7 @@ impl SiteStateTracker { .collect() } - pub fn apply_recommendations( + pub async fn apply_recommendations( &mut self, recommendations: Vec<(Recommendation, String)>, config: &StormguardConfig, @@ -487,6 +831,7 @@ impl SiteStateTracker { // Circuit queues host qdiscs; prefer the TreeGuard-style fallback path. if let Some(circuit_id) = queue.circuit_id.as_deref() { Self::handle_circuit_queue_recommendation(CircuitQueueRecommendationContext { + tree_generation: self.tree_generation, active_circuit_fallbacks: &mut self.active_circuit_fallbacks, site, config, @@ -496,7 +841,8 @@ impl SiteStateTracker { cooldown_secs, log_sender: &log_sender, bakery_sender: bakery_sender.clone(), - }); + }) + .await; continue; } @@ -536,6 +882,7 @@ impl SiteStateTracker { new_rate, config, &interface_name, + self.tree_generation, bakery_sender.clone(), ); Self::apply_htb_change( @@ -543,6 +890,11 @@ impl SiteStateTracker { &interface_name, class_handle, new_rate, + ( + Self::queue_minimum_rate(queue, recommendation.direction), + Self::planned_rate(&site.config, recommendation.direction), + ), + self.tree_generation, bakery_sender.clone(), ); Self::enter_cooldown( @@ -570,6 +922,7 @@ impl SiteStateTracker { new_rate, config, &interface_name, + self.tree_generation, bakery_sender.clone(), ); Self::apply_htb_change( @@ -577,6 +930,11 @@ impl SiteStateTracker { &interface_name, class_handle, new_rate, + ( + Self::queue_minimum_rate(queue, recommendation.direction), + Self::planned_rate(&site.config, recommendation.direction), + ), + self.tree_generation, bakery_sender.clone(), ); pending_site_updates.insert(site.config.name.clone()); @@ -614,8 +972,9 @@ impl SiteStateTracker { } } - fn handle_circuit_queue_recommendation(ctx: CircuitQueueRecommendationContext<'_>) { + async fn handle_circuit_queue_recommendation(ctx: CircuitQueueRecommendationContext<'_>) { let CircuitQueueRecommendationContext { + tree_generation, active_circuit_fallbacks, site, config, @@ -646,8 +1005,11 @@ impl SiteStateTracker { &config.circuit_fallback_sqm, config.circuit_fallback_persist, config.dry_run, + tree_generation, bakery_sender, - ) { + ) + .await + { Ok(outcome) => outcome, Err(e) => { warn!( @@ -661,7 +1023,14 @@ impl SiteStateTracker { } } RecommendationAction::Increase | RecommendationAction::IncreaseFast => { - match clear_circuit_fallback(circuit_id, config.dry_run, bakery_sender) { + match clear_circuit_fallback( + circuit_id, + config.dry_run, + tree_generation, + bakery_sender, + ) + .await + { Ok(outcome) => outcome, Err(e) => { warn!( @@ -756,6 +1125,16 @@ impl SiteStateTracker { } } + fn queue_minimum_rate( + queue: &lqos_queue_tracker::QueueNode, + direction: RecommendationDirection, + ) -> u64 { + match direction { + RecommendationDirection::Download => queue.download_bandwidth_mbps_min, + RecommendationDirection::Upload => queue.upload_bandwidth_mbps_min, + } + } + fn interface_name(config: &StormguardConfig, direction: RecommendationDirection) -> String { match direction { RecommendationDirection::Download => config.download_interface.clone(), @@ -808,30 +1187,55 @@ impl SiteStateTracker { new_rate: u64, config: &StormguardConfig, interface_name: &str, + tree_generation: u64, bakery_sender: Sender, ) { - for dependent in &site.dependent_nodes { - let max_rate = match direction { - RecommendationDirection::Download => dependent.original_max_download_mbps, - RecommendationDirection::Upload => dependent.original_max_upload_mbps, - }; - if max_rate < new_rate { - continue; - } + for (class_id, planned_rate, planned_ceil, rate) in + Self::dependent_adjustments(site, direction, new_rate) + { info!( - "Applying rate change to dependent {}: {} -> {}", - dependent.name, max_rate, new_rate + "Applying rate change to dependent class {}: {} Mbps", + class_id.as_tc_string(), + rate ); Self::apply_htb_change( config, interface_name, - dependent.class_id, - new_rate, + class_id, + rate, + (planned_rate, planned_ceil), + tree_generation, bakery_sender.clone(), ); } } + fn dependent_adjustments( + site: &crate::config::WatchingSite, + direction: RecommendationDirection, + target_rate: u64, + ) -> Vec<(TcHandle, u64, u64, u64)> { + site.dependent_nodes + .iter() + .filter_map(|dependent| { + let original_rate = match direction { + RecommendationDirection::Download => dependent.original_max_download_mbps, + RecommendationDirection::Upload => dependent.original_max_upload_mbps, + }; + let original_min_rate = match direction { + RecommendationDirection::Download => dependent.original_min_download_mbps, + RecommendationDirection::Upload => dependent.original_min_upload_mbps, + }; + (original_rate >= target_rate).then_some(( + dependent.class_id, + original_min_rate, + original_rate, + target_rate, + )) + }) + .collect() + } + fn site_override_update_from_state(site: &SiteState) -> SiteOverrideUpdate { SiteOverrideUpdate { site_name: site.config.name.clone(), @@ -878,14 +1282,19 @@ impl SiteStateTracker { interface_name: &str, class_handle: TcHandle, new_rate: u64, + planned_rates: (u64, u64), + tree_generation: u64, bakery_sender: Sender, ) { - if let Err(e) = bakery_sender.send(BakeryCommands::StormGuardAdjustment { - dry_run: config.dry_run, - interface_name: interface_name.to_owned(), - class_id: class_handle.as_tc_string(), + let request = Self::htb_change_request( + config, + interface_name, + class_handle, new_rate, - }) { + planned_rates, + tree_generation, + ); + if let Err(e) = Self::send_htb_change(request, bakery_sender, None) { warn!("Failed to send StormGuard adjustment command: {}", e); } @@ -923,13 +1332,73 @@ impl SiteStateTracker { // } // } } + + async fn apply_htb_change_confirmed( + config: &StormguardConfig, + interface_name: &str, + class_handle: TcHandle, + new_rate: u64, + planned_rates: (u64, u64), + tree_generation: u64, + bakery_sender: Sender, + ) -> anyhow::Result<()> { + let request = Self::htb_change_request( + config, + interface_name, + class_handle, + new_rate, + planned_rates, + tree_generation, + ); + let (reply_sender, reply_receiver) = std::sync::mpsc::channel(); + Self::send_htb_change(request, bakery_sender, Some(reply_sender))?; + wait_for_bakery_reply(reply_receiver, "replay persisted StormGuard class change").await + } + + fn htb_change_request( + config: &StormguardConfig, + interface_name: &str, + class_handle: TcHandle, + new_rate: u64, + planned_rates: (u64, u64), + tree_generation: u64, + ) -> HtbChangeRequest { + HtbChangeRequest { + tree_generation, + dry_run: config.dry_run, + interface_name: interface_name.to_owned(), + class_handle, + new_rate, + planned_rate: planned_rates.0, + planned_ceil: planned_rates.1, + } + } + + fn send_htb_change( + request: HtbChangeRequest, + bakery_sender: Sender, + reply: Option>>, + ) -> anyhow::Result<()> { + bakery_sender + .send(BakeryCommands::StormGuardAdjustment { + tree_generation: request.tree_generation, + dry_run: request.dry_run, + interface_name: request.interface_name, + class_id: request.class_handle.as_tc_string(), + new_rate: request.new_rate, + planned_rate: request.planned_rate, + planned_ceil: request.planned_ceil, + reply, + }) + .context("failed to send StormGuard class change to Bakery") + } } #[cfg(test)] mod tests { use super::*; use crate::config::StormguardConfig as RuntimeStormguardConfig; - use crate::config::WatchingSite; + use crate::config::{WatchingSite, WatchingSiteDependency}; use lqos_config::StormguardStrategy; use std::collections::HashMap; @@ -1023,6 +1492,44 @@ mod tests { assert_eq!(update.upload_bandwidth_mbps, None); } + #[test] + fn dependent_adjustment_carries_original_directional_rate_pair() -> anyhow::Result<()> { + let mut site = site_state(50, 50, 100, 100).config; + site.dependent_nodes.push(WatchingSiteDependency { + name: "Dependent A".to_string(), + class_id: TcHandle::from_string("1:2")?, + original_min_download_mbps: 7, + original_min_upload_mbps: 3, + original_max_download_mbps: 37, + original_max_upload_mbps: 19, + }); + let download = + SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Download, 15); + let upload = + SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Upload, 15); + assert_eq!(download, [(TcHandle::from_string("1:2")?, 7, 37, 15)]); + assert_eq!(upload, [(TcHandle::from_string("1:2")?, 3, 19, 15)]); + Ok(()) + } + + #[test] + fn dependent_adjustment_skips_classes_below_target_rate() -> anyhow::Result<()> { + let mut site = site_state(50, 50, 100, 100).config; + site.dependent_nodes.push(WatchingSiteDependency { + name: "Dependent A".to_string(), + class_id: TcHandle::from_string("1:2")?, + original_min_download_mbps: 7, + original_min_upload_mbps: 3, + original_max_download_mbps: 37, + original_max_upload_mbps: 19, + }); + assert!( + SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Download, 40,) + .is_empty() + ); + Ok(()) + } + #[test] fn skipped_circuit_outcomes_do_not_enter_cooldown() { assert!(!SiteStateTracker::circuit_outcome_enters_cooldown(&false));