1 files changed, 207 insertions, 23 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index fb13ad9..96f7651 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -1,7 +1,5 @@
 use anyhow::Result;
 use async_trait::async_trait;
-use automerge::ChangeHash;
-use automerge::sync::Message;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::boxed::Box;
@@ -11,6 +9,8 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 
 pub(crate) use crate::document::replica::Epoch;
 
+use crate::document::ChangeHash;
+
 // Used by the transports that don't authenticate their peers themselves (`tor`).
 #[cfg_attr(not(feature = "tor"), allow(dead_code))]
 pub(crate) mod challenge;
@@ -18,6 +18,7 @@ pub(crate) mod firewall;
 pub(crate) mod identity;
 #[cfg(feature = "libp2p")]
 pub(crate) mod libp2p;
+#[cfg(feature = "mycelium")]
 pub(crate) mod mycelium;
 pub(crate) mod paths;
 #[cfg(test)]
@@ -28,6 +29,67 @@ pub(crate) mod tor;
 pub(crate) mod traffic;
 pub(crate) mod wire;
 
+#[cfg(not(any(feature = "mycelium", feature = "libp2p", feature = "tor")))]
+compile_error!(
+    "continuum needs a transport: build with at least one of the mycelium, libp2p and tor features, e.g. `cargo build --features mycelium`"
+);
+
+/// The options of the transports this build has, each checked by its transport when the
+/// configuration is loaded (`config::Settings::check`, after the transports enabled):
+/// Mycelium's always (its conflicts are errors wherever they come from, as before);
+/// libp2p's and Tor's if the node uses them. Options of transports not built aren't
+/// checked: they're ignored, and said so (`config::Loaded::unused_transport_options`).
+pub(crate) fn check(settings: &crate::config::Settings) -> Result<()> {
+    let uses = |name: &str| settings.transports.iter().any(|used| used == name);
+    let _ = uses;
+    #[cfg(feature = "mycelium")]
+    mycelium::check(settings)?;
+    #[cfg(feature = "libp2p")]
+    if uses("libp2p") {
+        libp2p::check(settings)?;
+    }
+    #[cfg(feature = "tor")]
+    if uses("tor") {
+        tor::TorTransport::check(settings)?;
+    }
+    Ok(())
+}
+
+/// Starts the transport `name` (one of `config::TRANSPORTS`) as `settings` configure
+/// it: its scheme and the transport, or None if `stop` came first (some take a while to
+/// start: a supervised Mycelium daemon, arti). `key`: the node key, which some prove
+/// themselves.
+pub(crate) async fn start(
+    name: &str,
+    settings: &crate::config::Settings,
+    key: &std::sync::Arc<crate::identity::NodeKey>,
+    stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<(&'static str, std::sync::Arc<dyn Transport>)>> {
+    use std::sync::Arc;
+    let _ = (key, &stop);
+    let started: Option<(&'static str, Arc<dyn Transport>)> = match name {
+        #[cfg(feature = "mycelium")]
+        "mycelium" => mycelium::start(settings, stop)
+            .await?
+            .map(|transport| (mycelium::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+        #[cfg(feature = "libp2p")]
+        "libp2p" => {
+            let _ = stop;
+            let config = libp2p::Libp2pConfig::from_settings(settings, false)?;
+            Some((
+                libp2p::SCHEME,
+                Arc::new(libp2p::Libp2pTransport::start(config)?) as Arc<dyn Transport>,
+            ))
+        }
+        #[cfg(feature = "tor")]
+        "tor" => tor::TorTransport::from_settings(settings, key, stop)
+            .await?
+            .map(|transport| (tor::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+        other => anyhow::bail!("transports: no {other} in this build"),
+    };
+    Ok(started)
+}
+
 /// Random identifier of one run of a node (from start to exit).
 pub(crate) type RunId = [u8; 16];
 
@@ -137,6 +199,9 @@ pub(crate) struct Status {
     /// The sender would sync with the recipient: counts it as an active member on its
     /// own epoch.
     pub(crate) syncs_with_you: bool,
+    /// The sender is shutting down: its farewell (`keepalive::farewell`). The
+    /// recipient holds it dormant at once, rather than after `dormant_after` of silence.
+    pub(crate) leaving: bool,
 }
 
 /// Cheap to clone: large parts are shared (`bytes::Bytes`).
@@ -154,8 +219,17 @@ pub(crate) enum ParsedMessage {
     Me(IdentityProof),
     /// Asks for a PONG with this nonce, over the same path (`transport::paths`).
     Ping(u64),
-    /// Answers a PING.
-    Pong(u64),
+    /// Answers a PING: its nonce, and when it was answered by the answerer's clock
+    /// (`at`), for comparing clocks (`transport::paths`).
+    Pong {
+        nonce: u64,
+        at: crate::clock::UnixMillis,
+    },
+    /// Asks a member how it is linked to its peers, for a map of the swarm
+    /// (`swarm::links`).
+    Links(LinksRequest),
+    /// Answers a LINKS.
+    LinksReply(LinksReply),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -281,6 +355,8 @@ pub(crate) enum MessageKind {
     Me,
     Ping,
     Pong,
+    Links,
+    LinksReply,
 }
 
 impl ParsedMessage {
@@ -295,7 +371,9 @@ impl ParsedMessage {
             ParsedMessage::Who(_) => MessageKind::Who,
             ParsedMessage::Me(_) => MessageKind::Me,
             ParsedMessage::Ping(_) => MessageKind::Ping,
-            ParsedMessage::Pong(_) => MessageKind::Pong,
+            ParsedMessage::Pong { .. } => MessageKind::Pong,
+            ParsedMessage::Links(_) => MessageKind::Links,
+            ParsedMessage::LinksReply(_) => MessageKind::LinksReply,
         }
     }
 
@@ -310,14 +388,79 @@ impl ParsedMessage {
                 Some((status.epoch, status.epoch_seq))
             }
             ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
+            ParsedMessage::Links(request) => Some((request.epoch, request.epoch_seq)),
+            ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)),
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
-            | ParsedMessage::Pong(_) => None,
+            | ParsedMessage::Pong { .. } => None,
         }
     }
 }
 
+/// LINKS: what a member is asked for a map of the swarm (`swarm::links`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct LinksRequest {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// Repeated in the reply: which request it answers.
+    pub(crate) nonce: u64,
+    /// For this long (seconds), a LINKS_REPLY again whenever the links change (a
+    /// subscription; the next LINKS renews it). 0: this reply only, and a subscription
+    /// ends.
+    pub(crate) subscribe_secs: u32,
+}
+
+/// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct LinksReply {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    /// The request's.
+    pub(crate) nonce: u64,
+    /// At most `swarm::links::MAX_LINKS`.
+    pub(crate) links: Vec<Link>,
+}
+
+/// One of a member's links, in a LINKS_REPLY.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct Link {
+    /// The peer's node ID.
+    pub(crate) peer: String,
+    /// The transport its messages go over now (the scheme of the path in use), if any.
+    pub(crate) transport: Option<String>,
+    /// Only through a relay (libp2p).
+    pub(crate) relayed: bool,
+    /// The path in use is down: none to the peer answers (`paths`), whatever its
+    /// connection was.
+    pub(crate) down: bool,
+    /// The round trip on that path, in microseconds, if measured.
+    pub(crate) rtt_us: Option<u32>,
+    /// Silent for long (`peers.md`, dormancy).
+    pub(crate) dormant: bool,
+    /// The peer's other transports (`swarm::links::MAX_OTHERS` at most): where else
+    /// it proved itself, those paths up.
+    pub(crate) others: Vec<OtherPath>,
+    /// How far the peer's clock is ahead of the member's (`paths::ClockOffset`), if
+    /// measured: with every member's, a map of the swarm shows whose clock is off.
+    pub(crate) clock: Option<paths::ClockOffset>,
+}
+
+/// A transport of a link besides the one in use, in a LINKS_REPLY.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct OtherPath {
+    /// Its scheme (`mycelium`, `libp2p`, `tor`).
+    pub(crate) transport: String,
+    /// The round trip of its fastest path, in microseconds, if any is measured.
+    pub(crate) rtt_us: Option<u32>,
+    /// Only through a relay (libp2p): no direct connection at any of its paths.
+    pub(crate) relayed: bool,
+}
+
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) struct HelloRequest {
     /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in
@@ -361,17 +504,34 @@ pub(crate) struct Destination {
     pub(crate) proven: bool,
     /// The transport's connection to it, if the transport has connections.
     pub(crate) connection: Option<Connection>,
+    /// While it is connected only through a relay: how hole punching went.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) punching: Option<Punching>,
     /// Every address it proved itself at, best first, as the paths layer measures them
     /// (`transport::paths`).
     pub(crate) paths: Vec<stats::PathInfo>,
 }
 
+/// Hole punching to a peer connected only through a relay (libp2p), once a punch failed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Punching {
+    /// Punches that failed in a row.
+    pub(crate) failed: usize,
+    /// When it is tried again, in seconds from now (0: due).
+    pub(crate) next_in_secs: u64,
+}
+
 /// How a transport with connections (libp2p) is connected to a peer.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "camelCase")]
