mod common;
use common::connect_nodes;
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::sync::mpsc;
use unb::{handler, Handler, HandlerError, Reply, Request};
use unb_client::pair;
use unb_core::{Envelope, Kind, NodeIdentity, RouteSnapshot, PROTOCOL_VERSION};
use unb_runtime::Pipe;
use unb_runtime::Wire;
use unb_server::Node;
#[derive(Deserialize, JsonSchema)]
struct Probe {}
#[handler]
async fn weather(_request: Request<Probe>) -> Result<Reply<Value>, HandlerError> {
Ok(Reply::new(json!({ "temp_c": 21 })))
}
#[handler]
async fn late_ok(_request: Request<Probe>) -> Result<Reply<Value>, HandlerError> {
Ok(Reply::new(json!({ "ok": true })))
}
fn weather_node(name: &str) -> std::sync::Arc<Node> {
Node::builder(name)
.service(weather)
.insecure_accept_declared_peer_identities()
.build()
.unwrap()
}
fn v1_frame(id: &str, kind: Kind, corr: Option<&str>, subject: &str, payload: Value) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: id.into(),
target: if kind.is_application_request() {
"weather-1".into()
} else {
String::new()
},
subject: subject.into(),
kind,
corr: corr.map(str::to_owned),
seq: None,
hops: None,
body_token: None,
payload: Envelope::encode_payload(&payload),
path: Vec::new(),
headers: Default::default(),
}
}
#[tokio::test(flavor = "multi_thread")]
async fn a_v1_hello_client_session_is_served_requests_without_identity() {
let node = weather_node("weather-1");
let (client_tx, node_rx) = mpsc::channel::<Envelope>(64);
let (node_tx, mut client_rx) = mpsc::channel::<Envelope>(64);
node.serve_transport(Pipe::Local {
rx: node_rx,
tx: node_tx,
initiator: false,
})
.await;
client_tx
.send(v1_frame(
"f1",
Kind::Hello,
None,
"",
json!({ "versions": [1] }),
))
.await
.unwrap();
let welcome = client_rx.recv().await.unwrap();
assert_eq!(welcome.kind, Kind::Welcome);
assert_eq!(
welcome.payload_json()["version"],
1,
"a deployed v1 client keeps negotiating version 1"
);
client_tx
.send(v1_frame(
"f2",
Kind::Request,
Some("s1"),
"weather",
json!({ "city": "hue" }),
))
.await
.unwrap();
let response = client_rx.recv().await.unwrap();
assert_eq!(response.kind, Kind::Response);
assert_eq!(response.corr.as_deref(), Some("s1"));
assert_eq!(response.payload_json()["temp_c"], 21);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_v2_client_wire_without_identity_is_served_immediately() {
let node = weather_node("weather-1");
let (client_side, node_side) = pair();
node.serve_transport(node_side).await;
let client = Wire::open(client_side);
common::ready_client(&client).await;
let mut call = common::stream(&client, "/weather-1/weather", Kind::Request, json!({})).await;
let corr = call.operation().as_str().to_owned();
let response = call.next().await.unwrap().unwrap();
assert_eq!(response.corr.as_deref(), Some(corr.as_str()));
assert_eq!(response.payload_json()["temp_c"], 21);
}
#[tokio::test(flavor = "multi_thread")]
async fn empty_nodes_become_ready_and_a_runtime_feature_is_immediately_callable() {
let hub = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let leaf = Node::builder("leaf")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
connect_nodes(&leaf, "hub", &hub).await;
leaf.add_service(late_ok.at_subject("late.subject"))
.await
.unwrap();
let (client_side, hub_side) = pair();
hub.serve_transport(hub_side).await;
let client = Wire::open(client_side);
common::ready_client(&client).await;
common::wait_until("hub reaches the leaf node", || {
hub.reachable_names().contains(&"leaf".to_string())
})
.await;
let mut call = common::stream(&client, "/leaf/late.subject", Kind::Request, json!({})).await;
let response = call.next().await.unwrap().unwrap();
assert_eq!(response.payload_json()["ok"], true);
}
#[tokio::test(flavor = "multi_thread")]
async fn application_traffic_before_ready_fails_the_candidate() {
let node = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let premature = async {
let mut observer = ghost.observe();
while let Ok(envelope) = observer.recv().await {
if envelope.kind == Kind::Identify {
let _ = ghost
.open_stream("/hub/weather", Kind::Request, json!({}))
.await;
return;
}
}
};
let (connected, ()) = tokio::join!(node.connect_transport("ghost", dial_side), premature);
assert!(
connected.is_err(),
"a request before readiness must fail the candidate"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn identify_after_application_traffic_closes_the_client_session() {
let node = weather_node("weather-1");
let (client_side, node_side) = pair();
node.serve_transport(node_side).await;
let client = Wire::open(client_side);
common::ready_client(&client).await;
let mut call = common::stream(&client, "/weather-1/weather", Kind::Request, json!({})).await;
call.next().await.unwrap().unwrap();
let identity = NodeIdentity {
node_id: "late-node".into(),
instance_id: "late-1".into(),
epoch: 1,
proof: Value::Null,
};
client
.control(
Kind::Identify,
Envelope::encode_payload(&serde_json::to_value(&identity).unwrap()),
)
.await
.unwrap();
client.closed().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn route_control_as_the_first_frame_closes_the_session() {
let node = weather_node("weather-1");
let (client_side, node_side) = pair();
node.serve_transport(node_side).await;
let client = Wire::open(client_side);
common::ready_client(&client).await;
let snapshot = RouteSnapshot::canonical(1, Vec::new());
client
.control(
Kind::RouteSnapshot,
Envelope::encode_payload(&serde_json::to_value(&snapshot).unwrap()),
)
.await
.unwrap();
client.closed().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn a_rejected_peer_identity_fails_connect() {
struct RejectEveryPeer;
impl unb_server::PeerLayer for RejectEveryPeer {
fn admit(
&self,
_request: unb_server::PeerRequest,
_next: unb_server::PeerNext,
) -> std::pin::Pin<
Box<
dyn std::future::Future<
Output = Result<unb_server::PeerRequest, unb_server::HandlerError>,
> + Send
+ '_,
>,
> {
Box::pin(async move {
Err(unb_server::HandlerError::new(
unb_core::ErrorCode::Unauthorized,
"untrusted",
))
})
}
}
let hub = Node::builder("hub")
.peer_layer(RejectEveryPeer)
.build()
.unwrap();
let leaf = Node::builder("leaf")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, accept_side) = pair();
let (connected, _) = tokio::join!(
leaf.connect_transport("hub", dial_side),
hub.serve_transport(accept_side)
);
assert!(connected.is_err(), "the acceptor's policy rejected us");
}
#[tokio::test(flavor = "multi_thread")]
async fn a_malformed_identity_fails_the_candidate() {
let node = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let malformed = async {
let mut observer = ghost.observe();
while let Ok(envelope) = observer.recv().await {
if envelope.kind == Kind::Identify {
let _ = ghost
.control(
Kind::Identify,
Envelope::encode_payload(&json!({ "not": "an identity" })),
)
.await;
return;
}
}
};
let (connected, ()) = tokio::join!(node.connect_transport("ghost", dial_side), malformed);
assert!(connected.is_err());
}
#[tokio::test(flavor = "multi_thread")]
async fn a_malformed_initial_snapshot_fails_the_candidate() {
let node = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let identity = NodeIdentity {
node_id: "ghost".into(),
instance_id: "ghost-1".into(),
epoch: 1,
proof: Value::Null,
};
let bad_snapshot = async {
let mut accepted = false;
let mut observer = ghost.observe();
while let Ok(envelope) = observer.recv().await {
match envelope.kind {
Kind::Identify => {
let payload =
Envelope::encode_payload(&serde_json::to_value(&identity).unwrap());
ghost.control(Kind::Identify, payload).await.unwrap();
ghost
.control(
Kind::IdentityAccepted,
Envelope::encode_payload(&Value::Null),
)
.await
.unwrap();
}
Kind::IdentityAccepted => accepted = true,
_ => {}
}
if accepted {
let snapshot = json!({
"generation": 1,
"routes": [{
"subject": "trap",
"owner": "ghost",
"owner_instance": "ghost-1",
"owner_epoch": 1,
"owner_revision": 0,
"distance": 1,
"path": ["ghost", "hub"]
}]
});
let _ = ghost
.control(Kind::RouteSnapshot, Envelope::encode_payload(&snapshot))
.await;
return;
}
}
};
let (connected, ()) = tokio::join!(node.connect_transport("ghost", dial_side), bad_snapshot);
assert!(
connected.is_err(),
"a snapshot whose path contains the receiver must fail establishment"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_close_during_synchronization_fails_connect_without_hanging() {
let node = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let close_mid_sync = async {
let mut observer = ghost.observe();
while let Ok(envelope) = observer.recv().await {
if envelope.kind == Kind::Identify {
ghost.shutdown();
return;
}
}
};
let started = std::time::Instant::now();
let (connected, ()) = tokio::join!(node.connect_transport("ghost", dial_side), close_mid_sync);
assert!(connected.is_err());
assert!(
started.elapsed() < std::time::Duration::from_secs(10),
"connect must not hang past its deadline"
);
}
async fn close_after_route_snapshot_without_ack(ghost: &Wire) {
let identity = NodeIdentity {
node_id: "ghost".into(),
instance_id: "ghost-1".into(),
epoch: 1,
proof: Value::Null,
};
let mut observer = ghost.observe();
while let Ok(envelope) = observer.recv().await {
match envelope.kind {
Kind::Identify => {
ghost
.control(
Kind::Identify,
Envelope::encode_payload(&serde_json::to_value(&identity).unwrap()),
)
.await
.unwrap();
ghost
.control(
Kind::IdentityAccepted,
Envelope::encode_payload(&Value::Null),
)
.await
.unwrap();
}
Kind::IdentityAccepted => {
let snapshot = RouteSnapshot::canonical(
1,
vec![unb_core::RouteAdvertisement {
destination: "ghost".into(),
owner: "ghost".into(),
owner_instance: "ghost-1".into(),
owner_epoch: 1,
owner_revision: 0,
distance: 0,
path: vec!["ghost".into()],
}],
);
ghost
.control(
Kind::RouteSnapshot,
Envelope::encode_payload(&serde_json::to_value(&snapshot).unwrap()),
)
.await
.unwrap();
}
Kind::RouteSnapshot => {
ghost.shutdown();
return;
}
_ => {}
}
}
}
#[tokio::test(flavor = "multi_thread")]
async fn a_disconnect_before_route_ack_fails_checked_and_unchecked_connect() {
let checked = Node::builder("checked")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let (connected, ()) = tokio::join!(
checked.connect_transport("ghost", dial_side),
close_after_route_snapshot_without_ack(&ghost),
);
assert!(connected.is_err());
let unchecked = Node::builder("unchecked")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let (connected, ()) = tokio::join!(
unchecked.connect_transport_unchecked(dial_side),
close_after_route_snapshot_without_ack(&ghost),
);
assert!(connected.is_err());
}
#[tokio::test(flavor = "multi_thread")]
async fn an_expected_peer_name_mismatch_fails_connect() {
let hub = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let leaf = Node::builder("leaf")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, accept_side) = pair();
let (connected, _) = tokio::join!(
leaf.connect_transport("some-other-node", dial_side),
hub.serve_transport(accept_side)
);
assert!(
connected.is_err(),
"the declared identity does not match the dialed peer name"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_fully_established_ghost_peer_is_routable() {
let node = Node::builder("hub")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (dial_side, ghost_side) = pair();
let ghost = Wire::open(ghost_side);
let (connected, ()) = tokio::join!(
node.connect_transport("ghost", dial_side),
common::ghost_establish_with(&ghost, "ghost", &["ghost.subject"])
);
connected.unwrap();
assert!(node.reachable_names().contains(&"ghost".to_string()));
}
#[test]
fn building_without_an_identity_trust_policy_fails() {
assert!(Node::builder("unconfigured").build().is_err());
}
#[tokio::test(flavor = "multi_thread")]
async fn self_link_is_rejected_even_with_insecure_identity_trust() {
let node = Node::builder("self")
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
assert!(node.link(&node).await.is_err());
}