Check banned users on order arrival

crates/autopilot/src/domain/mod.rs

@@ -2,6 +2,7 @@ pub mod auction;
 pub mod blockchain;
 pub mod competition;
 pub mod fee;
+pub mod order_notify;
 pub mod quote;
 pub mod settlement;
 

crates/autopilot/src/domain/order_notify.rs

@@ -0,0 +1,93 @@
+//! Fan-out of orders as they arrive in the orderbook.
+//!
+//! Producers (currently the Postgres order listener) publish orders as soon as
+//! they show up, consumers subscribe independently so work that would
+//! otherwise happen on the auction cut's critical path (warming the banned
+//! users cache, ...) can start early.
+
+use {crate::domain::OrderUid, tokio::sync::broadcast, tokio_stream::wrappers::BroadcastStream};
+
+/// How many arrivals a consumer may fall behind before it starts missing
+/// orders. Consumers are best effort, so lagging is not fatal. In mainnet we're
+/// (at the time of writing) running around ~1 order/second, so 16 should be
+/// plenty space for new orders coming in.
+const CAPACITY: usize = 16;
+
+/// Publishing end of the order arrival fan-out.
+#[derive(Clone)]
+pub struct Notifier(broadcast::Sender<OrderUid>);
+
+impl Notifier {
+    pub fn new() -> Self {
+        Self(broadcast::Sender::new(CAPACITY))
+    }
+
+    /// Publishes a newly arrived order. Arrivals published while nobody is
+    /// subscribed are dropped.
+    pub fn publish(&self, order: OrderUid) {
+        if let Err(_) = self.0.send(order) {
+            tracing::error!("failed to send order uid to subscribers");
+        }
+    }
+
+    /// Stream of the orders arriving from now on, ending once all publishers
+    /// are gone.
+    ///
+    /// A consumer that can't keep up is told how many orders it missed
+    /// ([`BroadcastStreamRecvError::Lagged`]) instead of slowing down the
+    /// publisher or its fellow consumers.
+    pub fn subscribe(&self) -> BroadcastStream<OrderUid> {
+        BroadcastStream::new(self.0.subscribe())
+    }
+}
+
+impl Default for Notifier {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use {
+        super::*,
+        tokio_stream::{StreamExt, wrappers::errors::BroadcastStreamRecvError},
+    };
+
+    fn order(n: usize) -> OrderUid {
+        let mut uid = [0u8; 56];
+        uid[..size_of::<usize>()].copy_from_slice(&n.to_le_bytes());
+        OrderUid(uid)
+    }
+
+    #[tokio::test]
+    async fn every_subscriber_sees_every_arrival() {
+        let arrivals = Notifier::new();
+        let (mut first, mut second) = (arrivals.subscribe(), arrivals.subscribe());
+
+        arrivals.publish(order(1));
+        arrivals.publish(order(2));
+
+        assert_eq!(first.next().await, Some(Ok(order(1))));
+        assert_eq!(first.next().await, Some(Ok(order(2))));
+        assert_eq!(second.next().await, Some(Ok(order(1))));
+        assert_eq!(second.next().await, Some(Ok(order(2))));
+    }
+
+    #[tokio::test]
+    async fn lagging_subscriber_is_told_what_it_missed() {
+        let arrivals = Notifier::new();
+        let mut subscription = arrivals.subscribe();
+
+        for n in 0..=CAPACITY {
+            arrivals.publish(order(n));
+        }
+
+        // The oldest arrival got dropped instead of stalling the publisher.
+        assert_eq!(
+            subscription.next().await,
+            Some(Err(BroadcastStreamRecvError::Lagged(1)))
+        );
+        assert_eq!(subscription.next().await, Some(Ok(order(1))));
+    }
+}

crates/autopilot/src/infra/banned.rs

@@ -0,0 +1,32 @@
+pub use order_validation::banned::*;
+use {
+    crate::domain::order_notify,
+    futures::StreamExt,
+    std::sync::Arc,
+    tokio_stream::wrappers::errors::BroadcastStreamRecvError,
+};
+
+/// Spawns a task that warms the cache with the owner of every arriving order
+/// so the auction cut doesn't pay for the remote lookup on its critical path.
+///
+/// Best effort: the receiver of an order is not part of its UID, so it remains
+/// a cut time lookup.
+pub fn spawn_cache_prewarming(arrivals: &order_notify::Notifier, users: Arc<Users>) {
+    let mut arrivals = arrivals.subscribe();
+    tokio::spawn(async move {
+        while let Some(arrival) = arrivals.next().await {
+            let owner = match arrival {
+                Ok(order) => order.owner(),
+                Err(BroadcastStreamRecvError::Lagged(skipped)) => {
+                    tracing::debug!(skipped, "lagged behind new orders, skipping prewarming");
+                    continue;
+                }
+            };
+            // Owners we already know about make up the bulk of the arrivals.
+            if users.cached(&owner).is_none() {
+                users.banned([owner]).await;
+            }
+        }
+        tracing::error!("banned users cache prewarming task terminated unexpectedly");
+    });
+}

