Merge pull request #1083 from LibreQoE/fix_1082

Prevent transient XDP IP mapping urgent warnings during startup

src/LibreQoS.py

@@ -51,13 +51,37 @@ from liblqos_python import is_lqosd_alive, clear_ip_mappings, delete_ip_mapping,
     shaping_cpu_count, \
     Bakery
 
-# Optional: urgent issue submission (available in newer liblqos_python)
+# Optional: urgent issue helpers (available in newer liblqos_python)
 try:
     from liblqos_python import submit_urgent_issue  # type: ignore
 except Exception:
     def submit_urgent_issue(*_args, **_kwargs):
         return False
 
+try:
+    from liblqos_python import clear_urgent_issue_by_identity  # type: ignore
+except ImportError:
+    _missing_clear_urgent_issue_by_identity_warned = False
+
+    def clear_urgent_issue_by_identity(*_args, **_kwargs):
+        global _missing_clear_urgent_issue_by_identity_warned
+        if not _missing_clear_urgent_issue_by_identity_warned:
+            logging.warning("Unable to clear recovered urgent issues: liblqos_python helper is unavailable")
+            _missing_clear_urgent_issue_by_identity_warned = True
+        return False
+
+try:
+    from liblqos_python import xdp_ip_mapping_ready  # type: ignore
+except ImportError:
+    _missing_xdp_ip_mapping_ready_warned = False
+
+    def xdp_ip_mapping_ready():
+        global _missing_xdp_ip_mapping_ready_warned
+        if not _missing_xdp_ip_mapping_ready_warned:
+            logging.warning("Unable to check XDP IP mapping readiness: liblqos_python helper is unavailable")
+            _missing_xdp_ip_mapping_ready_warned = True
+        return True
+
 try:
     from liblqos_python import calculate_topology_source_generation  # type: ignore
 except Exception:
@@ -78,6 +102,11 @@ class ValidationFailure(RefreshFailure):
     pass
 
 
+XDP_IP_MAPPING_APPLY_FAILED = "XDP_IP_MAPPING_APPLY_FAILED"
+XDP_MAPPING_READY_TIMEOUT_SECONDS = 5.0
+XDP_MAPPING_READY_INTERVAL_SECONDS = 0.1
+
+
 def report_refresh_failure(code, message, context=None, dedupe_key=None):
     logging.error(message)
     print("ERROR: " + message)
@@ -98,6 +127,56 @@ def report_refresh_failure(code, message, context=None, dedupe_key=None):
         pass
     raise RefreshFailure(message)
 
+
+def wait_for_xdp_ip_mapping_ready(
+    ready=xdp_ip_mapping_ready,
+    sleep=time.sleep,
+    now=time.monotonic,
+    timeout_seconds=XDP_MAPPING_READY_TIMEOUT_SECONDS,
+    interval_seconds=XDP_MAPPING_READY_INTERVAL_SECONDS,
+):
+    deadline = now() + timeout_seconds
+    while now() < deadline:
+        if ready():
+            return
+        sleep(interval_seconds)
+    raise TimeoutError("XDP IP mapping BPF maps are not ready")
+
+
+def apply_xdp_ip_mappings(
+    ip_map_batch,
+    context,
+    ready=xdp_ip_mapping_ready,
+    sleep=time.sleep,
+    report_failure=report_refresh_failure,
+    clear_recovered_issue=clear_urgent_issue_by_identity,
+):
+    try:
+        wait_for_xdp_ip_mapping_ready(ready=ready, sleep=sleep)
+    except (TimeoutError, OSError) as exc:
+        report_failure(
+            XDP_IP_MAPPING_APPLY_FAILED,
+            "Failed to apply XDP IP mappings: " + str(exc),
+            context,
+            XDP_IP_MAPPING_APPLY_FAILED,
+        )
+        return
+    try:
+        ip_map_batch.finish_ip_mappings()
+        ip_map_batch.submit()
+    except OSError as exc:
+        report_failure(
+            XDP_IP_MAPPING_APPLY_FAILED,
+            "Failed to apply XDP IP mappings: " + str(exc),
+            context,
+            XDP_IP_MAPPING_APPLY_FAILED,
+        )
+        return
+    try:
+        clear_recovered_issue(XDP_IP_MAPPING_APPLY_FAILED, XDP_IP_MAPPING_APPLY_FAILED)
+    except OSError as exc:
+        logging.warning("Unable to clear recovered XDP IP mapping urgent issue: %s", exc)
+
 R2Q = 10
 #MAX_R2Q = 200_000
 MAX_R2Q = 60_000 # See https://lartc.vger.kernel.narkive.com/NKaH1ZNG/htb-quantum-of-class-100001-is-small-consider-r2q-change
