liminal_sdk/remote/
handles.rs

1use alloc::string::{String, ToString};
2use alloc::sync::Arc;
3use alloc::vec::Vec;
4
5use serde::Serialize;
6use serde::de::DeserializeOwned;
7use spin::Mutex;
8
9#[cfg(feature = "std")]
10use crate::DeliveryAck;
11use crate::connection::{
12    ConnectionEvent, ConnectionLifecycle, ConnectionPool, ConnectionState, ResumeRequest,
13    SubscriptionId,
14};
15use crate::embedded::{
16    EmbeddedChannelHandle, EmbeddedConversationHandle, EmptyLifecycleStream, ReadyResult,
17    SdkSubscription,
18};
19use crate::{
20    ChannelHandle, ConversationHandle, ConversationId, PressureResponse, SchemaValidate, SdkError,
21};
22
23pub use super::participant::RemoteParticipantHandle;
24
25use super::config::SdkConfig;
26use super::protocol::{
27    RemoteTransport, WireConversationRequest, WirePublishRequest, WireResumeRequest,
28    deserialize_payload,
29};
30use super::{RemoteConfig, ServerAddress};
31mod negotiated;
32
33/// The refusal typed `subscribe` returns instead of an instantly-dry stream.
34///
35/// No ledger row records this gap: `WIRING-LEDGER.md`'s W6 is the BROWSER
36/// conversation surface (a TypeScript transport lane), not the Rust typed
37/// subscription, and nothing else in `docs/` names it. The refusal says so
38/// rather than borrowing a row that does not cover it.
39fn unwired_typed_subscribe() -> SdkError {
40    SdkError::Unwired {
41        surface: "typed subscribe delivery (ChannelHandle::subscribe on a remote handle)"
42            .to_string(),
43        seam: "no ledger row records this gap; W6 is the browser conversation surface, not this"
44            .to_string(),
45        alternative: "for real delivery use SubscriptionStream::open plus \
46                      SubscriptionStream::recv_timeout, or \
47                      WebSocketSubscriptionStream::open on the WebSocket transport"
48            .to_string(),
49    }
50}
51
52#[derive(Debug)]
53struct RemoteChannelState {
54    lifecycle: ConnectionLifecycle,
55    pool: ConnectionPool,
56    next_subscription: u64,
57}
58
59/// Channel handle that communicates through SDK-internal wire protocol transport.
60///
61/// The handle inherits whatever transport its [`RemoteConfig`] carries. A config
62/// straight out of `RemoteConfig::new` has never been connected, so every
63/// operation on a handle built from it refuses with
64/// [`SdkError::NotConnected`] rather than reporting a success that reached no
65/// wire. Connect the config first.
66#[derive(Clone, Debug)]
67pub struct RemoteChannelHandle {
68    server_address: ServerAddress,
69    channel_name: String,
70    state: Arc<Mutex<RemoteChannelState>>,
71    transport: Arc<dyn RemoteTransport>,
72}
73
74impl RemoteChannelHandle {
75    /// Creates a remote channel handle from validated configuration.
76    ///
77    /// # Errors
78    ///
79    /// Returns [`SdkError`] if the connection pool cannot be created.
80    pub fn new(config: &RemoteConfig) -> Result<Self, SdkError> {
81        Ok(Self {
82            server_address: config.server_address.clone(),
83            channel_name: config.channel_name.clone(),
84            state: Arc::new(Mutex::new(RemoteChannelState {
85                lifecycle: ConnectionLifecycle::new(),
86                pool: ConnectionPool::new(config.pool_config)?,
87                next_subscription: 0,
88            })),
89            transport: Arc::clone(&config.transport),
90        })
91    }
92
93    /// Returns current lifecycle state from the SDK-003 state machine.
94    #[must_use]
95    pub fn connection_state(&self) -> ConnectionState {
96        self.state.lock().lifecycle.state().clone()
97    }
98
99    /// Records that an externally authorized real reconnect attempt started.
100    ///
101    /// # Errors
102    ///
103    /// Returns [`SdkError`] when the lifecycle rejects the transition.
104    pub fn reconnect_started(&self) -> Result<(), SdkError> {
105        self.state.lock().lifecycle.reconnect_started()
106    }
107
108    /// Marks the remote channel connected and builds subscription resume requests.
109    ///
110    /// # Errors
111    ///
112    /// Returns [`SdkError`] if lifecycle transition or recovery state is invalid.
113    pub fn connected(&self) -> Result<Vec<ResumeRequest>, SdkError> {
114        // Compute the resume requests under the lock, then release it before any
115        // transport I/O so concurrent callers do not spin-wait during resume calls.
116        let requests = {
117            let mut state = self.state.lock();
118            let previous = state.lifecycle.state().clone();
119            state.lifecycle.connected()?;
120            let event = ConnectionEvent::new(previous, state.lifecycle.state().clone());
121            state.pool.resume_requests_for_transition(&event)?
122        };
123        for request in &requests {
124            let wire_request = WireResumeRequest::new(*request);
125            self.transport.resume(&self.server_address, &wire_request)?;
126        }
127        Ok(requests)
128    }
129
130    /// Marks a subscription active and assigns it to the configured pool.
131    ///
132    /// # Errors
133    ///
134    /// Returns [`SdkError`] if a subscription id overflows or pool assignment fails.
135    pub fn track_subscription(&self) -> Result<SubscriptionId, SdkError> {
136        let mut state = self.state.lock();
137        let id = SubscriptionId::new(state.next_subscription);
138        state.next_subscription =
139            state
140                .next_subscription
141                .checked_add(1)
142                .ok_or_else(|| SdkError::Store {
143                    description: "subscription id overflow".to_string(),
144                })?;
145        state.pool.assign_subscription(id)?;
146        Ok(id)
147    }
148
149    /// Records an acknowledged sequence for subscription recovery.
150    ///
151    /// # Errors
152    ///
153    /// Returns [`SdkError`] if the subscription is not active in the pool.
154    pub fn acknowledge(
155        &self,
156        subscription_id: SubscriptionId,
157        sequence: u64,
158    ) -> Result<(), SdkError> {
159        let mut state = self.state.lock();
160        state.pool.acknowledge(subscription_id, sequence)
161    }
162
163    /// Returns remote server address used by this handle.
164    #[must_use]
165    pub const fn server_address(&self) -> &ServerAddress {
166        &self.server_address
167    }
168}
169
170impl ChannelHandle for RemoteChannelHandle {
171    type Subscription<M>
172        = SdkSubscription<M>
173    where
174        M: DeserializeOwned;
175
176    type ReplyFuture<'a, Resp>
177        = ReadyResult<Resp>
178    where
179        Self: 'a,
180        Resp: DeserializeOwned + 'a;
181
182    fn publish<M>(&self, message: M) -> Result<PressureResponse, SdkError>
183    where
184        M: Serialize + SchemaValidate,
185    {
186        let request = WirePublishRequest::new(&self.channel_name, &message)?;
187        self.transport.publish(&self.server_address, &request)
188    }
189
190    /// REFUSES. Typed subscription delivery is not implemented on this handle.
191    ///
192    /// The returned stream yields exactly one item — `Err(SdkError::Unwired)` —
193    /// and then ends. It does not touch the transport, so no `Subscribe` is left
194    /// outstanding on the server for a stream that could never deliver.
195    ///
196    /// What this replaces was worse than unimplemented. The old body performed
197    /// the real wire `Subscribe`, registered a subscription in the pool, and
198    /// then returned `SdkSubscription::empty()` — a stream whose `poll_next` is
199    /// `Ready(None)` immediately and forever. A caller's
200    /// `while let Some(m) = sub.next().await` exited on the first poll while the
201    /// server pumped `Deliver` frames at a connection nothing was reading.
202    ///
203    /// For real delivery use the lower-level surface that works today:
204    /// `SubscriptionStream::open` plus `SubscriptionStream::recv_timeout` (or
205    /// `WebSocketSubscriptionStream::open` on the WebSocket transport).
206    fn subscribe<M>(&self) -> Self::Subscription<M>
207    where
208        M: DeserializeOwned,
209    {
210        SdkSubscription::error(unwired_typed_subscribe())
211    }
212
213    fn request_reply<Req, Resp>(&self, request: Req) -> ReadyResult<Resp>
214    where
215        Req: Serialize + SchemaValidate,
216        Resp: DeserializeOwned,
217    {
218        ReadyResult::new(self.request_reply_blocking(&request))
219    }
220}
221
222impl RemoteChannelHandle {
223    /// Publishes a message with an idempotency key and returns a genuine delivery
224    /// ack.
225    ///
226    /// The idempotency key drives dedup-on-delivery on the server: a re-publish of
227    /// the same key is delivered to subscribers at most once. The returned
228    /// [`DeliveryAck`] reports whether this publish was genuinely accepted by a
229    /// subscriber ([`DeliveryAck::is_accepted`]), which a caller such as the aion
230    /// outbox uses to treat the send as done only on real acceptance. This is
231    /// distinct from the backpressure-only [`publish`](ChannelHandle::publish).
232    ///
233    /// # Errors
234    ///
235    /// Returns [`SdkError`] when the message cannot be serialized, the round trip
236    /// fails, or the transport cannot produce a genuine delivery ack.
237    ///
238    /// Only with std: without it no transport could carry the key, so no keyed
239    /// publish is offered rather than one that drops the key (Hermes 7 Oct 08:32).
240    #[cfg(feature = "std")]
241    pub fn publish_with_idempotency_key<M>(
242        &self,
243        message: &M,
244        idempotency_key: &str,
245    ) -> Result<DeliveryAck, SdkError>
246    where
247        M: Serialize + SchemaValidate,
248    {
249        let request =
250            WirePublishRequest::with_idempotency_key(&self.channel_name, message, idempotency_key)?;
251        self.transport
252            .publish_with_delivery(&self.server_address, &request)
253    }
254
255    /// Publishes `message` as the reply to `request` and returns a genuine
256    /// delivery ack (Fable 4c).
257    ///
258    /// `request` is the id a delivered message carried
259    /// ([`DeliveredMessage::message`](crate::DeliveredMessage::message)): the
260    /// reply names it as its causal parent, `<boot>:<id>`, so the server's tap
261    /// can pair the two and measure the dwell between them. A request that
262    /// carried no id (a server before protocol 1.2, or a durable message
263    /// replayed from the store) gives a reply with no parent, never an
264    /// invented one.
265    ///
266    /// # Errors
267    ///
268    /// Returns [`SdkError`] when the message cannot be serialized, the round trip
269    /// fails, or the transport cannot produce a genuine delivery ack.
270    #[cfg(feature = "std")]
271    pub fn reply_to<M>(
272        &self,
273        request: Option<liminal::protocol::DeliveredId>,
274        message: &M,
275    ) -> Result<DeliveryAck, SdkError>
276    where
277        M: Serialize + SchemaValidate,
278    {
279        let reply = WirePublishRequest::reply(&self.channel_name, message, request)?;
280        self.transport
281            .publish_with_delivery(&self.server_address, &reply)
282    }
283
284    /// Performs a correlated request-reply round trip over the transport and
285    /// deserializes the reply.
286    ///
287    /// The round trip rides the conversation request-reply path: the channel name
288    /// is used as the conversation correlation id and subject, so the reply the
289    /// server returns is matched back to this request by `conversation_id` on the
290    /// single synchronous connection. Schema validation still runs on the request
291    /// so a request-reply enforces the same typing contract as `publish`.
292    ///
293    /// # Errors
294    ///
295    /// Returns [`SdkError`] when the request cannot be serialized, the round trip
296    /// fails, or the reply cannot be deserialized into `Resp`.
297    fn request_reply_blocking<Req, Resp>(&self, request: &Req) -> Result<Resp, SdkError>
298    where
299        Req: Serialize + SchemaValidate,
300        Resp: DeserializeOwned,
301    {
302        let conversation_id = ConversationId::new(self.channel_name.clone());
303        let wire_request = WireConversationRequest::new(&conversation_id, request)?;
304        let reply = self
305            .transport
306            .request_reply_conversation(&self.server_address, &wire_request)?;
307        deserialize_payload(&reply)
308    }
309}
310
311/// Conversation handle that communicates through SDK-internal wire protocol transport.
312///
313/// As with [`RemoteChannelHandle`], a handle built from a config that was never
314/// connected refuses every send, request, and receive with
315/// [`SdkError::NotConnected`] instead of reporting a success that reached no
316/// wire.
317#[derive(Clone)]
318pub struct RemoteConversationHandle {
319    server_address: ServerAddress,
320    conversation_id: ConversationId,
321    lifecycle: Arc<Mutex<ConnectionLifecycle>>,
322    transport: Arc<dyn RemoteTransport>,
323    /// Buffered reply bytes from the most recent [`request`](Self::request) round
324    /// trip, drained by the next [`receive`](ConversationHandle::receive). The
325    /// synchronous transport completes the round trip inside `request`, so the
326    /// correlated reply is held here until the caller deserializes it.
327    pending_reply: Arc<Mutex<Option<Vec<u8>>>>,
328}
329
330impl core::fmt::Debug for RemoteConversationHandle {
331    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
332        formatter
333            .debug_struct("RemoteConversationHandle")
334            .field("server_address", &self.server_address)
335            .field("conversation_id", &self.conversation_id)
336            .field(
337                "pending_reply_bytes",
338                &self.pending_reply.lock().as_ref().map(Vec::len),
339            )
340            .finish_non_exhaustive()
341    }
342}
343
344impl RemoteConversationHandle {
345    /// Creates a remote conversation handle from validated configuration.
346    #[must_use]
347    pub fn new(config: &RemoteConfig) -> Self {
348        Self {
349            server_address: config.server_address.clone(),
350            conversation_id: config.conversation_id.clone(),
351            lifecycle: Arc::new(Mutex::new(ConnectionLifecycle::new())),
352            transport: Arc::clone(&config.transport),
353            pending_reply: Arc::new(Mutex::new(None)),
354        }
355    }
356
357    /// Sends a typed request on this conversation and blocks for its correlated
358    /// reply, buffering the reply for the next [`receive`](ConversationHandle::receive).
359    ///
360    /// This is the request leg of the conversation request-reply pattern. It is
361    /// kept distinct from [`send`](ConversationHandle::send), which stays
362    /// fire-and-forget (the server is silent on success): only `request` sets the
363    /// reply-requested flag and waits for the server's correlated answer. The aion
364    /// dispatch model (`send` request, then `receive` reply) maps onto a `request`
365    /// followed by `receive`.
366    ///
367    /// # Errors
368    ///
369    /// Returns [`SdkError`] when the request cannot be serialized or the round
370    /// trip fails. A worker refusal is [`SdkError::WorkerRefused`], with its
371    /// code and exact words, distinct from a route's admission refusal.
372    pub fn request<Req>(&self, request: Req) -> Result<(), SdkError>
373    where
374        Req: Serialize,
375    {
376        let wire_request = WireConversationRequest::new(&self.conversation_id, &request)?;
377        let reply = self
378            .transport
379            .request_reply_conversation(&self.server_address, &wire_request)?;
380        *self.pending_reply.lock() = Some(reply);
381        Ok(())
382    }
383
384    /// Returns current lifecycle state from the SDK-003 state machine.
385    #[must_use]
386    pub fn connection_state(&self) -> ConnectionState {
387        self.lifecycle.lock().state().clone()
388    }
389
390    /// Returns remote server address used by this handle.
391    #[must_use]
392    pub const fn server_address(&self) -> &ServerAddress {
393        &self.server_address
394    }
395}
396
397impl ConversationHandle for RemoteConversationHandle {
398    type ReceiveFuture<'a, M>
399        = ReadyResult<M>
400    where
401        Self: 'a,
402        M: DeserializeOwned + 'a;
403
404    type LifecycleStream = EmptyLifecycleStream;
405
406    fn send<M>(&self, message: M) -> Result<(), SdkError>
407    where
408        M: Serialize,
409    {
410        let request = WireConversationRequest::new(&self.conversation_id, &message)?;
411        self.transport
412            .send_conversation(&self.server_address, &request)
413    }
414
415    fn receive<M>(&self) -> ReadyResult<M>
416    where
417        M: DeserializeOwned,
418    {
419        ReadyResult::new(self.receive_blocking())
420    }
421
422    fn lifecycle(&self) -> Self::LifecycleStream {
423        EmptyLifecycleStream
424    }
425}
426
427impl RemoteConversationHandle {
428    /// Drains and deserializes the correlated reply buffered by the most recent
429    /// [`request`](Self::request).
430    ///
431    /// # Errors
432    ///
433    /// Returns [`SdkError::Conversation`] when no reply is pending (no `request`
434    /// has completed since the last `receive`), or [`SdkError::Serialization`]
435    /// when the buffered reply cannot be deserialized into `M`.
436    fn receive_blocking<M>(&self) -> Result<M, SdkError>
437    where
438        M: DeserializeOwned,
439    {
440        let payload = self
441            .pending_reply
442            .lock()
443            .take()
444            .ok_or_else(|| SdkError::Conversation {
445                conversation_id: self.conversation_id.as_str().to_string(),
446                description: "no correlated reply is pending; call request before receive"
447                    .to_string(),
448            })?;
449        deserialize_payload(&payload)
450    }
451}
452
453/// Runtime-selected channel handle that keeps deployment differences behind configuration.
454#[derive(Clone, Debug)]
455pub enum SdkChannelHandle {
456    /// Embedded direct in-process handle.
457    Embedded(EmbeddedChannelHandle),
458    /// Remote protocol-backed handle.
459    Remote(RemoteChannelHandle),
460}
461
462impl SdkChannelHandle {
463    /// Creates a channel handle selected by SDK configuration.
464    ///
465    /// # Errors
466    ///
467    /// Returns [`SdkError`] if the selected handle cannot be initialized.
468    pub fn new(config: &SdkConfig) -> Result<Self, SdkError> {
469        match config {
470            SdkConfig::Embedded(config) => Ok(Self::Embedded(EmbeddedChannelHandle::new(config))),
471            SdkConfig::Remote(config) => Ok(Self::Remote(RemoteChannelHandle::new(config)?)),
472        }
473    }
474}
475
476impl ChannelHandle for SdkChannelHandle {
477    type Subscription<M>
478        = SdkSubscription<M>
479    where
480        M: DeserializeOwned;
481
482    type ReplyFuture<'a, Resp>
483        = ReadyResult<Resp>
484    where
485        Self: 'a,
486        Resp: DeserializeOwned + 'a;
487
488    fn publish<M>(&self, message: M) -> Result<PressureResponse, SdkError>
489    where
490        M: Serialize + SchemaValidate,
491    {
492        match self {
493            Self::Embedded(handle) => handle.publish(message),
494            Self::Remote(handle) => handle.publish(message),
495        }
496    }
497
498    /// BOTH ARMS REFUSE. Typed subscription delivery is not implemented in
499    /// either deployment mode, so whichever mode configuration selected, the
500    /// returned stream yields exactly one `Err(SdkError::Unwired)` and then
501    /// ends.
502    ///
503    /// The remote arm's refusal names `SubscriptionStream::open` as the surface
504    /// that works. The embedded arm's names the same remote surface, because
505    /// embedded mode has no in-process delivery path to offer instead. That arm
506    /// used to return a locally empty stream — described here as "what embedded
507    /// mode has always meant", which was true of the implementation and
508    /// misleading to a caller: an empty stream is indistinguishable from a live
509    /// subscription on a quiet channel.
510    fn subscribe<M>(&self) -> Self::Subscription<M>
511    where
512        M: DeserializeOwned,
513    {
514        match self {
515            Self::Embedded(handle) => handle.subscribe(),
516            Self::Remote(handle) => handle.subscribe(),
517        }
518    }
519
520    fn request_reply<Req, Resp>(&self, request: Req) -> ReadyResult<Resp>
521    where
522        Req: Serialize + SchemaValidate,
523        Resp: DeserializeOwned,
524    {
525        match self {
526            Self::Embedded(handle) => handle.request_reply(request),
527            Self::Remote(handle) => handle.request_reply(request),
528        }
529    }
530}
531
532/// Runtime-selected conversation handle that keeps deployment differences behind configuration.
533#[derive(Clone, Debug)]
534pub enum SdkConversationHandle {
535    /// Embedded direct in-process handle.
536    Embedded(EmbeddedConversationHandle),
537    /// Remote protocol-backed handle.
538    Remote(RemoteConversationHandle),
539}
540
541impl SdkConversationHandle {
542    /// Creates a conversation handle selected by SDK configuration.
543    ///
544    /// # Errors
545    ///
546    /// Returns [`SdkError`] if the selected handle cannot be initialized.
547    pub fn new(config: &SdkConfig) -> Result<Self, SdkError> {
548        match config {
549            SdkConfig::Embedded(config) => {
550                Ok(Self::Embedded(EmbeddedConversationHandle::new(config)))
551            }
552            SdkConfig::Remote(config) => Ok(Self::Remote(RemoteConversationHandle::new(config))),
553        }
554    }
555}
556
557impl ConversationHandle for SdkConversationHandle {
558    type ReceiveFuture<'a, M>
559        = ReadyResult<M>
560    where
561        Self: 'a,
562        M: DeserializeOwned + 'a;
563
564    type LifecycleStream = EmptyLifecycleStream;
565
566    fn send<M>(&self, message: M) -> Result<(), SdkError>
567    where
568        M: Serialize,
569    {
570        match self {
571            Self::Embedded(handle) => handle.send(message),
572            Self::Remote(handle) => handle.send(message),
573        }
574    }
575
576    fn receive<M>(&self) -> ReadyResult<M>
577    where
578        M: DeserializeOwned,
579    {
580        match self {
581            Self::Embedded(handle) => handle.receive(),
582            Self::Remote(handle) => handle.receive(),
583        }
584    }
585
586    fn lifecycle(&self) -> Self::LifecycleStream {
587        EmptyLifecycleStream
588    }
589}
liminal_sdk/remote/handles.rs

liminal_sdk/remote/
handles.rs
