feat(datapath): add TTL support for message forwarding (agntcy#1714)

# Description

Add a hop-by-hop TTL (time-to-live) field to the SLIMHeader that
controls how many times a message can be forwarded between SLIM nodes

Fixes: agntcy#1715 

## Type of Change

- [ ] Bugfix
- [x] New Feature
- [ ] Breaking Change
- [ ] Refactor
- [ ] Documentation
- [ ] Other (please describe)

## Checklist

- [x] I have read the [contributing
guidelines](/agntcy/repo-template/blob/main/CONTRIBUTING.md)
- [x] Existing issues have been referenced (where applicable)
- [x] I have verified this change is not present in other open pull
requests
- [x] Functionality is documented
- [x] All code style checks pass
- [x] New code contribution is covered by automated tests
- [x] All new and existing tests pass

Signed-off-by: Mauro Sardara <msardara@cisco.com>

Author: msardara

data-plane/core/datapath/benches/name_benchmark.rs

@@ -3,6 +3,7 @@
 
 use criterion::{Criterion, black_box, criterion_group, criterion_main};
 use slim_datapath::api::{EncodedName, NameId, ProtoName, SlimHeader, StringName};
+use slim_datapath::messages::utils::DEFAULT_TTL;
 use slim_datapath::tables::SubscriptionTable;
 use slim_datapath::tables::subscription_table::SubscriptionTableImpl;
 use slim_datapath::tables::{ConnType, MatchFilter};
@@ -38,6 +39,7 @@ fn make_slim_header() -> SlimHeader {
         incoming_conn: None,
         error: None,
         header_mac: None,
+        ttl: DEFAULT_TTL,
     }
 }
 

data-plane/core/datapath/src/api/gen/dataplane.proto.v1.rs

@@ -84,15 +84,17 @@ pub struct SlimHeader {
     pub fanout: u32,
     #[prost(string, tag = "5")]
     pub version: ::prost::alloc::string::String,
-    #[prost(uint64, optional, tag = "6")]
-    pub recv_from: ::core::option::Option<u64>,
+    #[prost(uint32, tag = "6")]
+    pub ttl: u32,
     #[prost(uint64, optional, tag = "7")]
-    pub forward_to: ::core::option::Option<u64>,
+    pub recv_from: ::core::option::Option<u64>,
     #[prost(uint64, optional, tag = "8")]
+    pub forward_to: ::core::option::Option<u64>,
+    #[prost(uint64, optional, tag = "9")]
     pub incoming_conn: ::core::option::Option<u64>,
-    #[prost(bool, optional, tag = "9")]
+    #[prost(bool, optional, tag = "10")]
     pub error: ::core::option::Option<bool>,
-    #[prost(bytes = "vec", optional, tag = "10")]
+    #[prost(bytes = "vec", optional, tag = "11")]
     pub header_mac: ::core::option::Option<::prost::alloc::vec::Vec<u8>>,
 }
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]

data-plane/core/datapath/src/errors.rs