@@ -2757,20 +2836,14 @@ def refreshShapers():
                 print("Observe mode active; skipping XDP-CPUMAP-TC IP filter apply after clearing mappings")
                 numXdpCommands = 0
             else:
-                ipMapBatch.finish_ip_mappings()
-                try:
-                    ipMapBatch.submit()
-                except Exception as e:
-                    report_refresh_failure(
-                        "XDP_IP_MAPPING_APPLY_FAILED",
-                        "Failed to apply XDP IP mappings: " + str(e),
-                        {
-                            "required_ip_mappings": requiredIpMappings,
-                            "queued_requests": numXdpCommands,
-                            "on_a_stick": on_a_stick(),
-                        },
-                        "XDP_IP_MAPPING_APPLY_FAILED",
-                    )
+                apply_xdp_ip_mappings(
+                    ipMapBatch,
+                    {
+                        "required_ip_mappings": requiredIpMappings,
+                        "queued_requests": numXdpCommands,
+                        "on_a_stick": on_a_stick(),
+                    },
+                )
             #for command in xdpCPUmapCommands:
             #	logging.info(command)
             #	commands = command.split(' ')

src/rust/lqos_bus/src/bus/protocol.rs

@@ -171,6 +171,19 @@ mod tests {
         assert_eq!(decoded.requests, session.requests);
     }
 
