1 files changed, 134 insertions, 15 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 3f77306..25ad56d 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -11,13 +11,20 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 
 pub(crate) use crate::document::replica::Epoch;
 
+// Used by the transports that don't authenticate their peers themselves (`tor`).
+#[cfg_attr(not(feature = "tor"), allow(dead_code))]
+pub(crate) mod challenge;
 pub(crate) mod firewall;
 pub(crate) mod identity;
-pub(crate) mod mycelium;
 #[cfg(feature = "libp2p")]
-pub(crate) mod p2p;
+pub(crate) mod libp2p;
+pub(crate) mod mycelium;
+pub(crate) mod paths;
 #[cfg(test)]
 pub(crate) mod sim;
+pub(crate) mod stats;
+#[cfg(feature = "tor")]
+pub(crate) mod tor;
 pub(crate) mod traffic;
 pub(crate) mod wire;
 
@@ -115,9 +122,9 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
     hasher.finalize().into()
 }
 
-/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
+/// STATUS and STATUS_REPLY. Addressed to one peer: `syncs_with_you` is about the recipient.
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub(crate) struct Keepalive {
+pub(crate) struct Status {
     /// Only comparable between nodes on the same epoch.
     #[serde(with = "crate::transport::wire::codec::heads")]
     pub(crate) heads: HeadsSummary,
@@ -132,18 +139,23 @@ pub(crate) struct Keepalive {
     pub(crate) syncs_with_you: bool,
 }
 
-#[derive(Debug)]
+/// Cheap to clone: large parts are shared (`bytes::Bytes`).
+#[derive(Debug, Clone)]
 pub(crate) enum ParsedMessage {
     Hello(HelloRequest),
     HelloReply(HelloReply),
     Sync(SyncEnvelope),
-    Ping(Keepalive),
-    Pong(Keepalive),
+    Status(Status),
+    StatusReply(Status),
     Blob(BlobMessage),
     /// Asks the recipient who it is, saying who the sender is (`transport::identity`).
     Who(IdentityProof),
     /// Answers a WHO.
     Me(IdentityProof),
+    /// Asks for a PONG with this nonce, over the same path (`transport::paths`).
+    Ping(u64),
+    /// Answers a PING.
+    Pong(u64),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -262,11 +274,13 @@ pub(crate) enum MessageKind {
     Hello,
     HelloReply,
     Sync,
-    Ping,
-    Pong,
+    Status,
+    StatusReply,
     Blob,
     Who,
     Me,
+    Ping,
+    Pong,
 }
 
 impl ParsedMessage {
@@ -275,26 +289,31 @@ impl ParsedMessage {
             ParsedMessage::Hello(_) => MessageKind::Hello,
             ParsedMessage::HelloReply(_) => MessageKind::HelloReply,
             ParsedMessage::Sync(_) => MessageKind::Sync,
-            ParsedMessage::Ping(_) => MessageKind::Ping,
-            ParsedMessage::Pong(_) => MessageKind::Pong,
+            ParsedMessage::Status(_) => MessageKind::Status,
+            ParsedMessage::StatusReply(_) => MessageKind::StatusReply,
             ParsedMessage::Blob(_) => MessageKind::Blob,
             ParsedMessage::Who(_) => MessageKind::Who,
             ParsedMessage::Me(_) => MessageKind::Me,
+            ParsedMessage::Ping(_) => MessageKind::Ping,
+            ParsedMessage::Pong(_) => MessageKind::Pong,
         }
     }
 
     /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them
-    /// but WHO and ME (which the node never sees).
+    /// but WHO, ME, PING and PONG (which the node never sees).
     pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> {
         match self {
             ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
             ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
             ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
-            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => {
-                Some((keepalive.epoch, keepalive.epoch_seq))
+            ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
+                Some((status.epoch, status.epoch_seq))
             }
             ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
-            ParsedMessage::Who(_) | ParsedMessage::Me(_) => None,
+            ParsedMessage::Who(_)
+            | ParsedMessage::Me(_)
+            | ParsedMessage::Ping(_)
+            | ParsedMessage::Pong(_) => None,
         }
     }
 }
@@ -330,6 +349,40 @@ pub(crate) struct ListedPeer {
     pub(crate) addresses: Vec<String>,
 }
 
+/// Where messages to a peer go now (`Transport::destination`).
+#[derive(Debug, Clone, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Destination {
+    /// As the identity layer keeps it: `<scheme>:<transport identity>`.
+    pub(crate) address: String,
+    /// The peer proved itself there; else it is only said to be there, and sends wait
+    /// for its proof.
+    pub(crate) proven: bool,
+    /// The transport's connection to it, if the transport has connections.
+    pub(crate) connection: Option<Connection>,
+    /// Every address it proved itself at, best first, as the paths layer measures them
+    /// (`transport::paths`).
+    pub(crate) paths: Vec<stats::PathInfo>,
+}
+
+/// How a transport with connections (libp2p) is connected to a peer.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum Connection {
+    Direct,
+    /// Only through a relay.
+    Relayed,
+}
+
+impl Connection {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::Direct => "direct",
+            Self::Relayed => "relayed",
+        }
+    }
+}
+
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) struct HelloReply {
     pub(crate) swarm_id: String,
@@ -504,6 +557,9 @@ pub(crate) struct EncodedMessage {
     pub(crate) kind: MessageKind,
     /// Bytes the message takes on the wire (without the transport's own framing).
     pub(crate) payload_len: usize,
+    /// The message itself, under the paths layer (`transport::paths`): to encode it
+    /// again for another transport when a send there fails. None from a transport.
+    pub(crate) message: Option<std::sync::Arc<ParsedMessage>>,
 }
 
 /// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`).
@@ -513,6 +569,28 @@ pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn
     + 'static;
 pub(crate) type MessageHandler = Box<HandlerFn>;
 
+/// Whether `message` is a PING or PONG. The paths layer times them, so a transport's
+/// 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(_))
+}
+
+/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own.
+pub(crate) fn hand_on_probe(
+    handler: &std::sync::Arc<HandlerFn>,
+    sender: String,
+    message: ParsedMessage,
+    payload_len: usize,
+) {
+    let handler = std::sync::Arc::clone(handler);
+    tokio::spawn(async move {
+        if let Err(error) = handler(sender.clone(), message, payload_len).await {
+            tracing::warn!(peer = %sender, error = %format!("{error:#}"), "Failed to process inbound message");
+        }
+    });
+}
+
 #[async_trait]
 pub(crate) trait Transport: Send + Sync {
     /// This node's identity on the transport (for Mycelium: its public key).
@@ -584,6 +662,47 @@ pub(crate) trait Transport: Send + Sync {
         let _ = (peer, addresses);
     }
 
+    /// Where messages to `peer` go now. None if nowhere, and by default, for a
+    /// transport whose identities are addresses (`peer` is where they go).
+    fn destination(&self, peer: &str) -> Option<Destination> {
+        let _ = peer;
+        None
+    }
+
+    /// Where messages to `target` (a transport identity) go now, as an address that
+    /// names `target` too: the remote address of the connection in use, say. By default
+    /// nowhere but `target` itself (Mycelium).
+    fn location(&self, target: &str) -> Option<String> {
+        let _ = target;
+        None
+    }
+
+    /// How this node is connected to `target` (a transport identity), if it is. By
+    /// default not at all, for a transport without connections (Mycelium).
+    fn connection(&self, target: &str) -> Option<Connection> {
+        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> {
+        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.
+    fn answers_pings(&self) -> bool {
+        false
+    }
+
+    /// Figures about the transport itself (`stats`). By default none.
+    fn stats(&self) -> stats::TransportStats {
+        stats::TransportStats::default()
+    }
+
     /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's
     /// identity, as `local_id` gives it on that node.
     async fn resolve_seed(&self, seed: &str) -> Result<String>;