Merge pull request #1117 from LibreQoE/change_jun_26

Route TreeGuard virtual override cleanup through lqosd writer

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;

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 {

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<String>,
+    },
+}
+
 /// 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<OverrideMutation>,
+    },
+
     /// 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());
+    }
 }

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<String>,
+}
+
 /// 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<TreeGuardRuntimeNodeBranchSnapshot>),
+
+    /// Result from a bus-backed override mutation.
+    OverrideMutationResult(OverrideMutationResult),
 }

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;
 

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<String> {
+        if node_names.is_empty() {
+            return Vec::new();
+        }
+
+        let requested_names = node_names
+            .iter()
+            .map(String::as_str)
+            .collect::<HashSet<_>>();
+        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::<Vec<_>>();
+        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.

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>
                 });
                 BusResponse::TreeGuardRuntimeNodeBranch(snapshot)
             }
+            BusRequest::ApplyOverrideMutationBatch { layer, mutations } => {
+                override_writer::apply_bus_mutation_batch(*layer, mutations.clone())
+            }
             BusRequest::ApiReady => {
                 tool_status::api_seen();
                 BusResponse::Ack

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<Sender<OverrideWriterCommand>> = 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<OverrideMutation>,
+        reply: Sender<Result<OverrideMutationResult, OverrideWriterError>>,
+    },
+}
+
+/// 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::<OverrideWriterCommand>(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<OverrideWriterCommand>) {
+    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<OverrideMutation>,
+) -> Result<OverrideMutationResult, OverrideWriterError> {
+    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<OverrideMutation>,
+) -> 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<T>(
+    reply_rx: Receiver<Result<T, OverrideWriterError>>,
+) -> Result<T, OverrideWriterError> {
+    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<OverrideMutationResult, OverrideWriterError> {
+    retry_lock_contention(|| apply_mutations_once(layer, mutations))
+}
+
+fn retry_lock_contention<T>(
+    mut operation: impl FnMut() -> anyhow::Result<T>,
+) -> Result<T, OverrideWriterError> {
+    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<OverrideMutationResult> {
+    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<String>,
+) {
+    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())
+        );
+    }
+}

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<String, TreeguardWarningGroupState>,
 }
 
+#[derive(Clone)]
+struct PendingLinkCleanup {
+    node_name: String,
+    reason: String,
+}
+
+struct PendingLinkCleanupContext<'a> {
+    status: &'a mut TreeguardStatusData,
+    activity: &'a mut VecDeque<TreeguardActivityEntry>,
+    now_unix: u64,
+    dry_run: bool,
+    runtime_virtualized_nodes: &'a mut FxHashSet<String>,
+    pending_link_operations: &'a mut FxHashMap<String, PendingLinkOperation>,
+    link_virtualization_backoff_until_unix: &'a mut FxHashMap<String, u64>,
+    managed_nodes: &'a mut FxHashSet<String>,
+    link_states: &'a mut FxHashMap<String, LinkState>,
+}
+
 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<Option<usize>> =
@@ -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<TreeguardActivityEntry>,
-    now_unix: u64,
-    node_name: &str,
+fn queue_link_cleanup(
+    pending_link_cleanups: &mut Vec<PendingLinkCleanup>,
+    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<String>, FxHashMap<String, String>) {
+    let mut cleanup_reasons_by_node: FxHashMap<String, Vec<String>> = 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");

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<bool, TreeguardError> {
-    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`.