diff --git a/src/transport/async.rs b/src/transport/async.rs index bc622e4c..4b080d68 100644 --- a/src/transport/async.rs +++ b/src/transport/async.rs @@ -853,46 +853,28 @@ impl AsyncTcpMessageBus { 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 AsyncTcpMessageBus { } } 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 AsyncTcpMessageBus { 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( + /// 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, - id: &K, - execution_id: Option, + order_id: Option, + request_id: Option, 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 diff --git a/src/transport/async/registry.rs b/src/transport/async/registry.rs index 4f5206e2..f71e5d9b 100644 --- a/src/transport/async/registry.rs +++ b/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,6 +136,7 @@ impl SenderHash { /// 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(&self, id: &K, f: impl FnOnce(&Route) -> R) -> Option { self.read().get(id).map(f) } @@ -138,13 +144,44 @@ impl SenderHash { /// 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( + &self, + id: &K, + item: RoutedItem, + alias: Option<&A>, + aliases: &SenderHash, + ) -> 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 SenderHash { 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> { + 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()); } diff --git a/src/transport/async/registry_tests.rs b/src/transport/async/registry_tests.rs index 1d39205c..bc7f49e8 100644 --- a/src/transport/async/registry_tests.rs +++ b/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::::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::::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::::new(); + let aliases = SenderHash::::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(); diff --git a/src/transport/sync.rs b/src/transport/sync.rs index 81715c32..26239d0d 100644 --- a/src/transport/sync.rs +++ b/src/transport/sync.rs @@ -5,7 +5,7 @@ use std::collections::{hash_map, HashMap, HashSet}; use std::io::prelude::*; use std::net::TcpStream; -use std::sync::{Arc, Mutex, OnceLock, PoisonError, RwLock}; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard}; use std::thread::{self, JoinHandle}; use std::time::Duration; @@ -122,7 +122,7 @@ impl SharedChannels { } } - fn subscribers(&self) -> std::sync::MutexGuard<'_, Vec> { + fn subscribers(&self) -> MutexGuard<'_, Vec> { self.subscribers.lock().unwrap_or_else(PoisonError::into_inner) } @@ -301,6 +301,12 @@ impl NoticeBroadcaster { } } +/// Lock the order-update slot, recovering from poisoning: the slot holds no +/// invariant a panic could break. +fn lock_slot(slot: &Mutex>>) -> MutexGuard<'_, Option>> { + slot.lock().unwrap_or_else(PoisonError::into_inner) +} + #[derive(Debug)] pub struct TcpMessageBus { connection: Connection, @@ -379,7 +385,7 @@ impl TcpMessageBus { // After the flag: `create_order_update_subscription` checks it under // the same lock, so no stream can register once this slot is emptied. // The subscription holds a sender clone, so only a sent item ends it. - if let Some(entry) = self.order_update_stream.lock().unwrap().take() { + if let Some(entry) = lock_slot(&self.order_update_stream).take() { let _ = entry.sender.send(Error::Shutdown.into()); } @@ -468,13 +474,7 @@ impl TcpMessageBus { } fn clear_order_update_stream(&self, lease: &LeaseRef) { - let mut stream = if let Ok(stream) = self.order_update_stream.lock() { - stream - } else { - warn!("failed to lock order_update_stream"); - return; - }; - + let mut stream = lock_slot(&self.order_update_stream); let removed = stream.as_ref().is_some_and(|registered| registered.lease.is(lease)); if removed { *stream = None; @@ -636,11 +636,11 @@ impl TcpMessageBus { fn process_response_with_id(&self, id: Option, message: ResponseMessage, routed: bool) { match id { Some(WireId::Request(request_id)) if self.requests.contains(&request_id) => { - self.requests.send(&request_id, message.into()).unwrap(); + let _ = self.requests.deliver(&request_id, message.into()); return; } Some(WireId::Order(order_id)) if self.orders.contains(&order_id) => { - self.orders.send(&order_id, message.into()).unwrap(); + let _ = self.orders.deliver(&order_id, message.into()); return; } _ => {} @@ -655,20 +655,21 @@ impl TcpMessageBus { /// Deliver a pre-classified Notice or Error to the request or order /// subscription its id names. fn deliver(&self, id: WireId, item: RoutedItem, sent_to_update_stream: bool) { - match id { - WireId::Request(request_id) if self.requests.contains(&request_id) => { - let _ = self.requests.send(&request_id, item); - } - WireId::Order(order_id) if self.orders.contains(&order_id) => { - let _ = self.orders.send(&order_id, item); + let unrouted = match id { + WireId::Request(request_id) => self.requests.deliver(&request_id, item), + WireId::Order(order_id) => self.orders.deliver(&order_id, item), + }; + if let Err(item) = unrouted { + if !sent_to_update_stream { + log_orphan(id, &item); } - _ if sent_to_update_stream => {} - _ => log_orphan(id, &item), } } fn process_orders(&self, message: ResponseMessage) { let strategy = order_routing_strategy(message.message_type()); + let message_order_id = message.order_id().map(OrderId::from); + let message_request_id = message.request_id().and_then(RequestId::from_raw); match strategy { OrderRoutingStrategy::OrderUpdateOnly => { @@ -676,63 +677,28 @@ impl TcpMessageBus { } OrderRoutingStrategy::ExecutionData => { let sent_to_update_stream = self.send_order_update(&message); - - // Try order_id channel first, then request_id, storing execution_id mapping - if let Some(order_id) = message.order_id().map(OrderId::from) { - if self.orders.contains(&order_id) { - self.store_execution_mapping(&message, &self.orders, order_id); - if let Err(e) = self.orders.send(&order_id, message.into()) { - warn!("error routing message for order_id({order_id}): {e}"); - } - return; + let execution_id = message.execution_id(); + if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), execution_id.as_ref()) { + if !sent_to_update_stream { + warn!("could not route message {item:?}"); } } - if let Some(request_id) = message.request_id().and_then(RequestId::from_raw) { - if self.requests.contains(&request_id) { - self.store_execution_mapping(&message, &self.requests, request_id); - if let Err(e) = self.requests.send(&request_id, message.into()) { - warn!("error routing message for request_id({request_id}): {e}"); - } - return; - } - } - if !sent_to_update_stream { - warn!("could not route message {message:?}"); - } } OrderRoutingStrategy::ExecutionDataEnd => { - if let Some(order_id) = message.order_id().map(OrderId::from) { - if self.orders.contains(&order_id) { - if let Err(e) = self.orders.send(&order_id, message.into()) { - warn!("error routing message for order_id({order_id}): {e}"); - } - return; - } - } - if let Some(request_id) = message.request_id().and_then(RequestId::from_raw) { - if self.requests.contains(&request_id) { - if let Err(e) = self.requests.send(&request_id, message.into()) { - warn!("error routing message for request_id({request_id}): {e}"); - } - return; - } + if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), None) { + warn!("could not route message {item:?}"); } - warn!("could not route message {message:?}"); } OrderRoutingStrategy::OrderOrShared => { let sent_to_update_stream = self.send_order_update(&message); - if let Some(order_id) = message.order_id().map(OrderId::from) { + if let Some(order_id) = message_order_id { if self.orders.contains(&order_id) { - if let Err(e) = self.orders.send(&order_id, message.into()) { - warn!("error routing message for order_id({order_id}): {e}"); - } - return; - } - if self.shared_channels.is_shared_response(IncomingMessages::OpenOrder) { + let _ = self.orders.deliver(&order_id, message.into()); + } else { self.shared_channels.send_message(message.message_type(), &message); - return; } + return; } if !sent_to_update_stream { warn!("could not route message {message:?}"); @@ -741,12 +707,14 @@ impl TcpMessageBus { OrderRoutingStrategy::ByExecutionId => { let sent_to_update_stream = self.send_order_update(&message); - if let Some(execution_id) = message.execution_id() { - if let Err(e) = self.executions.send(&execution_id, message.into()) { - warn!("error sending commission report for execution {execution_id}: {e}"); + 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 !sent_to_update_stream { + warn!("could not route commission report {item:?}"); } - } else if !sent_to_update_stream { - warn!("could not route commission report {message:?}"); } } OrderRoutingStrategy::SharedOnly => { @@ -792,20 +760,24 @@ impl TcpMessageBus { Ok(subscription) } - /// Alias the execution id to `id`'s channel in `channels`. The insert runs - /// under `channels`' read lock, so a concurrent cleanup either removes the - /// registration first (no alias is stored) or prunes the alias after it. - fn store_execution_mapping( + /// 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, - message: &ResponseMessage, - channels: &SenderHash, - id: K, - ) { - if let Some(execution_id) = message.execution_id() { - channels.with_entry(&id, |entry| { - self.executions.insert(execution_id, entry.sender.clone(), entry.lease.clone()) - }); + order_id: Option, + request_id: Option, + item: RoutedItem, + execution_id: Option<&String>, + ) -> Result<(), RoutedItem> { + 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), + }) } // Sends an order update message to the order update stream if it exists. @@ -815,17 +787,16 @@ impl TcpMessageBus { } fn send_order_update_item(&self, item: RoutedItem) -> bool { - if let Ok(order_update_stream) = self.order_update_stream.lock() { - if let Some(entry) = order_update_stream.as_ref() { - if let Err(e) = entry.sender.send(item) { - warn!("error sending to order update stream: {e}"); - return false; - } - warn_if_backlogged(format_args!("order update stream"), entry.sender.len()); - return true; - } + let order_update_stream = lock_slot(&self.order_update_stream); + let Some(entry) = order_update_stream.as_ref() else { + return false; + }; + if let Err(e) = entry.sender.send(item) { + warn!("error sending to order update stream: {e}"); + return false; } - false + warn_if_backlogged(format_args!("order update stream"), entry.sender.len()); + true } // The cleanup thread receives signals as subscribers are cancelled or @@ -961,7 +932,7 @@ impl MessageBus for TcpMessageBus { } fn create_order_update_subscription(&self) -> Result { - let mut order_update_stream = self.order_update_stream.lock().unwrap(); + let mut order_update_stream = lock_slot(&self.order_update_stream); // Not `ensure_connected`: nothing is written, and the stream may be // created while a reconnect is in progress. @@ -1043,6 +1014,34 @@ impl Entry { fn closed(&self) -> bool { self.bound.as_ref().is_some_and(BoundState::closed) } + + /// An unbounded registration sharing this one's channel and lease. + fn alias(&self) -> Self { + Self::new(self.sender.clone(), self.lease.clone()) + } +} + +impl Entry { + /// Send `item`, subject to a bounded route's cap ([`BoundState::admit`]). + /// Never blocks. + fn deliver(&self, id: &impl std::fmt::Debug, item: RoutedItem) { + let item = match &self.bound { + Some(bound) => match bound.admit(&item, self.sender.len()) { + Admit::Deliver => item, + Admit::Overflow => bound.overflow_error(), + Admit::Discard => { + trace!("discarding item for closed route {id:?}"); + return; + } + }, + None => item, + }; + if let Err(err) = self.sender.send(item) { + warn!("error sending: {id:?}, {err}") + } else { + warn_if_backlogged(format_args!("subscription queue for {id:?}"), self.sender.len()); + } + } } #[derive(Debug)] @@ -1057,6 +1056,17 @@ impl SenderHash RwLockReadGuard<'_, HashMap>> { + self.senders.read().unwrap_or_else(PoisonError::into_inner) + } + + fn write(&self) -> RwLockWriteGuard<'_, HashMap>> { + self.senders.write().unwrap_or_else(PoisonError::into_inner) + } + #[cfg(test)] pub fn copy_sender(&self, id: K) -> Option> { self.with_entry(&id, |entry| entry.sender.clone()) @@ -1069,13 +1079,14 @@ impl SenderHash(&self, id: &K, f: impl FnOnce(&Entry) -> R) -> Option { - let senders = self.senders.read().unwrap(); + let senders = self.read(); senders.get(id).map(f) } pub fn insert(&self, id: K, sender: Sender, lease: LeaseRef) { - self.senders.write().unwrap().insert(id, Entry::new(sender, lease)); + self.insert_entry(id, Entry::new(sender, lease)); } /// Like [`insert`](Self::insert), with an unread-item cap: see [`BoundState::admit`]. @@ -1084,14 +1095,18 @@ impl SenderHash) { + self.write().insert(id, entry); } /// Remove the entry for `id` only if it holds `lease`. Returns whether an /// entry was removed. Used by signal cleanup so a stale signal cannot /// remove a newer registration under the same key. pub fn remove_if_same(&self, id: K, lease: &LeaseRef) -> bool { - match self.senders.write().unwrap().entry(id) { + match self.write().entry(id) { hash_map::Entry::Occupied(registered) if registered.get().lease.is(lease) => { registered.remove(); true @@ -1102,54 +1117,64 @@ impl SenderHash usize { - let mut senders = self.senders.write().unwrap(); + let mut senders = self.write(); let before = senders.len(); senders.retain(|_, registered| !registered.lease.is(lease)); before - senders.len() } pub fn contains(&self, id: &K) -> bool { - let senders = self.senders.read().unwrap(); + let senders = self.read(); senders.contains_key(id) } pub fn len(&self) -> usize { - let senders = self.senders.read().unwrap(); + let senders = self.read(); senders.len() } pub fn clear(&self) { - let mut senders = self.senders.write().unwrap(); + let mut senders = self.write(); senders.clear(); } } impl SenderHash { - /// Deliver `message` to `id`'s channel, subject to a bounded route's cap - /// ([`BoundState::admit`]). Never blocks. - pub fn send(&self, id: &K, message: RoutedItem) -> Result<(), Error> { - let senders = self.senders.read().unwrap(); - debug!("senders: {senders:?}"); + /// Deliver `item` to `id`'s route; hands it back when nothing is + /// registered under `id`. + pub fn deliver(&self, id: &K, item: RoutedItem) -> Result<(), RoutedItem> { + 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 + /// cleanup either removes the registration first (no alias is stored) or + /// prunes the alias after it. 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 fn deliver_aliased( + &self, + id: &K, + item: RoutedItem, + alias: Option<&A>, + aliases: &SenderHash, + ) -> Result<(), RoutedItem> { + self.deliver_then(id, item, |entry| { + if let Some(alias) = alias { + aliases.insert_entry(alias.clone(), entry.alias()); + } + }) + } + + /// 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(&Entry)) -> Result<(), RoutedItem> { + let senders = self.read(); let Some(entry) = senders.get(id) else { - warn!("no recipient found for: {id:?}, {message:?}"); - return Ok(()); + return Err(item); }; - let message = match &entry.bound { - Some(bound) => match bound.admit(&message, entry.sender.len()) { - Admit::Deliver => message, - Admit::Overflow => bound.overflow_error(), - Admit::Discard => { - trace!("discarding item for closed route {id:?}"); - return Ok(()); - } - }, - None => message, - }; - if let Err(err) = entry.sender.send(message) { - warn!("error sending: {id:?}, {err}") - } else { - warn_if_backlogged(format_args!("subscription queue for {id:?}"), entry.sender.len()); - } + entry.deliver(id, item); + then(entry); Ok(()) } @@ -1160,7 +1185,7 @@ impl SenderHash { where F: Fn() -> RoutedItem, { - let mut senders = self.senders.write().unwrap(); + let mut senders = self.write(); for entry in senders.values().filter(|entry| !entry.closed()) { if let Err(e) = entry.sender.send(message_fn()) { warn!("error sending notification: {e}"); diff --git a/src/transport/sync_tests.rs b/src/transport/sync_tests.rs index 27f017b3..76e1f7ed 100644 --- a/src/transport/sync_tests.rs +++ b/src/transport/sync_tests.rs @@ -3473,3 +3473,121 @@ fn test_bounded_request_end_marker_at_limit_still_ends() -> Result<(), Error> { assert!(sub.try_next().is_none(), "frames after the end marker are discarded"); Ok(()) } + +/// A request route, its receiver, and the lease that keeps it live. +fn sender_hash_route() -> (SenderHash, Receiver, Lease) { + let routes = SenderHash::new(); + let (sender, receiver) = channel::unbounded(); + let lease = Lease::new(); + routes.insert(RequestId::nth(1), sender, lease.downgrade()); + (routes, receiver, lease) +} + +#[test] +fn sender_hash_deliver_hands_the_item_back_when_unrouted() { + let routes = SenderHash::::new(); + let item = routes.deliver(&RequestId::nth(1), RoutedItem::Error(Error::Cancelled)); + assert!(matches!(item, Err(RoutedItem::Error(Error::Cancelled)))); +} + +#[test] +fn sender_hash_deliver_aliased_aliases_the_route() { + let (routes, receiver, lease) = sender_hash_route(); + let aliases = SenderHash::::new(); + 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: Vec<_> = receiver.try_iter().collect(); + assert!( + items.len() == 2 && matches!(items[1], RoutedItem::Error(Error::ConnectionReset)), + "{items:?}" + ); + assert!(aliases.lease(alias).unwrap().is(&lease.downgrade()), "alias holds another lease"); +} + +#[test] +fn sender_hash_deliver_aliased_without_alias_registers_none() { + let (routes, receiver, _lease) = sender_hash_route(); + let aliases = SenderHash::::new(); + + routes + .deliver_aliased(&RequestId::nth(1), RoutedItem::Error(Error::Cancelled), None, &aliases) + .unwrap(); + + assert_eq!(receiver.try_iter().count(), 1); + assert_eq!(aliases.len(), 0); +} + +#[test] +fn sender_hash_deliver_aliased_hands_the_item_back_when_unrouted() { + let routes = SenderHash::::new(); + let aliases = SenderHash::::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_eq!(aliases.len(), 0, "unrouted item was aliased"); +} + +/// A panic under the route lock must not take every later route and +/// teardown down with it. +#[test] +fn sender_hash_recovers_from_a_poisoned_lock() { + let (routes, receiver, _lease) = sender_hash_route(); + std::thread::scope(|scope| { + let _ = scope + .spawn(|| { + let _guard = routes.senders.write().unwrap(); + panic!("poison the route lock"); + }) + .join(); + }); + assert!(routes.senders.is_poisoned()); + + routes.deliver(&RequestId::nth(1), RoutedItem::Error(Error::Cancelled)).unwrap(); + routes.clear(); + + assert_eq!(receiver.try_iter().count(), 1); + assert_eq!(routes.len(), 0); +} + +/// A panic under the order-update slot's lock must not silently drop later +/// order updates or leave the slot uncleanable. +#[test] +fn test_order_update_stream_survives_a_poisoned_lock() -> Result<(), Error> { + let (stream, bus) = make_bus(); + let stream_sub = bus.create_order_update_subscription()?; + std::thread::scope(|scope| { + let _ = scope + .spawn(|| { + let _guard = bus.order_update_stream.lock().unwrap(); + panic!("poison the order-update slot"); + }) + .join(); + }); + assert!(bus.order_update_stream.is_poisoned()); + + stream.push_inbound(binary_proto( + crate::messages::IncomingMessages::OpenOrder as i32, + &crate::proto::OpenOrder { + order_id: Some(42), + ..Default::default() + }, + )); + bus.dispatch()?; + assert!(stream_sub.next_timeout(TICK).is_some(), "update stream missed open order"); + + let registered = lock_slot(&bus.order_update_stream).as_ref().unwrap().lease.clone(); + bus.clear_order_update_stream(®istered); + assert!(lock_slot(&bus.order_update_stream).is_none(), "poisoned slot not cleared"); + Ok(()) +}