feat(swarm): add swarm rendezvous messages for clients sharing records

src/chunk.rs

@@ -9,6 +9,11 @@
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 
+use crate::swarm::{
+    SwarmAnnounceRequest, SwarmAnnounceResponse, SwarmAnswerRequest, SwarmAnswerResponse,
+    SwarmLeaveRequest, SwarmLeaveResponse,
+};
+
 /// Protocol identifier for chunk operations.
 pub const CHUNK_PROTOCOL_ID: &str = "autonomi.ant.chunk.v1";
 
@@ -201,6 +206,24 @@ pub enum ChunkMessageBody {
     GetOrCloserRequest(ChunkGetOrCloserRequest),
     /// Response to a get-or-closer request.
     GetOrCloserResponse(ChunkGetOrCloserResponse),
+    /// Join or stay in a file's swarm room, hand over WebRTC offers and
+    /// collect signals ([`crate::swarm`]).
+    ///
+    /// Appended after every existing variant for the reason given on
+    /// [`Self::QuoteRequestV2`]: a node built before swarms existed rejects it
+    /// cleanly as an unknown discriminant, and the client falls back to
+    /// fetching from nodes alone.
+    SwarmAnnounceRequest(SwarmAnnounceRequest),
+    /// Response to a swarm announce.
+    SwarmAnnounceResponse(SwarmAnnounceResponse),
+    /// Answer an offer the node delivered.
+    SwarmAnswerRequest(SwarmAnswerRequest),
+    /// Response to a swarm answer.
+    SwarmAnswerResponse(SwarmAnswerResponse),
+    /// Leave a swarm room.
+    SwarmLeaveRequest(SwarmLeaveRequest),
+    /// Response to a swarm leave.
+    SwarmLeaveResponse(SwarmLeaveResponse),
 }
 
 /// Whether a peer's user agent says it answers
@@ -358,15 +381,24 @@ impl ChunkMessage {
     ///
     /// Returns [`ProtocolError::MessageTooLarge`] if the input exceeds the
     /// size limit, or [`ProtocolError::DeserializationFailed`] if postcard
-    /// cannot parse the data.
+    /// cannot parse the data or bytes follow the message's end: a sender
+    /// cannot pad a small message out to the size limit and have it taken.
     pub fn decode(data: &[u8]) -> Result<Self, ProtocolError> {
         if data.len() > MAX_WIRE_MESSAGE_SIZE {
             return Err(ProtocolError::MessageTooLarge {
                 size: data.len(),
                 max_size: MAX_WIRE_MESSAGE_SIZE,
             });
         }
-        postcard::from_bytes(data).map_err(|e| ProtocolError::DeserializationFailed(e.to_string()))
+        let (message, rest) = postcard::take_from_bytes(data)
+            .map_err(|e| ProtocolError::DeserializationFailed(e.to_string()))?;
+        if !rest.is_empty() {
+            return Err(ProtocolError::DeserializationFailed(format!(
+                "{} bytes after the end of the message",
+                rest.len()
+            )));
+        }
+        Ok(message)
     }
 }
 
@@ -864,6 +896,7 @@ impl std::error::Error for ProtocolError {}
 #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 mod tests {
     use super::*;
