bitwarden_ipc/

ipc_client.rs

use std::sync::{Arc, Mutex};

use bitwarden_threading::cancellation_token::CancellationToken;
use serde::de::DeserializeOwned;
use thiserror::Error;
use tokio::select;

use crate::{
    constants::CHANNEL_BUFFER_CAPACITY,
    error::{
        AlreadyRunningError, ErrorKind, IpcErrorKind, ReceiveError, SendError, SubscribeError,
        TypedReceiveError,
    },
    message::{IncomingMessage, OutgoingMessage, PayloadTypeName, TypedIncomingMessage},
    rpc::{
        exec::{handler::ErasedRpcHandler, handler_registry::RpcHandlerRegistry},
        request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestPayload},
        response_message::OutgoingRpcResponseMessage,
    },
    serde_utils,
    traits::{CommunicationBackend, CryptoProvider, SessionRepository},
};

/// A subscription to receive messages over IPC.
/// The subcription will start buffering messages after its creation and return them
/// when receive() is called. Messages received before the subscription was created will not be
/// returned.
pub struct IpcClientSubscription {
    pub(crate) receiver: tokio::sync::broadcast::Receiver<IncomingMessage>,
    pub(crate) topic: Option<String>,
}

/// A subscription to receive messages over IPC.
/// The subcription will start buffering messages after its creation and return them
/// when receive() is called. Messages received before the subscription was created will not be
/// returned.
pub struct IpcClientTypedSubscription<Payload: DeserializeOwned + PayloadTypeName>(
    IpcClientSubscription,
    std::marker::PhantomData<Payload>,
);

/// Internal shared state for the IPC client.
struct IpcClientInner<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    crypto: Crypto,
    communication: Com,
    sessions: Ses,

    handlers: RpcHandlerRegistry,
    incoming: Mutex<Option<tokio::sync::broadcast::Receiver<IncomingMessage>>>,
    cancellation_token: Mutex<Option<CancellationToken>>,
}

/// An IPC client that handles communication between different components and clients.
/// It uses a crypto provider to encrypt and decrypt messages, a communication backend to send and
/// receive messages, and a session repository to persist sessions.
///
/// This is the concrete implementation of the [`IpcClient`](crate::IpcClient) trait.
pub struct IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    inner: Arc<IpcClientInner<Crypto, Com, Ses>>,
}

impl<Crypto, Com, Ses> Clone for IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<Crypto, Com, Ses> IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    /// Create a new IPC client with the provided crypto provider, communication backend, and
    /// session repository.
    pub fn new(crypto: Crypto, communication: Com, sessions: Ses) -> Self {
        Self {
            inner: Arc::new(IpcClientInner {
                crypto,
                communication,
                sessions,

                handlers: RpcHandlerRegistry::new(),
                incoming: Mutex::new(None),
                cancellation_token: Mutex::new(None),
            }),
        }
    }
}

