1 files changed, 81 insertions, 21 deletions

diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index c597d77..1df84ec 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -44,7 +44,7 @@ use crate::transport::paths::{Gate, PathHandler, Paths, scheme_of};
 use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
 use crate::transport::traffic::Traffic;
 use crate::transport::{
-    Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind,
+    Connection, Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind,
     ParsedMessage, Transport,
 };
 
@@ -58,8 +58,9 @@ const ASK_INTERVAL: Duration = Duration::from_secs(5);
 const OTHER_PATHS_ASK_INTERVAL: Duration = Duration::from_secs(60);
 /// How long resolving a seed waits for its ME.
 const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
-/// Addresses kept per node from hints: about one per transport.
-const MAX_HINTS: usize = 8;
+/// Addresses kept per node from hints: as many as a member lists
+/// (`members::MAX_ADDRESSES`).
+const MAX_HINTS: usize = 16;
 /// Addresses with a proven node, nodes with hints, and addresses asked, each at most:
 /// anyone can send us proofs (keys are free), so the books are bounded.
 const MAX_ENTRIES: usize = 4096;
@@ -187,6 +188,8 @@ impl Book {
         }
         IdentityStats {
             proven,
+            // Asked of the transports outside this lock (`IdentityTransport::stats`).
+            relayed: std::collections::BTreeMap::new(),
             hinted: self
                 .hints
                 .keys()
@@ -242,12 +245,15 @@ impl Book {
 
 impl IdentityTransport {
     /// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
-    /// `firewall`: the debug firewall, if any, which paths applies by address.
+    /// `firewall`: the debug firewall, if any, which paths applies by address; `clock`:
+    /// the node's wall clock; `keepalive`: the node's keepalive interval (`Paths::new`).
     pub(crate) fn new(
         transport: Arc<dyn Transport>,
         config: IdentityConfig,
         traffic: Arc<Traffic>,
         firewall: Option<Arc<Firewall>>,
+        clock: Arc<crate::clock::Clock>,
+        keepalive: Duration,
     ) -> Self {
         let node_id = config.key.id();
         let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| {
@@ -259,6 +265,8 @@ impl IdentityTransport {
                         .collect(),
                     Some(Arc::clone(&traffic)),
                     firewall.is_some().then_some(gate),
+                    clock,
+                    keepalive,
                 )),
                 firewall,
                 key: config.key,
@@ -294,6 +302,22 @@ impl Shared {
 }
 
 impl Shared {
+    /// Of the nodes in `proven`, those each transport reaches only through a relay, by
+    /// transport (`IdentityStats::relayed`); asked of the transports, so not under the
+    /// book's lock.
+    fn relayed(
+        &self,
+        proven: &HashMap<String, Vec<String>>,
+    ) -> std::collections::BTreeMap<String, usize> {
+        let mut relayed = std::collections::BTreeMap::new();
+        for address in proven.values().flatten() {
+            if self.paths.connection(address) == Some(Connection::Relayed) {
+                *relayed.entry(scheme_of(address).to_owned()).or_insert(0) += 1;
+            }
+        }
+        relayed
+    }
+
     fn lock(&self) -> MutexGuard<'_, Book> {
         self.book.lock_or_recover()
     }
@@ -435,24 +459,25 @@ impl Shared {
     }
 
     /// Where to send to `peer`: an address it proved itself at, else one it is said to
-    /// be at; the most preferred transport's first.
-    fn destination(&self, peer: &str) -> Option<String> {
-        self.proven_at(peer)
+    /// be at; the most preferred transport's first. `sending` as `Paths::rank` has it:
+    /// for a message about to go, or only a look.
+    fn destination(&self, peer: &str, sending: bool) -> Option<String> {
+        self.proven_at(peer, sending)
             .into_iter()
             .next()
-            .or_else(|| self.hinted_at(peer).into_iter().next())
+            .or_else(|| self.hinted_at(peer, sending).into_iter().next())
     }
 
     /// The addresses `peer` proved itself at, best first (`Paths::rank`).
-    fn proven_at(&self, peer: &str) -> Vec<String> {
+    fn proven_at(&self, peer: &str, sending: bool) -> Vec<String> {
         let proven = self.lock().proven.get(peer).cloned().unwrap_or_default();
-        self.paths.rank(&proven)
+        self.paths.rank(&proven, sending)
     }
 
     /// Where `peer` is said to be, best first.
-    fn hinted_at(&self, peer: &str) -> Vec<String> {
+    fn hinted_at(&self, peer: &str, sending: bool) -> Vec<String> {
         let hints = self.lock().hints.get(peer).cloned().unwrap_or_default();
-        self.paths.rank(&hints)
+        self.paths.rank(&hints, sending)
     }
 }
 
@@ -470,6 +495,10 @@ impl Transport for IdentityTransport {
         self.shared.paths.delivery_limit()
     }
 
+    fn sync_reply_timeout(&self) -> Duration {
+        self.shared.paths.sync_reply_timeout()
+    }
+
     fn local_addresses(&self) -> Vec<String> {
         self.shared.paths.local_addresses()
     }
@@ -533,7 +562,7 @@ impl Transport for IdentityTransport {
     }
 
     fn can_reach(&self, peer: &str) -> bool {
-        self.shared.destination(peer).is_some()
+        self.shared.destination(peer, false).is_some()
     }
 
     fn transports(&self) -> Vec<String> {
@@ -548,16 +577,24 @@ impl Transport for IdentityTransport {
     /// The address `encode` picks, the transport's connection there, and every path
     /// to `peer` as paths measures it, best first.
     fn destination(&self, peer: &str) -> Option<Destination> {
-        let proven = self.shared.proven_at(peer);
+        let proven = self.shared.proven_at(peer, false);
         let (address, is_proven) = match proven.first() {
             Some(address) => (address.clone(), true),
-            None => (self.shared.hinted_at(peer).into_iter().next()?, false),
+            None => (
+                self.shared.hinted_at(peer, false).into_iter().next()?,
+                false,
+            ),
         };
         let connection = self.shared.paths.connection(&address);
+        let punching = match connection {
+            Some(Connection::Relayed) => self.shared.paths.punching(&address),
+            _ => None,
+        };
         Some(Destination {
             address: self.shared.paths.location(&address),
             proven: is_proven,
             connection,
+            punching,
             paths: self.shared.paths.info(&proven),
         })
     }
@@ -569,15 +606,29 @@ impl Transport for IdentityTransport {
 
     /// The best path's, where messages to `peer` go.
     fn rtt(&self, peer: &str) -> Option<Duration> {
-        let address = self.shared.proven_at(peer).into_iter().next()?;
+        let address = self.shared.proven_at(peer, false).into_iter().next()?;
         self.shared.paths.rtt(&address)
     }
 
+    /// The narrowest of its proven paths'.
+    fn clock_offset(&self, peer: &str) -> Option<crate::transport::paths::ClockOffset> {
+        self.shared
+            .proven_at(peer, false)
+            .iter()
+            .filter_map(|address| self.shared.paths.clock_offset(address))
+            .min_by_key(|offset| offset.within_ms)
+    }
+
     /// The identity layer's, and each transport's.
     fn stats(&self) -> TransportStats {
         let mut stats = self.shared.paths.stats();
+        let (mut identity, proven) = {
+            let books = self.shared.lock();
+            (books.stats(), books.proven.clone())
+        };
+        identity.relayed = self.shared.relayed(&proven);
         stats.merge(TransportStats {
-            identity: Some(self.shared.lock().stats()),
+            identity: Some(identity),
             ..TransportStats::default()
         });
         stats
@@ -587,12 +638,20 @@ impl Transport for IdentityTransport {
         self.shared.paths.discovered_seeds()
     }
 
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        self.shared.paths.on_seeds_found(found)
+    }
+
+    fn on_reachable(&self, reachable: std::sync::Arc<tokio::sync::Notify>) {
+        self.shared.paths.on_reachable(reachable)
+    }
+
     /// A node ID resolves to itself if this node knows where it is; anything else is
     /// an address (`<scheme>:...`) or what one of the transports resolves to one (the
     /// first that does), asked WHO.
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         if identity::is_node_id(seed) {
-            if self.shared.destination(seed).is_some() {
+            if self.shared.destination(seed, false).is_some() {
                 return Ok(seed.to_owned());
             }
             bail!("a node ID alone doesn't say where that node is; give its address");
@@ -628,9 +687,10 @@ impl Transport for IdentityTransport {
     }
 
     /// For the address `destination` picks (a send there is refused until `peer`
-    /// proved itself there, `send_encoded`).
+    /// proved itself there, `send_encoded`). Only a look at the choice: the send makes
+    /// it (encoding again for another path if it changes).
     fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        let Some(address) = self.shared.destination(peer) else {
+        let Some(address) = self.shared.destination(peer, false) else {
             bail!("no address known for {peer} on this node's transports");
         };
         let mut encoded = self.shared.paths.encode(&address, message)?;
@@ -641,7 +701,7 @@ impl Transport for IdentityTransport {
     /// To the addresses `peer` proved itself at, best first, the next when a send
     /// fails (`Paths::send_first`); refused if it proved itself nowhere yet.
     async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
-        let proven = self.shared.proven_at(&message.peer);
+        let proven = self.shared.proven_at(&message.peer, true);
         if !proven.is_empty() {
             self.shared.ask_other_paths(&message.peer);
         }