phoenix_channel_runtime/
protocol.rs

1use std::collections::HashMap;
2
3use serde_json::{Value, json};
4use thiserror::Error;
5
6use crate::{Frame, Payload};
7
8const PHOENIX_TOPIC: &str = "phoenix";
9
10/// Local lifecycle state for a channel topic.
11#[derive(Clone, Copy, Debug, Eq, PartialEq)]
12pub enum ChannelState {
13    /// A `phx_join` reply is pending.
14    Joining,
15    /// The server accepted the join.
16    Joined,
17    /// A `phx_leave` reply is pending.
18    Leaving,
19    /// The server closed the channel.
20    Closed,
21    /// The server reported a channel error.
22    Errored,
23    /// The transport disconnected while the channel was active.
24    Disconnected,
25}
26
27/// Status carried by a Phoenix `phx_reply`.
28#[derive(Clone, Copy, Debug, Eq, PartialEq)]
29pub enum ReplyStatus {
30    /// The server accepted the operation.
31    Ok,
32    /// The server rejected the operation.
33    Error,
34}
35
36/// An outbound frame and the reference allocated to correlate its reply.
37#[derive(Clone, Debug, PartialEq)]
38pub struct Outbound {
39    /// Unique request reference.
40    pub reference: String,
41    /// Frame to encode and send.
42    pub frame: Frame,
43}
44
45/// Semantic event produced from an incoming Phoenix frame or connection reset.
46#[derive(Clone, Debug, PartialEq)]
47pub enum ProtocolEvent {
48    /// A topic join succeeded.
49    Joined {
50        /// Joined topic.
51        topic: String,
52        /// Correlated request reference.
53        reference: String,
54        /// Server join response.
55        response: Payload,
56    },
57    /// A topic join was rejected.
58    JoinError {
59        /// Rejected topic.
60        topic: String,
61        /// Correlated request reference.
62        reference: String,
63        /// Server error response.
64        response: Payload,
65    },
66    /// A topic leave completed.
67    Left {
68        /// Left topic.
69        topic: String,
70        /// Correlated request reference.
71        reference: String,
72        /// Server leave response.
73        response: Payload,
74    },
75    /// An application push received a reply.
76    Reply {
77        /// Topic to which the push was sent.
78        topic: String,
79        /// Application event name.
80        event: String,
81        /// Correlated request reference.
82        reference: String,
83        /// Phoenix reply status.
84        status: ReplyStatus,
85        /// Server response payload.
86        response: Payload,
87    },
88    /// An application or Presence message that was not a correlated reply.
89    Message(Frame),
90    /// The server sent `phx_close` for a topic.
91    ChannelClosed {
92        /// Closed topic.
93        topic: String,
94        /// Close payload.
95        payload: Payload,
96    },
97    /// The server sent `phx_error` for a topic.
98    ChannelError {
99        /// Errored topic.
100        topic: String,
101        /// Error payload.
102        payload: Payload,
103    },
104    /// The server acknowledged a heartbeat.
105    HeartbeatAck {
106        /// Correlated heartbeat reference.
107        reference: String,
108        /// Heartbeat reply status.
109        status: ReplyStatus,
110    },
111    /// A pending request was interrupted by transport reset.
112    RequestInterrupted {
113        /// Topic on which the request was active.
114        topic: String,
115        /// Join, leave, or application event name.
116        event: String,
117        /// Interrupted request reference.
118        reference: String,
119    },
120    /// A frame belonged to an earlier join generation.
121    StaleMessage(Frame),
122    /// A reply had no matching pending operation.
123    UnmatchedReply(Frame),
124}
125
126#[derive(Clone, Debug)]
127struct Channel {
128    state: ChannelState,
129    join_ref: String,
130    params: Value,
131}
132
133#[derive(Clone, Debug)]
134enum Pending {
135    Join { topic: String },
136    Leave { topic: String },
137    Push { topic: String, event: String },
138    Heartbeat,
139}
140
141/// Pure Phoenix Channels protocol state.
142///
143/// The caller owns I/O, clocks, authentication refresh, and retry scheduling.
144/// That makes the same state machine usable from browser WASM, Tokio, smol, or
145/// a UI framework executor.
146#[derive(Debug, Default)]
147pub struct Protocol {
148    next_reference: u64,
149    channels: HashMap<String, Channel>,
150    pending: HashMap<String, Pending>,
151}
152
153impl Protocol {
154    /// Creates empty protocol state.
155    pub fn new() -> Self {
156        Self::default()
157    }
158
159    /// Returns the local state of `topic`, if it is known.
160    pub fn channel_state(&self, topic: &str) -> Option<ChannelState> {
161        self.channels.get(topic).map(|channel| channel.state)
162    }
163
164    /// Starts joining a topic with the supplied parameters.
165    ///
166    /// The returned frame must be sent and later passed back through
167    /// [`Protocol::receive`] when its reply arrives.
168    pub fn join(
169        &mut self,
170        topic: impl Into<String>,
171        params: Value,
172    ) -> Result<Outbound, ProtocolError> {
173        let topic = topic.into();
174        if self.channels.get(&topic).is_some_and(|channel| {
175            matches!(
176                channel.state,
177                ChannelState::Joining | ChannelState::Joined | ChannelState::Leaving
178            )
179        }) {
180            return Err(ProtocolError::AlreadyActive(topic));
181        }
182
183        let reference = self.allocate_reference();
184        self.channels.insert(
185            topic.clone(),
186            Channel {
187                state: ChannelState::Joining,
188                join_ref: reference.clone(),
189                params: params.clone(),
190            },
191        );
192        self.pending.insert(
193            reference.clone(),
194            Pending::Join {
195                topic: topic.clone(),
196            },
197        );
198
199        Ok(Outbound {
200            reference: reference.clone(),
201            frame: Frame::new(
202                Some(reference.clone()),
203                Some(reference),
204                topic,
205                "phx_join",
206                params,
207            ),
208        })
209    }
210
211    /// Starts a new join generation for a disconnected, errored, or closed topic.
212    pub fn rejoin(
213        &mut self,
214        topic: impl Into<String>,
215        refreshed_params: Value,
216    ) -> Result<Outbound, ProtocolError> {
217        let topic = topic.into();
218        match self.channels.get(&topic).map(|channel| channel.state) {
219            Some(ChannelState::Disconnected | ChannelState::Errored | ChannelState::Closed) => {
220                self.discard_channel(&topic);
221                self.join(topic, refreshed_params)
222            }
223            Some(_) => Err(ProtocolError::AlreadyActive(topic)),
224            None => self.join(topic, refreshed_params),
225        }
226    }
227
228    /// Rejoins every inactive topic using the parameters stored by its prior join.
229    pub fn rejoin_all_with_stored_params(&mut self) -> Vec<Outbound> {
230        let channels = self
231            .channels
232            .iter()
233            .filter(|(_, channel)| {
234                matches!(
235                    channel.state,
236                    ChannelState::Disconnected | ChannelState::Errored | ChannelState::Closed
237                )
238            })
239            .map(|(topic, channel)| (topic.clone(), channel.params.clone()))
240            .collect::<Vec<_>>();
241
242        channels
243            .into_iter()
244            .filter_map(|(topic, params)| self.rejoin(topic, params).ok())
245            .collect()
246    }
247
248    /// Starts leaving a joined topic.
249    pub fn leave(&mut self, topic: &str) -> Result<Outbound, ProtocolError> {
250        let reference = self.allocate_reference();
251        let channel = self
252            .channels
253            .get_mut(topic)
254            .ok_or_else(|| ProtocolError::UnknownTopic(topic.to_owned()))?;
255        if channel.state != ChannelState::Joined {
256            return Err(ProtocolError::NotJoined(topic.to_owned()));
257        }
258
259        channel.state = ChannelState::Leaving;
260        let join_ref = channel.join_ref.clone();
261        self.pending.insert(
262            reference.clone(),
263            Pending::Leave {
264                topic: topic.to_owned(),
265            },
266        );
267
268        Ok(Outbound {
269            reference: reference.clone(),
270            frame: Frame::new(
271                Some(join_ref),
272                Some(reference.clone()),
273                topic,
274                "phx_leave",
275                json!({}),
276            ),
277        })
278    }
279
280    /// Builds and tracks an application push that expects a reply.
281    pub fn push(
282        &mut self,
283        topic: &str,
284        event: impl Into<String>,
285        payload: impl Into<Payload>,
286    ) -> Result<Outbound, ProtocolError> {
287        let event = event.into();
288        let join_ref = self
289            .channels
290            .get(topic)
291            .filter(|channel| channel.state == ChannelState::Joined)
292            .map(|channel| channel.join_ref.clone())
293            .ok_or_else(|| ProtocolError::NotJoined(topic.to_owned()))?;
294        let reference = self.allocate_reference();
295        self.pending.insert(
296            reference.clone(),
297            Pending::Push {
298                topic: topic.to_owned(),
299                event: event.clone(),
300            },
301        );
302
303        Ok(Outbound {
304            reference: reference.clone(),
305            frame: Frame::new(
306                Some(join_ref),
307                Some(reference.clone()),
308                topic,
309                event,
310                payload,
311            ),
312        })
313    }
314
315    /// Builds a push that does not request a reply.
316    ///
317    /// The frame has no `ref`, so it is not tracked as an in-flight request.
318    pub fn cast(
319        &self,
320        topic: &str,
321        event: impl Into<String>,
322        payload: impl Into<Payload>,
323    ) -> Result<Frame, ProtocolError> {
324        let join_ref = self
325            .channels
326            .get(topic)
327            .filter(|channel| channel.state == ChannelState::Joined)
328            .map(|channel| channel.join_ref.clone())
329            .ok_or_else(|| ProtocolError::NotJoined(topic.to_owned()))?;
330
331        Ok(Frame::new(Some(join_ref), None, topic, event, payload))
332    }
333
334    /// Stops correlating a push reply, for example after an API timeout.
335    ///
336    /// Join and leave requests cannot be forgotten through this method because
337    /// their replies also transition channel state.
338    pub fn forget_push(&mut self, reference: &str) -> bool {
339        if matches!(self.pending.get(reference), Some(Pending::Push { .. })) {
340            self.pending.remove(reference);
341            true
342        } else {
343            false
344        }
345    }
346
347    /// Stops correlating a heartbeat reply, for example after an explicit ping
348    /// has timed out at the client API boundary.
349    pub fn forget_heartbeat(&mut self, reference: &str) -> bool {
350        if matches!(self.pending.get(reference), Some(Pending::Heartbeat)) {
351            self.pending.remove(reference);
352            true
353        } else {
354            false
355        }
356    }
357
358    /// Removes all local state associated with a topic.
359    ///
360    /// This is used when a channel handle is dropped. Any later replies for
361    /// the removed join generation are treated as unmatched messages.
362    pub fn discard_channel(&mut self, topic: &str) -> bool {
363        let removed = self.channels.remove(topic).is_some();
364        self.pending.retain(|_, pending| match pending {
365            Pending::Join {
366                topic: pending_topic,
367            }
368            | Pending::Leave {
369                topic: pending_topic,
370            }
371            | Pending::Push {
372                topic: pending_topic,
373                ..
374            } => pending_topic != topic,
375            Pending::Heartbeat => true,
376        });
377        removed
378    }
379
380    /// Builds and tracks a Phoenix heartbeat.
381    pub fn heartbeat(&mut self) -> Outbound {
382        let reference = self.allocate_reference();
383        self.pending.insert(reference.clone(), Pending::Heartbeat);
384        Outbound {
385            reference: reference.clone(),
386            frame: Frame::new(
387                None,
388                Some(reference.clone()),
389                PHOENIX_TOPIC,
390                "heartbeat",
391                json!({}),
392            ),
393        }
394    }
395
396    /// Applies an incoming frame and returns its semantic protocol event.
397    pub fn receive(&mut self, frame: Frame) -> Result<ProtocolEvent, ProtocolError> {
398        if self.channels.get(&frame.topic).is_some_and(|channel| {
399            frame.join_ref.is_some() && frame.join_ref.as_deref() != Some(channel.join_ref.as_str())
400        }) {
401            return Ok(ProtocolEvent::StaleMessage(frame));
402        }
403
404        if frame.event == "phx_reply" {
405            return self.receive_reply(frame);
406        }
407
408        match frame.event.as_str() {
409            "phx_close" => {
410                if let Some(channel) = self.channels.get_mut(&frame.topic) {
411                    channel.state = ChannelState::Closed;
412                }
413                Ok(ProtocolEvent::ChannelClosed {
414                    topic: frame.topic,
415                    payload: frame.payload,
416                })
417            }
418            "phx_error" => {
419                if let Some(channel) = self.channels.get_mut(&frame.topic) {
420                    channel.state = ChannelState::Errored;
421                }
422                Ok(ProtocolEvent::ChannelError {
423                    topic: frame.topic,
424                    payload: frame.payload,
425                })
426            }
427            _ => Ok(ProtocolEvent::Message(frame)),
428        }
429    }
430
431    /// Marks active channels disconnected and returns interrupted requests.
432    pub fn reset_connection(&mut self) -> Vec<ProtocolEvent> {
433        let interrupted = self
434            .pending
435            .drain()
436            .filter_map(|(reference, pending)| match pending {
437                Pending::Join { topic } => Some(ProtocolEvent::RequestInterrupted {
438                    topic,
439                    event: "phx_join".into(),
440                    reference,
441                }),
442                Pending::Leave { topic } => Some(ProtocolEvent::RequestInterrupted {
443                    topic,
444                    event: "phx_leave".into(),
445                    reference,
446                }),
447                Pending::Push { topic, event } => Some(ProtocolEvent::RequestInterrupted {
448                    topic,
449                    event,
450                    reference,
451                }),
452                Pending::Heartbeat => None,
453            })
454            .collect();
455
456        for channel in self.channels.values_mut() {
457            channel.state = ChannelState::Disconnected;
458        }
459        interrupted
460    }
461
462    fn receive_reply(&mut self, frame: Frame) -> Result<ProtocolEvent, ProtocolError> {
463        let Some(reference) = frame.reference.clone() else {
464            return Ok(ProtocolEvent::UnmatchedReply(frame));
465        };
466        let Some(pending) = self.pending.remove(&reference) else {
467            return Ok(ProtocolEvent::UnmatchedReply(frame));
468        };
469        let (status, response) = decode_reply_payload(&frame.payload)?;
470
471        match pending {
472            Pending::Join { topic } => {
473                if let Some(channel) = self.channels.get_mut(&topic) {
474                    channel.state = match status {
475                        ReplyStatus::Ok => ChannelState::Joined,
476                        ReplyStatus::Error => ChannelState::Errored,
477                    };
478                }
479                match status {
480                    ReplyStatus::Ok => Ok(ProtocolEvent::Joined {
481                        topic,
482                        reference,
483                        response,
484                    }),
485                    ReplyStatus::Error => Ok(ProtocolEvent::JoinError {
486                        topic,
487                        reference,
488                        response,
489                    }),
490                }
491            }
492            Pending::Leave { topic } => {
493                self.channels.remove(&topic);
494                Ok(ProtocolEvent::Left {
495                    topic,
496                    reference,
497                    response,
498                })
499            }
500            Pending::Push { topic, event } => Ok(ProtocolEvent::Reply {
501                topic,
502                event,
503                reference,
504                status,
505                response,
506            }),
507            Pending::Heartbeat => Ok(ProtocolEvent::HeartbeatAck { reference, status }),
508        }
509    }
510
511    fn allocate_reference(&mut self) -> String {
512        self.next_reference = self.next_reference.wrapping_add(1);
513        if self.next_reference == 0 {
514            self.next_reference = 1;
515        }
516        self.next_reference.to_string()
517    }
518}
519
520fn decode_reply_payload(payload: &Payload) -> Result<(ReplyStatus, Payload), ProtocolError> {
521    match payload {
522        Payload::Json(value) => {
523            let object = value
524                .as_object()
525                .ok_or(ProtocolError::InvalidReplyPayload)?;
526            let status = decode_reply_status(object.get("status").and_then(Value::as_str))?;
527            let response = object.get("response").cloned().unwrap_or_else(|| json!({}));
528            Ok((status, Payload::Json(response)))
529        }
530        Payload::Reply { status, response } => {
531            Ok((decode_reply_status(Some(status))?, (**response).clone()))
532        }
533        Payload::Binary(_) => Err(ProtocolError::InvalidReplyPayload),
534    }
535}
536
537fn decode_reply_status(status: Option<&str>) -> Result<ReplyStatus, ProtocolError> {
538    match status {
539        Some("ok") => Ok(ReplyStatus::Ok),
540        Some("error") => Ok(ReplyStatus::Error),
541        _ => Err(ProtocolError::InvalidReplyStatus),
542    }
543}
544
545/// Invalid operation or reply for the current protocol state.
546#[derive(Debug, Error, PartialEq)]
547pub enum ProtocolError {
548    /// A join was requested for a topic that is already active.
549    #[error("topic is already active: {0}")]
550    AlreadyActive(String),
551    /// An operation required a joined topic.
552    #[error("topic is not joined: {0}")]
553    NotJoined(String),
554    /// An operation referred to a topic absent from protocol state.
555    #[error("unknown topic: {0}")]
556    UnknownTopic(String),
557    /// A `phx_reply` payload did not have the expected reply envelope.
558    #[error("phx_reply payload must be an object")]
559    InvalidReplyPayload,
560    /// A `phx_reply` status was neither `ok` nor `error`.
561    #[error("phx_reply status must be ok or error")]
562    InvalidReplyStatus,
563}
564
565#[cfg(test)]
566mod tests {
567    use proptest::prelude::*;
568    use serde_json::json;
569
570    use super::*;
571
572    fn reply(outbound: &Outbound, status: &str, response: Value) -> Frame {
573        Frame::new(
574            outbound.frame.join_ref.clone(),
575            Some(outbound.reference.clone()),
576            outbound.frame.topic.clone(),
577            "phx_reply",
578            json!({"status": status, "response": response}),
579        )
580    }
581
582    fn joined_protocol() -> Protocol {
583        let mut protocol = Protocol::new();
584        let join = protocol
585            .join("room:lobby", json!({"token": "abc"}))
586            .unwrap();
587        protocol.receive(reply(&join, "ok", json!({}))).unwrap();
588        protocol
589    }
590
591    proptest! {
592        #[test]
593        fn arbitrary_state_transitions_never_panic(actions in proptest::collection::vec(any::<u8>(), 0..512)) {
594            let mut protocol = Protocol::new();
595            for action in actions {
596                match action % 7 {
597                    0 => { let _ = protocol.join("room:lobby", json!({})); }
598                    1 => {
599                        let frame = Frame::new(
600                            Some((action / 7).to_string()),
601                            Some(action.to_string()),
602                            "room:lobby",
603                            if action & 1 == 0 { "phx_error" } else { "event" },
604                            json!({}),
605                        );
606                        let _ = protocol.receive(frame);
607                    }
608                    2 => { let _ = protocol.reset_connection(); }
609                    3 => { let _ = protocol.discard_channel("room:lobby"); }
610                    4 => { let _ = protocol.heartbeat(); }
611                    5 => { let _ = protocol.leave("room:lobby"); }
612                    _ => { let _ = protocol.push("room:lobby", "event", json!({})); }
613                }
614            }
615        }
616    }
617
618    #[test]
619    fn joins_a_topic_and_correlates_the_reply() {
620        let mut protocol = Protocol::new();
621        let outbound = protocol
622            .join("room:lobby", json!({"token": "abc"}))
623            .unwrap();
624
625        assert_eq!(outbound.frame.event, "phx_join");
626        assert_eq!(outbound.frame.join_ref, outbound.frame.reference);
627        assert_eq!(
628            protocol.channel_state("room:lobby"),
629            Some(ChannelState::Joining)
630        );
631
632        let event = protocol
633            .receive(reply(&outbound, "ok", json!({"ready": true})))
634            .unwrap();
635        assert_eq!(
636            event,
637            ProtocolEvent::Joined {
638                topic: "room:lobby".into(),
639                reference: outbound.reference,
640                response: json!({"ready": true}).into(),
641            }
642        );
643        assert_eq!(
644            protocol.channel_state("room:lobby"),
645            Some(ChannelState::Joined)
646        );
647    }
648
649    #[test]
650    fn correlates_push_replies_without_swallowing_server_events() {
651        let mut protocol = joined_protocol();
652        let push = protocol
653            .push("room:lobby", "new_message", json!({"body": "hello"}))
654            .unwrap();
655        let broadcast = Frame::new(
656            push.frame.join_ref.clone(),
657            None,
658            "room:lobby",
659            "presence_changed",
660            json!({"online": 2}),
661        );
662
663        assert_eq!(
664            protocol.receive(broadcast.clone()).unwrap(),
665            ProtocolEvent::Message(broadcast)
666        );
667        assert_eq!(
668            protocol
669                .receive(reply(&push, "ok", json!({"id": 99})))
670                .unwrap(),
671            ProtocolEvent::Reply {
672                topic: "room:lobby".into(),
673                event: "new_message".into(),
674                reference: push.reference,
675                status: ReplyStatus::Ok,
676                response: json!({"id": 99}).into(),
677            }
678        );
679    }
680
681    #[test]
682    fn forgets_a_timed_out_heartbeat() {
683        let mut protocol = Protocol::new();
684        let heartbeat = protocol.heartbeat();
685        assert!(protocol.forget_heartbeat(&heartbeat.reference));
686        assert!(!protocol.forget_heartbeat(&heartbeat.reference));
687        assert_eq!(
688            protocol
689                .receive(reply(&heartbeat, "ok", json!({})))
690                .unwrap(),
691            ProtocolEvent::UnmatchedReply(reply(&heartbeat, "ok", json!({})))
692        );
693    }
694
695    #[test]
696    fn rejects_pushes_before_join_completes() {
697        let mut protocol = Protocol::new();
698        protocol.join("room:lobby", json!({})).unwrap();
699
700        assert_eq!(
701            protocol.push("room:lobby", "event", json!({})).unwrap_err(),
702            ProtocolError::NotJoined("room:lobby".into())
703        );
704    }
705
706    #[test]
707    fn resets_and_rejoins_with_a_fresh_join_reference() {
708        let mut protocol = joined_protocol();
709        let push = protocol
710            .push("room:lobby", "new_message", json!({}))
711            .unwrap();
712        let old_join_ref = push.frame.join_ref.clone();
713
714        let interrupted = protocol.reset_connection();
715        assert_eq!(interrupted.len(), 1);
716        assert_eq!(
717            protocol.channel_state("room:lobby"),
718            Some(ChannelState::Disconnected)
719        );
720
721        let rejoins = protocol.rejoin_all_with_stored_params();
722        assert_eq!(rejoins.len(), 1);
723        assert_ne!(rejoins[0].frame.join_ref, old_join_ref);
724    }
725
726    #[test]
727    fn identifies_messages_from_an_old_join_generation() {
728        let mut protocol = joined_protocol();
729        protocol.reset_connection();
730        let rejoin = protocol.rejoin_all_with_stored_params().remove(0);
731        protocol.receive(reply(&rejoin, "ok", json!({}))).unwrap();
732
733        let stale = Frame::new(
734            Some("1".into()),
735            None,
736            "room:lobby",
737            "new_message",
738            json!({}),
739        );
740        assert_eq!(
741            protocol.receive(stale.clone()).unwrap(),
742            ProtocolEvent::StaleMessage(stale)
743        );
744    }
745
746    #[test]
747    fn rejects_replies_from_an_old_join_generation() {
748        let mut protocol = joined_protocol();
749        let push = protocol
750            .push("room:lobby", "save", json!({"version": 1}))
751            .unwrap();
752        let channel_error = Frame::new(
753            push.frame.join_ref.clone(),
754            None,
755            "room:lobby",
756            "phx_error",
757            json!({}),
758        );
759        protocol.receive(channel_error).unwrap();
760
761        let rejoin = protocol
762            .rejoin("room:lobby", json!({"token": "refreshed"}))
763            .unwrap();
764        protocol.receive(reply(&rejoin, "ok", json!({}))).unwrap();
765
766        let stale_reply = reply(&push, "ok", json!({"version": 2}));
767        assert_eq!(
768            protocol.receive(stale_reply.clone()).unwrap(),
769            ProtocolEvent::StaleMessage(stale_reply)
770        );
771    }
772}
phoenix_channel_runtime/protocol.rs

phoenix_channel_runtime/
protocol.rs
