(fast6) prepare order-notification pipeline in runloop - #4861
Conversation
MartinquaXD
changed the title
|
Claude finished @MartinquaXD's task in 1m 33s —— View job PR Review: prepare order-notification pipeline in runloop
The refactor is clean and behaviourally equivalent to before: order notifications now flow through an unbounded One robustness issue, posted inline:
Nothing else stood out — no blocking concerns. |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| async fn on_new_order(&self, _: OrderUid) { | ||
| self.0.notify_one(); | ||
| async fn on_new_order(&self, order: OrderUid) { | ||
| self.0.unbounded_send(order).unwrap() |
There was a problem hiding this comment.
The .unwrap() on unbounded_send is a landmine that contradicts the best-effort contract documented on the Listener trait ("Notifications are best effort ... whatever a listener does must be recoverable at auction cut time").
unbounded_send only errors when the receiver has been dropped. If the run-loop's spawn_order_listener task ever ends (e.g. it panics), the receiver drops and the next order notification panics here. That panic propagates out of dispatch's join_all and kills the entire Notifier task — which has no panic recovery, so all order notifications stop permanently, including the unrelated CachePrewarmer for banned users.
The old notify_one() could never fail. Consider handling the send error gracefully instead:
| self.0.unbounded_send(order).unwrap() | |
| if self.0.unbounded_send(order).is_err() { | |
| tracing::warn!("run loop order listener is gone; dropping notification"); | |
| } | |
| } |
(the closing } in the suggestion replaces line 14's } — adjust if you keep the original brace layout)
There was a problem hiding this comment.
I understand why the current unwrap is there, maybe halfway would be an expect with the explanatio
MartinquaXD
changed the title
Replaces the plain `Arc<tokio::sync::Notify>` that the DB notifier used to wake the run loop with an `mpsc::UnboundedSender<OrderUid>`. The run loop constructs its own wake `Notify` internally and spawns a background task (`spawn_order_listener`) that pulls order uids off the channel and notifies the run loop. For now the listener just wakes the loop and drops the uid; a later PR will hand the uid to the fast-path handler. Also switches `RunLoop::new` to return `Arc<Self>` and `run_forever` to take `self: Arc<Self>` so the listener can hold a strong reference. Signed-off-by: MartinquaXD <martin@cow.fi>
e97797e to
ffaa4b7
Compare
| infra::order_notify::Notifier::new(banned_users.clone(), wake_runloop.clone()) | ||
| // New-order notifications from the DB fan out through this channel to | ||
| // the run loop (which wakes) and, later on, to the fast-path handler. | ||
| let (new_orders_sender, new_orders_receiver) = mpsc::unbounded(); |
There was a problem hiding this comment.
This unbounded makes me nervous, why not running bounded and applying backpressure?
2 participants
Description
The current order notification handling was very simple: listener sees new order -> sends notify to the run loop waker.
With the introduction of the fast path handling in the run loop it's not sufficient to know that an order was created, we also want to know which order AND we have to buffer them in a channel in case many of them get placed at once.
Changes
autopilot now spawns a background task that awaits order notifications and notifies the waker.
Kicking off the actual fast path handling happens in a follow up PR.
How to test
effectively no behavioural change