feat: per-batch scanner lock for concurrent discovery (#2)

* feat: hold scanner lock per-batch instead of per-scan in discover_bpf

Previously the scanner mutex was held for the entire batch-send phase,
serializing all concurrent discovery requests. Now each batch acquires
and releases the lock independently, with a per-request stealth profile
clone instead of mutating shared state. Concurrent scans interleave
their batches rather than fully serializing.

* fix: remove passthrough mode — BPF-only, no degraded operation

Passthrough mode silently ran with raw socket TX when BPF had issues,
producing discovery results without timing data. Remove the config
option entirely so limpet either starts with full BPF or fails hard.

---------

Co-authored-by: Simon Morley <simon.morley+debbie@mac.com>

Author: simonmorley

src/cli/mod.rs

@@ -178,18 +178,12 @@ use crate::ScanResult;
 /// On non-Linux, returns the engine directly (connect-only for dev).
 #[cfg(target_os = "linux")]
 fn create_engine(interface: Option<String>) -> Result<Engine, String> {
-    Engine::new(ScanEngineConfig {
-        interface,
-        passthrough: false,
-    })
+    Engine::new(ScanEngineConfig { interface })
 }
 
 #[cfg(not(target_os = "linux"))]
 fn create_engine(interface: Option<String>) -> Result<Engine, String> {
-    Ok(Engine::new(ScanEngineConfig {
-        interface,
-        passthrough: false,
-    }))
+    Ok(Engine::new(ScanEngineConfig { interface }))
 }
 
 // ─────────────────────────────────────────────────────────────────────────────

src/engine.rs

@@ -22,9 +22,6 @@ use crate::{PortState, ScanResult, ScannedPort, TimingBackend, TimingRequest, Ti
 pub struct ScanEngineConfig {
     /// Network interface for XDP (None = auto-detect from /proc/net/route).
     pub interface: Option<String>,
-    /// When true, use raw socket TX only (no AF_XDP redirect).
-    /// BPF timestamps packets but XDP_PASS lets them through to the kernel.
-    pub passthrough: bool,
 }
 
 /// Scanning engine. Shared via `Arc<Engine>` between discovery and timing loops.
@@ -44,8 +41,6 @@ pub struct ScanEngine {
     scanner: Arc<Mutex<SynScanner>>,
     interface: String,
     backend: TimingBackend,
-    #[allow(dead_code)]
-    passthrough: bool,
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -108,10 +103,6 @@ impl Engine {
     /// - SYN scanner creation fails (no IPv4 on interface, raw socket error)
     #[cfg(target_os = "linux")]
     pub fn new(config: ScanEngineConfig) -> Result<Self, String> {
-        if config.passthrough {
-            tracing::info!("passthrough mode: using raw socket TX, no AF_XDP redirect");
-        }
-
         let (backend, collector) = crate::timing::detect_timing_backend(&config.interface)
             .map_err(|e| {
                 format!(
@@ -127,12 +118,11 @@ impl Engine {
         tracing::info!(
             backend = %backend,
             interface = %iface,
-            passthrough = config.passthrough,
             "BPF timing backend initialised"
         );
 
         let bpf = Arc::new(Mutex::new(collector));
-        let scanner = Self::create_scanner(&iface, &bpf, config.passthrough).ok_or_else(|| {
+        let scanner = Self::create_scanner(&iface, &bpf).ok_or_else(|| {
             format!(
                 "SYN scanner creation failed on interface '{iface}'. \
                  Check that the interface has an IPv4 address and raw socket permissions."
@@ -144,7 +134,6 @@ impl Engine {
             scanner,
             interface: iface,
             backend,
-            passthrough: config.passthrough,
         }))
     }
 
@@ -158,18 +147,13 @@ impl Engine {
         }
     }
 
-    /// Create a `SynScanner` with the appropriate sender.
-    ///
-    /// In passthrough mode, uses `RawSocketSender` only (no AF_XDP redirect).
-    /// Otherwise creates a `HybridSender` and registers its AF_XDP fd in xsk_map.
+    /// Create a `SynScanner` with raw socket TX + BPF timestamps.
     #[cfg(target_os = "linux")]
     fn create_scanner(
         iface: &str,
-        bpf: &Arc<Mutex<BpfTimingCollector>>,
-        passthrough: bool,
+        _bpf: &Arc<Mutex<BpfTimingCollector>>,
     ) -> Option<Arc<Mutex<SynScanner>>> {
         use crate::scanner::afxdp_sender::AfXdpSend;
-        use crate::scanner::hybrid_sender::HybridSender;
         use crate::scanner::raw_socket_sender::RawSocketSender;
         use crate::scanner::syn_sender::interface_source_ip;
 
@@ -181,39 +165,11 @@ impl Engine {
             }
         };
 
-        let xdp_sender: Box<dyn AfXdpSend> = if passthrough {
-            match RawSocketSender::new(src_ip, Some(iface)) {
-                Ok(s) => Box::new(s),
-                Err(e) => {
-                    tracing::error!(error = %e, "raw socket sender failed");
-                    return None;
-                }
-            }
-        } else {
-            match HybridSender::new(iface, 0, src_ip) {
-                Ok(sender) => {
-                    // Register AF_XDP socket in BPF xsk_map.
-                    // try_lock() is safe here — called at init, no contention.
-                    if let Ok(bpf_guard) = bpf.try_lock() {
-                        if let Err(e) = bpf_guard.register_xsk_fd(sender.fd()) {
-                            tracing::warn!(error = %e, "xsk_map registration failed");
-                        }
-                    }
-                    Box::new(sender)
-                }
-                Err(e) => {
-                    tracing::warn!(
-                        error = %e,
-                        "hybrid sender unavailable — falling back to raw socket TX"
-                    );
-                    match RawSocketSender::new(src_ip, Some(iface)) {
-                        Ok(s) => Box::new(s),
-                        Err(e) => {
-                            tracing::error!(error = %e, "raw socket fallback also failed");
-                            return None;
-                        }
-                    }
-                }
+        let xdp_sender: Box<dyn AfXdpSend> = match RawSocketSender::new(src_ip, Some(iface)) {
+            Ok(s) => Box::new(s),
+            Err(e) => {
+                tracing::error!(error = %e, "raw socket sender failed");
+                return None;
             }
         };
 
@@ -331,21 +287,19 @@ impl ScanEngine {
         let timeout = Duration::from_millis(timeout_ms as u64);
         let target_ip_u32 = u32::from_be_bytes(target_ip.octets());
 
-        // Apply per-request pacing to the shared scanner. We hold the lock
-        // for the entire batch-send phase, so no concurrent access sees the
-        // changed profile. Restored to Aggressive after sending.
-        let mut scanner_guard = self.scanner.lock().await;
-        let original_profile = scanner_guard.profile().clone();
+        // Build per-request stealth profile (no shared state mutation).
+        // Scanner lock is held per-batch, not per-scan, so concurrent
+        // discovery requests can interleave their batches.
         let mut scan_profile = StealthProfile::linux_6x_default();
         request.pacing.apply_to(&mut scan_profile);
-        scanner_guard.set_profile(scan_profile);
 
         let mut all_probes = Vec::new();
         for batch in ports.chunks(batch_size) {
+            let mut scanner_guard = self.scanner.lock().await;
+            scanner_guard.set_profile(scan_profile.clone());
             let result = match scanner_guard.send_syn_batch(target_ip, batch) {
                 Ok(r) => r,
                 Err(e) => {
-                    scanner_guard.set_profile(original_profile);
                     drop(scanner_guard);
                     return ScanResult {
                         request_id: request.request_id,
@@ -363,12 +317,9 @@ impl ScanEngine {
 
             // Drain AF_XDP RX ring between batches to prevent overflow
             scanner_guard.poll_rx(0);
+            drop(scanner_guard);
         }
 
-        // Restore Aggressive profile for timing probes
-        scanner_guard.set_profile(original_profile);
-        drop(scanner_guard);
-
         // Early-return polling: check every 5ms if all probes have responses.
         // Falls back to full timeout as the deadline.
         {
@@ -736,7 +687,6 @@ mod tests {
         // Either way, it must not panic.
         let result = Engine::new(ScanEngineConfig {
             interface: None,
-            passthrough: true,
         });
         match result {
             Ok(engine) => {
@@ -757,7 +707,6 @@ mod tests {
     fn test_engine_new_returns_connect_only_on_non_linux() {
         let engine = Engine::new(ScanEngineConfig {
             interface: None,
-            passthrough: true,
         });
         assert_eq!(engine.backend_str(), "connect");
     }