bump number of heads in keepalive messages

Instead of arbitrarily limiting the number to 16, allow filling
keepalive messages as long as they don't cause mycelium to chunk it.

6 files changed, 1069 insertions, 11 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index ce41776..3317185 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -57,12 +57,18 @@ pub(crate) struct PeerStatus {
     epoch: Option<EpochInfo>,
     /// Whether that epoch equals ours, i.e. whether we sync with this peer.
     same_epoch: Option<bool>,
+    /// Number of heads in its last keepalive; absent before one, or when it had more
+    /// than 32 (then only their digest is sent).
+    heads: Option<usize>,
 }
 
 #[derive(Debug, Serialize, ToSchema)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerListResponse {
     local_epoch: EpochInfo,
+    /// Number of heads of our document. Several heads are concurrent changes no later
+    /// change has merged yet.
+    local_heads: usize,
     peers: Vec<PeerStatus>,
 }
 
@@ -122,12 +128,15 @@ pub(crate) async fn list_peers(
             has_pending_delivery: ps.sync.outbound.is_some(),
             epoch: ps.epoch.map(EpochInfo::from),
             same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
+            heads: ps.reported_heads,
         });
     }
 
     peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
+    let local_heads = state.doc.write().await.doc.get_heads().len();
     Ok(Json(PeerListResponse {
         local_epoch: local_epoch.into(),
+        local_heads,
         peers,
     }))
 }
diff --git a/src/state.rs b/src/state.rs
index 8ff8b9d..4d82150 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -126,6 +126,9 @@ pub(crate) struct PeerState {
 
     /// Whether the peer said, in its most recent keepalive, that it would sync with us.
     pub(crate) syncs_with_us: Option<bool>,
+
+    /// How many heads its most recent keepalive listed (None if it sent a digest).
+    pub(crate) reported_heads: Option<usize>,
 }
 
 /// One Automerge sync with a peer, from a reset to the next. Everything a reset must
@@ -330,6 +333,7 @@ impl PeerState {
             epoch: None,
             reports_ready: None,
             syncs_with_us: None,
+            reported_heads: None,
         }
     }
 
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 9fe9b07..368c969 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -131,6 +131,7 @@ async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepal
         let mut peer_state = peer_state.lock().await;
         peer_state.reports_ready = Some(remote.ready);
         peer_state.syncs_with_us = Some(remote.syncs_with_you);
+        peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
     }
     crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
 }
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index aee3ee3..d7e7135 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -12,14 +12,16 @@ pub use crate::document::replica::Epoch;
 pub mod firewall;
 pub mod mycelium;
 
-/// Most heads listed in a keepalive; more are sent as a digest. 16 heads are about
-/// 520 bytes, well within one Mycelium chunk (1300 bytes).
-pub const MAX_LISTED_HEADS: usize = 16;
+/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on
+/// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see
+/// the size test in `mycelium`). There are rarely more heads than nodes that wrote
+/// concurrently since the last merging write.
+pub const MAX_LISTED_HEADS: usize = 32;
 
 /// A document's heads, compactly: the heads themselves while there are few, otherwise a