#[async_trait::async_trait]
impl<Crypto, Com, Ses> crate::ipc_client_trait::IpcClient for IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    async fn start(
        &self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<(), AlreadyRunningError> {
        if self.is_running() {
            return Err(AlreadyRunningError);
        }

        let cancellation_token = cancellation_token.unwrap_or_default();
        self.inner
            .cancellation_token
            .lock()
            .expect("Failed to lock cancellation token mutex")
            .replace(cancellation_token.clone());

        let com_receiver = self.inner.communication.subscribe().await;
        let (client_tx, client_rx) = tokio::sync::broadcast::channel(CHANNEL_BUFFER_CAPACITY);

        self.inner
            .incoming
            .lock()
            .expect("Failed to lock incoming mutex")
            .replace(client_rx);

        let inner = self.inner.clone();
        let future = async move {
            loop {
                let rpc_topic = RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned();
                select! {
                    _ = cancellation_token.cancelled() => {
                        tracing::debug!("Cancellation signal received, stopping IPC client");
                        break;
                    }
                    received = inner.crypto.receive(&com_receiver, &inner.communication, &inner.sessions) => {
                        match received {
                            Ok(message) if message.topic == Some(rpc_topic) => {
                                handle_rpc_request(&inner, message)
                            }
                            Ok(message) => {
                                if client_tx.send(message).is_err() {
                                    tracing::error!("Failed to save incoming message");
                                    break;
                                };
                            }
                            Err(error) if matches!(error.kind(), ErrorKind::Fatal) => {
                                tracing::error!(?error, "Fatal error receiving message, stopping IPC client");
                                break;
                            }
                            Err(error) => {
                                tracing::warn!(?error, "Recoverable error receiving message, continuing");
                            }
                        }
                    }
                }
            }
            tracing::debug!("IPC client shutting down");
            stop_inner(&inner);
        };

        #[cfg(not(target_arch = "wasm32"))]
        tokio::spawn(future);

        #[cfg(target_arch = "wasm32")]
        wasm_bindgen_futures::spawn_local(future);

        Ok(())
    }

    fn is_running(&self) -> bool {
        let has_incoming = self
            .inner
            .incoming
            .lock()
            .expect("Failed to lock incoming mutex")
            .as_ref()
            .map(|receiver| !receiver.is_closed())
            .unwrap_or(false);
        let has_cancellation_token = self
            .inner
            .cancellation_token
            .lock()
            .expect("Failed to lock cancellation token mutex")
            .is_some();
        has_incoming && has_cancellation_token
    }

    async fn send(&self, message: OutgoingMessage) -> Result<(), SendError> {
        let result = self
            .inner
            .crypto
            .send(&self.inner.communication, &self.inner.sessions, message)
            .await;

        if let Err(ref error) = result {
            match error.kind() {
                ErrorKind::Fatal => {
                    tracing::error!(?error, "Fatal error sending message, stopping IPC client");
                    stop_inner(&self.inner);
                }
                // An unreachable destination is an expected condition and not logged
                ErrorKind::Unreachable => {}
                // Every other recoverable send failure is still surfaced.
                ErrorKind::Other => {
                    tracing::debug!(
                        ?error,
                        "Recoverable error sending message, IPC client will continue running"
                    );
                }
            }
        }

        result.map_err(|e| match e.kind() {
            ErrorKind::Unreachable => SendError::Unreachable,
            _ => SendError::Other(format!("{e:?}")),
        })
    }

    async fn subscribe(
        &self,
        topic: Option<String>,
    ) -> Result<IpcClientSubscription, SubscribeError> {
        Ok(IpcClientSubscription {
            receiver: self
                .inner
                .incoming
                .lock()
                .expect("Failed to lock incoming mutex")
                .as_ref()
                .ok_or(SubscribeError::NotStarted)?
                .resubscribe(),
            topic,
        })
    }

    async fn register_rpc_handler_erased(&self, name: &str, handler: Box<dyn ErasedRpcHandler>) {
        self.inner
            .handlers
            .register_erased(name.to_owned(), handler)
            .await;
    }
}

fn stop_inner<Crypto, Com, Ses>(inner: &IpcClientInner<Crypto, Com, Ses>)
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let mut cancellation_token = inner
        .cancellation_token
        .lock()
        .expect("Failed to lock cancellation token mutex");

    if let Some(cancellation_token) = cancellation_token.take() {
        cancellation_token.cancel();
    }
}

fn handle_rpc_request<Crypto, Com, Ses>(
    inner: &Arc<IpcClientInner<Crypto, Com, Ses>>,
    incoming_message: IncomingMessage,
) where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let inner = inner.clone();
    let future = async move {
        #[derive(Debug, Error)]
        enum HandleError {
            #[error("Failed to deserialize request message: {0}")]
            Deserialize(String),

            #[error("Failed to serialize response message: {0}")]
            Serialize(String),
        }

        async fn handle(
            incoming_message: IncomingMessage,
            handlers: &RpcHandlerRegistry,
        ) -> Result<OutgoingMessage, HandleError> {
            let request = RpcRequestPayload::from_slice(incoming_message.payload.clone()).map_err(
                |e: serde_utils::DeserializeError| HandleError::Deserialize(e.to_string()),
            )?;

            let response = handlers.handle(&request).await;

            let response_message = OutgoingRpcResponseMessage {
                request_id: request.request_id(),
                request_type: request.request_type(),
                result: response,
            };

            // Publish the response on the dedicated topic the requester is subscribed to.
            let payload = serde_utils::to_vec(&response_message)
                .map_err(|e: serde_utils::SerializeError| HandleError::Serialize(e.to_string()))?;

            Ok(OutgoingMessage {
                payload,
                destination: incoming_message.source.into(),
                topic: Some(request.response_topic().to_owned()),
            })
        }

        match handle(incoming_message, &inner.handlers).await {
            Ok(outgoing_message) => {
                // Send response directly through the crypto provider (not through the trait)
                // since we're inside the background task and don't have a trait object.
                let result = inner
                    .crypto
                    .send(&inner.communication, &inner.sessions, outgoing_message)
                    .await;
                if result.is_err() {
                    tracing::error!("Failed to send response message");
                }
            }
            Err(error) => {
                tracing::error!(%error, "Error handling RPC request");
            }
        }
    };

    #[cfg(not(target_arch = "wasm32"))]
    tokio::spawn(future);

    #[cfg(target_arch = "wasm32")]
    wasm_bindgen_futures::spawn_local(future);
}

