polyc-host 2026.9.3

//! Generic dedicated-thread bridge between the tokio control plane and a
//! Commonware-runtime-backed durable store.
//!
//! Built slice by slice against the invariants every host migration
//! (`PersonaHost`, `EventLogHost`) depends on:
//!
//! - **INV-H1**: dropping a [`HostHandle`] never deadlocks — the command
//!   sender is dropped before the dedicated thread is joined.
//! - **INV-H2**: an eager [`Body::open`] failure surfaces as
//!   [`spawn_host`]'s `Err`, before any command can be sent — never as a
//!   silently-empty host.
//! - **INV-H3**: a lazy [`Body::open`] (returns `Ok(())` immediately)
//!   followed by a command that fails inside [`Body::handle`] never crashes
//!   the command loop.
//! - **INV-H4**: a command already queued on the channel when
//!   `shutdown.cancel()` fires is still processed by the post-cancellation
//!   drain, not lost.
//! - **INV-H5**: [`Body::on_drain_complete`] fires strictly after the drain
//!   has run every command it queued.
//! - **INV-H6**: a caller-supplied `(Sender, Receiver)` pair
//!   ([`spawn_host_with_channel`]) drives the exact same ready
//!   handshake/loop/drain/`on_drain_complete`/`Drop` machinery as an
//!   internally-created one ([`spawn_host`]) — the only difference is who
//!   constructs the channel. `EventLogHost`'s 4 shards need this: every
//!   shard's `Sender` must exist and be handed to every OTHER shard's `Body`
//!   before any shard's thread starts (a chicken-and-egg an internally-built
//!   channel can't resolve), so the caller builds all its channels up front
//!   and hands each shard its own `(Sender, Receiver)` pair.

use std::path::PathBuf;
use std::thread::JoinHandle;

use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;

/// Configuration for one dedicated host thread.
#[derive(Debug, Clone)]
pub struct HostOptions {
    /// Name given to the dedicated OS thread (surfaces in a panic backtrace
    /// or `top`/`ps` listing, so it should name the store it hosts).
    pub thread_name: &'static str,
    /// Directory the Commonware runtime roots its storage at.
    pub storage_dir: PathBuf,
    /// Bounded backlog of in-flight commands on the channel from the tokio
    /// side to the dedicated thread.
    pub command_backlog: usize,
}

/// Per-host behavior plugged into the generic dedicated-thread bridge.
#[async_trait::async_trait]
pub trait Body: Send + 'static {
    /// The command type carried from the tokio side to the dedicated thread.
    type Cmd: Send + 'static;

    /// Called once, on the dedicated thread, inside the Commonware runtime,
    /// before the command loop starts (see INV-H2 and INV-H3).
    ///
    /// # Errors
    ///
    /// Returns a human-readable reason the store could not be opened. Only
    /// meaningful for a host that opens eagerly; a lazy host returns
    /// `Ok(())` immediately and never returns `Err` here.
    async fn open(&mut self, ctx: &commonware_runtime::tokio::Context) -> Result<(), String>;

    /// Dispatch one command against this host's state.
    async fn handle(&mut self, ctx: &commonware_runtime::tokio::Context, cmd: Self::Cmd);

    /// Called once, after the post-cancellation drain has run every command
    /// that was already queued when shutdown fired (INV-H5). Defaults to a
    /// no-op; `EventLogHost`'s per-shard "sync every open log" epilogue is
    /// exactly what this hook exists for.
    async fn on_drain_complete(&mut self, ctx: &commonware_runtime::tokio::Context) {
        let _ = ctx;
    }
}

/// Error starting a dedicated host thread.
#[derive(Debug, thiserror::Error)]
pub enum SpawnError {
    /// The dedicated thread panicked before it could report either outcome
    /// of [`Body::open`].
    #[error("dedicated host thread failed before it could report readiness")]
    RuntimeStart,
    /// [`Body::open`] returned an error: the host's store could not be
    /// opened or recovered (INV-H2).
    #[error("host body failed to open: {0}")]
    Open(String),
}

/// The dedicated host has shut down; a [`call`] could not be served.
#[derive(Debug, thiserror::Error)]
#[error("host is shut down")]
pub struct Closed;

