oc2_hlapi_v2 0.2.0

//! `BaseCodec`: framing, timeouts, reconnection, blobs, and stream clearing.
//!
//! The framing tests run on both the rpc (response) and event channels, and with both timeout
//! error kinds that real streams produce (`WouldBlock` on Unix, `TimedOut` on Windows).

mod common;

use std::io;
use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
use std::sync::atomic::Ordering;
use std::thread;
use std::time::{Duration, Instant};

use common::*;
use oc2_hlapi_v2::Codec;
use oc2_hlapi_v2::codecs::BaseCodecError;
use oc2_hlapi_v2::messages::*;
use serde_json::{Value, json};

type Err = BaseCodecError<io::Error>;

#[derive(Clone, Copy, Debug)]
enum Chan {
    Rpc,
    Event,
}

impl Chan {
    fn stream(self, h: &Host) -> &MemStream {
        match self {
            Chan::Rpc => &h.rpc,
            Chan::Event => &h.event,
        }
    }
    /// A message on this channel carrying `n` in its generation field.
    fn msg(self, n: u32) -> Value {
        match self {
            Chan::Rpc => json!({"type": "list", "id": 0, "gen": n, "data": []}),
            Chan::Event => json!({"type": "devicesChanged", "gen": n}),
        }
    }
    /// Read one message, returning the number from its generation field.
    fn read(self, c: &mut MockCodec, t: Option<Duration>) -> Result<Option<u32>, Err> {
        match self {
            Chan::Rpc => Ok(c.read_response(t)?.map(|r| r.generation.unwrap())),
            Chan::Event => Ok(c.read_event(t)?.map(|e| e.generation)),
        }
    }
}

/// Run `f` for every channel and timeout error kind, naming the combination on failure.
fn each(f: impl Fn(Chan, TimeoutKind)) {
    for chan in [Chan::Rpc, Chan::Event] {
        for kind in KINDS {
            if let Err(e) = catch_unwind(AssertUnwindSafe(|| f(chan, kind))) {
                eprintln!("failed for {chan:?} / {kind:?}");
                resume_unwind(e);
            }
        }
    }
}

fn ms(n: u64) -> Option<Duration> {
    Some(Duration::from_millis(n))
}

fn is_eof(e: &Err) -> bool {
    matches!(e, BaseCodecError::Io(e) if e.kind() == io::ErrorKind::UnexpectedEof)
}

// --- framing ------------------------------------------------------------------------------------

#[test]
fn reads_a_framed_message() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        chan.stream(&h).send_json(&chan.msg(5));
        assert_eq!(chan.read(&mut c, Some(LONG)).unwrap(), Some(5));
    });
}

#[test]
fn reads_several_messages_in_order_from_one_write() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let mut bytes = Vec::new();
        for n in 1..=3 {
            bytes.extend(framed(&chan.msg(n)));
        }
        chan.stream(&h).send(&bytes).unwrap();
        for n in 1..=3 {
            assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(n));
        }
        assert_eq!(chan.read(&mut c, Some(Duration::ZERO)).unwrap(), None);
    });
}

#[test]
fn the_leading_nul_is_optional() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let mut bytes = serde_json::to_vec(&chan.msg(1)).unwrap();
        bytes.push(0);
        chan.stream(&h).send(&bytes).unwrap();
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(1));
    });
}

#[test]
fn empty_frames_are_ignored() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let mut bytes = vec![0, 0, 0, 0];
        bytes.extend(framed(&chan.msg(8)));
        bytes.extend([0, 0]);
        chan.stream(&h).send(&bytes).unwrap();
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(8));
        assert_eq!(chan.read(&mut c, Some(Duration::ZERO)).unwrap(), None);
    });
}

#[test]
fn garbage_frames_are_dropped() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let mut bytes = b"\0this is not json\0\0{\"truncated\": \0".to_vec();
        bytes.extend(framed(&chan.msg(3)));
        chan.stream(&h).send(&bytes).unwrap();
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(3));
    });
}

