Harden SMB startup, quieter protocol errors, and add tests/benches

Three resilience changes plus the tests and benchmarks for them.

Setup action: dump the spiceio log on failure (both unexpected exit and
timeout). The log was being written to RUNNER_TEMP but never echoed, so
when startup stalled past the grace window there was no way to diagnose
which phase hung. Also doubled the grace from 30s to 60s — the previous
good run took ~12s, leaving only ~2.5x headroom for a slow server day.

TCP connect timeout: TcpStream::connect had no spiceio-level timeout, so
a server dropping SYNs left the OS waiting 75-90s and stalled pool init
past any sensible CI window. Wrapped in tokio::time::timeout(15s) with
explicit TimedOut error.

Pool connection retry: extracted retry_with_backoff helper used by
SmbPool::connect with a 250ms/750ms/2s schedule (4 attempts). A flaky
connection during startup no longer takes down the whole pool init.

Quieter protocol-layer logging: smb_status_to_io_error was emitting an
error log for every SMB status, including expected ones (NotFound on HEAD
probes, SharingViolation during WAL cleanup). Removed the unconditional
log; mapped statuses return their typed io::Error silently and the
catchall arm still logs for truly unknown statuses. Added
STATUS_SHARING_VIOLATION (0xC0000043) -> ErrorKind::ResourceBusy.

Tests (+21, now 142 total):
- smb_status_to_io_error: full mapping coverage including the new
  ResourceBusy case, unknown-status fallback, STATUS_SUCCESS panic guard,
  path preservation
- retry_with_backoff: first-attempt success, success after transient
  failures, exhaustion preserving last error, empty-backoff edge case,
  elapsed-time floor from the schedule, and the structural invariant on
  CONNECT_RETRY_BACKOFF
- parse_compound_response (moved from client.rs to protocol.rs as pub):
  single/multi-message, empty, truncated header, malformed next_command

Benches (+3 in protocol_bench.rs):
- parse_compound_response over n=2,4,8 chained messages
- pipelined_read_decode at (depth, chunk_size) = (8,64K), (64,64K), (64,8K)
  — the GetObject hot-path inner loop with throughput reporting
- pipelined_write_encode at (depth, chunk_size) = (8,64K), (64,64K),
  (64,1M) — the WAL pipelined-write inner loop

Author: lukekim

.github/actions/setup/action.yml

@@ -132,9 +132,12 @@ runs:
         # auto-increment if the requested port was busy).
         echo "Waiting for spiceio on ${SPICEIO_BIND}..."
         ENDPOINT=""
-        for i in $(seq 1 30); do
+        for i in $(seq 1 60); do
           if ! kill -0 "$PID" 2>/dev/null; then
             echo "::error::spiceio exited unexpectedly"
+            echo "::group::spiceio log"
+            cat "$SPICEIO_LOG" 2>/dev/null || echo "(log file missing)"
+            echo "::endgroup::"
             exit 1
           fi
           ENDPOINT=$(grep 'listening on' "$SPICEIO_LOG" 2>/dev/null | grep -o 'http://[^ ]*' | tail -1 || true)
@@ -146,6 +149,10 @@ runs:
           fi
           sleep 1
         done
-        echo "::error::spiceio failed to start within 30s"
+        echo "::error::spiceio failed to start within 60s"
+        echo "::group::spiceio log"
+        cat "$SPICEIO_LOG" 2>/dev/null || echo "(log file missing)"
+        echo "::endgroup::"
+        kill "$PID" 2>/dev/null || true
         exit 1
 

benches/protocol_bench.rs

@@ -152,6 +152,133 @@ fn bench_encode_set_info_rename(c: &mut Criterion) {
     group.finish();
 }
 