/// Send one command carrying a fresh `oneshot` ack, and await the reply.
///
/// This is the plumbing every host's own caller-facing method builds on:
/// `build` wraps the ack half into that host's own `Cmd` variant, `call`
/// sends it and awaits the reply, and a channel that is closed on either
/// leg — the command couldn't be enqueued, or the ack was dropped without a
/// reply — collapses to one [`Closed`] error. What a caller does with that
/// error (fail open, fail closed, propagate it) stays entirely up to the
/// caller; this helper only ever reports whether the round trip happened.
///
/// # Errors
///
/// Returns [`Closed`] if the dedicated thread has shut down: either the
/// command could not be enqueued, or the ack was dropped without a reply.
pub async fn call<Cmd, T>(
    tx: &mpsc::Sender<Cmd>,
    build: impl FnOnce(oneshot::Sender<T>) -> Cmd,
) -> Result<T, Closed> {
    let (ack, ack_rx) = oneshot::channel();
    tx.send(build(ack)).await.map_err(|_| Closed)?;
    ack_rx.await.map_err(|_| Closed)
}

/// Handle to a durable host running on its dedicated Commonware-runtime
/// thread.
#[derive(Debug)]
pub struct HostHandle<Cmd> {
    /// Outbound command channel; `Option` so `Drop` can close it before
    /// joining (INV-H1).
    tx: Option<mpsc::Sender<Cmd>>,
    /// The dedicated OS thread, joined on drop.
    thread: Option<JoinHandle<()>>,
}

impl<Cmd> HostHandle<Cmd> {
    /// The outbound command channel, if the host has not already begun
    /// shutting down. Feed this to [`call`] to build a caller-facing method
    /// on top.
    #[must_use]
    pub const fn sender(&self) -> Option<&mpsc::Sender<Cmd>> {
        self.tx.as_ref()
    }
}

