Merge pull request #977 from LibreQoE/fix_454

Fix adaptive SQM override rate regression and require Ubuntu 24.04 networkd hotfix on affected hosts

src/build_dpkg.sh

@@ -39,6 +39,7 @@ LQOS_FILES=(
   scheduler.py
   ShapedDevices.example.csv
   shaping_skip_report.py
+  systemd_hotfix.sh
   virtual_tree_nodes.py
   mikrotikDHCPRouterList.template.csv
   integrationUISPbandwidths.template.csv
@@ -115,6 +116,18 @@ popd > /dev/null || exit
 pushd "$DEBIAN_DIR" > /dev/null || exit
 cat <<EOF > postinst
 #!/bin/bash
+if /opt/libreqos/src/systemd_hotfix.sh should-offer >/dev/null 2>&1; then
+cat >&2 <<'HOTFIX'
+LibreQoS requires the Ubuntu 24.04 systemd-networkd hotfix on this host before installation can continue.
+
+Run:
+  HOTFIX_BASE_URL=https://downloads.libreqos.com sudo /opt/libreqos/src/systemd_hotfix.sh install
+
+Then re-run the LibreQoS package installation.
+HOTFIX
+exit 1
+fi
+
 # Install Python Dependencies
 pushd /opt/libreqos
 # - Setup Python dependencies as a post-install task

src/build_rust.sh

@@ -135,6 +135,8 @@ fi
 echo "-----------------------------------------------------------------"
 echo "Don't forget to setup /etc/lqos.conf!"
 echo "Template .service files can be found in bin/"
+echo "Use ./systemd_hotfix.sh to evaluate or install the Ubuntu 24.04 networkd hotfix bundle."
+echo "LibreQoS package installs on affected Ubuntu 24.04 hosts now require the hotfix first."
 echo ""
 echo "Run sudo rust/remove_pinned_maps.sh before you restart lqosd"
 echo "This ensures that any data-format changes will apply correctly."

src/integrationSplynx.py

@@ -41,9 +41,23 @@ def build_online_ip_maps(customersOnline):
 			temp = ipv6sForService.get(service_id, [])
 			if ipv6 not in temp:
 				temp.append(ipv6)
-			ipv6sForService[service_id] = temp
+				ipv6sForService[service_id] = temp
 	return ipv4sForService, ipv6sForService
 
+
+def stable_splynx_device_id(service_id, equipment_id=None):
+	"""
+	Build a deterministic device ID for Splynx-generated shaped devices.
+
+	For the current importer model, each service maps to a single device row, so the
+	service ID is the stable identity. The optional equipment_id parameter is reserved
+	for a future equipment-aware importer mode.
+	"""
+	base_id = f"splynx_service_{service_id}"
+	if equipment_id is None:
+		return base_id
+	return f"{base_id}_equipment_{equipment_id}"
+
 def supplement_existing_devices_with_online_ips(net, allServices, service_ids_handled, customersOnline, cust_id_to_name, allocated_ipv4s, allocated_ipv6s, device_by_service_id=None):
 	"""
 	For services already handled (devices created from static IPs), supplement missing
@@ -91,7 +105,7 @@ def supplement_existing_devices_with_online_ips(net, allServices, service_ids_ha
 						matched_via_supplementation += 1
 	return matched_via_supplementation
 
-def create_devices_from_online_for_unhandled_services(net, allServices, service_ids_handled, customersOnline, cust_id_to_name, downloadForTariffID, uploadForTariffID, device_counter, allocated_ipv4s, allocated_ipv6s, parent_selector=None, device_by_service_id=None):
+def create_devices_from_online_for_unhandled_services(net, allServices, service_ids_handled, customersOnline, cust_id_to_name, downloadForTariffID, uploadForTariffID, allocated_ipv4s, allocated_ipv6s, parent_selector=None, device_by_service_id=None):
 	"""
 	For services that didn't produce devices via static IPs, create devices for those
 	with online IPs. Optionally select a parent node via parent_selector(serviceItem).
@@ -119,33 +133,32 @@ def create_devices_from_online_for_unhandled_services(net, allServices, service_
 				except Exception:
 					parent_id = None
 			# Create customer circuit node
-			customer = NetworkNode(
-				type=NodeType.client,
-				id=circuit_id,
-				parentId=parent_id,
-				displayName=customer_name,
+				customer = NetworkNode(
+					type=NodeType.client,
+					id=circuit_id,
+					parentId=parent_id,
+					displayName=customer_name,
 				address=customer_name,
 				customerName=customer_name,
-				download=service_download,
-				upload=service_upload
-			)
-			net.addRawNode(customer)
-			# Create device node under the client circuit
-			device = NetworkNode(
-				id=device_counter[0],
-				displayName=customer_name,
-				type=NodeType.device,
-				parentId=circuit_id,
-				mac=service.get('mac', ''),
-				ipv4=ipv4,
-				ipv6=ipv6
-			)
-			net.addRawNode(device)
-			if device_by_service_id is not None:
-				device_by_service_id[service['id']] = device
-			device_counter[0] += 1
-			service_ids_handled.append(service['id'])
-			matched_via_alternate_method += 1
+					download=service_download,
+					upload=service_upload
+				)
+				net.addRawNode(customer)
+				# Create device node under the client circuit
+				device = NetworkNode(
+					id=stable_splynx_device_id(service['id']),
+					displayName=customer_name,
+					type=NodeType.device,
+					parentId=circuit_id,
+					mac=service.get('mac', ''),
+					ipv4=ipv4,
+					ipv6=ipv6
+				)
+				net.addRawNode(device)
+				if device_by_service_id is not None:
+					device_by_service_id[service['id']] = device
+				service_ids_handled.append(service['id'])
+				matched_via_alternate_method += 1
 	return matched_via_alternate_method
 
 def run_splynx_pipeline(strategy_name: str):
@@ -339,7 +352,6 @@ def run_splynx_pipeline(strategy_name: str):
 	build_infrastructure()
 	allocated_ipv4s = {}
 	allocated_ipv6s = {}
-	device_counter = [200000]
 	service_ids_handled = []
 	device_by_service_id = {}
 	static_created = 0
@@ -349,7 +361,7 @@ def run_splynx_pipeline(strategy_name: str):
 			if ipv4_list or ipv6_list:
 				parent_node_id = select_parent(serviceItem)
 				circuit_id = createClientAndDevice(
-					net, serviceItem, cust_id_to_name, downloadForTariffID, uploadForTariffID, device_counter, parent_node_id, ipv4_list, ipv6_list
+					net, serviceItem, cust_id_to_name, downloadForTariffID, uploadForTariffID, parent_node_id, ipv4_list, ipv6_list
 				)
 				service_ids_handled.append(serviceItem['id'])
 				# Last added node is the device
@@ -363,7 +375,7 @@ def run_splynx_pipeline(strategy_name: str):
 	)
 	matched_via_alternate_method = create_devices_from_online_for_unhandled_services(
 		net, allServices, service_ids_handled, customersOnline, cust_id_to_name,
-		downloadForTariffID, uploadForTariffID, device_counter, allocated_ipv4s, allocated_ipv6s,
+		downloadForTariffID, uploadForTariffID, allocated_ipv4s, allocated_ipv6s,
 		parent_selector=select_parent, device_by_service_id=device_by_service_id
 	)
 
@@ -592,7 +604,7 @@ def extractServiceIPs(serviceItem, cust_id_to_name, allocated_ipv4s, allocated_i
 	
 	return ipv4_list, ipv6_list
 
-def createClientAndDevice(net, serviceItem, cust_id_to_name, downloadForTariffID, uploadForTariffID, device_counter, parent_node_id, ipv4_list, ipv6_list):
+def createClientAndDevice(net, serviceItem, cust_id_to_name, downloadForTariffID, uploadForTariffID, parent_node_id, ipv4_list, ipv6_list):
 	"""
 	Create client and device nodes for a service.
 	"""
@@ -621,7 +633,7 @@ def createClientAndDevice(net, serviceItem, cust_id_to_name, downloadForTariffID
 	
 	# Always create a device for each service
 	device = NetworkNode(
-		id=device_counter[0],
+		id=stable_splynx_device_id(serviceItem['id']),
 		displayName=customer_name,
 		type=NodeType.device,
 		parentId=circuit_id,
@@ -630,7 +642,6 @@ def createClientAndDevice(net, serviceItem, cust_id_to_name, downloadForTariffID
 		ipv6=ipv6_list
 	)
 	net.addRawNode(device)
-	device_counter[0] += 1
 	
 	return circuit_id
 

src/rust/lqos_bus/src/bus/unix_socket_server.rs

@@ -16,6 +16,10 @@ use tracing::{debug, error, info, warn};
 use super::BUS_SOCKET_DIRECTORY;
 use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame};
 
+fn dropped_reply_response_count(reply: &BusReply) -> usize {
+    reply.responses.len()
+}
+
 /// Implements a Tokio-friendly server using Unix Sockets and the bus protocol.
 /// Requests are handled and then forwarded to the handler.
 pub struct UnixSocketServer {}
@@ -111,8 +115,11 @@ impl UnixSocketServer {
                 if let Some((reply_channel, msg)) = ret {
                   let mut response = BusReply { responses: Vec::with_capacity(8) };
                   handle_bus_requests(&[msg], &mut response.responses);
-                  if let Err(e) = reply_channel.send(response) {
-                      warn!("Unable to send response back to client: {:?}", e);
+                  if let Err(reply) = reply_channel.send(response) {
+                      warn!(
+                          dropped_response_count = dropped_reply_response_count(&reply),
+                          "Unable to send response back to client; receiver dropped"
+                      );
                   }
                 }
               },
@@ -226,3 +233,18 @@ pub enum UnixSocketServerError {
     #[error("Unable to write to socket")]
     WriteFail,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::dropped_reply_response_count;
+    use crate::{BusReply, BusResponse};
+
+    #[test]
+    fn dropped_reply_summary_only_counts_responses() {
+        let reply = BusReply {
+            responses: vec![BusResponse::Ack, BusResponse::Ack, BusResponse::Ack],
+        };
+
+        assert_eq!(dropped_reply_response_count(&reply), 3);
+    }
+}

src/rust/lqos_overrides/src/overrides_file.rs

@@ -60,6 +60,13 @@ pub enum CircuitAdjustment {
         /// Replacement maximum upload bandwidth in Mbps.
         max_upload_bandwidth: Option<f32>,
     },
+    /// Replaces the SQM override token for a specific device without changing any other fields.
+    DeviceAdjustSqm {
+        /// Device identifier to update.
+        device_id: String,
+        /// Replacement SQM override token. `None` or empty removes the override.
+        sqm_override: Option<String>,
+    },
     /// Removes a circuit from generated output by circuit ID.
     RemoveCircuit {
         /// Circuit identifier to remove.
@@ -198,10 +205,16 @@ fn merge_persistent_devices_by_id(
     for dev in operator_devices {
         by_id.insert(dev.device_id.as_str(), dev.clone());
     }
-    for dev in stormguard_devices {
+    for dev in stormguard_devices
+        .iter()
+        .filter(|dev| !persistent_device_carries_sqm(dev))
+    {
         by_id.insert(dev.device_id.as_str(), dev.clone());
     }
-    for dev in treeguard_devices {
+    for dev in treeguard_devices
+        .iter()
+        .filter(|dev| !persistent_device_carries_sqm(dev))
+    {
         by_id.insert(dev.device_id.as_str(), dev.clone());
     }
 
@@ -219,7 +232,10 @@ fn merge_persistent_devices_by_id(
         }
     }
 
-    for dev in stormguard_devices {
+    for dev in stormguard_devices
+        .iter()
+        .filter(|dev| !persistent_device_carries_sqm(dev))
+    {
         let did = dev.device_id.as_str();
         if seen.contains(did) {
             continue;
@@ -230,7 +246,10 @@ fn merge_persistent_devices_by_id(
         }
     }
 
-    for dev in treeguard_devices {
+    for dev in treeguard_devices
+        .iter()
+        .filter(|dev| !persistent_device_carries_sqm(dev))
+    {
         let did = dev.device_id.as_str();
         if seen.contains(did) {
             continue;
@@ -244,6 +263,108 @@ fn merge_persistent_devices_by_id(
     out
 }
 
+fn persistent_device_carries_sqm(device: &ShapedDevice) -> bool {
+    device
+        .sqm_override
+        .as_deref()
+        .is_some_and(|sqm| !sqm.trim().is_empty())
+}
+
+fn legacy_sqm_adjustments_from_devices(devices: &[ShapedDevice]) -> Vec<CircuitAdjustment> {
+    devices
+        .iter()
+        .filter_map(|device| {
+            let sqm_override = device.sqm_override.as_deref()?.trim();
+            if sqm_override.is_empty() {
+                return None;
+            }
+            Some(CircuitAdjustment::DeviceAdjustSqm {
+                device_id: device.device_id.clone(),
+                sqm_override: Some(sqm_override.to_string()),
+            })
+        })
+        .collect()
+}
+
+fn circuit_adjustment_merge_key(adj: &CircuitAdjustment) -> (&'static str, &str) {
+    match adj {
+        CircuitAdjustment::CircuitAdjustSpeed { circuit_id, .. } => ("circuit_speed", circuit_id),
+        CircuitAdjustment::DeviceAdjustSpeed { device_id, .. } => ("device_speed", device_id),
+        CircuitAdjustment::DeviceAdjustSqm { device_id, .. } => ("device_sqm", device_id),
+        CircuitAdjustment::RemoveCircuit { circuit_id } => ("remove_circuit", circuit_id),
+        CircuitAdjustment::RemoveDevice { device_id } => ("remove_device", device_id),
+        CircuitAdjustment::ReparentCircuit { circuit_id, .. } => ("reparent_circuit", circuit_id),
+    }
+}
+
+fn merge_circuit_adjustments_owned(
+    operator_adjustments: &[CircuitAdjustment],
+    stormguard_adjustments: &[CircuitAdjustment],
+    treeguard_adjustments: &[CircuitAdjustment],
+    stormguard_devices: &[ShapedDevice],
+    treeguard_devices: &[ShapedDevice],
+) -> Vec<CircuitAdjustment> {
+    use std::collections::{HashMap, HashSet};
+
+    let mut by_key: HashMap<(&'static str, &str), CircuitAdjustment> = HashMap::new();
+    let stormguard_combined: Vec<CircuitAdjustment> = stormguard_adjustments
+        .iter()
+        .cloned()
+        .chain(legacy_sqm_adjustments_from_devices(stormguard_devices))
+        .collect();
+    let treeguard_combined: Vec<CircuitAdjustment> = treeguard_adjustments
+        .iter()
+        .cloned()
+        .chain(legacy_sqm_adjustments_from_devices(treeguard_devices))
+        .collect();
+
+    for adj in operator_adjustments {
+        by_key.insert(circuit_adjustment_merge_key(adj), adj.clone());
+    }
+    for adj in &stormguard_combined {
+        by_key.insert(circuit_adjustment_merge_key(adj), adj.clone());
+    }
+    for adj in &treeguard_combined {
+        by_key.insert(circuit_adjustment_merge_key(adj), adj.clone());
+    }
+
+    let mut out = Vec::new();
+    let mut seen: HashSet<(&'static str, &str)> = HashSet::new();
+
+    for adj in operator_adjustments {
+        let key = circuit_adjustment_merge_key(adj);
+        if seen.contains(&key) {
+            continue;
+        }
+        if let Some(merged) = by_key.get(&key) {
+            out.push(merged.clone());
+            seen.insert(key);
+        }
+    }
+    for adj in &stormguard_combined {
+        let key = circuit_adjustment_merge_key(adj);
+        if seen.contains(&key) {
+            continue;
+        }
+        if let Some(merged) = by_key.get(&key) {
+            out.push(merged.clone());
+            seen.insert(key);
+        }
+    }
+    for adj in &treeguard_combined {
+        let key = circuit_adjustment_merge_key(adj);
+        if seen.contains(&key) {
+            continue;
+        }
+        if let Some(merged) = by_key.get(&key) {
+            out.push(merged.clone());
+            seen.insert(key);
+        }
+    }
+
+    out
+}
+
 fn merge_network_adjustments_owned(
     operator_adjustments: &[NetworkAdjustment],
     stormguard_adjustments: &[NetworkAdjustment],
@@ -376,12 +497,20 @@ fn merge_owned_sections(
     stormguard: OverrideFile,
     treeguard: OverrideFile,
 ) -> OverrideFile {
+    let operator_circuit = std::mem::take(&mut operator.circuit_adjustments);
     let operator_devices = std::mem::take(&mut operator.persistent_devices);
     operator.persistent_devices = merge_persistent_devices_by_id(
         &operator_devices,
         &stormguard.persistent_devices,
         &treeguard.persistent_devices,
     );
+    operator.circuit_adjustments = merge_circuit_adjustments_owned(
+        &operator_circuit,
+        &stormguard.circuit_adjustments,
+        &treeguard.circuit_adjustments,
+        &stormguard.persistent_devices,
+        &treeguard.persistent_devices,
+    );
 
     let operator_network = std::mem::take(&mut operator.network_adjustments);
     operator.network_adjustments = merge_network_adjustments_owned(
@@ -515,6 +644,62 @@ impl OverrideFile {
         self.circuit_adjustments.push(adj);
     }
 
+    /// Add or replace an SQM override token for `device_id`. Returns true if changed.
+    pub fn set_device_sqm_override_return_changed(
+        &mut self,
+        device_id: String,
+        sqm_override: Option<String>,
+    ) -> bool {
+        let normalized = sqm_override.and_then(|sqm| {
+            let trimmed = sqm.trim();
+            if trimmed.is_empty() {
+                None
+            } else {
+                Some(trimmed.to_string())
+            }
+        });
+
+        if self.circuit_adjustments.iter().any(|adj| {
+            matches!(
+                adj,
+                CircuitAdjustment::DeviceAdjustSqm {
+                    device_id: current,
+                    sqm_override: existing_sqm,
+                } if current == &device_id && existing_sqm == &normalized
+            )
+        }) {
+            return false;
+        }
+
+        self.circuit_adjustments.retain(|adj| {
+            !matches!(
+                adj,
+                CircuitAdjustment::DeviceAdjustSqm {
+                    device_id: current, ..
+                } if current == &device_id
+            )
+        });
+        self.circuit_adjustments.push(CircuitAdjustment::DeviceAdjustSqm {
+            device_id,
+            sqm_override: normalized,
+        });
+        true
+    }
+
+    /// Remove any SQM override adjustments for `device_id`. Returns number removed.
+    pub fn remove_device_sqm_override_by_device_count(&mut self, device_id: &str) -> usize {
+        let before = self.circuit_adjustments.len();
+        self.circuit_adjustments.retain(|adj| {
+            !matches!(
+                adj,
+                CircuitAdjustment::DeviceAdjustSqm {
+                    device_id: current, ..
+                } if current == device_id
+            )
+        });
+        before.saturating_sub(self.circuit_adjustments.len())
+    }
+
     /// Remove a circuit adjustment by index. Returns true if removed.
     pub fn remove_circuit_adjustment_by_index(&mut self, index: usize) -> bool {
         if index < self.circuit_adjustments.len() {
@@ -725,43 +910,111 @@ mod tests {
         }
     }
 
+    fn shaped_device_with_comment(device_id: &str, comment: &str) -> ShapedDevice {
+        ShapedDevice {
+            device_id: device_id.to_string(),
+            comment: comment.to_string(),
+            ..ShapedDevice::default()
+        }
+    }
+
     #[test]
     fn effective_merge_treeguard_wins_for_persistent_devices() {
         let mut operator = OverrideFile::default();
-        operator
-            .add_persistent_shaped_device_return_changed(shaped_device_with_sqm("dev1", "cake"));
-        operator
-            .add_persistent_shaped_device_return_changed(shaped_device_with_sqm("dev2", "cake"));
+        operator.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev1", "operator",
+        ));
+        operator.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev2", "operator",
+        ));
 
         let mut stormguard = OverrideFile::default();
-        stormguard.add_persistent_shaped_device_return_changed(shaped_device_with_sqm(
-            "dev1", "fq_codel",
+        stormguard.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev1", "stormguard",
         ));
-        stormguard.add_persistent_shaped_device_return_changed(shaped_device_with_sqm(
-            "dev3", "fq_codel",
+        stormguard.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev3", "stormguard",
         ));
 
         let mut treeguard = OverrideFile::default();
-        treeguard
-            .add_persistent_shaped_device_return_changed(shaped_device_with_sqm("dev1", "cake"));
-        treeguard
-            .add_persistent_shaped_device_return_changed(shaped_device_with_sqm("dev4", "cake"));
+        treeguard.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev1", "treeguard",
+        ));
+        treeguard.add_persistent_shaped_device_return_changed(shaped_device_with_comment(
+            "dev4", "treeguard",
+        ));
 
         let merged = merge_owned_sections(operator, stormguard, treeguard);
-        let sqm_by_id: std::collections::HashMap<&str, &str> = merged
+        let comments_by_id: std::collections::HashMap<&str, &str> = merged
             .persistent_devices
             .iter()
-            .filter_map(|d| {
-                d.sqm_override
-                    .as_deref()
-                    .map(|sqm| (d.device_id.as_str(), sqm))
-            })
+            .map(|d| (d.device_id.as_str(), d.comment.as_str()))
             .collect();
 
-        assert_eq!(sqm_by_id.get("dev1"), Some(&"cake"));
+        assert_eq!(comments_by_id.get("dev1"), Some(&"treeguard"));
+        assert_eq!(comments_by_id.get("dev2"), Some(&"operator"));
+        assert_eq!(comments_by_id.get("dev3"), Some(&"stormguard"));
+        assert_eq!(comments_by_id.get("dev4"), Some(&"treeguard"));
+    }
+
+    #[test]
+    fn effective_merge_moves_adaptive_legacy_sqm_into_circuit_adjustments() {
+        let mut operator = OverrideFile::default();
+        let mut operator_device = shaped_device_with_sqm("operator_dev", "cake");
+        operator_device.download_max_mbps = 100.0;
+        operator.add_persistent_shaped_device_return_changed(operator_device);
+
+        let mut stormguard = OverrideFile::default();
+        let mut sg_device = shaped_device_with_sqm("dev1", "fq_codel");
+        sg_device.download_max_mbps = 10.0;
+        stormguard.add_persistent_shaped_device_return_changed(sg_device);
+
+        let mut treeguard = OverrideFile::default();
+        let mut tg_device = shaped_device_with_sqm("dev2", "cake");
+        tg_device.download_max_mbps = 20.0;
+        treeguard.add_persistent_shaped_device_return_changed(tg_device);
+
+        let merged = merge_owned_sections(operator, stormguard, treeguard);
+
+        let merged_ids: Vec<&str> = merged
+            .persistent_devices()
+            .iter()
+            .map(|d| d.device_id.as_str())
+            .collect();
+        assert_eq!(merged_ids, vec!["operator_dev"]);
+
+        let sqm_by_id: std::collections::HashMap<&str, &str> = merged
+            .circuit_adjustments()
+            .iter()
+            .filter_map(|adj| match adj {
+                CircuitAdjustment::DeviceAdjustSqm {
+                    device_id,
+                    sqm_override,
+                } => sqm_override.as_deref().map(|sqm| (device_id.as_str(), sqm)),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(sqm_by_id.get("dev1"), Some(&"fq_codel"));
         assert_eq!(sqm_by_id.get("dev2"), Some(&"cake"));
-        assert_eq!(sqm_by_id.get("dev3"), Some(&"fq_codel"));
-        assert_eq!(sqm_by_id.get("dev4"), Some(&"cake"));
+    }
+
+    #[test]
+    fn set_device_sqm_override_is_idempotent() {
+        let mut of = OverrideFile::default();
+        assert!(of.set_device_sqm_override_return_changed(
+            "dev1".to_string(),
+            Some("fq_codel".to_string())
+        ));
+        assert!(!of.set_device_sqm_override_return_changed(
+            "dev1".to_string(),
+            Some("fq_codel".to_string())
+        ));
+        assert!(of.set_device_sqm_override_return_changed(
+            "dev1".to_string(),
+            Some("cake".to_string())
+        ));
+        assert_eq!(of.remove_device_sqm_override_by_device_count("dev1"), 1);
+        assert_eq!(of.remove_device_sqm_override_by_device_count("dev1"), 0);
     }
 
     #[test]

src/rust/lqos_python/src/lib.rs

@@ -668,6 +668,10 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(overrides_persistent_devices, m)?)?;
     m.add_function(wrap_pyfunction!(overrides_persistent_devices_effective, m)?)?;
     m.add_function(wrap_pyfunction!(overrides_circuit_adjustments, m)?)?;
+    m.add_function(wrap_pyfunction!(
+        overrides_circuit_adjustments_effective,
+        m
+    )?)?;
     m.add_function(wrap_pyfunction!(overrides_network_adjustments, m)?)?;
     m.add_function(wrap_pyfunction!(
         overrides_network_adjustments_effective,
@@ -1084,6 +1088,115 @@ fn overrides_circuit_adjustments(py: Python<'_>) -> PyResult<Vec<PyObject>> {
                     d.set_item("max_upload_bandwidth", *v)?;
                 }
             }
+            lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+                device_id,
+                sqm_override,
+            } => {
+                d.set_item("type", "device_adjust_sqm")?;
+                d.set_item("device_id", device_id.clone())?;
+                if let Some(value) = sqm_override {
+                    d.set_item("sqm_override", value.clone())?;
+                }
+            }
+            lqos_overrides::CircuitAdjustment::RemoveCircuit { circuit_id } => {
+                d.set_item("type", "remove_circuit")?;
+                d.set_item("circuit_id", circuit_id.clone())?;
+            }
+            lqos_overrides::CircuitAdjustment::RemoveDevice { device_id } => {
+                d.set_item("type", "remove_device")?;
+                d.set_item("device_id", device_id.clone())?;
+            }
+            lqos_overrides::CircuitAdjustment::ReparentCircuit {
+                circuit_id,
+                parent_node,
+            } => {
+                d.set_item("type", "reparent_circuit")?;
+                d.set_item("circuit_id", circuit_id.clone())?;
+                d.set_item("parent_node", parent_node.clone())?;
+            }
+        }
+        let obj: PyObject = d.unbind().into();
+        out.push(obj);
+    }
+
+    Ok(out)
+}
+
+/// Returns the list of circuit adjustments as Python dicts, using the effective overrides view
+/// (operator + adaptive layers when enabled).
+#[pyfunction]
+fn overrides_circuit_adjustments_effective(py: Python<'_>) -> PyResult<Vec<PyObject>> {
+    let config = lqos_config::load_config().map_err(|e| PyOSError::new_err(e.to_string()))?;
+    let apply_stormguard = config
+        .stormguard
+        .as_ref()
+        .is_some_and(|sg| sg.enabled && !sg.dry_run);
+    let apply_treeguard = config.treeguard.enabled;
+
+    let overrides =
+        match lqos_overrides::OverrideStore::load_effective(apply_stormguard, apply_treeguard) {
+            Ok(o) => o,
+            Err(e) => return Err(PyOSError::new_err(e.to_string())),
+        };
+
+    let mut out: Vec<PyObject> = Vec::new();
+    for adj in overrides.circuit_adjustments().iter() {
+        let d = PyDict::new(py);
+        match adj {
+            lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed {
+                circuit_id,
+                min_download_bandwidth,
+                max_download_bandwidth,
+                min_upload_bandwidth,
+                max_upload_bandwidth,
+            } => {
+                d.set_item("type", "circuit_adjust_speed")?;
+                d.set_item("circuit_id", circuit_id.clone())?;
+                if let Some(v) = min_download_bandwidth {
+                    d.set_item("min_download_bandwidth", *v)?;
+                }
+                if let Some(v) = max_download_bandwidth {
+                    d.set_item("max_download_bandwidth", *v)?;
+                }
+                if let Some(v) = min_upload_bandwidth {
+                    d.set_item("min_upload_bandwidth", *v)?;
+                }
+                if let Some(v) = max_upload_bandwidth {
+                    d.set_item("max_upload_bandwidth", *v)?;
+                }
+            }
+            lqos_overrides::CircuitAdjustment::DeviceAdjustSpeed {
+                device_id,
+                min_download_bandwidth,
+                max_download_bandwidth,
+                min_upload_bandwidth,
+                max_upload_bandwidth,
+            } => {
+                d.set_item("type", "device_adjust_speed")?;
+                d.set_item("device_id", device_id.clone())?;
+                if let Some(v) = min_download_bandwidth {
+                    d.set_item("min_download_bandwidth", *v)?;
+                }
+                if let Some(v) = max_download_bandwidth {
+                    d.set_item("max_download_bandwidth", *v)?;
+                }
+                if let Some(v) = min_upload_bandwidth {
+                    d.set_item("min_upload_bandwidth", *v)?;
+                }
+                if let Some(v) = max_upload_bandwidth {
+                    d.set_item("max_upload_bandwidth", *v)?;
+                }
+            }
+            lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+                device_id,
+                sqm_override,
+            } => {
+                d.set_item("type", "device_adjust_sqm")?;
+                d.set_item("device_id", device_id.clone())?;
+                if let Some(value) = sqm_override {
+                    d.set_item("sqm_override", value.clone())?;
+                }
+            }
             lqos_overrides::CircuitAdjustment::RemoveCircuit { circuit_id } => {
                 d.set_item("type", "remove_circuit")?;
                 d.set_item("circuit_id", circuit_id.clone())?;

src/rust/lqos_stormguard/src/adaptive_actions.rs

@@ -135,7 +135,8 @@ pub fn clear_circuit_fallback(
 
 pub fn load_persisted_circuit_fallbacks() -> Result<HashMap<String, PersistedCircuitFallback>> {
     let overrides = OverrideStore::load_layer(OverrideLayer::Stormguard)?;
-    Ok(group_circuit_fallbacks(overrides.persistent_devices()))
+    let current_devices = ConfigShapedDevices::load()?.devices;
+    Ok(group_circuit_fallbacks(&overrides, &current_devices))
 }
 
 fn current_site_override(
@@ -185,7 +186,19 @@ fn conflicting_sqm_owner(devices: &[ShapedDevice]) -> Result<Option<String>> {
 }
 
 fn has_sqm_override_for_device_ids(overrides: &OverrideFile, device_ids: &HashSet<&str>) -> bool {
-    overrides.persistent_devices().iter().any(|device| {
+    overrides.circuit_adjustments().iter().any(|adj| {
+        matches!(
+            adj,
+            lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+                device_id,
+                sqm_override,
+            } if device_ids.contains(device_id.as_str())
+                && sqm_override
+                    .as_deref()
+                    .map(str::trim)
+                    .is_some_and(|token| !token.is_empty())
+        )
+    }) || overrides.persistent_devices().iter().any(|device| {
         device_ids.contains(device.device_id.as_str())
             && device
                 .sqm_override
@@ -196,21 +209,15 @@ fn has_sqm_override_for_device_ids(overrides: &OverrideFile, device_ids: &HashSe
 
 fn set_devices_sqm_override(base_devices: &[ShapedDevice], sqm_override: &str) -> Result<bool> {
     let mut overrides = OverrideStore::load_layer(OverrideLayer::Stormguard)?;
-    let effective = OverrideStore::load_effective(false, true)?;
-    let effective_devices: HashMap<&str, &ShapedDevice> = effective
-        .persistent_devices()
-        .iter()
-        .map(|device| (device.device_id.as_str(), device))
-        .collect();
     let mut changed = false;
     for base_device in base_devices {
-        let mut device = effective_devices
-            .get(base_device.device_id.as_str())
-            .copied()
-            .cloned()
-            .unwrap_or_else(|| base_device.clone());
-        device.sqm_override = Some(sqm_override.to_string());
-        if overrides.add_persistent_shaped_device_return_changed(device) {
+        if overrides.set_device_sqm_override_return_changed(
+            base_device.device_id.clone(),
+            Some(sqm_override.to_string()),
+        ) {
+            changed = true;
+        }
+        if overrides.remove_persistent_shaped_device_by_device_count(&base_device.device_id) > 0 {
             changed = true;
         }
     }
@@ -227,8 +234,9 @@ fn clear_device_overrides(device_ids: &[String]) -> Result<bool> {
     let mut overrides = OverrideStore::load_layer(OverrideLayer::Stormguard)?;
     let mut removed_any = false;
     for device_id in device_ids {
-        let removed = overrides.remove_persistent_shaped_device_by_device_count(device_id);
-        if removed > 0 {
+        let removed_adjustments = overrides.remove_device_sqm_override_by_device_count(device_id);
+        let removed_devices = overrides.remove_persistent_shaped_device_by_device_count(device_id);
+        if removed_adjustments > 0 || removed_devices > 0 {
             removed_any = true;
         }
     }
@@ -321,28 +329,75 @@ fn ip_list(devices: &[ShapedDevice]) -> String {
     ips.join(",")
 }
 
-fn group_circuit_fallbacks(devices: &[ShapedDevice]) -> HashMap<String, PersistedCircuitFallback> {
+fn group_circuit_fallbacks(
+    overrides: &OverrideFile,
+    current_devices: &[ShapedDevice],
+) -> HashMap<String, PersistedCircuitFallback> {
+    let devices_by_id: HashMap<&str, &ShapedDevice> = current_devices
+        .iter()
+        .map(|device| (device.device_id.as_str(), device))
+        .collect();
     let mut grouped: BTreeMap<String, PersistedCircuitFallback> = BTreeMap::new();
-    for device in devices {
-        let Some(token) = device.sqm_override.as_deref().map(str::trim) else {
+
+    for adj in overrides.circuit_adjustments() {
+        let lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+            device_id,
+            sqm_override,
+        } = adj
+        else {
+            continue;
+        };
+        let Some(token) = sqm_override.as_deref().map(str::trim) else {
+            continue;
+        };
+        let Some(device) = devices_by_id.get(device_id.as_str()) else {
             continue;
         };
         if token.is_empty() || device.circuit_id.trim().is_empty() {
             continue;
         }
 
-        let entry =
-            grouped
-                .entry(device.circuit_id.clone())
-                .or_insert_with(|| PersistedCircuitFallback {
-                    sqm_override: token.to_string(),
-                    devices: Vec::new(),
-                });
+        let entry = grouped
+            .entry(device.circuit_id.clone())
+            .or_insert_with(|| PersistedCircuitFallback {
+                sqm_override: token.to_string(),
+                devices: Vec::new(),
+            });
+        if entry.sqm_override != token {
+            continue;
+        }
+        entry.devices.push((*device).clone());
+    }
+
+    for device in overrides.persistent_devices() {
+        let Some(token) = device.sqm_override.as_deref().map(str::trim) else {
+            continue;
+        };
+        let Some(current_device) = devices_by_id.get(device.device_id.as_str()) else {
+            continue;
+        };
+        if token.is_empty() || current_device.circuit_id.trim().is_empty() {
+            continue;
+        }
+
+        let entry = grouped
+            .entry(current_device.circuit_id.clone())
+            .or_insert_with(|| PersistedCircuitFallback {
+                sqm_override: token.to_string(),
+                devices: Vec::new(),
+            });
 
         if entry.sqm_override != token {
             continue;
         }
-        entry.devices.push(device.clone());
+        if entry
+            .devices
+            .iter()
+            .any(|existing| existing.device_id == current_device.device_id)
+        {
+            continue;
+        }
+        entry.devices.push((*current_device).clone());
     }
     grouped.into_iter().collect()
 }
@@ -370,7 +425,21 @@ mod tests {
         let mut d3 = sample_device("dev3", "c2");
         d3.sqm_override = Some("cake".to_string());
 
-        let grouped = group_circuit_fallbacks(&[d1, d2, d3]);
+        let mut overrides = OverrideFile::default();
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev1".to_string(),
+            Some("fq_codel".to_string())
+        ));
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev2".to_string(),
+            Some("fq_codel".to_string())
+        ));
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev3".to_string(),
+            Some("cake".to_string())
+        ));
+
+        let grouped = group_circuit_fallbacks(&overrides, &[d1, d2, d3]);
         assert_eq!(grouped.len(), 2);
         assert_eq!(
             grouped
@@ -405,7 +474,21 @@ mod tests {
         let mut d3 = sample_device("dev3", "c2");
         d3.sqm_override = Some(" ".to_string());
 
-        let grouped = group_circuit_fallbacks(&[d1, d2, d3]);
+        let mut overrides = OverrideFile::default();
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev1".to_string(),
+            Some("fq_codel".to_string())
+        ));
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev2".to_string(),
+            Some("cake".to_string())
+        ));
+        assert!(overrides.set_device_sqm_override_return_changed(
+            "dev3".to_string(),
+            Some(" ".to_string())
+        ));
+
+        let grouped = group_circuit_fallbacks(&overrides, &[d1, d2, d3]);
         assert_eq!(
             grouped
                 .get("c1")

src/rust/lqos_stormguard/src/lib.rs

@@ -11,6 +11,7 @@
 
 use lqos_bakery::BakeryCommands;
 use lqos_bus::StormguardDebugEntry;
+use lqos_config::NetworkJsonTransport;
 use lqos_queue_tracker::QUEUE_STRUCTURE_CHANGED_STORMGUARD;
 use parking_lot::Mutex;
 use std::time::Duration;
@@ -36,6 +37,7 @@ pub static STORMGUARD_DEBUG: Mutex<Vec<StormguardDebugEntry>> = Mutex::new(Vec::
 /// nothing to do.
 pub async fn start_stormguard(
     bakery: crossbeam_channel::Sender<BakeryCommands>,
+    network_map_provider: fn() -> Vec<(usize, NetworkJsonTransport)>,
 ) -> anyhow::Result<()> {
     let _ = tokio::time::sleep(Duration::from_secs(1)).await;
 
@@ -90,9 +92,12 @@ pub async fn start_stormguard(
         if let (Some(cfg), Some(tracker)) = (&config, &mut site_state_tracker) {
             let (active_ping_sample, active_ping_updated) = active_ping.latest();
             // Update all the ring buffers
-            tracker
-                .read_new_tick_data(cfg, active_ping_sample, active_ping_updated)
-                .await;
+            tracker.read_new_tick_data(
+                cfg,
+                active_ping_sample,
+                active_ping_updated,
+                network_map_provider(),
+            );
 
             // Check for state changes
             tracker.check_state(cfg);

src/rust/lqos_stormguard/src/site_state.rs

@@ -21,7 +21,8 @@ use crate::site_state::stormguard_state::StormguardState;
 use crate::{MOVING_AVERAGE_BUFFER_SIZE, READING_ACCUMULATOR_SIZE};
 use crossbeam_channel::Sender;
 use lqos_bakery::BakeryCommands;
-use lqos_bus::{BusRequest, BusResponse, StormguardDebugDirection, StormguardDebugEntry, TcHandle};
+use lqos_bus::{StormguardDebugDirection, StormguardDebugEntry, TcHandle};
+use lqos_config::NetworkJsonTransport;
 use lqos_queue_tracker::QUEUE_STRUCTURE;
 use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant};
@@ -185,63 +186,53 @@ impl SiteStateTracker {
         }
     }
 
-    pub async fn read_new_tick_data(
+    pub fn read_new_tick_data(
         &mut self,
         config: &StormguardConfig,
         active_ping_sample: Option<TimedRtt>,
         active_ping_updated: bool,
+        all_nodes: Vec<(usize, NetworkJsonTransport)>,
     ) {
-        let requests = vec![BusRequest::GetFullNetworkMap];
-        let Ok(responses) = lqos_bus::bus_request(requests).await else {
-            info!("Failed to get lqosd stats");
-            return;
-        };
-
         for site in self.sites.values_mut() {
             site.current_throughput = (0.0, 0.0);
             site.clear_tick_rtt_state();
         }
 
-        for response in responses {
-            let BusResponse::NetworkMap(all_nodes) = response else {
+        for (_, node_info) in all_nodes {
+            let Some(target) = self.sites.get_mut(&node_info.name) else {
                 continue;
             };
-            for (_, node_info) in all_nodes {
-                let Some(target) = self.sites.get_mut(&node_info.name) else {
-                    continue;
-                };
 
-                // Record throughput (Mbps)
-                let down_mbps = (node_info.current_throughput.0 as f64 * 8.0) / 1_000_000.0;
-                let up_mbps = (node_info.current_throughput.1 as f64 * 8.0) / 1_000_000.0;
-                target.throughput_down.add(down_mbps);
-                target.throughput_up.add(up_mbps);
-                target.current_throughput = (down_mbps, up_mbps);
+            // Record throughput (Mbps)
+            let down_mbps = (node_info.current_throughput.0 as f64 * 8.0) / 1_000_000.0;
+            let up_mbps = (node_info.current_throughput.1 as f64 * 8.0) / 1_000_000.0;
+            target.throughput_down.add(down_mbps);
+            target.throughput_up.add(up_mbps);
+            target.current_throughput = (down_mbps, up_mbps);
 
-                // Retransmits (as a percentage of TCP packets)
-                let retransmits_down = if node_info.current_tcp_packets.0 > 0 {
-                    node_info.current_retransmits.0 as f64 / node_info.current_tcp_packets.0 as f64
-                } else {
-                    0.0
-                };
-                let retransmits_up = if node_info.current_tcp_packets.1 > 0 {
-                    node_info.current_retransmits.1 as f64 / node_info.current_tcp_packets.1 as f64
-                } else {
-                    0.0
-                };
-                target.retransmits_down.add(retransmits_down);
-                target.retransmits_up.add(retransmits_up);
+            // Retransmits (as a percentage of TCP packets)
+            let retransmits_down = if node_info.current_tcp_packets.0 > 0 {
+                node_info.current_retransmits.0 as f64 / node_info.current_tcp_packets.0 as f64
+            } else {
+                0.0
+            };
+            let retransmits_up = if node_info.current_tcp_packets.1 > 0 {
+                node_info.current_retransmits.1 as f64 / node_info.current_tcp_packets.1 as f64
+            } else {
+                0.0
+            };
+            target.retransmits_down.add(retransmits_down);
+            target.retransmits_up.add(retransmits_up);
 
-                // Round-Trip Time
-                if !node_info.rtts.is_empty() {
-                    let mut my_round_trip_times = node_info.rtts.clone();
-                    my_round_trip_times.sort_by(|a, b| a.total_cmp(b));
-                    let samples = my_round_trip_times.len();
-                    let mut idx = ((samples as f32) * 0.9).floor() as usize;
-                    idx = idx.min(samples.saturating_sub(1));
-                    let p90 = my_round_trip_times[idx] as f64;
-                    target.record_passive_rtt_sample(p90);
-                }
+            // Round-Trip Time
+            if !node_info.rtts.is_empty() {
+                let mut my_round_trip_times = node_info.rtts.clone();
+                my_round_trip_times.sort_by(|a, b| a.total_cmp(b));
+                let samples = my_round_trip_times.len();
+                let mut idx = ((samples as f32) * 0.9).floor() as usize;
+                idx = idx.min(samples.saturating_sub(1));
+                let p90 = my_round_trip_times[idx] as f64;
+                target.record_passive_rtt_sample(p90);
             }
         }
 

src/rust/lqosd/src/main.rs

@@ -336,7 +336,12 @@ fn main() -> Result<()> {
                         Err(e) => error!("Insight control channel failed to start: {:#}", e),
                     }
 
-                    match lqos_stormguard::start_stormguard(bakery_sender_for_async).await {
+                    match lqos_stormguard::start_stormguard(
+                        bakery_sender_for_async,
+                        shaped_devices_tracker::full_network_map_snapshot,
+                    )
+                    .await
+                    {
                         Ok(_) => info!("StormGuard started successfully"),
                         Err(e) => error!("StormGuard failed to start: {:#}", e),
                     }

src/rust/lqosd/src/node_manager/local_api/flow_map.rs

@@ -1,4 +1,18 @@
 use crate::throughput_tracker::flow_data;
+
 pub fn flow_map_data() -> Vec<(f64, f64, String, u64, f32)> {
     flow_data::RECENT_FLOWS.lat_lon_endpoints()
 }
+
+pub fn endpoints_by_country_data() -> Vec<(
+    String,
+    lqos_utils::units::DownUpOrder<u64>,
+    [f32; 2],
+    String,
+)> {
+    flow_data::RECENT_FLOWS.country_summary()
+}
+
+pub fn endpoint_latlon_data() -> Vec<(f64, f64, String, u64, f32)> {
+    flow_data::RECENT_FLOWS.lat_lon_endpoints()
+}

src/rust/lqosd/src/node_manager/ws/ticker.rs

@@ -143,7 +143,7 @@ async fn one_second_cadence(
             ),
             ticker_with_timeout(
                 "endpoint_latlon",
-                endpoint_latlon::endpoint_latlon(channels.clone(), bus_tx.clone())
+                endpoint_latlon::endpoint_latlon(channels.clone())
             ),
             ticker_with_timeout(
                 "tree_summary",

src/rust/lqosd/src/node_manager/ws/ticker/endpoint_latlon.rs

@@ -1,15 +1,11 @@
 use std::sync::Arc;
 
+use crate::node_manager::local_api::flow_map;
 use crate::node_manager::ws::messages::WsResponse;
 use crate::node_manager::ws::publish_subscribe::PubSub;
 use crate::node_manager::ws::published_channels::PublishedChannels;
-use lqos_bus::{BusReply, BusRequest, BusResponse};
-use tokio::sync::mpsc::Sender;
 
-pub async fn endpoint_latlon(
-    channels: Arc<PubSub>,
-    bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
-) {
+pub async fn endpoint_latlon(channels: Arc<PubSub>) {
     if !channels
         .is_channel_alive(PublishedChannels::EndpointLatLon)
         .await
@@ -17,29 +13,10 @@ pub async fn endpoint_latlon(
         return;
     }
 
-    let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
-    let request = BusRequest::CurrentEndpointLatLon;
-    if let Err(e) = bus_tx.send((tx, request)).await {
-        tracing::warn!("EndpointLatLon: failed to send request to bus: {:?}", e);
-        return;
-    }
-    let replies = match rx.await {
-        Ok(r) => r,
-        Err(e) => {
-            tracing::warn!(
-                "EndpointLatLon: failed to receive response from bus: {:?}",
-                e
-            );
-            return;
-        }
+    let message = WsResponse::EndpointLatLon {
+        data: flow_map::endpoint_latlon_data(),
     };
-
-    for reply in replies.responses.into_iter() {
-        if let BusResponse::CurrentLatLon(points) = reply {
-            let message = WsResponse::EndpointLatLon { data: points };
-            channels
-                .send(PublishedChannels::EndpointLatLon, message)
-                .await;
-        }
-    }
+    channels
+        .send(PublishedChannels::EndpointLatLon, message)
+        .await;
 }

src/rust/lqosd/src/node_manager/ws/ticker/flow_endpoints.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::node_manager::local_api::flow_map;
 use crate::node_manager::ws::messages::{EtherProtocolsData, WsResponse};
 use crate::node_manager::ws::publish_subscribe::PubSub;
 use crate::node_manager::ws::published_channels::PublishedChannels;
@@ -8,7 +9,7 @@ use tokio::sync::mpsc::Sender;
 
 pub async fn endpoints_by_country(
     channels: Arc<PubSub>,
-    bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
+    _bus_tx: Sender<(tokio::sync::oneshot::Sender<lqos_bus::BusReply>, BusRequest)>,
 ) {
     if !channels
         .is_channel_alive(PublishedChannels::EndpointsByCountry)
@@ -17,30 +18,12 @@ pub async fn endpoints_by_country(
         return;
     }
 
-    let (tx, rx) = tokio::sync::oneshot::channel::<BusReply>();
-    let request = BusRequest::CurrentEndpointsByCountry;
-    if let Err(e) = bus_tx.send((tx, request)).await {
-        tracing::warn!("EndpointsByCountry: failed to send request to bus: {:?}", e);
-        return;
-    }
-    let replies = match rx.await {
-        Ok(r) => r,
-        Err(e) => {
-            tracing::warn!(
-                "EndpointsByCountry: failed to receive throughput from bus: {:?}",
-                e
-            );
-            return;
-        }
+    let message = WsResponse::EndpointsByCountry {
+        data: flow_map::endpoints_by_country_data(),
     };
-    for reply in replies.responses.into_iter() {
-        if let BusResponse::CurrentEndpointsByCountry(countries) = reply {
-            let message = WsResponse::EndpointsByCountry { data: countries };
-            channels
-                .send(PublishedChannels::EndpointsByCountry, message)
-                .await;
-        }
-    }
+    channels
+        .send(PublishedChannels::EndpointsByCountry, message)
+        .await;
 }
 
 pub async fn ether_protocols(

src/rust/lqosd/src/shaped_devices_tracker/mod.rs

@@ -150,17 +150,17 @@ pub fn get_one_network_map_layer(parent_idx: usize) -> BusResponse {
     }
 }
 
-pub fn get_full_network_map() -> BusResponse {
+pub fn full_network_map_snapshot() -> Vec<(usize, NetworkJsonTransport)> {
     let nj = NETWORK_JSON.read();
-    let data = {
-        nj.get_nodes_when_ready()
-            .iter()
-            .enumerate()
-            .map(|(i, n)| (i, n.clone_to_transit()))
-            .collect::<Vec<(usize, NetworkJsonTransport)>>()
-    };
+    nj.get_nodes_when_ready()
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (i, n.clone_to_transit()))
+        .collect()
+}
 
-    BusResponse::NetworkMap(data)
+pub fn get_full_network_map() -> BusResponse {
+    BusResponse::NetworkMap(full_network_map_snapshot())
 }
 
 pub fn get_top_n_root_queues(n_queues: usize) -> BusResponse {

src/rust/lqosd/src/throughput_tracker/mod.rs

@@ -1212,7 +1212,6 @@ pub fn current_endpoints_by_country() -> BusResponse {
 /// Current endpoint lat/lon
 pub fn current_lat_lon() -> BusResponse {
     let summary = flow_data::RECENT_FLOWS.lat_lon_endpoints();
-    debug!("CurrentLatLon payload: {:?}", summary);
     BusResponse::CurrentLatLon(summary)
 }
 

src/rust/lqosd/src/treeguard/actor.rs

@@ -23,12 +23,15 @@ use lqos_overrides::{NetworkAdjustment, OverrideFile, OverrideLayer, OverrideSto
 use lqos_utils::hash_to_i64;
 use lqos_utils::units::DownUpOrder;
 use lqos_utils::unix_time::{time_since_boot, unix_now};
+use parking_lot::RwLock;
 use std::collections::VecDeque;
 use std::sync::OnceLock;
 use std::time::{Duration, Instant};
 use tracing::{debug, warn};
 
 static TREEGUARD_SENDER: OnceLock<Sender<TreeguardCommand>> = OnceLock::new();
+static TREEGUARD_STATUS_CACHE: OnceLock<RwLock<TreeguardStatusData>> = OnceLock::new();
+static TREEGUARD_ACTIVITY_CACHE: OnceLock<RwLock<Vec<TreeguardActivityEntry>>> = OnceLock::new();
 
 const ACTIVITY_RING_CAPACITY: usize = 200;
 const UTIL_EWMA_ALPHA: f64 = 0.1;
@@ -60,6 +63,9 @@ pub(crate) fn start_treeguard_actor(
         return Ok(());
     }
 
+    let _ = TREEGUARD_STATUS_CACHE.set(RwLock::new(empty_status_snapshot()));
+    let _ = TREEGUARD_ACTIVITY_CACHE.set(RwLock::new(Vec::new()));
+
     let (tx, rx) = crossbeam_channel::bounded::<TreeguardCommand>(64);
     let _ = TREEGUARD_SENDER.set(tx);
 
@@ -70,6 +76,14 @@ pub(crate) fn start_treeguard_actor(
     Ok(())
 }
 
+pub(crate) fn cached_status_snapshot() -> Option<TreeguardStatusData> {
+    Some(TREEGUARD_STATUS_CACHE.get()?.read().clone())
+}
+
+pub(crate) fn cached_activity_snapshot() -> Option<Vec<TreeguardActivityEntry>> {
+    Some(TREEGUARD_ACTIVITY_CACHE.get()?.read().clone())
+}
+
 /// Requests a status snapshot from the TreeGuard actor.
 ///
 /// This function is not pure: it sends a message to the TreeGuard actor thread.
@@ -110,18 +124,9 @@ fn treeguard_actor_loop(
 ) {
     debug!("TreeGuard actor started");
 
-    let mut status = TreeguardStatusData {
-        enabled: false,
-        dry_run: true,
-        cpu_max_pct: None,
-        managed_nodes: 0,
-        managed_circuits: 0,
-        virtualized_nodes: 0,
-        fq_codel_circuits: 0,
-        last_action_summary: None,
-        warnings: Vec::new(),
-    };
+    let mut status = empty_status_snapshot();
     let mut activity: VecDeque<TreeguardActivityEntry> = VecDeque::new();
+    update_cached_snapshots(&status, &activity);
 
     let mut runtime_state = TreeguardRuntimeState::default();
 
@@ -143,6 +148,7 @@ fn treeguard_actor_loop(
                     &mut tick_seconds,
                     &mut runtime_state,
                 );
+                update_cached_snapshots(&status, &activity);
             }
             Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
                 warn!("TreeGuard actor command channel disconnected; exiting actor");
@@ -171,6 +177,32 @@ fn handle_command(
     }
 }
 
+fn empty_status_snapshot() -> TreeguardStatusData {
+    TreeguardStatusData {
+        enabled: false,
+        dry_run: true,
+        cpu_max_pct: None,
+        managed_nodes: 0,
+        managed_circuits: 0,
+        virtualized_nodes: 0,
+        fq_codel_circuits: 0,
+        last_action_summary: None,
+        warnings: Vec::new(),
+    }
+}
+
+fn update_cached_snapshots(
+    status: &TreeguardStatusData,
+    activity: &VecDeque<TreeguardActivityEntry>,
+) {
+    if let Some(cache) = TREEGUARD_STATUS_CACHE.get() {
+        *cache.write() = status.clone();
+    }
+    if let Some(cache) = TREEGUARD_ACTIVITY_CACHE.get() {
+        *cache.write() = activity.iter().cloned().rev().collect();
+    }
+}
+
 #[derive(Default)]
 struct TreeguardRuntimeState {
     link_states: FxHashMap<String, LinkState>,
@@ -364,17 +396,7 @@ fn run_tick(
 
     let operator_sqm_device_overrides: FxHashSet<String> = operator_overrides_snapshot
         .as_ref()
-        .map(|o| {
-            o.persistent_devices()
-                .iter()
-                .filter(|d| {
-                    d.sqm_override
-                        .as_deref()
-                        .is_some_and(|t| !t.trim().is_empty())
-                })
-                .map(|d| d.device_id.clone())
-                .collect()
-        })
+        .map(overrides_sqm_device_ids)
         .unwrap_or_default();
 
     // --- Link sampling + decisions (virtualization) ---
@@ -1310,11 +1332,7 @@ fn run_tick(
     // Cleanup for removed circuits or disabled circuits/TreeGuard.
     if !manage_circuits {
         let removed: Vec<String> = match OverrideStore::load_layer(OverrideLayer::Treeguard) {
-            Ok(of) => of
-                .persistent_devices()
-                .iter()
-                .map(|d| d.device_id.clone())
-                .collect(),
+            Ok(of) => overrides_sqm_device_ids(&of).into_iter().collect(),
             Err(e) => {
                 status.warnings.push(format!(
                     "TreeGuard circuits: unable to load TreeGuard overrides for cleanup: {e}"
@@ -1352,12 +1370,7 @@ fn run_tick(
         // Reconcile device IDs removed from allowlisted circuits.
         let treeguard_device_ids_with_overrides: FxHashSet<String> = treeguard_overrides_snapshot
             .as_ref()
-            .map(|of| {
-                of.persistent_devices()
-                    .iter()
-                    .map(|d| d.device_id.clone())
-                    .collect()
-            })
+            .map(overrides_sqm_device_ids)
             .unwrap_or_default();
         let removed: Vec<String> = treeguard_device_ids_with_overrides
             .iter()
@@ -1667,7 +1680,9 @@ fn run_tick(
                     let mut persisted_ok = false;
                     let mut can_apply_live = true;
                     if tg.circuits.persist_sqm_overrides {
-                        match overrides::set_devices_sqm_override(&devices, &token) {
+                        let device_ids: Vec<String> =
+                            devices.iter().map(|device| device.device_id.clone()).collect();
+                        match overrides::set_devices_sqm_override(&device_ids, &token) {
                             Ok(_) => {
                                 persisted_ok = true;
                             }
@@ -1795,7 +1810,73 @@ fn overrides_node_virtual(overrides: &OverrideFile, node_name: &str) -> Option<b
         })
 }
 
-/// Infers an SQM override token for a circuit, preferring persisted device overlays.
+/// Returns the current SQM override token for `device_id`, if present.
+///
+/// This function is pure: it has no side effects.
+fn overrides_device_sqm(overrides: &OverrideFile, device_id: &str) -> Option<String> {
+    for adj in overrides.circuit_adjustments() {
+        if let lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+            device_id: current,
+            sqm_override,
+        } = adj
+        {
+            if current != device_id {
+                continue;
+            }
+            return sqm_override
+                .as_deref()
+                .map(str::trim)
+                .filter(|token| !token.is_empty())
+                .map(str::to_string);
+        }
+    }
+
+    for dev in overrides.persistent_devices() {
+        if dev.device_id != device_id {
+            continue;
+        }
+        if let Some(token) = dev.sqm_override.as_deref().map(str::trim)
+            && !token.is_empty()
+        {
+            return Some(token.to_string());
+        }
+    }
+
+    None
+}
+
+/// Returns the set of device IDs carrying an SQM override in an overrides file.
+///
+/// This function is pure: it has no side effects.
+fn overrides_sqm_device_ids(overrides: &OverrideFile) -> FxHashSet<String> {
+    let mut out = FxHashSet::default();
+    for adj in overrides.circuit_adjustments() {
+        if let lqos_overrides::CircuitAdjustment::DeviceAdjustSqm {
+            device_id,
+            sqm_override,
+        } = adj
+            && sqm_override
+                .as_deref()
+                .map(str::trim)
+                .is_some_and(|token| !token.is_empty())
+        {
+            out.insert(device_id.clone());
+        }
+    }
+    for dev in overrides.persistent_devices() {
+        if dev
+            .sqm_override
+            .as_deref()
+            .map(str::trim)
+            .is_some_and(|token| !token.is_empty())
+        {
+            out.insert(dev.device_id.clone());
+        }
+    }
+    out
+}
+
+/// Infers an SQM override token for a circuit, preferring persisted override entries.
 ///
 /// This function is pure: it has no side effects.
 fn infer_circuit_sqm_override_token(
@@ -1810,15 +1891,9 @@ fn infer_circuit_sqm_override_token(
         .collect();
 
     if let Some(overrides) = overrides {
-        for dev in overrides.persistent_devices() {
-            if !circuit_device_ids.contains(dev.device_id.as_str()) {
-                continue;
-            }
-            if let Some(token) = dev.sqm_override.as_deref() {
-                let token = token.trim();
-                if !token.is_empty() {
-                    return Some(token.to_string());
-                }
+        for device_id in &circuit_device_ids {
+            if let Some(token) = overrides_device_sqm(overrides, device_id) {
+                return Some(token);
             }
         }
     }

src/rust/lqosd/src/treeguard/overrides.rs

@@ -3,7 +3,6 @@
 //! This module writes TreeGuard-owned changes to `lqos_overrides.treeguard.json`.
 
 use crate::treeguard::TreeguardError;
-use lqos_config::ShapedDevice;
 use lqos_overrides::{NetworkAdjustment, OverrideFile, OverrideLayer, OverrideStore};
 
 /// Returns the current virtual override value for `node_name`, if present.
@@ -75,14 +74,13 @@ pub(crate) fn clear_node_virtual(node_name: &str) -> Result<bool, TreeguardError
     Ok(true)
 }
 
-/// Persists a per-device SQM override token for a list of devices by storing persistent shaped
-/// device overlays.
+/// 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`.
 ///
 /// Returns `Ok(true)` if the file was changed, `Ok(false)` if no change was needed.
 pub(crate) fn set_devices_sqm_override(
-    base_devices: &[ShapedDevice],
+    device_ids: &[String],
     sqm_override: &str,
 ) -> Result<bool, TreeguardError> {
     let mut overrides = OverrideStore::load_layer(OverrideLayer::Treeguard).map_err(|e| {
@@ -92,10 +90,14 @@ pub(crate) fn set_devices_sqm_override(
     })?;
 
     let mut changed = false;
-    for base_device in base_devices {
-        let mut device = base_device.clone();
-        device.sqm_override = Some(sqm_override.to_string());
-        if overrides.add_persistent_shaped_device_return_changed(device) {
+    for device_id in device_ids {
+        if overrides.set_device_sqm_override_return_changed(
+            device_id.clone(),
+            Some(sqm_override.to_string()),
+        ) {
+            changed = true;
+        }
+        if overrides.remove_persistent_shaped_device_by_device_count(device_id) > 0 {
             changed = true;
         }
     }
@@ -112,7 +114,7 @@ pub(crate) fn set_devices_sqm_override(
     Ok(true)
 }
 
-/// Removes any persistent shaped device overlay entries for a list of device IDs.
+/// Removes any persisted SQM override entries for a list of device IDs.
 ///
 /// This function is not pure: it reads and writes `lqos_overrides.treeguard.json`.
 ///
@@ -126,8 +128,9 @@ pub(crate) fn clear_device_overrides(device_ids: &[String]) -> Result<bool, Tree
 
     let mut removed_any = false;
     for device_id in device_ids {
-        let removed = overrides.remove_persistent_shaped_device_by_device_count(device_id);
-        if removed > 0 {
+        let removed_adjustments = overrides.remove_device_sqm_override_by_device_count(device_id);
+        let removed_devices = overrides.remove_persistent_shaped_device_by_device_count(device_id);
+        if removed_adjustments > 0 || removed_devices > 0 {
             removed_any = true;
         }
     }

src/rust/lqosd/src/treeguard/status.rs

@@ -9,6 +9,10 @@ use lqos_config::load_config;
 /// This function is not pure: it sends a request to the TreeGuard actor, and may fall back to
 /// reading the current configuration from disk (via the config cache).
 pub async fn treeguard_status_snapshot() -> TreeguardStatusData {
+    if let Some(snapshot) = actor::cached_status_snapshot() {
+        return snapshot;
+    }
+
     if let Some(snapshot) = actor::request_status_snapshot().await {
         return snapshot;
     }
@@ -71,5 +75,9 @@ pub async fn treeguard_status_snapshot() -> TreeguardStatusData {
 /// This function is not pure: it sends a request to the TreeGuard actor.
 /// If the actor isn't available, it returns an empty list.
 pub async fn treeguard_activity_snapshot() -> Vec<TreeguardActivityEntry> {
+    if let Some(snapshot) = actor::cached_activity_snapshot() {
+        return snapshot;
+    }
+
     actor::request_activity_snapshot().await.unwrap_or_default()
 }

src/scheduler.py

@@ -12,7 +12,7 @@ from liblqos_python import automatic_import_uisp, automatic_import_splynx, queue
     automatic_import_powercode, automatic_import_sonar, influx_db_enabled, get_libreqos_directory, \
     blackboard_finish, blackboard_submit, automatic_import_wispgate, enable_insight_topology, insight_topology_role, \
     automatic_import_netzur, automatic_import_visp, calculate_hash, efficiency_core_ids, scheduler_alive, scheduler_error, \
-    overrides_persistent_devices_effective, overrides_circuit_adjustments, overrides_network_adjustments_effective, \
+    overrides_persistent_devices_effective, overrides_circuit_adjustments_effective, overrides_network_adjustments_effective, \
     scheduler_output
 
 from apscheduler.schedulers.background import BlockingScheduler
@@ -347,7 +347,7 @@ def apply_lqos_overrides():
 
     # 2) Circuit adjustments: speed changes, removals, reparenting
     try:
-        adjustments = overrides_circuit_adjustments()
+        adjustments = overrides_circuit_adjustments_effective()
     except Exception as e:
         print(f"Failed to read circuit adjustments: {e}")
         adjustments = []
@@ -360,6 +360,11 @@ def apply_lqos_overrides():
         except Exception:
             return current_str
 
+    def set_row_value(row, index, value):
+        while len(row) <= index:
+            row.append('')
+        row[index] = value
+
     if adjustments:
         for adj in adjustments:
             t = adj.get('type')
@@ -381,6 +386,15 @@ def apply_lqos_overrides():
                         r[9] = set_if_some(adj.get('min_upload_bandwidth'), r[9] if len(r) > 9 else '')
                         r[11] = set_if_some(adj.get('max_upload_bandwidth'), r[11] if len(r) > 11 else '')
                         changed = True
+            elif t == 'device_adjust_sqm':
+                did = adj.get('device_id', '')
+                sqm_override = (adj.get('sqm_override') or '').strip()
+                for r in merged_rows:
+                    if len(r) >= 3 and r[2] == did:
+                        current_sqm = r[13] if len(r) > 13 else ''
+                        if current_sqm != sqm_override:
+                            set_row_value(r, 13, sqm_override)
+                            changed = True
             elif t == 'remove_circuit':
                 cid = adj.get('circuit_id', '')
                 before = len(merged_rows)

src/systemd_hotfix.sh

@@ -0,0 +1,254 @@
+#!/bin/bash
+
+set -euo pipefail
+
+HOTFIX_BASE_URL="${HOTFIX_BASE_URL:-https://downloads.libreqos.com}"
+HOTFIX_PACKAGE_VERSION="${HOTFIX_PACKAGE_VERSION:-255.4-1ubuntu8.12+libreqos1}"
+SUPPORTED_UBUNTU_SYSTEMD_VERSION_GLOBS="${SUPPORTED_UBUNTU_SYSTEMD_VERSION_GLOBS:-255.4-1ubuntu8 255.4-1ubuntu8.*}"
+HOTFIX_MARKER="${HOTFIX_MARKER:-/opt/libreqos/src/.systemd_hotfix_installed}"
+ARCH="${ARCH:-amd64}"
+
+HOTFIX_CORE_PACKAGES=(
+  "libsystemd0"
+  "libsystemd-shared"
+  "libudev1"
+  "systemd-dev"
+  "systemd"
+  "systemd-sysv"
+  "systemd-resolved"
+  "systemd-timesyncd"
+  "udev"
+)
+
+HOTFIX_OPTIONAL_PACKAGES=(
+  "libpam-systemd"
+  "libnss-systemd"
+  "libnss-resolve"
+  "libnss-myhostname"
+)
+
+usage() {
+    cat <<EOF
+Usage: $0 <command>
+
+Commands:
+  status        Show whether this host should be offered the Noble systemd hotfix
+  should-offer  Exit 0 when the hotfix should be offered on this host
+  download      Download the hotfix bundle into a temporary directory
+  install       Download and install the hotfix bundle with apt
+  packages      Print the package filenames expected for this host
+  urls          Print the expected package URLs for this host
+
+Environment:
+  HOTFIX_BASE_URL                  Base URL hosting the .deb files
+  HOTFIX_PACKAGE_VERSION           Backported package version suffix
+  SUPPORTED_UBUNTU_SYSTEMD_VERSION_GLOBS Space-separated stock Ubuntu version globs eligible for replacement
+  HOTFIX_MARKER                    Marker file written after install
+  ARCH                             Debian architecture suffix, defaults to amd64
+EOF
+}
+
+log() {
+    printf '%s\n' "$*"
+}
+
+fail() {
+    printf 'Error: %s\n' "$*" >&2
+    exit 1
+}
+
+require_command() {
+    command -v "$1" >/dev/null 2>&1 || fail "Required command not found: $1"
+}
+
+current_systemd_version() {
+    dpkg-query -W -f='${Version}\n' systemd 2>/dev/null || true
+}
+
+package_is_installed() {
+    local package="$1"
+    dpkg-query -W -f='${db:Status-Abbrev}\n' "$package" 2>/dev/null | grep -q '^ii'
+}
+
+package_arch_suffix() {
+    local package="$1"
+
+    case "$package" in
+        systemd-dev)
+            printf 'all\n'
+            ;;
+        *)
+            printf '%s\n' "$ARCH"
+            ;;
+    esac
+}
+
+package_filename() {
+    local package="$1"
+    printf '%s_%s_%s.deb\n' \
+        "$package" \
+        "$HOTFIX_PACKAGE_VERSION" \
+        "$(package_arch_suffix "$package")"
+}
+
+resolved_hotfix_packages() {
+    local package
+
+    for package in "${HOTFIX_CORE_PACKAGES[@]}"; do
+        package_filename "$package"
+    done
+
+    for package in "${HOTFIX_OPTIONAL_PACKAGES[@]}"; do
+        if package_is_installed "$package"; then
+            package_filename "$package"
+        fi
+    done
+}
+
+is_supported_os() {
+    [[ -r /etc/os-release ]] || return 1
+    # shellcheck disable=SC1091
+    . /etc/os-release
+    [[ "${ID:-}" == "ubuntu" && "${VERSION_ID:-}" == "24.04" && "${VERSION_CODENAME:-}" == "noble" ]]
+}
+
+is_installed_hotfix() {
+    local version
+    version="$(current_systemd_version)"
+    [[ "$version" == *"+libreqos"* ]]
+}
+
+is_supported_stock_version() {
+    local version supported
+    version="$(current_systemd_version)"
+    for supported in $SUPPORTED_UBUNTU_SYSTEMD_VERSION_GLOBS; do
+        [[ "$version" == $supported ]] && return 0
+    done
+    return 1
+}
+
+uses_systemd_networkd() {
+    local enabled_state active_state
+
+    enabled_state="$(systemctl is-enabled systemd-networkd 2>/dev/null || true)"
+    active_state="$(systemctl is-active systemd-networkd 2>/dev/null || true)"
+
+    [[ "$enabled_state" == "enabled" || "$enabled_state" == "static" || "$active_state" == "active" ]]
+}
+
+print_urls() {
+    local package
+    while IFS= read -r package; do
+        printf '%s/%s\n' "$HOTFIX_BASE_URL" "$package"
+    done < <(resolved_hotfix_packages)
+}
+
+print_packages() {
+    resolved_hotfix_packages
+}
+
+status() {
+    local version
+    version="$(current_systemd_version)"
+
+    if ! is_supported_os; then
+        log "Host is not Ubuntu 24.04 Noble. Hotfix not applicable."
+        return 1
+    fi
+
+    if [[ -z "$version" ]]; then
+        log "systemd is not installed via dpkg query. Hotfix not applicable."
+        return 1
+    fi
+
+    if ! uses_systemd_networkd; then
+        log "systemd-networkd is not enabled or active. Hotfix not applicable."
+        return 1
+    fi
+
+    if is_installed_hotfix; then
+        log "LibreQoS hotfix already installed: $version"
+        return 1
+    fi
+
+    if is_supported_stock_version; then
+        log "Hotfix should be offered. Installed systemd version: $version"
+        return 0
+    fi
+
+    log "Installed systemd version is not in the supported replacement list: $version"
+    return 1
+}
+
+download_bundle() {
+    local workdir package
+    require_command curl
+    workdir="$(mktemp -d /tmp/libreqos-systemd-hotfix.XXXXXX)"
+    while IFS= read -r package; do
+        curl -fL --retry 3 --output "${workdir}/${package}" "${HOTFIX_BASE_URL}/${package}"
+    done < <(resolved_hotfix_packages)
+    printf '%s\n' "$workdir"
+}
+
+write_marker() {
+    local package
+    {
+        printf 'installed_at=%s\n' "$(date -Iseconds)"
+        printf 'package_version=%s\n' "$HOTFIX_PACKAGE_VERSION"
+        printf 'base_url=%s\n' "$HOTFIX_BASE_URL"
+        printf 'systemd_version=%s\n' "$(current_systemd_version)"
+        while IFS= read -r package; do
+            printf 'package_file=%s\n' "$package"
+            printf 'package_url=%s/%s\n' "$HOTFIX_BASE_URL" "$package"
+        done < <(resolved_hotfix_packages)
+    } | sudo tee "$HOTFIX_MARKER" >/dev/null
+}
+
+install_bundle() {
+    local workdir package_paths=() package
+
+    status >/dev/null || fail "Hotfix is not applicable on this host."
+    require_command sudo
+    require_command apt-get
+
+    workdir="$(download_bundle)"
+    while IFS= read -r package; do
+        package_paths+=("${workdir}/${package}")
+    done < <(resolved_hotfix_packages)
+
+    sudo apt-get install -y "${package_paths[@]}"
+    write_marker
+    log "Hotfix installed. Reboot before validating networkd behavior."
+}
+
+main() {
+    local command="${1:-}"
+
+    case "$command" in
+        status)
+            status
+            ;;
+        should-offer)
+            status >/dev/null
+            ;;
+        download)
+            status >/dev/null || fail "Hotfix is not applicable on this host."
+            download_bundle
+            ;;
+        packages)
+            print_packages
+            ;;
+        install)
+            install_bundle
+            ;;
+        urls)
+            print_urls
+            ;;
+        *)
+            usage
+            exit 1
+            ;;
+    esac
+}
+
+main "$@"

src/test_integration_splynx.py

@@ -0,0 +1,113 @@
+import importlib
+import sys
+import types
+import unittest
+
+
+def install_splynx_stubs():
+    lqlib = types.ModuleType("liblqos_python")
+    lqlib.allowed_subnets = lambda: ["0.0.0.0/0"]
+    lqlib.ignore_subnets = lambda: []
+    lqlib.generated_pn_download_mbps = lambda: 1000
+    lqlib.generated_pn_upload_mbps = lambda: 1000
+    lqlib.circuit_name_use_address = lambda: False
+    lqlib.upstream_bandwidth_capacity_download_mbps = lambda: 1000
+    lqlib.upstream_bandwidth_capacity_upload_mbps = lambda: 1000
+    lqlib.find_ipv6_using_mikrotik = lambda: False
+    lqlib.exclude_sites = lambda: []
+    lqlib.bandwidth_overhead_factor = lambda: 1.0
+    lqlib.committed_bandwidth_multiplier = lambda: 1.0
+    lqlib.exception_cpes = lambda: []
+    lqlib.promote_to_root_list = lambda: []
+    lqlib.client_bandwidth_multiplier = lambda: 1.0
+    lqlib.splynx_api_key = lambda: ""
+    lqlib.splynx_api_secret = lambda: ""
+    lqlib.splynx_api_url = lambda: "http://example.invalid"
+    lqlib.splynx_strategy = lambda: "flat"
+    lqlib.overwrite_network_json_always = False
+    sys.modules["liblqos_python"] = lqlib
+
+
+install_splynx_stubs()
+integrationSplynx = importlib.import_module("integrationSplynx")
+integrationCommon = importlib.import_module("integrationCommon")
+
+
+class TestIntegrationSplynxStableIds(unittest.TestCase):
+    def test_stable_splynx_device_id_uses_service_id(self):
+        self.assertEqual(
+            integrationSplynx.stable_splynx_device_id(93),
+            "splynx_service_93",
+        )
+
+    def test_stable_splynx_device_id_supports_future_equipment_suffix(self):
+        self.assertEqual(
+            integrationSplynx.stable_splynx_device_id(93, "5045"),
+            "splynx_service_93_equipment_5045",
+        )
+
+    def test_create_client_and_device_uses_stable_device_id(self):
+        net = integrationCommon.NetworkGraph()
+        circuit_id = integrationSplynx.createClientAndDevice(
+            net,
+            serviceItem={
+                "id": 93,
+                "customer_id": 29547,
+                "tariff_id": 32,
+                "mac": "2CC81BBDC9AB, 00E04C687E6A",
+            },
+            cust_id_to_name={29547: "Charles Massey and Alex Soto"},
+            downloadForTariffID={32: 330.0},
+            uploadForTariffID={32: 330.0},
+            parent_node_id="ap_244",
+            ipv4_list=["66.185.224.210", "66.185.224.210/32"],
+            ipv6_list=[],
+        )
+
+        self.assertEqual(circuit_id, 93)
+        self.assertEqual(len(net.nodes), 3)
+        client = net.nodes[1]
+        device = net.nodes[2]
+        self.assertEqual(client.id, 93)
+        self.assertEqual(device.id, "splynx_service_93")
+        self.assertEqual(device.parentId, 93)
+        self.assertEqual(device.ipv4, ["66.185.224.210", "66.185.224.210/32"])
+
+    def test_online_fallback_uses_stable_device_id(self):
+        net = integrationCommon.NetworkGraph()
+        matched = integrationSplynx.create_devices_from_online_for_unhandled_services(
+            net,
+            allServices=[
+                {
+                    "id": 93,
+                    "customer_id": 29547,
+                    "status": "active",
+                    "tariff_id": 32,
+                    "mac": "2CC81BBDC9AB",
+                }
+            ],
+            service_ids_handled=[],
+            customersOnline=[
+                {
+                    "service_id": 93,
+                    "ipv4": "66.185.224.210",
+                    "ipv6": "",
+                }
+            ],
+            cust_id_to_name={29547: "Charles Massey and Alex Soto"},
+            downloadForTariffID={32: 330.0},
+            uploadForTariffID={32: 330.0},
+            allocated_ipv4s={},
+            allocated_ipv6s={},
+            parent_selector=lambda service: "ap_244",
+            device_by_service_id={},
+        )
+
+        self.assertEqual(matched, 1)
+        self.assertEqual(len(net.nodes), 3)
+        self.assertEqual(net.nodes[2].id, "splynx_service_93")
+        self.assertEqual(net.nodes[2].ipv4, ["66.185.224.210"])
+
+
+if __name__ == "__main__":
+    unittest.main()

src/test_scheduler.py

@@ -32,7 +32,7 @@ def install_scheduler_stubs():
     lqlib.scheduler_error = Mock()
     lqlib.scheduler_output = Mock()
     lqlib.overrides_persistent_devices_effective = lambda: []
-    lqlib.overrides_circuit_adjustments = lambda: []
+    lqlib.overrides_circuit_adjustments_effective = lambda: []
     lqlib.overrides_network_adjustments_effective = lambda: []
     sys.modules["liblqos_python"] = lqlib
 
@@ -228,5 +228,60 @@ class TestSchedulerErrorReporting(unittest.TestCase):
         )
 
 
+class TestSchedulerOverrideMerge(unittest.TestCase):
+    def test_merge_rows_replaces_matching_device_id(self):
+        existing = [["93", "Name", "splynx_service_93", "Name", "AP", "MAC", "1.1.1.1", "", "1", "1", "330", "330", "", ""]]
+        override = [["93", "Name", "splynx_service_93", "Name", "AP", "MAC", "1.1.1.1/32", "", "1", "1", "330", "330", "", "fq_codel/fq_codel"]]
+
+        merged, changed = scheduler.merge_rows_replace_by_device_id(existing, override)
+
+        self.assertTrue(changed)
+        self.assertEqual(len(merged), 1)
+        self.assertEqual(merged[0][2], "splynx_service_93")
+        self.assertEqual(merged[0][6], "1.1.1.1/32")
+        self.assertEqual(merged[0][13], "fq_codel/fq_codel")
+
+    def test_merge_rows_appends_unmatched_non_splynx_override(self):
+        existing = [["93", "Name", "splynx_service_93", "Name", "AP", "MAC", "1.1.1.1", "", "1", "1", "330", "330", "", ""]]
+        override = [["145", "Other", "legacy_device_1", "Other", "AP", "MAC2", "2.2.2.2", "", "1", "1", "300", "300", "", ""]]
+
+        merged, changed = scheduler.merge_rows_replace_by_device_id(existing, override)
+
+        self.assertTrue(changed)
+        self.assertEqual(len(merged), 2)
+        self.assertEqual(merged[1][2], "legacy_device_1")
+
+    def test_apply_lqos_overrides_device_adjust_sqm_only_updates_sqm_column(self):
+        header = [
+            "Circuit ID", "Circuit Name", "Device ID", "Device Name", "Parent Node", "MAC",
+            "IPv4", "IPv6", "Download Min Mbps", "Upload Min Mbps", "Download Max Mbps",
+            "Upload Max Mbps", "Comment", "SQM"
+        ]
+        rows = [[
+            "93", "Name", "splynx_service_93", "Name", "AP", "MAC", "1.1.1.1", "",
+            "1", "1", "330", "330", "", ""
+        ]]
+
+        with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"):
+            with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
+                with patch.object(scheduler, "overrides_persistent_devices_effective", return_value=[]):
+                    with patch.object(
+                        scheduler,
+                        "overrides_circuit_adjustments_effective",
+                        return_value=[{
+                            "type": "device_adjust_sqm",
+                            "device_id": "splynx_service_93",
+                            "sqm_override": "fq_codel/fq_codel",
+                        }],
+                    ):
+                        with patch.object(scheduler, "write_shaped_devices_csv") as mock_write:
+                            scheduler.apply_lqos_overrides()
+
+        written_rows = mock_write.call_args.args[2]
+        self.assertEqual(written_rows[0][10], "330")
+        self.assertEqual(written_rows[0][11], "330")
+        self.assertEqual(written_rows[0][13], "fq_codel/fq_codel")
+
+
 if __name__ == "__main__":
     unittest.main()