+// ── Parser benches for new public API ────────────────────────────────────────
+
+/// Bench parsing of a compound response (chained SMB2 messages in one frame).
+/// Compound responses are the wire format for create+read+close and similar
+/// batched operations — relevant CPU cost when the S3 layer issues compounds.
+fn bench_parse_compound_response(c: &mut Criterion) {
+    let mut group = c.benchmark_group("parse_compound_response");
+    // Body size of 16 bytes is representative of close/create-response payloads.
+    let body_len = 16usize;
+    let entry_size = SMB2_HEADER_SIZE + body_len;
+    for n in [2usize, 4, 8] {
+        let mut data = vec![0u8; entry_size * n];
+        for i in 0..n {
+            let mut hdr = Header::new(Command::Create, i as u64);
+            hdr.next_command = if i + 1 < n { entry_size as u32 } else { 0 };
+            let mut buf = BytesMut::with_capacity(SMB2_HEADER_SIZE);
+            hdr.encode(&mut buf);
+            let start = i * entry_size;
+            data[start..start + SMB2_HEADER_SIZE].copy_from_slice(&buf);
+            for b in &mut data[start + SMB2_HEADER_SIZE..start + entry_size] {
+                *b = 0xAB;
+            }
+        }
+        group.bench_with_input(
+            criterion::BenchmarkId::from_parameter(n),
+            &data,
+            |b, data| b.iter(|| parse_compound_response(black_box(data))),
+        );
+    }
+    group.finish();
+}
+
+/// Build one framed SMB2 read response message (header + read response body +
+/// data) ready for `Header::decode` + `decode_read_response_owned`.
+fn build_read_response_msg(msg_id: u64, data_len: usize) -> Vec<u8> {
+    let body_len = 16 + data_len;
+    let mut msg = vec![0u8; SMB2_HEADER_SIZE + body_len];
+
+    let mut hdr_buf = BytesMut::with_capacity(SMB2_HEADER_SIZE);
+    let mut hdr = Header::new(Command::Read, msg_id);
+    hdr.status = 0;
+    hdr.encode(&mut hdr_buf);
+    msg[..SMB2_HEADER_SIZE].copy_from_slice(&hdr_buf);
+
+    let body = &mut msg[SMB2_HEADER_SIZE..];
+    // StructureSize = 17
+    body[0..2].copy_from_slice(&17u16.to_le_bytes());
+    // DataOffset (from start of SMB2 message)
+    let data_offset = (SMB2_HEADER_SIZE + 16) as u16;
+    body[2..4].copy_from_slice(&data_offset.to_le_bytes());
+    // DataLength
+    body[4..8].copy_from_slice(&(data_len as u32).to_le_bytes());
+    // Remaining 8 bytes (DataRemaining + Flags) stay zero. Data bytes stay zero.
+    msg
+}
+
+/// Bench the CPU-bound per-batch work of `pipelined_read`: header decode,
+/// slot computation from message_id, and `decode_read_response_owned`. This
+/// is the inner loop of GetObject streaming once the wire bytes are in.
+fn bench_pipelined_read_decode(c: &mut Criterion) {
+    let mut group = c.benchmark_group("pipelined_read_decode");
+    // (depth, chunk_size) — depth=64 matches PIPELINE_DEPTH in ops.rs.
+    let cases = [(8usize, 65536usize), (64, 65536), (64, 8192)];
+    for (depth, chunk_size) in cases {
+        let base_msg_id = 1_000u64;
+        let messages: Vec<Vec<u8>> = (0..depth)
+            .map(|i| build_read_response_msg(base_msg_id + i as u64, chunk_size))
+            .collect();
+        group.throughput(criterion::Throughput::Bytes((depth * chunk_size) as u64));
+        group.bench_with_input(
+            criterion::BenchmarkId::from_parameter(format!("d{depth}_c{chunk_size}")),
+            &messages,
+            |b, messages| {
+                b.iter(|| {
+                    let n = messages.len();
+                    let mut slots: Vec<Option<bytes::Bytes>> = (0..n).map(|_| None).collect();
+                    for msg in messages.iter() {
+                        let header = Header::decode(black_box(msg)).unwrap();
+                        let slot = header.message_id.wrapping_sub(base_msg_id) as usize;
+                        let body = msg[SMB2_HEADER_SIZE..].to_vec();
+                        slots[slot] = decode_read_response_owned(body);
+                    }
+                    slots
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
+/// Bench the CPU-bound per-batch work of `pipelined_write`: header construction
+/// (with credit charge), `encode_write_request`, and `build_request` framing.
+/// This is the inner loop of WAL pipelined writes before any I/O happens.
+fn bench_pipelined_write_encode(c: &mut Criterion) {
+    let mut group = c.benchmark_group("pipelined_write_encode");
+    let file_id = [1u8; 16];
+    // (depth, chunk_size) — depth=64 matches WRITE_PIPELINE_DEPTH in ops.rs.
+    let cases = [(8usize, 65536usize), (64, 65536), (64, 1024 * 1024)];
+    for (depth, chunk_size) in cases {
+        let chunk = vec![0u8; chunk_size];
+        group.throughput(criterion::Throughput::Bytes((depth * chunk_size) as u64));
+        group.bench_with_input(
+            criterion::BenchmarkId::from_parameter(format!("d{depth}_c{chunk_size}")),
+            &chunk,
+            |b, chunk| {
+                b.iter(|| {
+                    let mut packets = Vec::with_capacity(depth);
+                    let mut offset = 0u64;
+                    for i in 0..depth {
+                        let mut hdr = Header::new(Command::Write, i as u64)
+                            .with_credit_charge(chunk.len() as u32);
+                        hdr.tree_id = 42;
+                        hdr.session_id = 0xdead_beef;
+                        let packet = build_request(&hdr, |buf| {
+                            encode_write_request(buf, &file_id, offset, black_box(chunk));
+                        });
+                        packets.push(packet);
+                        offset += chunk.len() as u64;
+                    }
+                    packets
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
 fn bench_parse_directory_entries(c: &mut Criterion) {
     // Build 50 entries
     let mut data = Vec::new();
@@ -192,6 +319,9 @@ criterion_group!(
     bench_decode_read_response,
     bench_decode_read_response_owned,
     bench_build_request,
+    bench_parse_compound_response,
+    bench_pipelined_read_decode,
+    bench_pipelined_write_encode,
     bench_parse_directory_entries,
 );
 criterion_main!(benches);

src/smb/client.rs

@@ -15,6 +15,11 @@ use bytes::{BufMut, BytesMut};
 /// the SMB server is slow or unresponsive under heavy load.
 const SMB_READ_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// Timeout for the initial TCP handshake to the SMB server. Without this,
+/// a server that drops SYNs leaves the OS waiting ~75-90s, which stalls
+/// pool initialization past any sensible CI window.
+const SMB_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+
 use super::auth;
 use super::protocol::*;
 
@@ -71,16 +76,25 @@ impl SmbClient {
     /// Connect to the SMB server and authenticate.
     pub async fn connect(config: SmbConfig) -> io::Result<Arc<Self>> {
         let addr = format!("{}:{}", config.server, config.port);
-        let stream = match TcpStream::connect(&addr).await {
-            Ok(s) => {
-                crate::slog!("[spiceio] smb tcp connected: {addr}");
-                s
-            }
-            Err(e) => {
-                crate::serr!("[spiceio] smb tcp connect failed: {addr}: {e}");
-                return Err(e);
-            }
-        };
+        let stream =
+            match tokio::time::timeout(SMB_CONNECT_TIMEOUT, TcpStream::connect(&addr)).await {
+                Ok(Ok(s)) => {
+                    crate::slog!("[spiceio] smb tcp connected: {addr}");
+                    s
+                }
+                Ok(Err(e)) => {
+                    crate::serr!("[spiceio] smb tcp connect failed: {addr}: {e}");
+                    return Err(e);
+                }
+                Err(_) => {
+                    let msg = format!(
+                        "smb tcp connect timed out after {}s: {addr}",
+                        SMB_CONNECT_TIMEOUT.as_secs()
+                    );
+                    crate::serr!("[spiceio] {msg}");
+                    return Err(io::Error::new(io::ErrorKind::TimedOut, msg));
+                }
+            };
         stream.set_nodelay(true)?;
 
         // Enlarge socket buffers to 1 MB for large read/write throughput.
@@ -1193,8 +1207,10 @@ fn update_preauth_hash(hash: &mut [u8; 64], message: &[u8]) {
 }
 
 fn smb_status_to_io_error(status: u32, path: &str) -> io::Error {
-    crate::serr!("[spiceio] smb error 0x{status:08X}: {path}");
-    // Map raw status codes directly to avoid losing info through NtStatus enum
+    // Map raw status codes directly to avoid losing info through NtStatus enum.
+    // We deliberately do NOT log here — many of these are expected (NotFound on
+    // HEAD probes, SharingViolation during cleanup) and the io::Error string
+    // carries the status code for any caller that wants to surface it.
     match status {
         0xC000_000F // STATUS_NO_SUCH_FILE
         | 0xC000_0034 // STATUS_OBJECT_NAME_NOT_FOUND
@@ -1212,7 +1228,15 @@ fn smb_status_to_io_error(status: u32, path: &str) -> io::Error {
             format!("already exists: {path}"),
         ),
 
-        _ => io::Error::other(format!("SMB error 0x{status:08X} for {path}")),
+        0xC000_0043 => io::Error::new( // STATUS_SHARING_VIOLATION
+            io::ErrorKind::ResourceBusy,
+            format!("sharing violation: {path}"),
+        ),
+
+        _ => {
+            crate::serr!("[spiceio] smb error 0x{status:08X}: {path}");
+            io::Error::other(format!("SMB error 0x{status:08X} for {path}"))
+        }
     }
 }
 
@@ -1232,45 +1256,81 @@ fn sign_message(msg: &mut [u8], key: &[u8; 16]) {
     msg[SIGNATURE_OFFSET..SIGNATURE_OFFSET + 16].copy_from_slice(&signature);
 }
 
-/// Parse a compound response (multiple SMB2 messages in one frame).
-fn parse_compound_response(msg: &[u8]) -> Vec<(Header, Vec<u8>)> {
-    let mut results = Vec::new();
-    let mut offset = 0;
+// Need this for from_raw_fd
 
-    loop {
-        if offset + SMB2_HEADER_SIZE > msg.len() {
-            break;
-        }
-        let header = match Header::decode(&msg[offset..]) {
-            Some(h) => h,
-            None => break,
-        };
+#[cfg(test)]
+mod tests {
+    use super::*;
 
-        let next = header.next_command as usize;
-        let body_start = offset + SMB2_HEADER_SIZE;
-        let body_end = if next > 0 {
-            let end = offset + next;
-            if end > msg.len() || end < body_start {
-                break;
-            }
-            end
-        } else {
-            msg.len()
-        };
-        if body_start > body_end || body_end > msg.len() {
-            break;
-        }
+    fn assert_kind_and_path(err: &io::Error, kind: io::ErrorKind, needle: &str) {
+        assert_eq!(err.kind(), kind, "wrong kind for {err}");
+        let s = err.to_string();
+        assert!(s.contains(needle), "expected {needle:?} in {s:?}");
+    }
 
-        let body = msg[body_start..body_end].to_vec();
-        results.push((header, body));
+    #[test]
+    fn maps_no_such_file_to_not_found() {
+        let e = smb_status_to_io_error(0xC000_000F, "a\\b");
+        assert_kind_and_path(&e, io::ErrorKind::NotFound, "a\\b");
+    }
 
-        if next == 0 {
-            break;
-        }
-        offset += next;
+    #[test]
+    fn maps_object_name_not_found_to_not_found() {
+        let e = smb_status_to_io_error(0xC000_0034, "missing.txt");
+        assert_kind_and_path(&e, io::ErrorKind::NotFound, "missing.txt");
     }
 
-    results
-}
+    #[test]
+    fn maps_object_path_not_found_to_not_found() {
+        let e = smb_status_to_io_error(0xC000_003A, "dir\\file");
+        assert_kind_and_path(&e, io::ErrorKind::NotFound, "dir\\file");
+    }
 
-// Need this for from_raw_fd
+    #[test]
+    fn maps_object_name_invalid_to_not_found() {
+        let e = smb_status_to_io_error(0xC000_0033, "bad?name");
+        assert_kind_and_path(&e, io::ErrorKind::NotFound, "bad?name");
+    }
+
+    #[test]
+    fn maps_access_denied_to_permission_denied() {
+        let e = smb_status_to_io_error(0xC000_0022, "secret");
+        assert_kind_and_path(&e, io::ErrorKind::PermissionDenied, "secret");
+    }
+
+    #[test]
+    fn maps_name_collision_to_already_exists() {
+        let e = smb_status_to_io_error(0xC000_0035, "dup");
+        assert_kind_and_path(&e, io::ErrorKind::AlreadyExists, "dup");
+    }
+
+    #[test]
+    fn maps_sharing_violation_to_resource_busy() {
+        let e = smb_status_to_io_error(0xC000_0043, ".spiceio-wal\\01-0000");
+        assert_kind_and_path(&e, io::ErrorKind::ResourceBusy, ".spiceio-wal\\01-0000");
+    }
+
+    #[test]
+    fn unknown_status_falls_back_to_other_and_includes_hex() {
+        let e = smb_status_to_io_error(0xC000_00BB, "x");
+        assert_eq!(e.kind(), io::ErrorKind::Other);
+        let s = e.to_string();
+        assert!(s.contains("0xC00000BB"), "expected hex in: {s}");
+        assert!(s.contains("x"), "expected path in: {s}");
+    }
+
+    #[test]
+    fn success_status_zero_falls_through_to_other() {
+        // STATUS_SUCCESS is not really an error, but the mapper must never panic.
+        let e = smb_status_to_io_error(0x0000_0000, "ok");
+        assert_eq!(e.kind(), io::ErrorKind::Other);
+    }
+
+    #[test]
+    fn error_path_is_preserved_verbatim() {
+        // Path containing backslashes, dots, and the WAL prefix must round-trip.
+        let path = ".spiceio-wal\\01778725545751751000-0000";
+        let e = smb_status_to_io_error(0xC000_0043, path);
+        assert!(e.to_string().contains(path));
+    }
+}