Skip to content

(fast6) prepare order-notification pipeline in runloop #4861

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MartinquaXD wants to merge 1 commit into from
Open

(fast6) prepare order-notification pipeline in runloop #4861

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions crates/autopilot/src/infra/order_notify/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ mod run_loop;
use {
self::run_loop::RunLoopWaker,
crate::{domain::OrderUid, infra::order_notify::banned::CachePrewarmer},
futures::future::join_all,
futures::{channel::mpsc, future::join_all},
order_validation::banned::Users,
sqlx::PgPool,
std::{sync::Arc, time::Duration},
Expand All @@ -26,10 +26,13 @@ pub struct Notifier {
}

impl Notifier {
pub fn new(banned_users: Arc<Users>, run_loop_wake: Arc<tokio::sync::Notify>) -> Self {
pub fn new(
banned_users: Arc<Users>,
run_loop_new_order_listener: mpsc::UnboundedSender<OrderUid>,
) -> Self {
Self {
listeners: vec![
Box::new(RunLoopWaker(run_loop_wake)),
Box::new(RunLoopWaker(run_loop_new_order_listener)),
Box::new(CachePrewarmer(banned_users)),
],
}
Expand Down
8 changes: 4 additions & 4 deletions crates/autopilot/src/infra/order_notify/run_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
use {
crate::{domain::OrderUid, infra::order_notify::Listener},
std::sync::Arc,
futures::channel::mpsc,
};

/// "Wakes" up (i.e. notifies) the run-loop to start when a new block or order
/// appears.
pub struct RunLoopWaker(pub Arc<tokio::sync::Notify>);
pub struct RunLoopWaker(pub mpsc::UnboundedSender<OrderUid>);

#[async_trait::async_trait]
impl Listener for RunLoopWaker {
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()

@claude claude Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

Suggested change
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)

@jmg-duarte jmg-duarte Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand why the current unwrap is there, maybe halfway would be an expect with the explanatio

}
}
10 changes: 6 additions & 4 deletions crates/autopilot/src/run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ use {
contracts::{GPv2Settlement, WETH9},
ethrpc::{Web3, block_stream::block_number_to_block_number_hash},
event_indexing::block_retriever::BlockRetriever,
futures::channel::mpsc,
http_client::HttpClientFactory,
model::DomainSeparator,
num::ToPrimitive,
Expand Down Expand Up @@ -471,9 +472,10 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
config.banned_users.max_cache_size.get().to_u64().unwrap(),
));

// Wakes the run loop on new orders (via the notifier) and new blocks.
let wake_runloop = Arc::new(tokio::sync::Notify::new());
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();

@jmg-duarte jmg-duarte Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This unbounded makes me nervous, why not running bounded and applying backpressure?

infra::order_notify::Notifier::new(banned_users.clone(), new_orders_sender)
.spawn(db_write.pool.clone());

let penalty_cap_calculator = match &config.penalty_cap {
Expand Down Expand Up @@ -659,7 +661,7 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
startup,
},
awaiter,
wake_runloop,
new_orders_receiver,
);
run.run_forever(shutdown_controller).await;

Expand Down
28 changes: 21 additions & 7 deletions crates/autopilot/src/run_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ use {
crate::{
domain::{
self,
OrderUid,
auction::Id,
competition::{
self,
Expand All @@ -31,7 +32,7 @@ use {
database::order_events::OrderEventLabel,
eth_domain_types::Address,
ethrpc::block_stream::{BlockInfo, CurrentBlockWatcher},
futures::{StreamExt, TryFutureExt},
futures::{StreamExt, TryFutureExt, channel::mpsc},
itertools::Itertools,
num::ToPrimitive,
rand::seq::SliceRandom,
Expand Down Expand Up @@ -184,11 +185,12 @@ impl RunLoop {
trusted_tokens: AutoUpdatingTokenList,
probes: Probes,
maintenance: MaintenanceSync,
wake_runloop: Arc<tokio::sync::Notify>,
) -> Self {
new_orders_listener: mpsc::UnboundedReceiver<OrderUid>,
) -> Arc<Self> {
let max_winners = config.max_winners_per_auction.get();
let weth = eth.contracts().wrapped_native_token();

let wake_runloop = Arc::new(tokio::sync::Notify::new());
Self::spawn_block_listener(eth.current_block().clone(), wake_runloop.clone());

let settle_coordinator = Arc::new(SettleCallCoordinator::new(
Expand All @@ -198,7 +200,7 @@ impl RunLoop {
config.max_settlement_transaction_wait,
));

Self {
let self_ = Arc::new(Self {
delta_state: std::sync::Mutex::new(DeltaState::new(
config.auction_delta_checkpoint_interval,
)),
Expand All @@ -213,14 +215,26 @@ impl RunLoop {
wake_notify: wake_runloop,
drivers,
settle_coordinator,
}
});
Self::spawn_order_listener(self_.clone(), new_orders_listener);
self_
}

/// Spawns a background task that listens to the creation of new orders
/// and wakes the run loop for each incoming order.
fn spawn_order_listener(self: Arc<Self>, mut receiver: mpsc::UnboundedReceiver<OrderUid>) {
tokio::spawn(async move {
while let Some(_order_uid) = receiver.next().await {
self.wake_notify.notify_one();
}
});
}

pub async fn run_forever(self, mut control: ShutdownController) {
pub async fn run_forever(self: Arc<Self>, mut control: ShutdownController) {
let mut last_auction = None;
let mut last_block = None;

let self_arc = Arc::new(self);
let self_arc = self;
let leader = if self_arc.config.enable_leader_lock {
Some(
self_arc
Expand Down
Loading