perf(datapath): eliminate allocations from publish routing hot path (agntcy#1577)

# Description

The publish routing hot path allocates unnecessarily on every inbound
publish message. `SlimHeader::get_dst()` converts the proto `Name` into
an internal `Name`, allocating `Arc<[String; 3]>` and cloning 3 strings
from the wire message. The routing path (subscription table matching)
never uses these strings — it only needs the four `u64` hash components
already present as `EncodedName` in the proto.

This PR threads a zero-allocation `EncodedName` (a `Copy` struct of 4
`u64`s) through the entire publish routing path, eliminating all heap
allocation from the hot path.

**Benchmark results (Apple M-series, release profile):**

| Benchmark | Time | Change |
|---|---|---|
| `routing_via_name` (before) | ~108 ns | baseline |
| `routing_via_encoded_name` (after) | ~38 ns | **~65% faster** |

Closes agntcy#1386

## Changes

**Phase 1 — Benchmark baseline (separate commit)**
- Add criterion benchmark suite covering `Name` construction and the
full publish routing loop (`routing_via_name` baseline)

**Phase 2 — Code changes**
- Add `SlimHeader::get_encoded_dst()` / `get_encoded_source()` returning
`EncodedName` (Copy, zero allocation)
- Add `From<&Name> for EncodedName` conversion
- Replace `SubscriptionTable::match_one` / `match_all` to accept
`&EncodedName` instead of `&Name`
- Replace `InternalName([u64; 4])` with `SubscriptionName([u64; 3])`,
eliminating unsafe `transmute` tricks and the redundant 4th component
(the id is never part of the subscription group key)
- Replace `representative_name: Name` in `NameState` with
`representative_strings: [String; 3]` — the human-readable name this
`NameState` represents in the subscription table — reconstructing a
`Name` only in the cold `for_each` path
- Simplify `process_publish` and `match_and_forward_msg`; retain
`get_dst()` call in the debug log for human-readable output
- Add `DataPathError::NoMatchEncoded` for the allocation-free error path

## Type of Change

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

## Checklist

- [ ] 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
- [ ] 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: Sam Betts <1769706+Tehsmash@users.noreply.github.com>

Author: Tehsmash

data-plane/core/datapath/Cargo.toml

@@ -49,3 +49,7 @@ tracing-test = { workspace = true, features = ["no-env-filter"] }
 [[bench]]
 name = "pool_benchmark"
 harness = false
+
+[[bench]]
+name = "name_benchmark"
+harness = false

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

@@ -0,0 +1,118 @@
+// Copyright AGNTCY Contributors (https://github.com/agntcy)
+// SPDX-License-Identifier: Apache-2.0
+
+use criterion::{Criterion, black_box, criterion_group, criterion_main};
+use slim_datapath::api::{EncodedName, ProtoName, SlimHeader, StringName};
+use slim_datapath::messages::encoder::Name;
+use slim_datapath::tables::SubscriptionTable;
+use slim_datapath::tables::subscription_table::SubscriptionTableImpl;
+
+fn make_proto_name() -> ProtoName {
+    ProtoName {
+        name: Some(EncodedName {
+            component_0: 0x1234_5678_9abc_def0,
+            component_1: 0xfeed_cafe_dead_beef,
+            component_2: 0x0101_0101_0101_0101,
+            component_3: u64::MAX,
+        }),
+        str_name: Some(StringName {
+            str_component_0: "org".to_string(),
+            str_component_1: "namespace".to_string(),
+            str_component_2: "agent".to_string(),
+        }),
+    }
+}
+
+/// Build a SlimHeader whose destination matches the subscription used in routing benchmarks
+/// (org/namespace/agent, id=42). Hashes are derived from Name::from_strings to ensure they
+/// match whatever SubscriptionTableImpl stores.
+fn make_slim_header() -> SlimHeader {
+    let name = Name::from_strings(["org", "namespace", "agent"]).with_id(42);
+    let c = name.components();
+    let proto_name = ProtoName {
+        name: Some(EncodedName {
+            component_0: c[0],
+            component_1: c[1],
+            component_2: c[2],
+            component_3: c[3],
+        }),
+        str_name: Some(StringName {
+            str_component_0: "org".to_string(),
+            str_component_1: "namespace".to_string(),
+            str_component_2: "agent".to_string(),
+        }),
+    };
+    SlimHeader {
+        source: Some(proto_name.clone()),
+        destination: Some(proto_name),
+        identity: String::new(),
+        fanout: 1,
+        recv_from: None,
+        forward_to: None,
+        incoming_conn: None,
+        error: None,
+    }
+}
+
+fn bench_name_from_strings(c: &mut Criterion) {
+    c.bench_function("name_from_strings", |b| {
+        b.iter(|| black_box(Name::from_strings(black_box(["org", "namespace", "agent"]))))
+    });
+}
+
+fn bench_name_from_proto(c: &mut Criterion) {
+    let proto_name = make_proto_name();
+    c.bench_function("name_from_proto", |b| {
+        b.iter(|| black_box(Name::from(black_box(&proto_name))))
+    });
+}
+
+fn bench_name_clone_with_strings(c: &mut Criterion) {
+    let name = Name::from_strings(["org", "namespace", "agent"]);
+    c.bench_function("name_clone_with_strings", |b| {
+        b.iter(|| black_box(name.clone()))
+    });
+}
+
+/// Baseline: full routing flow via the Name-based path.
+/// get_dst() allocates Arc<[String; 3]> + clones 3 Strings.
+fn bench_routing_via_name(c: &mut Criterion) {
+    let table = SubscriptionTableImpl::default();
+    let sub = Name::from_strings(["org", "namespace", "agent"]).with_id(42);
+    table.add_subscription(sub, 1, true, 1).unwrap();
+    let slim_header = make_slim_header();
+
+    c.bench_function("routing_via_name", |b| {
+        b.iter(|| {
+            let dst = black_box(slim_header.get_dst());
+            black_box(table.match_one(black_box(&EncodedName::from(&dst)), 0))
+        })
+    });
+}
+
+/// Optimized: full routing flow via the EncodedName path.
+/// get_encoded_dst() is a Copy of 4 u64s — zero heap allocation.
+fn bench_routing_via_encoded_name(c: &mut Criterion) {
+    let table = SubscriptionTableImpl::default();
+    let sub = Name::from_strings(["org", "namespace", "agent"]).with_id(42);
+    table.add_subscription(sub, 1, true, 1).unwrap();
+    let slim_header = make_slim_header();
+
+    c.bench_function("routing_via_encoded_name", |b| {
+        b.iter(|| {
+            let encoded = black_box(slim_header.get_encoded_dst());
+            black_box(table.match_one(black_box(&encoded), 0))
+        })
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_name_from_strings,
+    bench_name_from_proto,
+    bench_name_clone_with_strings,
+    bench_routing_via_name,
+    bench_routing_via_encoded_name,
+);
+
+criterion_main!(benches);

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

@@ -9,6 +9,7 @@ pub use proto::dataplane::v1::CommandPayload;
 pub use proto::dataplane::v1::Content;
 pub use proto::dataplane::v1::DiscoveryReplyPayload;
 pub use proto::dataplane::v1::DiscoveryRequestPayload;
+pub use proto::dataplane::v1::EncodedName;
 pub use proto::dataplane::v1::GroupAckPayload;
 pub use proto::dataplane::v1::GroupAddPayload;
 pub use proto::dataplane::v1::GroupNackPayload;
@@ -29,6 +30,7 @@ pub use proto::dataplane::v1::SessionHeader;
 pub use proto::dataplane::v1::SessionMessageType as ProtoSessionMessageType;
 pub use proto::dataplane::v1::SessionType as ProtoSessionType;
 pub use proto::dataplane::v1::SlimHeader;
+pub use proto::dataplane::v1::StringName;
 pub use proto::dataplane::v1::Subscribe as ProtoSubscribe;
 pub use proto::dataplane::v1::SubscriptionAck as ProtoSubscriptionAck;
 pub use proto::dataplane::v1::Unsubscribe as ProtoUnsubscribe;

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

@@ -27,6 +27,8 @@ pub enum DataPathError {
     // Subscription / matching
     #[error("no matching found for {0}")]
     NoMatch(Name),
+    #[error("no matching found for [{:x}, {:x}, {:x}, {:x}]", .0[0], .0[1], .0[2], .0[3])]
+    NoMatchEncoded([u64; 4]),
     #[error("subscription not found")]
     SubscriptionNotFound(Name),
     #[error("subscription id not found: {0}")]

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

@@ -8,6 +8,7 @@ use super::tables::SubscriptionTable;
 use super::tables::connection_table::ConnectionTable;
 use super::tables::remote_subscription_table::RemoteSubscriptions;
 use super::tables::subscription_table::SubscriptionTableImpl;
+use crate::api::EncodedName;
 use crate::errors::DataPathError;
 use crate::messages::Name;
 use crate::tables::remote_subscription_table::SubscriptionInfo;
@@ -129,16 +130,16 @@ where
 
     pub fn on_publish_msg_match(
         &self,
-        name: Name,
+        encoded: EncodedName,
         incoming_conn: u64,
         fanout: u32,
     ) -> Result<Vec<u64>, DataPathError> {
         if fanout == 1 {
             self.subscription_table
-                .match_one(&name, incoming_conn)
+                .match_one(&encoded, incoming_conn)
                 .map(|out| vec![out])
         } else {
-            self.subscription_table.match_all(&name, incoming_conn)
+            self.subscription_table.match_all(&encoded, incoming_conn)
         }
     }
 
@@ -177,15 +178,15 @@ mod tests {
         );
 
         assert_eq!(
-            fwd.on_publish_msg_match(name.clone().with_id(1), 100, 1)
+            fwd.on_publish_msg_match(EncodedName::from(&name.clone().with_id(1)), 100, 1)
                 .unwrap(),
             vec![12]
         );
 
         let expected = name.clone().with_id(2);
 
-        let err = fwd.on_publish_msg_match(expected.clone(), 100, 1);
-        assert!(matches!(err, Err(DataPathError::NoMatch(_))));
+        let err = fwd.on_publish_msg_match(EncodedName::from(&expected), 100, 1);
+        assert!(matches!(err, Err(DataPathError::NoMatchEncoded(_))));
 
         assert!(
             fwd.on_subscription_msg(name.clone(), 10, false, false, 1)

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

@@ -30,7 +30,7 @@ use crate::api::ProtoSubscriptionAckType as SubscriptionAckType;
 use crate::api::ProtoUnsubscribeType as UnsubscribeType;
 use crate::api::proto::dataplane::v1::Message;
 use crate::api::{
-    LinkNegotiationPayload, ProtoLink, ProtoLinkMessageType as LinkType, ProtoLinkType,
+    EncodedName, LinkNegotiationPayload, ProtoLink, ProtoLinkMessageType as LinkType, ProtoLinkType,
 };
 use semver;
 
@@ -435,15 +435,11 @@ impl MessageProcessor {
         &self,
         #[cfg(feature = "otel_tracing")] mut msg: Message,
         #[cfg(not(feature = "otel_tracing"))] msg: Message,
-        name: Name,
         in_connection: u64,
         fanout: u32,
     ) -> Result<(), DataPathError> {
-        debug!(
-            %name,
-            %fanout,
-            "match and forward message"
-        );
+        let header = msg.get_slim_header();
+        debug!(name = %header.get_dst(), %fanout, "match and forward message");
 
         // if the message already contains an output connection, use that one
         // without performing any match in the subscription table
@@ -452,9 +448,11 @@ impl MessageProcessor {
             return self.send_msg(msg, val).await;
         }
 
+        let encoded = header.get_encoded_dst();
+
         match self
             .forwarder()
-            .on_publish_msg_match(name, in_connection, fanout)
+            .on_publish_msg_match(encoded, in_connection, fanout)
         {
             Ok(out_vec) => {
                 let len = out_vec.len();
@@ -639,17 +637,11 @@ impl MessageProcessor {
         );
         //////////////////////////////////////////////////////
 
-        // get header
-        let header = msg.get_slim_header();
-
-        let dst = header.get_dst();
-
         // this function may panic, but at this point we are sure we are processing
         // a publish message
         let fanout = msg.get_fanout();
 
-        self.match_and_forward_msg(msg, dst, in_connection, fanout)
-            .await
+        self.match_and_forward_msg(msg, in_connection, fanout).await
     }
 
     pub(crate) async fn send_subscription_ack(
@@ -1227,7 +1219,7 @@ impl MessageProcessor {
                                     .into_iter()
                                     .filter(|(name, _)| {
                                         mp.forwarder()
-                                            .on_publish_msg_match(name.clone(), u64::MAX, u32::MAX)
+                                            .on_publish_msg_match(EncodedName::from(name), u64::MAX, u32::MAX)
                                             .is_err()
                                     })
                                     .collect();
@@ -1345,6 +1337,7 @@ mod tests {
 
     use super::*;
     use crate::api::ProtoSubscriptionAck;
+    use crate::messages::Name;
     use crate::tables::remote_subscription_table::SubscriptionInfo;
     use tonic::Status;
 
@@ -2298,9 +2291,10 @@ mod tests {
             .unwrap();
 
         // The subscription should have been restored in the routing table.
-        let result = processor
-            .forwarder()
-            .on_publish_msg_match(sub_name, u64::MAX, 1);
+        let result =
+            processor
+                .forwarder()
+                .on_publish_msg_match(EncodedName::from(&sub_name), u64::MAX, 1);
         assert!(result.is_ok(), "recovered subscription should be routable");
         assert_eq!(result.unwrap(), vec![conn_id]);
     }

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

@@ -5,7 +5,7 @@ use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 use twox_hash::XxHash64;
 
-use crate::api::ProtoName;
+use crate::api::{EncodedName, ProtoName};
 
 #[derive(Clone)]
 pub struct Name {
@@ -82,6 +82,18 @@ impl From<&ProtoName> for Name {
     }
 }
 
+impl From<&Name> for EncodedName {
+    fn from(name: &Name) -> Self {
+        let c = name.components();
+        EncodedName {
+            component_0: c[0],
+            component_1: c[1],
+            component_2: c[2],
+            component_3: c[3],
+        }
+    }
+}
+
 impl Name {
     // NULL_COMPONENT is used to represent a component that is not set
     pub const NULL_COMPONENT: u64 = u64::MAX;