Check banned users on order arrival

crates/autopilot/src/infra/persistence/mod.rs

@@ -8,7 +8,7 @@ use {
         },
     },
     ::winner_selection::state::RankedItem,
-    alloy::primitives::B256,
+    alloy::primitives::{Address, B256},
     anyhow::Context,
     bigdecimal::{BigDecimal, ToPrimitive},
     boundary::database::byte_array::ByteArray,
@@ -99,10 +99,59 @@ impl Persistence {
         LeaderLock::new(self.postgres.pool.clone(), key, Duration::from_millis(200))
     }
 
+    /// Spawns a long running task that looks up the banned status of every
+    /// address pushed into the returned queue, warming the shared cache.
+    /// Lookups are batched: a batch is flushed once it reaches `BATCH_SIZE`
+    /// addresses or `BATCH_DELAY` after its first address, whichever comes
+    /// first.
+    fn spawn_banned_prefetch_task(
+        banned_users: Arc<crate::infra::banned::Users>,
+    ) -> mpsc::Sender<Address> {
+        const BATCH_SIZE: usize = 50;
+        const BATCH_DELAY: Duration = Duration::from_secs(1);
+        // A few batches of headroom; prefetching is best effort so there is
+        // no need for backpressure — dropping an address on overflow just
+        // moves the lookup back to the auction cut.
+        const QUEUE_SIZE: usize = BATCH_SIZE * 4;
+
+        let (sender, mut receiver) = mpsc::channel(QUEUE_SIZE);
+        tokio::spawn(async move {
+            while let Some(address) = receiver.recv().await {
+                let mut batch = HashSet::from([address]);
+                let deadline = tokio::time::sleep(BATCH_DELAY);
+                tokio::pin!(deadline);
+                loop {
+                    tokio::select! {
+                        () = &mut deadline => break,
+                        next = receiver.recv() => {
+                            let Some(address) = next else { break };
+                            batch.insert(address);
+                            if batch.len() >= BATCH_SIZE {
+                                break;
+                            }
+                        }
+                    }
+                }
+                banned_users.banned(batch).await;
+            }
+            tracing::error!("banned users prefetch task terminated unexpectedly");
+        });
+        sender
+    }
+
     /// Spawns a background task that listens for new order notifications from
     /// PostgreSQL and notifies via the provided Notify.
