disco: fix endpoint encoding

ts_disco_protocol/src/endpoint.rs

@@ -87,7 +87,7 @@ impl Endpoint {
         };
 
         Self {
-            addr: zerocopy::transmute!(ip.segments()),
+            addr: zerocopy::transmute!(ip.octets()),
             port: zerocopy::U16::new(sa.port()),
         }
     }
@@ -120,7 +120,7 @@ impl From<Endpoint> for SocketAddr {
 impl From<SocketAddrV6> for Endpoint {
     fn from(value: SocketAddrV6) -> Self {
         Self {
-            addr: zerocopy::transmute!(value.ip().segments()),
+            addr: zerocopy::transmute!(value.ip().octets()),
             port: value.port().into(),
         }
     }
@@ -131,3 +131,34 @@ impl From<SocketAddr> for Endpoint {
         Self::from_socket_addr(value)
     }
 }
+
+#[cfg(test)]
+mod test {
+    use core::str::FromStr;
+
+    use zerocopy::{FromBytes, IntoBytes};
+
+    use super::*;
+
+    #[test]
+    fn convert_basic() {
+        const BYTES: [u8; 18] = [
+            0x26, 0x00, 0xab, 0xcd, 0xef, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09,
+            0x0a, 0x0b, 0xa5, 0xb5,
+        ];
+
+        let addr = Ipv6Addr::from_str("2600:abcd:ef01:0203:0405:0607:0809:0a0b").unwrap();
+        let ep = Endpoint::read_from_bytes(&BYTES).unwrap();
+
+        assert_eq!(ep.addr_v6(), addr);
+        assert_eq!(ep.port(), 0xa5b5);
+
+        let sa = ep.socket_addr();
+        assert_eq!(sa.ip(), IpAddr::V6(addr));
+        assert_eq!(sa.port(), 0xa5b5);
+
+        let ep2 = Endpoint::from(sa);
+        assert_eq!(ep2, ep);
+        assert_eq!(ep2.as_bytes(), BYTES);
+    }
+}

ts_runtime/src/dataplane.rs

@@ -155,37 +155,52 @@ impl Message<StreamMessage<DiscoInternal, (), ()>> for DataplaneActor {
         };
 
         for mut buf in bufs {
-            let pkt = match Packet::from_encrypted_bytes_mut(buf.as_mut()) {
-                Ok(pkt) => pkt,
-                Err(e) => {
-                    tracing::error!(error = %e, "parsing disco message:\n{}",
-                        buf.iter().hexdump_string(Case::Lower)
-                    );
-                    return;
-                }
-            };
-
-            if let Err(e) = pkt.decrypt_in_place(&self.env.keys.disco_keys.private) {
-                tracing::error!(error = %e, "decrypting disco message");
-                return;
-            };
-
-            let pkt = yoke::Yoke::<&'static Packet<Plaintext>, _>::try_attach_to_cart(
-                buf.freeze(),
-                // SAFETY: we just parsed this from the same buffer, so type/version are set correctly.
-                |buf| unsafe { Packet::from_bytes_unchecked(buf) },
+            let pkt = tracing::debug_span!(
+                "disco_ingest",
+                ?transport_id,
+                ?ep,
+                sender_disco = tracing::field::Empty
             )
-            .unwrap();
+            .in_scope::<_, Option<IncomingDiscoMsg>>(|| {
+                let pkt = match Packet::from_encrypted_bytes_mut(buf.as_mut()) {
+                    Ok(pkt) => pkt,
+                    Err(e) => {
+                        tracing::error!(error = %e, "parsing disco message:\n{}",
+                            buf.iter().hexdump_string(Case::Lower)
+                        );
+                        return None;
+                    }
+                };
+
+                tracing::Span::current()
+                    .record("sender_disco", tracing::field::display(pkt.sender_pubkey()));
+
+                if let Err(e) = pkt.decrypt_in_place(&self.env.keys.disco_keys.private) {
+                    tracing::error!(error = %e, "decrypting disco message");
+                    return None;
+                };
+
+                let pkt = yoke::Yoke::<&'static Packet<Plaintext>, _>::try_attach_to_cart(
+                    buf.freeze(),
+                    // SAFETY: we just parsed this from the same buffer, so type/version are set correctly.
+                    |buf| unsafe { Packet::from_bytes_unchecked(buf) },
+                )
+                .unwrap();
 
-            let pkt = IncomingDiscoMsg {
-                transport: transport_id,
-                sender: ep.clone(),
-                packet: pkt,
-            };
+                let pkt = IncomingDiscoMsg {
+                    transport: transport_id,
+                    sender: ep.clone(),
+                    packet: pkt,
+                };
+
+                tracing::trace!(?pkt, "decrypted disco message");
 
-            tracing::trace!(?pkt, "decrypted disco message");
+                Some(pkt)
+            });
 
-            self.env.publish_noretain(pkt).await.unwrap();
+            if let Some(pkt) = pkt {
+                self.env.publish_noretain(pkt).await.unwrap();
+            }
         }
     }
 }