From 5c1b8740f5bbc6efb3743b77820f11a22bd428fc Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Wed, 19 Aug 2026 11:13:47 +0000 Subject: [PATCH 01/12] PR08: seam implementations and a single auction cycle --- Cargo.lock | 5 + crates/autopilot-svm/Cargo.toml | 6 + crates/autopilot-svm/src/domain/arbitrator.rs | 81 ++++++++ crates/autopilot-svm/src/domain/auction.rs | 13 +- crates/autopilot-svm/src/domain/cycle.rs | 64 ++++++ crates/autopilot-svm/src/domain/mod.rs | 4 +- crates/autopilot-svm/src/infra/competition.rs | 140 ++++++++++++++ crates/autopilot-svm/src/infra/db.rs | 6 +- crates/autopilot-svm/src/infra/driver/mod.rs | 2 - crates/autopilot-svm/src/infra/executor.rs | 70 +++++++ crates/autopilot-svm/src/infra/mod.rs | 7 +- crates/autopilot-svm/src/infra/observer.rs | 60 ++++++ crates/autopilot-svm/src/infra/provider.rs | 70 +++++++ crates/autopilot-svm/src/infra/trigger.rs | 54 ++++++ crates/autopilot-svm/src/lib.rs | 5 +- crates/autopilot-svm/src/tests.rs | 182 ++++++++++++++++++ crates/cow-solana-rpc/Cargo.toml | 1 + crates/cow-solana-rpc/src/lib.rs | 10 +- 18 files changed, 766 insertions(+), 14 deletions(-) create mode 100644 crates/autopilot-svm/src/domain/arbitrator.rs create mode 100644 crates/autopilot-svm/src/domain/cycle.rs create mode 100644 crates/autopilot-svm/src/infra/competition.rs create mode 100644 crates/autopilot-svm/src/infra/executor.rs create mode 100644 crates/autopilot-svm/src/infra/observer.rs create mode 100644 crates/autopilot-svm/src/infra/provider.rs create mode 100644 crates/autopilot-svm/src/infra/trigger.rs create mode 100644 crates/autopilot-svm/src/tests.rs diff --git a/Cargo.lock b/Cargo.lock index 8b71bc9402..b197897896 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1589,10 +1589,13 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "axum", "bigdecimal", "chain-types", "const-hex", + "cow-solana-rpc", "database", + "futures", "reqwest 0.13.4", "serde", "serde_json", @@ -1602,6 +1605,7 @@ dependencies = [ "tokio", "tracing", "url", + "winner-selection", ] [[package]] @@ -3961,6 +3965,7 @@ version = "0.1.0" dependencies = [ "solana-commitment-config", "solana-rpc-client", + "solana-rpc-client-api", "url", ] diff --git a/crates/autopilot-svm/Cargo.toml b/crates/autopilot-svm/Cargo.toml index 93e1364e6a..0e30dd042c 100644 --- a/crates/autopilot-svm/Cargo.toml +++ b/crates/autopilot-svm/Cargo.toml @@ -10,7 +10,9 @@ async-trait = { workspace = true } bigdecimal = { workspace = true } chain-types = { workspace = true } const-hex = { workspace = true } +cow-solana-rpc = { workspace = true } database = { workspace = true } +futures = { workspace = true } reqwest = { workspace = true, features = ["json"] } serde = { workspace = true } serde_json = { workspace = true } @@ -20,6 +22,10 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } tracing = { workspace = true } url = { workspace = true } +winner-selection = { workspace = true } + +[dev-dependencies] +axum = { workspace = true } [lints] workspace = true diff --git a/crates/autopilot-svm/src/domain/arbitrator.rs b/crates/autopilot-svm/src/domain/arbitrator.rs new file mode 100644 index 0000000000..61c691337f --- /dev/null +++ b/crates/autopilot-svm/src/domain/arbitrator.rs @@ -0,0 +1,81 @@ +//! Winner selection over the chain-generic arbitrator. + +use { + crate::{ + domain::{ + auction::Auction, + cycle::{Ranking, SolanaCycle, Solution}, + }, + run_loop::WinnerSelection, + }, + chain_types::{ + ChainTypes, + solana::{Pubkey, Solana}, + }, + winner_selection::{Arbitrator, AuctionContext}, +}; + +/// Runs the generic fair-combinatorial arbitrator over the drivers' +/// solutions. +pub struct SolanaArbitrator { + inner: Arbitrator, +} + +impl SolanaArbitrator { + pub fn new(max_winners: usize, wrapped_native: Pubkey) -> Self { + Self { + inner: Arbitrator { + max_winners, + wrapped_native, + }, + } + } +} + +impl WinnerSelection for SolanaArbitrator { + fn arbitrate(&self, solutions: Vec, auction: &Auction) -> Ranking { + let drivers = solutions + .iter() + .map(|solution| { + ( + (solution.inner.solver(), solution.inner.id()), + solution.driver_index, + ) + }) + .collect(); + + // An empty fee-policy list marks an order as part of the auction, the + // arbitrator only scores such orders. The auction carries no protocol + // fees. + // TODO: real fee policies arrive with the MVP protocol-fee support. + let fee_policies = auction + .orders + .iter() + .map(|order| (order.uid, Vec::new())) + .collect(); + // Every auction token is priced at the native denominator, i.e. 1:1 + // to lamports, so scores compare raw surplus. Ranking within one + // token pair is exact, comparisons across pairs are not. + // TODO: replace with native price estimation. + let native_prices = auction + .orders + .iter() + .flat_map(|order| [order.sell_token, order.buy_token]) + .map(|token| (token, Solana::NATIVE_PRICE_DENOMINATOR)) + .collect(); + let context = AuctionContext:: { + fee_policies, + surplus_capturing_jit_order_owners: Default::default(), + native_prices, + }; + + let inner = self.inner.arbitrate( + solutions + .into_iter() + .map(|solution| solution.inner) + .collect(), + &context, + ); + Ranking { inner, drivers } + } +} diff --git a/crates/autopilot-svm/src/domain/auction.rs b/crates/autopilot-svm/src/domain/auction.rs index 6d94142c23..1703201090 100644 --- a/crates/autopilot-svm/src/domain/auction.rs +++ b/crates/autopilot-svm/src/domain/auction.rs @@ -39,6 +39,9 @@ pub struct Auction { /// Autopilot-assigned id. Excluded from equality: the dedupe compares two /// cuts by content, and the id is allocated only for a fresh cut. pub id: i64, + /// Slot the auction was cut at. Excluded from equality for the same + /// reason: the tip moves every cycle, the content may not. + pub tip: u64, pub orders: Vec, } @@ -76,18 +79,24 @@ mod tests { } #[test] - fn auction_equality_ignores_the_id() { + fn auction_equality_ignores_id_and_tip() { let orders = vec![order(10)]; let a = Auction { id: 1, + tip: 100, orders: orders.clone(), }; - let b = Auction { id: 2, orders }; + let b = Auction { + id: 2, + tip: 200, + orders, + }; assert_eq!(a, b); assert_ne!( a, Auction { id: 1, + tip: 100, orders: vec![], } ); diff --git a/crates/autopilot-svm/src/domain/cycle.rs b/crates/autopilot-svm/src/domain/cycle.rs new file mode 100644 index 0000000000..da16632e68 --- /dev/null +++ b/crates/autopilot-svm/src/domain/cycle.rs @@ -0,0 +1,64 @@ +//! The Solana instantiation of the auction loop's type vocabulary. + +use { + crate::{ + domain::auction::Auction, + run_loop::{Cycle, RankingInfo}, + }, + chain_types::solana::{IntentHash, Pubkey, Solana}, + std::collections::{HashMap, HashSet}, + winner_selection::{Unscored, solution}, +}; + +/// Marker type binding the generic loop to the Solana vocabulary. +pub struct SolanaCycle; + +impl Cycle for SolanaCycle { + type Auction = Auction; + type OrderUid = IntentHash; + type Ranking = Ranking; + type Solution = Solution; + /// The chain progress marker is the slot. + type Tip = u64; +} + +/// One driver's solution, attributed to the driver that proposed it so the +/// executor can dispatch the settlement back to it. +pub struct Solution { + /// Index into the configured driver list. + pub driver_index: usize, + pub inner: solution::Solution, +} + +/// Key attributing a ranked solution back to its driver. The arbitrator +/// keeps `(solver, solution id)` through ranking, the driver index does not +/// survive it. +pub type SolutionKey = (Pubkey, u64); + +/// Arbitrated solutions plus the driver attribution the generic ranking +/// drops. +pub struct Ranking { + pub inner: winner_selection::Ranking, + /// Driver index per solution, keyed by `(solver, solution id)`. + pub drivers: HashMap, +} + +impl RankingInfo for Ranking { + fn winner_count(&self) -> usize { + self.inner.winners().count() + } + + fn winning_order_uids(&self) -> HashSet { + self.inner + .winners() + .flat_map(|solution| solution.orders().iter().map(|order| order.uid)) + .collect() + } + + fn considered_order_uids(&self) -> HashSet { + self.inner + .non_winners() + .flat_map(|solution| solution.orders().iter().map(|order| order.uid)) + .collect() + } +} diff --git a/crates/autopilot-svm/src/domain/mod.rs b/crates/autopilot-svm/src/domain/mod.rs index 24f3a1be78..056b7af2c3 100644 --- a/crates/autopilot-svm/src/domain/mod.rs +++ b/crates/autopilot-svm/src/domain/mod.rs @@ -1,3 +1,5 @@ -//! Domain types: the auction and its orders. +//! Domain types: the auction, its orders, and the cycle vocabulary. +pub mod arbitrator; pub mod auction; +pub mod cycle; diff --git a/crates/autopilot-svm/src/infra/competition.rs b/crates/autopilot-svm/src/infra/competition.rs new file mode 100644 index 0000000000..7cf74f3b93 --- /dev/null +++ b/crates/autopilot-svm/src/infra/competition.rs @@ -0,0 +1,140 @@ +//! Fans the auction out to the configured drivers. + +use { + crate::{ + domain::{ + auction::{Auction, Order, OrderKind}, + cycle::{SolanaCycle, Solution}, + }, + infra::driver::{Driver, dto}, + run_loop::SolverCompetition, + }, + async_trait::async_trait, + chain_types::solana::{IntentHash, Solana}, + std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, + }, + winner_selection::{Side, solution}, +}; + +/// Slots between the tip a solve request is stamped at and its deadline, +/// the driver's whole thinking budget (~6s at 400ms slots, the EVM fast +/// chains run 6s solve deadlines). +/// TODO: make configurable. +const SOLVE_DEADLINE_SLOTS: u64 = 15; + +/// Solana's target slot time. +const SLOT_DURATION: Duration = Duration::from_millis(400); + +/// Sends `/solve` to every driver and converts the answers into attributable +/// solutions for the arbitrator. +pub struct DriverCompetition { + drivers: Vec>, +} + +impl DriverCompetition { + pub fn new(drivers: Vec>) -> Self { + Self { drivers } + } +} + +#[async_trait] +impl SolverCompetition for DriverCompetition { + async fn solve(&self, auction: &Auction) -> Vec { + let request = &dto::SolveRequest { + id: auction.id, + deadline_slot: auction.tip + SOLVE_DEADLINE_SLOTS, + orders: auction.orders.iter().map(dto::Order::from).collect(), + }; + let by_uid: HashMap = auction + .orders + .iter() + .map(|order| (order.uid, order)) + .collect(); + + // The deadline sent to the drivers is also enforced here: a response + // landing after it is dropped, so one hung driver cannot hold the + // loop for the HTTP client's whole ceiling. + let budget = SLOT_DURATION * u32::try_from(SOLVE_DEADLINE_SLOTS).expect("small constant"); + let responses = futures::future::join_all(self.drivers.iter().enumerate().map( + |(driver_index, driver)| async move { + match tokio::time::timeout(budget, driver.solve(request)).await { + Ok(Ok(response)) => Some((driver_index, driver, response)), + Ok(Err(err)) => { + tracing::warn!(driver = %driver.name, ?err, "solve failed"); + None + } + Err(_) => { + tracing::warn!(driver = %driver.name, "solve missed the deadline"); + None + } + } + }, + )) + .await; + + let mut solutions = Vec::new(); + // `(solver, solution id)` attributes a ranked solution back to its + // driver, so a duplicate would misdispatch the settlement: keep the + // first and drop the rest. + let mut seen = HashSet::new(); + for (driver_index, driver, response) in responses.into_iter().flatten() { + for dto_solution in response.solutions { + let duplicate = !seen.insert((dto_solution.solver, dto_solution.solution_id)); + if duplicate { + tracing::warn!( + driver = %driver.name, + solution_id = dto_solution.solution_id, + "duplicate solver and solution id, dropped" + ); + continue; + } + match convert(driver_index, dto_solution, &by_uid) { + Some(solution) => solutions.push(solution), + None => tracing::warn!( + driver = %driver.name, + "solution names an order outside the auction, dropped" + ), + } + } + } + solutions + } +} + +/// A wire solution becomes an arbitrator solution by joining each traded +/// order against the auction for its limits and tokens. A trade naming an +/// order outside the auction voids the whole solution: its intent cannot be +/// scored. +fn convert( + driver_index: usize, + dto_solution: dto::Solution, + by_uid: &HashMap, +) -> Option { + let orders = dto_solution + .orders + .iter() + .map(|(uid, amounts)| { + let order = by_uid.get(uid)?; + Some(solution::Order:: { + uid: *uid, + sell_token: order.sell_token, + buy_token: order.buy_token, + sell_amount: order.sell_amount, + buy_amount: order.buy_amount, + executed_sell: amounts.executed_sell, + executed_buy: amounts.executed_buy, + side: match order.kind { + OrderKind::Sell => Side::Sell, + OrderKind::Buy => Side::Buy, + }, + }) + }) + .collect::>>()?; + Some(Solution { + driver_index, + inner: solution::Solution::new(dto_solution.solution_id, dto_solution.solver, orders), + }) +} diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index 065aee619f..f37833b9db 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -102,10 +102,10 @@ ORDER BY slot, tx_signature .context("read solana.settlements by auction") } -/// Cut an auction from the orders currently open for solving. -pub async fn cut(ex: impl PgExecutor<'_>, id: i64, now_unix: i64) -> Result { +/// Cut an auction at the given tip from the open orders. +pub async fn cut(ex: impl PgExecutor<'_>, id: i64, tip: u64, now_unix: i64) -> Result { let orders = orders_from_rows(open_orders(ex, now_unix).await?); - Ok(Auction { id, orders }) + Ok(Auction { id, tip, orders }) } /// A row the indexer wrote always converts (on-chain values fit the domain diff --git a/crates/autopilot-svm/src/infra/driver/mod.rs b/crates/autopilot-svm/src/infra/driver/mod.rs index 7f43cdb182..6da628e6bb 100644 --- a/crates/autopilot-svm/src/infra/driver/mod.rs +++ b/crates/autopilot-svm/src/infra/driver/mod.rs @@ -1,7 +1,5 @@ //! HTTP client for the Solana driver. -#![expect(dead_code, reason = "consumed by the auction loop wiring")] - pub mod dto; use { diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs new file mode 100644 index 0000000000..e3ff1671c9 --- /dev/null +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -0,0 +1,70 @@ +//! Dispatches winning solutions back to their drivers for settlement. + +use { + crate::{ + domain::cycle::{Ranking, SolanaCycle}, + infra::driver::{Driver, dto}, + run_loop::SettlementExecutor, + }, + async_trait::async_trait, + std::sync::Arc, +}; + +/// Slots a settlement may take after ranking before it counts as late. +/// TODO: make configurable. +const SUBMISSION_DEADLINE_SLOTS: u64 = 25; + +/// Sends `/settle` to each winner's driver. Submission runs detached, the +/// loop starts the next cycle while settlements land. +pub struct DriverExecutor { + drivers: Vec>, +} + +impl DriverExecutor { + pub fn new(drivers: Vec>) -> Self { + Self { drivers } + } +} + +#[async_trait] +impl SettlementExecutor for DriverExecutor { + fn submission_deadline(&self, tip: &u64) -> u64 { + tip + SUBMISSION_DEADLINE_SLOTS + } + + async fn execute(&self, auction_id: i64, ranking: &Ranking, deadline: u64) { + for winner in ranking.inner.winners() { + let key = (winner.solver(), winner.id()); + let Some(driver) = ranking + .drivers + .get(&key) + .and_then(|&index| self.drivers.get(index)) + else { + tracing::error!(solution_id = winner.id(), "winner without a driver"); + continue; + }; + let driver = Arc::clone(driver); + let request = dto::SettleRequest { + auction_id, + solution_id: winner.id(), + }; + tokio::spawn(async move { + match driver.settle(&request).await { + Ok(response) => tracing::info!( + driver = %driver.name, + auction_id, + deadline, + tx_signature = %response.tx_signature, + "settlement submitted" + ), + Err(err) => tracing::error!( + driver = %driver.name, + auction_id, + ?err, + "settlement failed" + ), + } + }); + } + } +} diff --git a/crates/autopilot-svm/src/infra/mod.rs b/crates/autopilot-svm/src/infra/mod.rs index 73835f0e28..0dc98dc5a7 100644 --- a/crates/autopilot-svm/src/infra/mod.rs +++ b/crates/autopilot-svm/src/infra/mod.rs @@ -1,5 +1,10 @@ -//! Infrastructure: database reads and the Postgres LISTEN session. +//! Infrastructure: database access, driver clients, and the loop seams. +pub mod competition; pub mod db; pub mod driver; +pub mod executor; pub mod listen; +pub mod observer; +pub mod provider; +pub mod trigger; diff --git a/crates/autopilot-svm/src/infra/observer.rs b/crates/autopilot-svm/src/infra/observer.rs new file mode 100644 index 0000000000..2dff2ab3f3 --- /dev/null +++ b/crates/autopilot-svm/src/infra/observer.rs @@ -0,0 +1,60 @@ +//! Competition bookkeeping. Everything is logged only: there are no +//! competition tables (auction snapshots, proposed executions, order +//! events) to write. +//! +//! TODO: persist the competition outcome once the tables exist. The +//! settlement attribution (`solana.settlements.solution_uid`) depends on +//! the persisted ranking. + +use { + crate::{ + domain::{auction::Auction, cycle::Ranking}, + run_loop::SettlementObserver, + }, + async_trait::async_trait, + chain_types::solana::IntentHash, + std::collections::HashSet, +}; + +pub struct LogObserver; + +#[async_trait] +impl SettlementObserver for LogObserver { + fn on_orders_ready(&self, auction: &Auction) { + tracing::debug!(orders = auction.orders.len(), "auction entered competition"); + } + + async fn persist_competition_ranking( + &self, + _auction: &Auction, + tip: &u64, + ranking: &Ranking, + deadline: u64, + ) -> anyhow::Result<()> { + tracing::info!( + tip, + deadline, + winners = ranking.inner.winners().count(), + ranked = ranking.inner.ranked.len(), + filtered_out = ranking.inner.filtered_out.len(), + "competition ranked" + ); + Ok(()) + } + + fn on_orders_matched(&self, executing: HashSet, considered: HashSet) { + tracing::debug!( + executing = executing.len(), + considered = considered.len(), + "orders matched" + ); + } + + fn on_competition_ended(&self, auction: &Auction, ranking: &Ranking) { + tracing::debug!( + auction_id = auction.id, + winners = ranking.inner.winners().count(), + "competition ended" + ); + } +} diff --git a/crates/autopilot-svm/src/infra/provider.rs b/crates/autopilot-svm/src/infra/provider.rs new file mode 100644 index 0000000000..2a4a4f75e3 --- /dev/null +++ b/crates/autopilot-svm/src/infra/provider.rs @@ -0,0 +1,70 @@ +//! Auction provider backed by the indexer-written tables. + +use { + crate::{domain::cycle::SolanaCycle, infra::db, run_loop::AuctionProvider}, + async_trait::async_trait, + sqlx::PgPool, + std::{ + sync::atomic::{AtomicI64, Ordering}, + time::{SystemTime, UNIX_EPOCH}, + }, +}; + +/// Cuts auctions from the open orders the indexer persisted. +pub struct DbAuctionProvider { + pool: PgPool, + /// Last allocated auction id. Ids are unix seconds, bumped past the + /// previous allocation when cycles land within the same second. Unique + /// only per process: no table allocates auction ids. + /// TODO: allocate from the auctions table sequence once competition + /// persistence lands, like the EVM `auctions.id` bigserial. + last_id: AtomicI64, +} + +impl DbAuctionProvider { + pub fn new(pool: PgPool) -> Self { + Self { + pool, + last_id: AtomicI64::new(0), + } + } + + /// Allocates the next auction id: the current unix second, or one past + /// the previous id when several cycles land within the same second, so + /// ids strictly increase within the process. + fn next_id(&self, now: i64) -> i64 { + let prev = self + .last_id + .update(Ordering::Relaxed, Ordering::Relaxed, |prev| { + now.max(prev + 1) + }); + now.max(prev + 1) + } +} + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock after the unix epoch") + .as_secs() + .try_into() + .expect("unix seconds fit i64") +} + +#[async_trait] +impl AuctionProvider for DbAuctionProvider { + /// The order data is push-fed by the indexer, there is no cache to + /// refresh. + async fn sync_to_tip(&self, _tip: &u64) -> anyhow::Result<()> { + Ok(()) + } + + async fn cut_auction(&self, tip: &u64) -> Option { + let now = now_unix(); + let auction = db::cut(&self.pool, self.next_id(now), *tip, now) + .await + .map_err(|err| tracing::warn!(?err, "failed to cut the auction")) + .ok()?; + (!auction.orders.is_empty()).then_some(auction) + } +} diff --git a/crates/autopilot-svm/src/infra/trigger.rs b/crates/autopilot-svm/src/infra/trigger.rs new file mode 100644 index 0000000000..29f97b4ac8 --- /dev/null +++ b/crates/autopilot-svm/src/infra/trigger.rs @@ -0,0 +1,54 @@ +//! Chain-driven cycle trigger: wakes the loop once per new slot. + +use { + crate::{domain::cycle::SolanaCycle, run_loop::CycleTrigger}, + async_trait::async_trait, + cow_solana_rpc::SolanaRPC, + std::{ + sync::atomic::{AtomicU64, Ordering}, + time::Duration, + }, +}; + +/// How often the trigger asks the node for the current slot. Half a slot, so +/// a new slot is typically observed within one poll of its arrival. +const POLL_INTERVAL: Duration = Duration::from_millis(200); + +/// Polls the node for the slot and yields each newly observed one. +pub struct SlotTrigger { + rpc: SolanaRPC, + tip: AtomicU64, +} + +impl SlotTrigger { + pub fn new(rpc: SolanaRPC) -> Self { + Self { + rpc, + tip: AtomicU64::new(0), + } + } +} + +#[async_trait] +impl CycleTrigger for SlotTrigger { + async fn next_cycle(&mut self) -> u64 { + loop { + match self.rpc.slot().await { + Ok(slot) => { + let previous = self.tip.fetch_max(slot, Ordering::Relaxed); + if slot > previous { + return slot; + } + } + Err(err) => tracing::warn!(?err, "failed to poll the slot"), + } + tokio::time::sleep(POLL_INTERVAL).await; + } + } + + // TODO: poll in the background so the tip observed after ranking is + // fresher than the cut tip once the submission deadline is enforced. + fn current_tip(&self) -> u64 { + self.tip.load(Ordering::Relaxed) + } +} diff --git a/crates/autopilot-svm/src/lib.rs b/crates/autopilot-svm/src/lib.rs index bd13bb1702..b86fc0ec35 100644 --- a/crates/autopilot-svm/src/lib.rs +++ b/crates/autopilot-svm/src/lib.rs @@ -1,8 +1,11 @@ //! Solana-side autopilot components built on a chain-generic auction loop. -#![expect(dead_code, reason = "consumed by the auction loop wiring")] +#![expect(dead_code, reason = "consumed by the binary wiring")] pub mod run_loop; mod domain; mod infra; + +#[cfg(test)] +mod tests; diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs new file mode 100644 index 0000000000..a8471828fa --- /dev/null +++ b/crates/autopilot-svm/src/tests.rs @@ -0,0 +1,182 @@ +//! The BE-184 checkpoint: one full auction cycle through [`AuctionLoop`] +//! against the real database and a mocked driver. + +use { + crate::{ + domain::{arbitrator::SolanaArbitrator, cycle::SolanaCycle}, + infra::{ + competition::DriverCompetition, + driver::{Driver, dto}, + executor::DriverExecutor, + observer::LogObserver, + provider::DbAuctionProvider, + }, + run_loop::{ + AuctionLoop, + AuctionProvider, + CycleTrigger, + RankingInfo, + SolverCompetition, + WinnerSelection, + }, + }, + async_trait::async_trait, + axum::{Json, Router, extract::State, routing::post}, + chain_types::solana::{IntentHash, Pubkey, Signature}, + database::byte_array::ByteArray, + sqlx::PgPool, + std::{collections::HashMap, net::SocketAddr, sync::Arc}, + tokio::sync::mpsc, + url::Url, +}; + +/// A trigger pinned to one slot: the test drives exactly one cycle. +struct FixedTrigger(u64); + +#[async_trait] +impl CycleTrigger for FixedTrigger { + async fn next_cycle(&mut self) -> u64 { + self.0 + } + + fn current_tip(&self) -> u64 { + self.0 + } +} + +#[derive(Clone)] +struct MockDriverState { + solution: dto::Solution, + settles: mpsc::UnboundedSender, +} + +async fn handle_solve(State(state): State) -> Json { + Json( + serde_json::to_value(&dto::SolveResponse { + solutions: vec![state.solution.clone()], + }) + .unwrap(), + ) +} + +async fn handle_settle( + State(state): State, + Json(request): Json, +) -> Json { + state.settles.send(request).unwrap(); + Json( + serde_json::to_value(&dto::SettleResponse { + tx_signature: Signature([9; 64]), + }) + .unwrap(), + ) +} + +/// Serves `/solve` with one canned solution and records every `/settle`. +async fn spawn_mock_driver(state: MockDriverState) -> SocketAddr { + let app = Router::new() + .route("/solve", post(handle_solve)) + .route("/settle", post(handle_settle)) + .with_state(state); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + addr +} + +async fn seed_open_order(pool: &PgPool, uid: [u8; 32], tip: i64) { + sqlx::query( + "TRUNCATE solana.trades, solana.settlements, solana.order_pda, solana.orders, \ + solana.indexer_state", + ) + .execute(pool) + .await + .unwrap(); + sqlx::query("INSERT INTO solana.indexer_state (slot) VALUES ($1)") + .bind(tip) + .execute(pool) + .await + .unwrap(); + sqlx::query( + r#" +INSERT INTO solana.orders (uid, owner, sell_token, buy_token, sell_token_account, + buy_token_account, sell_amount, buy_amount, valid_to, kind, + partially_fillable, app_data, creation_timestamp, order_pda) +VALUES ($1, $2, $2, $3, $2, $2, 1000, 500, $4, 'sell'::OrderKind, false, $2, now(), $5) + "#, + ) + .bind(uid) + .bind(ByteArray([0xAA; 32])) + .bind(ByteArray([0xAB; 32])) + .bind(i64::from(u32::MAX)) + .bind(ByteArray([0xB0; 32])) + .execute(pool) + .await + .unwrap(); + sqlx::query("INSERT INTO solana.order_pda (order_uid, created_by) VALUES ($1, $2)") + .bind(uid) + .bind(ByteArray([0xAA; 32])) + .execute(pool) + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "needs the solana.* schema applied to the local database"] +async fn solana_db_mock_cycle_dispatches_the_settlement() { + let pool = PgPool::connect("postgresql://").await.unwrap(); + let uid = [0x11; 32]; + let tip = 500_u64; + seed_open_order(&pool, uid, i64::try_from(tip).unwrap()).await; + + // Selling 1000 for at least 500 and receiving 600 clears the limit, so + // the solution scores its 100 surplus and wins. + let solution = dto::Solution { + solution_id: 7, + score: 100, + solver: Pubkey([0xCC; 32]), + orders: HashMap::from([( + IntentHash(uid), + dto::TradedAmounts { + executed_sell: 1000, + executed_buy: 600, + }, + )]), + }; + let wrapped_native = Pubkey([0xFE; 32]); + let (settles, mut settled) = mpsc::unbounded_channel(); + let addr = spawn_mock_driver(MockDriverState { solution, settles }).await; + let driver = Arc::new(Driver::new( + "mock".to_string(), + &Url::parse(&format!("http://{addr}")).unwrap(), + )); + + // Stage probes: pinpoint the failing phase before driving the loop. + { + let provider = DbAuctionProvider::new(pool.clone()); + let auction = provider.cut_auction(&tip).await.expect("auction cut"); + assert_eq!(auction.orders.len(), 1, "open order in the auction"); + let competition = DriverCompetition::new(vec![Arc::clone(&driver)]); + let solutions = competition.solve(&auction).await; + assert_eq!(solutions.len(), 1, "driver solution converted"); + let ranking = SolanaArbitrator::new(1, wrapped_native).arbitrate(solutions, &auction); + assert_eq!(ranking.winner_count(), 1, "solution won"); + } + + let mut auction_loop = AuctionLoop::new( + Box::new(FixedTrigger(tip)), + Box::new(DbAuctionProvider::new(pool)), + Box::new(DriverCompetition::new(vec![Arc::clone(&driver)])), + Box::new(SolanaArbitrator::new(1, wrapped_native)), + Box::new(DriverExecutor::new(vec![driver])), + Box::new(LogObserver), + ); + auction_loop.run_cycle().await; + + let settle = tokio::time::timeout(std::time::Duration::from_secs(5), settled.recv()) + .await + .expect("settlement dispatched before the timeout") + .expect("settle channel open"); + assert_eq!(settle.solution_id, 7); + assert!(settle.auction_id > 0); +} diff --git a/crates/cow-solana-rpc/Cargo.toml b/crates/cow-solana-rpc/Cargo.toml index 19d44b8e6e..3a6c3e1975 100644 --- a/crates/cow-solana-rpc/Cargo.toml +++ b/crates/cow-solana-rpc/Cargo.toml @@ -12,6 +12,7 @@ path = "src/lib.rs" [dependencies] solana-commitment-config = { workspace = true } solana-rpc-client = { workspace = true } +solana-rpc-client-api = { workspace = true } url = { workspace = true } [features] diff --git a/crates/cow-solana-rpc/src/lib.rs b/crates/cow-solana-rpc/src/lib.rs index dac40f87d5..f3b4b5baea 100644 --- a/crates/cow-solana-rpc/src/lib.rs +++ b/crates/cow-solana-rpc/src/lib.rs @@ -1,12 +1,9 @@ //! Solana JSON-RPC client wrapper. -pub use solana_commitment_config::CommitmentConfig; +pub use {solana_commitment_config::CommitmentConfig, solana_rpc_client_api::client_error::Error}; use {solana_rpc_client::nonblocking::rpc_client::RpcClient, std::time::Duration, url::Url}; pub struct SolanaRPC { - /// Not yet read: helper methods that use the underlying client will be - /// added in follow-up PRs. - #[expect(dead_code)] inner: RpcClient, } @@ -35,4 +32,9 @@ impl SolanaRPC { inner: RpcClient::new_mock(url), } } + + /// The node's current slot at the client's commitment level. + pub async fn slot(&self) -> Result { + self.inner.get_slot().await + } } From 760fcae6e8993d58ed4ab6373c15050e9153b232 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Wed, 19 Aug 2026 11:14:57 +0000 Subject: [PATCH 02/12] PR09: settlement observation over execution windows --- crates/autopilot-svm/src/infra/executor.rs | 26 +- crates/autopilot-svm/src/infra/mod.rs | 1 + crates/autopilot-svm/src/infra/observation.rs | 328 ++++++++++++++++++ crates/autopilot-svm/src/infra/observer.rs | 20 +- crates/autopilot-svm/src/tests.rs | 20 +- ...V2__solana_order_and_settlement_tables.sql | 13 + 6 files changed, 398 insertions(+), 10 deletions(-) create mode 100644 crates/autopilot-svm/src/infra/observation.rs diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs index e3ff1671c9..376541eb48 100644 --- a/crates/autopilot-svm/src/infra/executor.rs +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -3,7 +3,10 @@ use { crate::{ domain::cycle::{Ranking, SolanaCycle}, - infra::driver::{Driver, dto}, + infra::{ + driver::{Driver, dto}, + observation::SettlementTracker, + }, run_loop::SettlementExecutor, }, async_trait::async_trait, @@ -18,11 +21,13 @@ const SUBMISSION_DEADLINE_SLOTS: u64 = 25; /// loop starts the next cycle while settlements land. pub struct DriverExecutor { drivers: Vec>, + /// Dispatched settlements the observation side resolves or times out. + tracker: SettlementTracker, } impl DriverExecutor { - pub fn new(drivers: Vec>) -> Self { - Self { drivers } + pub fn new(drivers: Vec>, tracker: SettlementTracker) -> Self { + Self { drivers, tracker } } } @@ -48,6 +53,21 @@ impl SettlementExecutor for DriverExecutor { auction_id, solution_id: winner.id(), }; + // A window that cannot be opened must not block the settlement, + // the dispatch is the priority. + if let Err(err) = self + .tracker + .dispatched( + auction_id, + winner.solver(), + winner.id(), + deadline.saturating_sub(SUBMISSION_DEADLINE_SLOTS), + deadline, + ) + .await + { + tracing::error!(auction_id, ?err, "failed to open the settlement window"); + } tokio::spawn(async move { match driver.settle(&request).await { Ok(response) => tracing::info!( diff --git a/crates/autopilot-svm/src/infra/mod.rs b/crates/autopilot-svm/src/infra/mod.rs index 0dc98dc5a7..f686f9e957 100644 --- a/crates/autopilot-svm/src/infra/mod.rs +++ b/crates/autopilot-svm/src/infra/mod.rs @@ -5,6 +5,7 @@ pub mod db; pub mod driver; pub mod executor; pub mod listen; +pub mod observation; pub mod observer; pub mod provider; pub mod trigger; diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs new file mode 100644 index 0000000000..cc46e2928a --- /dev/null +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -0,0 +1,328 @@ +//! Settlement observation: tracks dispatched settlements as open windows in +//! `solana.settlement_executions` and resolves them against the +//! indexer-written `solana.settlements` rows. +//! +//! The executor opens a window per dispatched settlement. The indexer's +//! insert into `solana.settlements` fires the `solana_settlement_finalized` +//! NOTIFY (trigger in the schema), a [`ListenSession`] delivers it here, and +//! the window closes as `landed`. Windows whose submission deadline passes +//! without a settlement close as `timeout`. Windows live in the database, so +//! a restart mid-window loses nothing: the listen seed re-checks every open +//! window. + +use { + crate::infra::{db, listen::NotifyHandler}, + anyhow::{Context, Result}, + async_trait::async_trait, + chain_types::solana::Pubkey, + sqlx::PgPool, +}; + +/// A closed window's outcome, the `solana.settlement_executions.outcome` +/// value. The outcome records what the indexer observed on chain, never a +/// driver's report, which is why a landing observed after the deadline may +/// overwrite a timeout. The schema also allows `rejected`, unused until +/// typed `/settle` errors are consumed. +enum Outcome { + Landed, + Timeout, +} + +impl Outcome { + fn as_str(&self) -> &'static str { + match self { + Self::Landed => "landed", + Self::Timeout => "timeout", + } + } +} + +/// Writes and closes settlement-execution windows. +#[derive(Clone)] +pub struct SettlementTracker { + pool: PgPool, +} + +impl SettlementTracker { + pub fn new(pool: PgPool) -> Self { + Self { pool } + } + + /// Open a window for a dispatched settlement. `solution_uid` is the + /// winner's driver-local solution id until competition persistence + /// allocates uids. + pub async fn dispatched( + &self, + auction_id: i64, + solver: Pubkey, + solution_uid: u64, + start_slot: u64, + deadline_slot: u64, + ) -> Result<()> { + sqlx::query( + r#" +INSERT INTO solana.settlement_executions + (auction_id, solver, solution_uid, start_timestamp, start_slot, deadline_slot) +VALUES ($1, $2, $3, now(), $4, $5) +ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING + "#, + ) + .bind(auction_id) + .bind(solver.0) + .bind(to_db_u64(solution_uid)) + .bind(to_db_u64(start_slot)) + .bind(to_db_u64(deadline_slot)) + .execute(&self.pool) + .await + .context("open settlement execution window")?; + Ok(()) + } + + /// Close the solver's windows of the auction as landed, recording the + /// settlement's slot and signature. Keyed by solver so concurrent winners + /// of one auction each close on their own settlement. A window already + /// closed as timed out upgrades to landed: the settlement executed, just + /// late, and lateness stays visible as `end_slot` past `deadline_slot`. + async fn landed( + &self, + auction_id: i64, + solver: &[u8], + slot: i64, + signature: &[u8], + ) -> Result { + let closed = sqlx::query( + r#" +UPDATE solana.settlement_executions +SET outcome = $4, end_timestamp = now(), end_slot = $2, submitted_signature = $3 +WHERE auction_id = $1 AND solver = $6 AND (outcome IS NULL OR outcome = $5) + "#, + ) + .bind(auction_id) + .bind(slot) + .bind(signature) + .bind(Outcome::Landed.as_str()) + .bind(Outcome::Timeout.as_str()) + .bind(solver) + .execute(&self.pool) + .await + .context("close settlement execution window")? + .rows_affected(); + Ok(closed) + } + + /// Close every open window whose deadline is at or before the slot as + /// timed out, logging each. Driven by the competition cycle, so on a + /// chain with no active auctions a timeout surfaces with the next + /// competition, not at its deadline slot. + pub async fn flag_expired(&self, slot: u64) -> Result<()> { + let expired: Vec<(i64,)> = sqlx::query_as( + r#" +UPDATE solana.settlement_executions +SET outcome = $2, end_timestamp = now(), end_slot = $1 +WHERE outcome IS NULL AND deadline_slot <= $1 +RETURNING auction_id + "#, + ) + .bind(to_db_u64(slot)) + .bind(Outcome::Timeout.as_str()) + .fetch_all(&self.pool) + .await + .context("expire settlement execution windows")?; + for (auction_id,) in expired { + tracing::error!(auction_id, slot, "settlement missed its deadline"); + } + Ok(()) + } + + /// Auction ids of open windows, the seed re-read set. + async fn open_windows(&self) -> Result> { + let rows: Vec<(i64,)> = sqlx::query_as( + "SELECT auction_id FROM solana.settlement_executions WHERE outcome IS NULL", + ) + .fetch_all(&self.pool) + .await + .context("read open settlement execution windows")?; + Ok(rows.into_iter().map(|(auction_id,)| auction_id).collect()) + } +} + +/// Slots and solution ids stay far below `i64::MAX`. +fn to_db_u64(value: u64) -> i64 { + i64::try_from(value).expect("value exceeds i64") +} + +/// Resolves NOTIFY payloads (and reconnect re-reads) against the settlements +/// table, closing the matching windows. +pub struct SettlementObservation { + pool: PgPool, + tracker: SettlementTracker, +} + +impl SettlementObservation { + pub fn new(pool: PgPool, tracker: SettlementTracker) -> Self { + Self { pool, tracker } + } + + /// Close the matching windows for every settlement observed for the + /// auction, each keyed to its solver. + async fn resolve(&self, auction_id: i64) -> Result<()> { + for settlement in db::settlements_by_auction(&self.pool, auction_id).await? { + let closed = self + .tracker + .landed( + auction_id, + &settlement.solver.0, + settlement.slot, + &settlement.tx_signature.0, + ) + .await?; + if closed > 0 { + tracing::info!( + auction_id, + slot = settlement.slot, + tx_signature = %const_hex::encode(settlement.tx_signature.0), + "settlement observed on chain" + ); + } + } + Ok(()) + } +} + +#[async_trait] +impl NotifyHandler for SettlementObservation { + /// Re-check every open window: a NOTIFY missed while the connection was + /// down (or while the autopilot was not running) is recovered here. + async fn seed(&mut self) -> Result<()> { + for auction_id in self.tracker.open_windows().await? { + self.resolve(auction_id).await?; + } + Ok(()) + } + + async fn on_notify(&mut self, payload: &str) -> Result<()> { + let Ok(auction_id) = payload.parse::() else { + tracing::warn!(payload, "unparsable settlement notify payload"); + return Ok(()); + }; + self.resolve(auction_id).await + } +} + +#[cfg(test)] +mod tests { + use { + super::{SettlementObservation, SettlementTracker}, + crate::infra::listen::ListenSession, + chain_types::solana::Pubkey, + sqlx::PgPool, + std::time::Duration, + }; + + async fn wipe(pool: &PgPool) { + for table in [ + "solana.trades", + "solana.settlements", + "solana.settlement_executions", + ] { + sqlx::query(&format!("DELETE FROM {table}")) + .execute(pool) + .await + .unwrap(); + } + } + + async fn insert_settlement(pool: &PgPool, auction_id: i64) { + sqlx::query( + r#" +INSERT INTO solana.settlements (slot, tx_signature, instruction_index, solver, auction_id, solution_uid) +VALUES (10, $1, 0, $2, $3, NULL) + "#, + ) + .bind([9u8; 64]) + .bind([7u8; 32]) + .bind(auction_id) + .execute(pool) + .await + .unwrap(); + } + + async fn outcome(pool: &PgPool, auction_id: i64) -> Option { + sqlx::query_scalar("SELECT outcome FROM solana.settlement_executions WHERE auction_id = $1") + .bind(auction_id) + .fetch_one(pool) + .await + .unwrap() + } + + /// The full path: a dispatched settlement opens a window, the trigger's + /// NOTIFY (here: a bare INSERT, standing in for the indexer) closes it + /// as landed with the settlement's signature. + #[tokio::test] + #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] + async fn solana_db_settlement_notify_closes_the_window_as_landed() { + let pool = PgPool::connect("postgresql://").await.unwrap(); + wipe(&pool).await; + + let tracker = SettlementTracker::new(pool.clone()); + tracker + .dispatched(4242, Pubkey([7; 32]), 1, 90, 100) + .await + .unwrap(); + + let task = ListenSession::spawn( + pool.clone(), + "solana_settlement_finalized", + SettlementObservation::new(pool.clone(), tracker.clone()), + ); + + insert_settlement(&pool, 4242).await; + + for _ in 0..200 { + if outcome(&pool, 4242).await.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + task.abort(); + assert_eq!(outcome(&pool, 4242).await.as_deref(), Some("landed")); + let signature: Vec = sqlx::query_scalar( + "SELECT submitted_signature FROM solana.settlement_executions WHERE auction_id = 4242", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(signature, vec![9u8; 64]); + } + + /// Expiry closes only windows past their deadline, and closed windows + /// are not re-expired. + #[tokio::test] + #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] + async fn solana_db_expiry_times_out_only_past_deadlines() { + let pool = PgPool::connect("postgresql://").await.unwrap(); + wipe(&pool).await; + + let tracker = SettlementTracker::new(pool.clone()); + tracker + .dispatched(1, Pubkey([7; 32]), 1, 90, 100) + .await + .unwrap(); + tracker + .dispatched(2, Pubkey([7; 32]), 1, 90, 200) + .await + .unwrap(); + + tracker.flag_expired(150).await.unwrap(); + assert_eq!(outcome(&pool, 1).await.as_deref(), Some("timeout")); + assert_eq!(outcome(&pool, 2).await, None); + + // A settlement observed after the timeout upgrades the verdict: it + // executed, just late. + tracker + .landed(1, &[7u8; 32], 160, &[9u8; 64]) + .await + .unwrap(); + assert_eq!(outcome(&pool, 1).await.as_deref(), Some("landed")); + } +} diff --git a/crates/autopilot-svm/src/infra/observer.rs b/crates/autopilot-svm/src/infra/observer.rs index 2dff2ab3f3..abbdbcd494 100644 --- a/crates/autopilot-svm/src/infra/observer.rs +++ b/crates/autopilot-svm/src/infra/observer.rs @@ -1,4 +1,4 @@ -//! Competition bookkeeping. Everything is logged only: there are no +//! Competition bookkeeping. Outcomes are logged only: there are no //! competition tables (auction snapshots, proposed executions, order //! events) to write. //! @@ -9,6 +9,7 @@ use { crate::{ domain::{auction::Auction, cycle::Ranking}, + infra::observation::SettlementTracker, run_loop::SettlementObserver, }, async_trait::async_trait, @@ -16,7 +17,17 @@ use { std::collections::HashSet, }; -pub struct LogObserver; +/// Logs the competition phases and drives the settlement-timeout check off +/// the per-cycle tip. +pub struct LogObserver { + tracker: SettlementTracker, +} + +impl LogObserver { + pub fn new(tracker: SettlementTracker) -> Self { + Self { tracker } + } +} #[async_trait] impl SettlementObserver for LogObserver { @@ -31,6 +42,11 @@ impl SettlementObserver for LogObserver { ranking: &Ranking, deadline: u64, ) -> anyhow::Result<()> { + // Best effort: the expiry bookkeeping only touches previously + // dispatched windows and must not block the current dispatch. + if let Err(err) = self.tracker.flag_expired(*tip).await { + tracing::error!(?err, "failed to flag expired settlement windows"); + } tracing::info!( tip, deadline, diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index a8471828fa..f404791066 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -8,6 +8,7 @@ use { competition::DriverCompetition, driver::{Driver, dto}, executor::DriverExecutor, + observation::SettlementTracker, observer::LogObserver, provider::DbAuctionProvider, }, @@ -86,8 +87,8 @@ async fn spawn_mock_driver(state: MockDriverState) -> SocketAddr { async fn seed_open_order(pool: &PgPool, uid: [u8; 32], tip: i64) { sqlx::query( - "TRUNCATE solana.trades, solana.settlements, solana.order_pda, solana.orders, \ - solana.indexer_state", + "TRUNCATE solana.trades, solana.settlements, solana.settlement_executions, \ + solana.order_pda, solana.orders, solana.indexer_state", ) .execute(pool) .await @@ -163,13 +164,14 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { assert_eq!(ranking.winner_count(), 1, "solution won"); } + let tracker = SettlementTracker::new(pool.clone()); let mut auction_loop = AuctionLoop::new( Box::new(FixedTrigger(tip)), - Box::new(DbAuctionProvider::new(pool)), + Box::new(DbAuctionProvider::new(pool.clone())), Box::new(DriverCompetition::new(vec![Arc::clone(&driver)])), Box::new(SolanaArbitrator::new(1, wrapped_native)), - Box::new(DriverExecutor::new(vec![driver])), - Box::new(LogObserver), + Box::new(DriverExecutor::new(vec![driver], tracker.clone())), + Box::new(LogObserver::new(tracker.clone())), ); auction_loop.run_cycle().await; @@ -179,4 +181,12 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { .expect("settle channel open"); assert_eq!(settle.solution_id, 7); assert!(settle.auction_id > 0); + // The dispatch opened a settlement-execution window. + let open_windows: i64 = sqlx::query_scalar( + "SELECT count(*) FROM solana.settlement_executions WHERE outcome IS NULL", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(open_windows, 1); } diff --git a/database/sql-solana/V2__solana_order_and_settlement_tables.sql b/database/sql-solana/V2__solana_order_and_settlement_tables.sql index b5fd0282cb..35d2e1faca 100644 --- a/database/sql-solana/V2__solana_order_and_settlement_tables.sql +++ b/database/sql-solana/V2__solana_order_and_settlement_tables.sql @@ -77,6 +77,19 @@ CREATE TABLE solana.settlements ( ); CREATE INDEX solana_settlements_auction_id ON solana.settlements (auction_id); + +-- Wakes the autopilot's settlement observation: settlements are rare, so a +-- NOTIFY beats polling. The payload is the auction id the settlement claims. +CREATE FUNCTION solana.notify_settlement_finalized() RETURNS trigger AS $$ +BEGIN + PERFORM pg_notify('solana_settlement_finalized', NEW.auction_id::text); + RETURN NULL; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER solana_settlement_finalized +AFTER INSERT ON solana.settlements +FOR EACH ROW EXECUTE FUNCTION solana.notify_settlement_finalized(); -- Per-order accounting deltas of a settlement. order_uid completes the key -- because one settlement moves several orders. CREATE TABLE solana.trades ( From dc194bced0b2cb8322e515eb2d080411d5261171 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 20 Aug 2026 07:58:06 +0000 Subject: [PATCH 03/12] Make the solve budget a wall-clock duration like the EVM autopilot --- crates/autopilot-svm/src/domain/arbitrator.rs | 2 +- crates/autopilot-svm/src/infra/competition.rs | 22 ++++++++++--------- crates/autopilot-svm/src/tests.rs | 11 ++++++---- 3 files changed, 20 insertions(+), 15 deletions(-) diff --git a/crates/autopilot-svm/src/domain/arbitrator.rs b/crates/autopilot-svm/src/domain/arbitrator.rs index 61c691337f..7c9659de10 100644 --- a/crates/autopilot-svm/src/domain/arbitrator.rs +++ b/crates/autopilot-svm/src/domain/arbitrator.rs @@ -47,7 +47,7 @@ impl WinnerSelection for SolanaArbitrator { // An empty fee-policy list marks an order as part of the auction, the // arbitrator only scores such orders. The auction carries no protocol // fees. - // TODO: real fee policies arrive with the MVP protocol-fee support. + // TODO: real fee policies arrive with the protocol-fee support (post-MVP). let fee_policies = auction .orders .iter() diff --git a/crates/autopilot-svm/src/infra/competition.rs b/crates/autopilot-svm/src/infra/competition.rs index 7cf74f3b93..bcf587386a 100644 --- a/crates/autopilot-svm/src/infra/competition.rs +++ b/crates/autopilot-svm/src/infra/competition.rs @@ -19,12 +19,6 @@ use { winner_selection::{Side, solution}, }; -/// Slots between the tip a solve request is stamped at and its deadline, -/// the driver's whole thinking budget (~6s at 400ms slots, the EVM fast -/// chains run 6s solve deadlines). -/// TODO: make configurable. -const SOLVE_DEADLINE_SLOTS: u64 = 15; - /// Solana's target slot time. const SLOT_DURATION: Duration = Duration::from_millis(400); @@ -32,11 +26,16 @@ const SLOT_DURATION: Duration = Duration::from_millis(400); /// solutions for the arbitrator. pub struct DriverCompetition { drivers: Vec>, + /// How long a driver gets to answer `/solve`. + solve_deadline: Duration, } impl DriverCompetition { - pub fn new(drivers: Vec>) -> Self { - Self { drivers } + pub fn new(drivers: Vec>, solve_deadline: Duration) -> Self { + Self { + drivers, + solve_deadline, + } } } @@ -45,7 +44,10 @@ impl SolverCompetition for DriverCompetition { async fn solve(&self, auction: &Auction) -> Vec { let request = &dto::SolveRequest { id: auction.id, - deadline_slot: auction.tip + SOLVE_DEADLINE_SLOTS, + // The wire carries the deadline as a slot, so the wall-clock + // budget converts at the target slot time. + deadline_slot: auction.tip + + (self.solve_deadline.as_millis() / SLOT_DURATION.as_millis()) as u64, orders: auction.orders.iter().map(dto::Order::from).collect(), }; let by_uid: HashMap = auction @@ -57,7 +59,7 @@ impl SolverCompetition for DriverCompetition { // The deadline sent to the drivers is also enforced here: a response // landing after it is dropped, so one hung driver cannot hold the // loop for the HTTP client's whole ceiling. - let budget = SLOT_DURATION * u32::try_from(SOLVE_DEADLINE_SLOTS).expect("small constant"); + let budget = self.solve_deadline; let responses = futures::future::join_all(self.drivers.iter().enumerate().map( |(driver_index, driver)| async move { match tokio::time::timeout(budget, driver.solve(request)).await { diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index a8471828fa..0c61bba383 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -25,7 +25,7 @@ use { chain_types::solana::{IntentHash, Pubkey, Signature}, database::byte_array::ByteArray, sqlx::PgPool, - std::{collections::HashMap, net::SocketAddr, sync::Arc}, + std::{collections::HashMap, net::SocketAddr, sync::Arc, time::Duration}, tokio::sync::mpsc, url::Url, }; @@ -156,7 +156,7 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { let provider = DbAuctionProvider::new(pool.clone()); let auction = provider.cut_auction(&tip).await.expect("auction cut"); assert_eq!(auction.orders.len(), 1, "open order in the auction"); - let competition = DriverCompetition::new(vec![Arc::clone(&driver)]); + let competition = DriverCompetition::new(vec![Arc::clone(&driver)], Duration::from_secs(6)); let solutions = competition.solve(&auction).await; assert_eq!(solutions.len(), 1, "driver solution converted"); let ranking = SolanaArbitrator::new(1, wrapped_native).arbitrate(solutions, &auction); @@ -166,14 +166,17 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { let mut auction_loop = AuctionLoop::new( Box::new(FixedTrigger(tip)), Box::new(DbAuctionProvider::new(pool)), - Box::new(DriverCompetition::new(vec![Arc::clone(&driver)])), + Box::new(DriverCompetition::new( + vec![Arc::clone(&driver)], + Duration::from_secs(6), + )), Box::new(SolanaArbitrator::new(1, wrapped_native)), Box::new(DriverExecutor::new(vec![driver])), Box::new(LogObserver), ); auction_loop.run_cycle().await; - let settle = tokio::time::timeout(std::time::Duration::from_secs(5), settled.recv()) + let settle = tokio::time::timeout(Duration::from_secs(5), settled.recv()) .await .expect("settlement dispatched before the timeout") .expect("settle channel open"); From 6e3eb7b69994758185428e6ff3947b0cbd95a247 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 20 Aug 2026 09:09:58 +0000 Subject: [PATCH 04/12] Carry a wall-clock solve deadline and a settle submission deadline slot on the driver wire --- Cargo.lock | 1 + crates/autopilot-svm/Cargo.toml | 1 + crates/autopilot-svm/src/infra/competition.rs | 8 +------ crates/autopilot-svm/src/infra/driver/dto.rs | 22 +++++++++++++++---- crates/autopilot-svm/src/infra/executor.rs | 1 + 5 files changed, 22 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b197897896..a7b7fc765a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1592,6 +1592,7 @@ dependencies = [ "axum", "bigdecimal", "chain-types", + "chrono", "const-hex", "cow-solana-rpc", "database", diff --git a/crates/autopilot-svm/Cargo.toml b/crates/autopilot-svm/Cargo.toml index 0e30dd042c..63d83ab691 100644 --- a/crates/autopilot-svm/Cargo.toml +++ b/crates/autopilot-svm/Cargo.toml @@ -9,6 +9,7 @@ anyhow = { workspace = true } async-trait = { workspace = true } bigdecimal = { workspace = true } chain-types = { workspace = true } +chrono = { workspace = true, features = ["serde"] } const-hex = { workspace = true } cow-solana-rpc = { workspace = true } database = { workspace = true } diff --git a/crates/autopilot-svm/src/infra/competition.rs b/crates/autopilot-svm/src/infra/competition.rs index bcf587386a..b580a334cf 100644 --- a/crates/autopilot-svm/src/infra/competition.rs +++ b/crates/autopilot-svm/src/infra/competition.rs @@ -19,9 +19,6 @@ use { winner_selection::{Side, solution}, }; -/// Solana's target slot time. -const SLOT_DURATION: Duration = Duration::from_millis(400); - /// Sends `/solve` to every driver and converts the answers into attributable /// solutions for the arbitrator. pub struct DriverCompetition { @@ -44,10 +41,7 @@ impl SolverCompetition for DriverCompetition { async fn solve(&self, auction: &Auction) -> Vec { let request = &dto::SolveRequest { id: auction.id, - // The wire carries the deadline as a slot, so the wall-clock - // budget converts at the target slot time. - deadline_slot: auction.tip - + (self.solve_deadline.as_millis() / SLOT_DURATION.as_millis()) as u64, + deadline: chrono::Utc::now() + self.solve_deadline, orders: auction.orders.iter().map(dto::Order::from).collect(), }; let by_uid: HashMap = auction diff --git a/crates/autopilot-svm/src/infra/driver/dto.rs b/crates/autopilot-svm/src/infra/driver/dto.rs index ad1f510c0f..0a173e03ca 100644 --- a/crates/autopilot-svm/src/infra/driver/dto.rs +++ b/crates/autopilot-svm/src/infra/driver/dto.rs @@ -19,8 +19,8 @@ use { pub struct SolveRequest { /// Autopilot-assigned auction id. pub id: i64, - /// Slot after which a settlement for this auction is late. - pub deadline_slot: u64, + /// Wall-clock deadline for answering `/solve`. + pub deadline: chrono::DateTime, pub orders: Vec, } @@ -125,6 +125,8 @@ pub struct TradedAmounts { pub struct SettleRequest { pub auction_id: i64, pub solution_id: u64, + /// The last slot the settlement transaction may land in. + pub submission_deadline_slot: u64, } /// The driver's `/settle` answer. @@ -165,11 +167,11 @@ mod tests { fn dtos_round_trip_and_pin_the_wire_format() { let request = SolveRequest { id: 7, - deadline_slot: 100, + deadline: "2026-01-01T00:00:00Z".parse().unwrap(), orders: vec![Order::from(&order())], }; let json = serde_json::to_value(&request).unwrap(); - assert_eq!(json["deadlineSlot"], 100); + assert_eq!(json["deadline"], "2026-01-01T00:00:00Z"); assert_eq!( json["orders"][0]["uid"], "0x1111111111111111111111111111111111111111111111111111111111111111" @@ -217,5 +219,17 @@ mod tests { serde_json::from_value::(json).unwrap(), settle ); + + let settle_request = SettleRequest { + auction_id: 7, + solution_id: 3, + submission_deadline_slot: 125, + }; + let json = serde_json::to_value(&settle_request).unwrap(); + assert_eq!(json["submissionDeadlineSlot"], 125); + assert_eq!( + serde_json::from_value::(json).unwrap(), + settle_request + ); } } diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs index e3ff1671c9..13acde572a 100644 --- a/crates/autopilot-svm/src/infra/executor.rs +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -47,6 +47,7 @@ impl SettlementExecutor for DriverExecutor { let request = dto::SettleRequest { auction_id, solution_id: winner.id(), + submission_deadline_slot: deadline, }; tokio::spawn(async move { match driver.settle(&request).await { From fcb2ddd6a45a9ff4d7e8127f02dc8f1fb297e670 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 20 Aug 2026 09:44:04 +0000 Subject: [PATCH 05/12] Drop the auction tip field, the wall-clock deadline removed its last reader --- crates/autopilot-svm/src/domain/auction.rs | 13 ++----------- crates/autopilot-svm/src/infra/db.rs | 6 +++--- crates/autopilot-svm/src/infra/provider.rs | 4 ++-- 3 files changed, 7 insertions(+), 16 deletions(-) diff --git a/crates/autopilot-svm/src/domain/auction.rs b/crates/autopilot-svm/src/domain/auction.rs index 1703201090..c7c871d717 100644 --- a/crates/autopilot-svm/src/domain/auction.rs +++ b/crates/autopilot-svm/src/domain/auction.rs @@ -39,9 +39,6 @@ pub struct Auction { /// Autopilot-assigned id. Excluded from equality: the dedupe compares two /// cuts by content, and the id is allocated only for a fresh cut. pub id: i64, - /// Slot the auction was cut at. Excluded from equality for the same - /// reason: the tip moves every cycle, the content may not. - pub tip: u64, pub orders: Vec, } @@ -79,24 +76,18 @@ mod tests { } #[test] - fn auction_equality_ignores_id_and_tip() { + fn auction_equality_ignores_id() { let orders = vec![order(10)]; let a = Auction { id: 1, - tip: 100, orders: orders.clone(), }; - let b = Auction { - id: 2, - tip: 200, - orders, - }; + let b = Auction { id: 2, orders }; assert_eq!(a, b); assert_ne!( a, Auction { id: 1, - tip: 100, orders: vec![], } ); diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index f37833b9db..272d25bfaa 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -102,10 +102,10 @@ ORDER BY slot, tx_signature .context("read solana.settlements by auction") } -/// Cut an auction at the given tip from the open orders. -pub async fn cut(ex: impl PgExecutor<'_>, id: i64, tip: u64, now_unix: i64) -> Result { +/// Cut an auction from the open orders. +pub async fn cut(ex: impl PgExecutor<'_>, id: i64, now_unix: i64) -> Result { let orders = orders_from_rows(open_orders(ex, now_unix).await?); - Ok(Auction { id, tip, orders }) + Ok(Auction { id, orders }) } /// A row the indexer wrote always converts (on-chain values fit the domain diff --git a/crates/autopilot-svm/src/infra/provider.rs b/crates/autopilot-svm/src/infra/provider.rs index 2a4a4f75e3..ae0bd47724 100644 --- a/crates/autopilot-svm/src/infra/provider.rs +++ b/crates/autopilot-svm/src/infra/provider.rs @@ -59,9 +59,9 @@ impl AuctionProvider for DbAuctionProvider { Ok(()) } - async fn cut_auction(&self, tip: &u64) -> Option { + async fn cut_auction(&self, _tip: &u64) -> Option { let now = now_unix(); - let auction = db::cut(&self.pool, self.next_id(now), *tip, now) + let auction = db::cut(&self.pool, self.next_id(now), now) .await .map_err(|err| tracing::warn!(?err, "failed to cut the auction")) .ok()?; From d8c06b6545a3925cb6ae951b45613834e9ecafc1 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:30:03 +0000 Subject: [PATCH 06/12] Name the window operations after what they do --- crates/autopilot-svm/src/infra/executor.rs | 5 ++- crates/autopilot-svm/src/infra/observation.rs | 42 ++++++++++--------- crates/autopilot-svm/src/infra/observer.rs | 2 +- 3 files changed, 26 insertions(+), 23 deletions(-) diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs index 596a1b3338..f3a1c87d1f 100644 --- a/crates/autopilot-svm/src/infra/executor.rs +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -21,7 +21,8 @@ const SUBMISSION_DEADLINE_SLOTS: u64 = 25; /// loop starts the next cycle while settlements land. pub struct DriverExecutor { drivers: Vec>, - /// Dispatched settlements the observation side resolves or times out. + /// Opens a settlement-execution window per dispatched settlement, which + /// the observation side later resolves or times out. tracker: SettlementTracker, } @@ -58,7 +59,7 @@ impl SettlementExecutor for DriverExecutor { // the dispatch is the priority. if let Err(err) = self .tracker - .dispatched( + .open_dispatched_settlement_window( auction_id, winner.solver(), winner.id(), diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index cc46e2928a..fd0f700dd5 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -20,11 +20,13 @@ use { /// A closed window's outcome, the `solana.settlement_executions.outcome` /// value. The outcome records what the indexer observed on chain, never a -/// driver's report, which is why a landing observed after the deadline may -/// overwrite a timeout. The schema also allows `rejected`, unused until -/// typed `/settle` errors are consumed. +/// driver's report. The schema also allows `rejected`, unused until typed +/// `/settle` errors are consumed. enum Outcome { + /// The settlement was observed landed on chain. Landed, + /// The deadline passed with no observed settlement. A landing observed + /// later still overwrites this, chain truth wins over the sweep's timing. Timeout, } @@ -51,7 +53,7 @@ impl SettlementTracker { /// Open a window for a dispatched settlement. `solution_uid` is the /// winner's driver-local solution id until competition persistence /// allocates uids. - pub async fn dispatched( + pub async fn open_dispatched_settlement_window( &self, auction_id: i64, solver: Pubkey, @@ -69,9 +71,9 @@ ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING ) .bind(auction_id) .bind(solver.0) - .bind(to_db_u64(solution_uid)) - .bind(to_db_u64(start_slot)) - .bind(to_db_u64(deadline_slot)) + .bind(to_db_integer(solution_uid)) + .bind(to_db_integer(start_slot)) + .bind(to_db_integer(deadline_slot)) .execute(&self.pool) .await .context("open settlement execution window")?; @@ -83,7 +85,7 @@ ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING /// of one auction each close on their own settlement. A window already /// closed as timed out upgrades to landed: the settlement executed, just /// late, and lateness stays visible as `end_slot` past `deadline_slot`. - async fn landed( + async fn close_solvers_window_as_landed( &self, auction_id: i64, solver: &[u8], @@ -114,7 +116,7 @@ WHERE auction_id = $1 AND solver = $6 AND (outcome IS NULL OR outcome = $5) /// timed out, logging each. Driven by the competition cycle, so on a /// chain with no active auctions a timeout surfaces with the next /// competition, not at its deadline slot. - pub async fn flag_expired(&self, slot: u64) -> Result<()> { + pub async fn close_expired_windows_as_timeout(&self, slot: u64) -> Result<()> { let expired: Vec<(i64,)> = sqlx::query_as( r#" UPDATE solana.settlement_executions @@ -123,7 +125,7 @@ WHERE outcome IS NULL AND deadline_slot <= $1 RETURNING auction_id "#, ) - .bind(to_db_u64(slot)) + .bind(to_db_integer(slot)) .bind(Outcome::Timeout.as_str()) .fetch_all(&self.pool) .await @@ -135,7 +137,7 @@ RETURNING auction_id } /// Auction ids of open windows, the seed re-read set. - async fn open_windows(&self) -> Result> { + async fn open_window_auction_ids(&self) -> Result> { let rows: Vec<(i64,)> = sqlx::query_as( "SELECT auction_id FROM solana.settlement_executions WHERE outcome IS NULL", ) @@ -146,8 +148,8 @@ RETURNING auction_id } } -/// Slots and solution ids stay far below `i64::MAX`. -fn to_db_u64(value: u64) -> i64 { +/// Slots and solution ids stay far below `i64::MAX`, the column type. +fn to_db_integer(value: u64) -> i64 { i64::try_from(value).expect("value exceeds i64") } @@ -169,7 +171,7 @@ impl SettlementObservation { for settlement in db::settlements_by_auction(&self.pool, auction_id).await? { let closed = self .tracker - .landed( + .close_solvers_window_as_landed( auction_id, &settlement.solver.0, settlement.slot, @@ -194,7 +196,7 @@ impl NotifyHandler for SettlementObservation { /// Re-check every open window: a NOTIFY missed while the connection was /// down (or while the autopilot was not running) is recovered here. async fn seed(&mut self) -> Result<()> { - for auction_id in self.tracker.open_windows().await? { + for auction_id in self.tracker.open_window_auction_ids().await? { self.resolve(auction_id).await?; } Ok(()) @@ -266,7 +268,7 @@ VALUES (10, $1, 0, $2, $3, NULL) let tracker = SettlementTracker::new(pool.clone()); tracker - .dispatched(4242, Pubkey([7; 32]), 1, 90, 100) + .open_dispatched_settlement_window(4242, Pubkey([7; 32]), 1, 90, 100) .await .unwrap(); @@ -305,22 +307,22 @@ VALUES (10, $1, 0, $2, $3, NULL) let tracker = SettlementTracker::new(pool.clone()); tracker - .dispatched(1, Pubkey([7; 32]), 1, 90, 100) + .open_dispatched_settlement_window(1, Pubkey([7; 32]), 1, 90, 100) .await .unwrap(); tracker - .dispatched(2, Pubkey([7; 32]), 1, 90, 200) + .open_dispatched_settlement_window(2, Pubkey([7; 32]), 1, 90, 200) .await .unwrap(); - tracker.flag_expired(150).await.unwrap(); + tracker.close_expired_windows_as_timeout(150).await.unwrap(); assert_eq!(outcome(&pool, 1).await.as_deref(), Some("timeout")); assert_eq!(outcome(&pool, 2).await, None); // A settlement observed after the timeout upgrades the verdict: it // executed, just late. tracker - .landed(1, &[7u8; 32], 160, &[9u8; 64]) + .close_solvers_window_as_landed(1, &[7u8; 32], 160, &[9u8; 64]) .await .unwrap(); assert_eq!(outcome(&pool, 1).await.as_deref(), Some("landed")); diff --git a/crates/autopilot-svm/src/infra/observer.rs b/crates/autopilot-svm/src/infra/observer.rs index abbdbcd494..279c135d68 100644 --- a/crates/autopilot-svm/src/infra/observer.rs +++ b/crates/autopilot-svm/src/infra/observer.rs @@ -44,7 +44,7 @@ impl SettlementObserver for LogObserver { ) -> anyhow::Result<()> { // Best effort: the expiry bookkeeping only touches previously // dispatched windows and must not block the current dispatch. - if let Err(err) = self.tracker.flag_expired(*tip).await { + if let Err(err) = self.tracker.close_expired_windows_as_timeout(*tip).await { tracing::error!(?err, "failed to flag expired settlement windows"); } tracing::info!( From 9b8ebef9d6c1640d023ac8dbff89a9ebe2fa0a62 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:50:49 +0000 Subject: [PATCH 07/12] Write the settlement outcomes as literals --- crates/autopilot-svm/src/infra/observation.rs | 35 +++++-------------- 1 file changed, 8 insertions(+), 27 deletions(-) diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index fd0f700dd5..302e7544dd 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -18,28 +18,12 @@ use { sqlx::PgPool, }; -/// A closed window's outcome, the `solana.settlement_executions.outcome` -/// value. The outcome records what the indexer observed on chain, never a -/// driver's report. The schema also allows `rejected`, unused until typed -/// `/settle` errors are consumed. -enum Outcome { - /// The settlement was observed landed on chain. - Landed, - /// The deadline passed with no observed settlement. A landing observed - /// later still overwrites this, chain truth wins over the sweep's timing. - Timeout, -} - -impl Outcome { - fn as_str(&self) -> &'static str { - match self { - Self::Landed => "landed", - Self::Timeout => "timeout", - } - } -} - /// Writes and closes settlement-execution windows. +/// +/// `solana.settlement_executions.outcome` records what the indexer observed +/// on chain, never a driver's report, which is why a landing observed after +/// the deadline overwrites a timeout. The schema also allows `rejected`, +/// unused until typed `/settle` errors are consumed. #[derive(Clone)] pub struct SettlementTracker { pool: PgPool, @@ -95,15 +79,13 @@ ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING let closed = sqlx::query( r#" UPDATE solana.settlement_executions -SET outcome = $4, end_timestamp = now(), end_slot = $2, submitted_signature = $3 -WHERE auction_id = $1 AND solver = $6 AND (outcome IS NULL OR outcome = $5) +SET outcome = 'landed', end_timestamp = now(), end_slot = $2, submitted_signature = $3 +WHERE auction_id = $1 AND solver = $4 AND (outcome IS NULL OR outcome = 'timeout') "#, ) .bind(auction_id) .bind(slot) .bind(signature) - .bind(Outcome::Landed.as_str()) - .bind(Outcome::Timeout.as_str()) .bind(solver) .execute(&self.pool) .await @@ -120,13 +102,12 @@ WHERE auction_id = $1 AND solver = $6 AND (outcome IS NULL OR outcome = $5) let expired: Vec<(i64,)> = sqlx::query_as( r#" UPDATE solana.settlement_executions -SET outcome = $2, end_timestamp = now(), end_slot = $1 +SET outcome = 'timeout', end_timestamp = now(), end_slot = $1 WHERE outcome IS NULL AND deadline_slot <= $1 RETURNING auction_id "#, ) .bind(to_db_integer(slot)) - .bind(Outcome::Timeout.as_str()) .fetch_all(&self.pool) .await .context("expire settlement execution windows")?; From bc786bdef89b0455b71acff7f490f0fe3149c987 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:34:38 +0000 Subject: [PATCH 08/12] State the settlement attribution constraint --- crates/autopilot-svm/src/infra/observation.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index 302e7544dd..fd79467ae7 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -69,6 +69,10 @@ ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING /// of one auction each close on their own settlement. A window already /// closed as timed out upgrades to landed: the settlement executed, just /// late, and lateness stays visible as `end_slot` past `deadline_slot`. + /// + /// A settlement carries no solution uid, so a solver holding several + /// windows of one auction closes all of them on its first settlement. + /// Correct while one solver wins at most one solution per auction. async fn close_solvers_window_as_landed( &self, auction_id: i64, From 6df9bcd9dfd46b7bdb1b9c48485705f5b91221a7 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:41:48 +0000 Subject: [PATCH 09/12] Move the settlement window queries into the db layer --- crates/autopilot-svm/src/infra/db.rs | 89 +++++++++++ crates/autopilot-svm/src/infra/observation.rs | 149 ++++-------------- crates/autopilot-svm/src/lib.rs | 2 + crates/autopilot-svm/src/test_db.rs | 19 +++ crates/autopilot-svm/src/tests.rs | 8 +- 5 files changed, 141 insertions(+), 126 deletions(-) create mode 100644 crates/autopilot-svm/src/test_db.rs diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index 272d25bfaa..75845d65cc 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -102,6 +102,95 @@ ORDER BY slot, tx_signature .context("read solana.settlements by auction") } +/// A window closed as landed, for the caller's log line. +#[derive(Clone, Debug, sqlx::FromRow)] +pub struct LandedWindow { + pub solver: ByteArray<32>, + pub end_slot: i64, + pub submitted_signature: ByteArray<64>, +} + +/// Open a settlement-execution window for a dispatched settlement. +pub async fn open_settlement_window( + ex: impl PgExecutor<'_>, + auction_id: i64, + solver: Pubkey, + solution_uid: i64, + start_slot: i64, + deadline_slot: i64, +) -> Result<()> { + const QUERY: &str = r#" +INSERT INTO solana.settlement_executions + (auction_id, solver, solution_uid, start_timestamp, start_slot, deadline_slot) +VALUES ($1, $2, $3, now(), $4, $5) +ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING + "#; + sqlx::query(QUERY) + .bind(auction_id) + .bind(solver.0) + .bind(solution_uid) + .bind(start_slot) + .bind(deadline_slot) + .execute(ex) + .await + .context("open settlement execution window")?; + Ok(()) +} + +/// Close the auction's windows against the settlements the indexer recorded, +/// matching each window to its solver's settlement. A window already closed +/// as timed out upgrades to landed: the settlement executed, just late, and +/// lateness stays visible as `end_slot` past `deadline_slot`. +/// +/// A settlement carries no solution uid, so a solver holding several windows +/// of one auction closes all of them on its first settlement. Correct while +/// one solver wins at most one solution per auction. +pub async fn close_landed_windows( + ex: impl PgExecutor<'_>, + auction_id: i64, +) -> Result> { + const QUERY: &str = r#" +UPDATE solana.settlement_executions e +SET outcome = 'landed', end_timestamp = now(), end_slot = s.slot, + submitted_signature = s.tx_signature +FROM solana.settlements s +WHERE e.auction_id = $1 + AND s.auction_id = e.auction_id + AND s.solver = e.solver + AND (e.outcome IS NULL OR e.outcome = 'timeout') +RETURNING e.solver, e.end_slot, e.submitted_signature + "#; + sqlx::query_as(QUERY) + .bind(auction_id) + .fetch_all(ex) + .await + .context("close landed settlement execution windows") +} + +/// Close every open window whose deadline is at or before the slot as timed +/// out, returning their auction ids. +pub async fn expire_settlement_windows(ex: impl PgExecutor<'_>, slot: i64) -> Result> { + const QUERY: &str = r#" +UPDATE solana.settlement_executions +SET outcome = 'timeout', end_timestamp = now(), end_slot = $1 +WHERE outcome IS NULL AND deadline_slot <= $1 +RETURNING auction_id + "#; + sqlx::query_scalar(QUERY) + .bind(slot) + .fetch_all(ex) + .await + .context("expire settlement execution windows") +} + +/// Auction ids of open windows, the seed re-read set. +pub async fn open_window_auction_ids(ex: impl PgExecutor<'_>) -> Result> { + sqlx::query_scalar("SELECT auction_id FROM solana.settlement_executions WHERE outcome IS NULL") + .fetch_all(ex) + .await + .context("read open settlement execution windows") +} + /// Cut an auction from the open orders. pub async fn cut(ex: impl PgExecutor<'_>, id: i64, now_unix: i64) -> Result { let orders = orders_from_rows(open_orders(ex, now_unix).await?); diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index fd79467ae7..94d81be2dd 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -12,13 +12,13 @@ use { crate::infra::{db, listen::NotifyHandler}, - anyhow::{Context, Result}, + anyhow::Result, async_trait::async_trait, chain_types::solana::Pubkey, sqlx::PgPool, }; -/// Writes and closes settlement-execution windows. +/// Opens and expires settlement-execution windows. /// /// `solana.settlement_executions.outcome` records what the indexer observed /// on chain, never a driver's report, which is why a landing observed after @@ -45,57 +45,15 @@ impl SettlementTracker { start_slot: u64, deadline_slot: u64, ) -> Result<()> { - sqlx::query( - r#" -INSERT INTO solana.settlement_executions - (auction_id, solver, solution_uid, start_timestamp, start_slot, deadline_slot) -VALUES ($1, $2, $3, now(), $4, $5) -ON CONFLICT (auction_id, solver, solution_uid) DO NOTHING - "#, + db::open_settlement_window( + &self.pool, + auction_id, + solver, + to_db_integer(solution_uid), + to_db_integer(start_slot), + to_db_integer(deadline_slot), ) - .bind(auction_id) - .bind(solver.0) - .bind(to_db_integer(solution_uid)) - .bind(to_db_integer(start_slot)) - .bind(to_db_integer(deadline_slot)) - .execute(&self.pool) .await - .context("open settlement execution window")?; - Ok(()) - } - - /// Close the solver's windows of the auction as landed, recording the - /// settlement's slot and signature. Keyed by solver so concurrent winners - /// of one auction each close on their own settlement. A window already - /// closed as timed out upgrades to landed: the settlement executed, just - /// late, and lateness stays visible as `end_slot` past `deadline_slot`. - /// - /// A settlement carries no solution uid, so a solver holding several - /// windows of one auction closes all of them on its first settlement. - /// Correct while one solver wins at most one solution per auction. - async fn close_solvers_window_as_landed( - &self, - auction_id: i64, - solver: &[u8], - slot: i64, - signature: &[u8], - ) -> Result { - let closed = sqlx::query( - r#" -UPDATE solana.settlement_executions -SET outcome = 'landed', end_timestamp = now(), end_slot = $2, submitted_signature = $3 -WHERE auction_id = $1 AND solver = $4 AND (outcome IS NULL OR outcome = 'timeout') - "#, - ) - .bind(auction_id) - .bind(slot) - .bind(signature) - .bind(solver) - .execute(&self.pool) - .await - .context("close settlement execution window")? - .rows_affected(); - Ok(closed) } /// Close every open window whose deadline is at or before the slot as @@ -103,34 +61,12 @@ WHERE auction_id = $1 AND solver = $4 AND (outcome IS NULL OR outcome = 'timeout /// chain with no active auctions a timeout surfaces with the next /// competition, not at its deadline slot. pub async fn close_expired_windows_as_timeout(&self, slot: u64) -> Result<()> { - let expired: Vec<(i64,)> = sqlx::query_as( - r#" -UPDATE solana.settlement_executions -SET outcome = 'timeout', end_timestamp = now(), end_slot = $1 -WHERE outcome IS NULL AND deadline_slot <= $1 -RETURNING auction_id - "#, - ) - .bind(to_db_integer(slot)) - .fetch_all(&self.pool) - .await - .context("expire settlement execution windows")?; - for (auction_id,) in expired { + let slot = to_db_integer(slot); + for auction_id in db::expire_settlement_windows(&self.pool, slot).await? { tracing::error!(auction_id, slot, "settlement missed its deadline"); } Ok(()) } - - /// Auction ids of open windows, the seed re-read set. - async fn open_window_auction_ids(&self) -> Result> { - let rows: Vec<(i64,)> = sqlx::query_as( - "SELECT auction_id FROM solana.settlement_executions WHERE outcome IS NULL", - ) - .fetch_all(&self.pool) - .await - .context("read open settlement execution windows")?; - Ok(rows.into_iter().map(|(auction_id,)| auction_id).collect()) - } } /// Slots and solution ids stay far below `i64::MAX`, the column type. @@ -138,39 +74,27 @@ fn to_db_integer(value: u64) -> i64 { i64::try_from(value).expect("value exceeds i64") } -/// Resolves NOTIFY payloads (and reconnect re-reads) against the settlements -/// table, closing the matching windows. +/// Closes windows against the settlements the indexer records, driven by the +/// `solana_settlement_finalized` notifications. pub struct SettlementObservation { pool: PgPool, - tracker: SettlementTracker, } impl SettlementObservation { - pub fn new(pool: PgPool, tracker: SettlementTracker) -> Self { - Self { pool, tracker } + pub fn new(pool: PgPool) -> Self { + Self { pool } } - /// Close the matching windows for every settlement observed for the - /// auction, each keyed to its solver. + /// Close the auction's windows against its observed settlements. async fn resolve(&self, auction_id: i64) -> Result<()> { - for settlement in db::settlements_by_auction(&self.pool, auction_id).await? { - let closed = self - .tracker - .close_solvers_window_as_landed( - auction_id, - &settlement.solver.0, - settlement.slot, - &settlement.tx_signature.0, - ) - .await?; - if closed > 0 { - tracing::info!( - auction_id, - slot = settlement.slot, - tx_signature = %const_hex::encode(settlement.tx_signature.0), - "settlement observed on chain" - ); - } + for landed in db::close_landed_windows(&self.pool, auction_id).await? { + tracing::info!( + auction_id, + slot = landed.end_slot, + solver = %Pubkey(landed.solver.0), + tx_signature = %const_hex::encode(landed.submitted_signature.0), + "settlement observed on chain" + ); } Ok(()) } @@ -181,7 +105,7 @@ impl NotifyHandler for SettlementObservation { /// Re-check every open window: a NOTIFY missed while the connection was /// down (or while the autopilot was not running) is recovered here. async fn seed(&mut self) -> Result<()> { - for auction_id in self.tracker.open_window_auction_ids().await? { + for auction_id in db::open_window_auction_ids(&self.pool).await? { self.resolve(auction_id).await?; } Ok(()) @@ -206,19 +130,6 @@ mod tests { std::time::Duration, }; - async fn wipe(pool: &PgPool) { - for table in [ - "solana.trades", - "solana.settlements", - "solana.settlement_executions", - ] { - sqlx::query(&format!("DELETE FROM {table}")) - .execute(pool) - .await - .unwrap(); - } - } - async fn insert_settlement(pool: &PgPool, auction_id: i64) { sqlx::query( r#" @@ -249,7 +160,7 @@ VALUES (10, $1, 0, $2, $3, NULL) #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] async fn solana_db_settlement_notify_closes_the_window_as_landed() { let pool = PgPool::connect("postgresql://").await.unwrap(); - wipe(&pool).await; + crate::test_db::wipe(&pool).await; let tracker = SettlementTracker::new(pool.clone()); tracker @@ -260,7 +171,7 @@ VALUES (10, $1, 0, $2, $3, NULL) let task = ListenSession::spawn( pool.clone(), "solana_settlement_finalized", - SettlementObservation::new(pool.clone(), tracker.clone()), + SettlementObservation::new(pool.clone()), ); insert_settlement(&pool, 4242).await; @@ -288,7 +199,7 @@ VALUES (10, $1, 0, $2, $3, NULL) #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] async fn solana_db_expiry_times_out_only_past_deadlines() { let pool = PgPool::connect("postgresql://").await.unwrap(); - wipe(&pool).await; + crate::test_db::wipe(&pool).await; let tracker = SettlementTracker::new(pool.clone()); tracker @@ -306,8 +217,8 @@ VALUES (10, $1, 0, $2, $3, NULL) // A settlement observed after the timeout upgrades the verdict: it // executed, just late. - tracker - .close_solvers_window_as_landed(1, &[7u8; 32], 160, &[9u8; 64]) + insert_settlement(&pool, 1).await; + crate::infra::db::close_landed_windows(&pool, 1) .await .unwrap(); assert_eq!(outcome(&pool, 1).await.as_deref(), Some("landed")); diff --git a/crates/autopilot-svm/src/lib.rs b/crates/autopilot-svm/src/lib.rs index b86fc0ec35..7561686f6a 100644 --- a/crates/autopilot-svm/src/lib.rs +++ b/crates/autopilot-svm/src/lib.rs @@ -7,5 +7,7 @@ pub mod run_loop; mod domain; mod infra; +#[cfg(test)] +mod test_db; #[cfg(test)] mod tests; diff --git a/crates/autopilot-svm/src/test_db.rs b/crates/autopilot-svm/src/test_db.rs new file mode 100644 index 0000000000..52de984615 --- /dev/null +++ b/crates/autopilot-svm/src/test_db.rs @@ -0,0 +1,19 @@ +//! Database helpers for the crate's `solana_db_` tests. + +use sqlx::PgPool; + +pub(crate) async fn pool() -> PgPool { + PgPool::connect("postgresql://").await.unwrap() +} + +/// Empty every `solana.*` table a test can touch. Tests wipe at the start so +/// a failure leaves its state behind for inspection. +pub(crate) async fn wipe(pool: &PgPool) { + sqlx::query( + "TRUNCATE solana.trades, solana.settlements, solana.settlement_executions, \ + solana.order_pda, solana.orders, solana.indexer_state", + ) + .execute(pool) + .await + .unwrap(); +} diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index c3b453dff1..cf81a480f6 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -86,13 +86,7 @@ async fn spawn_mock_driver(state: MockDriverState) -> SocketAddr { } async fn seed_open_order(pool: &PgPool, uid: [u8; 32], tip: i64) { - sqlx::query( - "TRUNCATE solana.trades, solana.settlements, solana.settlement_executions, \ - solana.order_pda, solana.orders, solana.indexer_state", - ) - .execute(pool) - .await - .unwrap(); + crate::test_db::wipe(pool).await; sqlx::query("INSERT INTO solana.indexer_state (slot) VALUES ($1)") .bind(tip) .execute(pool) From ff84a2274647f67cd765790454e7c79e4bbdd01d Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:43:30 +0000 Subject: [PATCH 10/12] Drop the settlement read the single-query close replaced --- crates/autopilot-svm/src/infra/db.rs | 65 +--------------------------- 1 file changed, 1 insertion(+), 64 deletions(-) diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index 75845d65cc..98d424c921 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -59,20 +59,6 @@ ORDER BY o.uid .context("read open solana.orders") } -/// A row of `solana.settlements`, the indexer's record of one settlement -/// transaction. The transaction carries no solution uid, the indexer -/// attributes it from the recorded competition, so `solution_uid` is `None` -/// for settlements it cannot match. A settlement is finalized once its slot -/// is at or below `solana.indexer_state.finalized_slot`. -#[derive(Clone, Debug, sqlx::FromRow)] -pub struct Settlement { - pub slot: i64, - pub tx_signature: ByteArray<64>, - pub solver: ByteArray<32>, - pub auction_id: i64, - pub solution_uid: Option, -} - /// Latest slot the indexer fully processed. `None` before the indexer's first /// write. `solana.indexer_state` is a single-row table. pub async fn last_indexed_slot(ex: impl PgExecutor<'_>) -> Result> { @@ -83,25 +69,6 @@ pub async fn last_indexed_slot(ex: impl PgExecutor<'_>) -> Result> { .context("read solana.indexer_state slot") } -/// Settlements the indexer recorded for an auction. More than one row when the -/// auction had several winners. -pub async fn settlements_by_auction( - ex: impl PgExecutor<'_>, - auction_id: i64, -) -> Result> { - const QUERY: &str = r#" -SELECT slot, tx_signature, solver, auction_id, solution_uid -FROM solana.settlements -WHERE auction_id = $1 -ORDER BY slot, tx_signature - "#; - sqlx::query_as(QUERY) - .bind(auction_id) - .fetch_all(ex) - .await - .context("read solana.settlements by auction") -} - /// A window closed as landed, for the caller's log line. #[derive(Clone, Debug, sqlx::FromRow)] pub struct LandedWindow { @@ -249,7 +216,7 @@ fn to_amount(value: &BigDecimal) -> Result { #[cfg(test)] mod tests { use { - super::{last_indexed_slot, open_orders, settlements_by_auction}, + super::{last_indexed_slot, open_orders}, bigdecimal::BigDecimal, database::byte_array::ByteArray, sqlx::{PgPool, PgTransaction}, @@ -410,34 +377,4 @@ VALUES ($1, $2, CASE WHEN $3 THEN now() END, $4, $5) .unwrap(); assert_eq!(last_indexed_slot(&mut *tx).await.unwrap(), Some(42)); } - - #[tokio::test] - #[ignore = "needs the solana.* schema applied to the local database"] - async fn solana_db_settlements_by_auction_roundtrip() { - let pool = PgPool::connect("postgresql://").await.unwrap(); - let mut tx = pool.begin().await.unwrap(); - - for table in ["trades", "settlements"] { - sqlx::query(&format!("DELETE FROM solana.{table}")) - .execute(&mut *tx) - .await - .unwrap(); - } - sqlx::query( - r#" -INSERT INTO solana.settlements (slot, tx_signature, instruction_index, solver, auction_id, solution_uid) -VALUES (7, $1, 0, $2, 123, NULL) - "#, - ) - .bind(ByteArray([9u8; 64])) - .bind(ByteArray([10u8; 32])) - .execute(&mut *tx) - .await - .unwrap(); - let settlements = settlements_by_auction(&mut *tx, 123).await.unwrap(); - assert_eq!(settlements.len(), 1); - assert_eq!(settlements[0].auction_id, 123); - assert_eq!(settlements[0].solution_uid, None); - assert_eq!(settlements[0].slot, 7); - } } From b7d40da16156a5a7915b3d121bf6c6b6f8f2f7b7 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:53:10 +0000 Subject: [PATCH 11/12] Fold the settlement window types into one --- crates/autopilot-svm/src/infra/executor.rs | 12 +-- crates/autopilot-svm/src/infra/observation.rs | 85 ++++++++----------- crates/autopilot-svm/src/infra/observer.rs | 10 +-- crates/autopilot-svm/src/tests.rs | 8 +- 4 files changed, 52 insertions(+), 63 deletions(-) diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs index f3a1c87d1f..84dcf07785 100644 --- a/crates/autopilot-svm/src/infra/executor.rs +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -5,7 +5,7 @@ use { domain::cycle::{Ranking, SolanaCycle}, infra::{ driver::{Driver, dto}, - observation::SettlementTracker, + observation::SettlementWindows, }, run_loop::SettlementExecutor, }, @@ -23,12 +23,12 @@ pub struct DriverExecutor { drivers: Vec>, /// Opens a settlement-execution window per dispatched settlement, which /// the observation side later resolves or times out. - tracker: SettlementTracker, + windows: SettlementWindows, } impl DriverExecutor { - pub fn new(drivers: Vec>, tracker: SettlementTracker) -> Self { - Self { drivers, tracker } + pub fn new(drivers: Vec>, windows: SettlementWindows) -> Self { + Self { drivers, windows } } } @@ -58,8 +58,8 @@ impl SettlementExecutor for DriverExecutor { // A window that cannot be opened must not block the settlement, // the dispatch is the priority. if let Err(err) = self - .tracker - .open_dispatched_settlement_window( + .windows + .open_dispatched( auction_id, winner.solver(), winner.id(), diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index 94d81be2dd..5b686304a8 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -1,6 +1,6 @@ -//! Settlement observation: tracks dispatched settlements as open windows in -//! `solana.settlement_executions` and resolves them against the -//! indexer-written `solana.settlements` rows. +//! Settlement windows: dispatched settlements tracked in +//! `solana.settlement_executions` and resolved against the indexer-written +//! `solana.settlements` rows. //! //! The executor opens a window per dispatched settlement. The indexer's //! insert into `solana.settlements` fires the `solana_settlement_finalized` @@ -18,18 +18,22 @@ use { sqlx::PgPool, }; -/// Opens and expires settlement-execution windows. +/// The settlement-execution windows in `solana.settlement_executions`: the +/// executor opens one per dispatched settlement, the competition cycle +/// expires the ones past their deadline, and the +/// `solana_settlement_finalized` notifications close the ones the indexer +/// saw land. /// -/// `solana.settlement_executions.outcome` records what the indexer observed -/// on chain, never a driver's report, which is why a landing observed after -/// the deadline overwrites a timeout. The schema also allows `rejected`, -/// unused until typed `/settle` errors are consumed. +/// `outcome` records what the indexer observed on chain, never a driver's +/// report, which is why a landing observed after the deadline overwrites a +/// timeout. The schema also allows `rejected`, unused until typed `/settle` +/// errors are consumed. #[derive(Clone)] -pub struct SettlementTracker { +pub struct SettlementWindows { pool: PgPool, } -impl SettlementTracker { +impl SettlementWindows { pub fn new(pool: PgPool) -> Self { Self { pool } } @@ -37,7 +41,7 @@ impl SettlementTracker { /// Open a window for a dispatched settlement. `solution_uid` is the /// winner's driver-local solution id until competition persistence /// allocates uids. - pub async fn open_dispatched_settlement_window( + pub async fn open_dispatched( &self, auction_id: i64, solver: Pubkey, @@ -60,33 +64,16 @@ impl SettlementTracker { /// timed out, logging each. Driven by the competition cycle, so on a /// chain with no active auctions a timeout surfaces with the next /// competition, not at its deadline slot. - pub async fn close_expired_windows_as_timeout(&self, slot: u64) -> Result<()> { + pub async fn expire_past_deadline(&self, slot: u64) -> Result<()> { let slot = to_db_integer(slot); for auction_id in db::expire_settlement_windows(&self.pool, slot).await? { tracing::error!(auction_id, slot, "settlement missed its deadline"); } Ok(()) } -} - -/// Slots and solution ids stay far below `i64::MAX`, the column type. -fn to_db_integer(value: u64) -> i64 { - i64::try_from(value).expect("value exceeds i64") -} - -/// Closes windows against the settlements the indexer records, driven by the -/// `solana_settlement_finalized` notifications. -pub struct SettlementObservation { - pool: PgPool, -} - -impl SettlementObservation { - pub fn new(pool: PgPool) -> Self { - Self { pool } - } /// Close the auction's windows against its observed settlements. - async fn resolve(&self, auction_id: i64) -> Result<()> { + async fn close_landed(&self, auction_id: i64) -> Result<()> { for landed in db::close_landed_windows(&self.pool, auction_id).await? { tracing::info!( auction_id, @@ -100,13 +87,18 @@ impl SettlementObservation { } } +/// Slots and solution ids stay far below `i64::MAX`, the column type. +fn to_db_integer(value: u64) -> i64 { + i64::try_from(value).expect("value exceeds i64") +} + #[async_trait] -impl NotifyHandler for SettlementObservation { +impl NotifyHandler for SettlementWindows { /// Re-check every open window: a NOTIFY missed while the connection was /// down (or while the autopilot was not running) is recovered here. async fn seed(&mut self) -> Result<()> { for auction_id in db::open_window_auction_ids(&self.pool).await? { - self.resolve(auction_id).await?; + self.close_landed(auction_id).await?; } Ok(()) } @@ -116,14 +108,14 @@ impl NotifyHandler for SettlementObservation { tracing::warn!(payload, "unparsable settlement notify payload"); return Ok(()); }; - self.resolve(auction_id).await + self.close_landed(auction_id).await } } #[cfg(test)] mod tests { use { - super::{SettlementObservation, SettlementTracker}, + super::SettlementWindows, crate::infra::listen::ListenSession, chain_types::solana::Pubkey, sqlx::PgPool, @@ -162,17 +154,14 @@ VALUES (10, $1, 0, $2, $3, NULL) let pool = PgPool::connect("postgresql://").await.unwrap(); crate::test_db::wipe(&pool).await; - let tracker = SettlementTracker::new(pool.clone()); - tracker - .open_dispatched_settlement_window(4242, Pubkey([7; 32]), 1, 90, 100) + let windows = SettlementWindows::new(pool.clone()); + windows + .open_dispatched(4242, Pubkey([7; 32]), 1, 90, 100) .await .unwrap(); - let task = ListenSession::spawn( - pool.clone(), - "solana_settlement_finalized", - SettlementObservation::new(pool.clone()), - ); + let task = + ListenSession::spawn(pool.clone(), "solana_settlement_finalized", windows.clone()); insert_settlement(&pool, 4242).await; @@ -201,17 +190,17 @@ VALUES (10, $1, 0, $2, $3, NULL) let pool = PgPool::connect("postgresql://").await.unwrap(); crate::test_db::wipe(&pool).await; - let tracker = SettlementTracker::new(pool.clone()); - tracker - .open_dispatched_settlement_window(1, Pubkey([7; 32]), 1, 90, 100) + let windows = SettlementWindows::new(pool.clone()); + windows + .open_dispatched(1, Pubkey([7; 32]), 1, 90, 100) .await .unwrap(); - tracker - .open_dispatched_settlement_window(2, Pubkey([7; 32]), 1, 90, 200) + windows + .open_dispatched(2, Pubkey([7; 32]), 1, 90, 200) .await .unwrap(); - tracker.close_expired_windows_as_timeout(150).await.unwrap(); + windows.expire_past_deadline(150).await.unwrap(); assert_eq!(outcome(&pool, 1).await.as_deref(), Some("timeout")); assert_eq!(outcome(&pool, 2).await, None); diff --git a/crates/autopilot-svm/src/infra/observer.rs b/crates/autopilot-svm/src/infra/observer.rs index 279c135d68..d42e4ffe9b 100644 --- a/crates/autopilot-svm/src/infra/observer.rs +++ b/crates/autopilot-svm/src/infra/observer.rs @@ -9,7 +9,7 @@ use { crate::{ domain::{auction::Auction, cycle::Ranking}, - infra::observation::SettlementTracker, + infra::observation::SettlementWindows, run_loop::SettlementObserver, }, async_trait::async_trait, @@ -20,12 +20,12 @@ use { /// Logs the competition phases and drives the settlement-timeout check off /// the per-cycle tip. pub struct LogObserver { - tracker: SettlementTracker, + windows: SettlementWindows, } impl LogObserver { - pub fn new(tracker: SettlementTracker) -> Self { - Self { tracker } + pub fn new(windows: SettlementWindows) -> Self { + Self { windows } } } @@ -44,7 +44,7 @@ impl SettlementObserver for LogObserver { ) -> anyhow::Result<()> { // Best effort: the expiry bookkeeping only touches previously // dispatched windows and must not block the current dispatch. - if let Err(err) = self.tracker.close_expired_windows_as_timeout(*tip).await { + if let Err(err) = self.windows.expire_past_deadline(*tip).await { tracing::error!(?err, "failed to flag expired settlement windows"); } tracing::info!( diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index cf81a480f6..35f78aa6c7 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -8,7 +8,7 @@ use { competition::DriverCompetition, driver::{Driver, dto}, executor::DriverExecutor, - observation::SettlementTracker, + observation::SettlementWindows, observer::LogObserver, provider::DbAuctionProvider, }, @@ -158,7 +158,7 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { assert_eq!(ranking.winner_count(), 1, "solution won"); } - let tracker = SettlementTracker::new(pool.clone()); + let windows = SettlementWindows::new(pool.clone()); let mut auction_loop = AuctionLoop::new( Box::new(FixedTrigger(tip)), Box::new(DbAuctionProvider::new(pool.clone())), @@ -167,8 +167,8 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { Duration::from_secs(6), )), Box::new(SolanaArbitrator::new(1, wrapped_native)), - Box::new(DriverExecutor::new(vec![driver], tracker.clone())), - Box::new(LogObserver::new(tracker.clone())), + Box::new(DriverExecutor::new(vec![driver], windows.clone())), + Box::new(LogObserver::new(windows.clone())), ); auction_loop.run_cycle().await; From cf362a0b1a07b7adb92f5be055c419dfde327620 Mon Sep 17 00:00:00 2001 From: squadgazzz <22964585+squadgazzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:55:58 +0000 Subject: [PATCH 12/12] Connect the crate's tests through one helper --- crates/autopilot-svm/src/infra/db.rs | 10 +++++++--- crates/autopilot-svm/src/infra/listen.rs | 3 +-- crates/autopilot-svm/src/infra/observation.rs | 4 ++-- crates/autopilot-svm/src/tests.rs | 2 +- 4 files changed, 11 insertions(+), 8 deletions(-) diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index 98d424c921..a4c502a840 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -61,6 +61,10 @@ ORDER BY o.uid /// Latest slot the indexer fully processed. `None` before the indexer's first /// write. `solana.indexer_state` is a single-row table. +#[cfg_attr( + not(test), + expect(dead_code, reason = "consumed by the freshness gating") +)] pub async fn last_indexed_slot(ex: impl PgExecutor<'_>) -> Result> { const QUERY: &str = r#"SELECT slot FROM solana.indexer_state"#; sqlx::query_scalar(QUERY) @@ -219,7 +223,7 @@ mod tests { super::{last_indexed_slot, open_orders}, bigdecimal::BigDecimal, database::byte_array::ByteArray, - sqlx::{PgPool, PgTransaction}, + sqlx::PgTransaction, }; fn conversion_row() -> super::OrderRow { @@ -315,7 +319,7 @@ VALUES ($1, $2, CASE WHEN $3 THEN now() END, $4, $5) #[tokio::test] #[ignore = "needs the solana.* schema applied to the local database"] async fn solana_db_open_orders_applies_the_solvability_predicates() { - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; let mut tx = pool.begin().await.unwrap(); for table in ["trades", "order_pda", "orders"] { @@ -362,7 +366,7 @@ VALUES ($1, $2, CASE WHEN $3 THEN now() END, $4, $5) #[tokio::test] #[ignore = "needs the solana.* schema applied to the local database"] async fn solana_db_last_indexed_slot_roundtrip() { - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; let mut tx = pool.begin().await.unwrap(); sqlx::query(r#"DELETE FROM solana.indexer_state"#) diff --git a/crates/autopilot-svm/src/infra/listen.rs b/crates/autopilot-svm/src/infra/listen.rs index 3966300248..2aaad4e430 100644 --- a/crates/autopilot-svm/src/infra/listen.rs +++ b/crates/autopilot-svm/src/infra/listen.rs @@ -99,7 +99,6 @@ mod tests { super::{ListenSession, NotifyHandler}, anyhow::Result, async_trait::async_trait, - sqlx::PgPool, std::{ sync::{Arc, Mutex}, time::Duration, @@ -150,7 +149,7 @@ mod tests { #[ignore = "needs a local postgres"] async fn postgres_delivers_notify_fired_during_seed() { const CHANNEL: &str = "autopilot_svm_listen_test"; - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; let log: Log = Arc::default(); let (started_tx, started_rx) = oneshot::channel(); let (gate_tx, gate_rx) = oneshot::channel(); diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index 5b686304a8..d5eaf1eefd 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -151,7 +151,7 @@ VALUES (10, $1, 0, $2, $3, NULL) #[tokio::test] #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] async fn solana_db_settlement_notify_closes_the_window_as_landed() { - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; crate::test_db::wipe(&pool).await; let windows = SettlementWindows::new(pool.clone()); @@ -187,7 +187,7 @@ VALUES (10, $1, 0, $2, $3, NULL) #[tokio::test] #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] async fn solana_db_expiry_times_out_only_past_deadlines() { - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; crate::test_db::wipe(&pool).await; let windows = SettlementWindows::new(pool.clone()); diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index 35f78aa6c7..e4033fb96d 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -119,7 +119,7 @@ VALUES ($1, $2, $2, $3, $2, $2, 1000, 500, $4, 'sell'::OrderKind, false, $2, now #[tokio::test] #[ignore = "needs the solana.* schema applied to the local database"] async fn solana_db_mock_cycle_dispatches_the_settlement() { - let pool = PgPool::connect("postgresql://").await.unwrap(); + let pool = crate::test_db::pool().await; let uid = [0x11; 32]; let tip = 500_u64; seed_open_order(&pool, uid, i64::try_from(tip).unwrap()).await;