#[test]
fn a_message_split_across_many_writes_is_reassembled() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let bytes = framed(&chan.msg(21));
        let writer = thread::spawn(move || {
            for chunk in bytes.chunks(3) {
                chan.stream(&h).send(chunk).unwrap();
                thread::sleep(Duration::from_millis(2));
            }
            h // keep the connection open until the read is done
        });
        assert_eq!(chan.read(&mut c, Some(LONG)).unwrap(), Some(21));
        drop(writer.join().unwrap());
    });
}

#[test]
fn a_read_that_times_out_mid_frame_keeps_the_partial_frame() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let bytes = framed(&chan.msg(4));
        let (first, rest) = bytes.split_at(bytes.len() / 2);
        chan.stream(&h).send(first).unwrap();
        assert_eq!(chan.read(&mut c, ms(40)).unwrap(), None);
        chan.stream(&h).send(rest).unwrap();
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(4));
    });
}

#[test]
fn large_messages_are_read_whole() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let mut msg = chan.msg(6);
        msg["padding"] = json!("x".repeat(200_000));
        chan.stream(&h).send_json(&msg);
        assert_eq!(chan.read(&mut c, Some(LONG)).unwrap(), Some(6));
    });
}

// --- timeouts -----------------------------------------------------------------------------------

#[test]
fn an_idle_read_times_out_after_about_the_timeout() {
    each(|chan, kind| {
        let (mut c, _h) = codec(kind);
        let start = Instant::now();
        assert_eq!(chan.read(&mut c, ms(80)).unwrap(), None);
        let took = start.elapsed();
        assert!(
            took >= Duration::from_millis(75),
            "returned early: {took:?}"
        );
        assert!(took < Duration::from_secs(5), "returned late: {took:?}");
    });
}

#[test]
fn a_zero_timeout_polls() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let start = Instant::now();
        assert_eq!(chan.read(&mut c, Some(Duration::ZERO)).unwrap(), None);
        assert!(start.elapsed() < Duration::from_secs(2));
        chan.stream(&h).send_json(&chan.msg(1));
        assert_eq!(chan.read(&mut c, Some(Duration::ZERO)).unwrap(), Some(1));
    });
}

#[test]
fn a_poll_does_not_leave_the_stream_nonblocking() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        assert_eq!(chan.read(&mut c, Some(Duration::ZERO)).unwrap(), None);
        let sender = thread::spawn(move || {
            thread::sleep(Duration::from_millis(60));
            chan.stream(&h).send_json(&chan.msg(2));
            h
        });
        // Would return None at once if the stream were still nonblocking
        assert_eq!(chan.read(&mut c, Some(LONG)).unwrap(), Some(2));
        drop(sender.join().unwrap());
    });
}

#[test]
fn no_timeout_blocks_until_a_message_arrives() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let sender = thread::spawn(move || {
            thread::sleep(Duration::from_millis(80));
            chan.stream(&h).send_json(&chan.msg(9));
            h
        });
        let start = Instant::now();
        assert_eq!(chan.read(&mut c, None).unwrap(), Some(9));
        assert!(start.elapsed() >= Duration::from_millis(60));
        drop(sender.join().unwrap());
    });
}

#[test]
fn the_timeout_is_a_deadline_not_a_per_read_limit() {
    // A host that keeps sending frames the codec skips must not keep the read alive forever
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let feeder = thread::spawn(move || {
            let end = Instant::now() + Duration::from_millis(1500);
            while Instant::now() < end {
                if chan.stream(&h).send(b"\0noise\0").is_err() {
                    break;
                }
                thread::sleep(Duration::from_millis(15));
            }
        });
        let start = Instant::now();
        assert_eq!(chan.read(&mut c, ms(250)).unwrap(), None);
        assert!(
            start.elapsed() < Duration::from_millis(1200),
            "{:?}",
            start.elapsed()
        );
        drop(c);
        feeder.join().unwrap();
    });
}