+    use crate::swarm::{SwarmRefusal, SwarmSignal};
 
     #[test]
     fn test_put_request_encode_decode() {
@@ -1015,6 +1048,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_decode_rejects_bytes_after_the_message() {
+        let message = ChunkMessage {
+            request_id: 7,
+            body: ChunkMessageBody::GetRequest(ChunkGetRequest::new([3u8; 32])),
+        };
+        let mut bytes = message.encode().unwrap();
+        assert_eq!(ChunkMessage::decode(&bytes).unwrap().request_id, 7);
+        bytes.push(0);
+        let err = ChunkMessage::decode(&bytes).unwrap_err();
+        assert!(
+            matches!(err, ProtocolError::DeserializationFailed(_)),
+            "expected DeserializationFailed, got {err:?}"
+        );
+    }
+
     #[test]
     fn test_invalid_decode() {
         let invalid_data = vec![0xFF, 0xFF, 0xFF];
@@ -1227,6 +1276,104 @@ mod tests {
         );
     }
 
+    /// The swarm variants come after pointers (10 to 13) and get-or-closer
+    /// (14 and 15), so a node that predates them rejects them as unknown and
+    /// misreads nothing.
+    #[test]
+    fn swarm_variants_are_appended_after_get_or_closer() {
+        let room = [0x33; 32];
+        assert_eq!(
+            wire_discriminant(ChunkMessageBody::GetOrCloserResponse(
+                ChunkGetOrCloserResponse::Found {
+                    address: room,
+                    content: Bytes::new(),
+                }
+            )),
+            15,
+        );
+        let bodies = [
+            ChunkMessageBody::SwarmAnnounceRequest(SwarmAnnounceRequest {
+                room,
+                node: [3; 32],
+                membership: 1,
+                offers: Vec::new(),
+                wait_ms: 0,
+            }),
+            ChunkMessageBody::SwarmAnnounceResponse(SwarmAnnounceResponse::Refused(
+                SwarmRefusal::Busy,
+            )),
+            ChunkMessageBody::SwarmAnswerRequest(SwarmAnswerRequest {
+                room,
+                node: [3; 32],
+                to: [1; 16],
+                offer_id: [2; 16],
+                sdp: String::new(),
+            }),
+            ChunkMessageBody::SwarmAnswerResponse(SwarmAnswerResponse::Delivered),
+            ChunkMessageBody::SwarmLeaveRequest(SwarmLeaveRequest {
+                room,
+                node: [3; 32],
+                membership: 1,
+            }),
+            ChunkMessageBody::SwarmLeaveResponse(SwarmLeaveResponse::Left),
+        ];
+        let indices: Vec<u8> = bodies.into_iter().map(wire_discriminant).collect();
+        assert_eq!(indices, vec![16, 17, 18, 19, 20, 21]);
+    }
+
+    #[test]
+    fn a_swarm_announce_round_trips() {
+        let message = ChunkMessage {
+            request_id: 7,
+            body: ChunkMessageBody::SwarmAnnounceResponse(SwarmAnnounceResponse::Joined {
+                member: [4; 16],
+                members: 3,
+                offers_delivered: 2,
+                inbox: vec![
+                    SwarmSignal::Offer {
+                        from: [5; 16],
+                        offer_id: [6; 16],
+                        sdp: "v=0 offer".into(),
+                    },
+                    SwarmSignal::Answer {
+                        from: [7; 16],
+                        offer_id: [8; 16],
+                        sdp: "v=0 answer".into(),
+                    },
+                ],
+                ice_servers: vec!["stun:203.0.113.5:10000".into()],
+            }),
+        };
+        let decoded = ChunkMessage::decode(&message.encode().expect("encode")).expect("decode");
+        assert_eq!(decoded.request_id, 7);
+        match decoded.body {
+            ChunkMessageBody::SwarmAnnounceResponse(response) => {
+                assert_eq!(
+                    response,
+                    SwarmAnnounceResponse::Joined {
+                        member: [4; 16],
+                        members: 3,
+                        offers_delivered: 2,
+                        inbox: vec![
+                            SwarmSignal::Offer {
+                                from: [5; 16],
+                                offer_id: [6; 16],
+                                sdp: "v=0 offer".into(),
+                            },
+                            SwarmSignal::Answer {
+                                from: [7; 16],
+                                offer_id: [8; 16],
+                                sdp: "v=0 answer".into(),
+                            },
+                        ],
+                        ice_servers: vec!["stun:203.0.113.5:10000".into()],
+                    }
+                );
+            }
+            other => panic!("decoded {other:?}"),
+        }
+    }
+
     /// `ProtocolError` rides inside quote responses, so its variants carry the
     /// same append-only obligation as the message bodies.
     #[test]

src/lib.rs

@@ -17,6 +17,8 @@
 //!   [`chunk_protocol::send_and_await_chunk_response`] that both the
 //!   client and node test harness use to exchange chunk messages on
 //!   a `saorsa-core::P2PNode`.
+//! - [`swarm`] — messages that let clients downloading the same file find
+//!   each other through the nodes closest to it and share its records.
 //! - [`payment`] — on-wire payment artifacts: `PaymentProof`,
 //!   `SingleNodePayment` (with `pay` and `verify` co-located), and
 //!   ML-DSA-65 verification of quotes and merkle candidates.
@@ -50,6 +52,7 @@ pub mod error;
 pub mod logging;
 pub mod payment;
 pub mod pointer;
+pub mod swarm;
 
 // =============================================================================
 // Public surface re-exports
@@ -85,6 +88,15 @@ pub use pointer::{
     PointerTarget, PointerTargetKind, DATA_TYPE_POINTER, FINAL_COUNTER, POINTER_BODY_LEN,
     POINTER_FORMAT_VERSION, POINTER_WIRE_LEN, TARGET_WIRE_LEN,
 };
+pub use swarm::{
+    canonical_ip, constrain_swarm_sdp, is_global_ip, is_local_ip, MemberId, Membership, NodeId,
+    OfferId, RoomKey, SwarmAnnounceRequest, SwarmAnnounceResponse, SwarmAnswerRequest,
+    SwarmAnswerResponse, SwarmLeaveRequest, SwarmLeaveResponse, SwarmOffer, SwarmRefusal,
+    SwarmSignal, MAX_SWARM_CANDIDATES, MAX_SWARM_ERROR_BYTES, MAX_SWARM_ICE_SERVERS,
+    MAX_SWARM_ICE_SERVER_BYTES, MAX_SWARM_INBOX, MAX_SWARM_OFFERS, MAX_SWARM_REPLY_SDP_BYTES,
+    MAX_SWARM_REQUEST_BYTES, MAX_SWARM_SDP_BYTES, MAX_SWARM_WAIT_MS, SWARM_RENDEZVOUS_NODES,
+    SWARM_SIGNAL_TTL_MS,
+};
 
 // =============================================================================
 // Transitive-dep re-exports