-    pub fn spawn_order_listener(&self, notify: Arc<tokio::sync::Notify>) {
+    ///
+    /// Additionally warms the banned users cache for new orders' owners
+    /// (batched, best effort) so the auction cut doesn't pay for the remote
+    /// lookup on its critical path.
+    pub fn spawn_order_listener(
+        &self,
+        notify: Arc<tokio::sync::Notify>,
+        banned_users: Arc<crate::infra::banned::Users>,
+    ) {
         let pool = self.postgres.pool.clone();
+        let prefetch_queue = Self::spawn_banned_prefetch_task(banned_users);
         tokio::spawn(async move {
             loop {
                 let mut listener = match sqlx::postgres::PgListener::connect_with(&pool).await {
@@ -127,6 +176,25 @@ impl Persistence {
                         Ok(notification) => {
                             let order_uid = notification.payload();
                             tracing::debug!(order_uid, "received order notification from postgres");
+                            // Best effort: the receiver (distinct from the
+                            // owner in ~3% of orders) is not derivable from
+                            // the payload and remains a cut-time lookup
+                            match order_uid_from_notification(order_uid) {
+                                Some(uid) => {
+                                    if let Err(err) = prefetch_queue.try_send(uid.owner()) {
+                                        tracing::debug!(
+                                            ?err,
+                                            "failed to enqueue banned status prefetch"
+                                        );
+                                    }
+                                }
+                                None => {
+                                    tracing::warn!(
+                                        order_uid,
+                                        "malformed order notification payload"
+                                    )
+                                }
+                            }
                             notify.notify_one();
                         }
                         Err(err) => {
@@ -1034,6 +1102,13 @@ impl Persistence {
     }
 }
 
+/// Parses the payload of a `new_order` notification: the hex encoded order
+/// UID as emitted by the `order_insert_notify` database trigger.
+fn order_uid_from_notification(payload: &str) -> Option<domain::OrderUid> {
+    let bytes = alloy::hex::decode(payload).ok()?;
+    Some(domain::OrderUid(bytes.try_into().ok()?))
+}
+
 #[derive(prometheus_metric_storage::MetricStorage)]
 struct Metrics {
     /// Timing of db queries.

crates/autopilot/src/run.rs

@@ -449,23 +449,25 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
         config.price_estimation.max_quote_timeout,
     ));
 
+    let banned_users = Arc::new(infra::banned::Users::new(
+        eth.contracts().chainalysis_oracle().clone(),
+        config
+            .banned_users
+            .hermod
+            .clone()
+            .map(|hermod| infra::banned::HermodConfig {
+                url: hermod.url,
+                hmac_key: hermod.hmac_key,
+                api_key: hermod.api_key,
+            }),
+        config.banned_users.addresses,
+        config.banned_users.max_cache_size.get().to_u64().unwrap(),
+    ));
+
     let solvable_orders_cache = SolvableOrdersCache::new(
         config.min_order_validity_period,
         persistence.clone(),
-        infra::banned::Users::new(
-            eth.contracts().chainalysis_oracle().clone(),
-            config
-                .banned_users
-                .hermod
-                .clone()
-                .map(|hermod| infra::banned::HermodConfig {
-                    url: hermod.url,
-                    hmac_key: hermod.hmac_key,
-                    api_key: hermod.api_key,
-                }),
-            config.banned_users.addresses,
-            config.banned_users.max_cache_size.get().to_u64().unwrap(),
-        ),
+        banned_users.clone(),
         balance_fetcher.clone(),
         deny_listed_tokens.clone(),
         competition_native_price_updater.clone(),
@@ -626,6 +628,7 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
             startup,
         },
         awaiter,
+        banned_users,
     );
     run.run_forever(shutdown_controller).await;
 

crates/autopilot/src/run_loop.rs

@@ -129,6 +129,7 @@ impl RunLoop {
         trusted_tokens: AutoUpdatingTokenList,
         probes: Probes,
         maintenance: MaintenanceSync,
+        banned_users: Arc<infra::banned::Users>,
     ) -> Self {
         let max_winners = config.max_winners_per_auction.get();
         let weth = eth.contracts().wrapped_native_token();
@@ -137,7 +138,7 @@ impl RunLoop {
         let wake_notify = Arc::new(tokio::sync::Notify::new());
 
         // Spawn background tasks to listen for events
-        persistence.spawn_order_listener(wake_notify.clone());
+        persistence.spawn_order_listener(wake_notify.clone(), banned_users);
         Self::spawn_block_listener(eth.current_block().clone(), wake_notify.clone());
 
         Self {

crates/autopilot/src/solvable_orders.rs

@@ -131,7 +131,7 @@ impl Metrics {
 pub struct SolvableOrdersCache {
     min_order_validity_period: Duration,
     persistence: infra::Persistence,
-    banned_users: banned::Users,
+    banned_users: Arc<banned::Users>,
     balance_fetcher: Arc<dyn BalanceFetching>,
     deny_listed_tokens: DenyListedTokens,
     cache: Mutex<Option<Inner>>,
@@ -157,7 +157,7 @@ impl SolvableOrdersCache {
     pub fn new(
         min_order_validity_period: Duration,
         persistence: infra::Persistence,
-        banned_users: banned::Users,
+        banned_users: Arc<banned::Users>,
         balance_fetcher: Arc<dyn BalanceFetching>,
         deny_listed_tokens: DenyListedTokens,
         native_price_estimator: Arc<NativePriceUpdater>,

crates/order-validation/src/banned/cached.rs

@@ -25,6 +25,16 @@ struct Entry {
     last_updated: Instant,
 }
 
+impl Entry {
+    /// Creates a new [`Entry`] with `last_updated` set to [`Instant::now`]-
+    fn new(is_banned: bool) -> Self {
+        Self {
+            is_banned,
+            last_updated: Instant::now(),
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub(super) enum BackendError {
     #[error("chainalysis lookup failed")]
@@ -134,13 +144,7 @@ impl Cached {
     /// positive confirmation and at least one backend failed.
     async fn refresh(&self, address: Address) -> Option<(Address, Entry)> {
         let is_banned = self.fetch_all(address).await?;
-        Some((
-            address,
-            Entry {
-                is_banned,
-                last_updated: Instant::now(),
-            },
-        ))
+        Some((address, Entry::new(is_banned)))
     }
 
     /// Spawns a background task that periodically refreshes near-expiry cache