h3x 0.6.1

//! Reusable error-latching utilities for RPC/IPC.
//!
//! These types enforce the QUIC connection-error consistency requirement:
//! once a connection error occurs, **every** subsequent operation on the
//! same connection must observe the **same** error.
//!
//! # Design rules
//!
//! - **No eager `latch(e)` on the public surface.** Callers must always use
//!   `latch_with(|| ...)` / `guard*_with(...)` so that error construction is
//!   lazy and skipped entirely when an error is already latched.
//! - **No bare `check()?` on operation paths.** Use one of the `guard*`
//!   helpers on [`LifecycleExt`] so the check / execute / latch sequence
//!   stays in one place.
//! - **The latch is an implementation detail.** Application code never
//!   touches [`ConnectionErrorLatch`] directly — it interacts through
//!   [`LifecycleExt`] on the owning container type.
//!
//! # Design
//!
//! Container types (e.g. `RemoteConnection`, `IpcLifecycle`, session types)
//! own a [`ConnectionErrorLatch`] field and expose it to the crate through
//! the sealed [`HasLatch`] trait. Implementing `HasLatch` plus
//! [`quic::Lifecycle`] automatically gives the container all of
//! [`LifecycleExt`]'s guard/check/closed helpers via blanket impl.
//!
//! # Latch-aware lifecycle invariant
//!
//! A type that implements [`HasLatch`] **must** make its [`quic::Lifecycle`]
//! implementation latch-aware. In practice, `check()` should consult the latch
//! first (usually through [`LifecycleExt::check_with_probe`]), and `closed()`
//! should return or install the canonical latched error (usually through
//! [`LifecycleExt::resolve_closed`]).
//!
//! Implementing `HasLatch` but writing `check()` / `closed()` as if the latch
//! did not exist is an implementation bug: operation guards may have already
//! recorded a terminal error, and direct lifecycle callers must observe the
//! same first-wins error.
//!
//! `close(code, reason)` is intentionally not a latch propagation channel. It
//! only initiates local close; the terminal error is recorded by guarded
//! operation failures, liveness probes, or `closed()` resolution.
//!
//! The container authors its own `impl Lifecycle` using
//! [`check_with_probe`](LifecycleExt::check_with_probe) and
//! [`resolve_closed`](LifecycleExt::resolve_closed) as building blocks;
//! operation paths use [`guard`](LifecycleExt::guard) /
//! [`guard_with`](LifecycleExt::guard_with) /
//! [`guard_sync`](LifecycleExt::guard_sync) /
//! [`guard_sync_with`](LifecycleExt::guard_sync_with) exclusively.

use std::future::Future;

use crate::{
    quic::{self, ConnectionError},
    util::set_once::SetOnce,
};

// ---------------------------------------------------------------------------
// ConnectionErrorLatch
// ---------------------------------------------------------------------------

/// First-wins connection error latch.
///
/// Cloning yields a handle to the **same** underlying state — the latch is
/// `Arc`-backed via [`SetOnce`].
///
/// This type is a crate-internal primitive. Application code interacts with
/// it only through [`LifecycleExt`] on the owning container.
#[derive(Debug, Clone, Default)]
pub struct ConnectionErrorLatch {
    terminal_error: SetOnce<ConnectionError>,
}

impl ConnectionErrorLatch {
    pub fn new() -> Self {
        Self::default()
    }

    /// Lazy latch: only calls `f` when no error is latched at entry.
    ///
    /// If another setter wins the race after `f` constructs an error, returns
    /// the already-latched canonical (first-wins) error instead of the losing
    /// candidate.
    pub(crate) fn latch_with(&self, f: impl FnOnce() -> ConnectionError) -> ConnectionError {
        match self.terminal_error.peek() {
            Some(existing) => existing,
            None => {
                let error = f();
                match self.terminal_error.set(error.clone()) {
                    Ok(()) => error,
                    Err(rejected) => self.terminal_error.peek().unwrap_or(rejected),
                }
            }
        }
    }

