fix(supervisor-network): keep workload bytes read with a mediated CONNECT header

[dims]

Summary

A mediated open's first request could be read together with the synthesized CONNECT header and dropped, leaving the workload waiting until its own timeout. Take the header out of the BufReader with fill_buf/consume up to the terminator so the trailing bytes stay buffered for the relay.

Related Issue

No issue required: localized bug fix with a reproducing test.

Changes

• handle_mediated_connection: fill_buf/consume header scan with a three-byte overlap, instead of a full-width read that bypassed the BufReader.
• Test mediated_connect_keeps_workload_bytes_read_with_the_synthesized_header: the workload writes before the proxy's first read. On main it fails with upstream received "" instead of the workload's request; with the change it passes.

Testing

• cargo test -p openshell-supervisor-network mediated_connect_keeps_workload_bytes: red before, green after
• cargo fmt --check and cargo clippy --tests -- -D warnings on the crate
• Parser microbenchmark from the review, rustc -O: buffered scan 159 ns / 521 ns / 3.3 µs per 128 B / 1 KiB / 8 KiB header, main 198 ns / 1.0 µs / 7.5 µs, payload preserved at every size
• Unit tests added/updated
• E2E tests added/updated (not applicable)
• Seen in the field on Agent Substrate micro-VMs as about one first open after a restore in ten. 122 consecutive demo runs across two hosts passed with the one-byte version of this fix; the unpatched supervisor stalled about one run in six.

Checklist

• Follows Conventional Commits
• Commits are signed off (DCO)
• Architecture docs updated (not applicable)

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[dims]

cc @drew

[drew]

I reproduced the reported failure with the regression scenario: reverting the header read to client.read(&mut buf[used..]) made the upstream see an early EOF before the workload request. The PR's one-byte read and a buffered parser both passed that test.

I also compared the header parsers with an optimized standalone microbenchmark (rustc -O). Each input was a 128-byte, 1-KiB, or 8-KiB header followed immediately by BODY; the reader capacity was 8 KiB. Times are medians of five runs (20,000 / 5,000 / 300 iterations per run, respectively):

Header	main full-width read	PR one-byte read	Buffered scan
128 bytes	0.118 µs	2.371 µs	0.089 µs
1 KiB	0.592 µs	138.545 µs	0.361 µs
8 KiB	4.339 µs	8.716 ms	2.539 µs

The main parser lost BODY for the 128-byte and 1-KiB cases; both proposed parsers preserved it. At 8 KiB, main happened to preserve it because the header exactly filled the first read.

The one-byte fix is correct, but the existing terminator search rescans the accumulated header after every byte, giving quadratic work for long headers. I suggest using BufReader::fill_buf() and consume() only through the terminator, scanning the new chunk plus up to three bytes from the prior chunk. That preserved the regression test's workload bytes and stayed close to main's parser cost in this microbenchmark. These numbers measure parser CPU with std::io::BufReader, not end-to-end Tokio proxy throughput.

Standalone benchmark source

use std::hint::black_box;
use std::io::{BufRead, BufReader, Cursor, Read};
use std::time::Instant;

const MAX: usize = 8192;

fn parse(input: &[u8], mode: u8, verify_payload: bool) -> (usize, bool) {
    let mut reader = BufReader::with_capacity(MAX, Cursor::new(input));
    let mut buf = [0_u8; MAX];
    let mut used = 0;
    let header_end = loop {
        assert!(used < MAX);
        if mode == 2 {
            let available = reader.fill_buf().unwrap();
            assert!(!available.is_empty());
            let n = available.len().min(MAX - used);
            buf[used..used + n].copy_from_slice(&available[..n]);
            let start = used.saturating_sub(3);
            let end = buf[start..used + n]
                .windows(4)
                .position(|w| w == b"\r\n\r\n")
                .map(|p| start + p + 4);
            let consumed = end.map_or(n, |end| end - used);
            reader.consume(consumed);
            used += consumed;
            if let Some(end) = end { break end; }
        } else if mode == 3 {
            let n = reader.read(&mut buf[used..]).unwrap();
            assert!(n > 0);
            used += n;
            if buf[..used].windows(4).any(|w| w == b"\r\n\r\n") {
                break buf[..used].windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
            }
        } else {
            let n = reader.read(&mut buf[used..=used]).unwrap();
            assert_eq!(n, 1);
            used += n;
            let found = if mode == 0 {
                buf[..used].windows(4).any(|w| w == b"\r\n\r\n")
            } else {
                used >= 4 && &buf[used - 4..used] == b"\r\n\r\n"
            };
            if found {
                break buf[..used].windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
            }
        }
    };
    let preserved = if verify_payload {
        let mut body = [0_u8; 4];
        reader.read_exact(&mut body).is_ok() && &body == b"BODY"
    } else {
        false
    };
    (header_end, preserved)
}

fn main() {
    for header_len in [128, 1024, 8192] {
        let mut input = b"GET / HTTP/1.1\r\nX: ".to_vec();
        input.resize(header_len - 4, b'x');
        input.extend_from_slice(b"\r\n\r\nBODY");
        let iterations = match header_len { 128 => 20_000, 1024 => 5_000, _ => 300 };
        for (name, mode) in [("main full read", 3), ("PR byte scan", 0), ("buffered scan", 2)] {
            let (_, preserved) = parse(&input, mode, true);
            let mut samples = Vec::new();
            for _ in 0..5 {
                let started = Instant::now();
                for _ in 0..iterations {
                    assert_eq!(black_box(parse(black_box(&input), mode, false)).0, header_len);
                }
                samples.push(started.elapsed() / iterations);
            }
            samples.sort();
            println!("{header_len:5} bytes | {name:16} | median {:?}/header | payload preserved: {preserved}", samples[2]);
        }
    }
}

[dims]

I suggest using BufReader::fill_buf() and consume() only through the terminator, scanning the new chunk plus up to three bytes from the prior chunk.

Done!