From ffaa4b74ee444b8828b9bcd0b3dcf77f9c4f845f Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Thu, 3 Sep 2026 10:23:11 +0000 Subject: [PATCH] Order-notification pipeline into the run loop Replaces the plain `Arc` that the DB notifier used to wake the run loop with an `mpsc::UnboundedSender`. 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` and `run_forever` to take `self: Arc` so the listener can hold a strong reference. Signed-off-by: MartinquaXD --- .../autopilot/src/infra/order_notify/mod.rs | 9 ++++-- .../src/infra/order_notify/run_loop.rs | 8 +++--- crates/autopilot/src/run.rs | 10 ++++--- crates/autopilot/src/run_loop.rs | 28 ++++++++++++++----- 4 files changed, 37 insertions(+), 18 deletions(-) diff --git a/crates/autopilot/src/infra/order_notify/mod.rs b/crates/autopilot/src/infra/order_notify/mod.rs index 101b1bd028..cfbd39f37e 100644 --- a/crates/autopilot/src/infra/order_notify/mod.rs +++ b/crates/autopilot/src/infra/order_notify/mod.rs @@ -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}, @@ -26,10 +26,13 @@ pub struct Notifier { } impl Notifier { - pub fn new(banned_users: Arc, run_loop_wake: Arc) -> Self { + pub fn new( + banned_users: Arc, + run_loop_new_order_listener: mpsc::UnboundedSender, + ) -> Self { Self { listeners: vec![ - Box::new(RunLoopWaker(run_loop_wake)), + Box::new(RunLoopWaker(run_loop_new_order_listener)), Box::new(CachePrewarmer(banned_users)), ], } diff --git a/crates/autopilot/src/infra/order_notify/run_loop.rs b/crates/autopilot/src/infra/order_notify/run_loop.rs index 17ce0f692b..a6a8ab8c68 100644 --- a/crates/autopilot/src/infra/order_notify/run_loop.rs +++ b/crates/autopilot/src/infra/order_notify/run_loop.rs @@ -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); +pub struct RunLoopWaker(pub mpsc::UnboundedSender); #[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() } } diff --git a/crates/autopilot/src/run.rs b/crates/autopilot/src/run.rs index 0f0959adfc..89ec31308e 100644 --- a/crates/autopilot/src/run.rs +++ b/crates/autopilot/src/run.rs @@ -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, @@ -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(); + infra::order_notify::Notifier::new(banned_users.clone(), new_orders_sender) .spawn(db_write.pool.clone()); let penalty_cap_calculator = match &config.penalty_cap { @@ -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; diff --git a/crates/autopilot/src/run_loop.rs b/crates/autopilot/src/run_loop.rs index 1e18d98a7d..2c7b688afb 100644 --- a/crates/autopilot/src/run_loop.rs +++ b/crates/autopilot/src/run_loop.rs @@ -5,6 +5,7 @@ use { crate::{ domain::{ self, + OrderUid, auction::Id, competition::{ self, @@ -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, @@ -184,11 +185,12 @@ impl RunLoop { trusted_tokens: AutoUpdatingTokenList, probes: Probes, maintenance: MaintenanceSync, - wake_runloop: Arc, - ) -> Self { + new_orders_listener: mpsc::UnboundedReceiver, + ) -> Arc { 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( @@ -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, )), @@ -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, mut receiver: mpsc::UnboundedReceiver) { + 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, 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