impl IpcClientSubscription {
    /// Receive a message, optionally filtering by topic.
    /// Setting the cancellation_token to `None` will wait indefinitely.
    pub async fn receive(
        &mut self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<IncomingMessage, ReceiveError> {
        let cancellation_token = cancellation_token.unwrap_or_default();

        loop {
            select! {
                _ = cancellation_token.cancelled() => {
                    return Err(ReceiveError::Cancelled)
                }
                result = self.receiver.recv() => {
                    let received = result?;
                    if self.topic.is_none() || received.topic == self.topic {
                        return Ok::<IncomingMessage, ReceiveError>(received);
                    }
                }
            }
        }
    }
}

impl<Payload> IpcClientTypedSubscription<Payload>
where
    Payload: DeserializeOwned + PayloadTypeName,
{
    pub(crate) fn new(subscription: IpcClientSubscription) -> Self {
        Self(subscription, std::marker::PhantomData)
    }

    /// Receive a message.
    /// Setting the cancellation_token to `None` will wait indefinitely.
    pub async fn receive(
        &mut self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<TypedIncomingMessage<Payload>, TypedReceiveError> {
        let received = self.0.receive(cancellation_token).await?;
        received
            .try_into()
            .map_err(|e: serde_utils::DeserializeError| TypedReceiveError::Typing(e.to_string()))
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, time::Duration};

    use bitwarden_threading::time::sleep;
    use serde::{Deserialize, Serialize};

    use super::*;
    use crate::{
        IpcClientExt,
        endpoint::{Endpoint, HostId, Source},
        ipc_client_trait::IpcClient,
        message::PayloadTypeName,
        rpc::{
            request::RpcRequest,
            request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestMessage},
            response_message::IncomingRpcResponseMessage,
        },
        traits::{InMemorySessionRepository, NoEncryptionCryptoProvider, TestCommunicationBackend},
    };

    /// Error type for [`TestCryptoProvider`] that carries an explicit fatal/recoverable
    /// classification so tests can exercise both control-flow paths.
    #[derive(Debug, Clone)]
    struct TestCryptoError {
        // Read only through the derived `Debug` impl (the outer `SendError` wraps it via
        // `{e:?}`), which dead-code analysis does not count as a use.
        #[allow(dead_code)]
        message: String,
        fatal: bool,
    }

    impl IpcErrorKind for TestCryptoError {
        fn kind(&self) -> ErrorKind {
            if self.fatal {
                ErrorKind::Fatal
            } else {
                ErrorKind::Other
            }
        }
    }

    struct TestCryptoProvider {
        /// Simulate a send result. Set to `None` wait indefinitely
        send_result: Option<Result<(), TestCryptoError>>,
        /// Simulate a receive result. Set to `None` wait indefinitely
        receive_result: Option<Result<IncomingMessage, TestCryptoError>>,
    }

    type TestSessionRepository = InMemorySessionRepository<String>;
    impl CryptoProvider<TestCommunicationBackend, TestSessionRepository> for TestCryptoProvider {
        type Session = String;
        type SendError = TestCryptoError;
        type ReceiveError = TestCryptoError;

        async fn receive(
            &self,
            _receiver: &<TestCommunicationBackend as CommunicationBackend>::Receiver,
            _communication: &TestCommunicationBackend,
            _sessions: &TestSessionRepository,
        ) -> Result<IncomingMessage, Self::ReceiveError> {
            match &self.receive_result {
                Some(result) => {
                    // Yield (and throttle) so a recoverable error that makes the processing loop
                    // `continue` doesn't busy-spin and starve the single-threaded test runtime.
                    // Real backends await their underlying transport here, which has the same
                    // effect.
                    sleep(Duration::from_millis(5)).await;
                    result.clone()
                }
                None => {
                    // Simulate waiting for a message but never returning
                    sleep(Duration::from_secs(600)).await;
                    Err(TestCryptoError {
                        message: "Simulated timeout".to_string(),
                        fatal: true,
                    })
                }
            }
        }

        async fn send(
            &self,
            _communication: &TestCommunicationBackend,
            _sessions: &TestSessionRepository,
            _message: OutgoingMessage,
        ) -> Result<(), Self::SendError> {
            match &self.send_result {
                Some(result) => result.clone(),
                None => {
                    // Simulate waiting for a message to be send but never returning
                    sleep(Duration::from_secs(600)).await;
                    Err(TestCryptoError {
                        message: "Simulated timeout".to_string(),
                        fatal: true,
                    })
                }
            }
        }
    }

    #[tokio::test]
    async fn returns_send_error_when_crypto_provider_returns_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
            receive_result: Some(Err(TestCryptoError {
                message: "Should not have be called".to_string(),
                fatal: false,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();

        assert!(error.to_string().contains("Crypto error"));
    }

    #[tokio::test]
    async fn communication_provider_has_outgoing_message_when_sending_through_ipc_client() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;

        client.send(message.clone()).await.unwrap();

        let outgoing_messages = communication_provider.outgoing().await;
        assert_eq!(outgoing_messages, vec![message]);
    }

    #[tokio::test]
    async fn returns_received_message_when_received_from_backend() {
        let message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;

        let mut subscription = client
            .subscribe(None)
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(message.clone());
        let received_message = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, message);
    }

    #[tokio::test]
    async fn skips_non_matching_topics_and_returns_first_matching_message() {
        let non_matching_message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("non_matching_topic".to_owned()),
        };
        let matching_message = IncomingMessage {
            payload: vec![109],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("matching_topic".to_owned()),
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe(Some("matching_topic".to_owned()))
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(non_matching_message.clone());
        communication_provider.push_incoming(non_matching_message.clone());
        communication_provider.push_incoming(matching_message.clone());

        let received_message: IncomingMessage = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, matching_message);
    }