impl<Cmd> Drop for HostHandle<Cmd> {
    fn drop(&mut self) {
        // INV-H1: drop the sender first. This closes the command channel,
        // which ends the dedicated thread's command loop (`rx.recv()`
        // resolves to `None`) and lets `Runner::start` return — only THEN
        // join. Joining first would deadlock: the dedicated thread would
        // still be blocked waiting on this very channel to close, and it
        // never will while the sender we are about to block on joining is
        // still alive.
        drop(self.tx.take());
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

/// Spawn a dedicated Commonware-runtime thread hosting `body`, rooted at
/// `opts.storage_dir`, and returns once the thread has reported readiness.
///
/// # Errors
///
/// Returns [`SpawnError::Open`] if [`Body::open`] reports a failure (an
/// eager host's store could not be opened or recovered — INV-H2), or
/// [`SpawnError::RuntimeStart`] if the dedicated thread panics before it can
/// report either outcome.
///
/// # Panics
///
/// Panics if the OS refuses to spawn the dedicated thread (resource
/// exhaustion).
pub fn spawn_host<B: Body>(
    opts: HostOptions,
    shutdown: CancellationToken,
    body: B,
) -> Result<HostHandle<B::Cmd>, SpawnError> {
    let (tx, rx) = mpsc::channel::<B::Cmd>(opts.command_backlog);
    spawn_host_with_channel(opts, shutdown, body, tx, rx)
}

/// Spawn a dedicated Commonware-runtime thread hosting `body`, using a
/// caller-supplied `(Sender, Receiver)` pair instead of one
/// [`spawn_host`] creates internally (INV-H6).
///
/// This is the entry point a multi-shard host needs: every shard's `Sender`
/// must exist, and be cloned into every OTHER shard's `Body`, before any
/// shard's dedicated thread starts — a chicken-and-egg [`spawn_host`]'s
/// internally-created channel cannot resolve on its own. The caller builds
/// every channel up front (e.g. one `mpsc::channel` per shard), clones each
/// `Sender` into whichever peer `Body`s need it, then calls this once per
/// shard with that shard's own pair — `tx` is consumed into the returned
/// [`HostHandle`], exactly as it would be if [`spawn_host`] had built it.
///
/// `opts.command_backlog` is ignored here (the channel already exists);
/// everything else — the ready handshake, the command loop, the post-cancel
/// drain, [`Body::on_drain_complete`], and the returned [`HostHandle`]'s
/// drop-tx-then-join `Drop` (INV-H1) — is identical to [`spawn_host`].
///
/// # Errors
///
/// Returns [`SpawnError::Open`] if [`Body::open`] reports a failure (INV-H2),
/// or [`SpawnError::RuntimeStart`] if the dedicated thread panics before it
/// can report either outcome.
///
/// # Panics
///
/// Panics if the OS refuses to spawn the dedicated thread (resource
/// exhaustion).
pub fn spawn_host_with_channel<B: Body>(
    opts: HostOptions,
    shutdown: CancellationToken,
    body: B,
    tx: mpsc::Sender<B::Cmd>,
    rx: mpsc::Receiver<B::Cmd>,
) -> Result<HostHandle<B::Cmd>, SpawnError> {
    let HostOptions {
        thread_name,
        storage_dir,
        command_backlog: _,
    } = opts;
    let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<(), String>>();

    let thread = std::thread::Builder::new()
        .name(thread_name.to_owned())
        .spawn(move || run_dedicated(storage_dir, rx, &ready_tx, &shutdown, body))
        .expect("spawn dedicated host thread");

    match ready_rx.recv() {
        Ok(Ok(())) => Ok(HostHandle {
            tx: Some(tx),
            thread: Some(thread),
        }),
        Ok(Err(reason)) => {
            let _ = thread.join();
            Err(SpawnError::Open(reason))
        }
        Err(_) => {
            // The thread panicked before reporting either outcome.
            let _ = thread.join();
            Err(SpawnError::RuntimeStart)
        }
    }
}

/// Body of the dedicated thread: owns the Commonware tokio runtime, opens
/// `body`'s store per [`Body::open`]'s contract, then serves commands until
/// either the channel closes or `shutdown` cancels (checked first, biased —
/// so a `shutdown` racing a still-buffered command always takes the
/// shutdown branch, leaving that command for the drain below, not lost).
/// Once the loop ends, drains whatever was already queued (INV-H4) and
/// reports completion via [`Body::on_drain_complete`].
fn run_dedicated<B: Body>(
    storage_dir: PathBuf,
    rx: mpsc::Receiver<B::Cmd>,
    ready_tx: &std::sync::mpsc::Sender<Result<(), String>>,
    shutdown: &CancellationToken,
    mut body: B,
) {
    use commonware_runtime::Runner as _;

    let cfg = commonware_runtime::tokio::Config::default().with_storage_directory(storage_dir);
    let runner = commonware_runtime::tokio::Runner::new(cfg);

    runner.start(|context| async move {
        if let Err(reason) = body.open(&context).await {
            tracing::warn!(error = %reason, "dedicated host thread's body failed to open");
            let _ = ready_tx.send(Err(reason));
            return;
        }
        let _ = ready_tx.send(Ok(()));

        let mut rx = rx;
        loop {
            // `commonware_macros::select!` is always biased (a verbatim
            // rename of `tokio::select! { biased; ... }`), so `shutdown` is
            // always checked first: a shutdown racing an already-buffered
            // command always takes this branch, leaving that command for
            // the drain below rather than losing the race unpredictably.
            let cmd = commonware_macros::select! {
                () = shutdown.cancelled() => break,
                maybe = rx.recv() => match maybe {
                    Some(cmd) => cmd,
                    None => break,
                },
            };
            body.handle(&context, cmd).await;
        }

        // INV-H4: drain whatever was already queued when the loop above
        // ended, so a command that lost the race against cancellation (or
        // was buffered behind the channel closing) is not silently dropped.
        let mut drained = 0usize;
        while let Ok(cmd) = rx.try_recv() {
            body.handle(&context, cmd).await;
            drained += 1;
        }
        if drained > 0 {
            tracing::debug!(drained, "ran queued commands after shutdown");
        }
        // INV-H5: only after every drained command has actually run.
        body.on_drain_complete(&context).await;
    });
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]
    use super::*;
    use tokio_util::sync::CancellationToken;

    /// A scratch storage directory unique to one test. The Commonware
    /// runtime creates the directory itself, so tests never do — only clean
    /// up any stale directory a prior run of the same test may have left.
    fn scratch_dir(label: &str) -> std::path::PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "polyc-host-{label}-{}-{}",
            std::process::id(),
            uuid::Uuid::new_v4()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        dir
    }

    /// [`HostOptions`] for one test, with a fresh scratch directory.
    fn opts(label: &str, command_backlog: usize) -> HostOptions {
        // Leak the name into a `'static str`: fine in a test, which runs
        // once and the process exits shortly after.
        let thread_name: &'static str =
            Box::leak(format!("polyc-host-test-{label}").into_boxed_str());
        HostOptions {
            thread_name,
            storage_dir: scratch_dir(label),
            command_backlog,
        }
    }

    /// A [`Body`] whose `open` always fails.
    struct EagerFailBody;

