diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/identity.rs | 68 | ||||
| -rw-r--r-- | src/transport/paths.rs | 2 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 43 |
3 files changed, 80 insertions, 33 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs index b61fb0f..5d3e2f5 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -257,6 +257,28 @@ impl Book { } impl 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. The address + /// it went to. + async fn send_proven(&self, message: &EncodedMessage) -> Result<String> { + let proven = self.shared.proven_at(&message.peer, true); + if !proven.is_empty() { + self.shared.ask_other_paths(&message.peer); + } + if proven.is_empty() { + let hints = { + let mut book = self.shared.lock(); + book.counts.refused_unproven += 1; + book.hints.get(&message.peer).cloned().unwrap_or_default() + }; + for address in hints { + self.shared.ask(&address, false); + } + return Err(NotProven(message.peer.clone()).into()); + } + self.shared.paths.send_first(message, &proven).await + } + /// Over `transport`, whose addresses have `config.scheme`, and `config.more`. /// `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`). @@ -299,6 +321,9 @@ impl IdentityTransport { admitted: Mutex::default(), } }); + // The node's messages are noted here, with the address each went to or came + // from (`MessageLog`). + shared.traffic.log.set_addressed(); Self { shared } } } @@ -479,7 +504,16 @@ impl Shared { // proofs fill the book (`Book::bind`). let node = self.lock().heard(&address, Instant::now()); match node { - Some(node) => handler(node, message, payload_len).await, + Some(node) => { + self.traffic.log.note( + "received", + &node, + Some(&address), + &message, + payload_len, + ); + handler(node, message, payload_len).await + } None => { self.lock().counts.dropped_unproven += 1; debug!( @@ -749,29 +783,21 @@ impl Transport for IdentityTransport { Ok(encoded) } - /// 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. + /// As `send_proven` does, noted in the message log with the address + /// (`MessageLog`). async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - let proven = self.shared.proven_at(&message.peer, true); - if !proven.is_empty() { - self.shared.ask_other_paths(&message.peer); - } - if proven.is_empty() { - let hints = { - let mut book = self.shared.lock(); - book.counts.refused_unproven += 1; - book.hints.get(&message.peer).cloned().unwrap_or_default() + let sent = self.send_proven(message).await; + if let Some(parsed) = &message.message { + let (way, address) = match &sent { + Ok(address) => ("sent", Some(address.as_str())), + Err(_) => ("refused", None), }; - for address in hints { - self.shared.ask(&address, false); - } - return Err(NotProven(message.peer.clone()).into()); + self.shared + .traffic + .log + .note(way, &message.peer, address, parsed, message.payload_len); } - self.shared - .paths - .send_first(message, &proven) - .await - .map(|_| ()) + sent.map(|_| ()) } /// Every transport's consumer (`Paths::start_consumer`); returns when the first diff --git a/src/transport/paths.rs b/src/transport/paths.rs index a830143..f425216 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -987,7 +987,7 @@ impl Paths { let way = if sent.is_ok() { "sent" } else { "refused" }; traffic .log - .note(way, address, &message, encoded.payload_len); + .note(way, address, None, &message, encoded.payload_len); } sent } diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 948e5d5..bac0da7 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -150,12 +150,16 @@ const LOGGED: usize = 1000; const LOGGED_TEXT: usize = 8 * 1024; /// The node's messages one by one, for the dashboard's console (`api::dashboard`): -/// kept only while it looks (`watch`), the latest `LOGGED`. Those this layer sees: the -/// node's own, not the paths layer's PINGs and PONGs nor the identity layer's WHO and -/// ME (counted under their addresses, `ByKind`), but the PINGs and PONGs the node is -/// handed. +/// kept only while it looks (`watch`), the latest `LOGGED`. The node's own, each with +/// the address it went to or came from where the identity layer notes them (`addressed`, +/// `transport::identity`), else as `CountingTransport` sees them; and the paths layer's +/// PINGs and PONGs, under their addresses; not the identity layer's WHO and ME (counted +/// under their addresses, `ByKind`). #[derive(Debug, Default)] pub(crate) struct MessageLog { + /// The identity layer notes the node's messages, with their addresses + /// (`set_addressed`): `CountingTransport`, above it, leaves them to it. + addressed: std::sync::atomic::AtomicBool, /// How many look (`watch`). watching: AtomicUsize, /// Told of each message logged, for those sending them on as they come (`logged`). @@ -192,9 +196,9 @@ pub(crate) struct Logged { /// exchange. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) related: Option<String>, - /// The address it went to or came from, where the paths layer logged it under that - /// (its own PINGs and PONGs) and the node there is known: `peer` is the node then - /// (`api::dashboard`). + /// The address it went to or came from: of a node's message, where the identity + /// layer noted it; of the paths layer's PINGs and PONGs, logged under the address, + /// where the node there is known (`peer` is the node then, `api::dashboard`). #[serde(skip_serializing_if = "Option::is_none")] pub(crate) address: Option<String>, } @@ -254,10 +258,21 @@ impl MessageLog { self.inner.lock_or_recover().next } + /// The identity layer notes the node's messages from now on, with their addresses. + pub(crate) fn set_addressed(&self) { + self.addressed.store(true, Ordering::Relaxed); + } + + /// Whether the identity layer notes the node's messages (`set_addressed`). + pub(crate) fn addressed(&self) -> bool { + self.addressed.load(Ordering::Relaxed) + } + pub(crate) fn note( &self, way: &'static str, peer: &str, + address: Option<&str>, message: &ParsedMessage, bytes: usize, ) { @@ -290,7 +305,7 @@ impl MessageLog { bytes, text, related: related(peer, message), - address: None, + address: address.map(str::to_owned), }); if inner.entries.len() > LOGGED { inner.entries.pop_front(); @@ -523,11 +538,13 @@ impl Transport for CountingTransport { self.traffic .refused(&message.peer, message.kind, message.payload_len); } - if let Some(parsed) = &message.message { + if let Some(parsed) = &message.message + && !self.traffic.log.addressed() + { let way = if result.is_ok() { "sent" } else { "refused" }; self.traffic .log - .note(way, &message.peer, parsed, message.payload_len); + .note(way, &message.peer, None, parsed, message.payload_len); } result } @@ -537,7 +554,11 @@ impl Transport for CountingTransport { self.inner .start_consumer(Box::new(move |sender, message, payload_len| { traffic.received(&sender, message.kind(), payload_len); - traffic.log.note("received", &sender, &message, payload_len); + if !traffic.log.addressed() { + traffic + .log + .note("received", &sender, None, &message, payload_len); + } handler(sender, message, payload_len) })) .await |