+// Only libp2p and Tor have connections; without both (the cargo features) nothing makes
+// one.
+#[cfg_attr(not(any(feature = "libp2p", feature = "tor")), allow(dead_code))]
 pub(crate) enum Connection {
+    /// libp2p's to the peer itself; Tor's always (an onion circuit).
     Direct,
-    /// Only through a relay.
+    /// Only through a relay (libp2p).
     Relayed,
 }
 
@@ -489,7 +649,7 @@ pub(crate) enum SyncBody {
 
 /// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of
 /// a document at `to`. Every node holding `to` produces the same bytes for it
-/// (`swarm::blob::delta_bytes`).
+/// (`document::changes::delta_bytes`).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct DeltaSpec {
     /// Sorted.
@@ -523,23 +683,13 @@ impl DeltaSpec {
     }
 }
 
-/// An Automerge sync message in Automerge's own encoding. Decoded only where it is
-/// applied (`decode`), after the sender proved to be a member and off the consumer: a
-/// large message from anyone else costs nothing but its bytes.
+/// A sync message as the document's engine encodes it (`document::sync`): bytes to the
+/// transports. Decoded only where it is applied (`document::sync::decode`), after the sender
+/// proved to be a member and off the consumer: a large message from anyone else costs
+/// nothing but its bytes.
 #[derive(Clone)]
 pub(crate) struct SyncPayload(pub(crate) bytes::Bytes);
 
