diff --git a/src/rust/lqos_bus/src/bus/mod.rs b/src/rust/lqos_bus/src/bus/mod.rs index 91f01315..d2887559 100644 --- a/src/rust/lqos_bus/src/bus/mod.rs +++ b/src/rust/lqos_bus/src/bus/mod.rs @@ -13,14 +13,14 @@ pub use client::{LibreqosBusClient, bus_request, bus_request_with_timeout}; pub use queue_data::*; pub use reply::BusReply; pub use request::{ - BakeryCapacityReportInterface, BlackboardSystem, BusRequest, SchedulerProgressReport, - TopFlowType, UrgentSeverity, UrgentSource, + BakeryCapacityReportInterface, BlackboardSystem, BusRequest, OverrideLayerSelection, + OverrideMutation, SchedulerProgressReport, TopFlowType, UrgentSeverity, UrgentSource, }; #[allow(unused_imports)] pub use response::{ - AsnHeatmapData, BakeryStatsSnapshot, BusResponse, CircuitHeatmapData, SiteHeatmapData, - StormguardDebugDirection, StormguardDebugEntry, TreeGuardRuntimeNodeBranchSnapshot, - TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue, + AsnHeatmapData, BakeryStatsSnapshot, BusResponse, CircuitHeatmapData, OverrideMutationResult, + SiteHeatmapData, StormguardDebugDirection, StormguardDebugEntry, + TreeGuardRuntimeNodeBranchSnapshot, TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue, }; pub use session::BusSession; use thiserror::Error; diff --git a/src/rust/lqos_bus/src/bus/protocol.rs b/src/rust/lqos_bus/src/bus/protocol.rs index 6d48ac66..f9fdbcdb 100644 --- a/src/rust/lqos_bus/src/bus/protocol.rs +++ b/src/rust/lqos_bus/src/bus/protocol.rs @@ -158,7 +158,10 @@ mod tests { BUS_CHUNK_SIZE, MAX_FRAME_BYTES, decode_reply_cbor, decode_session_cbor, encode_reply_cbor, encode_session_cbor, read_frame, write_frame, }; - use crate::{BusReply, BusRequest, BusResponse, BusSession, bus::BusClientError}; + use crate::{ + BusReply, BusRequest, BusResponse, BusSession, OverrideLayerSelection, OverrideMutation, + bus::BusClientError, + }; use tokio::io::{AsyncWriteExt, duplex}; #[test] @@ -184,6 +187,21 @@ mod tests { assert_eq!(decoded.requests, session.requests); } + #[test] + fn cbor_round_trip_override_mutation_request() { + let session = BusSession { + requests: vec![BusRequest::ApplyOverrideMutationBatch { + layer: OverrideLayerSelection::Treeguard, + mutations: vec![OverrideMutation::ClearNodeVirtualBatch { + node_names: vec!["Node A".to_string(), "Node B".to_string()], + }], + }], + }; + let bytes = encode_session_cbor(&session).expect("encode_session_cbor"); + let decoded = decode_session_cbor(&bytes).expect("decode_session_cbor"); + assert_eq!(decoded.requests, session.requests); + } + #[test] fn cbor_round_trip_reply() { let reply = BusReply { diff --git a/src/rust/lqos_bus/src/bus/request.rs b/src/rust/lqos_bus/src/bus/request.rs index 07e3e3af..e7883476 100644 --- a/src/rust/lqos_bus/src/bus/request.rs +++ b/src/rust/lqos_bus/src/bus/request.rs @@ -6,6 +6,27 @@ use allocative::Allocative; use lqos_config::Tunables; use serde::{Deserialize, Serialize}; +/// Override layer selected for a bus-backed override mutation. +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Allocative)] +pub enum OverrideLayerSelection { + /// Operator-owned overrides (`lqos_overrides.json`). + Operator, + /// StormGuard-owned overrides (`lqos_overrides.stormguard.json`). + Stormguard, + /// TreeGuard-owned overrides (`lqos_overrides.treeguard.json`). + Treeguard, +} + +/// A single override mutation that `lqosd` should apply through its override writer actor. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Allocative)] +pub enum OverrideMutation { + /// Remove node-virtual state for multiple node names in one load/save cycle. + ClearNodeVirtualBatch { + /// Exact node names from `network.json`. + node_names: Vec, + }, +} + /// Per-interface Bakery qdisc-budget report entry. #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Allocative)] pub struct BakeryCapacityReportInterface { @@ -480,6 +501,14 @@ pub enum BusRequest { node_name: String, }, + /// Apply a batch of override mutations through the `lqosd` override writer actor. + ApplyOverrideMutationBatch { + /// Override layer to mutate. + layer: OverrideLayerSelection, + /// Mutations to apply in one load/save cycle. + mutations: Vec, + }, + /// Announce that the API is ready ApiReady, @@ -693,6 +722,7 @@ impl BusRequest { Self::TreeGuardSetNodeVirtual { .. } => "TreeGuardSetNodeVirtual", Self::TreeGuardGetNodeVirtualStatus { .. } => "TreeGuardGetNodeVirtualStatus", Self::TreeGuardGetNodeVirtualBranchState { .. } => "TreeGuardGetNodeVirtualBranchState", + Self::ApplyOverrideMutationBatch { .. } => "ApplyOverrideMutationBatch", Self::ApiReady => "ApiReady", Self::ChatbotReady => "ChatbotReady", Self::SchedulerReady => "SchedulerReady", @@ -832,7 +862,7 @@ pub enum TopFlowType { #[cfg(test)] mod tests { - use super::{BusRequest, TopFlowType}; + use super::{BusRequest, OverrideLayerSelection, OverrideMutation, TopFlowType}; #[test] fn request_kind_omits_payload_details() { @@ -850,4 +880,18 @@ mod tests { "TopFlows" ); } + + #[test] + fn override_write_request_kinds_and_timeout_classification_are_explicit() { + let batch_write_request = BusRequest::ApplyOverrideMutationBatch { + layer: OverrideLayerSelection::Treeguard, + mutations: vec![OverrideMutation::ClearNodeVirtualBatch { + node_names: vec!["Node A".to_string(), "Node B".to_string()], + }], + }; + + assert_eq!(batch_write_request.kind(), "ApplyOverrideMutationBatch"); + + assert!(!batch_write_request.can_fail_fast_on_timeout()); + } } diff --git a/src/rust/lqos_bus/src/bus/response.rs b/src/rust/lqos_bus/src/bus/response.rs index a0570bc9..33f53f48 100644 --- a/src/rust/lqos_bus/src/bus/response.rs +++ b/src/rust/lqos_bus/src/bus/response.rs @@ -11,6 +11,15 @@ use lqos_utils::{HeatmapBlocks, qoq_heatmap::QoqHeatmapBlocks, units::DownUpOrde use serde::{Deserialize, Serialize}; use std::net::IpAddr; +/// Result from a bus-backed override mutation. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Allocative)] +pub struct OverrideMutationResult { + /// True when the mutation changed the selected override layer on disk. + pub changed: bool, + /// Entity identifiers changed by the mutation, such as node names or device IDs. + pub changed_entities: Vec, +} + /// An urgent issue to be displayed prominently in the UI #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Allocative)] pub struct UrgentIssue { @@ -731,4 +740,7 @@ pub enum BusResponse { /// Latest Bakery runtime branch-state snapshot for a named TreeGuard node, if any. TreeGuardRuntimeNodeBranch(Option), + + /// Result from a bus-backed override mutation. + OverrideMutationResult(OverrideMutationResult), } diff --git a/src/rust/lqos_bus/src/lib.rs b/src/rust/lqos_bus/src/lib.rs index a8d0e5f5..410d709e 100644 --- a/src/rust/lqos_bus/src/lib.rs +++ b/src/rust/lqos_bus/src/lib.rs @@ -28,16 +28,17 @@ pub use bus::response::{ AsnHeatmapData, AsnListEntry, BakeryStatsSnapshot, CircuitCapacityRow, CircuitCount, CircuitHeatmapData, CountryListEntry, DeviceCounts, ExecutiveSummaryHeader, FlowMapPoint, FlowTimelineEntry, InsightLicenseSummary, LtsCapabilitiesSummary, NodeCapacity, - ProtocolListEntry, QueueStatsTotal, RetransmitSummary, SchedulerDetails, SearchResultEntry, - SiteHeatmapData, StormguardDebugDirection, StormguardDebugEntry, - TreeGuardRuntimeNodeBranchSnapshot, TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue, - WarningLevel, + OverrideMutationResult, ProtocolListEntry, QueueStatsTotal, RetransmitSummary, + SchedulerDetails, SearchResultEntry, SiteHeatmapData, StormguardDebugDirection, + StormguardDebugEntry, TreeGuardRuntimeNodeBranchSnapshot, + TreeGuardRuntimeNodeOperationSnapshot, UrgentIssue, WarningLevel, }; pub use bus::{ BUS_SOCKET_PATH, BakeryCapacityReportInterface, BlackboardSystem, BusClientError, BusReply, BusRequest, BusResponse, BusSession, CakeDiffTinTransit, CakeDiffTransit, CakeTransit, - LibreqosBusClient, QueueStoreTransit, SchedulerProgressReport, TopFlowType, UnixSocketServer, - UrgentSeverity, UrgentSource, bus_request, bus_request_with_timeout, + LibreqosBusClient, OverrideLayerSelection, OverrideMutation, QueueStoreTransit, + SchedulerProgressReport, TopFlowType, UnixSocketServer, UrgentSeverity, UrgentSource, + bus_request, bus_request_with_timeout, }; pub use tc_handle::TcHandle; diff --git a/src/rust/lqos_overrides/src/overrides_file.rs b/src/rust/lqos_overrides/src/overrides_file.rs index d19607b2..717d094f 100644 --- a/src/rust/lqos_overrides/src/overrides_file.rs +++ b/src/rust/lqos_overrides/src/overrides_file.rs @@ -1,4 +1,5 @@ use std::{ + collections::HashSet, fs::read_to_string, io::Write, path::{Path, PathBuf}, @@ -914,6 +915,33 @@ impl OverrideFile { before.saturating_sub(self.network_adjustments.len()) } + /// Remove virtual-node overrides for all names in `node_names` in one pass. + /// Returns the sorted list of node names that had at least one override removed. + pub fn remove_network_node_virtual_by_names(&mut self, node_names: &[String]) -> Vec { + if node_names.is_empty() { + return Vec::new(); + } + + let requested_names = node_names + .iter() + .map(String::as_str) + .collect::>(); + let mut removed_names = HashSet::new(); + self.network_adjustments.retain(|adj| match adj { + NetworkAdjustment::SetNodeVirtual { node_name, .. } + if requested_names.contains(node_name.as_str()) => + { + removed_names.insert(node_name.clone()); + false + } + _ => true, + }); + + let mut removed_names = removed_names.into_iter().collect::>(); + removed_names.sort(); + removed_names + } + /// Returns the stored topology parent override for `node_id`, if present. /// /// This function is pure: it has no side effects. diff --git a/src/rust/lqosd/src/main.rs b/src/rust/lqosd/src/main.rs index daffd99c..c6402e25 100644 --- a/src/rust/lqosd/src/main.rs +++ b/src/rust/lqosd/src/main.rs @@ -13,6 +13,7 @@ pub mod lts2_sys; mod memory_watchdog; mod network_devices_hooks; mod node_manager; +mod override_writer; mod preflight_checks; mod probe_provider; mod program_control; @@ -238,6 +239,7 @@ fn main() -> Result<()> { lqos_network_devices::start_daemon_mode(Some(std::sync::Arc::new( network_devices_hooks::LqosdNetworkDevicesHooks, )))?; + override_writer::start_override_writer_actor()?; let system_usage_tx = system_stats::start_system_stats()?; // Handle signals @@ -982,6 +984,9 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec }); BusResponse::TreeGuardRuntimeNodeBranch(snapshot) } + BusRequest::ApplyOverrideMutationBatch { layer, mutations } => { + override_writer::apply_bus_mutation_batch(*layer, mutations.clone()) + } BusRequest::ApiReady => { tool_status::api_seen(); BusResponse::Ack diff --git a/src/rust/lqosd/src/override_writer.rs b/src/rust/lqosd/src/override_writer.rs new file mode 100644 index 00000000..0606dcb5 --- /dev/null +++ b/src/rust/lqosd/src/override_writer.rs @@ -0,0 +1,339 @@ +//! Serialized override file writes for `lqosd`. +//! +//! The writer actor coordinates normal override mutations issued by `lqosd` and +//! bus clients. It keeps retry and batching behavior in one place while the +//! existing process file lock remains the cross-process compatibility guard. + +use crossbeam_channel::{Receiver, RecvTimeoutError, SendTimeoutError, Sender, bounded}; +use lqos_bus::{BusResponse, OverrideLayerSelection, OverrideMutation, OverrideMutationResult}; +use lqos_overrides::{OverrideFile, OverrideLayer, OverrideStore}; +use std::sync::OnceLock; +use std::time::Duration; +use thiserror::Error; +use tracing::{debug, warn}; + +const ACTOR_COMMAND_TIMEOUT: Duration = Duration::from_secs(10); +const OVERRIDE_LOCK_RETRY_ATTEMPTS: usize = 10; +const OVERRIDE_LOCK_INITIAL_RETRY_DELAY: Duration = Duration::from_millis(100); +const OVERRIDE_LOCK_MAX_RETRY_DELAY: Duration = Duration::from_millis(500); + +static OVERRIDE_WRITER_SENDER: OnceLock> = OnceLock::new(); + +/// Errors returned by the override writer actor. +#[derive(Debug, Error)] +pub(crate) enum OverrideWriterError { + /// The actor has not been started yet. + #[error("override writer actor is not running")] + NotRunning, + /// The actor command queue remained full past the request timeout. + #[error("override writer command timed out while queueing")] + QueueTimeout, + /// The actor command queue disconnected. + #[error("override writer actor stopped before accepting the command")] + QueueDisconnected, + /// The actor did not reply before the request timeout. + #[error("override writer timed out waiting for actor reply")] + ReplyTimeout, + /// The actor reply channel disconnected. + #[error("override writer reply channel closed")] + ReplyDisconnected, + /// Override file I/O or parsing failed. + #[error("{details}")] + Store { + /// Human-readable store error. + details: String, + }, +} + +#[derive(Debug)] +enum OverrideWriterCommand { + Apply { + layer: OverrideLayerSelection, + mutations: Vec, + reply: Sender>, + }, +} + +/// Starts the override writer actor. +/// +/// Side effects: spawns a long-running background thread and registers the actor sender globally. +pub(crate) fn start_override_writer_actor() -> Result<(), OverrideWriterError> { + if OVERRIDE_WRITER_SENDER.get().is_some() { + return Ok(()); + } + + let (tx, rx) = bounded::(64); + std::thread::Builder::new() + .name("override_writer".to_string()) + .spawn(move || actor_loop(rx)) + .map_err(|err| OverrideWriterError::Store { + details: err.to_string(), + })?; + let _ = OVERRIDE_WRITER_SENDER.set(tx); + Ok(()) +} + +fn actor_loop(rx: Receiver) { + debug!("override writer actor starting"); + while let Ok(command) = rx.recv() { + match command { + OverrideWriterCommand::Apply { + layer, + mutations, + reply, + } => { + let result = apply_mutations_with_retry(layer, &mutations); + let _ = reply.send(result); + } + } + } + warn!("override writer actor command channel disconnected; exiting actor"); +} + +/// Applies a batch of mutations through the override writer actor. +/// +/// Side effects: sends a command to the override writer actor. The actor may read and write an +/// override file. +pub(crate) fn apply_mutation_batch( + layer: OverrideLayerSelection, + mutations: Vec, +) -> Result { + let (reply_tx, reply_rx) = bounded(1); + send_command(OverrideWriterCommand::Apply { + layer, + mutations, + reply: reply_tx, + })?; + receive_reply(reply_rx) +} + +/// Handles one bus override mutation batch request. +/// +/// Side effects: sends a command to the override writer actor and returns a bus response. +pub(crate) fn apply_bus_mutation_batch( + layer: OverrideLayerSelection, + mutations: Vec, +) -> BusResponse { + match apply_mutation_batch(layer, mutations) { + Ok(result) => BusResponse::OverrideMutationResult(result), + Err(err) => BusResponse::Fail(err.to_string()), + } +} + +fn send_command(command: OverrideWriterCommand) -> Result<(), OverrideWriterError> { + let sender = OVERRIDE_WRITER_SENDER + .get() + .cloned() + .ok_or(OverrideWriterError::NotRunning)?; + sender + .send_timeout(command, ACTOR_COMMAND_TIMEOUT) + .map_err(|err| match err { + SendTimeoutError::Timeout(_) => OverrideWriterError::QueueTimeout, + SendTimeoutError::Disconnected(_) => OverrideWriterError::QueueDisconnected, + }) +} + +fn receive_reply( + reply_rx: Receiver>, +) -> Result { + match reply_rx.recv_timeout(ACTOR_COMMAND_TIMEOUT) { + Ok(result) => result, + Err(RecvTimeoutError::Timeout) => Err(OverrideWriterError::ReplyTimeout), + Err(RecvTimeoutError::Disconnected) => Err(OverrideWriterError::ReplyDisconnected), + } +} + +fn apply_mutations_with_retry( + layer: OverrideLayerSelection, + mutations: &[OverrideMutation], +) -> Result { + retry_lock_contention(|| apply_mutations_once(layer, mutations)) +} + +fn retry_lock_contention( + mut operation: impl FnMut() -> anyhow::Result, +) -> Result { + let mut delay = OVERRIDE_LOCK_INITIAL_RETRY_DELAY; + let mut last_error = None; + for attempt in 1..=OVERRIDE_LOCK_RETRY_ATTEMPTS { + match operation() { + Ok(result) => return Ok(result), + Err(err) if is_retryable_override_lock_error(&err) => { + last_error = Some(err.to_string()); + if attempt == OVERRIDE_LOCK_RETRY_ATTEMPTS { + break; + } + std::thread::sleep(delay); + delay = delay.saturating_mul(2).min(OVERRIDE_LOCK_MAX_RETRY_DELAY); + } + Err(err) => { + return Err(OverrideWriterError::Store { + details: err.to_string(), + }); + } + } + } + + Err(OverrideWriterError::Store { + details: format!( + "override file remained locked after {OVERRIDE_LOCK_RETRY_ATTEMPTS} attempts: {}", + last_error.unwrap_or_else(|| "unknown lock error".to_string()) + ), + }) +} + +fn apply_mutations_once( + layer: OverrideLayerSelection, + mutations: &[OverrideMutation], +) -> anyhow::Result { + if mutations.is_empty() { + return Ok(OverrideMutationResult { + changed: false, + changed_entities: Vec::new(), + }); + } + + let override_layer = to_override_layer(layer); + let mut overrides = OverrideStore::load_layer(override_layer)?; + let result = apply_mutations_to_file(&mut overrides, mutations); + if result.changed { + OverrideStore::save_layer(override_layer, &overrides)?; + } + Ok(result) +} + +fn to_override_layer(layer: OverrideLayerSelection) -> OverrideLayer { + match layer { + OverrideLayerSelection::Operator => OverrideLayer::Operator, + OverrideLayerSelection::Stormguard => OverrideLayer::Stormguard, + OverrideLayerSelection::Treeguard => OverrideLayer::Treeguard, + } +} + +fn is_retryable_override_lock_error(err: &anyhow::Error) -> bool { + let message = err.to_string().to_ascii_lowercase(); + message.contains("lqos_overrides_locked") || message.contains("locked by another process") +} + +fn apply_mutations_to_file( + overrides: &mut OverrideFile, + mutations: &[OverrideMutation], +) -> OverrideMutationResult { + let mut changed_entities = Vec::new(); + + for mutation in mutations { + apply_one_mutation(overrides, mutation, &mut changed_entities); + } + + changed_entities.sort(); + changed_entities.dedup(); + OverrideMutationResult { + changed: !changed_entities.is_empty(), + changed_entities, + } +} + +fn apply_one_mutation( + overrides: &mut OverrideFile, + mutation: &OverrideMutation, + changed_entities: &mut Vec, +) { + match mutation { + OverrideMutation::ClearNodeVirtualBatch { node_names } => { + changed_entities.extend(overrides.remove_network_node_virtual_by_names(node_names)); + } + } +} + +#[cfg(test)] +mod tests { + use super::{ + apply_bus_mutation_batch, apply_mutations_to_file, is_retryable_override_lock_error, + retry_lock_contention, + }; + use lqos_bus::{BusResponse, OverrideLayerSelection, OverrideMutation}; + use lqos_overrides::OverrideFile; + use std::cell::Cell; + + #[test] + fn clear_node_virtual_batch_reports_only_changed_nodes() { + let mut overrides = OverrideFile::default(); + overrides.set_network_node_virtual("Node A".to_string(), true); + overrides.set_network_node_virtual("Node B".to_string(), true); + + let result = apply_mutations_to_file( + &mut overrides, + &[OverrideMutation::ClearNodeVirtualBatch { + node_names: vec![ + "Node B".to_string(), + "Node Missing".to_string(), + "Node A".to_string(), + ], + }], + ); + + assert!(result.changed); + assert_eq!( + result.changed_entities, + vec!["Node A".to_string(), "Node B".to_string()] + ); + assert!(overrides.network_adjustments().is_empty()); + } + + #[test] + fn lock_contention_detection_is_specific_to_override_lock_errors() { + let lock_error = anyhow::anyhow!( + "LQOS_OVERRIDES_LOCKED: The LibreQoS overrides file lock is locked by another process" + ); + let parse_error = anyhow::anyhow!("expected value at line 1 column 1"); + + assert!(is_retryable_override_lock_error(&lock_error)); + assert!(!is_retryable_override_lock_error(&parse_error)); + } + + #[test] + fn retry_lock_contention_retries_then_succeeds() { + let attempts = Cell::new(0); + + let result = retry_lock_contention(|| { + attempts.set(attempts.get() + 1); + if attempts.get() < 3 { + anyhow::bail!( + "LQOS_OVERRIDES_LOCKED: The LibreQoS overrides file lock is locked by another process" + ); + } + Ok("updated") + }); + + assert_eq!(result.expect("retry should eventually succeed"), "updated"); + assert_eq!(attempts.get(), 3); + } + + #[test] + fn retry_lock_contention_fails_immediately_on_non_lock_errors() { + let attempts = Cell::new(0); + + let result = retry_lock_contention::<()>(|| { + attempts.set(attempts.get() + 1); + anyhow::bail!("invalid overrides json") + }); + + assert!(result.is_err()); + assert_eq!(attempts.get(), 1); + } + + #[test] + fn bus_mutation_reports_failure_when_actor_is_not_running() { + let response = apply_bus_mutation_batch( + OverrideLayerSelection::Treeguard, + vec![OverrideMutation::ClearNodeVirtualBatch { + node_names: vec!["Node A".to_string()], + }], + ); + + assert_eq!( + response, + BusResponse::Fail("override writer actor is not running".to_string()) + ); + } +} diff --git a/src/rust/lqosd/src/treeguard/actor.rs b/src/rust/lqosd/src/treeguard/actor.rs index e1f9b393..f9507c25 100644 --- a/src/rust/lqosd/src/treeguard/actor.rs +++ b/src/rust/lqosd/src/treeguard/actor.rs @@ -16,6 +16,7 @@ use crate::treeguard::{bakery, decisions, overrides}; use crossbeam_channel::{Receiver, Sender}; use fxhash::{FxHashMap, FxHashSet}; use lqos_bakery::{BakeryRuntimeNodeOperationFailureReason, BakeryRuntimeNodeOperationStatus}; +use lqos_bus::{OverrideLayerSelection, OverrideMutation}; use lqos_config::{NetworkJsonNode, load_config}; use lqos_overrides::{NetworkAdjustment, OverrideFile, OverrideLayer, OverrideStore}; use lqos_utils::hash_to_i64; @@ -61,6 +62,24 @@ struct TreeguardWarningLimiter { groups: BTreeMap, } +#[derive(Clone)] +struct PendingLinkCleanup { + node_name: String, + reason: String, +} + +struct PendingLinkCleanupContext<'a> { + status: &'a mut TreeguardStatusData, + activity: &'a mut VecDeque, + now_unix: u64, + dry_run: bool, + runtime_virtualized_nodes: &'a mut FxHashSet, + pending_link_operations: &'a mut FxHashMap, + link_virtualization_backoff_until_unix: &'a mut FxHashMap, + managed_nodes: &'a mut FxHashSet, + link_states: &'a mut FxHashMap, +} + impl TreeguardWarningLimiter { fn push( &mut self, @@ -890,6 +909,8 @@ fn run_tick( let managed_nodes = &mut runtime_state.managed_nodes; let managed_device_ids = &mut runtime_state.managed_device_ids; let duplicate_device_conflict_circuits = &mut runtime_state.duplicate_device_conflict_circuits; + let mut pending_link_cleanups = Vec::new(); + let mut selected_link_decisions_to_apply = Vec::new(); let top_level_auto_virtualize = tg.links.enabled && tg.links.top_level_auto_virtualize; if tg.enabled @@ -1048,27 +1069,11 @@ fn run_tick( }; removed.extend(runtime_virtualized_nodes.iter().cloned()); for node_name in removed { - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - &node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name, "TreeGuard disabled or links disabled", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - &node_name, - "TreeGuard disabled or links disabled", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(&node_name); - link_states.remove(&node_name); } } else { lqos_network_devices::with_network_json_read(|net_json| { @@ -1143,27 +1148,11 @@ fn run_tick( ); } for node_name in removed { - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - &node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name, "Node removed from allowlist", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - &node_name, - "Node removed from allowlist", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(&node_name); - link_states.remove(&node_name); } let nodes = net_json.get_nodes_when_ready(); let parent_by_index: Vec> = @@ -1213,27 +1202,11 @@ fn run_tick( status.warnings.push(format!( "TreeGuard links: node '{node_name}' has an operator virtual override; TreeGuard will not manage it." )); - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name.clone(), "Operator override present; TreeGuard will not manage this node.", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - node_name, - "Operator override present; TreeGuard will not manage this node.", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(node_name); - link_states.remove(node_name); continue; } @@ -1241,54 +1214,22 @@ fn run_tick( status.warnings.push(format!( "TreeGuard links allowlist: node '{node_name}' not found in network.json." )); - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name.clone(), "Node no longer exists in network.json", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - node_name, - "Node no longer exists in network.json", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(node_name); - link_states.remove(node_name); continue; }; let Some(node) = nodes.get(index) else { status.warnings.push(format!( "TreeGuard links allowlist: node '{node_name}' index not present." )); - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name.clone(), "Node index no longer exists in network.json", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - node_name, - "Node index no longer exists in network.json", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(node_name); - link_states.remove(node_name); continue; }; @@ -1296,27 +1237,11 @@ fn run_tick( status.warnings.push(format!( "TreeGuard links: node '{node_name}' is marked virtual in base network.json; TreeGuard will not manage it." )); - clear_legacy_treeguard_virtual_override( - status, - activity, - now_unix, - node_name, + queue_link_cleanup( + &mut pending_link_cleanups, + node_name.clone(), "Node is marked virtual in base network.json; TreeGuard refuses to manage base-virtual nodes.", ); - restore_runtime_virtualization_if_needed( - status, - activity, - now_unix, - node_name, - "Node is marked virtual in base network.json; TreeGuard refuses to manage base-virtual nodes.", - tg.dry_run, - runtime_virtualized_nodes, - pending_link_operations, - link_virtualization_backoff_until_unix, - link_states, - ); - managed_nodes.remove(node_name); - link_states.remove(node_name); continue; } @@ -1592,27 +1517,44 @@ fn run_tick( )); } - for decision in selected_link_decisions { - let Some(state) = link_states.get_mut(&decision.node_name) else { - continue; - }; - apply_link_virtualization_decision( - status, - activity, - now_unix, - &decision.node_name, - decision.target, - decision.is_top_level, - decision.reason, - tg.dry_run, - state, - pending_link_operations, - link_virtualization_backoff_until_unix, - ); - } + selected_link_decisions_to_apply.extend(selected_link_decisions); }); } + { + let mut link_cleanup_context = PendingLinkCleanupContext { + status, + activity, + now_unix, + dry_run: tg.dry_run, + runtime_virtualized_nodes, + pending_link_operations, + link_virtualization_backoff_until_unix, + managed_nodes, + link_states, + }; + process_pending_link_cleanups(&mut link_cleanup_context, &pending_link_cleanups); + } + + for decision in selected_link_decisions_to_apply { + let Some(state) = link_states.get_mut(&decision.node_name) else { + continue; + }; + apply_link_virtualization_decision( + status, + activity, + now_unix, + &decision.node_name, + decision.target, + decision.is_top_level, + decision.reason, + tg.dry_run, + state, + pending_link_operations, + link_virtualization_backoff_until_unix, + ); + } + // --- Circuit sampling + decisions (SQM switching) --- let manage_circuits = tg.enabled && tg.circuits.enabled; let circuit_inventory = &runtime_state.circuit_inventory; @@ -2039,48 +1981,123 @@ fn pause_for_bakery_reload_with_flag( false } -fn clear_legacy_treeguard_virtual_override( - status: &mut TreeguardStatusData, - activity: &mut VecDeque, - now_unix: u64, - node_name: &str, +fn queue_link_cleanup( + pending_link_cleanups: &mut Vec, + node_name: String, reason: &str, ) { - match overrides::clear_node_virtual(node_name) { - Ok(changed) => { - if changed { + pending_link_cleanups.push(PendingLinkCleanup { + node_name, + reason: reason.to_string(), + }); +} + +fn process_pending_link_cleanups( + context: &mut PendingLinkCleanupContext<'_>, + pending_link_cleanups: &[PendingLinkCleanup], +) { + if pending_link_cleanups.is_empty() { + return; + } + + let (node_names, cleanup_by_node) = collect_link_cleanup_nodes(pending_link_cleanups); + + let mutation = OverrideMutation::ClearNodeVirtualBatch { + node_names: node_names.clone(), + }; + match crate::override_writer::apply_mutation_batch( + OverrideLayerSelection::Treeguard, + vec![mutation], + ) { + Ok(result) => { + for node_name in result.changed_entities { + let reason = cleanup_by_node + .get(&node_name) + .cloned() + .unwrap_or_else(|| "TreeGuard cleanup".to_string()); push_activity( - activity, + context.activity, TreeguardActivityEntry { - time: now_unix.to_string(), + time: context.now_unix.to_string(), entity_type: "node".to_string(), - entity_id: node_name.to_string(), + entity_id: node_name, action: "clear_virtual_override".to_string(), persisted: true, - reason: reason.to_string(), + reason, ..Default::default() }, ); } } - Err(e) => { - status.warnings.push(format!( - "TreeGuard links: failed to clear legacy virtual override for node '{node_name}': {e}" + Err(err) => { + let details = err.to_string(); + context.status.warnings.push(format!( + "TreeGuard links: failed to clear {} legacy virtual override(s): {details}", + node_names.len() )); - push_activity( - activity, - TreeguardActivityEntry { - time: now_unix.to_string(), - entity_type: "node".to_string(), - entity_id: node_name.to_string(), - action: "clear_virtual_override_failed".to_string(), - persisted: false, - reason: format!("Overrides write failed: {e}"), - ..Default::default() - }, - ); + for node_name in &node_names { + push_activity( + context.activity, + TreeguardActivityEntry { + time: context.now_unix.to_string(), + entity_type: "node".to_string(), + entity_id: node_name.clone(), + action: "clear_virtual_override_failed".to_string(), + persisted: false, + reason: format!("Overrides write failed: {details}"), + ..Default::default() + }, + ); + } } } + + for node_name in node_names { + let reason = cleanup_by_node + .get(&node_name) + .cloned() + .unwrap_or_else(|| "TreeGuard cleanup".to_string()); + restore_runtime_virtualization_if_needed( + context.status, + context.activity, + context.now_unix, + &node_name, + &reason, + context.dry_run, + context.runtime_virtualized_nodes, + context.pending_link_operations, + context.link_virtualization_backoff_until_unix, + context.link_states, + ); + context.managed_nodes.remove(&node_name); + context.link_states.remove(&node_name); + } +} + +fn collect_link_cleanup_nodes( + pending_link_cleanups: &[PendingLinkCleanup], +) -> (Vec, FxHashMap) { + let mut cleanup_reasons_by_node: FxHashMap> = FxHashMap::default(); + let mut node_names = Vec::with_capacity(pending_link_cleanups.len()); + for cleanup in pending_link_cleanups { + cleanup_reasons_by_node + .entry(cleanup.node_name.clone()) + .or_default() + .push(cleanup.reason.clone()); + node_names.push(cleanup.node_name.clone()); + } + node_names.sort(); + node_names.dedup(); + + let cleanup_by_node = cleanup_reasons_by_node + .into_iter() + .map(|(node_name, mut reasons)| { + reasons.sort(); + reasons.dedup(); + (node_name, reasons.join("; ")) + }) + .collect(); + (node_names, cleanup_by_node) } fn current_link_virtual_state( @@ -3301,15 +3318,16 @@ fn update_above_since( mod tests { use super::{ CircuitSqmApplyContext, CircuitSqmTransition, CircuitTickContext, LinkVirtualState, - PendingLinkVirtualizationDecision, TreeguardRuntimeState, apply_circuit_sqm_change, - apply_link_virtualization_decision, base_circuit_sqm_state, build_circuit_inventory, - circuit_evaluation_batch_size, circuit_sqm_transition_from_decision, - clear_structural_ineligible_if_topology_changed, collect_circuit_batch, - direct_circuit_counts_by_node, empty_status_snapshot, enrolled_treeguard_circuits, - ensure_circuit_inventory, latched_structural_ineligible_reason, - out_of_scope_treeguard_sqm_device_ids, pause_for_bakery_reload_with_flag, - process_circuit_tick, run_tick, select_link_virtualization_candidates, - treeguard_manages_circuit_direction, try_consume_circuit_change_budget, + PendingLinkCleanup, PendingLinkVirtualizationDecision, TreeguardRuntimeState, + apply_circuit_sqm_change, apply_link_virtualization_decision, base_circuit_sqm_state, + build_circuit_inventory, circuit_evaluation_batch_size, + circuit_sqm_transition_from_decision, clear_structural_ineligible_if_topology_changed, + collect_circuit_batch, collect_link_cleanup_nodes, direct_circuit_counts_by_node, + empty_status_snapshot, enrolled_treeguard_circuits, ensure_circuit_inventory, + latched_structural_ineligible_reason, out_of_scope_treeguard_sqm_device_ids, + pause_for_bakery_reload_with_flag, process_circuit_tick, run_tick, + select_link_virtualization_candidates, treeguard_manages_circuit_direction, + try_consume_circuit_change_budget, }; use crate::node_manager::ws::messages::TreeguardActivityEntry; use crate::system_stats::SystemStats; @@ -3393,6 +3411,35 @@ mod tests { } } + #[test] + fn link_cleanup_collection_collapses_duplicate_nodes() { + let pending = vec![ + PendingLinkCleanup { + node_name: "Tower B".to_string(), + reason: "stale link".to_string(), + }, + PendingLinkCleanup { + node_name: "Tower A".to_string(), + reason: "removed from topology".to_string(), + }, + PendingLinkCleanup { + node_name: "Tower B".to_string(), + reason: "duplicate stale link".to_string(), + }, + ]; + + let (node_names, cleanup_by_node) = collect_link_cleanup_nodes(&pending); + + assert_eq!( + node_names, + vec!["Tower A".to_string(), "Tower B".to_string()] + ); + assert_eq!( + cleanup_by_node.get("Tower B").map(String::as_str), + Some("duplicate stale link; stale link") + ); + } + #[test] fn treeguard_circuit_inventory_excludes_dynamic_circuits() { let static_device = treeguard_test_device("static-circuit", "static-device", "Tower A"); diff --git a/src/rust/lqosd/src/treeguard/overrides.rs b/src/rust/lqosd/src/treeguard/overrides.rs index e7665497..4b8ea205 100644 --- a/src/rust/lqosd/src/treeguard/overrides.rs +++ b/src/rust/lqosd/src/treeguard/overrides.rs @@ -5,31 +5,6 @@ use crate::treeguard::TreeguardError; use lqos_overrides::{OverrideLayer, OverrideStore}; -/// Removes any node-virtualization overrides for a `network.json` node. -/// -/// This function is not pure: it reads and writes `lqos_overrides.treeguard.json`. -/// -/// Returns `Ok(true)` if the file was changed, `Ok(false)` if no change was needed. -pub(crate) fn clear_node_virtual(node_name: &str) -> Result { - let mut overrides = OverrideStore::load_layer(OverrideLayer::Treeguard).map_err(|e| { - TreeguardError::OverridesLoad { - details: e.to_string(), - } - })?; - - let removed = overrides.remove_network_node_virtual_by_name_count(node_name); - if removed == 0 { - return Ok(false); - } - - OverrideStore::save_layer(OverrideLayer::Treeguard, &overrides).map_err(|e| { - TreeguardError::OverridesSave { - details: e.to_string(), - } - })?; - Ok(true) -} - /// Persists a per-device SQM override token for a list of devices. /// /// This function is not pure: it reads and writes `lqos_overrides.treeguard.json`.