    /// Return the latched error, if any.
    pub(crate) fn check(&self) -> Result<(), ConnectionError> {
        match self.terminal_error.peek() {
            Some(error) => Err(error),
            None => Ok(()),
        }
    }

    /// Peek at the latched error without consuming it.
    pub(crate) fn peek(&self) -> Option<ConnectionError> {
        self.terminal_error.peek()
    }
}

// ---------------------------------------------------------------------------
// Sealed HasLatch — bridge between container types and LifecycleExt
// ---------------------------------------------------------------------------

pub(crate) mod sealed {
    use super::ConnectionErrorLatch;

    /// Crate-internal trait that lets [`super::LifecycleExt`] reach into a
    /// container's latch without making the latch a public API.
    ///
    /// Implementers must return a stable reference to the same
    /// [`ConnectionErrorLatch`] on every call, and their [`crate::quic::Lifecycle`]
    /// implementation must be latch-aware. See the module-level
    /// "Latch-aware lifecycle invariant" section.
    pub trait HasLatch {
        fn latch(&self) -> &ConnectionErrorLatch;
    }
}

pub(crate) use sealed::HasLatch;

// ---------------------------------------------------------------------------
// LifecycleExt — the only public surface for guard / check / closed
// ---------------------------------------------------------------------------

/// Extension trait layered on top of [`quic::Lifecycle`] that enforces the
/// first-wins error-latching discipline for all operation paths.
///
/// This trait is **sealed**: it is automatically implemented for any type
/// that implements both [`quic::Lifecycle`] and the crate-private
/// [`HasLatch`] marker.
///
/// Implementers typically do two things:
///
/// 1. Own a `latch: ConnectionErrorLatch` field and implement [`HasLatch`].
/// 2. Author their own `impl quic::Lifecycle`, using
///    [`check_with_probe`](Self::check_with_probe) inside `check` and
///    [`resolve_closed`](Self::resolve_closed) inside `closed`.
///
/// Callers use [`guard`](Self::guard) /
/// [`guard_with`](Self::guard_with) /
/// [`guard_sync`](Self::guard_sync) /
/// [`guard_sync_with`](Self::guard_sync_with) for operation paths. No other
/// mechanism for installing errors is exposed.
#[allow(async_fn_in_trait)]
pub trait LifecycleExt: quic::Lifecycle + HasLatch {
    /// Standard implementation of [`quic::Lifecycle::check`].
    ///
    /// Consults the latch first; if clean, invokes `probe` for any
    /// liveness signal the container wants to expose (e.g. a remoc channel
    /// being closed, or a parent lifecycle being dead). A `Some(error)`
    /// result is folded into the latch via `latch_with` so every subsequent
    /// observer sees the same canonical error.
    fn check_with_probe(
        &self,
        probe: impl FnOnce() -> Option<ConnectionError>,
    ) -> Result<(), ConnectionError> {
        self.latch().check()?;
        match probe() {
            None => Ok(()),
            Some(error) => Err(self.latch().latch_with(|| error)),
        }
    }

    /// Standard implementation of [`quic::Lifecycle::closed`].
    ///
    /// Returns the latched error immediately if one exists; otherwise awaits
    /// `wait` (the container-specific terminal-error future) and latches
    /// whatever it produces.
    async fn resolve_closed(&self, wait: impl Future<Output = ConnectionError>) -> ConnectionError {
        if let Some(error) = self.latch().peek() {
            return error;
        }
        let error = wait.await;
        self.latch().latch_with(|| error)
    }

    /// Guard an async operation whose error is already a [`ConnectionError`].
    ///
    /// Checks liveness via [`quic::Lifecycle::check`] first; on success runs
    /// `fut` and lazily latches any error produced.
    async fn guard<T>(
        &self,
        fut: impl Future<Output = Result<T, ConnectionError>>,
    ) -> Result<T, ConnectionError> {
        quic::Lifecycle::check(self)?;
        match fut.await {
            Ok(v) => Ok(v),
            Err(e) => Err(self.latch().latch_with(|| e)),
        }
    }