@@ -90,6 +90,9 @@ pub enum DataPathError {
 
     #[error("inter-node ephemeral key generation failed")]
     LinkKeyGeneration,
+
+    #[error("message TTL expired")]
+    TtlExpired,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

data-plane/core/datapath/src/header_mac.rs

@@ -148,12 +148,16 @@ fn preimage_upper_bound(header: &SlimHeader) -> usize {
     /// Byte size of error: Option<bool> encoded by [`push_bool_opt`]
     const ERROR_SIZE: usize = 2;
 
+    /// Byte size of TTL: u32 serialized by [`to_le_bytes`]
+    const TTL_SIZE: usize = 4;
+
     DOMAIN_V1.len()
         + LINK_UUID_SIZE
         + FANOUT_SIZE
         + RECV_FROM_SIZE
         + FORWARD_TO_SIZE
         + ERROR_SIZE
+        + TTL_SIZE
         + encoded_name_upper_bound(&header.source)
         + encoded_name_upper_bound(&header.destination)
 }
@@ -268,12 +272,14 @@ fn write_preimage(buf: &mut Vec<u8>, hdr: &SlimHeader, link_uuid: &[u8; 16]) {
     push_u64_opt(buf, hdr.recv_from);
     push_u64_opt(buf, hdr.forward_to);
     push_bool_opt(buf, hdr.error);
+    buf.extend_from_slice(&hdr.ttl.to_le_bytes());
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::api::proto::dataplane::v1::{EncodedName, Name, NameId, StringName};
+    use crate::messages::utils::DEFAULT_TTL;
 
     fn test_key() -> Vec<u8> {
         b"01234567890123456789012345678901".to_vec()
@@ -315,6 +321,7 @@ mod tests {
             incoming_conn: Some(999),
             error: Some(false),
             header_mac: None,
+            ttl: DEFAULT_TTL,
         }
     }
 

data-plane/core/datapath/src/link_ecdh.rs

@@ -59,6 +59,7 @@ pub fn derive_header_mac_from_ecdh(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::messages::utils::DEFAULT_TTL;
 
     #[test]
     fn ecdh_hkdf_matches_between_initiator_and_responder() {
@@ -79,6 +80,7 @@ mod tests {
             incoming_conn: None,
             error: None,
             header_mac: None,
+            ttl: DEFAULT_TTL,
         };
         a.sign_slim_header(&mut h, &lid).unwrap();
         b.verify_slim_header(&h, &lid).unwrap();

data-plane/core/datapath/src/message_processing.rs

@@ -1336,6 +1336,12 @@ impl MessageProcessor {
             // add incoming connection to the SLIM header
             msg.set_incoming_conn(Some(conn_index));
 
+            // TTL processing: decrement for remote messages (hop-by-hop)
+            if !category.is_local() && msg.decrement_ttl() == 0 {
+                debug!(%conn_index, "dropping message: TTL expired");
+                return Err(DataPathError::TtlExpired);
+            }
+
             #[cfg(feature = "otel_tracing")]
             otel_tracing::prepare_inbound_msg(
                 &mut msg,

data-plane/core/datapath/src/messages/utils.rs

@@ -57,6 +57,9 @@ pub const FALSE_VAL: &str = "FALSE";
 /// Value: Half of u32::MAX to allow a separate ID space for out-of-band messages.
 pub const MAX_PUBLISH_ID: u32 = u32::MAX / 2;
 
+/// Default TTL value for messages that do not have an explicit TTL set.
+pub const DEFAULT_TTL: u32 = 16;
+
 #[derive(Error, Debug, PartialEq)]
 pub enum MessageError {
     #[error("SLIM header not found")]
@@ -181,6 +184,7 @@ pub struct SlimHeaderFlags {
     pub forward_to: Option<u64>,
     pub incoming_conn: Option<u64>,
     pub error: Option<bool>,
+    pub ttl: u32,
 }
 
 impl Default for SlimHeaderFlags {
@@ -191,6 +195,7 @@ impl Default for SlimHeaderFlags {
             forward_to: None,
             incoming_conn: None,
             error: None,
+            ttl: DEFAULT_TTL,
         }
     }
 }
@@ -199,8 +204,8 @@ impl Display for SlimHeaderFlags {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "fanout: {}, recv_from: {:?}, forward_to: {:?}, incoming_conn: {:?}, error: {:?}",
-            self.fanout, self.recv_from, self.forward_to, self.incoming_conn, self.error
+            "fanout: {}, recv_from: {:?}, forward_to: {:?}, incoming_conn: {:?}, error: {:?}, ttl: {:?}",
+            self.fanout, self.recv_from, self.forward_to, self.incoming_conn, self.error, self.ttl
         )
     }
 }
@@ -219,6 +224,7 @@ impl SlimHeaderFlags {
             forward_to,
             incoming_conn,
             error,
+            ttl: DEFAULT_TTL,
         }
     }
 
@@ -253,6 +259,10 @@ impl SlimHeaderFlags {
             ..self
         }
     }
+
+    pub fn with_ttl(self, ttl: u32) -> Self {
+        Self { ttl, ..self }
+    }
 }
 
 /// SLIM Header
@@ -277,6 +287,7 @@ impl SlimHeader {
             incoming_conn: flags.incoming_conn,
             error: flags.error,
             header_mac: None,
+            ttl: flags.ttl,
         }
     }
 
@@ -365,6 +376,20 @@ impl SlimHeader {
         self.error = error;
     }
 
