feat(lint): centralize clippy::collapsible_if allowance in Cargo.toml and remove redundant file-level declarations

src/dynamic/sandbox/baseline.rs

@@ -152,6 +152,8 @@ fn bind_mount_ro(src: &Path, dst: &Path) -> io::Result<()> {
     let cdst = CString::new(dst.as_os_str().as_bytes())
         .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
 
+    // SAFETY: `csrc`/`cdst` are `CString`s that outlive the call, so the pointers
+    // reference valid NUL-terminated C strings. Return value checked below.
     let bind = unsafe {
         mount(
             csrc.as_ptr(),
@@ -165,6 +167,8 @@ fn bind_mount_ro(src: &Path, dst: &Path) -> io::Result<()> {
         return Err(io::Error::last_os_error());
     }
     // Best-effort read-only remount; leave the rw bind if it fails.
+    // SAFETY: `cdst` outlives the call; the other pointers are null, accepted by
+    // `mount(2)` for a remount.
     unsafe {
         mount(
             std::ptr::null(),

src/dynamic/sandbox/mod.rs

@@ -724,7 +724,7 @@ fn register_exit_cleanup() {
     unsafe extern "C" {
         fn atexit(f: extern "C" fn()) -> i32;
     }
-    // Safety: atexit(3) is async-signal-safe for registration; the handler
+    // SAFETY: atexit(3) is async-signal-safe for registration; the handler
     // itself runs on the main thread during normal shutdown, after all Rust
     // destructors, so std::process::Command is safe to call from it.
     unsafe { atexit(stop_all_containers) };
@@ -1870,6 +1870,7 @@ fn libc_kill(pid: i32, sig: i32) -> i32 {
     unsafe extern "C" {
         fn kill(pid: i32, sig: i32) -> i32;
     }
+    // SAFETY: `kill(2)` takes only scalar args and touches no caller memory.
     unsafe { kill(pid, sig) }
 }
 

src/dynamic/sandbox/process_linux.rs

@@ -29,6 +29,8 @@
 //! record into it, execve(2) drops the write end, and the parent's
 //! drain thread sees EOF and records the outcome.
 
+#![warn(clippy::undocumented_unsafe_blocks)]
+
 use crate::dynamic::sandbox::seccomp;
 use crate::dynamic::sandbox::seccomp::bpf::SockFilter;
 use crate::dynamic::sandbox::{AblationMask, ProcessHardeningProfile, SandboxOptions};
@@ -144,11 +146,14 @@ struct StatusPipe {
 
 impl StatusPipe {
     fn new() -> std::io::Result<Self> {
+        // SAFETY: declares the libc `pipe2(2)` ABI; the signature matches <unistd.h>.
         unsafe extern "C" {
             fn pipe2(pipefd: *mut i32, flags: i32) -> i32;
         }
         const O_CLOEXEC: i32 = 0o2_000_000;
         let mut fds = [-1_i32; 2];
+        // SAFETY: `fds` is a valid 2-element array the kernel writes into; `pipe2`
+        // reads no caller memory beyond that pointer. Return value checked below.
         let ret = unsafe { pipe2(fds.as_mut_ptr(), O_CLOEXEC) };
         if ret != 0 {
             return Err(std::io::Error::last_os_error());
@@ -161,15 +166,20 @@ impl StatusPipe {
 }
 
 fn close_fd(fd: RawFd) {
+    // SAFETY: declares the libc `close(2)` ABI; signature matches <unistd.h>.
     unsafe extern "C" {
         fn close(fd: i32) -> i32;
     }
+    // SAFETY: `fd` is an owned raw fd closed exactly once; the return value is
+    // intentionally ignored (best-effort close).
     unsafe { close(fd) };
 }
 
 /// Drain `read_fd` into a `HardeningOutcome`.  Wire format is the
 /// 15-byte fixed-width record produced by [`encode_outcome`].
 fn drain_outcome(read_fd: RawFd) -> Option<HardeningOutcome> {
+    // SAFETY: `read_fd` is an owned raw fd (the pipe read end) used nowhere else;
+    // `File` takes sole ownership and closes it on drop.
     let mut file = unsafe { std::fs::File::from_raw_fd(read_fd) };
     let mut buf = Vec::with_capacity(64);
     if file.read_to_end(&mut buf).is_err() {
@@ -276,6 +286,8 @@ struct Rlimit {
     max: u64,
 }
 
+// SAFETY: declares the libc syscall-wrapper ABI (setrlimit/prctl/unshare/chroot/
+// chdir/mount/write/__errno_location); signatures match the glibc/musl headers.
 unsafe extern "C" {
     fn setrlimit(resource: i32, rlim: *const Rlimit) -> i32;
     fn prctl(option: i32, arg2: u64, arg3: u64, arg4: u64, arg5: u64) -> i32;
@@ -294,6 +306,8 @@ unsafe extern "C" {
 }
 
 fn last_errno() -> i32 {
+    // SAFETY: `__errno_location` returns a valid pointer to the calling thread's
+    // errno; dereferencing it right after a failed syscall is the standard idiom.
     unsafe { *__errno_location() }
 }
 
@@ -302,6 +316,8 @@ fn apply_rlimit(resource: i32, bytes: u64) -> PrimitiveStatus {
         cur: bytes,
         max: bytes,
     };
+    // SAFETY: `&rl` points to a valid `Rlimit` for the duration of the call;
+    // `setrlimit` only reads it and returns a status checked below.
     let ret = unsafe { setrlimit(resource, &rl) };
     if ret == 0 {
         PrimitiveStatus::Applied
@@ -311,6 +327,8 @@ fn apply_rlimit(resource: i32, bytes: u64) -> PrimitiveStatus {
 }
 
 fn apply_no_new_privs() -> PrimitiveStatus {
+    // SAFETY: `prctl(PR_SET_NO_NEW_PRIVS, ..)` takes only scalar args and touches
+    // no caller memory; the return value is checked below.
     let ret = unsafe { prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) };
     if ret == 0 {
         PrimitiveStatus::Applied
@@ -326,6 +344,8 @@ fn apply_unshare_with_flags(flags: i32) -> PrimitiveStatus {
     // ablation drops individual flags via `AblationMask::no_userns` /
     // `no_pidns` so the escape-fixture matrix can prove the namespace
     // primitive carries its weight.
+    // SAFETY: `unshare` takes a scalar flag set and touches no caller memory;
+    // the return value is checked below.
     let ret = unsafe { unshare(flags) };
     if ret == 0 {
         PrimitiveStatus::Applied
@@ -354,11 +374,14 @@ fn apply_chroot(workdir: &[u8]) -> PrimitiveStatus {
     // `workdir` is NUL-terminated by `canonicalize_workdir` so we can
     // hand the bytes straight to `chroot(2)` without allocating in
     // pre_exec.
+    // SAFETY: `workdir` is NUL-terminated by `canonicalize_workdir`, so the
+    // pointer references a valid C string for the duration of the call.
     let ret = unsafe { chroot(workdir.as_ptr() as *const i8) };
     if ret != 0 {
         return PrimitiveStatus::Failed(last_errno());
     }
     let root = b"/\0";
+    // SAFETY: `root` is a NUL-terminated byte literal, a valid C string.
     let ret = unsafe { chdir(root.as_ptr() as *const i8) };
     if ret != 0 {
         return PrimitiveStatus::Failed(last_errno());
@@ -391,6 +414,9 @@ struct BindMount {
 fn apply_bind_mounts(mounts: &[BindMount]) {
     let none = b"none\0";
     for m in mounts {
+        // SAFETY: `source_nul`/`dest_nul` are NUL-terminated by
+        // `canonicalize_bind_mount` and `none` is a NUL-terminated literal, so
+        // every pointer references a valid C string for the duration of the call.
         let r = unsafe {
             mount(
                 m.source_nul.as_ptr() as *const i8,
@@ -403,6 +429,8 @@ fn apply_bind_mounts(mounts: &[BindMount]) {
         if r != 0 {
             continue;
         }
+        // SAFETY: `dest_nul` is NUL-terminated; the remaining pointers are null,
+        // which `mount(2)` accepts for a remount. Best-effort: result ignored.
         unsafe {
             mount(
                 std::ptr::null(),
@@ -541,7 +569,7 @@ pub fn install_pre_exec(
     let read_fd = pipe.as_ref().map(|p| p.read_fd);
     let plan_for_child = plan.clone();
 
-    // Safety: pre_exec runs after fork(2) and before execve(2).  We must
+    // SAFETY: pre_exec runs after fork(2) and before execve(2).  We must
     // not allocate, take any locks, or call into the Rust runtime.  The
     // captured `plan_for_child` is moved in; reading its already-allocated
     // fields is safe because no allocator call is needed.

src/dynamic/sandbox/seccomp/mod.rs

@@ -28,6 +28,8 @@
 //!   can't be filtered without a number, and any kernel that recognises
 //!   the name has the number too.  Tests assert the policy round-trips.
 
+#![warn(clippy::undocumented_unsafe_blocks)]
+
 pub mod bpf;
 pub mod syscalls;
 
@@ -42,6 +44,8 @@ const PR_SET_NO_NEW_PRIVS: i32 = 38;
 const PR_SET_SECCOMP: i32 = 22;
 const SECCOMP_MODE_FILTER: u64 = 2;
 
+// SAFETY: declares the libc `prctl(2)` / `__errno_location` ABI; signatures
+// match the glibc/musl headers.
 unsafe extern "C" {
     fn prctl(option: i32, arg2: u64, arg3: u64, arg4: u64, arg5: u64) -> i32;
     fn __errno_location() -> *mut i32;
@@ -142,12 +146,16 @@ pub fn install_compiled_filter(program: &[SockFilter]) -> std::io::Result<()> {
     // seccomp filter install.  The Phase 17 hardening sequence already
     // calls it earlier, but installing here too is idempotent and
     // protects direct callers.
+    // SAFETY: `prctl(PR_SET_NO_NEW_PRIVS, ..)` takes only scalar args and touches
+    // no caller memory; idempotent, result intentionally ignored.
     let _ = unsafe { prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) };
 
     let prog = SockFprog {
         len: program.len() as u16,
         filter: program.as_ptr(),
     };
+    // SAFETY: `prog` and the `program` slice it points to outlive the call; the
+    // pointer passed as u64 references a valid `SockFprog`. Return value checked below.
     let ret = unsafe {
         prctl(
             PR_SET_SECCOMP,
@@ -160,6 +168,8 @@ pub fn install_compiled_filter(program: &[SockFilter]) -> std::io::Result<()> {
     if ret == 0 {
         Ok(())
     } else {
+        // SAFETY: `__errno_location` returns a valid per-thread errno pointer,
+        // dereferenced immediately after the failed prctl call.
         Err(std::io::Error::from_raw_os_error(unsafe {
             *__errno_location()
         }))

src/dynamic/stubs/broker.rs

@@ -141,14 +141,23 @@ impl BrokerStub {
             .append(true)
             .create(true)
             .open(&self.log_path)?;
-        writeln!(
-            f,
-            "{}\t{}\t{}",
+        // Build the whole record (including the trailing newline) up front and
+        // emit it in a single `write_all`. A `writeln!` issues one syscall per
+        // format fragment, so a concurrent `drain_events` reader could observe a
+        // torn line (e.g. just `deliver` with no tab) and misclassify it. For a
+        // record small enough to land in one `write()` (the common case) the
+        // append-mode `write_all` is delivered atomically; very large records
+        // can still span multiple `write()`s, so the drain's newline-framing
+        // guard remains the backstop. Both the tab-and-newline-stripped
+        // destination and the newline-stripped payload guarantee the record
+        // occupies exactly one physical line regardless.
+        let line = format!(
+            "{}\t{}\t{}\n",
             action.replace('\t', " "),
-            destination.replace('\t', " "),
-            payload
-        )?;
-        Ok(())
+            destination.replace(['\t', '\n'], " "),
+            payload.replace('\n', " ")
+        );
+        f.write_all(line.as_bytes())
     }
 }
 
@@ -202,17 +211,38 @@ impl StubProvider for BrokerStub {
         }
 
         let mut events = Vec::new();
-        let mut bytes_read = 0_u64;
-        let mut buf = String::new();
+        let mut consumed = 0_u64;
+        let mut buf: Vec<u8> = Vec::new();
         loop {
             buf.clear();
-            let n = match reader.read_line(&mut buf) {
+            // Read raw bytes up to and including the next '\n'. Byte-oriented
+            // (rather than `read_line` into a `String`) so a non-UTF-8 payload
+            // written by an in-sandbox harness — e.g. Go's `string(msg.Data)`
+            // over the shared `NYX_*_LOG` — degrades to a lossy decode instead
+            // of erroring out. With `read_line` such a byte would return `Err`,
+            // and the `Err => break` arm would park the cursor on that line
+            // forever, permanently stalling the stream and dropping every
+            // record after it.
+            let n = match reader.read_until(b'\n', &mut buf) {
                 Ok(0) => break,
                 Ok(n) => n,
                 Err(_) => break,
             };
-            bytes_read += n as u64;
-            let line = buf.trim_end_matches(['\r', '\n']);
+            // A chunk that does not end in '\n' is the tail of an in-flight
+            // append: a writer thread is mid-record. Leave it unconsumed (do
+            // not advance the cursor past it) so the next drain re-reads it
+            // once it is complete. Without this guard the partial line would be
+            // skipped forever and, worse, `parse_broker_log_line` would
+            // misclassify a tab-less fragment like `deliver` as a `publish`.
+            if buf.last() != Some(&b'\n') {
+                break;
+            }
+            consumed += n as u64;
+            // Strip exactly the single '\n' line terminator. The log is
+            // newline-framed (never CRLF), so a trailing '\r' is payload data
+            // and must be preserved rather than greedily trimmed.
+            let decoded = String::from_utf8_lossy(&buf[..buf.len() - 1]);
+            let line = decoded.as_ref();
             if line.is_empty() {
                 continue;
             }
@@ -229,7 +259,7 @@ impl StubProvider for BrokerStub {
             };
             events.push(event);
         }
-        *cursor += bytes_read;
+        *cursor += consumed;
         events
     }
 }
@@ -3378,13 +3408,19 @@ fn append_broker_event(
         .append(true)
         .create(true)
         .open(log_path)?;
-    writeln!(
-        f,
-        "{}\t{}\t{}",
+    // Single `write_all` append: see `record_event` for why a `writeln!` is
+    // unsafe against concurrent drains, and for the atomicity caveat on very
+    // large records. The broker server threads append from multiple handlers,
+    // so a torn record is otherwise observable mid-flight. The destination
+    // (path/topic-derived, e.g. a percent-decoded `%0A`) is stripped of tabs
+    // and newlines and the payload of newlines so the record stays one line.
+    let line = format!(
+        "{}\t{}\t{}\n",
         action.replace('\t', " "),
-        destination.replace('\t', " "),
-        payload
-    )
+        destination.replace(['\t', '\n'], " "),
+        payload.replace('\n', " ")
+    );
+    f.write_all(line.as_bytes())
 }
 
 #[cfg(test)]
@@ -4154,7 +4190,10 @@ mod tests {
             "{delivery:?}"
         );
 
-        let events = stub.drain_events();
+        // The MSG frame reaches the wire before the server appends the matching
+        // `deliver` record (see `nats_deliver`), so draining the moment the
+        // payload arrives can race the log write. Poll until both records land.
+        let events = drain_events_until(&stub, 2, Duration::from_secs(5));
         let actions: Vec<&str> = events
             .iter()
             .map(|ev| ev.detail.get("action").unwrap().as_str())