1 files changed, 51 insertions, 13 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]