bitwarden_ipc/wasm/
communication_backend.rs

1use std::sync::Arc;
2
3use bitwarden_error::bitwarden_error;
4use bitwarden_threading::ThreadBoundRunner;
5use thiserror::Error;
6use tokio::sync::RwLock;
7use wasm_bindgen::prelude::*;
8
9use crate::{
10    error::{ErrorKind, IpcErrorKind},
11    message::{IncomingMessage, OutgoingMessage},
12    traits::{CommunicationBackend, CommunicationBackendReceiver},
13};
14
15#[allow(missing_docs)]
16#[derive(Debug, Error)]
17#[bitwarden_error(basic)]
18#[error("Failed to deserialize incoming message: {0}")]
19pub struct DeserializeError(String);
20
21#[allow(missing_docs)]
22#[derive(Debug, Error)]
23#[bitwarden_error(basic)]
24#[error("Incoming message channel failed: {0}")]
25pub struct ChannelError(String);
26
27/// Error type for the WASM communication backend's send and receive operations.
28///
29/// Distinguishes recoverable failures (which leave the shared IPC client running) from the fatal
30/// closed-channel state. Without this distinction the client's processing loop would treat a
31/// permanently-closed broadcast channel as recoverable and busy-loop on it, since a closed channel
32/// returns an error immediately and forever without ever awaiting.
33#[derive(Debug, Error)]
34pub enum WasmCommunicationError {
35    /// An error returned by the JavaScript backend (e.g. a failed send). Recoverable: the IPC
36    /// client keeps running so future operations can succeed.
37    #[error("{0}")]
38    Js(String),
39
40    /// The incoming message receiver fell behind and `0` messages were dropped. Recoverable: the
41    /// next receive resumes normally.
42    #[error("incoming message channel lagged, {0} messages were dropped")]
43    Lagged(u64),
44
45    /// The destination is not reachable: the JS backend reported "Destination unreachable" (e.g.
46    /// the desktop app is not connected).
47    #[error("Destination unreachable")]
48    Unreachable,
49
50    /// The communication channel was closed because all senders were dropped. This is fatal: the
51    /// IPC client's processing loop stops cleanly instead of busy-looping on the closed channel.
52    #[error("incoming message channel closed")]
53    Closed,
54}
55
56impl IpcErrorKind for WasmCommunicationError {
57    fn kind(&self) -> ErrorKind {
58        match self {
59            WasmCommunicationError::Unreachable => ErrorKind::Unreachable,
60            WasmCommunicationError::Closed => ErrorKind::Fatal,
61            _ => ErrorKind::Other,
62        }
63    }
64}
65
66#[wasm_bindgen(typescript_custom_section)]
67const TS_CUSTOM_TYPES: &'static str = r#"
68export interface IpcCommunicationBackendSender {
69    send(message: OutgoingMessage): Promise<void>;
70}
71"#;
72
73#[wasm_bindgen]
74extern "C" {
75    /// JavaScript interface for handling outgoing messages from the IPC framework.
76    #[wasm_bindgen(js_name = IpcCommunicationBackendSender, typescript_type = "IpcCommunicationBackendSender")]
77    pub type JsCommunicationBackendSender;
78
79    /// Used by the IPC framework to send an outgoing message.
80    #[wasm_bindgen(catch, method, structural)]
81    pub async fn send(
82        this: &JsCommunicationBackendSender,
83        message: OutgoingMessage,
84    ) -> Result<(), JsValue>;
85
86    /// Used by JavaScript to provide an incoming message to the IPC framework.
87    #[wasm_bindgen(catch, method, structural)]
88    pub async fn receive(this: &JsCommunicationBackendSender) -> Result<JsValue, JsValue>;
89}
90
91/// JavaScript implementation of the `CommunicationBackend` trait for IPC communication.
92#[wasm_bindgen(js_name = IpcCommunicationBackend)]
93pub struct JsCommunicationBackend {
94    sender: Arc<ThreadBoundRunner<JsCommunicationBackendSender>>,
95    receive_rx: tokio::sync::broadcast::Receiver<IncomingMessage>,
96    receive_tx: tokio::sync::broadcast::Sender<IncomingMessage>,
97}
98
99impl Clone for JsCommunicationBackend {
100    fn clone(&self) -> Self {
101        Self {
102            sender: self.sender.clone(),
103            receive_rx: self.receive_rx.resubscribe(),
104            receive_tx: self.receive_tx.clone(),
105        }
106    }
107}
108
109#[wasm_bindgen(js_class = IpcCommunicationBackend)]
110impl JsCommunicationBackend {
111    /// Creates a new instance of the JavaScript communication backend.
112    #[wasm_bindgen(constructor)]
113    pub fn new(sender: JsCommunicationBackendSender) -> Self {
114        let (receive_tx, receive_rx) = tokio::sync::broadcast::channel(20);
115        Self {
116            sender: Arc::new(ThreadBoundRunner::new(sender)),
117            receive_rx,
118            receive_tx,
119        }
120    }
121
122    /// Used by JavaScript to provide an incoming message to the IPC framework.
123    pub fn receive(&self, message: IncomingMessage) -> Result<(), JsValue> {
124        self.receive_tx
125            .send(message)
126            .map_err(|e| ChannelError(e.to_string()))?;
127        Ok(())
128    }
129}
130
131impl CommunicationBackend for JsCommunicationBackend {
132    type SendError = WasmCommunicationError;
133    type Receiver = RwLock<tokio::sync::broadcast::Receiver<IncomingMessage>>;
134
135    async fn send(&self, message: OutgoingMessage) -> Result<(), Self::SendError> {
136        // Both the thread-runner failure and the JS-side send failure are treated as recoverable:
137        // a single failed send must not tear down the shared IPC client.
138        self.sender
139            .run_in_thread(|sender| async move {
140                sender.send(message).await.map_err(|e| format!("{e:?}"))
141            })
142            .await
143            .map_err(|e| WasmCommunicationError::Js(e.to_string()))?
144            .map_err(|message| {
145                // The TS backend throws `Error("Destination unreachable")` when the peer transport
146                // is not connected. Map it to a dedicated variant so callers can suppress logging
147                // for this expected case while still surfacing every other send failure.
148                if message.contains("Destination unreachable") {
149                    WasmCommunicationError::Unreachable
150                } else {
151                    WasmCommunicationError::Js(message)
152                }
153            })
154    }
155
156    async fn subscribe(&self) -> Self::Receiver {
157        RwLock::new(self.receive_rx.resubscribe())
158    }
159}
160
161impl CommunicationBackendReceiver for RwLock<tokio::sync::broadcast::Receiver<IncomingMessage>> {
162    type ReceiveError = WasmCommunicationError;
163
164    async fn receive(&self) -> Result<IncomingMessage, Self::ReceiveError> {
165        use tokio::sync::broadcast::error::RecvError;
166
167        self.write().await.recv().await.map_err(|e| match e {
168            // All senders have been dropped; the channel can never produce another message and
169            // `recv()` would return immediately forever. Treat this as fatal so the processing
170            // loop stops instead of busy-looping.
171            RecvError::Closed => WasmCommunicationError::Closed,
172            // The receiver fell behind and missed `skipped` messages. The next `recv()` resumes
173            // normally, so this is recoverable.
174            RecvError::Lagged(skipped) => WasmCommunicationError::Lagged(skipped),
175        })
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use tokio::sync::{RwLock, broadcast};
182
183    use super::*;
184    use crate::{
185        endpoint::{Endpoint, HostId, Source},
186        error::IpcErrorKind,
187    };
188
189    fn test_message() -> IncomingMessage {
190        IncomingMessage {
191            payload: vec![],
192            source: Source::BrowserBackground { id: HostId::Own },
193            destination: Endpoint::BrowserBackground { id: HostId::Own },
194            topic: None,
195        }
196    }
197
198    #[tokio::test]
199    async fn receive_returns_fatal_closed_when_all_senders_are_dropped() {
200        let (tx, rx) = broadcast::channel::<IncomingMessage>(4);
201        let receiver = RwLock::new(rx);
202
203        // Dropping the only sender closes the channel permanently.
204        drop(tx);
205
206        let error = receiver.receive().await.unwrap_err();
207        assert!(matches!(error, WasmCommunicationError::Closed));
208        assert_eq!(error.kind(), ErrorKind::Fatal);
209    }
210
211    #[tokio::test]
212    async fn receive_returns_recoverable_lagged_when_receiver_falls_behind() {
213        let (tx, rx) = broadcast::channel::<IncomingMessage>(1);
214        let receiver = RwLock::new(rx);
215
216        // Overflow the buffer without receiving so the next recv reports lag.
217        for _ in 0..3 {
218            tx.send(test_message()).expect("send should not fail");
219        }
220
221        let error = receiver.receive().await.unwrap_err();
222        assert!(matches!(error, WasmCommunicationError::Lagged(_)));
223        assert_ne!(error.kind(), ErrorKind::Fatal);
224    }
225}
bitwarden_ipc/wasm/communication_backend.rs

bitwarden_ipc/wasm/
communication_backend.rs
