astatin3/unshell
1
0
Fork 0
mirror of https://github.com/Astatin3/unshell.git synced 2026-06-09 06:47:59 -06:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite protocol flow around compiled routing

Browse Source 
Compile routing prefixes once per endpoint, restore minimal pending-to-active hook transitions, and route call/data/fault packets from the header before decoding payloads for local delivery only. Document the remaining protocol-level pressure points in src/protocol/PROTOCOL_CHANGES.md.
This commit is contained in:
Michael Mikovsky
2026-04-25 12:15:38 -06:00
parent 62b22be39f
commit 3d92b5cf0d
9 changed files with 477 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/protocol/tree/endpoint/builders.rs
+18 -3
View File
@@ -5,13 +5,13 @@
use alloc::{collections::BTreeSet, string::String, vec::Vec};
use crate::protocol::tree::ActiveHook;
use crate::protocol::tree::{ActiveHook, HookKey};
use crate::protocol::{
CallMessage, DataMessage, FrameBytes, HookTarget, PacketHeader, PacketType, ValidationError,
encode_packet, validate_call, validate_header, validate_procedure_id,
};
use super::super::RouteDecision;
use super::super::{CompiledRoutes, RouteDecision};
use super::core::{ChildRoute, EndpointError, EndpointOutcome, ProtocolEndpoint};
use crate::protocol::tree::LeafSpec;
@@ -63,6 +63,8 @@ impl ProtocolEndpoint {
peer_path: header.dst_path.clone(),
procedure_id: call.procedure_id.clone(),
dst_leaf: header.dst_leaf.clone(),
local_ended: false,
peer_ended: false,
})
.is_err()
{
@@ -114,9 +116,15 @@ impl ProtocolEndpoint {
children: Vec<ChildRoute>,
leaves: Vec<LeafSpec>,
) -> Self {
let registered_children = children
.iter()
.filter(|child| child.state == super::core::ConnectionState::Registered)
.map(|child| child.path.clone())
.collect::<Vec<_>>();
Self {
routing: CompiledRoutes::new(&path, &registered_children, parent_path.is_some()),
path,
parent_path,
children,
leaves: leaves
.into_iter()
@@ -202,6 +210,13 @@ impl ProtocolEndpoint {
) -> Result<EndpointOutcome, EndpointError> {
let (header, message) = self.prepare_data(dst_path, hook_id, procedure_id, data, end_hook)?;
if end_hook {
let key = HookKey::new(self.path.clone(), hook_id);
if self.hooks.mark_local_end(&key) {
self.hooks.remove_active(&key);
}
}
match self.decide_route(&header.dst_path) {
RouteDecision::Local => self.handle_local_data(header, message),
route => Ok(EndpointOutcome::forward(route, encode_packet(&header, &message)?)),
src/protocol/tree/endpoint/core.rs
+2 -2
View File
@@ -16,7 +16,7 @@ use crate::protocol::{
CallMessage, DataMessage, FaultMessage, FrameBytes, FrameError, PacketHeader, ValidationError,
};
use super::super::{HookTable, RouteDecision};
use super::super::{CompiledRoutes, HookTable, RouteDecision};
/// Local connection state used for child route eligibility.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -182,8 +182,8 @@ pub trait Endpoint {
#[derive(Debug, Default)]
pub struct ProtocolEndpoint {
pub(crate) path: Vec<String>,
pub(crate) parent_path: Option<Vec<String>>,
pub(crate) children: Vec<ChildRoute>,
pub(crate) routing: CompiledRoutes,
pub(crate) leaves: BTreeMap<String, LeafSpec>,
pub(crate) endpoint_procedures: BTreeSet<String>,
pub(crate) hooks: HookTable,
src/protocol/tree/endpoint/hooks.rs
+14 -15
View File
@@ -9,7 +9,7 @@ use crate::protocol::{
DataMessage, FaultMessage, PacketHeader, PacketType, ProtocolFault, encode_packet,
};
use super::super::{HookKey, RouteDecision, route_destination};
use super::super::{HookKey, RouteDecision};
use super::core::{EndpointError, EndpointOutcome, Ingress, LocalEvent, ProtocolEndpoint};
impl ProtocolEndpoint {
@@ -23,6 +23,7 @@ impl ProtocolEndpoint {
return Ok(EndpointOutcome::dropped());
};
self.hooks.remove_pending(&key);
self.hooks.remove_active(&key);
let header = PacketHeader {
@@ -82,7 +83,7 @@ impl ProtocolEndpoint {
return Ok(EndpointOutcome::dropped());
}
if message.end_hook {
if message.end_hook && self.hooks.mark_peer_end(&key) {
self.hooks.remove_active(&key);
}
@@ -100,7 +101,16 @@ impl ProtocolEndpoint {
header.hook_id.expect("validated"),
&header.src_path,
) else {
return Ok(EndpointOutcome::dropped());
let key = HookKey::new(self.path.clone(), header.hook_id.expect("validated"));
let matches_pending = self
.hooks
.pending(&key)
.is_some_and(|pending| pending.caller_src_path == header.src_path);
if !matches_pending {
return Ok(EndpointOutcome::dropped());
}
self.hooks.remove_pending(&key);
return Ok(EndpointOutcome::event(LocalEvent::Fault { header, message }));
};
self.hooks.remove_active(&key);
@@ -110,18 +120,7 @@ impl ProtocolEndpoint {
/// Chooses the next hop using the protocol's longest-prefix routing rule.
pub(crate) fn decide_route(&self, dst_path: &[String]) -> RouteDecision {
let child_paths = self
.children
.iter()
.filter(|child| child.state == super::core::ConnectionState::Registered)
.map(|child| &child.path);
route_destination(
&self.path,
child_paths,
self.parent_path.is_some(),
dst_path,
)
self.routing.route(dst_path)
}
/// Validates whether a source path is attributable to the ingress side.
src/protocol/tree/endpoint/receive.rs
+36 -15
View File
@@ -4,11 +4,12 @@
//! the `Call`, `Data`, and `Fault` sections of `PROTOCOL.md`.
use crate::protocol::{
CallMessage, PacketType, ProtocolFault, decode_frame, introspection::INTROSPECTION_PROCEDURE_ID,
validate_call, validate_header,
CallMessage, PacketType, ProtocolFault, decode_frame, deserialize_archived_bytes,
introspection::INTROSPECTION_PROCEDURE_ID, validate_call, validate_header,
};
use crate::protocol::types::{ArchivedCallMessage, ArchivedDataMessage, ArchivedFaultMessage};
use super::super::{ActiveHook, HookKey, RouteDecision};
use super::super::{HookKey, PendingHook, RouteDecision};
use super::core::{
Endpoint, EndpointError, EndpointOutcome, Ingress, LocalEvent, ProtocolEndpoint,
};
@@ -57,18 +58,26 @@ impl ProtocolEndpoint {
if let Some(hook) = &message.response_hook
&& hook.return_path != self.path
&& self
{
if self
.hooks
.insert_active(ActiveHook {
.insert_pending(PendingHook {
return_path: hook.return_path.clone(),
hook_id: hook.hook_id,
peer_path: header.src_path.clone(),
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);
{
return self.emit_fault_if_possible(key, ProtocolFault::INTERNAL_ERROR);
}
if let Some(key) = &key
&& self.hooks.activate_pending(key).is_none()
{
return self.emit_fault_if_possible(Some(key.clone()), ProtocolFault::INTERNAL_ERROR);
}
}
Ok(EndpointOutcome::event(LocalEvent::Call { header, message }))
@@ -95,12 +104,10 @@ impl Endpoint for ProtocolEndpoint {
match header.packet_type {
PacketType::Call => {
let message = parsed.deserialize_call()?;
if !matches!(ingress, Ingress::Parent | Ingress::Local) {
return Ok(EndpointOutcome::dropped());
}
validate_call(header, &message)?;
match self.decide_route(&header.dst_path) {
RouteDecision::Child(index) => {
Ok(EndpointOutcome::forward(RouteDecision::Child(index), frame))
@@ -109,14 +116,24 @@ impl Endpoint for ProtocolEndpoint {
Ok(EndpointOutcome::forward(RouteDecision::Parent, frame))
}
RouteDecision::Drop => Ok(EndpointOutcome::dropped()),
RouteDecision::Local => self.handle_local_call(parsed.deserialize_header(), message),
RouteDecision::Local => {
let (header, payload) = parsed.into_parts();
let message =
deserialize_archived_bytes::<ArchivedCallMessage, CallMessage>(payload)?;
validate_call(&header, &message)?;
self.handle_local_call(header, message)
}
}
}
PacketType::Data => {
match self.decide_route(&header.dst_path) {
RouteDecision::Local => {
let message = parsed.deserialize_data()?;
self.handle_local_data(parsed.deserialize_header(), message)
let (header, payload) = parsed.into_parts();
let message = deserialize_archived_bytes::<
ArchivedDataMessage,
crate::protocol::DataMessage,
>(payload)?;
self.handle_local_data(header, message)
}
RouteDecision::Child(index) => {
Ok(EndpointOutcome::forward(RouteDecision::Child(index), frame))
@@ -130,8 +147,12 @@ impl Endpoint for ProtocolEndpoint {
PacketType::Fault => {
match self.decide_route(&header.dst_path) {
RouteDecision::Local => {
let message = parsed.deserialize_fault()?;
self.handle_local_fault(parsed.deserialize_header(), message)
let (header, payload) = parsed.into_parts();
let message = deserialize_archived_bytes::<
ArchivedFaultMessage,
crate::protocol::FaultMessage,
>(payload)?;
self.handle_local_fault(header, message)
}
RouteDecision::Child(index) => {
Ok(EndpointOutcome::forward(RouteDecision::Child(index), frame))
src/protocol/tree/hook.rs
+122 -26
View File
@@ -30,13 +30,18 @@ pub struct ActiveHook {
pub peer_path: Vec<String>,
pub procedure_id: String,
pub dst_leaf: Option<String>,
pub local_ended: bool,
pub peer_ended: bool,
}
/// Peer-scoped index key used by the non-host side of a hook.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct PeerHookKey {
hook_id: u64,
peer_path: Vec<String>,
/// Pending hook context used only for fault attribution before activation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingHook {
pub return_path: Vec<String>,
pub hook_id: u64,
pub caller_src_path: Vec<String>,
pub procedure_id: String,
pub dst_leaf: Option<String>,
}
/// Duplicate hook insertion error.
@@ -46,14 +51,16 @@ pub struct HookConflict;
/// Durable hook state tables.
#[derive(Debug)]
pub struct HookTable {
active: BTreeMap<HookKey, ActiveHook>,
active_by_peer: BTreeMap<PeerHookKey, HookKey>,
pending: BTreeMap<u64, BTreeMap<Vec<String>, PendingHook>>,
active: BTreeMap<u64, BTreeMap<Vec<String>, ActiveHook>>,
active_by_peer: BTreeMap<u64, BTreeMap<Vec<String>, Vec<String>>>,
next_id: u64,
}
impl Default for HookTable {
fn default() -> Self {
Self {
pending: BTreeMap::new(),
active: BTreeMap::new(),
active_by_peer: BTreeMap::new(),
next_id: 1,
@@ -73,35 +80,101 @@ impl HookTable {
id
}
/// Inserts a pending hook created by a received call.
pub fn insert_pending(&mut self, pending: PendingHook) -> Result<(), HookConflict> {
if self.pending(&HookKey::new(pending.return_path.clone(), pending.hook_id)).is_some()
|| self.active(&HookKey::new(pending.return_path.clone(), pending.hook_id)).is_some()
{
return Err(HookConflict);
}
self.pending
.entry(pending.hook_id)
.or_default()
.insert(pending.return_path.clone(), pending);
Ok(())
}
/// Inserts an already-active hook flow.
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.active.contains_key(&key) || self.active_by_peer.contains_key(&peer_key) {
if self.pending(&key).is_some() || self.active(&key).is_some() {
return Err(HookConflict);
}
self.active_by_peer.insert(peer_key, key.clone());
self.active.insert(key, active);
self.active_by_peer
.entry(active.hook_id)
.or_default()
.insert(active.peer_path.clone(), active.return_path.clone());
self.active
.entry(active.hook_id)
.or_default()
.insert(active.return_path.clone(), active);
Ok(())
}
/// Promotes one pending hook into active state after local acceptance.
pub fn activate_pending(&mut self, key: &HookKey) -> Option<()> {
let pending = self.remove_pending(key)?;
self.insert_active(ActiveHook {
return_path: pending.return_path,
hook_id: pending.hook_id,
peer_path: pending.caller_src_path,
procedure_id: pending.procedure_id,
dst_leaf: pending.dst_leaf,
local_ended: false,
peer_ended: false,
})
.ok()?;
Some(())
}
/// Removes a pending hook entry.
pub fn remove_pending(&mut self, key: &HookKey) -> Option<PendingHook> {
let hooks = self.pending.get_mut(&key.hook_id)?;
let pending = hooks.remove(key.return_path.as_slice())?;
if hooks.is_empty() {
self.pending.remove(&key.hook_id);
}
Some(pending)
}
/// Removes an active hook entry.
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(),
});
let hooks = self.active.get_mut(&key.hook_id)?;
let active = hooks.remove(key.return_path.as_slice())?;
if hooks.is_empty() {
self.active.remove(&key.hook_id);
}
if let Some(peer_index) = self.active_by_peer.get_mut(&key.hook_id) {
peer_index.remove(active.peer_path.as_slice());
if peer_index.is_empty() {
self.active_by_peer.remove(&key.hook_id);
}
}
Some(active)
}
/// Returns a pending hook by its host-scoped key.
#[must_use]
pub fn pending(&self, key: &HookKey) -> Option<&PendingHook> {
self.pending
.get(&key.hook_id)?
.get(key.return_path.as_slice())
}
/// Returns an active hook by its host-scoped key.
#[must_use]
pub fn active(&self, key: &HookKey) -> Option<&ActiveHook> {
self.active.get(key)
self.active.get(&key.hook_id)?.get(key.return_path.as_slice())
}
/// Returns mutable access to an active hook by its host-scoped key.
pub fn active_mut(&mut self, key: &HookKey) -> Option<&mut ActiveHook> {
self.active
.get_mut(&key.hook_id)?
.get_mut(key.return_path.as_slice())
}
/// Resolves one active hook key from either the host side or the peer side.
@@ -113,21 +186,44 @@ impl HookTable {
peer_path: &[String],
) -> Option<HookKey> {
let host_key = HookKey::new(return_path.to_vec(), hook_id);
if self.active.contains_key(&host_key) {
if self.active(&host_key).is_some() {
return Some(host_key);
}
self.active_by_peer
.get(&PeerHookKey {
hook_id,
peer_path: peer_path.to_vec(),
})
.get(&hook_id)?
.get(peer_path)
.cloned()
.map(|return_path| HookKey::new(return_path, hook_id))
}
/// Marks one locally-originated final data packet.
pub fn mark_local_end(&mut self, key: &HookKey) -> bool {
let Some(active) = self.active_mut(key) else {
return false;
};
active.local_ended = true;
active.peer_ended
}
/// Marks one peer-originated final data packet.
pub fn mark_peer_end(&mut self, key: &HookKey) -> bool {
let Some(active) = self.active_mut(key) else {
return false;
};
active.peer_ended = true;
active.local_ended
}
/// Returns whether one key still has pending or active state.
#[must_use]
pub fn contains(&self, key: &HookKey) -> bool {
self.pending(key).is_some() || self.active(key).is_some()
}
/// Returns the number of active hooks.
#[must_use]
pub fn active_len(&self) -> usize {
self.active.len()
self.active.values().map(BTreeMap::len).sum()
}
}
src/protocol/tree/mod.rs
+3 -3
View File
@@ -8,8 +8,8 @@ pub use endpoint::{
ChildRoute, ConnectionState, Endpoint, EndpointError, EndpointOutcome, Ingress, LeafSpec,
LocalEvent, ProtocolEndpoint,
};
pub use hook::{ActiveHook, HookConflict, HookKey, HookTable};
pub use hook::{ActiveHook, HookConflict, HookKey, HookTable, PendingHook};
pub use routing::{
DefaultRouteProvider, LeafNode, RouteDecision, RouteProvider, TreeNode, is_prefix,
route_destination,
CompiledRoutes, DefaultRouteProvider, LeafNode, RouteDecision, RouteProvider, TreeNode,
is_prefix, route_destination,
};
src/protocol/tree/routing.rs
+106 -1
View File
@@ -1,6 +1,6 @@
//! Path routing helpers and explicit enum tree declarations.
use alloc::{string::String, vec::Vec};
use alloc::{collections::BTreeMap, string::String, vec, vec::Vec};
/// Explicit test tree declaration used for configuration.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -67,6 +67,93 @@ pub enum RouteDecision {
Drop,
}
/// One compiled routing table for one endpoint boundary.
#[derive(Debug, Clone, Default)]
pub struct CompiledRoutes {
local_path: Vec<String>,
has_parent: bool,
nodes: Vec<RouteTrieNode>,
}
#[derive(Debug, Clone, Default)]
struct RouteTrieNode {
best_child: Option<usize>,
edges: BTreeMap<String, usize>,
}
impl CompiledRoutes {
/// Compiles the registered-child prefixes into a trie once.
#[must_use]
pub fn new(local_path: &[String], child_paths: &[Vec<String>], has_parent: bool) -> Self {
let mut table = Self {
local_path: local_path.to_vec(),
has_parent,
nodes: vec![RouteTrieNode::default()],
};
for (index, child_path) in child_paths.iter().enumerate() {
table.insert_child(index, child_path);
}
table
}
fn insert_child(&mut self, index: usize, child_path: &[String]) {
if !is_prefix(&self.local_path, child_path) || child_path.len() <= self.local_path.len() {
return;
}
let mut node_index = 0usize;
for segment in &child_path[self.local_path.len()..] {
let next_index = if let Some(next_index) = self.nodes[node_index].edges.get(segment) {
*next_index
} else {
let next_index = self.nodes.len();
self.nodes.push(RouteTrieNode::default());
self.nodes[node_index]
.edges
.insert(segment.clone(), next_index);
next_index
};
node_index = next_index;
}
self.nodes[node_index].best_child = Some(index);
}
/// Resolves one destination path using one segment walk.
#[must_use]
pub fn route(&self, dst_path: &[String]) -> RouteDecision {
if !is_prefix(&self.local_path, dst_path) {
return if self.has_parent {
RouteDecision::Parent
} else {
RouteDecision::Drop
};
}
let mut best_child = None;
let mut node_index = 0usize;
for segment in &dst_path[self.local_path.len()..] {
let Some(next_index) = self.nodes[node_index].edges.get(segment) else {
break;
};
node_index = *next_index;
if let Some(index) = self.nodes[node_index].best_child {
best_child = Some(index);
}
}
if let Some(index) = best_child {
return RouteDecision::Child(index);
}
if self.local_path == dst_path {
return RouteDecision::Local;
}
RouteDecision::Drop
}
}
/// Returns `true` if `prefix` is a path prefix of `path`.
pub fn is_prefix(prefix: &[String], path: &[String]) -> bool {
prefix.len() <= path.len()
@@ -165,6 +252,24 @@ mod tests {
);
}
#[test]
fn compiled_routes_choose_longest_prefix_without_child_scan() {
let table = CompiledRoutes::new(
&[String::from("a")],
&[
vec![String::from("a"), String::from("b")],
vec![String::from("a"), String::from("x")],
],
true,
);
assert_eq!(
table.route(&[String::from("a"), String::from("b"), String::from("c")]),
RouteDecision::Child(0)
);
assert_eq!(table.route(&[String::from("z")]), RouteDecision::Parent);
}
#[test]
fn tree_enum_flattens_paths() {
let tree = TreeNode::Root {