    #[async_trait::async_trait]
    impl Body for EagerFailBody {
        type Cmd = ();

        async fn open(&mut self, _ctx: &commonware_runtime::tokio::Context) -> Result<(), String> {
            Err("the store is unopenable".to_owned())
        }

        async fn handle(&mut self, _ctx: &commonware_runtime::tokio::Context, (): Self::Cmd) {}
    }

    /// INV-H2: an eager `Body::open` failure surfaces as `spawn_host`'s
    /// `Err`, before any command can be sent — never as a silently-empty
    /// host.
    #[test]
    fn an_eager_open_failure_surfaces_through_spawn_host() {
        let opts = HostOptions {
            thread_name: "polyc-host-test-eager-fail",
            storage_dir: scratch_dir("eager-fail"),
            command_backlog: 8,
        };
        let err = spawn_host(opts, CancellationToken::new(), EagerFailBody)
            .expect_err("an eager Body::open failure must fail spawn_host");
        match err {
            SpawnError::Open(reason) => assert_eq!(reason, "the store is unopenable"),
            SpawnError::RuntimeStart => panic!("expected Open, got RuntimeStart"),
        }
    }

    /// Command variants a lazily-opening test body understands.
    enum LazyCmd {
        /// Acks `id` back on `ack`.
        Ping {
            id: u32,
            ack: tokio::sync::oneshot::Sender<u32>,
        },
        /// Always acks a simulated internal failure — never panics.
        Fail {
            ack: tokio::sync::oneshot::Sender<Result<u32, String>>,
        },
    }

    /// A [`Body`] that opens lazily (returns `Ok(())` immediately), answers
    /// every `Ping`, and answers `Fail` with a simulated internal error
    /// rather than panicking.
    struct LazyBody;

    #[async_trait::async_trait]
    impl Body for LazyBody {
        type Cmd = LazyCmd;

        async fn open(&mut self, _ctx: &commonware_runtime::tokio::Context) -> Result<(), String> {
            Ok(())
        }

        async fn handle(&mut self, _ctx: &commonware_runtime::tokio::Context, cmd: Self::Cmd) {
            match cmd {
                LazyCmd::Ping { id, ack } => {
                    let _ = ack.send(id);
                }
                LazyCmd::Fail { ack } => {
                    let _ = ack.send(Err("simulated per-command failure".to_owned()));
                }
            }
        }
    }

    /// INV-H1: dropping a [`HostHandle`] never deadlocks — the command
    /// sender is dropped before the dedicated thread is joined. Proven by
    /// actually completing a round trip through the loop first, then
    /// dropping the handle: a join-before-drop bug would hang this test
    /// forever instead of returning.
    #[tokio::test]
    async fn dropping_the_handle_never_deadlocks() {
        let handle = spawn_host(opts("drop-order", 8), CancellationToken::new(), LazyBody)
            .expect("lazy open always succeeds");

        let (ack, ack_rx) = tokio::sync::oneshot::channel();
        handle
            .sender()
            .expect("handle is fresh")
            .send(LazyCmd::Ping { id: 42, ack })
            .await
            .expect("channel is open");
        assert_eq!(ack_rx.await.expect("the loop answers"), 42);

        // Dropping the ONLY sender (the handle's own) must close the
        // channel and let the dedicated thread's loop end — if `Drop`
        // joined before dropping it, this would hang forever.
        drop(handle);
    }

    /// INV-H3: a lazy `Body::open` followed by a command that fails inside
    /// `Body::handle` never crashes the command loop — a later command must
    /// still be served.
    #[tokio::test]
    async fn a_command_that_fails_inside_handle_does_not_crash_the_loop() {
        let handle = spawn_host(
            opts("handle-failure", 8),
            CancellationToken::new(),
            LazyBody,
        )
        .expect("lazy open always succeeds");

        let (fail_ack, fail_ack_rx) = tokio::sync::oneshot::channel();
        handle
            .sender()
            .expect("handle is fresh")
            .send(LazyCmd::Fail { ack: fail_ack })
            .await
            .expect("channel is open");
        assert_eq!(
            fail_ack_rx.await.expect("the loop still answers"),
            Err("simulated per-command failure".to_owned())
        );

        // The loop must still be alive and answering commands after the
        // internal failure above.
        let (ack, ack_rx) = tokio::sync::oneshot::channel();
        handle
            .sender()
            .expect("handle is fresh")
            .send(LazyCmd::Ping { id: 7, ack })
            .await
            .expect("channel is open");
        assert_eq!(ack_rx.await.expect("the loop answers"), 7);

        drop(handle);
    }