    #[tokio::test]
    async fn skips_unrelated_messages_and_returns_typed_message() {
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestPayload {
            some_data: String,
        }

        impl PayloadTypeName for TestPayload {
            const PAYLOAD_TYPE_NAME: &str = "TestPayload";
        }

        let unrelated = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let typed_message = crate::message::TypedIncomingMessage {
            payload: TestPayload {
                some_data: "Hello, world!".to_string(),
            },
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe_typed::<TestPayload>()
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(unrelated.clone());
        communication_provider.push_incoming(unrelated.clone());
        communication_provider.push_incoming(
            typed_message
                .clone()
                .try_into()
                .expect("Serialization should not fail"),
        );

        let received_message = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, typed_message);
    }

    #[tokio::test]
    async fn returns_error_if_related_message_was_not_deserializable() {
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestPayload {
            some_data: String,
        }

        impl PayloadTypeName for TestPayload {
            const PAYLOAD_TYPE_NAME: &str = "TestPayload";
        }

        let non_deserializable_message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("TestPayload".to_owned()),
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe_typed::<TestPayload>()
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(non_deserializable_message.clone());

        let result = subscription.receive(None).await;
        assert!(matches!(result, Err(TypedReceiveError::Typing(_))));
    }

    #[tokio::test]
    async fn ipc_client_stops_if_crypto_returns_fatal_send_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: true,
            })),
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();
        let is_running = client.is_running();

        assert!(error.to_string().contains("Crypto error"));
        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_keeps_running_if_crypto_returns_recoverable_send_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            // A recoverable send error (e.g. a handshake timeout because the peer is down) must
            // not tear down the shared client.
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
            // Block forever on receive so the loop stays alive while we inspect it.
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();
        let is_running = client.is_running();

        // The error is still surfaced to the caller...
        assert!(error.to_string().contains("Crypto error"));
        // ...but the client keeps running so future sends/requests can succeed.
        assert!(is_running);
    }

    #[tokio::test]
    async fn ipc_client_stops_if_crypto_returns_fatal_receive_error() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: true,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process the error
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
        assert!(cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_keeps_running_if_crypto_returns_recoverable_receive_error() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            // A recoverable receive error must not stop the processing loop.
            receive_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client time to hit the recoverable receive error (repeatedly).
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(is_running);
        assert!(!cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_is_not_running_if_cancellation_token_is_cancelled() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Cancel the token and give the client some time to process the cancellation
        cancellation_token.cancel();
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_is_running_if_no_errors_are_encountered() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(is_running);
        assert!(!cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_is_not_running_if_not_started() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_start_returns_error_if_already_running() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let first_result = client.start(Some(cancellation_token.clone())).await;
        assert_eq!(first_result, Ok(()));

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(client.is_running());

        let second_result = client.start(Some(cancellation_token.clone())).await;
        assert_eq!(second_result, Err(AlreadyRunningError));
    }

    mod request {
        use super::*;
        use crate::RpcHandler;

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestRequest {
            a: i32,
            b: i32,
        }

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestResponse {
            result: i32,
        }

        impl RpcRequest for TestRequest {
            type Response = TestResponse;

            const NAME: &str = "TestRequest";
        }

        struct TestHandler;

        impl RpcHandler for TestHandler {
            type Request = TestRequest;

            async fn handle(&self, request: Self::Request) -> TestResponse {
                TestResponse {
                    result: request.a + request.b,
                }
            }
        }

        /// A second request type whose response is a unit enum, which serde serializes to a bare
        /// JSON string. Deserializing it as [`TestResponse`] fails, which is what makes it a
        /// faithful stand-in for the real `GetBiometricsStatus` / `UnlockBiometrics` pair.
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct OtherRequest;

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        enum OtherResponse {
            Available,
        }

        impl RpcRequest for OtherRequest {
            type Response = OtherResponse;

            const NAME: &str = "OtherRequest";
        }

        #[tokio::test]
        async fn request_sends_message_and_returns_response() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;
            let request = TestRequest { a: 1, b: 2 };
            let response = TestResponse { result: 3 };

            // Send the request
            let request_clone = request.clone();
            let client_clone = client.clone();
            let result_handle = tokio::spawn(async move {
                client_clone
                    .request::<TestRequest>(
                        request_clone,
                        Endpoint::BrowserBackground { id: HostId::Own },
                        None,
                    )
                    .await
            });
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Read and verify the outgoing message
            let outgoing_messages = communication_provider.outgoing().await;
            let outgoing_request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing_messages[0].payload)
                    .expect("Deserialization should not fail");
            assert_eq!(outgoing_request.request_type, "TestRequest");
            assert_eq!(outgoing_request.request, request);

            // Simulate receiving a response
            let simulated_response = IncomingRpcResponseMessage {
                result: Ok(response),
                request_id: outgoing_request.request_id.clone(),
                request_type: outgoing_request.request_type.clone(),
            };
            let simulated_response = IncomingMessage {
                payload: serde_utils::to_vec(&simulated_response)
                    .expect("Serialization should not fail"),
                source: Source::BrowserBackground { id: HostId::Own },
                destination: Endpoint::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                },
                // The requester subscribes to the dedicated topic it minted, so the response must
                // be published there.
                topic: Some(outgoing_request.response_topic.clone()),
            };
            communication_provider.push_incoming(simulated_response);

            // Wait for the response
            let result = result_handle.await.unwrap();
            assert_eq!(result.unwrap().result, 3);
        }

        /// Two requests in flight at once must each receive their own response. Each request mints
        /// its own response topic, so `TestRequest` and `OtherRequest` never see each other's
        /// responses even though their response types differ (`OtherResponse::Available` is a bare
        /// JSON string that would not deserialize into `TestResponse`).
        #[tokio::test]
        async fn concurrent_request_of_a_different_type_does_not_disturb_a_pending_request() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;

            // Put both requests in flight at the same time.
            let test_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            let other_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<OtherRequest>(
                            OtherRequest,
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Recover both requests from the wire, keyed by type, so we can answer each on the
            // dedicated response topic it minted.
            let outgoing = communication_provider.outgoing().await;
            let requests: HashMap<String, RpcRequestMessage<serde_json::Value>> = outgoing
                .iter()
                .map(|message| {
                    let partial: RpcRequestMessage<serde_json::Value> =
                        serde_utils::from_slice(&message.payload)
                            .expect("Deserialization should not fail");
                    (partial.request_type.clone(), partial)
                })
                .collect();

            let respond = |topic: String, payload: Vec<u8>| {
                communication_provider.push_incoming(IncomingMessage {
                    payload,
                    source: Source::BrowserBackground { id: HostId::Own },
                    destination: Endpoint::Web {
                        tab_id: 9001,
                        document_id: "doc-1".to_string(),
                    },
                    topic: Some(topic),
                });
            };

            // Answer `OtherRequest` first, while `TestRequest` is still waiting. Each response goes
            // to its own dedicated topic, so the two can never be confused for one another.
            respond(
                requests["OtherRequest"].response_topic.clone(),
                serde_utils::to_vec(&IncomingRpcResponseMessage {
                    result: Ok(OtherResponse::Available),
                    request_id: requests["OtherRequest"].request_id.clone(),
                    request_type: "OtherRequest".to_string(),
                })
                .expect("Serialization should not fail"),
            );

            respond(
                requests["TestRequest"].response_topic.clone(),
                serde_utils::to_vec(&IncomingRpcResponseMessage {
                    result: Ok(TestResponse { result: 3 }),
                    request_id: requests["TestRequest"].request_id.clone(),
                    request_type: "TestRequest".to_string(),
                })
                .expect("Serialization should not fail"),
            );

            assert_eq!(
                other_handle
                    .await
                    .unwrap()
                    .expect("OtherRequest should succeed"),
                OtherResponse::Available
            );
            assert_eq!(
                test_handle
                    .await
                    .unwrap()
                    .expect("TestRequest should succeed"),
                TestResponse { result: 3 }
            );
        }

        /// A response that reaches this request's dedicated topic but fails to deserialize into the
        /// expected response type is a genuine error for *this* request: because the topic is
        /// dedicated, there is no other request it could belong to, so the failure is surfaced
        /// immediately.
        #[tokio::test]
        async fn malformed_response_surfaces_error_instead_of_hanging() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;

            let result_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Recover the dedicated response topic and answer it with a body that cannot be
            // deserialized into `TestResponse` (`OtherResponse::Available` serializes to a bare
            // string).
            let outgoing = communication_provider.outgoing().await;
            let request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing[0].payload)
                    .expect("Deserialization should not fail");
            let malformed = IncomingRpcResponseMessage {
                result: Ok(OtherResponse::Available),
                request_id: request.request_id.clone(),
                request_type: request.request_type.clone(),
            };
            communication_provider.push_incoming(IncomingMessage {
                payload: serde_utils::to_vec(&malformed).expect("Serialization should not fail"),
                source: Source::BrowserBackground { id: HostId::Own },
                destination: Endpoint::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                },
                topic: Some(request.response_topic.clone()),
            });

            let result = tokio::time::timeout(Duration::from_secs(5), result_handle)
                .await
                .expect("request must not hang on a malformed response")
                .unwrap();
            assert!(matches!(
                result,
                Err(crate::error::RequestError::Rpc(
                    crate::rpc::error::RpcError::ResponseDeserialization(_)
                ))
            ));
        }

        #[tokio::test]
        async fn incoming_rpc_message_handles_request_and_returns_response() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;
            let request_id = uuid::Uuid::new_v4().to_string();
            let request = TestRequest { a: 1, b: 2 };
            let response = TestResponse { result: 3 };

            // Register the handler
            client.register_rpc_handler(TestHandler).await;

            // Simulate receiving a request
            let response_topic = format!("RpcResponseMessage:{request_id}");
            let simulated_request = RpcRequestMessage {
                request,
                request_id: request_id.clone(),
                request_type: "TestRequest".to_string(),
                response_topic: response_topic.clone(),
            };
            let simulated_request_message = IncomingMessage {
                payload: serde_utils::to_vec(&simulated_request)
                    .expect("Serialization should not fail"),
                source: Source::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                    origin: "https://example.com".to_string(),
                },
                destination: Endpoint::BrowserBackground { id: HostId::Own },
                topic: Some(RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned()),
            };
            communication_provider.push_incoming(simulated_request_message);

            // Give the client some time to process the request
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Read and verify the outgoing message
            let outgoing_messages = communication_provider.outgoing().await;
            let outgoing_response: IncomingRpcResponseMessage<TestResponse> =
                serde_utils::from_slice(&outgoing_messages[0].payload)
                    .expect("Deserialization should not fail");

            // The response must be published on the dedicated topic carried by the request.
            assert_eq!(outgoing_messages[0].topic, Some(response_topic));
            assert_eq!(outgoing_response.request_type, "TestRequest");
            assert_eq!(outgoing_response.result, Ok(response));
        }
    }
}