vertep_runtime/
websocket.rs

1//! The WebSocket Event I/O Processor: the frames both ends speak and the delivery state of one
2//! peer link, shared by the browser shell (`vertep-wasm`) and the server endpoint
3//! ([`server`], feature `websocket`).
4//!
5//! Frames are JSON text messages tagged by `type`:
6//!
7//! - `hello` (client to server, first frame): the client's node IRI, the credential its
8//!   provider supplied, and the IRI prefixes of the actors it hosts.
9//! - `welcome` or `refused` (server to client): the connection is accepted, or its credential
10//!   is refused (the server then closes it).
11//! - `message`: a Vertep envelope: message id, target IRI, origin IRI, event name, payload,
12//!   send id and correlation id.
13//! - `ack`: the receiver accepted the message with that id into a mailbox.
14//!
15//! Delivery is at least once. A [`Link`] keeps every message it sent until the peer
16//! acknowledges it and sends all of them again after a reconnect; receivers deduplicate by
17//! message id (the runtime journals message ids and drops a duplicate, see
18//! [[openspec:event-journal#Message Deduplication]]).
19//!
20//! The runtime's transactional outbox holds a send until delivery succeeds: a send offered to a
21//! [`LinkMailbox`] is refused with "awaiting acknowledgement" until the peer's `ack` arrives, so
22//! the outbox retries it and its durable outbox mark passes it only once it is acknowledged.
23//! That makes the redelivery survive a crash of the sender, too.
24
25use std::collections::{BTreeMap, BTreeSet};
26use std::sync::{Arc, Mutex, MutexGuard};
27
28use futures::channel::mpsc::UnboundedSender;
29use serde::{Deserialize, Serialize};
30use vertep_core::datamodel::{Event, EventType};
31
32use crate::address::Address;
33use crate::json::JsonValue;
34use crate::mailbox::{Envelope, Mailbox, MailboxError};
35use crate::node::NodeHandle;
36use crate::transport::{Resolver, Transport, TransportError};
37
38#[cfg(feature = "websocket")]
39pub mod server;
40
41/// The type IRI of the WebSocket Event I/O Processor (alias `websocket`). Events that arrive
42/// over WebSocket carry it as `_event.origintype`.
43pub const WEBSOCKET_PROCESSOR: &str = "https://w3id.org/vertep/vt#WebSocketEventProcessor";
44
45/// The dead-letter reason of a send whose connection credential was refused.
46pub const REFUSED_REASON: &str = "transport-refused";
47
48/// A Vertep envelope on the wire.
49#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
50pub struct Message {
51    /// The message id, the receiver's deduplication key.
52    pub id: String,
53    /// The receiver's IRI.
54    pub target: String,
55    /// The sender's IRI.
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    pub origin: Option<String>,
58    /// The event name.
59    pub event: String,
60    /// The payload, in the runtime's lossless JSON encoding.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub data: Option<JsonValue>,
63    /// The send id (`_event.sendid`).
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub sendid: Option<String>,
66    /// A request's correlation id, or the one a reply answers.
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub correlation: Option<String>,
69}
70
71impl Message {
72    /// The wire form of `env`.
73    pub fn from_envelope(env: &Envelope) -> Self {
74        Message {
75            id: env.message_id.clone(),
76            target: env.to.clone(),
77            origin: env.event.origin.clone(),
78            event: env.event.name.clone(),
79            data: Some(JsonValue(env.event.data.clone())),
80            sendid: env.event.sendid.clone(),
81            correlation: env.correlation.clone(),
82        }
83    }
84
85    /// The envelope a receiver delivers: an external event whose origin type names this
86    /// processor and whose origin is the sender's IRI.
87    pub fn into_envelope(self) -> Envelope {
88        let mut event = Event::named(self.event, EventType::External);
89        event.origin = self.origin;
90        event.origintype = Some(WEBSOCKET_PROCESSOR.to_owned());
91        event.sendid = self.sendid;
92        if let Some(d) = self.data {
93            event.data = d.0;
94        }
95        Envelope {
96            message_id: self.id,
97            to: self.target,
98            event,
99            correlation: self.correlation,
100            timer: None,
101        }
102    }
103}
104
105/// One frame of the protocol.
106#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
107#[serde(tag = "type", rename_all = "lowercase")]
108pub enum Frame {
109    /// The client's first frame.
110    Hello {
111        /// The client node's IRI, stable for the life of its actor system.
112        node: String,
113        /// The credential the client's provider supplied.
114        #[serde(default, skip_serializing_if = "Option::is_none")]
115        credential: Option<String>,
116        /// IRI prefixes of the actors the client hosts.
117        #[serde(default)]
118        hosts: Vec<String>,
119    },
120    /// The server accepted the connection.
121    Welcome {
122        /// The server node's IRI.
123        node: String,
124        /// IRI prefixes of the actors the server hosts.
125        #[serde(default)]
126        hosts: Vec<String>,
127    },
128    /// The server refused the credential and closes the connection.
129    Refused {
130        /// Why.
131        reason: String,
132    },
133    /// A message.
134    Message(Message),
135    /// The receiver accepted message `id`.
136    Ack {
137        /// The acknowledged message id.
138        id: String,
139    },
140}
141
142impl Frame {
143    /// The frame as JSON text.
144    pub fn to_text(&self) -> String {
145        serde_json::to_string(self).unwrap_or_default()
146    }
147
148    /// Parses a frame.
149    ///
150    /// # Errors
151    ///
152    /// A message for text that is not a frame.
153    pub fn parse(text: &str) -> Result<Frame, String> {
154        serde_json::from_str(text).map_err(|e| format!("not a Vertep WebSocket frame: {e}"))
155    }
156}
157
158/// What became of a message offered to a [`Link`].
159#[derive(Clone, Debug, PartialEq, Eq)]
160pub enum Offer {
161    /// The peer acknowledged it.
162    Delivered,
163    /// It is sent (or waits for a connection) and not acknowledged yet.
164    Pending,
165    /// The peer refused the connection credential.
166    Refused(String),
167}
168
169/// What the receiving side of a [`Link`] should do with a frame.
170#[derive(Clone, Debug, PartialEq)]
171pub enum Inbound {
172    /// Deliver this envelope, then [`Link::ack`] it.
173    Deliver(Envelope),
174    /// The peer acknowledged message `id`; `watched` when [`Link::watch`] asked to hear of it.
175    Acked {
176        /// The message id.
177        id: String,
178        /// Whether someone waits for this acknowledgement.
179        watched: bool,
180    },
181    /// A handshake frame (`hello`, `welcome`); the connection code handles it.
182    Handshake(Frame),
183    /// The credential was refused: the watched messages that were pending, now failed.
184    Refused {
185        /// Why.
186        reason: String,
187        /// The watched message ids that were pending.
188        failed: Vec<String>,
189    },
190}
191
192#[derive(Debug)]
193struct Pending {
194    env: Envelope,
195    sent_on: Option<u64>,
196    order: u64,
197}
198
199#[derive(Default)]
200struct State {
201    connection: Option<(u64, UnboundedSender<String>)>,
202    pending: BTreeMap<String, Pending>,
203    acked: BTreeSet<String>,
204    watched: BTreeSet<String>,
205    settled: Vec<(String, Result<(), String>)>,
206    refused: Option<String>,
207    next_order: u64,
208    routes: BTreeSet<String>,
209}
210
211/// The delivery state of one peer: the unacknowledged messages, the current connection, and
212/// the acknowledgements received. It outlives connections, so a reconnect sends every pending
213/// message again. It is `Send + Sync`; the I/O lives in the connection code, which hands the
214/// link a channel of outgoing frame texts on connect.
215// @tg: implements:: [[openspec:wasm-bindings#Server Interop Over WebSocket]] {confidence: partial, note: "envelopes, acknowledgements and redelivery, shared by both ends"}
216#[derive(Default)]
217pub struct Link {
218    state: Mutex<State>,
219}
220
221impl std::fmt::Debug for Link {
222    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
223        let s = self.lock();
224        f.debug_struct("Link")
225            .field("connected", &s.connection.as_ref().map(|c| c.0))
226            .field("pending", &s.pending.len())
227            .field("refused", &s.refused)
228            .finish()
229    }
230}
231
232impl Link {
233    /// A link with no connection yet.
234    pub fn new() -> Self {
235        Link::default()
236    }
237
238    fn lock(&self) -> MutexGuard<'_, State> {
239        self.state
240            .lock()
241            .unwrap_or_else(std::sync::PoisonError::into_inner)
242    }
243
244    fn send_pending(s: &mut State, id: &str) {
245        let Some((generation, tx)) = s.connection.clone() else {
246            return;
247        };
248        if let Some(p) = s.pending.get_mut(id)
249            && p.sent_on != Some(generation)
250        {
251            p.sent_on = Some(generation);
252            let _ = tx.unbounded_send(Frame::Message(Message::from_envelope(&p.env)).to_text());
253        }
254    }
255
256    /// Offers `env` for delivery: sends it if connected and not yet sent on this connection,
257    /// and reports whether the peer has acknowledged it. Offering the same message id again is
258    /// how the outbox retries; it is sent once per connection.
259    pub fn offer(&self, env: Envelope) -> Offer {
260        let mut s = self.lock();
261        if s.acked.remove(&env.message_id) {
262            return Offer::Delivered;
263        }
264        if let Some(r) = &s.refused {
265            return Offer::Refused(r.clone());
266        }
267        let id = env.message_id.clone();
268        if !s.pending.contains_key(&id) {
269            let order = s.next_order;
270            s.next_order += 1;
271            s.pending.insert(
272                id.clone(),
273                Pending {
274                    env,
275                    sent_on: None,
276                    order,
277                },
278            );
279        }
280        Link::send_pending(&mut s, &id);
281        Offer::Pending
282    }
283
284    /// Asks to hear of the acknowledgement (or refusal) of message `id` through
285    /// [`Link::take_settled`] instead of a later [`Link::offer`].
286    pub fn watch(&self, id: &str) {
287        let mut s = self.lock();
288        if s.acked.remove(id) {
289            s.settled.push((id.to_owned(), Ok(())));
290        } else {
291            s.watched.insert(id.to_owned());
292        }
293    }
294
295    /// The watched messages acknowledged or failed since the last call.
296    pub fn take_settled(&self) -> Vec<(String, Result<(), String>)> {
297        std::mem::take(&mut self.lock().settled)
298    }
299
300    /// Whether message `id` waits for an acknowledgement.
301    pub fn is_pending(&self, id: &str) -> bool {
302        self.lock().pending.contains_key(id)
303    }
304
305    /// The number of messages waiting for an acknowledgement.
306    pub fn pending_len(&self) -> usize {
307        self.lock().pending.len()
308    }
309
310    /// A connection with number `generation` is open (and accepted): its writer takes the
311    /// frame texts sent into `tx`. Every pending message is sent again, oldest first, and a
312    /// refusal is forgotten.
313    pub fn connected(&self, generation: u64, tx: UnboundedSender<String>) {
314        let mut s = self.lock();
315        s.connection = Some((generation, tx));
316        s.refused = None;
317        let mut ids: Vec<(u64, String)> = s
318            .pending
319            .iter()
320            .map(|(k, p)| (p.order, k.clone()))
321            .collect();
322        ids.sort();
323        for (_, id) in ids {
324            Link::send_pending(&mut s, &id);
325        }
326    }
327
328    /// The connection `generation` closed (a later connection is left alone).
329    pub fn disconnected(&self, generation: u64) {
330        let mut s = self.lock();
331        if s.connection.as_ref().is_some_and(|c| c.0 == generation) {
332            s.connection = None;
333        }
334    }
335
336    /// Whether a connection is open.
337    pub fn is_connected(&self) -> bool {
338        self.lock().connection.is_some()
339    }
340
341    /// Sends an acknowledgement of message `id` on the current connection.
342    pub fn ack(&self, id: &str) {
343        let s = self.lock();
344        if let Some((_, tx)) = &s.connection {
345            let _ = tx.unbounded_send(Frame::Ack { id: id.to_owned() }.to_text());
346        }
347    }
348
349    /// Sends a frame on the current connection; false when there is none.
350    pub fn send_frame(&self, f: &Frame) -> bool {
351        let s = self.lock();
352        match &s.connection {
353            Some((_, tx)) => tx.unbounded_send(f.to_text()).is_ok(),
354            None => false,
355        }
356    }
357
358    /// The credential was refused: watched pending messages fail, the others are dropped from
359    /// the link (the outbox offers them again and learns of the refusal then). The refusal
360    /// holds until the next accepted connection.
361    pub fn refused(&self, reason: &str) -> Vec<String> {
362        let mut s = self.lock();
363        s.refused = Some(reason.to_owned());
364        let ids: Vec<String> = s.pending.keys().cloned().collect();
365        s.pending.clear();
366        let mut failed = Vec::new();
367        for id in ids {
368            if s.watched.remove(&id) {
369                s.settled.push((id.clone(), Err(reason.to_owned())));
370                failed.push(id);
371            }
372        }
373        failed
374    }
375
376    /// Whether the peer refused the credential (and no connection was accepted since).
377    pub fn refusal(&self) -> Option<String> {
378        self.lock().refused.clone()
379    }
380
381    /// Learns that actor `iri` is reachable through this link (the origin of a message it
382    /// sent).
383    pub fn learn_route(&self, iri: &str) {
384        self.lock().routes.insert(iri.to_owned());
385    }
386
387    /// Whether actor `iri` is known to be reachable through this link.
388    pub fn routes_to(&self, iri: &str) -> bool {
389        self.lock().routes.contains(iri)
390    }
391
392    /// Handles one received frame text.
393    ///
394    /// # Errors
395    ///
396    /// A message for text that is not a frame.
397    pub fn receive(&self, text: &str) -> Result<Inbound, String> {
398        Ok(match Frame::parse(text)? {
399            Frame::Message(m) => {
400                if let Some(o) = &m.origin {
401                    self.learn_route(o);
402                }
403                Inbound::Deliver(m.into_envelope())
404            }
405            Frame::Ack { id } => {
406                let mut s = self.lock();
407                let known = s.pending.remove(&id).is_some();
408                let watched = s.watched.remove(&id);
409                if watched {
410                    s.settled.push((id.clone(), Ok(())));
411                } else if known {
412                    s.acked.insert(id.clone());
413                }
414                Inbound::Acked { id, watched }
415            }
416            Frame::Refused { reason } => {
417                let failed = self.refused(&reason);
418                Inbound::Refused { reason, failed }
419            }
420            f @ (Frame::Hello { .. } | Frame::Welcome { .. }) => Inbound::Handshake(f),
421        })
422    }
423}
424
425/// The runtime's view of an actor reached through a [`Link`]: a mailbox that offers each
426/// message to the link. It accepts a message once the peer acknowledged it; until then it
427/// reports the delivery as failed, so the outbox keeps the send and retries it. A send refused
428/// by the peer's credential check is recorded as a dead letter (reason `transport-refused`)
429/// and settled.
430pub struct LinkMailbox {
431    link: Arc<Link>,
432    handle: Option<NodeHandle>,
433}
434
435impl std::fmt::Debug for LinkMailbox {
436    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
437        f.debug_struct("LinkMailbox")
438            .field("link", &self.link)
439            .finish_non_exhaustive()
440    }
441}
442
443impl LinkMailbox {
444    /// A mailbox over `link`; `handle` records dead letters of refused sends.
445    pub fn new(link: Arc<Link>, handle: Option<NodeHandle>) -> Self {
446        LinkMailbox { link, handle }
447    }
448}
449
450impl Mailbox for LinkMailbox {
451    fn enqueue(&self, env: Envelope) -> Result<(), MailboxError> {
452        let (to, copy) = (env.to.clone(), env.clone());
453        match self.link.offer(env) {
454            Offer::Delivered => Ok(()),
455            Offer::Pending => Err(MailboxError(
456                "awaiting the WebSocket peer's acknowledgement".into(),
457            )),
458            Offer::Refused(_) => {
459                if let (Some(h), Ok(a)) = (&self.handle, Address::parse(&to)) {
460                    h.record_dead_letter(&a, &copy, REFUSED_REASON);
461                }
462                Ok(())
463            }
464        }
465    }
466
467    fn receive(&self, _deferred: &[String]) -> Option<Envelope> {
468        None
469    }
470
471    fn ack(&self, _message_id: &str) -> Result<(), MailboxError> {
472        Ok(())
473    }
474
475    fn release(&self, _message_id: &str) {}
476
477    fn release_all(&self) {}
478
479    fn pending(&self) -> Vec<Envelope> {
480        Vec::new()
481    }
482
483    fn pending_deliverable(&self, _deferred: &[String]) -> usize {
484        0
485    }
486}
487
488/// Resolves the addresses a client sends to over its one [`Link`] to the server: every
489/// address (of the schemes it is registered for) that starts with one of `targets`, or every
490/// address when `targets` is empty. Addresses hosted locally never reach a resolver.
491pub struct ClientResolver {
492    link: Arc<Link>,
493    handle: Option<NodeHandle>,
494    targets: Vec<String>,
495}
496
497impl std::fmt::Debug for ClientResolver {
498    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
499        f.debug_struct("ClientResolver")
500            .field("targets", &self.targets)
501            .finish_non_exhaustive()
502    }
503}
504
505impl ClientResolver {
506    /// A resolver sending the addresses under `targets` over `link`.
507    pub fn new(link: Arc<Link>, handle: Option<NodeHandle>, targets: Vec<String>) -> Self {
508        ClientResolver {
509            link,
510            handle,
511            targets,
512        }
513    }
514
515    /// Whether `address` goes over the link.
516    pub fn claims(&self, address: &str) -> bool {
517        self.targets.is_empty() || self.targets.iter().any(|t| address.starts_with(t.as_str()))
518    }
519}
520
521impl Resolver for ClientResolver {
522    fn resolve(&self, address: &Address) -> Option<Arc<dyn Mailbox>> {
523        self.claims(address.as_str()).then(|| {
524            Arc::new(LinkMailbox::new(self.link.clone(), self.handle.clone())) as Arc<dyn Mailbox>
525        })
526    }
527}
528
529/// The WebSocket Event I/O Processor as a runtime transport (`<send type="websocket">` or its
530/// type IRI). It delivers into the mailbox the address resolved to, which for a remote actor
531/// is a [`LinkMailbox`]; `_ioprocessors` lists it.
532#[derive(Clone, Copy, Debug, Default)]
533pub struct WebSocketTransport;
534
535impl Transport for WebSocketTransport {
536    fn type_iri(&self) -> &str {
537        WEBSOCKET_PROCESSOR
538    }
539
540    fn aliases(&self) -> Vec<String> {
541        vec!["websocket".to_owned()]
542    }
543
544    fn deliver(&self, mailbox: &dyn Mailbox, env: Envelope) -> Result<(), TransportError> {
545        mailbox
546            .enqueue(env)
547            .map_err(|e| TransportError(e.to_string()))
548    }
549}
550
551#[cfg(test)]
552mod tests {
553    use super::*;
554    use futures::channel::mpsc::unbounded;
555    use vertep_core::datamodel::Value;
556
557    fn env(id: &str) -> Envelope {
558        let mut e = Event::named("ping", EventType::External);
559        e.origin = Some("https://client.example/a/1".into());
560        e.data = Value::Object(vec![("n".into(), Value::Number(1.0))]);
561        Envelope::new(
562            id,
563            &Address::parse("https://server.example/s/1").unwrap(),
564            e,
565        )
566    }
567
568    fn drain(rx: &mut futures::channel::mpsc::UnboundedReceiver<String>) -> Vec<Frame> {
569        let mut out = Vec::new();
570        while let Ok(t) = rx.try_recv() {
571            out.push(Frame::parse(&t).unwrap());
572        }
573        out
574    }
575
576    #[test]
577    fn frames_round_trip() {
578        let m = Message::from_envelope(&env("m1"));
579        let f = Frame::Message(m.clone());
580        assert_eq!(Frame::parse(&f.to_text()).unwrap(), f);
581        let delivered = m.into_envelope();
582        assert_eq!(
583            delivered.event.origintype.as_deref(),
584            Some(WEBSOCKET_PROCESSOR)
585        );
586        assert_eq!(
587            delivered.event.origin.as_deref(),
588            Some("https://client.example/a/1")
589        );
590        assert_eq!(delivered.event.data, env("m1").event.data);
591        assert!(Frame::parse(r#"{"type":"ack","id":"x"}"#).is_ok());
592        assert!(Frame::parse("nope").is_err());
593    }
594
595    #[test]
596    fn pending_until_acked_and_resent_after_reconnect() {
597        let link = Link::new();
598        assert_eq!(link.offer(env("m1")), Offer::Pending, "no connection yet");
599        let (tx, mut rx) = unbounded();
600        link.connected(1, tx);
601        assert_eq!(drain(&mut rx).len(), 1, "sent on connect");
602        assert_eq!(link.offer(env("m1")), Offer::Pending);
603        assert!(
604            drain(&mut rx).is_empty(),
605            "a retry does not resend on the same connection"
606        );
607        link.disconnected(1);
608        let (tx, mut rx) = unbounded();
609        link.connected(2, tx);
610        assert_eq!(drain(&mut rx).len(), 1, "resent after reconnect");
611        link.receive(r#"{"type":"ack","id":"m1"}"#).unwrap();
612        assert_eq!(link.offer(env("m1")), Offer::Delivered);
613        assert_eq!(link.pending_len(), 0);
614    }
615
616    #[test]
617    fn watched_messages_settle_and_fail_on_refusal() {
618        let link = Link::new();
619        let (tx, _rx) = unbounded();
620        link.connected(1, tx);
621        link.offer(env("a"));
622        link.watch("a");
623        link.offer(env("b"));
624        link.watch("b");
625        link.receive(r#"{"type":"ack","id":"a"}"#).unwrap();
626        assert_eq!(link.take_settled(), vec![("a".into(), Ok(()))]);
627        let r = link
628            .receive(r#"{"type":"refused","reason":"bad token"}"#)
629            .unwrap();
630        assert_eq!(
631            r,
632            Inbound::Refused {
633                reason: "bad token".into(),
634                failed: vec!["b".into()]
635            }
636        );
637        assert_eq!(
638            link.take_settled(),
639            vec![("b".into(), Err("bad token".into()))]
640        );
641        assert_eq!(link.offer(env("c")), Offer::Refused("bad token".into()));
642        let (tx, mut rx) = unbounded();
643        link.connected(2, tx);
644        assert_eq!(link.offer(env("c")), Offer::Pending, "accepted again");
645        assert!(rx.try_recv().is_ok(), "and sent");
646    }
647
648    #[test]
649    fn received_messages_teach_routes() {
650        let link = Link::new();
651        let text = Frame::Message(Message::from_envelope(&env("m"))).to_text();
652        assert!(matches!(link.receive(&text).unwrap(), Inbound::Deliver(_)));
653        assert!(link.routes_to("https://client.example/a/1"));
654    }
655}
vertep_runtime/websocket.rs

vertep_runtime/
websocket.rs