-/// digest of them. Either way, two summaries are equal exactly when the heads are (the
-/// number of heads decides the form, so equal heads always take the same one).
-#[derive(Debug, Clone, PartialEq, Eq)]
+/// digest of them. Two summaries are equal exactly when the heads are, whatever form
+/// each took (nodes with different limits still agree).
+#[derive(Debug, Clone)]
 pub enum HeadsSummary {
     /// Sorted.
     Heads(Vec<ChangeHash>),
@@ -31,13 +33,18 @@ impl HeadsSummary {
     pub fn of(mut heads: Vec<ChangeHash>) -> Self {
         heads.sort_unstable();
         if heads.len() <= MAX_LISTED_HEADS {
-            return Self::Heads(heads);
+            Self::Heads(heads)
+        } else {
+            Self::Digest(digest_of(&heads))
         }
-        let mut hasher = Sha256::new();
-        for head in &heads {
-            hasher.update(head.0);
+    }
+
+    /// The digest of the heads, computed if they are listed.
+    pub fn digest(&self) -> [u8; 32] {
+        match self {
+            Self::Heads(heads) => digest_of(heads),
+            Self::Digest(digest) => *digest,
         }
-        Self::Digest(hasher.finalize().into())
     }
 
     /// The heads, if listed.
@@ -49,6 +56,26 @@ impl HeadsSummary {
     }
 }
 
+impl PartialEq for HeadsSummary {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Heads(a), Self::Heads(b)) => a == b,
+            _ => self.digest() == other.digest(),
+        }
+    }
+}
+
+impl Eq for HeadsSummary {}
+
+/// SHA-256 over sorted heads.
+fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
+    let mut hasher = Sha256::new();
+    for head in sorted_heads {
+        hasher.update(head.0);
+    }
+    hasher.finalize().into()
+}
+
 /// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
 #[derive(Debug, Clone)]
 pub struct Keepalive {
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 9d1fb90..2683f82 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -973,3 +973,45 @@ pub enum TransportError {
         status: reqwest::StatusCode,
     },
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Mycelium splits messages into chunks of this size.
+    const MYCELIUM_CHUNK: usize = 1300;
+
+    #[test]
+    fn largest_keepalive_fits_in_one_chunk() {
+        let keepalive = Keepalive {
+            heads: HeadsSummary::Heads(
+                (0..MAX_LISTED_HEADS as u8)
+                    .map(|n| ChangeHash([n; 32]))
+                    .collect(),
+            ),
+            epoch: Epoch {
+                generation: u64::MAX,
+                id: [0xff; 32],
+            },
+            ready: true,
+            syncs_with_you: true,
+        };
+        let payload = keepalive_to_bytes(&keepalive).unwrap();
+        // Mycelium's message data: a length byte, the topic, then our payload.
+        let size = 1 + TOPIC_PONG.len().max(TOPIC_PING.len()) + payload.len();
+        assert!(
+            size + 50 <= MYCELIUM_CHUNK,
+            "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
+        );
+        assert_eq!(bytes_to_keepalive(&payload).unwrap().heads, keepalive.heads);
+    }
+
+    #[test]
+    fn listed_heads_equal_their_digest() {
+        let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect();
+        let listed = HeadsSummary::of(heads.clone());
+        let digest = HeadsSummary::Digest(listed.digest());
+        assert_eq!(listed, digest);
+        assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec()));
+    }
+}
diff --git a/src/transport/mycelium.rs.orig b/src/transport/mycelium.rs.orig
new file mode 100644
index 0000000..9d1fb90
--- /dev/null
+++ b/src/transport/mycelium.rs.orig
@@ -0,0 +1,975 @@
+use anyhow::{Context, Result, anyhow, bail};
+use async_trait::async_trait;
+use automerge::ChangeHash;
+use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
+use reqwest::StatusCode;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde_bytes::{ByteBuf, Bytes};
+use std::borrow::Cow;
+use std::boxed::Box;
+use std::fs::File;
+use std::io::Write;
+use std::net::TcpListener;
+use std::os::unix::io::FromRawFd;
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::time::Duration;
+use thiserror::Error;
+use tokio::process::{Child, Command};
+use url::Url;
+
+use crate::transport::EncodedMessage;
+use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
+use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS};
+use crate::transport::{MessageHandler, ParsedMessage, Transport};
+
+const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
+
+const TOPIC_PREFIX: &str = "continuum.v1.";
+const TOPIC_HELLO: &str = "continuum.v1.hello";
+const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
+const TOPIC_SYNC: &str = "continuum.v1.sync";
+const TOPIC_PING: &str = "continuum.v1.ping";
+const TOPIC_PONG: &str = "continuum.v1.pong";
+
+const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct InboundMessage {
+    id: String,
+    src_pk: String,
+    topic: String,
+    payload: String,
+}
+
+// Deserialize only for debugging
+#[derive(Debug, Serialize, Deserialize)]
+struct OutboundMessage {
+    dst: Destination,
+    topic: String,
+    payload: String,
+}
+
+// Deserialize only for debugging
+#[derive(Debug, Serialize, Deserialize)]
+struct Destination {
+    pk: String,
+}
+
+/// Represents where a 32-byte cryptographic key originates.
+#[derive(Debug, Clone)]
+pub enum KeySource {
+    Raw([u8; 32]),
+    Path(PathBuf),
+}
+
+/// Atomic structure ensuring private network credentials are always handled together.
+#[derive(Debug, Clone)]
+pub struct PrivateNetworkConfig {
+    pub name: String,
+    pub key: KeySource,
+}
+
+/// Comprehensive parameters required to spawn or target an instance.
+#[derive(Debug, Clone)]
+pub struct MyceliumSpawnConfig {
+    pub rest_url_opt: Option<Url>,
+    pub key_file_opt: Option<PathBuf>,
+    pub network_key_file: Option<PathBuf>,
+    pub network_name: String,
+    pub tcp_port: Option<u16>,
+    pub api_port: Option<u16>,
+    pub rpc_port: Option<u16>,
+    pub discovery_port: Option<u16>,
+    pub peers: Vec<String>,
+}
+
+impl Default for MyceliumSpawnConfig {
+    fn default() -> Self {
+        Self {
+            rest_url_opt: None,
+            key_file_opt: None,
+            network_key_file: None,
+            network_name: "something".to_string(),
+            tcp_port: None,
+            api_port: None,
+            rpc_port: None,
+            discovery_port: None,
+            peers: vec!["tls://127.0.0.1:9651".to_string()],
+        }
+    }
+}
+
+pub enum MyceliumRuntime {
+    Supervised(Box<SupervisedMycelium>),
+    External,
+}
+
+pub struct MyceliumSetup {
+    pub runtime: MyceliumRuntime,
+    pub rest_url: Url,
+}
+
+impl MyceliumSetup {
+    /// Evaluates the configuration payload and initializes the chosen execution strategy.
+    pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
+        // TODO warn for all ignored options if rest_url_opt is set
+        if let Some(url) = config.rest_url_opt {
+            tracing::info!(
+                "Configured to interface with an external standalone Mycelium instance at: {}",
+                url
+            );
+            return Ok(Self {
+                runtime: MyceliumRuntime::External,
+                rest_url: url,
+            });
+        }
+
+        tracing::info!(
+            "Configured to initialize an embedded user-space Mycelium background worker"
+        );
+
+        let key = match config.key_file_opt {
+            Some(key_file) => KeySource::Path(key_file),
+            None => {
+                let mut bytes = [0u8; 32];
+                getrandom::fill(&mut bytes)?;
+                KeySource::Raw(bytes)
+            }
+        };
+
+        let private_net_config = config
+            .network_key_file
+            .map(|net_key_path| PrivateNetworkConfig {
+                name: config.network_name.clone(),
+                key: KeySource::Path(net_key_path),
+            });
+
+        let supervised_node = SupervisedMycelium::spawn(
+            key,
+            private_net_config,
+            config.tcp_port,
+            config.api_port,
+            config.rpc_port,
+            config.discovery_port,
+            &config.peers,
+        )
+        .await
+        .context("Failed to securely bootstrap embedded Mycelium background worker")?;
+
+        let resolved_url = Url::parse(&format!(
+            "http://127.0.0.1:{}",
+            supervised_node.config.api_port
+        ))?;
+
+        Ok(Self {
+            runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)),
+            rest_url: resolved_url,
+        })
+    }
+
+    /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely.
+    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        match &mut self.runtime {
+            MyceliumRuntime::Supervised(node) => node.wait_for_exit().await,
+            MyceliumRuntime::External => {
+                std::future::pending::<std::io::Result<std::process::ExitStatus>>().await
+            }
+        }
+    }
+}
+
+/// The part of Mycelium's `/api/v1/admin` answer we use.
+#[derive(Deserialize, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct MyceliumAdminInfo {
+    pub node_pubkey: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct MyceliumNodeConfig {
+    pub tcp_port: u16,
+    pub api_port: u16,
+    pub rpc_port: u16,
+    pub discovery_port: Option<u16>,
+    pub private_network: Option<String>,
+}
+
+pub struct SupervisedMycelium {
+    child: Child,
+    pub config: MyceliumNodeConfig,
+    _key_file_handle: Option<File>,
+    _net_key_file_handle: Option<File>,
+}
+
+impl SupervisedMycelium {
+    async fn spawn(
+        node_key: KeySource,
+        private_network: Option<PrivateNetworkConfig>,
+        explicit_tcp: Option<u16>,
+        explicit_api: Option<u16>,
+        explicit_rpc: Option<u16>,
+        explicit_discovery_port: Option<u16>,
+        peers: &[String],
+    ) -> Result<Self> {
+        // Linux only: keys are handed over in memory (memfd, never on disk), and the
+        // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself.
+        if cfg!(not(target_os = "linux")) {
+            bail!(
+                "Supervising Mycelium is only supported on Linux; run myceliumd yourself \
+                 and pass --mycelium-rest-url"
+            );
+        }
+
+        let tcp_port = match explicit_tcp {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate TCP wire port")?,
+        };
+        let api_port = match explicit_api {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate API port")?,
+        };
+        let rpc_port = match explicit_rpc {
+            Some(p) => p,
+            None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?,
+        };
+        let discovery_port = match explicit_discovery_port {
+            Some(0) => Some(
+                Self::find_available_port().context("Failed to allocate peer discovery port")?,
+            ),
+            p => p,
+        };
+
+        let config = MyceliumNodeConfig {
+            tcp_port,
+            api_port,
+            rpc_port,
+            discovery_port,
+            private_network: private_network.as_ref().map(|pn| pn.name.clone()),
+        };
+
+        let parent_pid = std::process::id();
+
+        let (key_file_handle, key_file_path, node_fd) =
+            Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?;
+
+        let (net_key_file_handle, net_key_file_path, net_fd) = match private_network {
+            Some(net_cfg) => {
+                let (h, p, fd) =
+                    Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
+                (h, Some(p), fd)
+            }
+            None => (None, None, None),
+        };
+
+        let api_addr = format!("127.0.0.1:{}", config.api_port);
+        let rpc_addr = format!("127.0.0.1:{}", config.rpc_port);
+
+        let mut cmd = Command::new("mycelium");
+
+        let fds_to_inherit: Vec<i32> = [node_fd, net_fd].iter().filter_map(|&x| x).collect();
+        // The child must never outlive us. An orphaned Mycelium keeps running with our
+        // key; the next one we start then shares its address, and replies (including
+        // Mycelium's own message ACKs) go to whichever instance the routers picked,
+        // leaving our messages pending forever. Drop only runs on a clean shutdown, so
+        // let the kernel signal the child when we die, however we die.
+        let parent_pid = std::process::id() as libc::pid_t;
+        unsafe {
+            cmd.pre_exec(move || {
+                for &fd in &fds_to_inherit {
+                    libc::fcntl(fd, libc::F_SETFD, 0);
+                }
+                // Own process group: a Ctrl-C in the terminal goes to continuum only,
+                // which then stops Mycelium itself, instead of both exiting at once (which
+                // shows up as "Mycelium runtime process exited").
+                if libc::setpgid(0, 0) != 0 {
+                    return Err(std::io::Error::last_os_error());
+                }
+                #[cfg(target_os = "linux")]
+                {
+                    if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 {
+                        return Err(std::io::Error::last_os_error());
+                    }
+                    // We may have died between fork and prctl.
+                    if libc::getppid() != parent_pid {
+                        libc::_exit(1);
+                    }
+                }
+                Ok(())
+            });
+        }
+
+        cmd.arg("--key-file")
+            .arg(&key_file_path)
+            .arg("--tcp-listen-port")
+            .arg(config.tcp_port.to_string())
+            .arg("--disable-quic")
+            .arg("--api-addr")
+            .arg(&api_addr)
+            .arg("--jsonrpc-addr")
+            .arg(&rpc_addr)
+            .arg("--no-tun");
+
+        if let Some(ref discovery_port) = config.discovery_port {
+            cmd.arg("--peer-discovery-port")
+                .arg(discovery_port.to_string());
+        } else {
+            cmd.arg("--disable-peer-discovery");
+        }
+
+        if let Some(ref net_name) = config.private_network {
+            cmd.arg("--network-name").arg(net_name);
+        }
+        if let Some(ref net_key_path) = net_key_file_path {
+            cmd.arg("--network-key-file").arg(net_key_path);
+        }
+
+        if !peers.is_empty() {
+            cmd.arg("--peers").arg(peers.join(" "));
+        }
+
+        cmd.env("RUST_LOG", "info");
+
+        cmd.stdout(Stdio::piped());
+        cmd.stderr(Stdio::piped());
+
+        let mut child = cmd
+            .spawn()
+            .context("Failed to execute mycelium process daemon")?;
+
+        // 1. FIXED: Pass-through streaming for child process STDOUT
+        if let Some(mut stdout) = child.stdout.take() {
+            tokio::spawn(async move {
+                let mut host_stdout = tokio::io::stdout();
+                let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await;
+            });
+        }
+
+        // 2. FIXED: Pass-through streaming for child process STDERR
+        if let Some(mut stderr) = child.stderr.take() {
+            tokio::spawn(async move {
+                let mut host_stderr = tokio::io::stderr();
+                let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await;
+            });
+        }
+
+        let node = Self {
+            child,
+            config,
+            _key_file_handle: key_file_handle,
+            _net_key_file_handle: net_key_file_handle,
+        };
+
+        Ok(node)
+    }
+
+    fn resolve_key_strategy(
+        desc_tag: &str,
+        source: KeySource,
+        parent_pid: u32,
+    ) -> Result<(Option<File>, PathBuf, Option<i32>)> {
+        match source {
+            KeySource::Raw(raw_bytes) => {
+                let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?;
+                let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd));
+                Ok((Some(handle), pseudo_path, Some(fd)))
+            }
+            KeySource::Path(existing_path) => Ok((None, existing_path, None)),
+        }
+    }
+
+    fn find_available_port() -> Result<u16> {
+        let listener = TcpListener::bind("127.0.0.1:0")?;
+        let addr = listener.local_addr()?;
+        Ok(addr.port())
+    }
+
+    fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> {
+        #[cfg(target_os = "linux")]
+        {
+            let memfd_name = std::ffi::CString::new(desc_tag)?;
+            let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) };
+            if fd < 0 {
+                return Err(anyhow!(
+                    "memfd_create failed natively inside Linux kernel for tag: {}",
+                    desc_tag
+                ));
+            }
+            let mut file = unsafe { File::from_raw_fd(fd) };
+            file.write_all(&raw_key)?;
+            file.flush()?;
+            Ok((file, fd))
+        }
+
+        // Unreachable: `spawn` refuses to run elsewhere.
+        #[cfg(not(target_os = "linux"))]
+        {
+            let _ = (desc_tag, raw_key);
+            bail!("In-memory key files need Linux")
+        }
+    }
+
+    pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+        self.child.wait().await
+    }
+}
+
+impl Drop for SupervisedMycelium {
+    fn drop(&mut self) {
+        // The memfd key files go away with their handles.
+        let _ = self.child.start_kill();
+    }
+}
+
+fn is_pubkey(s: &str) -> bool {
+    s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit())
+}
+
+pub struct MyceliumTransport {
+    // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane
+    pub setup: tokio::sync::Mutex<MyceliumSetup>,
+    pub client: reqwest::Client,
+    pub poll_client: reqwest::Client,
+    pub messages_endpoint: Url,
+    pubkey_basepath: Url,
+    local_id: String,
+}
+
+impl MyceliumTransport {
+    pub async fn connect(setup: MyceliumSetup) -> Result<Self> {
+        let rest_url = setup.rest_url.clone();
+
+        let client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(10))
+            .build()?;
+        let poll_client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
+            .build()?;
+
+        let admin: MyceliumAdminInfo = client
+            .get(rest_url.join("/api/v1/admin")?)
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await
+            .context("Querying local Mycelium identity")?;
+
+        Ok(Self {
+            setup: tokio::sync::Mutex::new(setup),
+            client,
+            poll_client,
+            messages_endpoint: rest_url.join("/api/v1/messages")?,
+            pubkey_basepath: rest_url.join("/api/v1/pubkey/")?,
+            local_id: admin.node_pubkey,
+        })
+    }
+
+    async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> {
+        #[derive(Deserialize)]
+        struct PublicKeyResponse {
+            #[serde(rename = "publicKey")]
+            public_key: String,
+        }
+
+        let response: PublicKeyResponse = self
+            .client
+            .get(self.pubkey_basepath.join(&ip.to_string())?)
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        Ok(response.public_key)
+    }
+}
+
+#[async_trait]
+impl Transport for MyceliumTransport {
+    fn local_id(&self) -> &str {
+        &self.local_id
+    }
+
+    async fn resolve_seed(&self, hint: &str) -> Result<String> {
+        if is_pubkey(hint) {
+            return Ok(hint.to_owned());
+        }
+
+        if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() {
+            return self.lookup_pubkey(&ip).await;
+        }
+
+        let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0)))
+            .await
+            .context("DNS lookup timed out")??
+            .find_map(|addr| match addr.ip() {
+                std::net::IpAddr::V6(ip) => Some(ip),
+                _ => None,
+            })
+            .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?;
+
+        self.lookup_pubkey(&ip).await
+    }
+
+    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+        let outbound = format_message(peer, message)?;
+        Ok(EncodedMessage {
+            peer: peer.to_owned(),
+            body: serde_json::to_vec(&outbound)?.into(),
+        })
+    }
+
+    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
+        tracing::trace!(payload_bytes = message.body.len(), "Sending message");
+        let response = self
+            .client
+            .post(self.messages_endpoint.clone())
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .body(message.body.clone())
+            .send()
+            .await?;
+
+        let status = response.status();
+        if !status.is_success() {
+            tracing::error!(status = %status, "Mycelium rejected outbound message");
+            return Err(TransportError::MyceliumRejected {
+                peer: message.peer.clone(),
+                status,
+            }
+            .into());
+        }
+
+        // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
+        // a message that never arrived.
+        if tracing::enabled!(tracing::Level::DEBUG) {
+            let Ok(body) = response
+                .bytes()
+                .await
+                .map_err(anyhow::Error::from)
+                .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
+            else {
+                tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)");
+                return Ok(());
+            };
+            let Ok(outbound) = serde_json::from_slice::<OutboundMessage>(&message.body) else {
+                tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)");
+                return Ok(());
+            };
+            let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else {
+                tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)");
+                return Ok(());
+            };
+            let Some(id) = body.get("id").and_then(|id| id.as_str()) else {
+                tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)");
+                return Ok(());
+            };
+            tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message");
+        }
+        Ok(())
+    }
+
+    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        tracing::info!(
+            endpoint = self.messages_endpoint.as_str(),
+            "Starting Mycelium message consumer"
+        );
+
+        loop {
+            tracing::trace!("Polling Mycelium inbox");
+
+            let response = match self
+                .poll_client
+                .get(self.messages_endpoint.clone())
+                .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
+                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+                .send()
+                .await
+            {
+                Ok(resp) => resp,
+                Err(error) => {
+                    tracing::warn!(error = %error, "Mycelium inbox poll failed");
+                    tokio::time::sleep(Duration::from_millis(250)).await;
+                    continue;
+                }
+            };
+
+            let status = response.status();
+
+            if status == StatusCode::NO_CONTENT {
+                tracing::trace!("Mycelium inbox is empty");
+                tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
+                continue;
+            }
+
+            if !status.is_success() {
+                tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
+                tokio::time::sleep(Duration::from_millis(500)).await;
+                continue;
+            }
+
+            let raw_bytes = match response.bytes().await {
+                Ok(bytes) => bytes,
+                Err(error) => {
+                    tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    continue;
+                }
+            };
+
+            let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) {
+                Ok(msg) => msg,
+                Err(error) => {
+                    tracing::error!(
+                        error = %error,
+                        raw_payload = %String::from_utf8_lossy(&raw_bytes),
+                        "Failed to decode inbound Mycelium message structure"
+                    );
+                    tokio::time::sleep(Duration::from_millis(500)).await;
+                    continue;
+                }
+            };
+
+            match parse_message(&inbound_message) {
+                Ok((sender, parsed)) => {
+                    tracing::debug!(
+                        from = %inbound_message.src_pk,
+                        message = ?parsed,
+                        id = %inbound_message.id,
+                        //payload_bytes = raw_bytes.len(),
+                        "Received message"
+                    );
+                    if let Err(error) = handler(sender, parsed).await {
+                        tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
+                    }
+                }
+                Err(transport_error) => {
+                    tracing::error!(
+                        error = %transport_error,
+                        topic = %inbound_message.topic,
+                        "Dropping poisoned packet: parsing failed gracefully"
+                    );
+                    // No sleep required: we successfully popped a malformed packet and want to process the next one immediately
+                }
+            }
+        }
+    }
+
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        let mut guard = self.setup.lock().await;
+        guard.wait_for_exit().await
+    }
+}
+
+fn encode_topic(topic: &str) -> String {
+    BASE64.encode(topic.as_bytes())
+}
+
+fn decode_topic(topic: &str) -> Result<String, TransportError> {
+    let bytes = BASE64.decode(topic)?;
+    Ok(String::from_utf8(bytes)?)
+}
+
+fn encode_payload(payload: &Vec<u8>) -> String {
+    BASE64.encode(payload)
+}
+
+fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
+    let topic: &str = &decode_topic(topic_encoded)?;
+    let bytes = BASE64.decode(payload)?;
+    match topic {
+        TOPIC_HELLO => {
+            let msg = bytes_to_hello_request(&bytes)?;
+            Ok(ParsedMessage::Hello(msg))
+        }
+        TOPIC_HELLO_REPLY => {
+            let msg = bytes_to_hello_reply(&bytes)?;
+            Ok(ParsedMessage::HelloReply(msg))
+        }
+        TOPIC_SYNC => {
+            let msg = bytes_to_sync_envelope(&bytes)?;
+            Ok(ParsedMessage::Sync(msg))
+        }
+        TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
+        TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
+        t => Err(TransportError::UnknownTopic(t.to_string())),
+    }
+}
+
+fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> {
+    let sender = message.src_pk.clone();
+
+    if sender.is_empty() {
+        return Err(TransportError::NoSender);
+    }
+
+    let parsed = decode_payload(&message.topic, &message.payload)?;
+
+    Ok((sender, parsed))
+}
+
+fn format_message(
+    recipient: &str,
+    message: &ParsedMessage,
+) -> Result<OutboundMessage, TransportError> {
+    let (topic, payload_bytes) = match message {
+        ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
+        ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
+        ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
+        ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
+        ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
+    };
+    Ok(OutboundMessage {
+        dst: Destination {
+            pk: recipient.to_owned(),
+        },
+        topic: encode_topic(topic),
+        payload: encode_payload(&payload_bytes),
+    })
+}
+
+// =====================================================================
+// 1. WIRE SHADOWS
+// =====================================================================
+//
+// rmp_serde encodes structs as arrays. Fields added later are appended at the end and
+// marked #[serde(default)], so a message from a node that predates them still decodes
+// (an absent epoch is the genesis epoch).
+
+/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch.
+#[derive(Serialize, Deserialize, Default)]
+struct EpochShadow {
+    generation: u64,
+    id: ByteBuf,
+}
+
+fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow {
+    EpochShadow {
+        generation: epoch.generation,
+        id: ByteBuf::from(epoch.id.to_vec()),
+    }
+}
+
+fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> {
+    if shadow.generation == 0 && shadow.id.is_empty() {
+        return Ok(Epoch::GENESIS);
+    }
+    let id = shadow
+        .id
+        .as_slice()
+        .try_into()
+        .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?;
+    Ok(Epoch {
+        generation: shadow.generation,
+        id,
+    })
+}
+
+/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set.
+#[derive(Serialize, Deserialize)]
+struct KeepaliveShadow {
+    heads: Vec<ByteBuf>,
+    digest: ByteBuf,
+    epoch: EpochShadow,
+    ready: bool,
+    syncs_with_you: bool,
+}
+
+fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    let (heads, digest) = match &msg.heads {
+        HeadsSummary::Heads(heads) => (
+            heads
+                .iter()
+                .map(|head| ByteBuf::from(head.0.to_vec()))
+                .collect(),
+            ByteBuf::new(),
+        ),
+        HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())),
+    };
+    rmp_serde::to_vec(&KeepaliveShadow {
+        heads,
+        digest,
+        epoch: epoch_to_shadow(&msg.epoch),
+        ready: msg.ready,
+        syncs_with_you: msg.syncs_with_you,
+    })
+}
+
+fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
+    let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?;
+    let heads = if shadow.digest.is_empty() {
+        if shadow.heads.len() > MAX_LISTED_HEADS {
+            return Err(TransportError::InvalidPayload(format!(
+                "more than {MAX_LISTED_HEADS} heads listed"
+            )));
+        }
+        let heads = shadow
+            .heads
+            .iter()
+            .map(|head| ChangeHash::try_from(head.as_slice()))
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?;
+        HeadsSummary::Heads(heads)
+    } else {
+        let digest =
+            shadow.digest.as_slice().try_into().map_err(|_| {
+                TransportError::InvalidPayload("heads digest must be 32 bytes".into())
+            })?;
+        HeadsSummary::Digest(digest)
+    };
+    Ok(Keepalive {
+        heads,
+        epoch: epoch_from_shadow(shadow.epoch)?,
+        ready: shadow.ready,
+        syncs_with_you: shadow.syncs_with_you,
+    })
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct HelloRequestShadow {
+    #[serde(default)]
+    epoch: EpochShadow,
+    #[serde(default)]
+    auth: ByteBuf,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct HelloReplyShadow<'a> {
+    // Cow allows zero-copy references when saving, but creates an owned String when loading
+    pub swarm_id: Cow<'a, str>,
+    #[serde(default)]
+    pub peers: Vec<String>,
+    #[serde(default)]
+    epoch: EpochShadow,
+    #[serde(default)]
+    auth: ByteBuf,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct SyncEnvelopeShadow<'a> {
+    pub swarm_id: Cow<'a, str>,
+    #[serde(with = "automerge_msg_codec")]
+    pub message: automerge::sync::Message,
+    #[serde(default)]
+    epoch: EpochShadow,
+    /// Absent from nodes that predate it: never counts as ready.
+    #[serde(default)]
+    ready: bool,
+}
+
+// =====================================================================
+// 2. CUSTOM CODEC
+// =====================================================================
+mod automerge_msg_codec {
+    use super::*;
+    use automerge::sync::Message;
+
+    pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let bytes = msg.clone().encode();
+        Bytes::new(&bytes).serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let byte_buf = ByteBuf::deserialize(deserializer)?;
+        Message::decode(&byte_buf)
+            .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e)))
+    }
+}
+
+// =====================================================================
+// 3. CLEAN CONVERSION INTERFACE
+// =====================================================================
+
+// --- HelloRequest ---
+pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    rmp_serde::to_vec(&HelloRequestShadow {
+        epoch: epoch_to_shadow(&msg.epoch),
+        auth: ByteBuf::from(msg.auth.clone()),
+    })
+}
+pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
+    let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
+    Ok(HelloRequest {
+        epoch: epoch_from_shadow(shadow.epoch)?,
+        auth: shadow.auth.into_vec(),
+    })
+}
+
+// --- HelloReply ---
+pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    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),
+        auth: ByteBuf::from(msg.auth.clone()),
+    };
+    rmp_serde::to_vec(&shadow)
+}
+pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
+    let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
+    Ok(HelloReply {
+        swarm_id: shadow.swarm_id.into_owned(),
+        peers: shadow.peers,
+        epoch: epoch_from_shadow(shadow.epoch)?,
+        auth: shadow.auth.into_vec(),
+    })
+}
+
+// --- SyncEnvelope ---
+pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+    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),
+        ready: msg.ready,
+    };
+    rmp_serde::to_vec(&shadow)
+}
+pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
+    let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
+    Ok(SyncEnvelope {
+        swarm_id: shadow.swarm_id.into_owned(),
+        epoch: epoch_from_shadow(shadow.epoch)?,
+        message: shadow.message,
+        ready: shadow.ready,
+    })
+}
+
+#[derive(Error, Debug)]
+pub enum TransportError {
+    #[error("Mycelium delivered a message without a sender public key")]
+    NoSender,
+
+    #[error("Base64 decoding failed: {0}")]
+    Base64Decode(#[from] base64::DecodeError),
+
+    #[error("UTF-8 decoding failed: {0}")]
+    Utf8Decode(#[from] std::string::FromUtf8Error),
+
+    #[error("MessagePack serialization failed: {0}")]
+    MessagePackEncode(#[from] rmp_serde::encode::Error),
+
+    #[error("MessagePack deserialization failed: {0}")]
+    MessagePackDecode(#[from] rmp_serde::decode::Error),
+
+    #[error("Unknown topic: {0}")]
+    UnknownTopic(String),
+
+    #[error("Invalid payload: {0}")]
+    InvalidPayload(String),
+
+    #[error("Network HTTP request failed: {0}")]
+    Http(#[from] reqwest::Error),
+
+    #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")]
+    MyceliumRejected {
+        peer: String,
+        status: reqwest::StatusCode,
+    },
+}