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

bitwarden_ipc/wasm/
communication_backend.rs