#[test]
fn a_message_that_arrives_before_the_deadline_wins() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        let sender = thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            chan.stream(&h).send_json(&chan.msg(1));
            h
        });
        assert_eq!(chan.read(&mut c, Some(LONG)).unwrap(), Some(1));
        drop(sender.join().unwrap());
    });
}

// --- connection loss ----------------------------------------------------------------------------

#[test]
fn eof_is_an_error_and_the_codec_reconnects() {
    each(|chan, kind| {
        let (mut c, mut hosts, opens) = codec_with_spares(kind, 1);
        let spare = hosts.pop().unwrap();
        drop(hosts); // the first connection's peer goes away
        let err = chan.read(&mut c, ms(500)).unwrap_err();
        assert!(is_eof(&err), "{err:?}");
        assert_eq!(opens.load(Ordering::SeqCst), 2);
        // The new connection works
        chan.stream(&spare).send_json(&chan.msg(1));
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(1));
        assert_eq!(opens.load(Ordering::SeqCst), 2);
    });
}

#[test]
fn eof_without_a_new_connection_reports_the_transport_error() {
    each(|chan, kind| {
        let (mut c, h) = codec(kind);
        drop(h);
        let err = chan.read(&mut c, ms(500)).unwrap_err();
        assert!(matches!(err, BaseCodecError::Transport(_)), "{err:?}");
    });
}

#[test]
fn reconnecting_is_bounded() {
    // Every connection is closed by the peer immediately. One read must not retry forever.
    each(|chan, kind| {
        let transport = MockTransport::with(move || Ok(connection(kind).0));
        let opens = transport.opens.clone();
        let mut c = oc2_hlapi_v2::codecs::BaseCodec::new(transport).unwrap();
        let err = chan.read(&mut c, ms(500)).unwrap_err();
        assert!(is_eof(&err), "{err:?}");
        assert_eq!(
            opens.load(Ordering::SeqCst),
            2,
            "one open in new(), one reconnect"
        );
    });
}

#[test]
fn reconnecting_discards_a_half_received_frame() {
    each(|chan, kind| {
        let (mut c, mut hosts, _) = codec_with_spares(kind, 1);
        let spare = hosts.pop().unwrap();
        let first = hosts.pop().unwrap();
        // Half a message, then the connection dies
        let bytes = framed(&chan.msg(1));
        chan.stream(&first).send(&bytes[..bytes.len() / 2]).unwrap();
        drop(first);
        assert!(is_eof(&chan.read(&mut c, ms(500)).unwrap_err()));
        // The new connection's first message has no leading NUL, so a leftover prefix would be
        // glued onto it and the message lost.
        let mut fresh = serde_json::to_vec(&chan.msg(2)).unwrap();
        fresh.push(0);
        chan.stream(&spare).send(&fresh).unwrap();
        assert_eq!(chan.read(&mut c, ms(500)).unwrap(), Some(2));
    });
}

#[test]
fn reconnecting_replaces_all_three_streams() {
    for kind in KINDS {
        let (mut c, mut hosts, _) = codec_with_spares(kind, 1);
        let spare = hosts.pop().unwrap();
        let Host { rpc, blob, event } = hosts.pop().unwrap();
        drop(rpc);
        assert!(is_eof(&c.read_response(ms(500)).unwrap_err()));
        // The old connection's other streams were dropped along with its rpc stream
        spare.event.send_json(&Chan::Event.msg(2));
        assert_eq!(c.read_event(ms(500)).unwrap().unwrap().generation, 2);
        assert!(event.send(b"x").is_err(), "old event stream still in use");
        assert!(blob.send(b"x").is_err(), "old blob stream still in use");
        spare.blob.send(b"new").unwrap();
        let d = BlobDescriptor::describe(b"new").unwrap();
        assert_eq!(c.read_blob(d).unwrap(), b"new");
    }
}

#[test]
fn a_transport_that_fails_at_startup_fails_new() {
    let transport = MockTransport::sequence(vec![]);
    assert!(oc2_hlapi_v2::codecs::BaseCodec::new(transport).is_err());
}

