use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::{CoreError, Kind};
pub const MAX_ROUTE_PATH: usize = 8;
pub const MAX_ROUTES_PER_UPDATE: usize = 4096;
pub const MAX_SUBJECT_LEN: usize = 256;
pub const MAX_ROUTE_IDENTIFIER_LEN: usize = 256;
pub const MAX_DESTINATION_LEN: usize = MAX_ROUTE_IDENTIFIER_LEN;
pub fn validate_node_identifier(node: &str) -> Result<(), CoreError> {
if node.is_empty()
|| node.len() > MAX_ROUTE_IDENTIFIER_LEN
|| node == "az"
|| !node
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~'))
{
return Err(CoreError::Establishment(
"node identifier is empty, reserved, not an ASCII URI-segment identifier, or exceeds its length limit"
.into(),
));
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct NodeIdentity {
pub node_id: String,
pub instance_id: String,
pub epoch: u64,
#[serde(default, skip_serializing_if = "Value::is_null")]
pub proof: Value,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct RouteAdvertisement {
pub destination: String,
pub owner: String,
pub owner_instance: String,
pub owner_epoch: u64,
pub owner_revision: u64,
pub distance: u32,
pub path: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct RouteWithdrawal {
pub destination: String,
pub owner: String,
pub owner_instance: String,
pub owner_epoch: u64,
pub owner_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RouteSnapshot {
pub generation: u64,
pub routes: Vec<RouteAdvertisement>,
}
impl RouteSnapshot {
pub fn canonical(generation: u64, mut routes: Vec<RouteAdvertisement>) -> RouteSnapshot {
routes.sort();
RouteSnapshot { generation, routes }
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RouteDelta {
pub generation: u64,
pub upsert: Vec<RouteAdvertisement>,
pub withdraw: Vec<RouteWithdrawal>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RouteAckStatus {
Applied,
ResyncRequired,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RouteAck {
pub generation: u64,
pub status: RouteAckStatus,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionClass {
Unclassified,
NodeCandidate,
Client,
}
impl SessionClass {
pub fn classify_first(kind: Kind) -> SessionClass {
match kind {
Kind::Identify => SessionClass::NodeCandidate,
Kind::Request | Kind::Subscribe | Kind::Channel | Kind::Discover => {
SessionClass::Client
}
_ => SessionClass::Unclassified,
}
}
}
#[derive(Debug, Default, Clone, PartialEq)]
pub(crate) struct Establishment {
sent_identify: bool,
remote_identity: Option<NodeIdentity>,
accepted_remote: bool,
remote_accepted_us: bool,
sent_snapshot: bool,
sent_snapshot_generation: Option<u64>,
applied_remote_snapshot: bool,
remote_acked_snapshot: bool,
}
impl Establishment {
pub(crate) fn identify_sent(&mut self) -> Result<(), CoreError> {
if self.sent_identify {
return Err(CoreError::Establishment("identify was already sent".into()));
}
self.sent_identify = true;
Ok(())
}
pub fn on_identify(&mut self, identity: NodeIdentity) -> Result<(), CoreError> {
if self.remote_identity.is_some() {
return Err(CoreError::Establishment(
"peer identity was already declared".into(),
));
}
self.remote_identity = Some(identity);
Ok(())
}
pub(crate) fn local_accept(&mut self) -> Result<&NodeIdentity, CoreError> {
let Some(identity) = self.remote_identity.as_ref() else {
return Err(CoreError::Establishment(
"cannot accept an identity that was never declared".into(),
));
};
self.accepted_remote = true;
Ok(identity)
}
pub(crate) fn on_identity_accepted(&mut self) -> Result<(), CoreError> {
if !self.sent_identify {
return Err(CoreError::Establishment(
"identity_accepted before identify was sent".into(),
));
}
self.remote_accepted_us = true;
Ok(())
}
pub fn identities_accepted(&self) -> bool {
self.accepted_remote && self.remote_accepted_us
}
pub(crate) fn snapshot_sent(&mut self, generation: u64) -> Result<(), CoreError> {
if !self.identities_accepted() {
return Err(CoreError::Establishment(
"route snapshot before bilateral identity acceptance".into(),
));
}
self.sent_snapshot = true;
self.sent_snapshot_generation = Some(generation);
Ok(())
}
pub(crate) fn on_snapshot_applied(&mut self) -> Result<(), CoreError> {
if !self.identities_accepted() {
return Err(CoreError::Establishment(
"route snapshot before bilateral identity acceptance".into(),
));
}
self.applied_remote_snapshot = true;
Ok(())
}
pub(crate) fn on_route_ack(&mut self, ack: &RouteAck) -> Result<(), CoreError> {
if !self.sent_snapshot {
return Err(CoreError::Establishment(
"route_ack before the initial snapshot was sent".into(),
));
}
if ack.status != RouteAckStatus::Applied
|| self.sent_snapshot_generation != Some(ack.generation)
{
return Err(CoreError::Establishment(
"initial snapshot ack was not Applied for the exact generation".into(),
));
}
self.remote_acked_snapshot = true;
Ok(())
}
pub fn remote_identity(&self) -> Option<&NodeIdentity> {
self.remote_identity.as_ref()
}
pub(crate) fn ready(&self) -> bool {
self.identities_accepted() && self.applied_remote_snapshot && self.remote_acked_snapshot
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn node_identifier_validation_uses_route_identity_limits() {
assert!(validate_node_identifier("node").is_ok());
assert!(validate_node_identifier("node.a").is_ok());
assert!(validate_node_identifier("").is_err());
assert!(validate_node_identifier(&"n".repeat(MAX_ROUTE_IDENTIFIER_LEN + 1)).is_err());
for invalid in [
"az",
"node/a",
"node?a",
"node#a",
"node\na",
"node a",
"n%C3%B8de",
"nøde",
] {
assert!(
validate_node_identifier(invalid).is_err(),
"accepted path-unsafe node identifier {invalid:?}"
);
}
}
use crate::{Envelope, DEFAULT_HOPS, PROTOCOL_VERSION};
use bytes::Bytes;
fn identity(node: &str) -> NodeIdentity {
NodeIdentity {
node_id: node.into(),
instance_id: format!("{node}-inst"),
epoch: 1,
proof: Value::Null,
}
}
fn advertisement(
destination: &str,
owner: &str,
distance: u32,
path: &[&str],
) -> RouteAdvertisement {
RouteAdvertisement {
destination: destination.into(),
owner: owner.into(),
owner_instance: format!("{owner}-inst"),
owner_epoch: 1,
owner_revision: 0,
distance,
path: path.iter().map(|s| s.to_string()).collect(),
}
}
fn control_frame(kind: Kind, payload: Bytes) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: "f1".into(),
target: String::new(),
subject: String::new(),
kind,
corr: None,
seq: None,
hops: None,
body_token: None,
payload,
path: Vec::new(),
headers: Default::default(),
}
}
#[test]
fn identity_payload_round_trips_through_an_envelope() {
let sent = identity("node-a");
let frame = control_frame(
Kind::Identify,
Envelope::encode_payload(&serde_json::to_value(&sent).unwrap()),
);
let decoded = Envelope::decode(frame.encode()).unwrap();
assert_eq!(decoded.kind, Kind::Identify);
assert_eq!(decoded.parse_payload::<NodeIdentity>().unwrap(), sent);
}
#[test]
fn an_arbitrary_proof_round_trips_through_the_identify_frame() {
let mut sent = identity("node-proof");
sent.proof = serde_json::json!({ "region": "eu", "tags": [1, 2, 3], "ok": true });
let frame = control_frame(
Kind::Identify,
Envelope::encode_payload(&serde_json::to_value(&sent).unwrap()),
);
let decoded = Envelope::decode(frame.encode()).unwrap();
assert_eq!(
decoded.parse_payload::<NodeIdentity>().unwrap().proof,
sent.proof
);
}
#[test]
fn a_null_proof_is_omitted_from_the_identify_payload_and_defaults_back() {
let sent = identity("node-null");
let json = serde_json::to_value(&sent).unwrap();
assert!(
json.get("proof").is_none(),
"null proof is omitted on the wire"
);
let decoded: NodeIdentity = serde_json::from_value(json).unwrap();
assert_eq!(decoded.proof, Value::Null);
}
#[test]
fn snapshot_canonical_order_is_input_order_independent() {
let a = advertisement("chess", "leaf-a", 0, &["leaf-a"]);
let b = advertisement("chess", "leaf-a", 1, &["leaf-a", "hub"]);
let c = advertisement("weather", "leaf-b", 0, &["leaf-b"]);
let one = RouteSnapshot::canonical(1, vec![c.clone(), b.clone(), a.clone()]);
let two = RouteSnapshot::canonical(1, vec![b.clone(), a.clone(), c.clone()]);
assert_eq!(one, two);
assert_eq!(one.routes, vec![a, b, c]);
}
#[test]
fn snapshot_delta_and_ack_payloads_round_trip() {
let snapshot =
RouteSnapshot::canonical(3, vec![advertisement("chess", "leaf-a", 0, &["leaf-a"])]);
let frame = control_frame(
Kind::RouteSnapshot,
Envelope::encode_payload(&serde_json::to_value(&snapshot).unwrap()),
);
let decoded = Envelope::decode(frame.encode()).unwrap();
assert_eq!(decoded.parse_payload::<RouteSnapshot>().unwrap(), snapshot);
let delta = RouteDelta {
generation: 4,
upsert: vec![advertisement("go", "leaf-b", 1, &["leaf-b", "hub"])],
withdraw: vec![RouteWithdrawal {
destination: "leaf-a".into(),
owner: "leaf-a".into(),
owner_instance: "leaf-a-inst".into(),
owner_epoch: 1,
owner_revision: 0,
}],
};
let frame = control_frame(
Kind::RouteDelta,
Envelope::encode_payload(&serde_json::to_value(&delta).unwrap()),
);
let decoded = Envelope::decode(frame.encode()).unwrap();
assert_eq!(decoded.parse_payload::<RouteDelta>().unwrap(), delta);
let ack = RouteAck {
generation: 4,
status: RouteAckStatus::ResyncRequired,
};
let frame = control_frame(
Kind::RouteAck,
Envelope::encode_payload(&serde_json::to_value(&ack).unwrap()),
);
let decoded = Envelope::decode(frame.encode()).unwrap();
assert_eq!(decoded.parse_payload::<RouteAck>().unwrap(), ack);
}
#[test]
fn v2_control_kinds_encode_snake_case() {
for (kind, name) in [
(Kind::Identify, "\"identify\""),
(Kind::IdentityAccepted, "\"identity_accepted\""),
(Kind::RouteSnapshot, "\"route_snapshot\""),
(Kind::RouteDelta, "\"route_delta\""),
(Kind::RouteAck, "\"route_ack\""),
] {
assert_eq!(serde_json::to_string(&kind).unwrap(), name);
}
}
#[test]
fn a_missing_payload_on_a_control_frame_is_malformed() {
let frame = control_frame(Kind::Identify, Bytes::new());
let decoded = Envelope::decode(frame.encode()).unwrap();
assert!(matches!(
decoded.parse_payload::<NodeIdentity>(),
Err(CoreError::Malformed(_))
));
}
#[test]
fn the_legal_establishment_sequence_reaches_ready() {
let mut establishment = Establishment::default();
assert!(!establishment.ready());
establishment.identify_sent().unwrap();
establishment.on_identify(identity("peer")).unwrap();
assert_eq!(establishment.local_accept().unwrap().node_id, "peer");
establishment.on_identity_accepted().unwrap();
assert!(!establishment.ready());
establishment.snapshot_sent(7).unwrap();
establishment.on_snapshot_applied().unwrap();
assert!(!establishment.ready());
establishment
.on_route_ack(&RouteAck {
generation: 7,
status: RouteAckStatus::Applied,
})
.unwrap();
assert!(establishment.ready());
}
#[test]
fn identify_twice_is_rejected() {
let mut establishment = Establishment::default();
establishment.identify_sent().unwrap();
assert!(matches!(
establishment.identify_sent(),
Err(CoreError::Establishment(_))
));
establishment.on_identify(identity("peer")).unwrap();
assert!(matches!(
establishment.on_identify(identity("peer")),
Err(CoreError::Establishment(_))
));
}
#[test]
fn acceptance_before_identify_is_rejected() {
let mut establishment = Establishment::default();
assert!(matches!(
establishment.on_identity_accepted(),
Err(CoreError::Establishment(_))
));
assert!(matches!(
establishment.local_accept(),
Err(CoreError::Establishment(_))
));
}
#[test]
fn snapshots_before_bilateral_identity_acceptance_are_rejected() {
let mut establishment = Establishment::default();
establishment.identify_sent().unwrap();
establishment.on_identify(identity("peer")).unwrap();
establishment.local_accept().unwrap();
assert!(matches!(
establishment.snapshot_sent(1),
Err(CoreError::Establishment(_))
));
assert!(matches!(
establishment.on_snapshot_applied(),
Err(CoreError::Establishment(_))
));
}
#[test]
fn an_ack_before_the_snapshot_was_sent_is_rejected() {
let mut establishment = Establishment::default();
establishment.identify_sent().unwrap();
establishment.on_identify(identity("peer")).unwrap();
establishment.local_accept().unwrap();
establishment.on_identity_accepted().unwrap();
assert!(matches!(
establishment.on_route_ack(&RouteAck {
generation: 1,
status: RouteAckStatus::Applied,
}),
Err(CoreError::Establishment(_))
));
}
#[test]
fn initial_snapshot_requires_an_applied_ack_for_the_exact_generation() {
let ready_for_snapshot = || {
let mut establishment = Establishment::default();
establishment.identify_sent().unwrap();
establishment.on_identify(identity("peer")).unwrap();
establishment.local_accept().unwrap();
establishment.on_identity_accepted().unwrap();
establishment.snapshot_sent(4).unwrap();
establishment
};
for ack in [
RouteAck {
generation: 3,
status: RouteAckStatus::Applied,
},
RouteAck {
generation: 4,
status: RouteAckStatus::ResyncRequired,
},
] {
assert!(matches!(
ready_for_snapshot().on_route_ack(&ack),
Err(CoreError::Establishment(_))
));
}
}
#[test]
fn the_first_frame_classifies_the_session() {
assert_eq!(
SessionClass::classify_first(Kind::Identify),
SessionClass::NodeCandidate
);
for kind in [
Kind::Request,
Kind::Subscribe,
Kind::Channel,
Kind::Discover,
] {
assert_eq!(SessionClass::classify_first(kind), SessionClass::Client);
}
for kind in [Kind::Ping, Kind::Pong] {
assert_eq!(
SessionClass::classify_first(kind),
SessionClass::Unclassified
);
}
}
#[test]
fn the_hop_budget_matches_the_spec_and_is_independent_of_the_route_path_limit() {
assert_eq!(DEFAULT_HOPS, 8);
assert_eq!(MAX_ROUTE_PATH, 8);
assert_eq!(MAX_ROUTES_PER_UPDATE, 4096);
assert_eq!(MAX_SUBJECT_LEN, 256);
}
}