diff --git a/crates/autopilot/src/domain/mod.rs b/crates/autopilot/src/domain/mod.rs index ed1eb17793..26930605ea 100644 --- a/crates/autopilot/src/domain/mod.rs +++ b/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; diff --git a/crates/autopilot/src/domain/order_notify.rs b/crates/autopilot/src/domain/order_notify.rs new file mode 100644 index 0000000000..57e6ebfd11 --- /dev/null +++ b/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); + +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 { + 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::()].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)))); + } +} diff --git a/crates/autopilot/src/infra/banned.rs b/crates/autopilot/src/infra/banned.rs new file mode 100644 index 0000000000..9e672c3a14 --- /dev/null +++ b/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) { + 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"); + }); +} diff --git a/crates/autopilot/src/infra/mod.rs b/crates/autopilot/src/infra/mod.rs index 4d34b214b7..4e29254682 100644 --- a/crates/autopilot/src/infra/mod.rs +++ b/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}; diff --git a/crates/autopilot/src/infra/persistence/mod.rs b/crates/autopilot/src/infra/persistence/mod.rs index 953e5c4b34..d2489f5f57 100644 --- a/crates/autopilot/src/infra/persistence/mod.rs +++ b/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) { + /// 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, + 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 { + 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. diff --git a/crates/autopilot/src/run.rs b/crates/autopilot/src/run.rs index 9b0d8dfe5e..86ac0107c2 100644 --- a/crates/autopilot/src/run.rs +++ b/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; diff --git a/crates/autopilot/src/run_loop.rs b/crates/autopilot/src/run_loop.rs index 4063673ff8..16fd6a2943 100644 --- a/crates/autopilot/src/run_loop.rs +++ b/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 { diff --git a/crates/autopilot/src/solvable_orders.rs b/crates/autopilot/src/solvable_orders.rs index ad6e33573f..e764b21d98 100644 --- a/crates/autopilot/src/solvable_orders.rs +++ b/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, balance_fetcher: Arc, deny_listed_tokens: DenyListedTokens, cache: Mutex>, @@ -157,7 +157,7 @@ impl SolvableOrdersCache { pub fn new( min_order_validity_period: Duration, persistence: infra::Persistence, - banned_users: banned::Users, + banned_users: Arc, balance_fetcher: Arc, deny_listed_tokens: DenyListedTokens, native_price_estimator: Arc, diff --git a/crates/order-validation/src/banned/cached.rs b/crates/order-validation/src/banned/cached.rs index 2ac6e66dda..2323161c9e 100644 --- a/crates/order-validation/src/banned/cached.rs +++ b/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 { + 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
) -> HashSet
{ @@ -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 diff --git a/crates/order-validation/src/banned/mod.rs b/crates/order-validation/src/banned/mod.rs index b9754c5d0c..3732c313db 100644 --- a/crates/order-validation/src/banned/mod.rs +++ b/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 { + 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) -> HashSet
{