    /// INV-H4: a command already queued on the channel when
    /// `shutdown.cancel()` fires is still processed by the post-cancellation
    /// drain, not lost. Proven with an extra live sender clone that is never
    /// dropped during the test — the ONLY way the dedicated thread's loop
    /// can end is by observing `shutdown`, never by the channel closing —
    /// so this also proves the loop actually reacts to `shutdown` at all.
    #[tokio::test]
    async fn a_command_queued_before_shutdown_cancel_fires_is_still_processed() {
        let shutdown = CancellationToken::new();
        let handle = spawn_host(opts("post-cancel-drain", 8), shutdown.clone(), LazyBody)
            .expect("lazy open always succeeds");
        let extra_tx = handle.sender().expect("handle is fresh").clone();

        let (ack, ack_rx) = tokio::sync::oneshot::channel();
        extra_tx
            .send(LazyCmd::Ping { id: 9, ack })
            .await
            .expect("channel is open");

        shutdown.cancel();

        // The queued Ping, sent before cancellation, must still run.
        assert_eq!(ack_rx.await.expect("a queued command must still run"), 9);

        // The dedicated thread must exit on `shutdown` alone: `extra_tx` is
        // a live sender clone that is never dropped, so the channel itself
        // never closes. If the loop only ever ended when the channel
        // closed, this would hang forever.
        drop(handle);
        drop(extra_tx);
    }

    /// A [`Body`] that lets the test hold the loop mid-command via `gate`,
    /// so it can force a deterministic race: queue commands, cancel
    /// shutdown while the loop is still busy, then release and observe the
    /// order everything ran in.
    struct GatedBody {
        gate: std::sync::Arc<tokio::sync::Notify>,
        order: std::sync::Arc<std::sync::Mutex<Vec<&'static str>>>,
    }

    enum GatedCmd {
        /// Blocks on `gate` before acking — holds the loop mid-command.
        Slow {
            ack: tokio::sync::oneshot::Sender<()>,
        },
        /// Acks `id` immediately.
        Ping {
            id: u32,
            ack: tokio::sync::oneshot::Sender<u32>,
        },
    }

    #[async_trait::async_trait]
    impl Body for GatedBody {
        type Cmd = GatedCmd;

        async fn open(&mut self, _ctx: &commonware_runtime::tokio::Context) -> Result<(), String> {
            Ok(())
        }

        async fn handle(&mut self, _ctx: &commonware_runtime::tokio::Context, cmd: Self::Cmd) {
            match cmd {
                GatedCmd::Slow { ack } => {
                    self.gate.notified().await;
                    self.order.lock().unwrap().push("slow");
                    let _ = ack.send(());
                }
                GatedCmd::Ping { id, ack } => {
                    self.order.lock().unwrap().push("ping");
                    let _ = ack.send(id);
                }
            }
        }

        async fn on_drain_complete(&mut self, _ctx: &commonware_runtime::tokio::Context) {
            self.order.lock().unwrap().push("drained");
        }
    }

    /// INV-H5: `Body::on_drain_complete` fires strictly after the drain has
    /// run every command it queued.
    #[tokio::test]
    async fn on_drain_complete_fires_after_the_drain_has_run_every_command() {
        let gate = std::sync::Arc::new(tokio::sync::Notify::new());
        let order = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let shutdown = CancellationToken::new();

        let body = GatedBody {
            gate: gate.clone(),
            order: order.clone(),
        };
        let handle = spawn_host(opts("drain-complete", 8), shutdown.clone(), body)
            .expect("lazy open always succeeds");

        // Start a Slow command: the loop is now stuck inside `handle()`,
        // awaiting `gate`, so it cannot poll `rx.recv()` again until
        // released.
        let (slow_ack, slow_ack_rx) = tokio::sync::oneshot::channel();
        handle
            .sender()
            .expect("handle is fresh")
            .send(GatedCmd::Slow { ack: slow_ack })
            .await
            .expect("channel is open");

        // Queue two Pings behind it — they sit in the mpsc buffer.
        let (ack_a, ack_a_rx) = tokio::sync::oneshot::channel();
        let (ack_b, ack_b_rx) = tokio::sync::oneshot::channel();
        handle
            .sender()
            .expect("handle is fresh")
            .send(GatedCmd::Ping { id: 1, ack: ack_a })
            .await
            .expect("channel is open");
        handle
            .sender()
            .expect("handle is fresh")
            .send(GatedCmd::Ping { id: 2, ack: ack_b })
            .await
            .expect("channel is open");

        // Cancel while the loop is still stuck on Slow: the next iteration
        // sees `shutdown` already resolved (checked first, biased) and
        // breaks, WITHOUT ever taking a Ping off the select arm — leaving
        // both for the drain.
        shutdown.cancel();

        // Release Slow. It completes; the loop's next iteration then breaks
        // on `shutdown`; the drain runs both queued Pings.
        gate.notify_one();
        slow_ack_rx.await.expect("slow command completes");
        ack_a_rx.await.expect("drained");
        ack_b_rx.await.expect("drained");

        // Drop the handle: joins the dedicated thread, which only returns
        // once the drain loop AND `on_drain_complete` have both run.
        drop(handle);

        let seen = order.lock().unwrap().clone();
        assert_eq!(
            seen,
            vec!["slow", "ping", "ping", "drained"],
            "on_drain_complete must fire strictly after every drained command"
        );
    }