// --- rpc-only and event-only behaviour ------------------------------------------------------------

#[test]
fn responses_keep_their_fields() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.reply(&json!({"type": "error", "id": 12, "gen": 3, "data": "boom"}));
        let r = c.read_response(ms(500)).unwrap().unwrap();
        assert_eq!((r.id, r.generation), (12, Some(3)));
        assert!(matches!(r.data, ResponseData::Error { ref msg } if msg == "boom"));
    }
}

#[test]
fn json_that_is_not_a_response_is_an_error_not_a_hang() {
    for kind in KINDS {
        for junk in [
            json!([1, 2]),
            json!({"foo": 1}),
            json!({"type": "bogus"}),
            json!(null),
        ] {
            let (mut c, h) = codec(kind);
            h.reply(&junk);
            let r = within(move || {
                let r = c.read_response(Some(Duration::from_secs(5)));
                drop(h);
                r
            });
            assert!(matches!(r, Err(BaseCodecError::Serde(_))), "{junk}: {r:?}");
        }
    }
}

#[test]
fn events_the_crate_cannot_use_are_skipped() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.emit(&json!(null));
        h.emit(&json!([1]));
        h.emit(&json!({"type": "no generation"}));
        h.emit(&Chan::Event.msg(5));
        assert_eq!(c.read_event(ms(500)).unwrap().unwrap().generation, 5);
    }
}

#[test]
fn unknown_events_are_delivered_as_other() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.emit(&json!({"type": "someFutureEvent", "gen": 31}));
        let e = c.read_event(ms(500)).unwrap().unwrap();
        assert_eq!(e.generation, 31);
        assert!(matches!(e.data, EventData::Other(_)));
    }
}

#[test]
fn device_events_keep_their_fields() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.emit(&device_event(4, "tick", id(9), json!({"n": 1})));
        let e = c.read_event(ms(500)).unwrap().unwrap();
        let EventData::Device(d) = e.data else {
            panic!("wrong variant")
        };
        assert_eq!((d.typ.as_str(), d.device_id), ("tick", id(9)));
        assert_eq!(d.data, Some(json!({"n": 1})));
    }
}

#[test]
fn the_event_and_rpc_streams_are_independent() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.emit(&Chan::Event.msg(1));
        assert!(c.read_response(Some(Duration::ZERO)).unwrap().is_none());
        h.reply(&Chan::Rpc.msg(2));
        assert_eq!(c.read_event(ms(500)).unwrap().unwrap().generation, 1);
        assert_eq!(
            c.read_response(ms(500)).unwrap().unwrap().generation,
            Some(2)
        );
    }
}

// --- sending ------------------------------------------------------------------------------------

fn list_request(id: Option<u32>) -> Request {
    Request {
        id,
        data: RequestData::List,
    }
}

fn invoke_with_padding(pad: usize) -> Request {
    Request {
        id: None,
        data: RequestData::Invoke {
            invocation: MethodInvocation {
                device_id: id(1),
                name: "m".into(),
                parameters: Some(vec![json!("p".repeat(pad))]),
            },
            blobs: None,
        },
    }
}

fn encoded_len(r: &Request) -> usize {
    serde_json::to_vec(r).unwrap().len() + 2
}

#[test]
fn a_request_is_sent_as_nul_json_nul() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        c.send_request(list_request(Some(3)), &[]).unwrap();
        let raw = h.rpc.available();
        assert_eq!((raw[0], *raw.last().unwrap()), (0, 0));
        assert_eq!(
            raw.iter().filter(|&&b| b == 0).count(),
            2,
            "NULs only at the ends"
        );
        let v: Value = serde_json::from_slice(&raw[1..raw.len() - 1]).unwrap();
        assert_eq!(v, json!({"type": "list", "id": 3}));
        assert!(h.blob.available().is_empty());
    }
}