-impl SyncPayload {
-    pub(crate) fn encode(message: Message) -> Self {
-        Self(message.encode().into())
-    }
-
-    pub(crate) fn decode(&self) -> Result<Message> {
-        Message::decode(&self.0)
-            .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}"))
-    }
-}
-
 /// Sizes instead of contents: logged messages stay short.
 impl std::fmt::Debug for SyncPayload {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -585,7 +735,7 @@ impl<T: std::fmt::Display> std::fmt::Display for Addressed<T> {
 /// consumer hands them on as soon as they are read (`hand_on_probe`), not after the
 /// messages read before them are handled (which can take a while: sync).
 pub(crate) fn is_probe(message: &ParsedMessage) -> bool {
-    matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_))
+    matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong { .. })
 }
 
 /// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own; `scheme` is the
@@ -618,6 +768,13 @@ pub(crate) trait Transport: Send + Sync {
     /// a message sent is delivered or lost for good.
     fn delivery_limit(&self) -> std::time::Duration;
 
+    /// The first wait for an answer to a sync message (`Timing::sync_reply_timeout`): by
+    /// default `SYNC_REPLY_TIMEOUT`, longer on a transport whose round trips take
+    /// seconds.
+    fn sync_reply_timeout(&self) -> std::time::Duration {
+        crate::state::SYNC_REPLY_TIMEOUT
+    }
+
     /// Where this node can be reached, as addresses other nodes can use: published in
     /// the member list (`swarm::members`). By default its identity, for a transport
     /// whose identities are addresses (Mycelium: a public key).
@@ -668,6 +825,19 @@ pub(crate) trait Transport: Send + Sync {
         Vec::new()
     }
 
+    /// Has the transport tell `found` whenever it finds a seed by itself
+    /// (`discovered_seeds`). By default it never does.
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        let _ = found;
+    }
+
+    /// Has the transport tell `reachable` whenever others may reach this node where they
+    /// couldn't before (libp2p: a relay granted it a slot, or AutoNAT found it public).
+    /// By default it never does.
+    fn on_reachable(&self, reachable: std::sync::Arc<tokio::sync::Notify>) {
+        let _ = reachable;
+    }
+
     /// Whether this node has a way to reach `peer` at all: an address of it on one of
     /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`).
     /// By default yes, as for `local_addresses`.
@@ -712,6 +882,13 @@ pub(crate) trait Transport: Send + Sync {
         None
     }
 
+    /// How hole punching to `target` went, if a punch failed and it is still connected
+    /// only through a relay (libp2p). By default never.
+    fn punching(&self, target: &str) -> Option<Punching> {
+        let _ = target;
+        None
+    }
+
     /// The smoothed round-trip time of the path messages to `peer` go over, if
     /// measured (`transport::paths`). By default none.
     fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
@@ -719,6 +896,13 @@ pub(crate) trait Transport: Send + Sync {
         None
     }
 
+    /// How far `peer`'s clock is ahead of ours, as its PONGs say (`paths::ClockOffset`):
+    /// the narrowest bound of its paths. By default unknown.
+    fn clock_offset(&self, peer: &str) -> Option<paths::ClockOffset> {
+        let _ = peer;
+        None
+    }
+
     /// Whether the transport answers PINGs with PONGs itself (the paths layer,
     /// `transport::paths`); if not (a bare transport, in tests), the node does. By
     /// default not.