+    pub fn get_ttl(&self) -> u32 {
+        self.ttl
+    }
+
+    pub fn set_ttl(&mut self, ttl: u32) {
+        self.ttl = ttl;
+    }
+
+    /// Decrements TTL by 1. Returns the new value.
+    pub fn decrement_ttl(&mut self) -> u32 {
+        self.ttl = self.ttl.saturating_sub(1);
+        self.ttl
+    }
+
     // returns (incoming, recv_from, forward_to) for subscription processing
     pub(crate) fn get_connections(&self) -> (u64, Option<u64>, Option<u64>) {
         // when calling this function, incoming connection is set
@@ -865,6 +890,19 @@ impl ProtoMessage {
         self.get_slim_header_mut().set_error_flag(error);
     }
 
+    pub fn get_ttl(&self) -> u32 {
+        self.get_slim_header().get_ttl()
+    }
+
+    pub fn set_ttl(&mut self, ttl: u32) {
+        self.get_slim_header_mut().set_ttl(ttl);
+    }
+
+    /// Decrements TTL by 1. Returns the new value.
+    pub fn decrement_ttl(&mut self) -> u32 {
+        self.get_slim_header_mut().decrement_ttl()
+    }
+
     pub fn set_session_message_type(&mut self, message_type: SessionMessageType) {
         self.get_session_header_mut()
             .set_session_message_type(message_type);
@@ -1448,36 +1486,37 @@ impl ProtoMessageBuilder {
 
     /// Sets the fanout value
     pub fn fanout(mut self, fanout: u32) -> Self {
-        let flags = self.flags.take().unwrap_or_default();
-        self.flags = Some(flags.with_fanout(fanout));
+        self.flags.get_or_insert_default().fanout = fanout;
         self
     }
 
     /// Sets the recv_from connection
     pub fn recv_from(mut self, recv_from: u64) -> Self {
-        let flags = self.flags.take().unwrap_or_default();
-        self.flags = Some(flags.with_recv_from(recv_from));
+        self.flags.get_or_insert_default().recv_from = Some(recv_from);
         self
     }
 
     /// Sets the forward_to connection
     pub fn forward_to(mut self, forward_to: u64) -> Self {
-        let flags = self.flags.take().unwrap_or_default();
-        self.flags = Some(flags.with_forward_to(forward_to));
+        self.flags.get_or_insert_default().forward_to = Some(forward_to);
         self
     }
 
     /// Sets the incoming connection
     pub fn incoming_conn(mut self, incoming_conn: u64) -> Self {
-        let flags = self.flags.take().unwrap_or_default();
-        self.flags = Some(flags.with_incoming_conn(incoming_conn));
+        self.flags.get_or_insert_default().incoming_conn = Some(incoming_conn);
         self
     }
 
     /// Sets the error flag
     pub fn error(mut self, error: bool) -> Self {
-        let flags = self.flags.take().unwrap_or_default();
-        self.flags = Some(flags.with_error(error));
+        self.flags.get_or_insert_default().error = Some(error);
+        self
+    }
+
+    /// Sets the TTL (time-to-live) value
+    pub fn ttl(mut self, ttl: u32) -> Self {
+        self.flags.get_or_insert_default().ttl = ttl;
         self
     }
 
@@ -1547,6 +1586,7 @@ impl ProtoMessageBuilder {
             forward_to: header.forward_to,
             incoming_conn: header.incoming_conn,
             error: header.error,
+            ttl: header.ttl,
         };
         self.flags = Some(flags);
         self
@@ -2011,6 +2051,7 @@ mod tests {
             incoming_conn: None,
             error: None,
             header_mac: None,
+            ttl: DEFAULT_TTL,
         };
 
         // the operations to retrieve source and destination should fail with panic

proto/data-plane/v1/data_plane.proto

@@ -60,11 +60,12 @@ message SLIMHeader {
   string identity = 3;
   uint32 fanout = 4;
   string version = 5;
-  optional uint64 recv_from = 6;
-  optional uint64 forward_to = 7;
-  optional uint64 incoming_conn = 8;
-  optional bool error = 9;
-  optional bytes header_mac = 10;
+  uint32 ttl = 6;
+  optional uint64 recv_from = 7;
+  optional uint64 forward_to = 8;
+  optional uint64 incoming_conn = 9;
+  optional bool error = 10;
+  optional bytes header_mac = 11;
 }
 
 message SessionHeader {