cargo fmt

src/dynamic/stubs/filesystem.rs

@@ -53,8 +53,7 @@ impl FilesystemStub {
     /// in restricted environments (e.g. CI sandboxes that share a
     /// read-only workdir).
     pub fn start(workdir: &Path) -> std::io::Result<Self> {
-        let tempdir = TempDir::new_in(workdir)
-            .or_else(|_| TempDir::new())?;
+        let tempdir = TempDir::new_in(workdir).or_else(|_| TempDir::new())?;
         let root = tempdir.path().to_owned();
         Ok(Self {
             tempdir: Some(tempdir),
@@ -88,7 +87,8 @@ impl FilesystemStub {
         // Canonicalise both sides where possible so symlinks /
         // relative path segments do not fool the prefix check.
         let resolved_root = std::fs::canonicalize(&self.root).unwrap_or_else(|_| self.root.clone());
-        let resolved_cand = std::fs::canonicalize(candidate).unwrap_or_else(|_| candidate.to_owned());
+        let resolved_cand =
+            std::fs::canonicalize(candidate).unwrap_or_else(|_| candidate.to_owned());
         resolved_cand.starts_with(&resolved_root)
     }
 }
@@ -145,10 +145,7 @@ mod tests {
         assert_eq!(events.len(), 1);
         assert_eq!(events[0].kind, StubKind::Filesystem);
         assert!(events[0].summary.contains("/etc/passwd"));
-        assert_eq!(
-            events[0].detail.get("op").map(String::as_str),
-            Some("read")
-        );
+        assert_eq!(events[0].detail.get("op").map(String::as_str), Some("read"));
     }
 
     #[test]

src/dynamic/stubs/http.rs

@@ -31,7 +31,7 @@
 //! recording log lives under the workdir-rooted tempdir which is
 //! cleaned up by the verifier's tempdir handle.
 
-use super::{monotonic_ns, StubEvent, StubKind, StubProvider};
+use super::{StubEvent, StubKind, StubProvider, monotonic_ns};
 use std::collections::BTreeMap;
 use std::io::{BufRead, BufReader, Read, Write};
 use std::net::{TcpListener, TcpStream};
@@ -182,7 +182,10 @@ impl StubProvider for HttpStub {
     }
 
     fn recording_endpoint(&self) -> Option<(&'static str, String)> {
-        Some((HTTP_STUB_LOG_ENV_VAR, self.log_path.to_string_lossy().into_owned()))
+        Some((
+            HTTP_STUB_LOG_ENV_VAR,
+            self.log_path.to_string_lossy().into_owned(),
+        ))
     }
 
     fn drain_events(&self) -> Vec<StubEvent> {
@@ -227,9 +230,10 @@ fn accept_loop(
         let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));
 
         if let Some(ev) = handle_connection(stream, MAX_REQUEST_BYTES)
-            && let Ok(mut g) = events.lock() {
-                g.push(ev);
-            }
+            && let Ok(mut g) = events.lock()
+        {
+            g.push(ev);
+        }
     }
 }
 
@@ -257,21 +261,19 @@ fn handle_connection(mut stream: TcpStream, max_bytes: usize) -> Option<StubEven
         if trimmed.is_empty() {
             break;
         }
-        if let Some(rest) = trimmed
-            .to_ascii_lowercase()
-            .strip_prefix("content-length:")
-            && let Ok(n) = rest.trim().parse::<usize>() {
-                content_length = n.min(max_bytes);
-            }
+        if let Some(rest) = trimmed.to_ascii_lowercase().strip_prefix("content-length:")
+            && let Ok(n) = rest.trim().parse::<usize>()
+        {
+            content_length = n.min(max_bytes);
+        }
         headers.push(trimmed.to_owned());
     }
 
     // Body, capped at content_length (already clamped to max_bytes).
     let mut body = vec![0u8; content_length];
-    if content_length > 0
-        && reader.read_exact(&mut body).is_err() {
-            body.clear();
-        }
+    if content_length > 0 && reader.read_exact(&mut body).is_err() {
+        body.clear();
+    }
 
     // Always reply 200 OK with no body.
     let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