crates/autopilot/src/infra/mod.rs

@@ -1,12 +1,8 @@
 pub mod api;
+pub mod banned;
 pub mod blockchain;
 pub mod persistence;
 pub mod shadow;
 pub mod solvers;
 
-pub use {
-    blockchain::Ethereum,
-    order_validation::banned,
-    persistence::Persistence,
-    solvers::Driver,
-};
+pub use {blockchain::Ethereum, persistence::Persistence, solvers::Driver};

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

@@ -4,6 +4,7 @@ use {
         database::{Postgres, order_events::store_order_events},
         domain::{
             self,
+            order_notify,
             settlement::{SettlementEvent, TradeEvent, transaction::EncodedTrade},
         },
     },
@@ -100,8 +101,13 @@ impl Persistence {
     }
 
     /// 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>) {
+    /// PostgreSQL, notifies via the provided Notify and publishes the arriving
+    /// orders so interested components can act on them right away.
+    pub fn spawn_order_listener(
+        &self,
+        notify: Arc<tokio::sync::Notify>,
+        new_orders: order_notify::Notifier,
+    ) {
         let pool = self.postgres.pool.clone();
         tokio::spawn(async move {
             loop {
@@ -127,6 +133,15 @@ impl Persistence {
                         Ok(notification) => {
                             let order_uid = notification.payload();
                             tracing::debug!(order_uid, "received order notification from postgres");
+                            match order_uid_from_notification(order_uid) {
+                                Some(uid) => new_orders.publish(uid),
+                                None => {
+                                    tracing::warn!(
+                                        order_uid,
+                                        "malformed order notification payload"
+                                    )
+                                }
+                            }
                             notify.notify_one();
                         }
                         Err(err) => {
@@ -1034,6 +1049,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

@@ -15,7 +15,7 @@ use {
                 event_retriever::CoWSwapOnchainOrdersContract,
             },
         },
-        domain,
+        domain::{self, order_notify},
         event_updater::EventUpdater,
         infra,
         maintenance::Maintenance,
@@ -449,23 +449,28 @@ 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 order_notifications = order_notify::Notifier::new();
+    infra::banned::spawn_cache_prewarming(&order_notifications, banned_users.clone());
+
     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 +631,7 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
             startup,
         },
         awaiter,
+        order_notifications,
     );
     run.run_forever(shutdown_controller).await;
 

crates/autopilot/src/run_loop.rs

@@ -10,6 +10,7 @@ use {
                 Unscored,
                 winner_selection::{self, Ranking},
             },
+            order_notify,
             settlement::{ExecutionEnded, ExecutionStarted},
         },
         infra::{
@@ -129,6 +130,7 @@ impl RunLoop {
         trusted_tokens: AutoUpdatingTokenList,
         probes: Probes,
         maintenance: MaintenanceSync,
+        order_notifier: order_notify::Notifier,
     ) -> Self {
         let max_winners = config.max_winners_per_auction.get();
         let weth = eth.contracts().wrapped_native_token();
@@ -137,7 +139,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(), order_notifier);
         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")]
@@ -63,6 +73,11 @@ impl Cached {
         Some(cached)
     }
 
+    /// Returns the cached ban status of `address`, never hitting a backend.
+    pub(super) fn cached(&self, address: &Address) -> Option<bool> {
+        self.cache.get(address).map(|entry| entry.is_banned)
+    }
+
     /// Returns the subset reported as banned by any backend. Misses fan out
     /// to backends concurrently.
     pub(super) async fn check(&self, addresses: &HashSet<Address>) -> HashSet<Address> {
@@ -134,13 +149,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

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

@@ -58,6 +58,22 @@ impl Users {
         Self { list, remote: None }
     }
 
+    /// Returns the ban status of `address` if it is already known locally,
+    /// `None` if determining it requires a remote lookup.
+    pub fn cached(&self, address: &Address) -> Option<bool> {
+        if self.list.contains(address) {
+            return Some(true);
+        }
+        // The zero/burn address is never checked remotely, see [`Self::banned`].
+        if address.is_zero() {
+            return Some(false);
+        }
+        match &self.remote {
+            Some(remote) => remote.cached(address),
+            None => Some(false),
+        }
+    }
+
     /// Returns the subset of `addresses` that are banned. Cache misses hit
     /// the configured remote sources.
     pub async fn banned(&self, addresses: impl IntoIterator<Item = Address>) -> HashSet<Address> {