    /// Guard an async operation whose error must be lazily converted to a
    /// [`ConnectionError`].
    ///
    /// The conversion closure `map_err` is **only** invoked if the operation
    /// errors **and** no error has been latched yet.
    async fn guard_with<T, E, M>(
        &self,
        fut: impl Future<Output = Result<T, E>>,
        map_err: M,
    ) -> Result<T, ConnectionError>
    where
        M: FnOnce(E) -> ConnectionError,
    {
        quic::Lifecycle::check(self)?;
        match fut.await {
            Ok(v) => Ok(v),
            Err(e) => Err(self.latch().latch_with(|| map_err(e))),
        }
    }

    /// Guard a synchronous fallible closure.
    ///
    /// Useful for code paths where both the pre-check and the post-op
    /// error-latching would otherwise be written by hand (e.g. FD unwrapping
    /// after an async stream-open call).
    fn guard_sync<T>(
        &self,
        f: impl FnOnce() -> Result<T, ConnectionError>,
    ) -> Result<T, ConnectionError> {
        quic::Lifecycle::check(self)?;
        match f() {
            Ok(v) => Ok(v),
            Err(e) => Err(self.latch().latch_with(|| e)),
        }
    }

    /// Like [`guard_sync`](Self::guard_sync) but with a lazy error converter.
    fn guard_sync_with<T, E, M>(
        &self,
        f: impl FnOnce() -> Result<T, E>,
        map_err: M,
    ) -> Result<T, ConnectionError>
    where
        M: FnOnce(E) -> ConnectionError,
    {
        quic::Lifecycle::check(self)?;
        match f() {
            Ok(v) => Ok(v),
            Err(e) => Err(self.latch().latch_with(|| map_err(e))),
        }
    }
}

