Avoid formatting discarded scan errors (#950)

* perf(scanner): defer TCP error diagnostics

* bench(scanner): check open-only TCP and UDP paths

* fix(scanner): Import HashSet in tests after main merge

---------

Co-authored-by: Autumn Skerritt <skerraut@amazon.com>

Author: bee-san

.github/scan-benchmark/scan_bench.py

@@ -196,6 +196,7 @@ def ports(ranges: tuple[tuple[int, int], ...]) -> str:
 
     scenarios = [
         Scenario("tcp-1-port", "TCP, 1 open port", lo, ports=(TCP_PORTS[0],), repeat=3),
+        Scenario("tcp-open", "TCP, open listeners only", lo, ports=TCP_PORTS, repeat=3),
         Scenario("tcp-sweep", f"TCP, {ports(sweep)}, default batch", lo, sweep),
         Scenario(
             "tcp-sweep-excluded",
@@ -236,6 +237,12 @@ def ports(ranges: tuple[tuple[int, int], ...]) -> str:
     scenarios.append(
         Scenario("udp-sweep", f"UDP, {ports(sweep)}", lo, sweep, timeout_ms=500, udp=True)
     )
+    scenarios.append(
+        Scenario(
+            "udp-open", "UDP, echo responders only", lo, ports=UDP_PORTS,
+            timeout_ms=500, udp=True, repeat=9,
+        )
+    )
     if delayed:
         scenarios.append(
             Scenario(

Cargo.lock

@@ -43,6 +43,12 @@ anyhow = "1.0.40"
 text_placeholder = { version = "0.5", features = ["struct_context"] }
 once_cell = "1.21.4"
 
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.59", features = ["Win32_Networking_WinSock"] }
+
 [dev-dependencies]
 parameterized = "2.0.0"
 criterion = { version = "0.8", features = ["html_reports"] }

Cargo.toml

@@ -0,0 +1,163 @@
+use std::{collections::HashSet, fmt, io, net::IpAddr};
+
+/// Keep the OS error and its target intact until diagnostics need formatting.
+#[derive(Debug)]
+struct ScanError {
+    ip: IpAddr,
+    error: io::Error,
+}
+
+impl fmt::Display for ScanError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{} {}", self.error, self.ip)
+    }
+}
+
+impl std::error::Error for ScanError {}
+
+/// Preserve the normal I/O result layout; only diagnostics need target context.
+pub(super) fn diagnostic_error(error: io::Error, ip: IpAddr, enabled: bool) -> io::Error {
+    if enabled {
+        io::Error::other(ScanError { ip, error })
+    } else {
+        error
+    }
+}
+
+pub(super) struct ScanErrors {
+    enabled: bool,
+    limit: usize,
+    messages: HashSet<String>,
+}
+
+impl ScanErrors {
+    pub(super) fn new(enabled: bool, limit: usize) -> Self {
+        Self {
+            enabled,
+            limit,
+            messages: HashSet::new(),
+        }
+    }
+
+    pub(super) fn record(&mut self, failure: io::Error) {
+        if self.enabled && self.messages.len() < self.limit {
+            self.messages.insert(failure.to_string());
+        }
+    }
+
+    pub(super) fn messages(&self) -> &HashSet<String> {
+        &self.messages
+    }
+}
+
+/// Detect descriptor exhaustion without allocating or interpreting OS messages.
+#[cfg(unix)]
+pub(super) fn is_descriptor_exhaustion(error: &io::Error) -> bool {
+    matches!(error.raw_os_error(), Some(libc::EMFILE | libc::ENFILE))
+}
+
+#[cfg(windows)]
+pub(super) fn is_descriptor_exhaustion(error: &io::Error) -> bool {
+    error.raw_os_error() == Some(windows_sys::Win32::Networking::WinSock::WSAEMFILE)
+}
+
+// Retain the previous fallback for targets without Unix or Winsock error codes.
+#[cfg(not(any(unix, windows)))]
+pub(super) fn is_descriptor_exhaustion(error: &io::Error) -> bool {
+    error
+        .to_string()
+        .to_lowercase()
+        .contains("too many open files")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{fmt, net::SocketAddr};
+
+    #[derive(Debug)]
+    struct MustNotFormat;
+
+    impl fmt::Display for MustNotFormat {
+        fn fmt(&self, _: &mut fmt::Formatter<'_>) -> fmt::Result {
+            panic!("discarded diagnostics must not format errors")
+        }
+    }
+
+    impl std::error::Error for MustNotFormat {}
+
+    fn failure(error: io::Error) -> io::Error {
+        diagnostic_error(error, "127.0.0.1".parse().unwrap(), true)
+    }
+
+    #[test]
+    fn disabled_diagnostics_do_not_format_errors() {
+        let mut errors = ScanErrors::new(false, 1000);
+        errors.record(diagnostic_error(
+            io::Error::other(MustNotFormat),
+            "127.0.0.1".parse().unwrap(),
+            false,
+        ));
+        assert!(errors.messages().is_empty());
+    }
+
+    #[test]
+    fn disabled_diagnostics_preserve_the_original_os_error() {
+        let original = io::Error::from_raw_os_error(123);
+        let error = diagnostic_error(original, "127.0.0.1".parse().unwrap(), false);
+        assert_eq!(error.raw_os_error(), Some(123));
+    }
+
+    #[test]
+    fn full_diagnostics_do_not_format_more_errors() {
+        let mut errors = ScanErrors::new(true, 1);
+        errors.record(failure(io::ErrorKind::ConnectionRefused.into()));
+        errors.record(failure(io::Error::other(MustNotFormat)));
+        assert_eq!(errors.messages().len(), 1);
+    }
+
+    #[test]
+    fn enabled_diagnostics_keep_error_details_and_deduplicate_per_ip() {
+        let mut errors = ScanErrors::new(true, 1000);
+        for port in [80, 443] {
+            errors.record(diagnostic_error(
+                io::Error::other("test socket error"),
+                SocketAddr::from(([127, 0, 0, 1], port)).ip(),
+                true,
+            ));
+        }
+        assert_eq!(errors.messages().len(), 1);
+        assert!(errors.messages().contains("test socket error 127.0.0.1"));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn detects_unix_descriptor_limits_by_code() {
+        for code in [libc::EMFILE, libc::ENFILE] {
+            assert!(is_descriptor_exhaustion(&io::Error::from_raw_os_error(
+                code
+            )));
+        }
+        assert!(!is_descriptor_exhaustion(&io::Error::from_raw_os_error(
+            libc::ECONNREFUSED
+        )));
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn detects_winsock_descriptor_limits_by_code() {
+        use windows_sys::Win32::Networking::WinSock::{WSAECONNREFUSED, WSAEMFILE};
+        assert!(is_descriptor_exhaustion(&io::Error::from_raw_os_error(
+            WSAEMFILE
+        )));
+        assert!(!is_descriptor_exhaustion(&io::Error::from_raw_os_error(
+            WSAECONNREFUSED
+        )));
+    }
+
+    #[cfg(any(unix, windows))]
+    #[test]
+    fn descriptor_check_never_formats_an_error() {
+        assert!(!is_descriptor_exhaustion(&io::Error::other(MustNotFormat)));
+    }
+}

src/scanner/errors.rs

@@ -7,13 +7,16 @@ use log::debug;
 mod socket_iterator;
 use socket_iterator::SocketIterator;
 
+mod errors;
+use errors::{diagnostic_error, is_descriptor_exhaustion, ScanErrors};
+
 use colored::Colorize;
 use futures::stream::{FuturesUnordered, StreamExt};
 use std::collections::BTreeMap;
 use std::future::poll_fn;
 use std::task::Poll;
 use std::{
-    collections::{HashMap, HashSet},
+    collections::HashMap,
     io,
     net::{IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr},
     num::NonZeroU8,
@@ -215,7 +218,8 @@ impl Scanner {
     async fn scan(&self) -> Vec<PortStatus> {
         let ports = filter_excluded_ports(self.port_strategy.order(), &self.exclude_ports);
         let mut found_sockets: Vec<PortStatus> = Vec::new();
-        let mut errors: HashSet<String> = HashSet::new();
+        let mut errors =
+            ScanErrors::new(log::log_enabled!(log::Level::Debug), self.ips.len() * 1000);
 
         // Build UDP payload lookup once (only if we are scanning UDP).
         // This avoids cloning a big map into every spawned future and turns
@@ -250,7 +254,7 @@ impl Scanner {
             }
         }
 
-        debug!("Typical socket connection errors {errors:?}");
+        debug!("Typical socket connection errors {:?}", errors.messages());
         debug!("Sockets found: {:?}", found_sockets);
         found_sockets
     }
@@ -271,7 +275,7 @@ impl Scanner {
         mut sockets: SocketIterator<'_>,
         udp_payloads: &Option<Arc<UdpPayloadLookup>>,
         found_sockets: &mut Vec<PortStatus>,
-        errors: &mut HashSet<String>,
+        errors: &mut ScanErrors,
     ) {
         let mut ftrs = FuturesUnordered::new();
 
@@ -319,24 +323,19 @@ impl Scanner {
         &self,
         result: io::Result<PortStatus>,
         found_sockets: &mut Vec<PortStatus>,
-        errors: &mut HashSet<String>,
+        errors: &mut ScanErrors,
     ) {
         match result {
             Ok(status) => found_sockets.push(status),
-            Err(e) => {
-                let error_string = e.to_string();
-                if errors.len() < self.ips.len() * 1000 {
-                    errors.insert(error_string);
-                }
-            }
+            Err(error) => errors.record(error),
         }
     }
 
     /// Given a socket, scan it self.tries times.
     /// Turns the address into a SocketAddr
     /// Deals with the `<result>` type
-    /// If it experiences error ErrorKind::Other then too many files are open and it Panics!
-    /// Else any other error, it returns the error in Result as a string
+    /// Panics if the OS reports descriptor exhaustion.
+    /// Other failures retain their I/O error, with target context for debug diagnostics.
     /// If no errors occur, it returns the port number in Result to signify the port is open.
     /// This function mainly deals with the logic of Results handling.
     /// # Example
@@ -376,14 +375,14 @@ impl Scanner {
                         return Ok(PortStatus::Closed(socket));
                     }
 
-                    let mut error_string = e.to_string();
-
-                    assert!(!error_string.to_lowercase().contains("too many open files"), "Too many open files. Please reduce batch size. The default is 5000. Try -b 2500.");
+                    assert!(!is_descriptor_exhaustion(&e), "Too many open files. Please reduce batch size. The default is 5000. Try -b 2500.");
 
                     if nr_try == tries {
-                        error_string.push(' ');
-                        error_string.push_str(&socket.ip().to_string());
-                        return Err(io::Error::other(error_string));
+                        return Err(diagnostic_error(
+                            e,
+                            socket.ip(),
+                            log::log_enabled!(log::Level::Debug),
+                        ));
                     }
                 }
             };
@@ -606,6 +605,7 @@ mod tests {
     use super::*;
 
     use crate::input::{PortRanges, ScanOrder};
+    use std::collections::HashSet;
 
     // These tests never open sockets: they only build a `Scanner` (and its
     // futures), run a scan that has no socket to scan, or inspect the payload