4 files changed, 170 insertions, 28 deletions

diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 10121bb..7abcced 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -509,6 +509,13 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco
     })
 }
 
+/// Mycelium's envelope, for simulations that run on it (`transport::sim::Codec`).
+#[cfg(test)]
+pub(crate) const CODEC: crate::transport::sim::Codec = crate::transport::sim::Codec {
+    encode: encode_message,
+    decode: decode_message,
+};
+
 /// The message in a body made by [`encode_message`].
 #[cfg(test)]
 pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> {
@@ -702,20 +709,51 @@ mod tests {
         ));
     }
 
+    /// Every kind decodes from Mycelium's API body (its envelope: the topic, the payload
+    /// in base64) to what was encoded.
     #[test]
-    fn messages_round_trip_through_the_api_body() {
-        let status = ParsedMessage::Status(Status {
-            heads: HeadsSummary::of(vec![ChangeHash([1; 32])]),
-            epoch: Epoch::GENESIS,
-            epoch_seq: 0,
-            ready: true,
-            syncs_with_you: true,
-        });
-        let encoded = encode_message("peer", &status).unwrap();
-        assert!(matches!(
-            decode_message(&encoded.body).unwrap(),
-            ParsedMessage::Status(_)
-        ));
+    fn every_kind_round_trips_through_the_api_body() {
+        for message in wire::samples() {
+            let encoded = encode_message("peer", &message).unwrap();
+            assert_eq!(encoded.kind, message.kind());
+            let decoded = decode_message(&encoded.body).unwrap();
+            assert_eq!(
+                wire::encode(&decoded).unwrap(),
+                wire::encode(&message).unwrap(),
+                "{message:?}"
+            );
+            assert_eq!(encoded.payload_len, wire::encode(&message).unwrap().len());
+        }
+    }
+
+    /// A swarm whose every message goes through Mycelium's envelope: joining, writes
+    /// reaching everyone, and concurrent writes merging, as the simulations do with the
+    /// wire encoding alone.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn a_swarm_converges_over_the_envelope() {
+        use crate::simtest::{Sim, read, write};
+        use crate::swarm::bootstrap::Start;
+        use serde_json::json;
+        const SOON: Duration = Duration::from_secs(10);
+
+        let sim = Sim::with_codec(11, CODEC);
+        let a = sim.start("a", &[], Start::Genesis);
+        let b = sim.start("b", &["a"], Start::Join);
+        let c = sim.start("c", &["a"], Start::Join);
+        let (a, b, c) = (&a, &b, &c);
+        sim.wait_ready(&[a, b, c], SOON).await;
+        write(b, "users/bob", json!({"name": "Bob"})).await;
+        sim.wait_until("c has b's write", SOON, move || async move {
+            read(c, "users/bob/name").await == Some(json!("Bob"))
+        })
+        .await;
+        write(a, "users/alice", json!(1)).await;
+        write(c, "users/carol", json!(3)).await;
+        sim.wait_converged(&[a, b, c], SOON).await;
+        for node in [a, b, c] {
+            assert_eq!(read(node, "users/alice").await, Some(json!(1)));
+            assert_eq!(read(node, "users/carol").await, Some(json!(3)));
+        }
     }
 
     #[test]
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 4d0e775..a8c7c49 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -1228,7 +1228,7 @@ mod tests {
             Ok(seed.to_owned())
         }
         fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-            crate::transport::sim::encode_message(peer, message)
+            (crate::transport::sim::WIRE.encode)(peer, message)
         }
         async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> {
             Err(anyhow!("broken"))
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 2410bba..a558e9f 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -1,9 +1,9 @@
 //! 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; without the mycelium feature, through the wire
-//! encoding alone, as libp2p and Tor send it. The net delays, loses and duplicates them per
+//! Messages go through the wire encoding (`WIRE`, as libp2p and Tor send it), so the
+//! codec is exercised as in production; a net can use a transport's envelope instead
+//! (`SimNet::with_codec`; Mycelium's tests run a swarm on its envelope). 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.
@@ -25,8 +25,6 @@ use bytes::Bytes;
 use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, warn};
 
-#[cfg(feature = "mycelium")]
-pub(crate) use super::mycelium::{decode_message, encode_message};
 use super::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport};
 use crate::locks::MutexExt;
 use crate::testrand::Rng;