impl<T: quic::Lifecycle + HasLatch + ?Sized> LifecycleExt for T {}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::{
        borrow::Cow,
        sync::{Arc, Mutex, mpsc},
        thread,
        time::Duration,
    };

    use super::*;
    use crate::{error::Code, varint::VarInt};

    fn make_err(tag: u32) -> ConnectionError {
        ConnectionError::Transport {
            source: quic::TransportError {
                kind: VarInt::from_u32(tag),
                frame_type: VarInt::from_u32(0),
                reason: format!("test-{tag}").into(),
            },
        }
    }

    fn make_app_err(tag: u32) -> ConnectionError {
        ConnectionError::Application {
            source: quic::ApplicationError {
                code: Code::new(VarInt::from_u32(tag)),
                reason: format!("test-app-{tag}").into(),
            },
        }
    }

    fn error_kind(error: &ConnectionError) -> VarInt {
        match error {
            ConnectionError::Transport { source } => source.kind,
            _ => panic!("unexpected error shape"),
        }
    }

    fn ok_unit() -> Result<(), ConnectionError> {
        Ok(())
    }

    /// Minimal container used to exercise the sealed trait.
    struct TestLifecycle {
        latch: ConnectionErrorLatch,
        probe: Mutex<Option<ConnectionError>>,
        wait: Mutex<Option<ConnectionError>>,
    }

    impl TestLifecycle {
        fn new() -> Self {
            Self {
                latch: ConnectionErrorLatch::new(),
                probe: Mutex::new(None),
                wait: Mutex::new(None),
            }
        }

        fn set_probe(&self, error: ConnectionError) {
            *self.probe.lock().unwrap() = Some(error);
        }

        fn set_wait(&self, error: ConnectionError) {
            *self.wait.lock().unwrap() = Some(error);
        }
    }

    impl HasLatch for TestLifecycle {
        fn latch(&self) -> &ConnectionErrorLatch {
            &self.latch
        }
    }

    impl quic::Lifecycle for TestLifecycle {
        fn close(&self, _code: Code, _reason: Cow<'static, str>) {}

        fn check(&self) -> Result<(), ConnectionError> {
            self.check_with_probe(|| self.probe.lock().unwrap().take())
        }

        async fn closed(&self) -> ConnectionError {
            self.resolve_closed(async {
                self.wait
                    .lock()
                    .unwrap()
                    .take()
                    .unwrap_or_else(|| make_err(99))
            })
            .await
        }
    }

    #[test]
    fn latch_with_is_lazy_when_already_latched() {
        let latch = ConnectionErrorLatch::new();
        let first = latch.latch_with(|| make_err(1));
        assert!(matches!(
            &first,
            ConnectionError::Transport { source } if source.kind == VarInt::from_u32(1)
        ));

        let called = Mutex::new(false);
        let again = latch.latch_with(|| {
            *called.lock().unwrap() = true;
            make_err(2)
        });
        assert!(
            !*called.lock().unwrap(),
            "closure must not be invoked once latched"
        );
        assert!(matches!(
            &again,
            ConnectionError::Transport { source } if source.kind == VarInt::from_u32(1)
        ));
    }

    #[test]
    fn latch_with_returns_canonical_error_when_setter_races() {
        let latch = ConnectionErrorLatch::new();
        let slow_latch = latch.clone();
        let fast_latch = latch.clone();
        let (entered_tx, entered_rx) = mpsc::channel();

        let slow = thread::spawn(move || {
            slow_latch.latch_with(|| {
                entered_tx.send(()).unwrap();
                thread::sleep(Duration::from_millis(100));
                make_err(1)
            })
        });

        entered_rx.recv().unwrap();

        let fast = thread::spawn(move || fast_latch.latch_with(|| make_err(2)));

        let slow_error = slow.join().unwrap();
        let fast_error = fast.join().unwrap();
        let canonical = latch.check().unwrap_err();

        assert_eq!(error_kind(&slow_error), error_kind(&canonical));
        assert_eq!(error_kind(&fast_error), error_kind(&canonical));
    }

    #[test]
    fn cloned_latch_handles_share_canonical_error() {
        let latch = ConnectionErrorLatch::new();
        let clone = latch.clone();

        assert!(latch.check().is_ok());
        assert!(clone.check().is_ok());

        let installed = clone.latch_with(|| make_err(11));
        let observed = latch.check().unwrap_err();

        assert_eq!(error_kind(&installed), VarInt::from_u32(11));
        assert_eq!(error_kind(&observed), VarInt::from_u32(11));
    }

    #[test]
    fn close_does_not_install_terminal_error() {
        let lc = TestLifecycle::new();

        quic::Lifecycle::close(
            &lc,
            Code::new(VarInt::from_u32(19)),
            Cow::Borrowed("local close"),
        );

        assert!(lc.latch.check().is_ok());
        assert!(quic::Lifecycle::check(&lc).is_ok());
    }

    #[test]
    fn check_with_probe_folds_probe_into_latch() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(42));
        let e1 = quic::Lifecycle::check(&lc).unwrap_err();
        // probe now consumed — but latch remembers the error.
        let e2 = quic::Lifecycle::check(&lc).unwrap_err();
        match (&e1, &e2) {
            (
                ConnectionError::Transport { source: s1 },
                ConnectionError::Transport { source: s2 },
            ) => {
                assert_eq!(s1.kind, VarInt::from_u32(42));
                assert_eq!(s2.kind, VarInt::from_u32(42));
            }
            _ => panic!("unexpected error shape"),
        }
    }

    #[test]
    fn check_with_probe_no_error_when_clean() {
        let lc = TestLifecycle::new();
        assert!(quic::Lifecycle::check(&lc).is_ok());
    }

    #[test]
    fn check_with_probe_skips_probe_when_latched() {
        let lc = TestLifecycle::new();
        lc.latch.latch_with(|| make_err(12));

        let called = Mutex::new(false);
        let err = lc
            .check_with_probe(|| {
                *called.lock().unwrap() = true;
                Some(make_err(13))
            })
            .unwrap_err();

        assert_eq!(error_kind(&err), VarInt::from_u32(12));
        assert!(
            !*called.lock().unwrap(),
            "probe must not run after a terminal error is latched"
        );
    }

    #[test]
    fn check_with_probe_preserves_application_error_shape() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_app_err(25));

        let e1 = quic::Lifecycle::check(&lc).unwrap_err();
        let e2 = quic::Lifecycle::check(&lc).unwrap_err();

        match (&e1, &e2) {
            (
                ConnectionError::Application { source: s1 },
                ConnectionError::Application { source: s2 },
            ) => {
                assert_eq!(s1.code, Code::new(VarInt::from_u32(25)));
                assert_eq!(s2.code, Code::new(VarInt::from_u32(25)));
            }
            _ => panic!("unexpected error shape"),
        }
    }

    #[tokio::test]
    async fn resolve_closed_returns_latched_without_awaiting() {
        let lc = TestLifecycle::new();
        lc.latch.latch_with(|| make_err(7));
        // `wait` is never polled because the latch is already set; if it
        // were, the `take()` below would come back `None`.
        let never = Mutex::new(Some(make_err(8)));
        let got = lc
            .resolve_closed(async { never.lock().unwrap().take().unwrap() })
            .await;
        assert!(matches!(
            &got,
            ConnectionError::Transport { source } if source.kind == VarInt::from_u32(7)
        ));
        assert!(never.lock().unwrap().is_some(), "wait must not be polled");
    }

    #[tokio::test]
    async fn resolve_closed_returns_error_that_wins_during_wait() {
        let lc = TestLifecycle::new();
        let latch = lc.latch.clone();

        let got = lc
            .resolve_closed(async move {
                latch.latch_with(|| make_err(14));
                make_err(15)
            })
            .await;

        assert_eq!(error_kind(&got), VarInt::from_u32(14));
        assert_eq!(
            error_kind(&lc.latch.check().unwrap_err()),
            VarInt::from_u32(14)
        );
    }

    #[tokio::test]
    async fn resolve_closed_latches_wait_result() {
        let lc = TestLifecycle::new();
        lc.set_wait(make_err(5));
        let got = quic::Lifecycle::closed(&lc).await;
        assert!(matches!(
            &got,
            ConnectionError::Transport { source } if source.kind == VarInt::from_u32(5)
        ));
        // Second call must return the same error and must not hit the wait
        // future again.
        let again = quic::Lifecycle::closed(&lc).await;
        assert!(matches!(
            &again,
            ConnectionError::Transport { source } if source.kind == VarInt::from_u32(5)
        ));
    }

    #[tokio::test]
    async fn closed_after_probe_error_returns_latched_without_consuming_wait() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(40));

        let check_error = quic::Lifecycle::check(&lc).unwrap_err();
        assert_eq!(error_kind(&check_error), VarInt::from_u32(40));

        lc.set_wait(make_err(41));
        let closed_error = quic::Lifecycle::closed(&lc).await;

        assert_eq!(error_kind(&closed_error), VarInt::from_u32(40));
        assert!(
            lc.wait.lock().unwrap().is_some(),
            "closed must not poll wait once check has latched the terminal error"
        );
    }

    #[tokio::test]
    async fn closed_latches_default_wait_error_when_wait_is_unset() {
        let lc = TestLifecycle::new();

        let got = quic::Lifecycle::closed(&lc).await;

        assert_eq!(error_kind(&got), VarInt::from_u32(99));
        assert_eq!(
            error_kind(&lc.latch.check().unwrap_err()),
            VarInt::from_u32(99)
        );
    }

    #[tokio::test]
    async fn guard_success_returns_value_without_latching_error() {
        let lc = TestLifecycle::new();

        let out = lc.guard(async { Ok::<_, ConnectionError>(31) }).await;

        assert_eq!(out.unwrap(), 31);
        assert!(lc.latch.check().is_ok());
    }

    #[tokio::test]
    async fn guard_does_not_poll_operation_after_failed_check() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(16));

        let called = Arc::new(Mutex::new(false));
        let called2 = called.clone();
        let res: Result<(), ConnectionError> = lc
            .guard(async move {
                *called2.lock().unwrap() = true;
                Ok(())
            })
            .await;

        assert_eq!(error_kind(&res.unwrap_err()), VarInt::from_u32(16));
        assert!(
            !*called.lock().unwrap(),
            "operation future must not be polled after failed check"
        );
    }

    #[tokio::test]
    async fn guard_latches_operation_error() {
        let lc = TestLifecycle::new();

        let first: Result<(), ConnectionError> = lc.guard(async { Err(make_err(17)) }).await;
        let second: Result<(), ConnectionError> = lc.guard(async { Err(make_err(18)) }).await;

        assert_eq!(error_kind(&first.unwrap_err()), VarInt::from_u32(17));
        assert_eq!(error_kind(&second.unwrap_err()), VarInt::from_u32(17));
    }

    #[tokio::test]
    async fn guard_returns_error_latched_during_operation() {
        let lc = TestLifecycle::new();
        let latch = lc.latch.clone();

        let res: Result<(), ConnectionError> = lc
            .guard(async move {
                latch.latch_with(|| make_err(37));
                Err(make_err(38))
            })
            .await;

        assert_eq!(error_kind(&res.unwrap_err()), VarInt::from_u32(37));
        assert_eq!(
            error_kind(&lc.latch.check().unwrap_err()),
            VarInt::from_u32(37)
        );
    }

    #[tokio::test]
    async fn guard_with_skips_closure_when_latched() {
        let lc = TestLifecycle::new();
        lc.latch.latch_with(|| make_err(7));

        let called = Arc::new(Mutex::new(false));
        let called2 = called.clone();
        let res: Result<(), ConnectionError> = lc
            .guard_with(async { Result::<(), &'static str>::Err("x") }, move |_| {
                *called2.lock().unwrap() = true;
                make_err(8)
            })
            .await;
        assert!(res.is_err());
        assert!(
            !*called.lock().unwrap(),
            "map_err must stay lazy once latched"
        );
    }

    #[tokio::test]
    async fn guard_with_skips_operation_and_mapping_after_failed_check() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(26));

        let err = tokio::time::timeout(
            Duration::from_millis(50),
            lc.guard_with(
                std::future::pending::<Result<(), ConnectionError>>(),
                std::convert::identity,
            ),
        )
        .await
        .expect("guard_with must return before polling the pending operation")
        .unwrap_err();

        assert_eq!(error_kind(&err), VarInt::from_u32(26));
    }

    #[tokio::test]
    async fn guard_with_success_is_untouched() {
        let lc = TestLifecycle::new();
        let out: Result<i32, ConnectionError> = lc
            .guard_with(async { Ok::<_, &'static str>(7) }, |_| make_err(1))
            .await;
        assert_eq!(out.unwrap(), 7);
    }

    #[tokio::test]
    async fn guard_with_error_maps_and_latches_first_error() {
        let lc = TestLifecycle::new();
        let map_calls = Mutex::new(Vec::new());

        let first: Result<(), ConnectionError> = lc
            .guard_with(async { Err::<(), _>(32) }, |tag| {
                map_calls.lock().unwrap().push(tag);
                make_err(tag)
            })
            .await;

        assert_eq!(error_kind(&first.unwrap_err()), VarInt::from_u32(32));
        assert_eq!(map_calls.lock().unwrap().as_slice(), &[32]);

        let second_map_called = Mutex::new(false);
        let second: Result<(), ConnectionError> = lc
            .guard_with(async { Err::<(), _>(33) }, |_| {
                *second_map_called.lock().unwrap() = true;
                make_err(33)
            })
            .await;

        assert_eq!(error_kind(&second.unwrap_err()), VarInt::from_u32(32));
        assert!(
            !*second_map_called.lock().unwrap(),
            "map_err must stay lazy after an error is latched"
        );
    }

    #[test]
    fn guard_sync_success_returns_value_without_latching_error() {
        let lc = TestLifecycle::new();

        lc.guard_sync(ok_unit).unwrap();

        assert!(lc.latch.check().is_ok());
    }

    #[test]
    fn guard_sync_skips_operation_after_failed_check() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(28));

        let err = lc.guard_sync(ok_unit).unwrap_err();

        assert_eq!(error_kind(&err), VarInt::from_u32(28));
    }

    #[test]
    fn guard_sync_skips_closure_when_already_latched() {
        let lc = TestLifecycle::new();
        lc.latch.latch_with(|| make_err(29));

        let called = Mutex::new(false);
        let err = lc
            .guard_sync(|| {
                *called.lock().unwrap() = true;
                Ok::<_, ConnectionError>(())
            })
            .unwrap_err();

        assert_eq!(error_kind(&err), VarInt::from_u32(29));
        assert!(
            !*called.lock().unwrap(),
            "operation closure must not run after a terminal error is latched"
        );
    }

    #[test]
    fn guard_sync_latches_only_first_error() {
        let lc = TestLifecycle::new();
        let a = lc.guard_sync(|| Err::<(), _>(make_err(1))).unwrap_err();
        let b = lc.guard_sync(|| Err::<(), _>(make_err(2))).unwrap_err();
        match (&a, &b) {
            (
                ConnectionError::Transport { source: sa },
                ConnectionError::Transport { source: sb },
            ) => {
                assert_eq!(sa.kind, sb.kind);
                assert_eq!(sa.kind, VarInt::from_u32(1));
            }
            _ => panic!("unexpected error shape"),
        }
    }

    #[test]
    fn guard_sync_with_error_maps_and_latches_first_error() {
        let lc = TestLifecycle::new();
        let map_calls = Mutex::new(Vec::new());

        let first = lc
            .guard_sync_with(
                || Err::<(), _>(35),
                |tag| {
                    map_calls.lock().unwrap().push(tag);
                    make_err(tag)
                },
            )
            .unwrap_err();

        assert_eq!(error_kind(&first), VarInt::from_u32(35));
        assert_eq!(map_calls.lock().unwrap().as_slice(), &[35]);

        let second_map_called = Mutex::new(false);
        let second = lc
            .guard_sync_with(
                || Err::<(), _>(36),
                |_| {
                    *second_map_called.lock().unwrap() = true;
                    make_err(36)
                },
            )
            .unwrap_err();

        assert_eq!(error_kind(&second), VarInt::from_u32(35));
        assert!(
            !*second_map_called.lock().unwrap(),
            "map_err must stay lazy after an error is latched"
        );
    }

    #[test]
    fn guard_sync_with_success_does_not_map_error() {
        let lc = TestLifecycle::new();
        let called = Mutex::new(false);

        let out = lc
            .guard_sync_with(
                || Ok::<_, &'static str>(21),
                |_| {
                    *called.lock().unwrap() = true;
                    make_err(22)
                },
            )
            .unwrap();

        assert_eq!(out, 21);
        assert!(
            !*called.lock().unwrap(),
            "map_err must not run for successful operations"
        );
    }

    #[test]
    fn guard_sync_with_skips_operation_and_mapping_after_failed_check() {
        let lc = TestLifecycle::new();
        lc.set_probe(make_err(23));

        let op_called = Mutex::new(false);
        let map_called = Mutex::new(false);
        let err = lc
            .guard_sync_with(
                || {
                    *op_called.lock().unwrap() = true;
                    Err::<(), _>("not reached")
                },
                |_| {
                    *map_called.lock().unwrap() = true;
                    make_err(24)
                },
            )
            .unwrap_err();

        assert_eq!(error_kind(&err), VarInt::from_u32(23));
        assert!(
            !*op_called.lock().unwrap(),
            "operation must not run after failed check"
        );
        assert!(
            !*map_called.lock().unwrap(),
            "map_err must not run when operation is skipped"
        );
    }
}