diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/mod.rs | 64 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 68 | ||||
| -rw-r--r-- | src/transport/sim.rs | 419 |
3 files changed, 532 insertions, 19 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c5769ca..35e3499 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -11,6 +11,8 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +#[cfg(test)] +pub(crate) mod sim; pub mod traffic; /// Random identifier of one run of a node (from start to exit). @@ -86,6 +88,8 @@ pub struct Keepalive { /// Only comparable between nodes on the same epoch. pub heads: HeadsSummary, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub ready: bool, /// The sender would sync with the recipient: counts it as an active member on its @@ -131,6 +135,8 @@ impl std::fmt::Display for BlobId { pub struct BlobMessage { /// The sender's epoch (every message carries it). pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, pub id: BlobId, pub body: BlobBody, } @@ -215,6 +221,17 @@ impl ParsedMessage { ParsedMessage::Blob(blob) => blob.epoch, } } + + /// `SyncEnvelope::epoch_seq`, as every message carries it. + pub fn epoch_seq(&self) -> u32 { + match self { + ParsedMessage::Hello(hello) => hello.epoch_seq, + ParsedMessage::HelloReply(reply) => reply.epoch_seq, + ParsedMessage::Sync(envelope) => envelope.epoch_seq, + ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq, + ParsedMessage::Blob(blob) => blob.epoch_seq, + } + } } #[derive(Debug, Clone)] @@ -223,6 +240,8 @@ pub struct HelloRequest { /// another swarm doesn't answer. pub swarm_id: Option<String>, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, /// The sender is ready (same meaning as `SyncEnvelope::ready`). @@ -236,6 +255,8 @@ pub struct HelloReply { pub swarm_id: String, pub peers: Vec<String>, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, /// The sender is ready. @@ -246,6 +267,8 @@ pub struct HelloReply { /// How long before this HELLO the sender had last heard from the recipient, if ever /// (in milliseconds): tells a restarted node how long it was away. pub away_ms: Option<u64>, + /// The sender's own run (as its HELLOs carry it): a new one means it restarted. + pub sender_run_id: RunId, } #[derive(Debug, Clone)] @@ -253,6 +276,10 @@ pub struct SyncEnvelope { pub swarm_id: String, /// Epoch of the document the message was generated from. pub epoch: Epoch, + /// How many times the sender changed epoch in its current run, when it sent this + /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver + /// ignores an epoch older than the newest it has seen in the same session. + pub epoch_seq: u32, pub message: Message, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a @@ -301,3 +328,40 @@ pub trait Transport: Send + Sync { /// Asynchronously monitors for fatal network client or sub-process failures. async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus>; } + +#[cfg(test)] +mod tests { + use super::*; + use crate::testrand::for_seeds; + + /// Two summaries are equal exactly when the heads are, in any order, whichever form + /// (listed or digest) each took. + #[test] + fn heads_summary_equality_is_set_equality() { + for_seeds(300, |rng| { + let n = rng.between(0, 2 * MAX_LISTED_HEADS + 2); + let heads: Vec<ChangeHash> = (0..n).map(|_| ChangeHash(rng.bytes32())).collect(); + let summary = HeadsSummary::of(heads.clone()); + assert_eq!(summary.heads().is_some(), n <= MAX_LISTED_HEADS); + + let mut shuffled = heads.clone(); + rng.shuffle(&mut shuffled); + assert_eq!(summary, HeadsSummary::of(shuffled)); + assert_eq!(summary, HeadsSummary::Digest(summary.digest())); + + let mut other = heads.clone(); + match rng.below(3) { + 1 if !other.is_empty() => { + let i = rng.below(other.len()); + other.remove(i); + } + 2 if !other.is_empty() => { + let i = rng.below(other.len()); + other[i] = ChangeHash(rng.bytes32()); + } + _ => other.push(ChangeHash(rng.bytes32())), + } + assert_ne!(summary, HeadsSummary::of(other)); + }); + } +} diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 82e6016..efab475 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -561,13 +561,7 @@ impl Transport for MyceliumTransport { } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - let (outbound, payload_len) = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - kind: message.kind(), - payload_len, - }) + encode_message(peer, message) } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { @@ -712,6 +706,26 @@ impl Transport for MyceliumTransport { } } +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let (outbound, payload_len) = format_message(peer, message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + body: serde_json::to_vec(&outbound)?.into(), + kind: message.kind(), + payload_len, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + fn encode_topic(topic: &str) -> String { BASE64.encode(topic.as_bytes()) } @@ -802,16 +816,20 @@ fn format_message( struct EpochShadow { generation: u64, id: ByteBuf, + /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`). + #[serde(default)] + seq: u32, } -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { +fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow { EpochShadow { generation: epoch.generation, id: ByteBuf::from(epoch.id.to_vec()), + seq, } } -fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { +fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> { if shadow.generation == 0 && shadow.id.is_empty() { return Ok(Epoch::GENESIS); } @@ -850,7 +868,7 @@ fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Err rmp_serde::to_vec(&KeepaliveShadow { heads, digest, - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, syncs_with_you: msg.syncs_with_you, }) @@ -880,7 +898,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { }; Ok(Keepalive { heads, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, ready: shadow.ready, syncs_with_you: shadow.syncs_with_you, }) @@ -908,6 +927,7 @@ pub struct HelloReplyShadow<'a> { ready: bool, run_id: ByteBuf, away_ms: Option<u64>, + sender_run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -955,7 +975,7 @@ mod automerge_msg_codec { pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { swarm_id: msg.swarm_id.clone(), - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), @@ -965,7 +985,8 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; Ok(HelloRequest { swarm_id: shadow.swarm_id, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, @@ -984,11 +1005,12 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco let shadow = HelloReplyShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), away_ms: msg.away_ms, + sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()), }; rmp_serde::to_vec(&shadow) } @@ -997,11 +1019,13 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> Ok(HelloReply { swarm_id: shadow.swarm_id.into_owned(), peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, away_ms: shadow.away_ms, + sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?, }) } @@ -1010,7 +1034,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde:: let shadow = SyncEnvelopeShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, }; rmp_serde::to_vec(&shadow) @@ -1019,7 +1043,8 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; Ok(SyncEnvelope { swarm_id: shadow.swarm_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, message: shadow.message, ready: shadow.ready, }) @@ -1084,7 +1109,7 @@ fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> BlobBody::Missing => BlobBodyShadow::Missing, }; rmp_serde::to_vec(&BlobShadow { - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), id, body, }) @@ -1123,7 +1148,8 @@ fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> { BlobBodyShadow::Missing => BlobBody::Missing, }; Ok(BlobMessage { - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, id, body, }) @@ -1181,6 +1207,7 @@ mod tests { generation: u64::MAX, id: [0xff; 32], }, + epoch_seq: u32::MAX, ready: true, syncs_with_you: true, }; @@ -1203,6 +1230,7 @@ mod tests { let id = BlobId::Change(ChangeHash([9; 32])); let chunk = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Chunk { index: 5, @@ -1215,12 +1243,14 @@ mod tests { let decoded = bytes_to_blob(&bytes).unwrap(); assert_eq!(decoded.id, id); assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); assert!( matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) ); let manifest = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Manifest { size: 100, diff --git a/src/transport/sim.rs b/src/transport/sim.rs new file mode 100644 index 0000000..ce5977f --- /dev/null +++ b/src/transport/sim.rs @@ -0,0 +1,419 @@ +//! A simulated network for tests: many nodes in one process, each with a +//! [`SimTransport`] on a shared [`SimNet`]. +//! +//! Messages go through the Mycelium wire encoding (`mycelium::encode_message`), so the +//! codec is exercised as in production. The net delays, loses and duplicates them per +//! link, from a seeded generator; nodes can be taken down (everything to and from them +//! is lost) and brought up again; and a test can inject a message as if a node had sent +//! it, e.g. one that arrives late. +//! +//! Time is tokio's real clock, not a paused one: a node's document work runs on the +//! blocking pool (`spawn_blocking`), which a paused clock doesn't wait for, so timers +//! would fire while the work is still running. Tests use a short keepalive instead. +//! Delays and losses come from the seed; when a message's turn comes still depends on +//! the scheduler, so a run is reproducible in its network decisions, not in its +//! interleavings. + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use async_trait::async_trait; +use bytes::Bytes; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, warn}; + +use super::mycelium::{decode_message, encode_message}; +use super::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport}; +use crate::testrand::Rng; + +/// How one direction of a link behaves. +#[derive(Clone, Debug)] +pub(crate) struct LinkParams { + /// Probability that a message is lost. + pub(crate) loss: f64, + /// Probability that a message arrives twice (each copy with its own delay). + pub(crate) duplicate: f64, + /// Delay, uniform in this range. Overlapping delays reorder messages. + pub(crate) delay: (Duration, Duration), + /// Messages of these kinds are always lost: a peer that answers some messages but + /// not others (e.g. keepalives but not sync). + pub(crate) drop: Vec<MessageKind>, +} + +impl Default for LinkParams { + /// A clean, fast link: nothing lost, a few milliseconds. + fn default() -> Self { + Self { + loss: 0.0, + duplicate: 0.0, + delay: (Duration::from_millis(1), Duration::from_millis(5)), + drop: Vec::new(), + } + } +} + +/// A message on its way into a node. +struct Delivery { + from: String, + body: Bytes, + payload_len: usize, + /// Told once the node has handled the message (injections wait for it). + handled: Option<oneshot::Sender<()>>, +} + +/// What the net counted, for assertions. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct NetCounts { + pub(crate) sent: u64, + pub(crate) delivered: u64, + pub(crate) lost: u64, + pub(crate) duplicated: u64, +} + +struct NetInner { + rng: Rng, + default_link: LinkParams, + links: HashMap<(String, String), LinkParams>, + down: HashSet<String>, + inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>, + counts: NetCounts, +} + +/// The network all simulated nodes share. Cloning is cheap. +#[derive(Clone)] +pub(crate) struct SimNet { + inner: Arc<Mutex<NetInner>>, +} + +impl SimNet { + pub(crate) fn new(seed: u64) -> Self { + Self { + inner: Arc::new(Mutex::new(NetInner { + rng: Rng::new(seed), + default_link: LinkParams::default(), + links: HashMap::new(), + down: HashSet::new(), + inboxes: HashMap::new(), + counts: NetCounts::default(), + })), + } + } + + fn lock(&self) -> MutexGuard<'_, NetInner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// A transport for a new node `id` on this net. + pub(crate) fn transport(&self, id: &str) -> Arc<SimTransport> { + let (tx, rx) = mpsc::unbounded_channel(); + let previous = self.lock().inboxes.insert(id.to_owned(), tx); + assert!(previous.is_none(), "node {id} is already on the net"); + Arc::new(SimTransport { + id: id.to_owned(), + net: self.clone(), + inbox: tokio::sync::Mutex::new(Some(rx)), + }) + } + + /// How links without their own parameters behave. + pub(crate) fn set_default_link(&self, params: LinkParams) { + self.lock().default_link = params; + } + + /// How messages from `from` to `to` are carried (one direction). + pub(crate) fn set_link(&self, from: &str, to: &str, params: LinkParams) { + self.lock() + .links + .insert((from.to_owned(), to.to_owned()), params); + } + + /// Everything to and from `id` is lost until [`bring_up`](Self::bring_up), including + /// messages already on their way. The node itself keeps running. + pub(crate) fn take_down(&self, id: &str) { + self.lock().down.insert(id.to_owned()); + } + + pub(crate) fn bring_up(&self, id: &str) { + self.lock().down.remove(id); + } + + pub(crate) fn counts(&self) -> NetCounts { + self.lock().counts + } + + /// Hands a message to the net: decides now whether and when it arrives. + fn send(&self, from: &str, message: &EncodedMessage) { + let delays = { + let mut guard = self.lock(); + let inner = &mut *guard; + inner.counts.sent += 1; + if inner.down.contains(from) || inner.down.contains(&message.peer) { + inner.counts.lost += 1; + return; + } + let params = inner + .links + .get(&(from.to_owned(), message.peer.clone())) + .unwrap_or(&inner.default_link) + .clone(); + if params.drop.contains(&message.kind) || inner.rng.chance(params.loss) { + inner.counts.lost += 1; + return; + } + let copies = if inner.rng.chance(params.duplicate) { + inner.counts.duplicated += 1; + 2 + } else { + 1 + }; + (0..copies) + .map(|_| inner.rng.duration_in(params.delay)) + .collect::<Vec<_>>() + }; + for delay in delays { + let net = self.clone(); + let delivery = Delivery { + from: from.to_owned(), + body: message.body.clone(), + payload_len: message.payload_len, + handled: None, + }; + let to = message.peer.clone(); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + net.arrive(&to, delivery); + }); + } + } + + /// The message reaches `to`, unless either end went down meanwhile. + fn arrive(&self, to: &str, delivery: Delivery) { + let mut guard = self.lock(); + let inner = &mut *guard; + if inner.down.contains(to) || inner.down.contains(&delivery.from) { + inner.counts.lost += 1; + return; + } + // An unknown recipient (an address no node has) loses the message, as + // Mycelium would. + let Some(inbox) = inner.inboxes.get(to) else { + inner.counts.lost += 1; + return; + }; + if inbox.send(delivery).is_ok() { + inner.counts.delivered += 1; + } else { + inner.counts.lost += 1; + } + } + + /// Delivers `message` to `to` as if `from` had sent it, now, bypassing links and + /// `take_down`; returns once `to` has handled it. `from` needn't be a node on the + /// net. For messages a real node wouldn't send at that moment, e.g. one sent long + /// ago and delivered late. + pub(crate) async fn inject(&self, from: &str, to: &str, message: &ParsedMessage) -> Result<()> { + let encoded = encode_message(to, message)?; + let (handled_tx, handled_rx) = oneshot::channel(); + let delivery = Delivery { + from: from.to_owned(), + body: encoded.body, + payload_len: encoded.payload_len, + handled: Some(handled_tx), + }; + { + let inner = self.lock(); + let Some(inbox) = inner.inboxes.get(to) else { + bail!("no node {to} on the net"); + }; + if inbox.send(delivery).is_err() { + bail!("node {to} has no consumer running"); + } + } + handled_rx.await?; + Ok(()) + } + + fn is_known(&self, id: &str) -> bool { + self.lock().inboxes.contains_key(id) + } +} + +/// One node's end of the [`SimNet`]. Its identity is the ID it was created with; seeds +/// are node IDs. +pub(crate) struct SimTransport { + id: String, + net: SimNet, + inbox: tokio::sync::Mutex<Option<mpsc::UnboundedReceiver<Delivery>>>, +} + +#[async_trait] +impl Transport for SimTransport { + fn local_id(&self) -> &str { + &self.id + } + + /// A seed is a node ID; one no node has doesn't resolve (yet). + async fn resolve_seed(&self, seed: &str) -> Result<String> { + if self.net.is_known(seed) { + Ok(seed.to_owned()) + } else { + bail!("no node {seed} on the net") + } + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + encode_message(peer, message) + } + + /// Always accepted: the net loses messages silently, like Mycelium once it has + /// taken them. + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + self.net.send(&self.id, message); + Ok(()) + } + + /// Handles messages one at a time, in arrival order, as the Mycelium consumer does. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let Some(mut inbox) = self.inbox.lock().await.take() else { + bail!("consumer of {} already started", self.id); + }; + while let Some(delivery) = inbox.recv().await { + match decode_message(&delivery.body) { + Ok(message) => { + debug!(node = %self.id, from = %delivery.from, message = ?message, "Simulated delivery"); + if let Err(error) = handler(delivery.from, message, delivery.payload_len).await + { + warn!(node = %self.id, error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Err(error) => { + warn!(node = %self.id, error = %format!("{error:#}"), "Undecodable simulated message"); + } + } + if let Some(handled) = delivery.handled { + let _ = handled.send(()); + } + } + Ok(()) + } + + /// A simulated node has no daemon that could exit. + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + std::future::pending().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::document::replica::Epoch; + use crate::transport::{HeadsSummary, Keepalive}; + + fn ping() -> ParsedMessage { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: true, + syncs_with_you: false, + }) + } + + /// Collects what a node receives. + fn record(transport: Arc<SimTransport>) -> Arc<Mutex<Vec<String>>> { + let received = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&received); + tokio::spawn(async move { + transport + .start_consumer(Box::new(move |from, _message, _len| { + sink.lock().unwrap().push(from); + Box::pin(async { Ok(()) }) + })) + .await + }); + received + } + + #[tokio::test] + async fn delivers_decodes_and_loses_when_down() { + let net = SimNet::new(1); + let a = net.transport("a"); + let received = record(net.transport("b")); + + let settle = || tokio::time::sleep(Duration::from_millis(30)); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.take_down("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.bring_up("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + + assert_eq!(*received.lock().unwrap(), vec!["a", "a"]); + let counts = net.counts(); + assert_eq!((counts.sent, counts.delivered, counts.lost), (3, 2, 1)); + assert_eq!(a.resolve_seed("b").await.unwrap(), "b"); + assert!(a.resolve_seed("c").await.is_err()); + } + + #[tokio::test] + async fn loss_and_duplication_follow_the_seed() { + let run = |seed| async move { + let net = SimNet::new(seed); + net.set_default_link(LinkParams { + loss: 0.3, + duplicate: 0.2, + ..LinkParams::default() + }); + let a = net.transport("a"); + let _b = net.transport("b"); + for _ in 0..200 { + a.send_message("b", &ping()).await.unwrap(); + } + tokio::time::sleep(Duration::from_millis(50)).await; + net.counts() + }; + let first = run(7).await; + assert_eq!(first, run(7).await); + assert!(first.lost > 30 && first.lost < 90, "{first:?}"); + assert!(first.duplicated > 10, "{first:?}"); + } + + #[tokio::test] + async fn a_link_overrides_the_default_in_one_direction() { + let net = SimNet::new(1); + net.set_link( + "a", + "b", + LinkParams { + loss: 1.0, + ..LinkParams::default() + }, + ); + let a = net.transport("a"); + let b = net.transport("b"); + let at_a = record(Arc::clone(&a)); + let at_b = record(Arc::clone(&b)); + + a.send_message("b", &ping()).await.unwrap(); + b.send_message("a", &ping()).await.unwrap(); + tokio::time::sleep(Duration::from_millis(30)).await; + + assert!(at_b.lock().unwrap().is_empty()); + assert_eq!(*at_a.lock().unwrap(), vec!["b"]); + } + + #[tokio::test] + async fn inject_returns_once_handled() { + let net = SimNet::new(1); + let received = record(net.transport("b")); + net.inject("ghost", "b", &ping()).await.unwrap(); + assert_eq!(*received.lock().unwrap(), vec!["ghost"]); + assert!(net.inject("ghost", "nobody", &ping()).await.is_err()); + } +} |