murmer/
endpoint.rs

1//! Endpoint — the user-facing send handle that abstracts local vs remote actors.
2//!
3//! [`Endpoint<A>`] is the primary API for sending messages to actors. It is
4//! generic over the actor type `A`, which provides compile-time guarantees
5//! that the actor can handle the message being sent.
6//!
7//! # Location transparency
8//!
9//! Under the hood, an endpoint is either:
10//! - **Local** — wraps an `mpsc` channel to the actor's supervisor (zero serialization cost)
11//! - **Remote** — serializes the message with bincode, sends it over a QUIC stream,
12//!   and awaits the deserialized response
13//!
14//! The caller never knows which variant they hold — the `send()` API is identical.
15//!
16//! # Obtaining endpoints
17//!
18//! - `receptionist.start(label, actor, state)` returns an `Endpoint<A>` for the new actor
19//! - `receptionist.lookup::<A>(label)` returns `Option<Endpoint<A>>`
20//! - `ctx.endpoint()` inside a handler returns a self-endpoint
21//! - `listing.next().await` yields endpoints from a subscription stream
22
23use serde::Serialize;
24use serde::de::DeserializeOwned;
25use tokio::sync::{mpsc, oneshot};
26
27use crate::actor::{Actor, AsyncHandler, Handler, Message, RemoteMessage};
28use crate::instrument;
29use crate::wire::{
30    AsyncTypedEnvelope, EnvelopeProxy, ForwardedEnvelope, RemoteInvocation, ReplySender,
31    ResponseRegistry, SendError, TypedEnvelope, next_call_id,
32};
33
34// =============================================================================
35// ENDPOINT
36// =============================================================================
37
38/// The primary handle for interacting with an actor.
39///
40/// `Endpoint<A>` is generic over actor type `A`, providing compile-time
41/// guarantees that the actor can handle the message being sent. Endpoints
42/// are cheap to clone and can be shared across tasks.
43///
44/// Under the hood, an endpoint is either local (in-memory channel) or remote
45/// (serializes over QUIC). The caller never knows which — the API is identical.
46///
47/// # Examples
48///
49/// ```rust,ignore
50/// // Obtain from system.start() or system.lookup()
51/// let counter = system.start("counter/0", Counter, CounterState { count: 0 });
52///
53/// // Send a sync message
54/// let count = counter.send(Increment { amount: 5 }).await?;
55///
56/// // Send an async message
57/// let data = counter.send_async(FetchData { key: "x".into() }).await?;
58///
59/// // Clone and share across tasks
60/// let counter2 = counter.clone();
61/// tokio::spawn(async move {
62///     counter2.send(Increment { amount: 1 }).await.ok();
63/// });
64/// ```
65pub struct Endpoint<A: Actor> {
66    inner: EndpointInner<A>,
67}
68
69impl<A: Actor> Clone for Endpoint<A> {
70    fn clone(&self) -> Self {
71        Self {
72            inner: self.inner.clone(),
73        }
74    }
75}
76
77enum EndpointInner<A: Actor> {
78    Local {
79        mailbox_tx: mpsc::UnboundedSender<Box<dyn EnvelopeProxy<A>>>,
80    },
81    Remote {
82        label: String,
83        wire_tx: mpsc::UnboundedSender<RemoteInvocation>,
84        response_registry: ResponseRegistry,
85    },
86}
87
88impl<A: Actor> Clone for EndpointInner<A> {
89    fn clone(&self) -> Self {
90        match self {
91            Self::Local { mailbox_tx } => Self::Local {
92                mailbox_tx: mailbox_tx.clone(),
93            },
94            Self::Remote {
95                label,
96                wire_tx,
97                response_registry,
98            } => Self::Remote {
99                label: label.clone(),
100                wire_tx: wire_tx.clone(),
101                response_registry: response_registry.clone(),
102            },
103        }
104    }
105}
106
107impl<A: Actor + 'static> Endpoint<A> {
108    pub(crate) fn local(mailbox_tx: mpsc::UnboundedSender<Box<dyn EnvelopeProxy<A>>>) -> Self {
109        Self {
110            inner: EndpointInner::Local { mailbox_tx },
111        }
112    }
113
114    pub(crate) fn remote(
115        label: String,
116        wire_tx: mpsc::UnboundedSender<RemoteInvocation>,
117        response_registry: ResponseRegistry,
118    ) -> Self {
119        Self {
120            inner: EndpointInner::Remote {
121                label,
122                wire_tx,
123                response_registry,
124            },
125        }
126    }
127
128    /// Send a message to the actor. **Identical API for local and remote.**
129    ///
130    /// Compile-time checked:
131    /// - `A` must implement `Handler<M>`
132    /// - `M` must implement `RemoteMessage` (serializable)
133    ///
134    /// For local actors: uses the envelope pattern (zero serialization cost).
135    /// For remote actors: serializes, sends over wire, awaits response.
136    ///
137    /// # Examples
138    ///
139    /// ```rust,ignore
140    /// let count = counter.send(Increment { amount: 5 }).await?;
141    ///
142    /// // Error handling
143    /// match counter.send(Increment { amount: 1 }).await {
144    ///     Ok(new_count) => println!("Count: {new_count}"),
145    ///     Err(SendError::MailboxClosed) => println!("Actor stopped"),
146    ///     Err(e) => println!("Send failed: {e}"),
147    /// }
148    /// ```
149    pub async fn send<M>(&self, message: M) -> Result<M::Result, SendError>
150    where
151        A: Handler<M>,
152        M: RemoteMessage,
153        M::Result: Serialize + DeserializeOwned,
154    {
155        let actor_type = std::any::type_name::<A>();
156        match &self.inner {
157            EndpointInner::Local { mailbox_tx } => {
158                instrument::send_local(actor_type);
159                let (tx, rx) = oneshot::channel();
160                let envelope = TypedEnvelope {
161                    message,
162                    respond_to: tx,
163                };
164                mailbox_tx.send(Box::new(envelope)).map_err(|_| {
165                    instrument::send_error(actor_type, "mailbox_closed");
166                    SendError::MailboxClosed
167                })?;
168                rx.await.map_err(|_| {
169                    instrument::send_error(actor_type, "response_dropped");
170                    SendError::ResponseDropped
171                })
172            }
173            EndpointInner::Remote {
174                label,
175                wire_tx,
176                response_registry,
177            } => {
178                instrument::send_remote(actor_type);
179                #[cfg(feature = "monitor")]
180                let start = std::time::Instant::now(); // determinism-gate: allow — monitor instrumentation (remote-path measurement, not control flow)
181
182                let call_id = next_call_id();
183                let payload = bincode::serde::encode_to_vec(&message, bincode::config::standard())
184                    .map_err(|e| {
185                        instrument::send_error(actor_type, "serialization_failed");
186                        SendError::SerializationFailed(e.to_string())
187                    })?;
188
189                let (resp_tx, resp_rx) = oneshot::channel();
190                response_registry.register(call_id, resp_tx);
191
192                let invocation = RemoteInvocation {
193                    call_id,
194                    actor_label: label.clone(),
195                    message_type: M::TYPE_ID.to_string(),
196                    payload,
197                };
198
199                wire_tx.send(invocation).map_err(|_| {
200                    instrument::send_error(actor_type, "wire_closed");
201                    SendError::WireClosed
202                })?;
203
204                let response = resp_rx.await.map_err(|_| {
205                    instrument::send_error(actor_type, "response_dropped");
206                    SendError::ResponseDropped
207                })?;
208                let result_bytes = response.result.map_err(|e| {
209                    instrument::send_error(actor_type, "remote_error");
210                    SendError::RemoteError(e)
211                })?;
212                let (result, _): (M::Result, _) =
213                    bincode::serde::decode_from_slice(&result_bytes, bincode::config::standard())
214                        .map_err(|e| {
215                        instrument::send_error(actor_type, "deserialization_failed");
216                        SendError::DeserializationFailed(e.to_string())
217                    })?;
218
219                #[cfg(feature = "monitor")]
220                instrument::remote_roundtrip_duration(actor_type, start.elapsed());
221
222                Ok(result)
223            }
224        }
225    }
226
227    /// Send a message to an async handler. **Identical API for local and remote.**
228    ///
229    /// Like [`send`](Self::send) but dispatches through [`AsyncHandler<M>`] so the
230    /// handler can `.await` inside its body.
231    ///
232    /// # Examples
233    ///
234    /// ```rust,ignore
235    /// // For actors that implement AsyncHandler<FetchData>
236    /// let data = endpoint.send_async(FetchData { key: "users".into() }).await?;
237    /// ```
238    pub async fn send_async<M>(&self, message: M) -> Result<M::Result, SendError>
239    where
240        A: AsyncHandler<M>,
241        M: RemoteMessage,
242        M::Result: Serialize + DeserializeOwned,
243    {
244        let actor_type = std::any::type_name::<A>();
245        match &self.inner {
246            EndpointInner::Local { mailbox_tx } => {
247                instrument::send_local(actor_type);
248                let (tx, rx) = oneshot::channel();
249                let envelope = AsyncTypedEnvelope {
250                    message,
251                    respond_to: tx,
252                };
253                mailbox_tx.send(Box::new(envelope)).map_err(|_| {
254                    instrument::send_error(actor_type, "mailbox_closed");
255                    SendError::MailboxClosed
256                })?;
257                rx.await.map_err(|_| {
258                    instrument::send_error(actor_type, "response_dropped");
259                    SendError::ResponseDropped
260                })
261            }
262            EndpointInner::Remote {
263                label,
264                wire_tx,
265                response_registry,
266            } => {
267                instrument::send_remote(actor_type);
268                #[cfg(feature = "monitor")]
269                let start = std::time::Instant::now(); // determinism-gate: allow — monitor instrumentation (remote-path measurement, not control flow)
270
271                // Remote path is identical — serialization doesn't care about sync/async
272                let call_id = next_call_id();
273                let payload = bincode::serde::encode_to_vec(&message, bincode::config::standard())
274                    .map_err(|e| {
275                        instrument::send_error(actor_type, "serialization_failed");
276                        SendError::SerializationFailed(e.to_string())
277                    })?;
278
279                let (resp_tx, resp_rx) = oneshot::channel();
280                response_registry.register(call_id, resp_tx);
281
282                let invocation = RemoteInvocation {
283                    call_id,
284                    actor_label: label.clone(),
285                    message_type: M::TYPE_ID.to_string(),
286                    payload,
287                };
288
289                wire_tx.send(invocation).map_err(|_| {
290                    instrument::send_error(actor_type, "wire_closed");
291                    SendError::WireClosed
292                })?;
293
294                let response = resp_rx.await.map_err(|_| {
295                    instrument::send_error(actor_type, "response_dropped");
296                    SendError::ResponseDropped
297                })?;
298                let result_bytes = response.result.map_err(|e| {
299                    instrument::send_error(actor_type, "remote_error");
300                    SendError::RemoteError(e)
301                })?;
302                let (result, _): (M::Result, _) =
303                    bincode::serde::decode_from_slice(&result_bytes, bincode::config::standard())
304                        .map_err(|e| {
305                        instrument::send_error(actor_type, "deserialization_failed");
306                        SendError::DeserializationFailed(e.to_string())
307                    })?;
308
309                #[cfg(feature = "monitor")]
310                instrument::remote_roundtrip_duration(actor_type, start.elapsed());
311
312                Ok(result)
313            }
314        }
315    }
316
317    /// Internal: send a message using an existing `ReplySender` (for `ctx.forward()`).
318    /// Returns true if the envelope was queued successfully.
319    pub(crate) fn send_with_reply_sender<M>(
320        &self,
321        message: M,
322        reply_sender: ReplySender<M::Result>,
323    ) -> bool
324    where
325        A: Handler<M>,
326        M: Message + Send + 'static,
327        M::Result: Send + 'static,
328    {
329        match &self.inner {
330            EndpointInner::Local { mailbox_tx } => {
331                let envelope = ForwardedEnvelope {
332                    message,
333                    reply_sender,
334                };
335                mailbox_tx.send(Box::new(envelope)).is_ok()
336            }
337            EndpointInner::Remote { .. } => {
338                tracing::warn!("ctx.forward() to remote endpoint is not yet supported");
339                false
340            }
341        }
342    }
343}
murmer/endpoint.rs

murmer/
endpoint.rs