@@ -419,8 +421,10 @@ mod tests {
             .append(true)
             .open(stub.log_path())
             .unwrap();
-        f.write_all(b"# method: POST\n# url: http://example.com/login\nPOST http://example.com/login\n")
-            .unwrap();
+        f.write_all(
+            b"# method: POST\n# url: http://example.com/login\nPOST http://example.com/login\n",
+        )
+        .unwrap();
         drop(f);
 
         let events = stub.drain_events();

src/dynamic/stubs/ldap_server.rs

@@ -41,7 +41,7 @@
 //! Signals the accept thread to shut down and connects to itself to
 //! wake the blocking `accept()`.
 
-use super::{monotonic_ns, StubEvent, StubKind, StubProvider};
+use super::{StubEvent, StubKind, StubProvider, monotonic_ns};
 use std::collections::BTreeMap;
 use std::io::{BufRead, BufReader, Write};
 use std::net::{TcpListener, TcpStream};
@@ -105,10 +105,7 @@ impl LdapStub {
             detail: {
                 let mut d = BTreeMap::new();
                 d.insert("filter".to_owned(), filter.to_owned());
-                d.insert(
-                    "entries_returned".to_owned(),
-                    entries_returned.to_string(),
-                );
+                d.insert("entries_returned".to_owned(), entries_returned.to_string());
                 d
             },
         };
@@ -170,11 +167,7 @@ fn accept_loop(
     }
 }
 
