Add straceable protocol operation binaries

Factor the benchmark scenarios into standalone example binaries that can be traced directly without cargo run, and simplify the hot call and hook lookup paths using the benchmark output as guidance.

src/protocol/tree/endpoint/receive.rs

@@ -6,7 +6,7 @@ use crate::protocol::{
     introspection::INTROSPECTION_PROCEDURE_ID, validate_call, validate_header,
 };
 
-use super::super::{HookKey, PendingHook, RouteDecision};
+use super::super::{ActiveHook, HookKey, RouteDecision};
 use super::core::{
     Endpoint, EndpointError, EndpointOutcome, Ingress, LocalEvent, ProtocolEndpoint,
 };
@@ -22,26 +22,7 @@ impl ProtocolEndpoint {
             .as_ref()
             .map(|hook| HookKey::new(hook.return_path.clone(), hook.hook_id));
 
-        if let Some(hook) = &message.response_hook
-            && hook.return_path != self.path
-            && self
-                .hooks
-                .insert_pending(PendingHook {
-                    return_path: hook.return_path.clone(),
-                    hook_id: hook.hook_id,
-                    caller_src_path: header.src_path.clone(),
-                    procedure_id: message.procedure_id.clone(),
-                    dst_leaf: header.dst_leaf.clone(),
-                })
-                .is_err()
-        {
-            return self.emit_fault_if_possible(key, ProtocolFault::INTERNAL_ERROR);
-        }
-
         if message.procedure_id == INTROSPECTION_PROCEDURE_ID {
-            if let Some(key) = &key {
-                self.hooks.activate_pending(key);
-            }
             return self.handle_introspection(&header, key);
         }
 
@@ -71,10 +52,22 @@ impl ProtocolEndpoint {
             return self.emit_fault_if_possible(key, fault);
         }
 
-        if let Some(key) = &key
-            && self.hooks.activate_pending(key).is_none()
+        if let Some(hook) = &message.response_hook
+            && hook.return_path != self.path
+            && self
+                .hooks
+                .insert_active(ActiveHook {
+                    return_path: hook.return_path.clone(),
+                    hook_id: hook.hook_id,
+                    peer_path: header.src_path.clone(),
+                    procedure_id: message.procedure_id.clone(),
+                    dst_leaf: header.dst_leaf.clone(),
+                    local_ended: false,
+                    peer_ended: false,
+                })
+                .is_err()
         {
-            return self.emit_fault_if_possible(Some(key.clone()), ProtocolFault::INTERNAL_ERROR);
+            return self.emit_fault_if_possible(key, ProtocolFault::INTERNAL_ERROR);
         }
 
         Ok(EndpointOutcome::event(LocalEvent::Call { header, message }))

src/protocol/tree/hook.rs

@@ -41,12 +41,6 @@ pub struct ActiveHook {
     pub peer_ended: bool,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
-struct PeerHookKey {
-    hook_id: u64,
-    peer_path: Vec<String>,
-}
-
 /// Duplicate hook insertion error.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct HookConflict;
@@ -56,7 +50,7 @@ pub struct HookConflict;
 pub struct HookTable {
     pending: BTreeMap<HookKey, PendingHook>,
     active: BTreeMap<HookKey, ActiveHook>,
-    active_by_peer: BTreeMap<PeerHookKey, HookKey>,
+    active_by_peer: BTreeMap<u64, BTreeMap<Vec<String>, HookKey>>,
     next_id: u64,
 }
 
@@ -94,17 +88,19 @@ impl HookTable {
 
     pub fn insert_active(&mut self, active: ActiveHook) -> Result<(), HookConflict> {
         let key = HookKey::new(active.return_path.clone(), active.hook_id);
-        let peer_key = PeerHookKey {
-            hook_id: active.hook_id,
-            peer_path: active.peer_path.clone(),
-        };
         if self.pending.contains_key(&key)
             || self.active.contains_key(&key)
-            || self.active_by_peer.contains_key(&peer_key)
+            || self
+                .active_by_peer
+                .get(&active.hook_id)
+                .is_some_and(|peer_paths| peer_paths.contains_key(active.peer_path.as_slice()))
         {
             return Err(HookConflict);
         }
-        self.active_by_peer.insert(peer_key, key.clone());
+        self.active_by_peer
+            .entry(active.hook_id)
+            .or_default()
+            .insert(active.peer_path.clone(), key.clone());
         self.active.insert(key, active);
         Ok(())
     }
@@ -115,10 +111,12 @@ impl HookTable {
 
     pub fn remove_active(&mut self, key: &HookKey) -> Option<ActiveHook> {
         let active = self.active.remove(key)?;
-        self.active_by_peer.remove(&PeerHookKey {
-            hook_id: active.hook_id,
-            peer_path: active.peer_path.clone(),
-        });
+        if let Some(peer_paths) = self.active_by_peer.get_mut(&active.hook_id) {
+            peer_paths.remove(active.peer_path.as_slice());
+            if peer_paths.is_empty() {
+                self.active_by_peer.remove(&active.hook_id);
+            }
+        }
         Some(active)
     }
 
@@ -147,12 +145,7 @@ impl HookTable {
         if self.active.contains_key(&host_key) {
             return Some(host_key);
         }
-        self.active_by_peer
-            .get(&PeerHookKey {
-                hook_id,
-                peer_path: peer_path.to_vec(),
-            })
-            .cloned()
+        self.active_by_peer.get(&hook_id)?.get(peer_path).cloned()
     }
 
     pub fn mark_local_end(&mut self, key: &HookKey) -> bool {