@@ -88,11 +86,19 @@ struct NetInner {
 #[derive(Clone)]
 pub(crate) struct SimNet {
     inner: Arc<Mutex<NetInner>>,
+    codec: Codec,
 }
 
 impl SimNet {
+    /// Messages in the wire encoding alone (`WIRE`).
     pub(crate) fn new(seed: u64) -> Self {
+        Self::with_codec(seed, WIRE)
+    }
+
+    /// Messages put into bytes by `codec`.
+    pub(crate) fn with_codec(seed: u64, codec: Codec) -> Self {
         Self {
+            codec,
             inner: Arc::new(Mutex::new(NetInner {
                 rng: Rng::new(seed),
                 default_link: LinkParams::default(),
@@ -217,7 +223,7 @@ impl SimNet {
     /// 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 encoded = (self.codec.encode)(to, message)?;
         let (handled_tx, handled_rx) = oneshot::channel();
         let delivery = Delivery {
             from: from.to_owned(),
@@ -251,9 +257,22 @@ pub(crate) struct SimTransport {
     inbox: tokio::sync::Mutex<Option<mpsc::UnboundedReceiver<Delivery>>>,
 }
 
-/// Without Mycelium: a message in the wire encoding alone, as libp2p and Tor send it.
-#[cfg(not(feature = "mycelium"))]
-pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+/// How a simulated net puts messages into bytes and back (`SimNet::with_codec`): the
+/// wire encoding alone (`WIRE`, as libp2p and Tor send it), or a transport's envelope
+/// around it (Mycelium's: `mycelium::CODEC`), so a whole swarm runs on that envelope.
+#[derive(Clone, Copy)]
+pub(crate) struct Codec {
+    pub(crate) encode: fn(&str, &ParsedMessage) -> Result<EncodedMessage>,
+    pub(crate) decode: fn(&[u8]) -> Result<ParsedMessage>,
+}
+
+/// The wire encoding alone (`transport::wire`).
+pub(crate) const WIRE: Codec = Codec {
+    encode: wire_encode,
+    decode: wire_decode,
+};
+
+fn wire_encode(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
     let body = super::wire::encode(message)?;
     Ok(EncodedMessage {
         peer: peer.to_owned(),
@@ -265,9 +284,7 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco
     })
 }
 
-/// The message in a body made by [`encode_message`].
-#[cfg(not(feature = "mycelium"))]
-pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> {
+fn wire_decode(body: &[u8]) -> Result<ParsedMessage> {
     Ok(super::wire::decode(body)?)
 }
 
@@ -296,7 +313,7 @@ impl Transport for SimTransport {
     }
 
     fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        encode_message(peer, message)
+        (self.net.codec.encode)(peer, message)
     }
 
     /// Always accepted: the net loses messages silently, like Mycelium once it has
@@ -312,7 +329,7 @@ impl Transport for SimTransport {
             bail!("consumer of {} already started", self.id);
         };
         while let Some(delivery) = inbox.recv().await {
-            match decode_message(&delivery.body) {
+            match (self.net.codec.decode)(&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
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index fcbee85..53fd460 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -325,6 +325,80 @@ pub(crate) mod codec {
     }
 }
 
+/// One message of every kind, for the codecs' tests (this one's, and each transport's
+/// envelope around it).
+#[cfg(test)]
+pub(crate) fn samples() -> Vec<ParsedMessage> {
+    use crate::document::replica::Epoch;
+    use crate::transport::{
+        BlobBody, BlobId, BlobMessage, DeltaSpec, HeadsSummary, HelloReply, HelloRequest,
+        IdentityProof, ListedPeer, Status, SyncBody, SyncEnvelope,
+    };
+    use automerge::ChangeHash;
+    let epoch = Epoch {
+        generation: 2,
+        id: [3; 32],
+    };
+    let status = Status {
+        heads: HeadsSummary::of(vec![ChangeHash([1; 32])]),
+        epoch,
+        epoch_seq: 1,
+        ready: true,
+        syncs_with_you: true,
+    };
+    let proof = IdentityProof {
+        key: [8; 32],
+        signature: [9; 64],
+    };
+    vec![
+        ParsedMessage::Hello(HelloRequest {
+            swarm_id: Some("swarm".into()),
+            epoch,
+            epoch_seq: 1,
+            auth: vec![1, 2, 3],
+            ready: true,
+            run_id: [4; 16],
+            handshake: 7,
+        }),
+        ParsedMessage::HelloReply(HelloReply {
+            swarm_id: "swarm".into(),
+            peers: vec![ListedPeer {
+                id: "p".into(),
+                addresses: vec!["a".into()],
+            }],
+            epoch,
+            epoch_seq: 1,
+            auth: Vec::new(),
+            ready: false,
+            run_id: [4; 16],
+            away_ms: Some(5),
+            sender_run_id: [6; 16],
+        }),
+        ParsedMessage::Sync(SyncEnvelope {
+            swarm_id: "swarm".into(),
+            epoch,
+            epoch_seq: 1,
+            body: SyncBody::Delta(DeltaSpec {
+                from: Vec::new(),
+                to: vec![ChangeHash([2; 32])],
+            }),
+            ready: true,
+        }),
+        ParsedMessage::Status(status.clone()),
+        ParsedMessage::StatusReply(status),
+        ParsedMessage::Blob(BlobMessage {
+            epoch,
+            epoch_seq: 1,
+            id: BlobId::Content([5; 32]),
+            body: BlobBody::Missing,
+        }),
+        ParsedMessage::Who(proof.clone()),
+        ParsedMessage::Me(proof),
+        ParsedMessage::Ping(7),
+        ParsedMessage::Pong(u64::MAX),
+    ]
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -335,6 +409,19 @@ mod tests {
     };
     use automerge::ChangeHash;
 
+    /// Every kind decodes to what was encoded: re-encoded, the same bytes.
+    #[test]
+    fn every_kind_round_trips() {
+        let samples = samples();
+        assert_eq!(samples.len(), 10, "a kind is missing from `samples`");
+        for message in samples {
+            let bytes = encode(&message).unwrap();
+            let decoded = decode(&bytes).unwrap();
+            assert_eq!(decoded.kind(), message.kind());
+            assert_eq!(encode(&decoded).unwrap(), bytes, "{message:?}");
+        }
+    }
+
     #[test]
     fn handshakes_round_trip() {
         let epoch = Epoch {