sim_lib_web_bridge/
remote.rs

1//! Network transports (wasm, local server, remote server).
2//!
3//! These share the [`Transport`] contract with the fixture so they are
4//! interchangeable behind the session bridge. Each connects its runtime through
5//! `realize`/`EvalFabric` (HTTP bootstrap plus a WebSocket live channel for the
6//! server transports, the in-process fabric for wasm). Disconnected transports
7//! fail closed, so the session degrades to a visible state rather than a crash.
8
9use std::collections::BTreeSet;
10use std::sync::Arc;
11use std::time::Duration;
12
13use sim_kernel::{Consistency, Cx, Error, EvalMode, EvalReply, EvalRequest, Expr, Result, Symbol};
14use sim_lib_server::{
15    EvalSite, FrameEnvelope, ServerAddress, ServerFrame, connect_transport_site,
16    eval_reply_from_frame, server_frame_from_request,
17};
18use sim_lib_stream_core::{
19    BufferPolicy, ClockDomain, PushResult, StreamDirection, StreamEnvelope,
20    StreamInspectorSnapshot, StreamItem, StreamMedia, StreamMetadata, StreamStats,
21    TransportProfile, stream_inspector_route_local_symbol,
22};
23use sim_lib_stream_fabric::{StreamControl, stream_control_frame_from_control};
24use sim_lib_view::Operation;
25
26use crate::transport::{
27    BrowserStreamStatus, ChangeEvent, SessionStatus, StreamInspectorRecord, Transport,
28    TransportKind,
29};
30
31/// A network-backed transport that connects a runtime over `realize`.
32pub struct RemoteTransport {
33    kind: TransportKind,
34    status: SessionStatus,
35    endpoint: String,
36    address: ServerAddress,
37    offered_codecs: Vec<Symbol>,
38    codec: Option<Symbol>,
39    site: Option<Arc<dyn EvalSite>>,
40    resources: BTreeSet<Symbol>,
41    next_msg_id: u64,
42    in_flight: BTreeSet<u64>,
43    max_in_flight: usize,
44    timeout: Option<Duration>,
45}
46
47impl RemoteTransport {
48    /// A wasm transport targeting an in-browser runtime.
49    pub fn wasm() -> Self {
50        Self::new(
51            TransportKind::Wasm,
52            "wasm:local",
53            ServerAddress::Wasm {
54                region: "local".to_owned(),
55            },
56        )
57    }
58
59    /// A local-server transport (HTTP bootstrap + WebSocket live).
60    pub fn local_server(endpoint: impl Into<String>) -> Self {
61        let endpoint = endpoint.into();
62        Self::new(
63            TransportKind::LocalServer,
64            endpoint.clone(),
65            server_address_from_endpoint(&endpoint, true),
66        )
67    }
68
69    /// A remote-server transport (HTTP bootstrap + WebSocket live).
70    pub fn remote_server(endpoint: impl Into<String>) -> Self {
71        let endpoint = endpoint.into();
72        Self::new(
73            TransportKind::RemoteServer,
74            endpoint.clone(),
75            server_address_from_endpoint(&endpoint, false),
76        )
77    }
78
79    /// A local-server transport targeting an explicit server address.
80    pub fn local_server_address(endpoint: impl Into<String>, address: ServerAddress) -> Self {
81        Self::new(TransportKind::LocalServer, endpoint, address)
82    }
83
84    /// A remote-server transport targeting an explicit server address.
85    pub fn remote_server_address(endpoint: impl Into<String>, address: ServerAddress) -> Self {
86        Self::new(TransportKind::RemoteServer, endpoint, address)
87    }
88
89    fn new(kind: TransportKind, endpoint: impl Into<String>, address: ServerAddress) -> Self {
90        Self {
91            kind,
92            status: SessionStatus::Disconnected,
93            endpoint: endpoint.into(),
94            address,
95            offered_codecs: vec![Symbol::qualified("codec", "binary")],
96            codec: None,
97            site: None,
98            resources: BTreeSet::new(),
99            next_msg_id: 1,
100            in_flight: BTreeSet::new(),
101            max_in_flight: 8,
102            timeout: None,
103        }
104    }
105
106    /// The configured endpoint.
107    pub fn endpoint(&self) -> &str {
108        &self.endpoint
109    }
110
111    /// The server address this transport connects to.
112    pub fn address(&self) -> &ServerAddress {
113        &self.address
114    }
115
116    /// Override the codec offer set used during negotiation.
117    pub fn with_offered_codecs(mut self, offered_codecs: Vec<Symbol>) -> Self {
118        self.offered_codecs = offered_codecs;
119        self
120    }
121
122    /// Override the maximum number of correlated requests allowed in flight.
123    pub fn with_max_in_flight(mut self, max_in_flight: usize) -> Self {
124        self.max_in_flight = max_in_flight.max(1);
125        self
126    }
127
128    /// Override the request timeout used for server answers.
129    pub fn with_timeout(mut self, timeout: Duration) -> Self {
130        self.timeout = Some(timeout);
131        self
132    }
133
134    /// Request a remote connection and negotiate the server codec.
135    pub fn connect(&mut self, cx: &mut Cx) -> Result<()> {
136        self.status = SessionStatus::Connecting;
137        match connect_transport_site(cx, self.address.clone(), self.offered_codecs.clone()) {
138            Ok((site, codec)) => {
139                self.site = Some(site);
140                self.codec = Some(codec);
141                self.in_flight.clear();
142                self.status = SessionStatus::Connected;
143                Ok(())
144            }
145            Err(error) => {
146                self.site = None;
147                self.codec = None;
148                self.status = SessionStatus::Disconnected;
149                Err(error)
150            }
151        }
152    }
153
154    /// Mark the remote channel disconnected.
155    pub fn disconnect(&mut self) {
156        self.site = None;
157        self.codec = None;
158        self.in_flight.clear();
159        self.status = SessionStatus::Disconnected;
160    }
161
162    /// Mark the remote channel reconnecting.
163    pub fn begin_reconnect(&mut self) {
164        self.site = None;
165        self.codec = None;
166        self.in_flight.clear();
167        self.status = SessionStatus::Reconnecting;
168    }
169
170    /// Close the remote channel deliberately.
171    pub fn close(&mut self, cx: &mut Cx) -> Result<()> {
172        if let Some(site) = self.site.take() {
173            site.close_connection(cx)?;
174        }
175        self.codec = None;
176        self.resources.clear();
177        self.in_flight.clear();
178        self.status = SessionStatus::Closed;
179        Ok(())
180    }
181
182    /// Encode a stream-fabric control frame for server-backed transports.
183    pub fn stream_control_frame(
184        &self,
185        cx: &mut Cx,
186        codec: Symbol,
187        control: &StreamControl,
188    ) -> Result<ServerFrame> {
189        match self.kind {
190            TransportKind::LocalServer | TransportKind::RemoteServer => {
191                stream_control_frame_from_control(cx, codec, control, FrameEnvelope::default())
192            }
193            TransportKind::Fixture | TransportKind::Wasm | TransportKind::Fabric => {
194                Err(Error::HostError(format!(
195                    "{:?} transport does not use server stream-fabric frames",
196                    self.kind
197                )))
198            }
199        }
200    }
201
202    fn not_connected(&self) -> Error {
203        Error::HostError(format!(
204            "{:?} transport to {} is {:?}; no traffic can flow",
205            self.kind, self.endpoint, self.status
206        ))
207    }
208
209    fn request(&mut self, cx: &mut Cx, expr: Expr) -> Result<Expr> {
210        if !self.status.is_live() {
211            return Err(self.not_connected());
212        }
213        if self.in_flight.len() >= self.max_in_flight {
214            return Err(Error::HostError(format!(
215                "{:?} transport to {} has {} in-flight requests (limit {})",
216                self.kind,
217                self.endpoint,
218                self.in_flight.len(),
219                self.max_in_flight
220            )));
221        }
222        let Some(site) = self.site.clone() else {
223            return Err(self.not_connected());
224        };
225        let Some(codec) = self.codec.clone() else {
226            return Err(self.not_connected());
227        };
228
229        let msg_id = self.next_msg_id;
230        self.next_msg_id = self.next_msg_id.saturating_add(1);
231        self.in_flight.insert(msg_id);
232
233        let mut frame = server_frame_from_request(cx, &codec, web_session_request(expr))?;
234        frame.msg_id = Some(msg_id);
235        frame.envelope.reply_codec_hint = Some(codec.clone());
236        let reply = site.answer_with_timeout(cx, frame, self.timeout);
237        self.in_flight.remove(&msg_id);
238
239        let reply = match reply {
240            Ok(reply) => reply,
241            Err(error) => {
242                self.status = SessionStatus::Disconnected;
243                return Err(error);
244            }
245        };
246        if reply.correlate != Some(msg_id) {
247            return Err(Error::HostError(format!(
248                "server reply correlation {:?} did not match request {msg_id}",
249                reply.correlate
250            )));
251        }
252        let EvalReply { value, .. } = eval_reply_from_frame(cx, &reply)?;
253        let expr = value.object().as_expr(cx)?;
254        if let Some(message) = remote_error_message(&expr) {
255            return Err(Error::HostError(message));
256        }
257        Ok(expr)
258    }
259
260    fn stream_unavailable(&self, stream_id: &Symbol, operation: &str) -> Error {
261        Error::HostError(format!(
262            "cannot {operation} stream {stream_id}: {:?} transport to {} uses server eval requests for web-session resources",
263            self.kind, self.endpoint
264        ))
265    }
266}
267
268impl Transport for RemoteTransport {
269    fn kind(&self) -> TransportKind {
270        self.kind
271    }
272
273    fn status(&self) -> SessionStatus {
274        self.status
275    }
276
277    fn read(&mut self, cx: &mut Cx, resource: &Symbol) -> Result<Expr> {
278        let value = self.request(cx, web_session_read(resource))?;
279        self.resources.insert(resource.clone());
280        Ok(value)
281    }
282
283    fn realize_operation(
284        &mut self,
285        cx: &mut Cx,
286        resource: &Symbol,
287        operation: &Operation,
288    ) -> Result<Expr> {
289        let value = self.request(cx, web_session_realize(resource, operation))?;
290        self.resources.insert(resource.clone());
291        Ok(value)
292    }
293
294    fn commit_operation(
295        &mut self,
296        cx: &mut Cx,
297        resource: &Symbol,
298        operation: &Operation,
299        expected_current: Option<&Expr>,
300    ) -> Result<Expr> {
301        let value = self.request(
302            cx,
303            web_session_commit(resource, operation, expected_current),
304        )?;
305        self.resources.insert(resource.clone());
306        Ok(value)
307    }
308
309    fn drain_events(&mut self, cx: &mut Cx) -> Result<Vec<ChangeEvent>> {
310        let resources = self.resources.iter().cloned().collect::<Vec<_>>();
311        let mut events = Vec::new();
312        for resource in resources {
313            events.extend(parse_changes(
314                self.request(cx, web_session_changes(&resource))?,
315            )?);
316        }
317        Ok(events)
318    }
319
320    fn stream_subscribe(
321        &mut self,
322        _cx: &mut Cx,
323        stream_id: &Symbol,
324    ) -> Result<StreamInspectorRecord> {
325        Err(self.stream_unavailable(stream_id, "subscribe to"))
326    }
327
328    fn stream_read(
329        &mut self,
330        _cx: &mut Cx,
331        stream_id: &Symbol,
332        _limit: usize,
333    ) -> Result<Vec<StreamItem>> {
334        Err(self.stream_unavailable(stream_id, "read"))
335    }
336
337    fn stream_push(
338        &mut self,
339        _cx: &mut Cx,
340        stream_id: &Symbol,
341        _envelope: StreamEnvelope,
342    ) -> Result<PushResult> {
343        Err(self.stream_unavailable(stream_id, "push"))
344    }
345
346    fn stream_cancel(&mut self, _cx: &mut Cx, stream_id: &Symbol) -> Result<()> {
347        Err(self.stream_unavailable(stream_id, "cancel"))
348    }
349
350    fn stream_stats(&mut self, _cx: &mut Cx, stream_id: &Symbol) -> Result<StreamStats> {
351        Err(self.stream_unavailable(stream_id, "inspect stats for"))
352    }
353
354    fn stream_inspector(
355        &mut self,
356        _cx: &mut Cx,
357        stream_id: &Symbol,
358    ) -> Result<StreamInspectorRecord> {
359        let status = match self.status {
360            SessionStatus::Disconnected => BrowserStreamStatus::Disconnected,
361            SessionStatus::Reconnecting => BrowserStreamStatus::Reconnecting,
362            SessionStatus::Closed => BrowserStreamStatus::Cancelled,
363            SessionStatus::Connected => BrowserStreamStatus::Disconnected,
364            SessionStatus::Connecting => BrowserStreamStatus::Disconnected,
365        };
366        Ok(StreamInspectorRecord {
367            stream_id: stream_id.clone(),
368            status,
369            buffered: 0,
370            stats: StreamStats::default(),
371            diagnostics: Vec::new(),
372            snapshot: StreamInspectorSnapshot::new(
373                &StreamMetadata::new(
374                    stream_id.clone(),
375                    StreamMedia::Data,
376                    StreamDirection::Source,
377                    ClockDomain::ServerFrame.symbol(),
378                    BufferPolicy::bounded(1)?,
379                ),
380                stream_inspector_route_local_symbol(),
381                TransportProfile::remote_stream_fabric().name().clone(),
382                status.inspector_status(),
383                0,
384                &StreamStats::default(),
385                None,
386                Vec::new(),
387            ),
388        })
389    }
390}
391
392fn server_address_from_endpoint(endpoint: &str, local: bool) -> ServerAddress {
393    if let Some(region) = endpoint.strip_prefix("wasm:") {
394        return ServerAddress::Wasm {
395            region: region.to_owned(),
396        };
397    }
398    if let Some(thread) = endpoint
399        .strip_prefix("in-process:")
400        .and_then(|value| value.parse::<u64>().ok())
401    {
402        return ServerAddress::InProcess { thread };
403    }
404    if endpoint.starts_with("ws://") || endpoint.starts_with("wss://") {
405        return ServerAddress::Ws {
406            url: endpoint.to_owned(),
407        };
408    }
409    if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
410        return ServerAddress::Http {
411            url: endpoint.to_owned(),
412        };
413    }
414    if local {
415        ServerAddress::InProcess { thread: 0 }
416    } else {
417        ServerAddress::Http {
418            url: endpoint.to_owned(),
419        }
420    }
421}
422
423fn web_session_request(expr: Expr) -> EvalRequest {
424    EvalRequest {
425        expr,
426        result_shape: None,
427        required_capabilities: Vec::new(),
428        deadline: None,
429        consistency: Consistency::LocalFirst,
430        mode: EvalMode::Eval,
431        answer_limit: None,
432        stream_buffer: None,
433        stream: false,
434        trace: false,
435    }
436}
437
438fn web_session_read(resource: &Symbol) -> Expr {
439    Expr::Map(vec![
440        (
441            Expr::Symbol(Symbol::new("op")),
442            Expr::Symbol(Symbol::qualified("web-session", "read")),
443        ),
444        (
445            Expr::Symbol(Symbol::new("resource")),
446            Expr::Symbol(resource.clone()),
447        ),
448    ])
449}
450
451fn web_session_realize(resource: &Symbol, operation: &Operation) -> Expr {
452    Expr::Map(vec![
453        (
454            Expr::Symbol(Symbol::new("op")),
455            Expr::Symbol(Symbol::qualified("web-session", "realize")),
456        ),
457        (
458            Expr::Symbol(Symbol::new("resource")),
459            Expr::Symbol(resource.clone()),
460        ),
461        (
462            Expr::Symbol(Symbol::new("operation")),
463            operation.form.clone(),
464        ),
465    ])
466}
467
468fn web_session_commit(
469    resource: &Symbol,
470    operation: &Operation,
471    expected_current: Option<&Expr>,
472) -> Expr {
473    Expr::Map(vec![
474        (
475            Expr::Symbol(Symbol::new("op")),
476            Expr::Symbol(Symbol::qualified("web-session", "commit")),
477        ),
478        (
479            Expr::Symbol(Symbol::new("resource")),
480            Expr::Symbol(resource.clone()),
481        ),
482        (
483            Expr::Symbol(Symbol::new("operation")),
484            operation.form.clone(),
485        ),
486        (
487            Expr::Symbol(Symbol::new("expected-current")),
488            expected_current.cloned().unwrap_or(Expr::Nil),
489        ),
490    ])
491}
492
493fn web_session_changes(resource: &Symbol) -> Expr {
494    Expr::Map(vec![
495        (
496            Expr::Symbol(Symbol::new("op")),
497            Expr::Symbol(Symbol::qualified("web-session", "changes")),
498        ),
499        (
500            Expr::Symbol(Symbol::new("resource")),
501            Expr::Symbol(resource.clone()),
502        ),
503    ])
504}
505
506fn parse_changes(expr: Expr) -> Result<Vec<ChangeEvent>> {
507    let Expr::List(items) = expr else {
508        return Err(Error::TypeMismatch {
509            expected: "change event list",
510            found: "non-list",
511        });
512    };
513    items
514        .into_iter()
515        .map(|item| match item {
516            Expr::Symbol(resource) => Ok(ChangeEvent { resource }),
517            Expr::Map(entries) => entries
518                .into_iter()
519                .find_map(|(key, value)| {
520                    let is_resource =
521                        matches!(key, Expr::Symbol(symbol) if symbol.name.as_ref() == "resource");
522                    match value {
523                        Expr::Symbol(resource) if is_resource => Some(Ok(ChangeEvent { resource })),
524                        _ => None,
525                    }
526                })
527                .unwrap_or_else(|| {
528                    Err(Error::HostError(
529                        "change event is missing symbol resource".to_owned(),
530                    ))
531                }),
532            _ => Err(Error::TypeMismatch {
533                expected: "change event",
534                found: "non-change",
535            }),
536        })
537        .collect()
538}
539
540fn remote_error_message(expr: &Expr) -> Option<String> {
541    let Expr::Map(entries) = expr else {
542        return None;
543    };
544    let kind = entries.iter().find_map(|(key, value)| {
545        let is_error = matches!(key, Expr::Symbol(symbol) if symbol.name.as_ref() == "error");
546        match value {
547            Expr::Symbol(symbol) if is_error => Some(symbol.as_qualified_str()),
548            Expr::String(message) if is_error => Some(message.clone()),
549            _ => None,
550        }
551    })?;
552    let message = entries
553        .iter()
554        .find_map(|(key, value)| {
555            let is_message =
556                matches!(key, Expr::Symbol(symbol) if symbol.name.as_ref() == "message");
557            match value {
558                Expr::String(message) if is_message => Some(message.clone()),
559                _ => None,
560            }
561        })
562        .unwrap_or_else(|| kind.clone());
563    Some(format!("{kind}: {message}"))
564}
sim_lib_web_bridge/remote.rs

sim_lib_web_bridge/
remote.rs
