transport: deliver_aliased + collapsed order/request fallback (#930)

src/transport/async.rs

@@ -853,46 +853,28 @@ impl<S: AsyncStream> AsyncTcpMessageBus<S> {
         match strategy {
             OrderRoutingStrategy::OrderUpdateOnly => {}
             OrderRoutingStrategy::ExecutionData => {
-                // Try order_id channel first, then request_id, storing execution_id mapping
                 let execution_id = message.execution_id();
-                let mut item = RoutedItem::from(message);
-                if let Some(order_id) = message_order_id {
-                    item = match self.deliver_execution(&self.orders, &order_id, execution_id.clone(), item) {
-                        Ok(()) => return Ok(()),
-                        Err(item) => item,
-                    };
-                }
-                if let Some(request_id) = message_request_id {
-                    item = match self.deliver_execution(&self.requests, &request_id, execution_id, item) {
-                        Ok(()) => return Ok(()),
-                        Err(item) => item,
-                    };
-                }
-                if !routed {
-                    warn!("could not route ExecutionData message {item:?}");
+                if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), execution_id.as_ref()) {
+                    if !routed {
+                        warn!("could not route ExecutionData message {item:?}");
+                    }
                 }
             }
             OrderRoutingStrategy::ExecutionDataEnd => {
-                let mut item = RoutedItem::from(message);
-                if let Some(order_id) = message_order_id {
-                    item = match self.orders.deliver(&order_id, item) {
-                        Ok(()) => return Ok(()),
-                        Err(item) => item,
-                    };
-                }
-                if let Some(request_id) = message_request_id {
-                    item = match self.requests.deliver(&request_id, item) {
-                        Ok(()) => return Ok(()),
-                        Err(item) => item,
-                    };
+                if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), None) {
+                    warn!("could not route ExecutionDataEnd message {item:?}");
                 }
-                warn!("could not route ExecutionDataEnd message {item:?}");
             }
             OrderRoutingStrategy::OrderOrShared => {
                 if let Some(order_id) = message_order_id {
-                    if self.orders.deliver(&order_id, message.clone().into()).is_ok()
-                        || self.shared_channels.send_message(message.message_type(), &message)
-                    {
+                    // `contains` first, not `deliver`'s hand-back: the shared
+                    // fallback needs the message, and a second lookup is
+                    // cheaper than cloning every frame.
+                    if self.orders.contains(&order_id) {
+                        let _ = self.orders.deliver(&order_id, message.into());
+                        return Ok(());
+                    }
+                    if self.shared_channels.send_message(message.message_type(), &message) {
                         return Ok(());
                     }
                 }
@@ -901,8 +883,14 @@ impl<S: AsyncStream> AsyncTcpMessageBus<S> {
                 }
             }
             OrderRoutingStrategy::ByExecutionId => {
-                if let Some(execution_id) = message.execution_id() {
-                    let _ = self.executions.deliver(&execution_id, message.into());
+                let unrouted = match message.execution_id() {
+                    Some(execution_id) => self.executions.deliver(&execution_id, message.into()),
+                    None => Err(message.into()),
+                };
+                if let Err(item) = unrouted {
+                    if !routed {
+                        warn!("could not route commission report {item:?}");
+                    }
                 }
             }
             OrderRoutingStrategy::SharedOnly => {
@@ -927,32 +915,24 @@ impl<S: AsyncStream> AsyncTcpMessageBus<S> {
         Ok(())
     }
 
-    /// Deliver an execution to `id`'s route in `routes` and alias the route
-    /// under `execution_id`, for the commission report that follows. Hands
-    /// the item back when nothing is registered under `id`.
-    fn deliver_execution<K: std::hash::Hash + Eq + std::fmt::Display + std::fmt::Debug>(
+    /// Deliver to `order_id`'s route, else `request_id`'s, aliasing the route
+    /// under `execution_id` for the commission report that follows. Hands
+    /// the item back when neither is registered.
+    fn deliver_to_order_or_request(
         &self,
-        routes: &SenderHash<K>,
-        id: &K,
-        execution_id: Option<String>,
+        order_id: Option<OrderId>,
+        request_id: Option<RequestId>,
         item: RoutedItem,
+        execution_id: Option<&String>,
     ) -> Result<(), RoutedItem> {
-        let mut item = Some(item);
-        let delivered = routes.with_route(id, |route| {
-            if let Some(item) = item.take() {
-                route.deliver(id, item);
-            }
-            (route.sender.clone(), route.lease.clone())
-        });
-        match (delivered, item) {
-            (Some((sender, lease)), _) => {
-                if let Some(execution_id) = execution_id {
-                    self.executions.insert(execution_id, Route::unbounded(sender, lease));
-                }
-                Ok(())
-            }
-            (None, item) => Err(item.expect("undelivered item")),
+        match order_id {
+            Some(id) => self.orders.deliver_aliased(&id, item, execution_id, &self.executions),
+            None => Err(item),
         }
+        .or_else(|item| match request_id {
+            Some(id) => self.requests.deliver_aliased(&id, item, execution_id, &self.executions),
+            None => Err(item),
+        })
     }
 
     /// Register a channel of `class` under `id` in `routes`, optionally with

src/transport/async/registry.rs

@@ -76,6 +76,11 @@ impl Route {
         self.bound.as_ref().is_some_and(|bound| bound.state.closed())
     }
 
+    /// An unbounded route sharing this one's channel and lease.
+    fn alias(&self) -> Self {
+        Self::unbounded(self.sender.clone(), self.lease.clone())
+    }
+
     /// Whether this route holds `lease` and every holder of it is gone.
     pub(super) fn released(&self, lease: &LeaseRef) -> bool {
         self.lease.is(lease) && !self.lease.is_live()
@@ -131,20 +136,52 @@ impl<K: Hash + Eq + Display + Debug> SenderHash<K> {
 
     /// Run `f` on `id`'s route while holding the read lock, so no removal can
     /// land between the lookup and `f`.
+    #[cfg(test)]
     pub(super) fn with_route<R>(&self, id: &K, f: impl FnOnce(&Route) -> R) -> Option<R> {
         self.read().get(id).map(f)
     }
 
     /// Deliver `item` to `id`'s route; hands it back when nothing is
     /// registered under `id`.
     pub(super) fn deliver(&self, id: &K, item: RoutedItem) -> Result<(), RoutedItem> {
-        match self.read().get(id) {
-            Some(route) => {
-                route.deliver(id, item);
-                Ok(())
+        self.deliver_then(id, item, |_| {})
+    }
+
+    /// Deliver `item` to `id`'s route, then alias that route under `alias`
+    /// in `aliases` while still holding `id`'s read lock, so a concurrent
+    /// clear can't land between the delivery and the alias. Lock order:
+    /// `self` (read), then `aliases` (write); nothing takes them the other
+    /// way round. Hands the item back when nothing is registered under `id`.
+    pub(super) fn deliver_aliased<A: Hash + Eq + Clone + Display + Debug>(
+        &self,
+        id: &K,
+        item: RoutedItem,
+        alias: Option<&A>,
+        aliases: &SenderHash<A>,
+    ) -> Result<(), RoutedItem> {
+        self.deliver_then(id, item, |route| {
+            if let Some(alias) = alias {
+                aliases.insert(alias.clone(), route.alias());
             }
-            None => Err(item),
-        }
+        })
+    }
+
+    /// Deliver `item` to `id`'s route and run `then` on it, both under the
+    /// read lock.
+    fn deliver_then(&self, id: &K, item: RoutedItem, then: impl FnOnce(&Route)) -> Result<(), RoutedItem> {
+        let routes = self.read();
+        let Some(route) = routes.get(id) else {
+            return Err(item);
+        };
+        route.deliver(id, item);
+        then(route);
+        Ok(())
+    }
+
+    /// Whether a route is registered under `id`. For a caller whose miss
+    /// path still needs the message `deliver` would consume.
+    pub(super) fn contains(&self, id: &K) -> bool {
+        self.read().contains_key(id)
     }
 
     /// Remove `id`'s registration if it matches `pred`; returns whether it
@@ -212,11 +249,6 @@ impl<K: Hash + Eq + Display + Debug> SenderHash<K> {
         self.with_route(id, |route| route.sender.clone())
     }
 
-    #[cfg(test)]
-    pub(super) fn contains(&self, id: &K) -> bool {
-        self.read().contains_key(id)
-    }
-
     #[cfg(test)]
     pub(super) fn len(&self) -> usize {
         self.read().len()
@@ -319,6 +351,10 @@ impl SharedChannels {
         self.channels.read().unwrap_or_else(PoisonError::into_inner)
     }
 
+    fn channels_mut(&self) -> RwLockWriteGuard<'_, HashMap<OutgoingMessages, SharedChannel>> {
+        self.channels.write().unwrap_or_else(PoisonError::into_inner)
+    }
+
     /// Every live shared subscription has just been failed: start a new
     /// generation so their later drops cannot touch the next session's counts.
     pub(super) async fn reset_counts(&self) {
@@ -362,8 +398,12 @@ impl SharedChannels {
 
     /// Deliver `item()` once to every channel, then drop the senders so every
     /// subscription ends after it. Later sends are no-ops.
+    ///
+    /// The counterpart of sync `SharedChannels::fail_all` (notify and clear
+    /// under one lock); named `close` because async channels persist across
+    /// subscriptions, and a closed registry refuses later subscribes.
     pub(super) fn close(&self, item: impl Fn() -> RoutedItem) {
-        let mut channels = self.channels.write().unwrap_or_else(PoisonError::into_inner);
+        let mut channels = self.channels_mut();
         for sender in channels.values().filter_map(|channel| channel.sender.get()) {
             let _ = sender.send(item());
         }

src/transport/async/registry_tests.rs

@@ -60,6 +60,58 @@ fn deliver_hands_the_item_back_when_unrouted() {
     assert!(matches!(item, Err(RoutedItem::Error(Error::Cancelled))));
 }
 
+#[test]
+fn deliver_aliased_aliases_the_route() {
+    let routes = SenderHash::new();
+    let aliases = SenderHash::<String>::new();
+    let (route, mut receiver, lease) = route();
+    routes.insert(RequestId::nth(1), route);
+    let alias = "exec-1".to_string();
+
+    routes
+        .deliver_aliased(&RequestId::nth(1), RoutedItem::Error(Error::Cancelled), Some(&alias), &aliases)
+        .unwrap();
+    aliases.deliver(&alias, Error::ConnectionReset.into()).unwrap();
+
+    let items = items(&mut receiver);
+    assert!(items.len() == 2 && is_reset(&items[1]), "{items:?}");
+    assert!(
+        aliases.with_route(&alias, |route| route.lease.is(&lease.downgrade())).unwrap(),
+        "alias holds another lease"
+    );
+}
+
+#[test]
+fn deliver_aliased_without_alias_registers_none() {
+    let routes = SenderHash::new();
+    let aliases = SenderHash::<String>::new();
+    let (route, mut receiver, _lease) = route();
+    routes.insert(RequestId::nth(1), route);
+
+    routes
+        .deliver_aliased(&RequestId::nth(1), RoutedItem::Error(Error::Cancelled), None, &aliases)
+        .unwrap();
+
+    assert_eq!(items(&mut receiver).len(), 1);
+    assert!(aliases.is_empty());
+}
+
+#[test]
+fn deliver_aliased_hands_the_item_back_when_unrouted() {
+    let routes = SenderHash::<RequestId>::new();
+    let aliases = SenderHash::<String>::new();
+
+    let item = routes.deliver_aliased(
+        &RequestId::nth(1),
+        RoutedItem::Error(Error::Cancelled),
+        Some(&"exec-1".to_string()),
+        &aliases,
+    );
+
+    assert!(matches!(item, Err(RoutedItem::Error(Error::Cancelled))));
+    assert!(aliases.is_empty(), "unrouted item was aliased");
+}
+
 #[test]
 fn remove_if_same_spares_a_replacement() {
     let routes = SenderHash::new();