#[test]
fn a_request_without_an_id_has_no_id_field() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    c.send_request(list_request(None), &[]).unwrap();
    let v = h.rpc.json(ms(500)).unwrap().unwrap();
    assert_eq!(v, json!({"type": "list"}));
}

#[test]
fn requests_up_to_4096_bytes_including_the_nuls_are_accepted() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    let base = encoded_len(&invoke_with_padding(0));
    c.send_request(invoke_with_padding(4096 - base), &[])
        .unwrap();
    assert_eq!(h.rpc.available().len(), 4096);
}

#[test]
fn requests_over_4096_bytes_are_rejected() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        let base = encoded_len(&invoke_with_padding(0));
        let r = invoke_with_padding(4096 - base + 1);
        assert_eq!(encoded_len(&r), 4097);
        let err = c.send_request(r, &[b"blob data"]).unwrap_err();
        assert!(matches!(err, BaseCodecError::RequestTooBig), "{err:?}");
        // Nothing may have been written: blobs without their request would desync the host
        assert!(h.rpc.available().is_empty());
        assert!(h.blob.available().is_empty());
    }
}

#[test]
fn blobs_are_written_before_the_request_and_the_request_in_one_write() {
    for kind in KINDS {
        let (mut c, h, log) = codec_logged(kind);
        c.send_request(list_request(Some(1)), &[b"one", b"second"])
            .unwrap();
        assert_eq!(*log.lock().unwrap(), ["blob", "blob", "rpc"]);
        assert_eq!(h.blob.available(), b"onesecond");
    }
}

#[test]
fn sending_to_a_dead_connection_fails_and_reconnects() {
    for kind in KINDS {
        let (mut c, mut hosts, opens) = codec_with_spares(kind, 1);
        let spare = hosts.pop().unwrap();
        drop(hosts);
        let err = c.send_request(list_request(Some(1)), &[]).unwrap_err();
        assert!(
            matches!(&err, BaseCodecError::Io(e) if e.kind() == io::ErrorKind::BrokenPipe),
            "{err:?}"
        );
        assert_eq!(opens.load(Ordering::SeqCst), 2);
        c.send_request(list_request(Some(2)), &[]).unwrap();
        assert_eq!(spare.rpc.json(ms(500)).unwrap().unwrap()["id"], 2);
    }
}

// --- blobs --------------------------------------------------------------------------------------

#[test]
fn reads_a_blob() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.blob_out(b"hello");
        let d = BlobDescriptor::describe(b"hello").unwrap();
        assert_eq!(c.read_blob(d).unwrap(), b"hello");
    }
}

#[test]
fn blobs_are_cut_by_length_not_by_content() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    h.blob_out(b"ab\0cdef\0\0");
    let first = BlobDescriptor::describe(b"ab\0c").unwrap();
    let second = BlobDescriptor::describe(b"def\0\0").unwrap();
    assert_eq!(c.read_blob(first).unwrap(), b"ab\0c");
    assert_eq!(c.read_blob(second).unwrap(), b"def\0\0");
}

#[test]
fn an_empty_blob_does_not_wait_for_data() {
    let (mut c, _h) = codec(TimeoutKind::WouldBlock);
    let blob = within(move || c.read_blob(BlobDescriptor::describe(b"").unwrap()));
    assert_eq!(blob.unwrap(), b"");
}

#[test]
fn a_blob_that_arrives_in_pieces_is_assembled() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    let data: Vec<u8> = (0..=255).cycle().take(10_000).collect();
    let d = BlobDescriptor::describe(&data).unwrap();
    let sender = {
        let data = data.clone();
        thread::spawn(move || {
            for chunk in data.chunks(997) {
                h.blob_out(chunk);
                thread::sleep(Duration::from_millis(1));
            }
            h
        })
    };
    assert_eq!(c.read_blob(d).unwrap(), data);
    drop(sender.join().unwrap());
}