    /// `call` is the plumbing every host's own caller-facing method builds
    /// on: it must complete a normal round trip, and collapse a closed
    /// channel (the host has shut down) to `Closed` rather than hanging or
    /// panicking.
    #[tokio::test]
    async fn call_round_trips_and_reports_closed_once_the_host_is_gone() {
        let shutdown = CancellationToken::new();
        let handle = spawn_host(opts("call-helper", 8), shutdown.clone(), LazyBody)
            .expect("lazy open always succeeds");
        // A clone kept alive on purpose: it lets us still try a `call`
        // after the host is gone without racing the channel's own closure.
        let tx = handle.sender().expect("handle is fresh").clone();

        let pong = call(&tx, |ack| LazyCmd::Ping { id: 5, ack })
            .await
            .expect("the channel is open");
        assert_eq!(pong, 5);

        // Shut the host down via cancellation, not by dropping every
        // sender (this clone stays alive throughout). `drop(handle)` joins
        // the dedicated thread, which only returns once its loop has
        // actually ended — so by the time `drop` returns, `rx` is
        // guaranteed gone; no race with the send below.
        shutdown.cancel();
        drop(handle);

        let closed = call(&tx, |ack| LazyCmd::Ping { id: 6, ack }).await;
        assert!(matches!(closed, Err(Closed)));
    }

    /// INV-H6: a caller-supplied `(Sender, Receiver)` pair drives the exact
    /// same machinery an internally-created one does — the multi-shard
    /// shape `EventLogHost` needs, where the caller must hold its own extra
    /// `Sender` clone (handed to peer shards) before the dedicated thread
    /// ever starts. Proven by building the channel here, keeping a peer
    /// clone alive the whole time, and running the exact same round-trip +
    /// drop-tx-then-join checks the internally-created-channel tests above
    /// run against `spawn_host`.
    #[tokio::test]
    async fn a_caller_supplied_channel_works_identically_to_an_internal_one() {
        let (tx, rx) = tokio::sync::mpsc::channel::<LazyCmd>(8);
        // Stands in for a peer shard's clone of this shard's sender — held
        // alive independently of the `HostHandle` below, exactly like
        // `EventLogHost`'s `peer_senders`.
        let peer_clone = tx.clone();

        let handle = spawn_host_with_channel(
            opts("with-channel", 8),
            CancellationToken::new(),
            LazyBody,
            tx,
            rx,
        )
        .expect("lazy open always succeeds");

        let (ack, ack_rx) = tokio::sync::oneshot::channel();
        peer_clone
            .send(LazyCmd::Ping { id: 11, ack })
            .await
            .expect("channel is open");
        assert_eq!(ack_rx.await.expect("the loop answers"), 11);

        // Drop the peer clone FIRST: while it's alive the channel has two
        // live senders (it and the handle's own), so the dedicated thread's
        // `rx.recv()` cannot resolve to `None` yet — dropping the handle
        // while `peer_clone` is still around would make `HostHandle::drop`'s
        // `thread.join()` block forever, exactly the multi-sender hazard
        // `EventLogHost` solves with `host_shutdown` cancellation instead of
        // relying on channel closure. Once `peer_clone` is gone, the
        // handle's own sender is the last one, so dropping it (INV-H1) closes
        // the channel and the thread exits cleanly.
        drop(peer_clone);
        drop(handle);
    }
}