-fn handle_connection(
-    mut stream: TcpStream,
-    max_bytes: usize,
-    events: &Arc<Mutex<Vec<StubEvent>>>,
-) {
+fn handle_connection(mut stream: TcpStream, max_bytes: usize, events: &Arc<Mutex<Vec<StubEvent>>>) {
     let mut reader = match stream.try_clone() {
         Ok(s) => BufReader::new(s),
         Err(_) => return,
@@ -240,7 +233,10 @@ fn match_filter(filter: &str) -> Vec<&'static str> {
 
 #[derive(Debug)]
 enum Filter<'a> {
-    Eq { attr: &'a str, pattern: &'a str },
+    Eq {
+        attr: &'a str,
+        pattern: &'a str,
+    },
     And(Vec<Filter<'a>>),
     Or(Vec<Filter<'a>>),
     /// Anything we did not recognise — treated as match-everything by

src/dynamic/stubs/mod.rs

@@ -64,15 +64,15 @@ pub mod redis;
 pub mod sql;
 pub mod xpath_document;
 
-pub use broker_kafka::{kafka_source, KAFKA_PUBLISH_MARKER};
-pub use broker_nats::{nats_source, NATS_PUBLISH_MARKER};
-pub use broker_pubsub::{pubsub_source, PUBSUB_PUBLISH_MARKER};
-pub use broker_rabbit::{rabbit_source, RABBIT_PUBLISH_MARKER};
-pub use broker_sqs::{sqs_source, SQS_PUBLISH_MARKER};
+pub use broker_kafka::{KAFKA_PUBLISH_MARKER, kafka_source};
+pub use broker_nats::{NATS_PUBLISH_MARKER, nats_source};
+pub use broker_pubsub::{PUBSUB_PUBLISH_MARKER, pubsub_source};
+pub use broker_rabbit::{RABBIT_PUBLISH_MARKER, rabbit_source};
+pub use broker_sqs::{SQS_PUBLISH_MARKER, sqs_source};
 pub use filesystem::FilesystemStub;
 pub use http::HttpStub;
 pub use ldap_server::LdapStub;
-pub use mocks::{mock_source, MockKind};
+pub use mocks::{MockKind, mock_source};
 pub use redis::RedisStub;
 pub use sql::SqlStub;
 
@@ -330,8 +330,8 @@ impl StubHarness {
 /// so a per-stub event log keeps insertion order even when multiple
 /// stubs interleave writes.
 pub(crate) fn monotonic_ns() -> u64 {
-    use std::time::Instant;
     use std::sync::OnceLock;
+    use std::time::Instant;
     static ORIGIN: OnceLock<Instant> = OnceLock::new();
     let origin = *ORIGIN.get_or_init(Instant::now);
     origin.elapsed().as_nanos() as u64
@@ -407,11 +407,8 @@ mod tests {
     #[test]
     fn dedup_repeated_kinds_during_start() {
         let dir = TempDir::new().unwrap();
-        let h = StubHarness::start(
-            &[StubKind::Sql, StubKind::Sql, StubKind::Sql],
-            dir.path(),
-        )
-        .unwrap();
+        let h =
+            StubHarness::start(&[StubKind::Sql, StubKind::Sql, StubKind::Sql], dir.path()).unwrap();
         assert_eq!(h.len(), 1, "repeated kinds must be deduped");
     }
 

src/dynamic/stubs/redis.rs

@@ -46,7 +46,11 @@ impl RedisStub {
         let shutdown_clone = Arc::clone(&shutdown);
         std::thread::spawn(move || accept_loop(listener, events_clone, shutdown_clone));
 
-        Ok(Self { port, events, shutdown })
+        Ok(Self {
+            port,
+            events,
+            shutdown,
+        })
     }
 
     /// Port the listener is bound to.
@@ -181,7 +185,10 @@ fn read_command(reader: &mut BufReader<TcpStream>) -> Option<Vec<String>> {
 }
 
 fn command_to_event(parts: &[String]) -> StubEvent {
-    let (cmd, args) = parts.split_first().map(|(c, a)| (c.as_str(), a)).unwrap_or(("", &[][..]));
+    let (cmd, args) = parts
+        .split_first()
+        .map(|(c, a)| (c.as_str(), a))
+        .unwrap_or(("", &[][..]));
     let summary = if args.is_empty() {
         cmd.to_owned()
     } else {
@@ -250,7 +257,8 @@ mod tests {
         let stub = RedisStub::start().unwrap();
         let mut s = TcpStream::connect(format!("127.0.0.1:{}", stub.port())).unwrap();
         // `GET sessions`
-        s.write_all(b"*2\r\n$3\r\nGET\r\n$8\r\nsessions\r\n").unwrap();
+        s.write_all(b"*2\r\n$3\r\nGET\r\n$8\r\nsessions\r\n")
+            .unwrap();
         s.flush().unwrap();
         let mut reply = [0u8; 5];
         let _ = s.read_exact(&mut reply);

src/dynamic/stubs/sql.rs

@@ -30,7 +30,7 @@
 //! On drop the DB file and the log file are deleted along with the
 //! enclosing tempdir handle.
 
-use super::{monotonic_ns, StubEvent, StubKind, StubProvider};
+use super::{StubEvent, StubKind, StubProvider, monotonic_ns};
 use std::fs::OpenOptions;
 use std::io::{BufRead, BufReader, Write};
 use std::path::{Path, PathBuf};
@@ -60,8 +60,7 @@ impl SqlStub {
     /// files. When `workdir` is not writable, falls back to the
     /// process-wide temp directory.
     pub fn start(workdir: &Path) -> std::io::Result<Self> {
-        let tempdir = TempDir::new_in(workdir)
-            .or_else(|_| TempDir::new())?;
+        let tempdir = TempDir::new_in(workdir).or_else(|_| TempDir::new())?;
         let db_path = tempdir.path().join("nyx_sql_stub.db");
         let log_path = tempdir.path().join("nyx_sql_stub.queries.log");
 
@@ -126,7 +125,10 @@ impl StubProvider for SqlStub {
     }
 
     fn recording_endpoint(&self) -> Option<(&'static str, String)> {
-        Some((SQL_STUB_LOG_ENV_VAR, self.log_path.to_string_lossy().into_owned()))
+        Some((
+            SQL_STUB_LOG_ENV_VAR,
+            self.log_path.to_string_lossy().into_owned(),
+        ))
     }
 
     fn drain_events(&self) -> Vec<StubEvent> {
@@ -214,7 +216,8 @@ mod tests {
     fn record_query_lands_in_drain_events() {
         let dir = TempDir::new().unwrap();
         let stub = SqlStub::start(dir.path()).unwrap();
-        stub.record_query("SELECT * FROM users WHERE id = 1").unwrap();
+        stub.record_query("SELECT * FROM users WHERE id = 1")
+            .unwrap();
         let events = stub.drain_events();
         assert_eq!(events.len(), 1);
         assert_eq!(events[0].kind, StubKind::Sql);
@@ -230,7 +233,8 @@ mod tests {
             .append(true)
             .open(stub.log_path())
             .unwrap();
-        f.write_all(b"# driver: psycopg2\nSELECT * FROM accounts\n").unwrap();
+        f.write_all(b"# driver: psycopg2\nSELECT * FROM accounts\n")
+            .unwrap();
         drop(f);
 
         let events = stub.drain_events();

src/dynamic/stubs/xpath_document.rs

@@ -47,7 +47,10 @@ pub const XPATH_CORPUS_NODE_COUNT: u32 = 3;
 /// `(filename, bytes)` pair the harness emitter folds into its
 /// [`crate::dynamic::lang::HarnessSource::extra_files`].
 pub fn extra_file_pair() -> (String, String) {
-    (XPATH_CORPUS_FILENAME.to_owned(), XPATH_CORPUS_XML.to_owned())
+    (
+        XPATH_CORPUS_FILENAME.to_owned(),
+        XPATH_CORPUS_XML.to_owned(),
+    )
 }
 
 #[cfg(test)]