#[test]
fn a_corrupt_blob_is_an_error() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    h.blob_out(b"hellp");
    let d = BlobDescriptor::describe(b"hello").unwrap();
    let err = c.read_blob(d).unwrap_err();
    assert!(
        matches!(err, BaseCodecError::BlobChecksumInvalid),
        "{err:?}"
    );
}

#[test]
fn a_blob_cut_short_by_eof_is_an_error() {
    let (mut c, h) = codec(TimeoutKind::WouldBlock);
    h.blob_out(b"hel");
    drop(h);
    let d = BlobDescriptor::describe(b"hello").unwrap();
    assert!(c.read_blob(d).is_err());
}

// --- clearing -----------------------------------------------------------------------------------

#[test]
fn clearing_the_response_stream_drops_stale_replies() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        for n in 0..3 {
            h.reply(&Chan::Rpc.msg(n));
        }
        h.rpc.send(&vec![b'x'; 20_000]).unwrap(); // more than one internal read
        c.clear_response_stream().unwrap();
        assert!(c.read_response(Some(Duration::ZERO)).unwrap().is_none());
        assert!(h.rpc.available().is_empty());
        // and the stream is usable (blocking) afterwards
        let h2 = thread::spawn(move || {
            thread::sleep(Duration::from_millis(40));
            h.reply(&Chan::Rpc.msg(7));
            h
        });
        assert_eq!(
            c.read_response(Some(LONG)).unwrap().unwrap().generation,
            Some(7)
        );
        drop(h2.join().unwrap());
    }
}

#[test]
fn clearing_the_response_stream_also_drops_a_half_read_frame() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        let bytes = framed(&Chan::Rpc.msg(1));
        h.rpc.send(&bytes[..bytes.len() / 2]).unwrap();
        assert!(c.read_response(ms(40)).unwrap().is_none()); // gave up on it
        c.clear_response_stream().unwrap();
        // A reply without a leading NUL would be glued to the leftover and lost
        let mut fresh = serde_json::to_vec(&Chan::Rpc.msg(2)).unwrap();
        fresh.push(0);
        h.rpc.send(&fresh).unwrap();
        assert_eq!(
            c.read_response(ms(500)).unwrap().unwrap().generation,
            Some(2)
        );
    }
}

#[test]
fn clearing_an_empty_stream_does_not_block() {
    for kind in KINDS {
        within(move || {
            let (mut c, _h) = codec(kind);
            c.clear_response_stream().unwrap();
            c.clear_blob_stream().unwrap();
        });
    }
}

#[test]
fn clearing_the_blob_stream_drops_stale_blob_data() {
    for kind in KINDS {
        let (mut c, h) = codec(kind);
        h.blob_out(&vec![7; 10_000]);
        c.clear_blob_stream().unwrap();
        assert!(h.blob.available().is_empty());
        h.blob_out(b"fresh");
        let d = BlobDescriptor::describe(b"fresh").unwrap();
        assert_eq!(c.read_blob(d).unwrap(), b"fresh");
    }
}

#[test]
fn clearing_a_dead_connection_fails_and_reconnects() {
    for kind in KINDS {
        let (mut c, mut hosts, opens) = codec_with_spares(kind, 2);
        let spare = hosts.pop().unwrap();
        drop(hosts.pop()); // second spare, unused
        drop(hosts);
        let err = c.clear_response_stream().unwrap_err();
        assert!(
            is_eof(&err) && err.to_string().contains("response stream"),
            "{err}"
        );
        assert_eq!(opens.load(Ordering::SeqCst), 2);
        let _ = spare;
    }
}

#[test]
fn clearing_a_dead_blob_stream_names_the_blob_stream() {
    for kind in KINDS {
        let (mut c, mut hosts, _) = codec_with_spares(kind, 1);
        let _spare = hosts.pop().unwrap();
        let Host { rpc, blob, event } = hosts.pop().unwrap();
        drop(blob);
        let err = c.clear_blob_stream().unwrap_err();
        assert!(
            is_eof(&err) && err.to_string().contains("blob stream"),
            "{err}"
        );
        drop((rpc, event));
    }
}