use std::collections::{HashMap, VecDeque};
use std::time::Duration;
use bytes::Bytes;
use serde_json::json;
use web_time::Instant;
use crate::{
CoreError, Envelope, ErrorCode, Kind, SessionClass, TargetPath, DEFAULT_HOPS, PROTOCOL_VERSION,
};
pub const SUPPORTED_VERSIONS: &[u16] = &[PROTOCOL_VERSION];
const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
#[derive(Debug, Clone, PartialEq)]
#[allow(
clippy::large_enum_variant,
reason = "boxing the frame-dominant variant would allocate on every received frame"
)]
pub(crate) enum SessionInput {
Start { initiator: bool },
FrameReceived(Envelope),
Deadline(Instant),
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum SessionEffect {
SendFrame(Envelope),
HandshakeEstablished { version: u16 },
Deliver(Envelope),
StreamClosed { corr: String },
Closed { code: ErrorCode, message: String },
Deadline(Option<Instant>),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum State {
Idle,
HelloSent,
AwaitingHello,
Established,
Closed,
}
struct Stream {
next_seq: u64,
}
pub struct SessionReducer {
state: State,
initiator: bool,
version: Option<u16>,
outputs: VecDeque<SessionEffect>,
streams: HashMap<String, Stream>,
next_frame: u64,
next_stream: u64,
ping_outstanding: bool,
keepalive_deadline: Option<Instant>,
class: SessionClass,
peer_ready: bool,
authoritative_ordering: bool,
}
impl SessionReducer {
pub fn new() -> SessionReducer {
SessionReducer {
state: State::Idle,
initiator: false,
version: None,
outputs: VecDeque::new(),
streams: HashMap::new(),
next_frame: 0,
next_stream: 0,
ping_outstanding: false,
keepalive_deadline: None,
class: SessionClass::Unclassified,
peer_ready: false,
authoritative_ordering: false,
}
}
pub(crate) fn authoritative() -> SessionReducer {
SessionReducer {
authoritative_ordering: true,
..SessionReducer::new()
}
}
pub(crate) fn handle_input(
&mut self,
now: Instant,
input: SessionInput,
) -> Result<(), CoreError> {
match input {
SessionInput::Start { initiator } => self.start(initiator),
SessionInput::FrameReceived(envelope) => {
self.on_frame(envelope);
if self.state == State::Established {
self.ping_outstanding = false;
self.keepalive_deadline = Some(now + KEEPALIVE_INTERVAL);
}
}
SessionInput::Deadline(now) => self.on_timeout(now),
}
Ok(())
}
pub(crate) fn poll_effect(&mut self) -> SessionEffect {
self.outputs
.pop_front()
.unwrap_or(SessionEffect::Deadline(self.keepalive_deadline))
}
pub(crate) fn deadline(&self) -> Option<Instant> {
self.keepalive_deadline
}
pub(crate) fn class(&self) -> SessionClass {
self.class
}
#[cfg(test)]
fn is_established(&self) -> bool {
self.state == State::Established
}
#[cfg(test)]
fn is_closed(&self) -> bool {
self.state == State::Closed
}
pub(crate) fn mark_peer_ready(&mut self) {
self.peer_ready = true;
}
fn on_timeout(&mut self, now: Instant) {
if self.state != State::Established {
return;
}
let Some(deadline) = self.keepalive_deadline else {
return;
};
if now < deadline {
return;
}
self.keepalive_deadline = Some(now + KEEPALIVE_INTERVAL);
if self.ping_outstanding {
self.close_with_error(
ErrorCode::PeerUnreachable,
"peer sent no frames within the keepalive window",
);
} else {
self.ping_outstanding = true;
let ping = self.frame(Kind::Ping, Bytes::new());
self.outputs.push_back(SessionEffect::SendFrame(ping));
}
}
fn start(&mut self, initiator: bool) {
if self.state != State::Idle {
return;
}
self.initiator = initiator;
if initiator {
let payload = json!({ "versions": SUPPORTED_VERSIONS });
let hello = self.frame(Kind::Hello, Envelope::encode_payload(&payload));
self.outputs.push_back(SessionEffect::SendFrame(hello));
self.state = State::HelloSent;
} else {
self.state = State::AwaitingHello;
}
}
fn on_frame(&mut self, envelope: Envelope) {
match (self.state, envelope.kind) {
(State::AwaitingHello, Kind::Hello) => self.on_hello(envelope),
(State::HelloSent, Kind::Welcome) => self.on_welcome(envelope),
(State::Established, _) => self.on_established_frame(envelope),
(State::Closed, _) => {}
_ => self.close_with_error(ErrorCode::Protocol, "frame before handshake completed"),
}
}
fn on_hello(&mut self, envelope: Envelope) {
let payload = envelope.payload_json();
let Some(versions) = payload["versions"].as_array() else {
self.close_with_error(ErrorCode::Protocol, "hello versions must be an array");
return;
};
if versions.is_empty() {
self.close_with_error(ErrorCode::Protocol, "hello versions must not be empty");
return;
}
let Some(offered) = versions
.iter()
.map(|value| value.as_u64().and_then(|value| u16::try_from(value).ok()))
.collect::<Option<Vec<_>>>()
else {
self.close_with_error(
ErrorCode::Protocol,
"hello versions must contain only u16 integers",
);
return;
};
let Some(version) = SUPPORTED_VERSIONS
.iter()
.rev()
.find(|v| offered.contains(v))
.copied()
else {
self.close_with_error(ErrorCode::VersionMismatch, "no mutual protocol version");
return;
};
let welcome = self.frame(
Kind::Welcome,
Envelope::encode_payload(&json!({ "version": version })),
);
self.outputs.push_back(SessionEffect::SendFrame(welcome));
self.state = State::Established;
self.version = Some(version);
self.outputs
.push_back(SessionEffect::HandshakeEstablished { version });
}
fn on_welcome(&mut self, envelope: Envelope) {
let version = envelope.payload_json()["version"]
.as_u64()
.and_then(|v| u16::try_from(v).ok());
match version {
Some(version) if SUPPORTED_VERSIONS.contains(&version) => {
self.state = State::Established;
self.version = Some(version);
self.outputs
.push_back(SessionEffect::HandshakeEstablished { version });
}
_ => self.close_with_error(
ErrorCode::VersionMismatch,
"welcome named an unsupported version",
),
}
}
fn on_established_frame(&mut self, envelope: Envelope) {
if matches!(envelope.kind, Kind::Hello | Kind::Welcome) {
self.close_with_error(ErrorCode::Protocol, "handshake frame after establishment");
return;
}
if self.authoritative_ordering && !self.accept_frame_order(envelope.kind) {
return;
}
match envelope.kind {
Kind::Ping => {
let pong = self.frame(Kind::Pong, Bytes::new());
self.outputs.push_back(SessionEffect::SendFrame(pong));
}
Kind::Pong | Kind::Hello | Kind::Welcome => {}
Kind::Identify
| Kind::IdentityAccepted
| Kind::RouteSnapshot
| Kind::RouteDelta
| Kind::RouteAck => self.outputs.push_back(SessionEffect::Deliver(envelope)),
Kind::Request | Kind::Subscribe | Kind::Channel | Kind::Discover => {
let Some(corr) = envelope.corr.clone() else {
let error = self.frame(
Kind::Error,
Envelope::encode_payload(&json!({ "code": ErrorCode::Protocol, "message": "stream-opening frame without corr" })),
);
self.outputs.push_back(SessionEffect::SendFrame(error));
return;
};
if self.streams.contains_key(&corr) {
let error = self.frame(
Kind::Error,
Envelope::encode_payload(&json!({ "code": ErrorCode::Protocol, "message": format!("stream {corr} is already open") })),
);
self.outputs.push_back(SessionEffect::SendFrame(error));
return;
}
self.streams.insert(corr, Stream { next_seq: 0 });
self.outputs.push_back(SessionEffect::Deliver(envelope));
}
Kind::Event => {
if envelope
.corr
.as_ref()
.is_some_and(|corr| self.streams.contains_key(corr))
{
self.outputs.push_back(SessionEffect::Deliver(envelope));
}
}
Kind::Response | Kind::Error => {
let Some(corr) = envelope.corr.clone() else {
return;
};
if self.streams.remove(&corr).is_some() {
self.outputs.push_back(SessionEffect::Deliver(envelope));
self.outputs.push_back(SessionEffect::StreamClosed { corr });
}
}
Kind::Cancel => {
let Some(corr) = envelope.corr.clone() else {
return;
};
if self.streams.remove(&corr).is_some() {
self.outputs.push_back(SessionEffect::Deliver(envelope));
self.outputs.push_back(SessionEffect::StreamClosed { corr });
}
}
}
}
fn accept_frame_order(&mut self, kind: Kind) -> bool {
match (self.class, SessionClass::classify_first(kind)) {
(SessionClass::Unclassified, SessionClass::Client) => {
self.class = SessionClass::Client;
true
}
(SessionClass::Unclassified, SessionClass::NodeCandidate) => {
self.class = SessionClass::NodeCandidate;
true
}
(SessionClass::Client, SessionClass::NodeCandidate) => {
self.close_with_error(
ErrorCode::Protocol,
"node control frame after application traffic on a client session",
);
false
}
(SessionClass::NodeCandidate, SessionClass::Client) => {
if self.peer_ready {
return true;
}
self.close_with_error(
ErrorCode::Protocol,
"application frame before the node connection was ready",
);
false
}
_ if self.class == SessionClass::Client
&& matches!(
kind,
Kind::IdentityAccepted
| Kind::RouteSnapshot
| Kind::RouteDelta
| Kind::RouteAck
) =>
{
self.close_with_error(
ErrorCode::Protocol,
"node control frame after application traffic on a client session",
);
false
}
_ => true,
}
}
pub(crate) fn open_stream_from_body(
&mut self,
path: &str,
kind: Kind,
body_token: Option<String>,
hops: Option<u8>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<String, CoreError> {
if self.state != State::Established {
return Err(CoreError::NotEstablished);
}
if !matches!(kind, Kind::Request | Kind::Subscribe | Kind::Channel) {
return Err(CoreError::BadKind(format!("{kind:?}")));
}
let target_path = TargetPath::parse_application(path)?;
Envelope::ensure_headers_wire_safe(&headers)?;
self.next_stream += 1;
let parity_base = if self.initiator { 1 } else { 2 };
let corr = format!("s{}", parity_base + (self.next_stream - 1) * 2);
self.streams.insert(corr.clone(), Stream { next_seq: 0 });
let mut envelope = self.frame(kind, Bytes::new());
envelope.target = target_path.target().to_owned();
envelope.subject = target_path.subject().to_owned();
envelope.corr = Some(corr.clone());
envelope.hops = Some(hops.unwrap_or(DEFAULT_HOPS));
envelope.headers = headers;
envelope.body_token = body_token;
self.outputs.push_back(SessionEffect::SendFrame(envelope));
Ok(corr)
}
pub(crate) fn open_discovery(
&mut self,
path: &str,
payload: Bytes,
hops: Option<u8>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<String, CoreError> {
if self.state != State::Established {
return Err(CoreError::NotEstablished);
}
let target_path = TargetPath::parse_discovery(path)?;
Envelope::ensure_headers_wire_safe(&headers)?;
self.next_stream += 1;
let parity_base = if self.initiator { 1 } else { 2 };
let corr = format!("s{}", parity_base + (self.next_stream - 1) * 2);
self.streams.insert(corr.clone(), Stream { next_seq: 0 });
let mut envelope = self.frame(Kind::Discover, payload);
envelope.target = target_path.target().to_owned();
envelope.corr = Some(corr.clone());
envelope.hops = Some(hops.unwrap_or(DEFAULT_HOPS));
envelope.headers = headers;
self.outputs.push_back(SessionEffect::SendFrame(envelope));
Ok(corr)
}
pub(crate) fn send_body(
&mut self,
corr: &str,
body: Option<String>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<(), CoreError> {
let stream = self
.streams
.get_mut(corr)
.ok_or_else(|| CoreError::UnknownStream(corr.into()))?;
Envelope::ensure_headers_wire_safe(&headers)?;
stream.next_seq += 1;
let seq = stream.next_seq;
let mut envelope = self.frame(Kind::Event, Bytes::new());
envelope.corr = Some(corr.to_string());
envelope.seq = Some(seq);
envelope.headers = headers;
envelope.body_token = body;
self.outputs.push_back(SessionEffect::SendFrame(envelope));
Ok(())
}
pub(crate) fn send_discovery_event(
&mut self,
corr: &str,
payload: Bytes,
) -> Result<(), CoreError> {
let stream = self
.streams
.get_mut(corr)
.ok_or_else(|| CoreError::UnknownStream(corr.into()))?;
stream.next_seq += 1;
let seq = stream.next_seq;
let mut envelope = self.frame(Kind::Event, payload);
envelope.corr = Some(corr.to_string());
envelope.seq = Some(seq);
self.outputs.push_back(SessionEffect::SendFrame(envelope));
Ok(())
}
pub(crate) fn respond_body(
&mut self,
corr: &str,
body: Option<String>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<(), CoreError> {
self.terminal_frame(
corr,
Kind::Response,
Bytes::new(),
body,
Vec::new(),
headers,
)
}
pub(crate) fn respond_discovery(
&mut self,
corr: &str,
payload: Bytes,
) -> Result<(), CoreError> {
self.terminal_frame(
corr,
Kind::Response,
payload,
None,
Vec::new(),
serde_json::Map::new(),
)
}
pub fn fail(&mut self, corr: &str, code: ErrorCode, message: &str) -> Result<(), CoreError> {
self.terminal_frame(
corr,
Kind::Error,
Envelope::encode_payload(&json!({ "code": code, "message": message })),
None,
Vec::new(),
serde_json::Map::new(),
)
}
pub(crate) fn respond_terminal(
&mut self,
corr: &str,
kind: Kind,
payload: Bytes,
path: Vec<String>,
) -> Result<(), CoreError> {
self.terminal_frame(corr, kind, payload, None, path, serde_json::Map::new())
}
pub fn fail_closed(&mut self, corr: &str, code: ErrorCode, message: &str) {
let mut envelope = self.frame(
Kind::Error,
Envelope::encode_payload(&json!({ "code": code, "message": message })),
);
envelope.corr = Some(corr.to_string());
self.outputs.push_back(SessionEffect::SendFrame(envelope));
}
pub(crate) fn respond_terminal_with(
&mut self,
corr: &str,
kind: Kind,
payload: Bytes,
path: Vec<String>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<(), CoreError> {
self.terminal_frame(corr, kind, payload, None, path, headers)
}
pub(crate) fn respond_terminal_body(
&mut self,
corr: &str,
body: Option<String>,
path: Vec<String>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<(), CoreError> {
self.terminal_frame(corr, Kind::Response, Bytes::new(), body, path, headers)
}
fn terminal_frame(
&mut self,
corr: &str,
kind: Kind,
payload: Bytes,
body_token: Option<String>,
path: Vec<String>,
headers: serde_json::Map<String, serde_json::Value>,
) -> Result<(), CoreError> {
if !matches!(kind, Kind::Response | Kind::Error) {
return Err(CoreError::BadKind(format!("{kind:?}")));
}
Envelope::ensure_headers_wire_safe(&headers)?;
if self.streams.remove(corr).is_none() {
return Err(CoreError::UnknownStream(corr.into()));
}
let mut envelope = self.frame(kind, payload);
envelope.corr = Some(corr.to_string());
envelope.path = path;
envelope.headers = headers;
envelope.body_token = body_token;
self.outputs.push_back(SessionEffect::SendFrame(envelope));
self.outputs.push_back(SessionEffect::StreamClosed {
corr: corr.to_string(),
});
Ok(())
}
pub fn cancel(&mut self, corr: &str) -> Result<(), CoreError> {
if self.streams.remove(corr).is_none() {
return Err(CoreError::UnknownStream(corr.into()));
}
let mut envelope = self.frame(Kind::Cancel, Bytes::new());
envelope.corr = Some(corr.to_string());
self.outputs.push_back(SessionEffect::SendFrame(envelope));
self.outputs.push_back(SessionEffect::StreamClosed {
corr: corr.to_string(),
});
Ok(())
}
pub(crate) fn operation_stream_failed(&mut self, corr: &str, code: ErrorCode, message: &str) {
if self.streams.remove(corr).is_none() {
return;
}
let mut envelope = self.frame(
Kind::Error,
Envelope::encode_payload(&json!({ "code": code, "message": message })),
);
envelope.corr = Some(corr.to_owned());
self.outputs.push_back(SessionEffect::Deliver(envelope));
self.outputs.push_back(SessionEffect::StreamClosed {
corr: corr.to_owned(),
});
}
pub fn control(&mut self, kind: Kind, payload: Bytes) -> Result<(), CoreError> {
if !matches!(
kind,
Kind::Identify
| Kind::IdentityAccepted
| Kind::RouteSnapshot
| Kind::RouteDelta
| Kind::RouteAck
) {
return Err(CoreError::BadKind(format!("{kind:?}")));
}
if self.state != State::Established {
return Err(CoreError::NotEstablished);
}
let frame = self.frame(kind, payload);
self.outputs.push_back(SessionEffect::SendFrame(frame));
Ok(())
}
fn close_with_error(&mut self, code: ErrorCode, message: &str) {
let error = self.frame(
Kind::Error,
Envelope::encode_payload(&json!({ "code": code, "message": message })),
);
self.outputs.push_back(SessionEffect::SendFrame(error));
self.state = State::Closed;
self.outputs.push_back(SessionEffect::Closed {
code,
message: message.to_string(),
});
}
fn frame(&mut self, kind: Kind, payload: Bytes) -> Envelope {
self.next_frame += 1;
Envelope {
v: self.version.unwrap_or(PROTOCOL_VERSION),
id: format!("f{}", self.next_frame),
target: String::new(),
subject: String::new(),
kind,
corr: None,
seq: None,
hops: None,
body_token: None,
payload,
path: Vec::new(),
headers: Default::default(),
}
}
}
impl Default for SessionReducer {
fn default() -> SessionReducer {
SessionReducer::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::Value;
fn now() -> Instant {
Instant::now()
}
fn bytes(value: Value) -> Bytes {
Envelope::encode_payload(&value)
}
fn received(kind: Kind, subject: &str, corr: Option<&str>, payload: Value) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: "f9".into(),
target: if kind.is_application_request() {
"target-node".into()
} else {
String::new()
},
subject: subject.into(),
kind,
corr: corr.map(str::to_owned),
seq: None,
hops: None,
body_token: None,
payload: bytes(payload),
path: Vec::new(),
headers: Default::default(),
}
}
fn drain_frames(core: &mut SessionReducer) -> (Vec<Envelope>, Vec<SessionEffect>) {
let mut frames = Vec::new();
let mut other = Vec::new();
loop {
match core.poll_effect() {
SessionEffect::SendFrame(envelope) => frames.push(envelope),
SessionEffect::Deadline(_) => break,
output => other.push(output),
}
}
(frames, other)
}
#[test]
fn handshake_establishes_both_ends_with_negotiated_version() {
let mut dialer = SessionReducer::new();
let mut listener = SessionReducer::new();
dialer
.handle_input(now(), SessionInput::Start { initiator: true })
.unwrap();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let (hello, _) = drain_frames(&mut dialer);
assert_eq!(hello[0].kind, Kind::Hello);
assert_eq!(
hello[0].payload_json(),
json!({ "versions": [1] }),
"hello negotiates version and nothing else"
);
listener
.handle_input(now(), SessionInput::FrameReceived(hello[0].clone()))
.unwrap();
let (welcome, listener_events) = drain_frames(&mut listener);
assert_eq!(welcome[0].kind, Kind::Welcome);
assert_eq!(
listener_events[0],
SessionEffect::HandshakeEstablished { version: 1 }
);
dialer
.handle_input(now(), SessionInput::FrameReceived(welcome[0].clone()))
.unwrap();
let (_, dialer_events) = drain_frames(&mut dialer);
assert_eq!(
dialer_events[0],
SessionEffect::HandshakeEstablished { version: 1 }
);
assert!(dialer.is_established() && listener.is_established());
}
#[test]
fn a_v1_only_hello_negotiates_version_1_for_deployed_clients() {
let mut listener = SessionReducer::new();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let hello = received(Kind::Hello, "", None, json!({ "versions": [1] }));
listener
.handle_input(now(), SessionInput::FrameReceived(hello))
.unwrap();
let (welcome, events) = drain_frames(&mut listener);
assert_eq!(welcome[0].kind, Kind::Welcome);
assert_eq!(welcome[0].payload_json()["version"], 1);
assert_eq!(
events[0],
SessionEffect::HandshakeEstablished { version: 1 }
);
}
#[test]
fn a_hello_with_custom_headers_and_token_payload_establishes_on_version_alone() {
let mut listener = SessionReducer::new();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let mut hello = received(
Kind::Hello,
"",
None,
json!({ "versions": [1], "token": "not-protocol" }),
);
hello
.headers
.insert("actor".into(), Value::String("jwt-abc".into()));
listener
.handle_input(now(), SessionInput::FrameReceived(hello))
.unwrap();
let (welcome, events) = drain_frames(&mut listener);
assert_eq!(welcome[0].kind, Kind::Welcome);
assert_eq!(
events[0],
SessionEffect::HandshakeEstablished { version: 1 }
);
assert!(listener.is_established());
}
#[test]
fn no_mutual_version_closes_with_version_mismatch() {
let mut listener = SessionReducer::new();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let alien_hello = received(Kind::Hello, "", None, json!({"versions": [99]}));
listener
.handle_input(now(), SessionInput::FrameReceived(alien_hello))
.unwrap();
let (frames, events) = drain_frames(&mut listener);
assert_eq!(frames[0].kind, Kind::Error);
assert_eq!(frames[0].payload_json()["code"], "VERSION_MISMATCH");
assert!(matches!(
events[0],
SessionEffect::Closed {
code: ErrorCode::VersionMismatch,
..
}
));
}
#[test]
fn malformed_hello_versions_are_protocol_errors() {
for versions in [
Value::Null,
json!(1),
json!([]),
json!([1, "2"]),
json!([1.0]),
json!([-1]),
json!([65536]),
] {
let mut listener = SessionReducer::new();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
listener
.handle_input(
now(),
SessionInput::FrameReceived(received(
Kind::Hello,
"",
None,
json!({ "versions": versions }),
)),
)
.unwrap();
let (frames, events) = drain_frames(&mut listener);
assert_eq!(frames[0].payload_json()["code"], "PROTOCOL");
assert!(matches!(
events[0],
SessionEffect::Closed {
code: ErrorCode::Protocol,
..
}
));
}
}
#[test]
fn frame_before_handshake_is_a_protocol_error() {
let mut listener = SessionReducer::new();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let early = received(Kind::Request, "chess", Some("s1"), Value::Null);
listener
.handle_input(now(), SessionInput::FrameReceived(early))
.unwrap();
let (frames, events) = drain_frames(&mut listener);
assert_eq!(frames[0].payload_json()["code"], "PROTOCOL");
assert!(matches!(
events[0],
SessionEffect::Closed {
code: ErrorCode::Protocol,
..
}
));
}
#[test]
fn unsupported_welcome_closes_the_dialer() {
let mut dialer = SessionReducer::new();
dialer
.handle_input(now(), SessionInput::Start { initiator: true })
.unwrap();
drain_frames(&mut dialer);
let bad_welcome = received(Kind::Welcome, "", None, json!({"version": 99}));
dialer
.handle_input(now(), SessionInput::FrameReceived(bad_welcome))
.unwrap();
let (_, events) = drain_frames(&mut dialer);
assert!(matches!(
events[0],
SessionEffect::Closed {
code: ErrorCode::VersionMismatch,
..
}
));
}
#[test]
fn welcome_must_select_an_offered_version_without_u16_truncation() {
let mut dialer = SessionReducer::new();
dialer
.handle_input(now(), SessionInput::Start { initiator: true })
.unwrap();
drain_frames(&mut dialer);
let wrapped = received(Kind::Welcome, "", None, json!({"version": 65538}));
dialer
.handle_input(now(), SessionInput::FrameReceived(wrapped))
.unwrap();
let (_, events) = drain_frames(&mut dialer);
assert!(matches!(events[0], SessionEffect::Closed { .. }));
}
#[test]
fn an_established_session_delivers_controls_and_rejects_a_second_hello() {
let establish = || {
let mut core = SessionReducer::new();
core.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
core.handle_input(
now(),
SessionInput::FrameReceived(received(
Kind::Hello,
"",
None,
json!({"versions": [1]}),
)),
)
.unwrap();
drain_frames(&mut core);
core
};
let mut control = establish();
control
.handle_input(
now(),
SessionInput::FrameReceived(received(Kind::Identify, "", None, json!({}))),
)
.unwrap();
let (_, events) = drain_frames(&mut control);
assert!(matches!(events[0], SessionEffect::Deliver(_)));
assert!(!control.is_closed());
let mut hello = establish();
hello
.handle_input(
now(),
SessionInput::FrameReceived(received(
Kind::Hello,
"",
None,
json!({"versions": [1]}),
)),
)
.unwrap();
assert!(hello.is_closed());
}
fn established_pair() -> (SessionReducer, SessionReducer) {
let mut dialer = SessionReducer::new();
let mut listener = SessionReducer::new();
dialer
.handle_input(now(), SessionInput::Start { initiator: true })
.unwrap();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let (hello, _) = drain_frames(&mut dialer);
listener
.handle_input(now(), SessionInput::FrameReceived(hello[0].clone()))
.unwrap();
let (welcome, _) = drain_frames(&mut listener);
dialer
.handle_input(now(), SessionInput::FrameReceived(welcome[0].clone()))
.unwrap();
drain_frames(&mut dialer);
(dialer, listener)
}
fn pump(from: &mut SessionReducer, to: &mut SessionReducer) -> Vec<SessionEffect> {
let (frames, mut events) = drain_frames(from);
for frame in frames {
to.handle_input(now(), SessionInput::FrameReceived(frame))
.unwrap();
}
let (_, to_events) = drain_frames(to);
events.extend(to_events);
events
}
fn open_application(
session: &mut SessionReducer,
subject: &str,
kind: Kind,
body: Option<&str>,
) -> Result<String, CoreError> {
session.open_stream_from_body(
&format!("/node/{subject}"),
kind,
body.map(str::to_owned),
None,
Default::default(),
)
}
#[test]
fn request_response_round_trip_with_odd_corr_for_the_dialer() {
let (mut dialer, mut listener) = established_pair();
let corr =
open_application(&mut dialer, "chess", Kind::Request, Some("request-body")).unwrap();
assert_eq!(corr, "s1", "dialer-opened streams use odd corr ids");
let events = pump(&mut dialer, &mut listener);
let SessionEffect::Deliver(request) = &events[0] else {
panic!("{events:?}")
};
assert_eq!(request.subject, "chess");
assert_eq!(request.target, "node");
assert_eq!(request.hops, Some(DEFAULT_HOPS));
assert_eq!(request.body_token.as_deref(), Some("request-body"));
assert!(request.payload.is_empty());
listener
.respond_body(&corr, Some("response-body".into()), Default::default())
.unwrap();
let events = pump(&mut listener, &mut dialer);
let SessionEffect::Deliver(response) = &events[1] else {
panic!("{events:?}")
};
assert_eq!(response.kind, Kind::Response);
assert_eq!(response.body_token.as_deref(), Some("response-body"));
assert!(response.payload.is_empty());
assert!(events.contains(&SessionEffect::StreamClosed { corr: corr.clone() }));
}
#[test]
fn open_stream_from_rejects_frame_splitting_headers() {
let (mut dialer, _listener) = established_pair();
let mut headers = serde_json::Map::new();
headers.insert("x-trace".into(), json!("span\r\nunb-corr: forged"));
let error = dialer
.open_stream_from_body("/node/chess", Kind::Request, None, None, headers)
.unwrap_err();
assert!(matches!(error, CoreError::Malformed(_)), "{error}");
assert!(
drain_frames(&mut dialer).0.is_empty(),
"a rejected open emits no frame"
);
}
#[test]
fn acceptor_opened_streams_use_even_corr_ids() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut listener, "browser-ui", Kind::Request, None).unwrap();
assert_eq!(
corr, "s2",
"peer symmetry: the acceptor originates without corr collision"
);
let events = pump(&mut listener, &mut dialer);
assert!(matches!(&events[0], SessionEffect::Deliver(env) if env.subject == "browser-ui"));
}
#[test]
fn subscribe_streams_events_in_order_until_terminal_response() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut dialer, "todo", Kind::Subscribe, Some("query")).unwrap();
pump(&mut dialer, &mut listener);
listener
.send_body(&corr, Some("event-1".into()), Default::default())
.unwrap();
listener
.send_body(&corr, Some("event-2".into()), Default::default())
.unwrap();
listener
.respond_body(&corr, Some("done".into()), Default::default())
.unwrap();
let events = pump(&mut listener, &mut dialer);
let delivered: Vec<&Envelope> = events
.iter()
.filter_map(|output| match output {
SessionEffect::Deliver(envelope) => Some(envelope),
_ => None,
})
.collect();
assert_eq!(delivered.len(), 3);
assert_eq!((delivered[0].seq, delivered[1].seq), (Some(1), Some(2)));
assert_eq!(delivered[0].body_token.as_deref(), Some("event-1"));
assert_eq!(delivered[1].body_token.as_deref(), Some("event-2"));
assert_eq!(delivered[2].kind, Kind::Response);
assert!(events.contains(&SessionEffect::StreamClosed { corr }));
}
#[test]
fn cancel_tears_down_both_sides_and_late_sends_fail() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut dialer, "todo", Kind::Subscribe, None).unwrap();
pump(&mut dialer, &mut listener);
dialer.cancel(&corr).unwrap();
let events = pump(&mut dialer, &mut listener);
assert!(events.iter().any(
|output| matches!(output, SessionEffect::Deliver(env) if env.kind == Kind::Cancel)
));
assert!(
events
.iter()
.filter(|output| matches!(output, SessionEffect::StreamClosed { .. }))
.count()
>= 2
);
let late = listener.send_body(&corr, Some("late".into()), Default::default());
assert!(matches!(late, Err(CoreError::UnknownStream(_))));
}
#[test]
fn channel_carries_items_in_both_directions_on_one_corr() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut dialer, "assistant", Kind::Channel, None).unwrap();
pump(&mut dialer, &mut listener);
dialer
.send_body(&corr, Some("say".into()), Default::default())
.unwrap();
let to_listener = pump(&mut dialer, &mut listener);
assert!(to_listener.iter().any(
|output| matches!(output, SessionEffect::Deliver(env) if env.body_token.as_deref() == Some("say"))
));
listener
.send_body(&corr, Some("reply".into()), Default::default())
.unwrap();
let to_dialer = pump(&mut listener, &mut dialer);
assert!(to_dialer.iter().any(
|output| matches!(output, SessionEffect::Deliver(env) if env.body_token.as_deref() == Some("reply"))
));
}
#[test]
fn fail_delivers_a_terminal_error_frame() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut dialer, "chess", Kind::Request, None).unwrap();
pump(&mut dialer, &mut listener);
listener
.fail(&corr, ErrorCode::InvalidInput, "unknown field \"frm\"")
.unwrap();
let events = pump(&mut listener, &mut dialer);
let SessionEffect::Deliver(error) = &events[1] else {
panic!("{events:?}")
};
assert_eq!(error.kind, Kind::Error);
assert_eq!(error.payload_json()["code"], "INVALID_INPUT");
assert!(events.contains(&SessionEffect::StreamClosed { corr }));
}
#[test]
fn a_duplicate_corr_open_is_refused_and_the_original_stream_survives() {
let (mut dialer, mut listener) = established_pair();
let corr = open_application(&mut dialer, "todo", Kind::Subscribe, None).unwrap();
pump(&mut dialer, &mut listener);
let duplicate = received(Kind::Request, "todo", Some(&corr), Value::Null);
listener
.handle_input(now(), SessionInput::FrameReceived(duplicate))
.unwrap();
let (frames, events) = drain_frames(&mut listener);
assert_eq!(frames[0].kind, Kind::Error);
assert_eq!(frames[0].payload_json()["code"], "PROTOCOL");
assert!(
frames[0].corr.is_none(),
"the refusal must not tear down the live stream"
);
assert!(events.is_empty(), "the duplicate open is not delivered");
listener
.send_body(&corr, Some("event".into()), Default::default())
.unwrap();
let events = pump(&mut listener, &mut dialer);
assert!(
events.iter().any(
|output| matches!(output, SessionEffect::Deliver(env) if env.body_token.as_deref() == Some("event"))
),
"the original stream still delivers: {events:?}"
);
}
#[test]
fn frames_for_unknown_streams_are_ignored() {
let (mut dialer, _) = established_pair();
let stray = received(Kind::Response, "", Some("s99"), json!({"ok": true}));
dialer
.handle_input(now(), SessionInput::FrameReceived(stray))
.unwrap();
let (frames, events) = drain_frames(&mut dialer);
assert!(
frames.is_empty() && events.is_empty(),
"idempotent close: late frames are dropped"
);
}
#[test]
fn open_stream_before_handshake_is_refused() {
let mut core = SessionReducer::new();
let refused = open_application(&mut core, "chess", Kind::Request, None);
assert!(matches!(refused, Err(CoreError::NotEstablished)));
}
#[test]
fn established_core_answers_ping_with_pong() {
let mut dialer = SessionReducer::new();
let mut listener = SessionReducer::new();
dialer
.handle_input(now(), SessionInput::Start { initiator: true })
.unwrap();
listener
.handle_input(now(), SessionInput::Start { initiator: false })
.unwrap();
let (hello, _) = drain_frames(&mut dialer);
listener
.handle_input(now(), SessionInput::FrameReceived(hello[0].clone()))
.unwrap();
drain_frames(&mut listener);
let ping = received(Kind::Ping, "", None, Value::Null);
listener
.handle_input(now(), SessionInput::FrameReceived(ping))
.unwrap();
let (frames, _) = drain_frames(&mut listener);
assert_eq!(frames[0].kind, Kind::Pong);
}
#[test]
fn keepalive_pings_after_one_interval_and_closes_after_the_second() {
let (_, mut listener) = established_pair();
let first = listener.deadline().unwrap();
listener
.handle_input(first, SessionInput::Deadline(first))
.unwrap();
let (frames, effects) = drain_frames(&mut listener);
assert_eq!(frames.len(), 1);
assert_eq!(frames[0].kind, Kind::Ping);
assert!(effects.is_empty());
let second = listener.deadline().unwrap();
assert_eq!(second.duration_since(first), KEEPALIVE_INTERVAL);
listener
.handle_input(second, SessionInput::Deadline(second))
.unwrap();
let (_, effects) = drain_frames(&mut listener);
assert!(effects.iter().any(|effect| matches!(
effect,
SessionEffect::Closed {
code: ErrorCode::PeerUnreachable,
..
}
)));
}
#[test]
fn any_inbound_frame_clears_the_outstanding_keepalive_ping() {
let (_, mut listener) = established_pair();
let first = listener.deadline().unwrap();
listener
.handle_input(first, SessionInput::Deadline(first))
.unwrap();
drain_frames(&mut listener);
let inbound = received(Kind::Pong, "", None, Value::Null);
let activity = first + Duration::from_secs(1);
listener
.handle_input(activity, SessionInput::FrameReceived(inbound))
.unwrap();
drain_frames(&mut listener);
let next = listener.deadline().unwrap();
listener
.handle_input(next, SessionInput::Deadline(next))
.unwrap();
let (frames, effects) = drain_frames(&mut listener);
assert_eq!(frames[0].kind, Kind::Ping);
assert!(!effects
.iter()
.any(|effect| matches!(effect, SessionEffect::Closed { .. })));
}
}