+    #[test]
+    fn cbor_round_trip_clear_urgent_issue_by_identity_request() {
+        let session = BusSession {
+            requests: vec![BusRequest::ClearUrgentIssueByIdentity {
+                code: "XDP_IP_MAPPING_APPLY_FAILED".to_string(),
+                dedupe_key: "XDP_IP_MAPPING_APPLY_FAILED".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

@@ -539,6 +539,14 @@ pub enum BusRequest {
     /// Clear a specific urgent issue by ID
     ClearUrgentIssue(u64),
 
+    /// Clear urgent issues matching a code and dedupe key
+    ClearUrgentIssueByIdentity {
+        /// Machine-readable issue code
+        code: String,
+        /// Dedupe key used when the issue was submitted
+        dedupe_key: String,
+    },
+
     /// Clear all urgent issues
     ClearAllUrgentIssues,
 

src/rust/lqos_python/src/lib.rs

@@ -942,6 +942,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(add_ip_mapping, m)?)?;
     m.add_function(wrap_pyfunction!(validate_shaped_devices, m)?)?;
     m.add_function(wrap_pyfunction!(wait_for_bus_ready, m)?)?;
+    m.add_function(wrap_pyfunction!(xdp_ip_mapping_ready, m)?)?;
     m.add_function(wrap_pyfunction!(is_libre_already_running, m)?)?;
     m.add_function(wrap_pyfunction!(create_lock_file, m)?)?;
     m.add_function(wrap_pyfunction!(free_lock_file, m)?)?;
@@ -1084,6 +1085,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(scheduler_error, m)?)?;
     m.add_function(wrap_pyfunction!(scheduler_output, m)?)?;
     m.add_function(wrap_pyfunction!(submit_urgent_issue, m)?)?;
+    m.add_function(wrap_pyfunction!(clear_urgent_issue_by_identity, m)?)?;
     m.add_function(wrap_pyfunction!(xdp_ip_mapping_capacity, m)?)?;
     m.add_function(wrap_pyfunction!(is_insight_enabled, m)?)?;
     m.add_function(wrap_pyfunction!(log_info, m)?)?;
@@ -1379,6 +1381,13 @@ fn xdp_ip_mapping_capacity() -> PyResult<usize> {
     Ok(lqos_sys::ip_mapping_capacity())
 }
 
+/// Returns whether the pinned BPF maps required for XDP IP mapping updates are ready.
+#[pyfunction]
+fn xdp_ip_mapping_ready() -> PyResult<bool> {
+    lqos_sys::ip_mapping_subsystem_ready()
+        .map_err(|e| PyOSError::new_err(format!("Unable to inspect XDP IP mapping maps: {e}")))
+}
+
 /// Requests Rust-side validation of `ShapedDevices.csv`
 #[pyfunction]
 fn validate_shaped_devices() -> PyResult<String> {
@@ -3865,6 +3874,52 @@ fn submit_urgent_issue(
     Ok(false)
 }
 
+/// Clear urgent issues matching a code and dedupe key.
+#[pyfunction]
+fn clear_urgent_issue_by_identity(_py: Python, code: String, dedupe_key: String) -> PyResult<bool> {
+    if let Ok(reply) = run_query(vec![BusRequest::ClearUrgentIssueByIdentity {
+        code: code.clone(),
+        dedupe_key: dedupe_key.clone(),
+    }]) {
+        for resp in reply.iter() {
+            if let BusResponse::Ack = resp {
+                return Ok(true);
+            }
+        }
+    }
+
+    let Ok(reply) = run_query(vec![BusRequest::GetUrgentIssues]) else {
+        return Ok(false);
+    };
+    let mut ids_to_clear = Vec::new();
+    for resp in reply.iter() {
+        if let BusResponse::UrgentIssues(issues) = resp {
+            ids_to_clear.extend(
+                issues
+                    .iter()
+                    .filter(|issue| {
+                        issue.code == code
+                            && issue.dedupe_key.as_deref().unwrap_or(&issue.code) == dedupe_key
+                    })
+                    .map(|issue| issue.id),
+            );
+        }
+    }
+
+    if ids_to_clear.is_empty() {
+        return Ok(false);
+    }
+
+    let requests = ids_to_clear
+        .into_iter()
+        .map(BusRequest::ClearUrgentIssue)
+        .collect::<Vec<_>>();
+    let Ok(reply) = run_query(requests) else {
+        return Ok(false);
+    };
+    Ok(reply.iter().any(|resp| matches!(resp, BusResponse::Ack)))
+}
+
 /// Log an informational message via the lqosd bus (appears in lqosd logs).
 #[pyfunction]
 fn log_info(_py: Python, message: String) -> PyResult<bool> {

src/rust/lqos_sys/src/lib.rs

@@ -34,6 +34,7 @@ pub use kernel_wrapper::LibreQoSKernels;
 pub use linux::num_possible_cpus;
 pub use lqos_kernel::interface_name_to_index;
 pub use lqos_kernel::ip_mapping_capacity;
+pub use lqos_kernel::ip_mapping_subsystem_ready;
 pub use lqos_kernel::max_tracked_ips;
 pub use lqos_kernel::unload_xdp_from_interface;
 pub use tc_classify_control::{initialize_tc_classify_bypass, set_tc_classify_bypass};

src/rust/lqos_sys/src/lqos_kernel.rs

@@ -31,6 +31,13 @@ pub(crate) mod bpf {
     include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
 }
 
+#[derive(Clone, Copy)]
+struct PinnedMapAbiSpec {
+    path: &'static str,
+    expected_key_size: u32,
+    expected_value_size: u32,
+}
+
 /// Returns the value set in the C XDP system's MAX_TRACKED_IPS
 /// constant.
 pub fn max_tracked_ips() -> usize {
@@ -42,27 +49,7 @@ fn pinned_map_max_entries(path: &str) -> Result<Option<u32>> {
         return Ok(None);
     }
 
-    let path_c = CString::new(path)?;
-    let fd = unsafe { bpf_obj_get(path_c.as_ptr()) };
-    if fd < 0 {
-        warn!("Unable to open pinned BPF map '{path}' for capacity check (fd={fd})");
-        return Ok(None);
-    }
-
-    let mut info = MaybeUninit::<bpf_map_info>::zeroed();
-    let mut len: u32 = std::mem::size_of::<bpf_map_info>() as u32;
-    let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) };
-    unsafe {
-        close(fd);
-    }
-    if err != 0 {
-        return Err(Error::msg(format!(
-            "Unable to query pinned BPF map '{path}' capacity (err={err})."
-        )));
-    }
-
-    let info = unsafe { info.assume_init() };
-    Ok(Some(info.max_entries))
+    Ok(pinned_map_info(path)?.map(|info| info.max_entries))
 }
 
 /// Returns the currently available IP-mapping capacity.
@@ -80,6 +67,155 @@ pub fn ip_mapping_capacity() -> usize {
     }
 }
 
+fn ip_mapping_pinned_map_abi_specs() -> [PinnedMapAbiSpec; 3] {
+    let expected_key_size = std::mem::size_of::<crate::ip_mapping::IpHashKey>() as u32;
+    let expected_value_size = std::mem::size_of::<crate::ip_mapping::IpHashData>() as u32;
+    [
+        PinnedMapAbiSpec {
+            path: "/sys/fs/bpf/map_ip_to_cpu_and_tc",
+            expected_key_size,
+            expected_value_size,
+        },
+        PinnedMapAbiSpec {
+            path: "/sys/fs/bpf/ip_to_cpu_and_tc_hotcache",
+            expected_key_size: std::mem::size_of::<XdpIpAddress>() as u32,
+            expected_value_size,
+        },
+        PinnedMapAbiSpec {
+            path: "/sys/fs/bpf/ip_mapping_epoch",
+            expected_key_size: std::mem::size_of::<u32>() as u32,
+            expected_value_size: std::mem::size_of::<u32>() as u32,
+        },
+    ]
+}
+
+fn pinned_map_info(path: &str) -> Result<Option<bpf_map_info>> {
+    let path_c = CString::new(path)?;
+    let fd = unsafe { bpf_obj_get(path_c.as_ptr()) };
+    if fd < 0 {
+        return Ok(None);
+    }
+
+    let mut info = MaybeUninit::<bpf_map_info>::zeroed();
+    let mut len: u32 = std::mem::size_of::<bpf_map_info>() as u32;
+    let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) };
+    unsafe {
+        close(fd);
+    }
+    if err != 0 {
+        return Err(Error::msg(format!(
+            "Unable to query pinned BPF map '{path}' info (err={err})."
+        )));
+    }
+    let info = unsafe { info.assume_init() };
+    Ok(Some(info))
+}
+
+fn pinned_map_key_value_sizes(path: &str) -> Result<Option<(u32, u32)>> {
+    Ok(pinned_map_info(path)?.map(|info| (info.key_size, info.value_size)))
+}
+
+fn ip_mapping_subsystem_ready_with<F>(mut map_sizes: F) -> Result<bool>
+where
+    F: FnMut(&str) -> Result<Option<(u32, u32)>>,
+{
+    for spec in ip_mapping_pinned_map_abi_specs() {
+        let Some((key_size, value_size)) = map_sizes(spec.path)? else {
+            return Ok(false);
+        };
+        if key_size != spec.expected_key_size || value_size != spec.expected_value_size {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+/// Returns whether the pinned maps required for XDP IP mapping updates are ready.
+///
+/// This function has side effects: it opens the relevant pinned BPF maps and
+/// queries their ABI metadata. It returns `Ok(false)` for maps that are not yet
+/// pinned or have an incompatible ABI, and `Err` for unexpected inspection
+/// failures.
+pub fn ip_mapping_subsystem_ready() -> Result<bool> {
+    ip_mapping_subsystem_ready_with(pinned_map_key_value_sizes)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    #[test]
+    fn ip_mapping_ready_requires_all_expected_maps() {
+        let specs = ip_mapping_pinned_map_abi_specs();
+        let sizes: HashMap<&str, (u32, u32)> = specs
+            .iter()
+            .map(|spec| {
+                (
+                    spec.path,
+                    (spec.expected_key_size, spec.expected_value_size),
+                )
+            })
+            .collect();
+
+        let ready = ip_mapping_subsystem_ready_with(|path| Ok(sizes.get(path).copied()))
+            .expect("readiness check should succeed");
+
+        assert!(ready);
+    }
+
+    #[test]
+    fn ip_mapping_ready_is_false_when_a_map_is_missing() {
+        let ready = ip_mapping_subsystem_ready_with(|path| {
+            if path == "/sys/fs/bpf/ip_mapping_epoch" {
+                Ok(None)
+            } else {
+                let spec = ip_mapping_pinned_map_abi_specs()
+                    .into_iter()
+                    .find(|spec| spec.path == path)
+                    .expect("test path should be part of IP mapping ABI specs");
+                Ok(Some((spec.expected_key_size, spec.expected_value_size)))
+            }
+        })
+        .expect("missing map should not be an inspection error");
+
+        assert!(!ready);
+    }
+
+    #[test]
+    fn ip_mapping_ready_is_false_when_a_map_abi_differs() {
+        let ready = ip_mapping_subsystem_ready_with(|path| {
+            let spec = ip_mapping_pinned_map_abi_specs()
+                .into_iter()
+                .find(|spec| spec.path == path)
+                .expect("test path should be part of IP mapping ABI specs");
+            let value_size = if path == "/sys/fs/bpf/map_ip_to_cpu_and_tc" {
+                spec.expected_value_size + 1
+            } else {
+                spec.expected_value_size
+            };
+            Ok(Some((spec.expected_key_size, value_size)))
+        })
+        .expect("ABI mismatch should not be an inspection error");
+
+        assert!(!ready);
+    }
+
+    #[test]
+    fn ip_mapping_ready_returns_inspection_error() {
+        let err = ip_mapping_subsystem_ready_with(|path| {
+            if path == "/sys/fs/bpf/map_ip_to_cpu_and_tc" {
+                Err(Error::msg("map info failed"))
+            } else {
+                Ok(None)
+            }
+        })
+        .expect_err("inspection errors should be returned to the caller");
+
+        assert!(err.to_string().contains("map info failed"));
+    }
+}
+
 pub fn check_root() -> Result<()> {
     unsafe {
         if geteuid() == 0 {
@@ -99,25 +235,10 @@ fn remove_incompatible_pinned_map(
         return Ok(());
     }
 
-    let path_c = CString::new(path)?;
-    let fd = unsafe { bpf_obj_get(path_c.as_ptr()) };
-    if fd < 0 {
-        warn!("Unable to open pinned BPF map '{path}' for ABI check (fd={fd})");
+    let Some(info) = pinned_map_info(path)? else {
+        warn!("Unable to open pinned BPF map '{path}' for ABI check");
         return Ok(());
-    }
-
-    let mut info = MaybeUninit::<bpf_map_info>::zeroed();
-    let mut len: u32 = std::mem::size_of::<bpf_map_info>() as u32;
-    let err = unsafe { bpf_obj_get_info_by_fd(fd, info.as_mut_ptr() as *mut c_void, &mut len) };
-    unsafe {
-        close(fd);
-    }
-    if err != 0 {
-        return Err(Error::msg(format!(
-            "Unable to query pinned BPF map '{path}' info (err={err})."
-        )));
-    }
-    let info = unsafe { info.assume_init() };
+    };
 
     if info.key_size != expected_key_size || info.value_size != expected_value_size {
         warn!(
@@ -135,23 +256,13 @@ fn remove_incompatible_pinned_map(
 }
 
 fn ensure_ip_mapping_maps_abi() -> Result<()> {
-    let expected_key_size = std::mem::size_of::<crate::ip_mapping::IpHashKey>() as u32;
-    let expected_value_size = std::mem::size_of::<crate::ip_mapping::IpHashData>() as u32;
-    remove_incompatible_pinned_map(
-        "/sys/fs/bpf/map_ip_to_cpu_and_tc",
-        expected_key_size,
-        expected_value_size,
-    )?;
-    remove_incompatible_pinned_map(
-        "/sys/fs/bpf/ip_to_cpu_and_tc_hotcache",
-        std::mem::size_of::<XdpIpAddress>() as u32,
-        expected_value_size,
-    )?;
-    remove_incompatible_pinned_map(
-        "/sys/fs/bpf/ip_mapping_epoch",
-        std::mem::size_of::<u32>() as u32,
-        std::mem::size_of::<u32>() as u32,
-    )?;
+    for spec in ip_mapping_pinned_map_abi_specs() {
+        remove_incompatible_pinned_map(
+            spec.path,
+            spec.expected_key_size,
+            spec.expected_value_size,
+        )?;
+    }
     remove_incompatible_pinned_map(
         "/sys/fs/bpf/tc_classify_control",
         std::mem::size_of::<u32>() as u32,

src/rust/lqosd/src/main.rs

@@ -1047,6 +1047,10 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>
                 urgent::clear(*id);
                 BusResponse::Ack
             }
+            BusRequest::ClearUrgentIssueByIdentity { code, dedupe_key } => {
+                urgent::clear_by_identity(code, dedupe_key);
+                BusResponse::Ack
+            }
             BusRequest::ClearAllUrgentIssues => {
                 urgent::clear_all();
                 BusResponse::Ack
@@ -1390,3 +1394,33 @@ fn search_result_to_bus(entry: node_manager::SearchResult) -> lqos_bus::SearchRe
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use lqos_bus::{UrgentSeverity, UrgentSource};
+
+    #[test]
+    fn bus_clear_urgent_issue_by_identity_reaches_urgent_store() {
+        let code = "TEST_BUS_XDP_IP_MAPPING_APPLY_FAILED";
+        urgent::clear_by_identity(code, code);
+        urgent::submit(
+            UrgentSource::LibreQoS,
+            UrgentSeverity::Error,
+            code.to_string(),
+            "mapping failed".to_string(),
+            None,
+            Some(code.to_string()),
+        );
+
+        let requests = [BusRequest::ClearUrgentIssueByIdentity {
+            code: code.to_string(),
+            dedupe_key: code.to_string(),
+        }];
+        let mut responses = Vec::new();
+        handle_bus_requests(&requests, &mut responses);
+
+        assert!(matches!(responses.as_slice(), [BusResponse::Ack]));
+        assert!(!urgent::list().iter().any(|issue| issue.code == code));
+    }
+}

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

@@ -4,14 +4,13 @@ use anyhow::Result;
 use arc_swap::ArcSwap;
 use fxhash::{FxHashMap, FxHashSet};
 use lqos_bus::{BusResponse, Circuit};
+#[cfg(test)]
+use lqos_config::load_active_runtime_shaping_inputs;
 use lqos_config::{
     NetworkJsonNode, NetworkJsonTransport, TopologyRuntimeShapingPayloadIdentity,
     TopologyRuntimeStatusFile, TopologyShapingInputsFile,
-    load_active_runtime_shaping_inputs_from_status, load_config,
-    topology_runtime_status_path,
+    load_active_runtime_shaping_inputs_from_status, load_config, topology_runtime_status_path,
 };
-#[cfg(test)]
-use lqos_config::load_active_runtime_shaping_inputs;
 use lqos_queue_tracker::EFFECTIVE_NODE_RATES;
 use lqos_utils::file_watcher::FileWatcher;
 use lqos_utils::hash_to_i64;
@@ -106,7 +105,9 @@ fn publish_shaping_inputs(shaping_inputs: TopologyShapingInputsFile) {
     invalidate_executive_cache_snapshot();
 }
 
-fn topology_status_identity(status: &TopologyRuntimeStatusFile) -> TopologyRuntimeShapingPayloadIdentity {
+fn topology_status_identity(
+    status: &TopologyRuntimeStatusFile,
+) -> TopologyRuntimeShapingPayloadIdentity {
     status.shaping_payload_identity()
 }
 
@@ -286,7 +287,9 @@ fn load_topology_runtime_status_payload() {
     let status = match TopologyRuntimeStatusFile::load(config.as_ref()) {
         Ok(status) => status,
         Err(err) => {
-            warn!("Unable to load topology_runtime_status.json: {err}; keeping last-known-good effective parent cache");
+            warn!(
+                "Unable to load topology_runtime_status.json: {err}; keeping last-known-good effective parent cache"
+            );
             return;
         }
     };

src/rust/lqosd/src/urgent.rs

@@ -17,6 +17,10 @@ fn now_unix() -> u64 {
     unix_now().unwrap_or_default()
 }
 
+fn urgent_identity_key(code: &str, dedupe_key: Option<&str>) -> String {
+    dedupe_key.unwrap_or(code).to_string()
+}
+
 fn prune_expired(q: &mut VecDeque<UrgentIssue>) {
     let now = now_unix();
     while let Some(front) = q.front() {
@@ -44,8 +48,8 @@ pub fn submit(
     let mut guard = URGENT.lock();
     prune_expired(&mut guard);
 
-    // Dedupe: same (code, dedupe_key) within window → update timestamp/message
-    let key = dedupe_key.clone().unwrap_or_else(|| code.clone());
+    // Dedupe: same (code, dedupe_key) within window updates timestamp/message.
+    let key = urgent_identity_key(&code, dedupe_key.as_deref());
     if let Some(existing) = guard.iter_mut().rev().find(|i| {
         (i.code == code)
             && (i.dedupe_key.as_deref().unwrap_or("") == key)
@@ -90,7 +94,79 @@ pub fn clear(id: u64) -> bool {
     }
 }
 
+/// Clears urgent issues matching the same code and dedupe key used for submission.
+pub fn clear_by_identity(code: &str, dedupe_key: &str) -> usize {
+    let key = urgent_identity_key(code, Some(dedupe_key));
+    let mut guard = URGENT.lock();
+    let before = guard.len();
+    guard.retain(|issue| {
+        issue.code != code || issue.dedupe_key.as_deref().unwrap_or("") != key.as_str()
+    });
+    before.saturating_sub(guard.len())
+}
+
 pub fn clear_all() {
     let mut guard = URGENT.lock();
     guard.clear();
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn clear_by_identity_removes_only_matching_issues() {
+        let code = "TEST_XDP_IP_MAPPING_APPLY_FAILED";
+        let other_code = "TEST_OTHER";
+        let other_dedupe = "TEST_XDP_IP_MAPPING_APPLY_FAILED_OTHER_DEDUPE";
+        clear_by_identity(code, code);
+        clear_by_identity(code, other_dedupe);
+        clear_by_identity(other_code, code);
+        submit(
+            UrgentSource::LibreQoS,
+            UrgentSeverity::Error,
+            code.to_string(),
+            "mapping failed".to_string(),
+            None,
+            Some(code.to_string()),
+        );
+        submit(
+            UrgentSource::LibreQoS,
+            UrgentSeverity::Error,
+            code.to_string(),
+            "same code different dedupe".to_string(),
+            None,
+            Some(other_dedupe.to_string()),
+        );
+        submit(
+            UrgentSource::LibreQoS,
+            UrgentSeverity::Error,
+            other_code.to_string(),
+            "other failed".to_string(),
+            None,
+            Some(code.to_string()),
+        );
+
+        let cleared = clear_by_identity(code, code);
+        let matching_issue_survived = list()
+            .into_iter()
+            .any(|issue| issue.code == code && issue.dedupe_key.as_deref() == Some(other_dedupe));
+
+        assert_eq!(cleared, 1);
+        assert!(matching_issue_survived);
+        assert_eq!(clear_by_identity(code, other_dedupe), 1);
+        assert_eq!(clear_by_identity(other_code, code), 1);
+    }
+
+    #[test]
+    fn urgent_identity_key_falls_back_to_code() {
+        assert_eq!(
+            urgent_identity_key("CODE_ONLY", None),
+            "CODE_ONLY".to_string()
+        );
+        assert_eq!(
+            urgent_identity_key("CODE", Some("DEDUP")),
+            "DEDUP".to_string()
+        );
+    }
+}

src/test_xdp_mapping_recovery.py

@@ -0,0 +1,124 @@
+import unittest
+
+import LibreQoS
+
+
+class FakeIpMapBatch:
+    def __init__(self, submit_error=None):
+        self.finished = False
+        self.submitted = False
+        self.submit_error = submit_error
+
+    def finish_ip_mappings(self):
+        self.finished = True
+
+    def submit(self):
+        self.submitted = True
+        if self.submit_error is not None:
+            raise self.submit_error
+
+
+class XdpMappingRecoveryTests(unittest.TestCase):
+    def test_ready_on_first_check_does_not_sleep(self):
+        sleeps = []
+
+        LibreQoS.wait_for_xdp_ip_mapping_ready(
+            ready=lambda: True,
+            sleep=lambda seconds: sleeps.append(seconds),
+            now=lambda: 0.0,
+        )
+
+        self.assertEqual(sleeps, [])
+
+    def test_readiness_after_one_false_check_sleeps_once(self):
+        checks = iter([False, True])
+        sleeps = []
+        times = iter([0.0, 0.0, 0.0])
+
+        LibreQoS.wait_for_xdp_ip_mapping_ready(
+            ready=lambda: next(checks),
+            sleep=lambda seconds: sleeps.append(seconds),
+            now=lambda: next(times),
+            timeout_seconds=5.0,
+            interval_seconds=0.1,
+        )
+
+        self.assertEqual(sleeps, [0.1])
+
+    def test_persistent_false_readiness_times_out(self):
+        times = iter([0.0, 0.0, 0.2])
+
+        with self.assertRaises(TimeoutError):
+            LibreQoS.wait_for_xdp_ip_mapping_ready(
+                ready=lambda: False,
+                sleep=lambda _seconds: None,
+                now=lambda: next(times),
+                timeout_seconds=0.1,
+                interval_seconds=0.1,
+            )
+
+    def test_readiness_os_error_reports_before_submit(self):
+        batch = FakeIpMapBatch()
+        reports = []
+
+        def raise_os_error():
+            raise OSError("Unable to inspect XDP IP mapping maps")
+
+        LibreQoS.apply_xdp_ip_mappings(
+            batch,
+            {"queued_requests": 1},
+            ready=raise_os_error,
+            sleep=lambda _seconds: None,
+            report_failure=lambda *args: reports.append(args),
+            clear_recovered_issue=lambda *_args: True,
+        )
+
+        self.assertFalse(batch.finished)
+        self.assertFalse(batch.submitted)
+        self.assertEqual(reports[0][0], LibreQoS.XDP_IP_MAPPING_APPLY_FAILED)
+
+    def test_submit_failure_reports_mapping_apply_failure(self):
+        batch = FakeIpMapBatch(submit_error=OSError("Unable to open BPF map"))
+        reports = []
+
+        LibreQoS.apply_xdp_ip_mappings(
+            batch,
+            {"queued_requests": 1},
+            ready=lambda: True,
+            sleep=lambda _seconds: None,
+            report_failure=lambda *args: reports.append(args),
+            clear_recovered_issue=lambda *_args: True,
+        )
+
+        self.assertTrue(batch.finished)
+        self.assertTrue(batch.submitted)
+        self.assertEqual(reports[0][0], LibreQoS.XDP_IP_MAPPING_APPLY_FAILED)
+
+    def test_success_clears_recovered_mapping_issue(self):
+        batch = FakeIpMapBatch()
+        cleared = []
+
+        LibreQoS.apply_xdp_ip_mappings(
+            batch,
+            {"queued_requests": 1},
+            ready=lambda: True,
+            sleep=lambda _seconds: None,
+            report_failure=lambda *_args: self.fail("unexpected failure report"),
+            clear_recovered_issue=lambda code, dedupe_key: cleared.append((code, dedupe_key)) or True,
+        )
+
+        self.assertTrue(batch.finished)
+        self.assertTrue(batch.submitted)
+        self.assertEqual(
+            cleared,
+            [
+                (
+                    LibreQoS.XDP_IP_MAPPING_APPLY_FAILED,
+                    LibreQoS.XDP_IP_MAPPING_APPLY_FAILED,
+                )
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()