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

.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
 

.github/workflows/ci.yml

@@ -96,6 +96,37 @@ jobs:
           SPICEIO_REGION: ${{ vars.SPICEIO_REGION || 'us-west-1' }}
         run: ./scripts/test-sccache.sh
 
+      - name: Run extended S3 operations test
+        # Multipart, range, multi-delete, conditional writes, list-under-load,
+        # streaming cancellation. Uses bucket 'extended' and port 18336 so it
+        # doesn't collide with the sccache test above.
+        if: ${{ env.HAS_SMB_PASS == 'true' }}
+        env:
+          SPICEIO_SMB_SERVER: ${{ vars.SPICEIO_SMB_SERVER || '192.168.3.148' }}
+          SPICEIO_SMB_USER: ${{ vars.SPICEIO_SMB_USER || 'runner' }}
+          SPICEIO_SMB_PASS: ${{ secrets.UNAS_SMB_PASS }}
+          SPICEIO_SMB_SHARE: ${{ vars.SPICEIO_SMB_SHARE || 'ai_platform_dev' }}
+          SPICEIO_BUCKET: extended
+          SPICEIO_REGION: ${{ vars.SPICEIO_REGION || 'us-west-1' }}
+          SPICEIO_BIND: 127.0.0.1:18336
+          AWS_DEFAULT_REGION: ${{ vars.SPICEIO_REGION || 'us-west-1' }}
+        run: ./scripts/test-extended.sh
+
+      - name: Run concurrent stress test
+        # Concurrent writes/reads/contention, write-then-read patterns,
+        # mixed read/write on same key, and large-file pipelined I/O.
+        if: ${{ env.HAS_SMB_PASS == 'true' }}
+        env:
+          SPICEIO_SMB_SERVER: ${{ vars.SPICEIO_SMB_SERVER || '192.168.3.148' }}
+          SPICEIO_SMB_USER: ${{ vars.SPICEIO_SMB_USER || 'runner' }}
+          SPICEIO_SMB_PASS: ${{ secrets.UNAS_SMB_PASS }}
+          SPICEIO_SMB_SHARE: ${{ vars.SPICEIO_SMB_SHARE || 'ai_platform_dev' }}
+          SPICEIO_BUCKET: stress
+          SPICEIO_REGION: ${{ vars.SPICEIO_REGION || 'us-west-1' }}
+          SPICEIO_BIND: 127.0.0.1:18335
+          AWS_DEFAULT_REGION: ${{ vars.SPICEIO_REGION || 'us-west-1' }}
+        run: ./scripts/stress-concurrent.sh
+
       - name: Build release artifact
         run: cargo build --release --locked --bin spiceio
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "spiceio"
-version = "0.5.1"
+version = "0.5.2"
 edition = "2024"
 description = "S3-compatible API proxy to SMB file shares"
 license = "Apache-2.0"

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);

scripts/stress-concurrent.sh

@@ -34,7 +34,7 @@ SPICEIO_PID=""
 cleanup() {
     echo ""
     echo "[stress] cleaning up..."
-    aws --endpoint-url "$ENDPOINT" --no-sign-request \
+    aws --endpoint-url "$ENDPOINT" --no-sign-request --region "$REGION" \
         s3 rm "s3://${BUCKET}/${PREFIX}/" --recursive --quiet 2>/dev/null || true
     if [[ -n "$SPICEIO_PID" ]]; then
         kill "$SPICEIO_PID" 2>/dev/null || true