fix: Fix TX frame pool exhaustion causing silent packet drops

mantis/src/core/ebpf/xsk_manager.rs

@@ -169,7 +169,7 @@ impl XskPair {
         // Zero-copy mode: NIC DMA directly into UMEM; fill queue starvation = drop;
         //   requires driver zero-copy support (igb: kernel 6.14+).
         let bind_flags = match config.xsk_bind_mode {
-            XskBindMode::Copy => BindFlags::XDP_USE_NEED_WAKEUP,
+            XskBindMode::Copy => BindFlags::XDP_COPY | BindFlags::XDP_USE_NEED_WAKEUP,
             XskBindMode::Zero => BindFlags::XDP_ZEROCOPY | BindFlags::XDP_USE_NEED_WAKEUP,
         };
         let mode_str = match config.xsk_bind_mode {
@@ -420,7 +420,9 @@ impl XskPair {
         // Drain everything above that threshold into the fill queue in one shot.
         // produce() is all-or-nothing per call, so iterate in BATCH-sized chunks
         // until the fill queue is full; return any remaining frames to the pool.
-        const RESERVED_FOR_TX: usize = 64;
+        // Reserve 2× the TX batch size so back-to-back 64-packet bursts never
+        // drain the pool to zero before completions are returned by the NIC.
+        const RESERVED_FOR_TX: usize = 128;
         const BATCH: usize = 256;
 
         let all_frames: Vec<FrameDesc> = {
@@ -454,27 +456,33 @@ impl XskPair {
     }
 
     fn process_tx_queue(&mut self, forward_rx: &Receiver<Vec<u8>>) -> Result<usize, EbpfError> {
-        let mut packets_to_send = Vec::with_capacity(64);
-        while let Ok(packet) = forward_rx.try_recv() {
-            packets_to_send.push(packet);
-            if packets_to_send.len() >= 64 {
-                break;
-            }
-        }
-
-        if packets_to_send.is_empty() {
-            return Ok(0);
-        }
-
+        // Drain completed TX frames first to maximise pool availability.
         let _ = self.process_comp_queue();
 
+        // Check pool BEFORE consuming from the channel.  If we consumed first and
+        // then found pool == 0 we would silently drop the packets (they cannot be
+        // put back into the channel).  By checking first, packets stay in the
+        // channel and are retried on the next loop iteration.
         let pool_size = {
             let pool = self.frame_pool.lock();
             pool.len()
         };
 
         if pool_size == 0 {
-            log!(EbpfLog::FramePoolExhausted(packets_to_send.len()));
+            return Ok(0);
+        }
+
+        // Consume at most min(pool_size, 64) packets so we never over-commit.
+        let max_to_send = pool_size.min(64);
+        let mut packets_to_send = Vec::with_capacity(max_to_send);
+        while let Ok(packet) = forward_rx.try_recv() {
+            packets_to_send.push(packet);
+            if packets_to_send.len() >= max_to_send {
+                break;
+            }
+        }
+
+        if packets_to_send.is_empty() {
             return Ok(0);
         }
 

mantis/src/detection/ml/flow_tracker.rs

@@ -347,11 +347,11 @@ impl FlowTracker {
     }
 
     // Export flows that are finished or idle.
-    // Active flows alive >= force_interval_us are cloned and exported,
+    // Active flows alive >= window_interval_us are cloned and exported,
     // then reset in place so the next window starts accumulating immediately.
-    // This ensures active attack flows reach ML/logger within force_interval_us
+    // This ensures active attack flows reach ML/logger within window_interval_us
     // and keeps training and inference feature distributions identical.
-    pub fn drain_ready_flows(&mut self, timeout_us: u64, force_interval_us: u64) -> Vec<FlowData> {
+    pub fn drain_ready_flows(&mut self, timeout_us: u64, window_interval_us: u64) -> Vec<FlowData> {
         let now = time::SystemTime::now()
             .duration_since(time::UNIX_EPOCH)
             .map(|d| d.as_micros() as u64)
@@ -371,7 +371,7 @@ impl FlowTracker {
             .collect();
 
         for flow in self.flows.values_mut() {
-            if now.saturating_sub(flow.start_time_us) >= force_interval_us {
+            if now.saturating_sub(flow.start_time_us) >= window_interval_us {
                 if flow.packet_count() > 0 {
                     result.push(flow.clone());
                     flow.reset(now);
@@ -442,9 +442,11 @@ fn detect_initiator(payload: &[u8], protocol: u8, src_port: u16, dst_port: u16)
         }
     }
 
-    // DNS over UDP (port 53): flags byte 2, MSB = QR bit
+    // DNS/NBNS over UDP (port 53/137): FLAGS byte 2, MSB = QR bit (RFC 1035 / RFC 1002).
     //   0 = query (initiator), 1 = response (responder)
-    if protocol == 17 && (src_port == 53 || dst_port == 53) && payload.len() >= 3 {
+    // NBNS has an identical header layout to DNS, so the same byte offset applies.
+    if protocol == 17 && (src_port == 53 || dst_port == 53 || src_port == 137 || dst_port == 137) && payload.len() >= 3
+    {
         return Some((payload[2] >> 7) == 0);
     }
 

mantis/src/model/log/ebpf.rs

@@ -21,9 +21,6 @@ loggable! {
         #[error("XSK thread shutting down")]
         XSKShutdown => tracing::Level::INFO,
 
-        #[error("Frame pool exhausted! Pending TX: {send_len} packets")]
-        FramePoolExhausted { send_len: usize } => tracing::Level::WARN,
-
         #[error("No frames available for TX")]
         NoFramesAvailable => tracing::Level::WARN,