diff --git a/crates/autopilot-svm/src/domain/arbitrator.rs b/crates/autopilot-svm/src/domain/arbitrator.rs index af7a068c1b..2fa5fb14a6 100644 --- a/crates/autopilot-svm/src/domain/arbitrator.rs +++ b/crates/autopilot-svm/src/domain/arbitrator.rs @@ -70,6 +70,11 @@ impl WinnerSelection for SolanaArbitrator { for winner in inner.winners() { tracing::info!(solver = %winner.solver(), solution = winner.id(), "winner"); } - Ranking { inner, drivers } + let reference_scores = self.inner.compute_reference_scores(&inner); + Ranking { + inner, + drivers, + reference_scores, + } } } diff --git a/crates/autopilot-svm/src/domain/cycle.rs b/crates/autopilot-svm/src/domain/cycle.rs index 99212ca026..5b27c8a1dd 100644 --- a/crates/autopilot-svm/src/domain/cycle.rs +++ b/crates/autopilot-svm/src/domain/cycle.rs @@ -7,7 +7,7 @@ use { }, chain_types::solana::{IntentHash, Pubkey, Solana}, std::collections::{HashMap, HashSet}, - winner_selection::{Unscored, solution}, + winner_selection::{Unscored, solution, state::Ranked}, }; /// Marker type binding the generic loop to the Solana vocabulary. @@ -45,6 +45,26 @@ pub struct Ranking { pub inner: winner_selection::Ranking, /// Driver index per solution, keyed by `(solver, solution id)`. pub drivers: HashMap, + /// Per winning solver, the winners' total score with that solver's + /// solutions removed: the rewards baseline. + pub reference_scores: HashMap, +} + +impl Ranking { + /// Every solution with its persisted uid: the position in `ranked` + /// followed by `filtered_out`, so winners come first. Everything + /// persisted references this uid, it disambiguates solver-assigned ids + /// across drivers. + pub fn enumerated( + &self, + ) -> impl Iterator, Solana>)> { + self.inner + .ranked + .iter() + .chain(self.inner.filtered_out.iter()) + .enumerate() + .map(|(uid, solution)| (i64::try_from(uid).unwrap_or(i64::MAX), solution)) + } } impl RankingInfo for Ranking { diff --git a/crates/autopilot-svm/src/infra/db.rs b/crates/autopilot-svm/src/infra/db.rs index fd713caa63..6bb21ed1eb 100644 --- a/crates/autopilot-svm/src/infra/db.rs +++ b/crates/autopilot-svm/src/infra/db.rs @@ -1,12 +1,13 @@ //! Database access for the Solana autopilot. use { - crate::domain::auction::{Auction, Order, OrderKind}, + crate::domain::auction::{Order, OrderKind}, anyhow::{Context, Result}, bigdecimal::{BigDecimal, ToPrimitive}, chain_types::solana::{AppData, IntentHash, Pubkey}, database::byte_array::ByteArray, - sqlx::PgExecutor, + solana_sdk::clock::MAX_PROCESSING_AGE, + sqlx::{PgExecutor, Postgres, QueryBuilder}, }; /// The channel the schema's `solana.settlements` trigger notifies. @@ -167,14 +168,12 @@ WHERE auction_id = $1 AND solver = $2 AND solution_uid = $3 AND outcome IS NULL 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. +/// Close the auction's windows against the settlements the indexer recorded. +/// A settlement carries no solution uid, so a window is matched through its +/// solution's trade executions: the solver's settlement that traded one of +/// them is the one that executed it. 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`. pub async fn close_landed_windows( ex: impl PgExecutor<'_>, auction_id: i64, @@ -188,6 +187,15 @@ 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') + AND EXISTS ( + SELECT 1 + FROM solana.proposed_trade_executions pte + JOIN solana.trades t + ON t.tx_signature = s.tx_signature + AND t.instruction_index = s.instruction_index + AND t.order_uid = pte.order_uid + WHERE pte.auction_id = e.auction_id AND pte.solution_uid = e.solution_uid + ) RETURNING e.solver, e.end_slot, e.submitted_signature "#; sqlx::query_as(QUERY) @@ -221,19 +229,201 @@ pub async fn open_window_auction_ids(ex: impl PgExecutor<'_>) -> Result .context("read open settlement execution windows") } -/// Cut an auction from the open orders. +/// Orders inside a winning solution whose settlement transaction may still +/// land: a blockhash lifetime past the deadline slot has not run out, no +/// settlement of the auction traded the order yet, and the driver did not +/// reject the solution before sending it. A timed-out window keeps the hold, +/// its transaction may land until the blockhash expires. Landing is checked +/// per order through the settlement's trades: `settlements.solution_uid` is +/// unattributed, and one solver may win several solutions of one auction. +pub async fn in_flight_orders( + ex: impl PgExecutor<'_>, + tip_slot: i64, +) -> Result>> { + const QUERY: &str = r#" +SELECT DISTINCT pte.order_uid +FROM solana.competition_auctions ca +JOIN solana.proposed_solutions ps ON ps.auction_id = ca.id AND ps.is_winner +JOIN solana.proposed_trade_executions pte + ON pte.auction_id = ca.id AND pte.solution_uid = ps.uid +WHERE ca.deadline_slot >= $1 + AND NOT EXISTS ( + SELECT 1 + FROM solana.settlements s + JOIN solana.trades t + ON t.tx_signature = s.tx_signature AND t.instruction_index = s.instruction_index + WHERE s.auction_id = ca.id AND t.order_uid = pte.order_uid + ) + AND NOT EXISTS ( + SELECT 1 FROM solana.settlement_executions se + WHERE se.auction_id = ca.id AND se.solution_uid = ps.uid AND se.outcome = 'rejected' + ) + "#; + // The oldest deadline whose transaction can still land at the tip. + let lifetime = i64::try_from(MAX_PROCESSING_AGE).expect("blockhash lifetime fits i64"); + let landable_deadline = tip_slot.saturating_sub(lifetime); + sqlx::query_scalar(QUERY) + .bind(landable_deadline) + .fetch_all(ex) + .await + .context("read in-flight orders") +} + +/// The solvable orders for a fresh auction cut. pub async fn cut( ex: impl PgExecutor<'_>, - id: i64, now_unix: i64, block_height: Option, -) -> Result { - let orders = orders_from_rows(open_orders(ex, now_unix, block_height).await?); - Ok(Auction { - id, - orders, - native_prices: Default::default(), - }) +) -> Result> { + Ok(orders_from_rows( + open_orders(ex, now_unix, block_height).await?, + )) +} + +/// Replace the current auction and answer the id its identity column +/// allocated, the source of sequential auction ids. +pub async fn replace_current_auction( + pool: &sqlx::PgPool, + tip_slot: i64, + json: &serde_json::Value, +) -> Result { + let mut tx = pool.begin().await.context("begin auction replacement")?; + sqlx::query("DELETE FROM solana.auctions") + .execute(&mut *tx) + .await + .context("delete the previous auction")?; + let id = sqlx::query_scalar( + "INSERT INTO solana.auctions (tip_slot, json) VALUES ($1, $2) RETURNING id", + ) + .bind(tip_slot) + .bind(sqlx::types::Json(json)) + .fetch_one(&mut *tx) + .await + .context("insert the current auction")?; + tx.commit().await.context("commit auction replacement")?; + Ok(id) +} + +/// One execution inside a proposed solution. +pub struct ProposedTrade { + pub order_uid: ByteArray<32>, + pub executed_sell: BigDecimal, + pub executed_buy: BigDecimal, +} + +/// One proposed solution of a competition, with its executions. +pub struct ProposedSolution { + /// Autopilot-generated, unique within the auction. + pub uid: i64, + /// Solver-assigned, unique only within one driver response. + pub id: i64, + pub solver: ByteArray<32>, + pub is_winner: bool, + pub filtered_out: bool, + pub score: BigDecimal, + pub trades: Vec, +} + +/// A winning solver's reference score: the winners' total score with that +/// solver's solutions removed. +pub struct ReferenceScore { + pub solver: ByteArray<32>, + pub score: BigDecimal, +} + +/// A competition outcome as persisted after ranking. +pub struct Competition { + pub auction_id: i64, + pub tip_slot: i64, + pub deadline_slot: i64, + pub order_uids: Vec>, + pub price_tokens: Vec>, + pub price_values: Vec, + pub solutions: Vec, + pub reference_scores: Vec, +} + +/// Persist a competition: the auction snapshot and every proposed solution +/// with its executions, in one transaction. +pub async fn persist_competition(pool: &sqlx::PgPool, competition: &Competition) -> Result<()> { + let mut tx = pool.begin().await.context("begin competition persist")?; + sqlx::query( + "INSERT INTO solana.competition_auctions (id, tip_slot, deadline_slot, order_uids, \ + price_tokens, price_values) VALUES ($1, $2, $3, $4, $5, $6)", + ) + .bind(competition.auction_id) + .bind(competition.tip_slot) + .bind(competition.deadline_slot) + .bind(&competition.order_uids) + .bind(&competition.price_tokens) + .bind(&competition.price_values) + .execute(&mut *tx) + .await + .context("insert competition auction")?; + if !competition.solutions.is_empty() { + let mut insert = QueryBuilder::::new( + "INSERT INTO solana.proposed_solutions (auction_id, uid, id, solver, is_winner, \ + filtered_out, score) ", + ); + insert.push_values(&competition.solutions, |mut row, solution| { + row.push_bind(competition.auction_id) + .push_bind(solution.uid) + .push_bind(solution.id) + .push_bind(solution.solver) + .push_bind(solution.is_winner) + .push_bind(solution.filtered_out) + .push_bind(&solution.score); + }); + insert + .build() + .execute(&mut *tx) + .await + .context("insert proposed solutions")?; + } + let trades: Vec<_> = competition + .solutions + .iter() + .flat_map(|solution| { + solution + .trades + .iter() + .map(move |trade| (solution.uid, trade)) + }) + .collect(); + if !trades.is_empty() { + let mut insert = QueryBuilder::::new( + "INSERT INTO solana.proposed_trade_executions (auction_id, solution_uid, order_uid, \ + executed_sell, executed_buy) ", + ); + insert.push_values(trades, |mut row, (solution_uid, trade)| { + row.push_bind(competition.auction_id) + .push_bind(solution_uid) + .push_bind(trade.order_uid) + .push_bind(&trade.executed_sell) + .push_bind(&trade.executed_buy); + }); + insert + .build() + .execute(&mut *tx) + .await + .context("insert proposed trade executions")?; + } + if !competition.reference_scores.is_empty() { + let mut insert = QueryBuilder::::new( + "INSERT INTO solana.reference_scores (auction_id, solver, reference_score) ", + ); + insert.push_values(&competition.reference_scores, |mut row, reference| { + row.push_bind(competition.auction_id) + .push_bind(reference.solver) + .push_bind(&reference.score); + }); + insert + .build() + .execute(&mut *tx) + .await + .context("insert reference scores")?; + } + tx.commit().await.context("commit competition persist") } /// A row the indexer wrote always converts (on-chain values fit the domain @@ -289,7 +479,7 @@ fn to_amount(value: &BigDecimal) -> Result { #[cfg(test)] mod tests { use { - super::{last_indexed_slot, open_orders}, + super::{in_flight_orders, last_indexed_slot, open_orders}, bigdecimal::BigDecimal, database::byte_array::ByteArray, sqlx::PgTransaction, @@ -454,6 +644,137 @@ WHERE uid = $1 assert_eq!(uids(orders), vec![1, 5, 6]); } + /// Held: an order of a winning solution through its deadline slot plus + /// the blockhash lifetime, a timed-out window included. Released: after + /// that, once a settlement of the auction trades the order, or once the + /// driver rejects its solution before sending. A solver's second winning + /// solution keeps its hold when the first one lands. Orders of + /// non-winning solutions are never held. + #[tokio::test] + #[ignore = "needs the solana.* schema applied to the local database"] + async fn solana_db_in_flight_orders_follow_the_winning_settlement() { + let pool = crate::test_db::pool().await; + let mut tx = pool.begin().await.unwrap(); + for table in [ + "trades", + "settlements", + "settlement_executions", + "proposed_trade_executions", + "proposed_solutions", + "competition_auctions", + ] { + sqlx::query(&format!("DELETE FROM solana.{table}")) + .execute(&mut *tx) + .await + .unwrap(); + } + let winner = ByteArray([0xEE; 32]); + sqlx::query( + "INSERT INTO solana.competition_auctions (id, tip_slot, deadline_slot, order_uids, \ + price_tokens, price_values) VALUES (77, 1, 100, '{}', '{}', '{}')", + ) + .execute(&mut *tx) + .await + .unwrap(); + for (uid, solver, is_winner, order) in [ + (0i64, winner, true, 1u8), + (1, ByteArray([0xEF; 32]), false, 2), + (2, winner, true, 3), + ] { + sqlx::query( + "INSERT INTO solana.proposed_solutions (auction_id, uid, id, solver, is_winner, \ + filtered_out, score) VALUES (77, $1, 7, $2, $3, false, 1)", + ) + .bind(uid) + .bind(solver) + .bind(is_winner) + .execute(&mut *tx) + .await + .unwrap(); + sqlx::query( + "INSERT INTO solana.proposed_trade_executions (auction_id, solution_uid, \ + order_uid, executed_sell, executed_buy) VALUES (77, $1, $2, 10, 20)", + ) + .bind(uid) + .bind(ByteArray([order; 32])) + .execute(&mut *tx) + .await + .unwrap(); + } + async fn held(tx: &mut PgTransaction<'_>, tip: i64) -> Vec { + let mut held: Vec = in_flight_orders(&mut **tx, tip) + .await + .unwrap() + .iter() + .map(|uid| uid.0[0]) + .collect(); + held.sort_unstable(); + held + } + + assert_eq!(held(&mut tx, 100).await, vec![1, 3]); + assert_eq!(held(&mut tx, 250).await, vec![1, 3]); + assert_eq!(held(&mut tx, 251).await, Vec::::new()); + + sqlx::query( + "INSERT INTO solana.settlement_executions (auction_id, solver, solution_uid, \ + start_timestamp, start_slot, deadline_slot) VALUES (77, $1, 0, now(), 1, 100)", + ) + .bind(winner) + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!(held(&mut tx, 50).await, vec![1, 3]); + sqlx::query( + "UPDATE solana.settlement_executions SET outcome = 'rejected', end_slot = 2, \ + end_timestamp = now() WHERE auction_id = 77", + ) + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!(held(&mut tx, 50).await, vec![3]); + sqlx::query("UPDATE solana.settlement_executions SET outcome = NULL WHERE auction_id = 77") + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!(held(&mut tx, 50).await, vec![1, 3]); + sqlx::query( + "UPDATE solana.settlement_executions SET outcome = 'timeout', end_slot = 100, \ + end_timestamp = now() WHERE auction_id = 77", + ) + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!(held(&mut tx, 150).await, vec![1, 3]); + + // The solver's settlement trades order 1 only: order 3, its other + // winning solution, stays held until its own trade lands. + sqlx::query( + "INSERT INTO solana.settlements (slot, tx_signature, instruction_index, solver, \ + auction_id) VALUES (10, $1, 0, $2, 77)", + ) + .bind([9u8; 64]) + .bind(winner) + .execute(&mut *tx) + .await + .unwrap(); + for order in [1u8, 3] { + sqlx::query( + "INSERT INTO solana.trades (tx_signature, instruction_index, order_uid, \ + sell_amount, buy_amount, fee_amount) VALUES ($1, 0, $2, 10, 20, 0)", + ) + .bind([9u8; 64]) + .bind(ByteArray([order; 32])) + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!( + held(&mut tx, 50).await, + if order == 1 { vec![3] } else { vec![] } + ); + } + } + #[tokio::test] #[ignore = "needs the solana.* schema applied to the local database"] async fn solana_db_last_indexed_slot_roundtrip() { diff --git a/crates/autopilot-svm/src/infra/executor.rs b/crates/autopilot-svm/src/infra/executor.rs index 5af43c6c87..9eeeed5db9 100644 --- a/crates/autopilot-svm/src/infra/executor.rs +++ b/crates/autopilot-svm/src/infra/executor.rs @@ -5,7 +5,6 @@ use { domain::cycle::{Ranking, SolanaCycle}, infra::{ driver::{Driver, dto}, - inflight::InFlightOrders, observation::SettlementWindows, sponsor::Sponsor, }, @@ -13,6 +12,7 @@ use { }, async_trait::async_trait, std::sync::Arc, + winner_selection::state::RankedItem, }; /// Sends `/settle` to each winner's driver. Submission runs detached, the @@ -26,9 +26,6 @@ pub struct DriverExecutor { /// without creations, and one containing a pending sponsored order fails /// at the driver. sponsor: Option, - /// Orders dispatched here are held out of auction cuts until their - /// submission deadline passes. - inflight: InFlightOrders, } impl DriverExecutor { @@ -36,13 +33,11 @@ impl DriverExecutor { drivers: Vec>, windows: SettlementWindows, sponsor: Option, - inflight: InFlightOrders, ) -> Self { Self { drivers, windows, sponsor, - inflight, } } } @@ -50,7 +45,10 @@ impl DriverExecutor { #[async_trait] impl SettlementExecutor for DriverExecutor { async fn execute(&self, auction_id: i64, ranking: &Ranking, tip: &u64, deadline: u64) { - for winner in ranking.inner.winners() { + for (uid, winner) in ranking + .enumerated() + .filter(|(_, winner)| winner.is_winner()) + { let key = (winner.solver(), winner.id()); let Some(driver) = ranking .drivers @@ -90,24 +88,19 @@ impl SettlementExecutor for DriverExecutor { submission_deadline_slot: deadline, creations, }; - // Held before the dispatch: the next cut must not re-auction - // these orders while the settlement can still land. - let uids: Vec<_> = winner.orders().iter().map(|order| order.uid).collect(); - self.inflight - .hold(auction_id, winner.solver(), uids.iter().copied(), deadline); // A window that cannot be opened must not block the settlement, - // the dispatch is the priority. + // the dispatch is the priority. The window carries the generated + // solution uid, the driver request keeps the driver-local id its + // solution cache is keyed by. if let Err(err) = self .windows - .open_dispatched(auction_id, winner.solver(), winner.id(), *tip, deadline) + .open_dispatched(auction_id, winner.solver(), uid, *tip, deadline) .await { tracing::error!(auction_id, ?err, "failed to open the settlement window"); } - let inflight = self.inflight.clone(); let windows = self.windows.clone(); let solver = winner.solver(); - let solution_uid = winner.id(); tokio::spawn(async move { match driver.settle(&request).await { Ok(response) => tracing::info!( @@ -117,9 +110,9 @@ impl SettlementExecutor for DriverExecutor { tx_signature = %response.tx_signature, "settlement submitted" ), - // No transaction went out, so the orders can re-enter - // the next cut instead of waiting out the hold, and the - // window has nothing left to observe. + // No transaction went out, so the window closes as + // rejected: the orders re-enter the next cut and nothing + // is left to observe. Err(err) if err.settlement_provably_unsent() => { tracing::warn!( driver = %driver.name, @@ -127,11 +120,7 @@ impl SettlementExecutor for DriverExecutor { ?err, "settlement rejected before submission" ); - inflight.release(uids); - if let Err(err) = windows - .close_rejected(auction_id, solver, solution_uid) - .await - { + if let Err(err) = windows.close_rejected(auction_id, solver, uid).await { tracing::error!( auction_id, ?err, diff --git a/crates/autopilot-svm/src/infra/inflight.rs b/crates/autopilot-svm/src/infra/inflight.rs deleted file mode 100644 index accdc59134..0000000000 --- a/crates/autopilot-svm/src/infra/inflight.rs +++ /dev/null @@ -1,175 +0,0 @@ -//! In-memory hold-out of orders with a settlement in flight. -//! -//! An order dispatched for settlement must not re-enter the next auction -//! while the first settlement can still land: a second winner would -//! double-settle it. The driver stops waiting at the submission deadline, -//! but the transaction it sent stays landable until its blockhash expires, -//! up to `MAX_PROCESSING_AGE` slots later, so held orders expire only at -//! the deadline plus that lifetime. Two events end a hold early: the -//! settlement is observed on chain (the auction's orders are done), or the -//! driver rejects provably before any send. Everything else, a submit error -//! or an exceeded deadline, may have left the transaction on the wire and -//! keeps the hold. -//! -//! The map lives in memory: a restart forgets it and reopens the window -//! until the entries would have expired. - -use { - chain_types::solana::{IntentHash, Pubkey}, - solana_sdk::clock::MAX_PROCESSING_AGE, - std::{ - collections::{HashMap, HashSet}, - sync::{Arc, Mutex}, - }, -}; - -/// A dispatched settlement, keyed the way the indexer identifies a landed -/// one. -#[derive(Clone, Copy, PartialEq, Eq)] -struct Settlement { - auction_id: i64, - solver: Pubkey, -} - -/// Why an order is held, and until when. -#[derive(Clone, Copy)] -struct Hold { - settlement: Settlement, - /// The last slot the settlement's transaction could still land. - expires_at: u64, -} - -/// Order uids held out of auction cuts until their settlement transaction -/// cannot land any more. -#[derive(Clone, Default)] -pub struct InFlightOrders(Arc>>); - -impl InFlightOrders { - /// Hold the settlement's orders until the deadline slot plus the - /// blockhash lifetime, the last slot its transaction could still land. - /// An order already held keeps the hold that expires last. - pub fn hold( - &self, - auction_id: i64, - solver: Pubkey, - uids: impl IntoIterator, - deadline_slot: u64, - ) { - let hold = Hold { - settlement: Settlement { auction_id, solver }, - expires_at: deadline_slot.saturating_add(MAX_PROCESSING_AGE as u64), - }; - let mut held = self.0.lock().expect("mutex poisoned"); - for uid in uids { - held.entry(uid) - .and_modify(|current| { - if hold.expires_at > current.expires_at { - *current = hold; - } - }) - .or_insert(hold); - } - } - - /// Release the orders: their settlement provably never went out, so no - /// second settlement can collide. - pub fn release(&self, uids: impl IntoIterator) { - let mut held = self.0.lock().expect("mutex poisoned"); - for uid in uids { - held.remove(&uid); - } - } - - /// Release the orders of a settlement observed on chain. The auction is - /// settled for this solver, nothing else can execute these orders. - pub fn release_landed(&self, auction_id: i64, solver: Pubkey) { - let settlement = Settlement { auction_id, solver }; - self.0 - .lock() - .expect("mutex poisoned") - .retain(|_, hold| hold.settlement != settlement); - } - - /// The orders still held at the tip. Expired entries are pruned on the - /// way. - pub fn held_at(&self, tip: u64) -> HashSet { - let mut held = self.0.lock().expect("mutex poisoned"); - held.retain(|_, hold| hold.expires_at >= tip); - held.keys().copied().collect() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - const SOLVER: Pubkey = Pubkey([9; 32]); - - #[test] - fn holds_through_the_blockhash_lifetime_past_the_deadline() { - let inflight = InFlightOrders::default(); - let uid = IntentHash([7; 32]); - inflight.hold(1, SOLVER, [uid], 100); - let expiry = 100 + MAX_PROCESSING_AGE as u64; - - assert!(inflight.held_at(100).contains(&uid)); - assert!(inflight.held_at(expiry).contains(&uid)); - assert!(inflight.held_at(expiry + 1).is_empty()); - } - - #[test] - fn released_orders_re_enter_immediately() { - let inflight = InFlightOrders::default(); - let uid = IntentHash([7; 32]); - let other = IntentHash([8; 32]); - inflight.hold(1, SOLVER, [uid, other], 100); - inflight.release([uid]); - - let held = inflight.held_at(50); - assert!(!held.contains(&uid)); - assert!(held.contains(&other)); - } - - #[test] - fn a_landed_settlement_releases_its_orders() { - let inflight = InFlightOrders::default(); - let uid = IntentHash([7; 32]); - let unrelated = IntentHash([8; 32]); - inflight.hold(1, SOLVER, [uid], 100); - inflight.hold(2, SOLVER, [unrelated], 100); - - inflight.release_landed(1, SOLVER); - let held = inflight.held_at(50); - assert!(!held.contains(&uid)); - assert!(held.contains(&unrelated)); - - // A landing for a solver without a dispatch is a no-op. - inflight.release_landed(3, Pubkey([1; 32])); - assert!(inflight.held_at(50).contains(&unrelated)); - } - - #[test] - fn a_second_dispatch_keeps_the_later_expiry() { - let inflight = InFlightOrders::default(); - let uid = IntentHash([7; 32]); - inflight.hold(1, SOLVER, [uid], 100); - inflight.hold(2, SOLVER, [uid], 90); - - assert!( - inflight - .held_at(100 + MAX_PROCESSING_AGE as u64) - .contains(&uid) - ); - } - - #[test] - fn a_landing_keeps_an_order_the_later_dispatch_still_holds() { - let inflight = InFlightOrders::default(); - let uid = IntentHash([7; 32]); - inflight.hold(1, SOLVER, [uid], 100); - inflight.hold(2, SOLVER, [uid], 200); - - inflight.release_landed(1, SOLVER); - assert!(inflight.held_at(50).contains(&uid)); - } -} diff --git a/crates/autopilot-svm/src/infra/mod.rs b/crates/autopilot-svm/src/infra/mod.rs index b491908e90..08dceb60fd 100644 --- a/crates/autopilot-svm/src/infra/mod.rs +++ b/crates/autopilot-svm/src/infra/mod.rs @@ -5,7 +5,6 @@ pub mod config; pub mod db; pub mod driver; pub mod executor; -pub mod inflight; pub mod listen; pub mod observation; pub mod observer; diff --git a/crates/autopilot-svm/src/infra/observation.rs b/crates/autopilot-svm/src/infra/observation.rs index b5ef096c2d..b9ba306adb 100644 --- a/crates/autopilot-svm/src/infra/observation.rs +++ b/crates/autopilot-svm/src/infra/observation.rs @@ -11,7 +11,7 @@ //! window. use { - crate::infra::{db, inflight::InFlightOrders, listen::NotifyHandler}, + crate::infra::{db, listen::NotifyHandler}, anyhow::Result, async_trait::async_trait, chain_types::solana::{Pubkey, Signature}, @@ -31,23 +31,20 @@ use { #[derive(Clone)] pub struct SettlementWindows { pool: PgPool, - /// A settlement observed on chain releases its orders from the hold-out. - inflight: InFlightOrders, } impl SettlementWindows { - pub fn new(pool: PgPool, inflight: InFlightOrders) -> Self { - Self { pool, inflight } + 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. + /// autopilot-generated uid the competition persisted. pub async fn open_dispatched( &self, auction_id: i64, solver: Pubkey, - solution_uid: u64, + solution_uid: i64, start_slot: u64, deadline_slot: u64, ) -> Result<()> { @@ -55,7 +52,7 @@ impl SettlementWindows { &self.pool, auction_id, solver, - to_db_integer(solution_uid), + solution_uid, to_db_integer(start_slot), to_db_integer(deadline_slot), ) @@ -70,10 +67,9 @@ impl SettlementWindows { &self, auction_id: i64, solver: Pubkey, - solution_uid: u64, + solution_uid: i64, ) -> Result<()> { - db::reject_settlement_window(&self.pool, auction_id, solver, to_db_integer(solution_uid)) - .await + db::reject_settlement_window(&self.pool, auction_id, solver, solution_uid).await } /// Close every open window whose deadline is at or before the slot as @@ -99,7 +95,6 @@ impl SettlementWindows { tx_signature = %Signature(landed.submitted_signature.0), "settlement observed on chain" ); - self.inflight.release_landed(auction_id, solver); } Ok(()) } @@ -134,20 +129,50 @@ impl NotifyHandler for SettlementWindows { mod tests { use { super::SettlementWindows, - crate::infra::{db, inflight::InFlightOrders, listen::ListenSession}, - chain_types::solana::{IntentHash, Pubkey}, + crate::infra::{db, listen::ListenSession}, + chain_types::solana::Pubkey, sqlx::PgPool, std::time::Duration, }; - async fn insert_settlement(pool: &PgPool, auction_id: i64) { + /// A persisted execution of order `[order; 32]` inside solution `uid` of + /// the auction. + async fn insert_execution(pool: &PgPool, auction_id: i64, uid: i64, order: u8) { + sqlx::query( + "INSERT INTO solana.proposed_trade_executions (auction_id, solution_uid, order_uid, \ + executed_sell, executed_buy) VALUES ($1, $2, $3, 10, 20)", + ) + .bind(auction_id) + .bind(uid) + .bind([order; 32]) + .execute(pool) + .await + .unwrap(); + } + + /// A landed settlement of the auction by solver 7 under signature + /// `[signature; 64]`, trading the given orders. The trades go in first: + /// the indexer commits both together, so the NOTIFY the settlement fires + /// never sees a settlement without its trades. + async fn insert_settlement(pool: &PgPool, auction_id: i64, signature: u8, orders: &[u8]) { + for order in orders { + sqlx::query( + "INSERT INTO solana.trades (tx_signature, instruction_index, order_uid, \ + sell_amount, buy_amount, fee_amount) VALUES ($1, 0, $2, 10, 20, 0)", + ) + .bind([signature; 64]) + .bind([*order; 32]) + .execute(pool) + .await + .unwrap(); + } 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([signature; 64]) .bind([7u8; 32]) .bind(auction_id) .execute(pool) @@ -163,6 +188,21 @@ VALUES (10, $1, 0, $2, $3, NULL) .unwrap() } + /// Outcome and signature of every window of the auction, by solution uid. + async fn windows_of( + pool: &PgPool, + auction_id: i64, + ) -> Vec<(i64, Option, Option>)> { + sqlx::query_as( + "SELECT solution_uid, outcome, submitted_signature FROM solana.settlement_executions \ + WHERE auction_id = $1 ORDER BY solution_uid", + ) + .bind(auction_id) + .fetch_all(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. @@ -173,14 +213,12 @@ VALUES (10, $1, 0, $2, $3, NULL) crate::test_db::wipe(&pool).await; let solver = Pubkey([7; 32]); - let uid = IntentHash([1; 32]); - let inflight = InFlightOrders::default(); - inflight.hold(4242, solver, [uid], 100); - let windows = SettlementWindows::new(pool.clone(), inflight.clone()); + let windows = SettlementWindows::new(pool.clone()); windows .open_dispatched(4242, solver, 1, 90, 100) .await .unwrap(); + insert_execution(&pool, 4242, 1, 1).await; let task = ListenSession::spawn( pool.clone(), @@ -188,7 +226,7 @@ VALUES (10, $1, 0, $2, $3, NULL) windows.clone(), ); - insert_settlement(&pool, 4242).await; + insert_settlement(&pool, 4242, 9, &[1]).await; for _ in 0..200 { if outcome(&pool, 4242).await.is_some() { @@ -198,8 +236,6 @@ VALUES (10, $1, 0, $2, $3, NULL) } task.abort(); assert_eq!(outcome(&pool, 4242).await.as_deref(), Some("landed")); - // The observed landing released the held order. - assert!(inflight.held_at(90).is_empty()); let signature: Vec = sqlx::query_scalar( "SELECT submitted_signature FROM solana.settlement_executions WHERE auction_id = 4242", ) @@ -217,7 +253,7 @@ VALUES (10, $1, 0, $2, $3, NULL) let pool = crate::test_db::pool().await; crate::test_db::wipe(&pool).await; - let windows = SettlementWindows::new(pool.clone(), InFlightOrders::default()); + let windows = SettlementWindows::new(pool.clone()); windows .open_dispatched(1, Pubkey([7; 32]), 1, 90, 100) .await @@ -233,7 +269,8 @@ VALUES (10, $1, 0, $2, $3, NULL) // A settlement observed after the timeout upgrades the verdict: it // executed, just late. - insert_settlement(&pool, 1).await; + insert_execution(&pool, 1, 1, 1).await; + insert_settlement(&pool, 1, 9, &[1]).await; crate::infra::db::close_landed_windows(&pool, 1) .await .unwrap(); @@ -248,7 +285,7 @@ VALUES (10, $1, 0, $2, $3, NULL) let pool = crate::test_db::pool().await; crate::test_db::wipe(&pool).await; - let windows = SettlementWindows::new(pool.clone(), InFlightOrders::default()); + let windows = SettlementWindows::new(pool.clone()); windows .open_dispatched(1, Pubkey([7; 32]), 1, 90, 100) .await @@ -269,13 +306,14 @@ VALUES (10, $1, 0, $2, $3, NULL) let pool = crate::test_db::pool().await; crate::test_db::wipe(&pool).await; - let windows = SettlementWindows::new(pool.clone(), InFlightOrders::default()); + let windows = SettlementWindows::new(pool.clone()); windows .open_dispatched(1, Pubkey([7; 32]), 1, 90, 100) .await .unwrap(); - insert_settlement(&pool, 1).await; + insert_execution(&pool, 1, 1, 1).await; + insert_settlement(&pool, 1, 9, &[1]).await; crate::infra::db::close_landed_windows(&pool, 1) .await .unwrap(); @@ -284,4 +322,48 @@ VALUES (10, $1, 0, $2, $3, NULL) windows.close_rejected(1, Pubkey([7; 32]), 1).await.unwrap(); assert_eq!(outcome(&pool, 1).await.as_deref(), Some("landed")); } + + /// A solver holding two windows of one auction: the settlement that + /// traded the first solution's order closes only that window, the second + /// window closes on its own settlement with its own signature. + #[tokio::test] + #[ignore = "needs the solana.* schema applied locally, run with --test-threads 1"] + async fn solana_db_a_landing_closes_only_the_window_it_executed() { + let pool = crate::test_db::pool().await; + crate::test_db::wipe(&pool).await; + + let solver = Pubkey([7; 32]); + let windows = SettlementWindows::new(pool.clone()); + for (uid, order) in [(1, 1u8), (2, 2)] { + windows + .open_dispatched(1, solver, uid, 90, 100) + .await + .unwrap(); + insert_execution(&pool, 1, uid, order).await; + } + + insert_settlement(&pool, 1, 9, &[1]).await; + crate::infra::db::close_landed_windows(&pool, 1) + .await + .unwrap(); + assert_eq!( + windows_of(&pool, 1).await, + vec![ + (1, Some("landed".to_string()), Some(vec![9u8; 64])), + (2, None, None), + ] + ); + + insert_settlement(&pool, 1, 8, &[2]).await; + crate::infra::db::close_landed_windows(&pool, 1) + .await + .unwrap(); + assert_eq!( + windows_of(&pool, 1).await, + vec![ + (1, Some("landed".to_string()), Some(vec![9u8; 64])), + (2, Some("landed".to_string()), Some(vec![8u8; 64])), + ] + ); + } } diff --git a/crates/autopilot-svm/src/infra/observer.rs b/crates/autopilot-svm/src/infra/observer.rs index dafbe02c90..ce73910a86 100644 --- a/crates/autopilot-svm/src/infra/observer.rs +++ b/crates/autopilot-svm/src/infra/observer.rs @@ -1,22 +1,19 @@ -//! Competition bookkeeping. Auction progress is written to -//! `solana.order_events`, everything else is logged only: there are no -//! competition tables (auction snapshots, proposed executions) to write. -//! -//! TODO: persist the competition outcome once the tables exist. The -//! settlement attribution (`solana.settlements.solution_uid`) depends on -//! the persisted ranking. +//! Competition bookkeeping: auction progress in `solana.order_events`, the +//! ranked outcome in the competition tables. use { crate::{ domain::{auction::Auction, cycle::Ranking}, - infra::{observation::SettlementWindows, order_events}, + infra::{db, observation::SettlementWindows, order_events}, run_loop::SettlementObserver, }, async_trait::async_trait, + bigdecimal::BigDecimal, chain_types::solana::IntentHash, - database::solana::OrderEventLabel, + database::{byte_array::ByteArray, solana::OrderEventLabel}, sqlx::PgPool, std::{collections::HashSet, sync::Mutex}, + winner_selection::state::RankedItem, }; /// Writes order events, logs the competition phases, and drives the @@ -41,12 +38,7 @@ impl CompetitionObserver { /// Store the events without blocking the cycle: a lost event degrades the /// status endpoint, never the competition. fn store_events(&self, uids: Vec, label: OrderEventLabel) { - let pool = self.pool.clone(); - tokio::spawn(async move { - if let Err(err) = order_events::store(&pool, uids, label).await { - tracing::error!(?err, ?label, "failed to store order events"); - } - }); + order_events::store_detached(self.pool.clone(), uids, label); } } @@ -82,7 +74,7 @@ impl SettlementObserver for CompetitionObserv async fn persist_competition_ranking( &self, - _auction: &Auction, + auction: &Auction, tip: &u64, ranking: &Ranking, deadline: u64, @@ -92,13 +84,61 @@ impl SettlementObserver for CompetitionObserv if let Err(err) = self.windows.expire_past_deadline(*tip).await { tracing::error!(?err, "failed to flag expired settlement windows"); } + let solutions = ranking + .enumerated() + .map(|(uid, solution)| db::ProposedSolution { + uid, + id: i64::try_from(solution.id()).unwrap_or(i64::MAX), + solver: ByteArray(solution.solver().0), + is_winner: solution.is_winner(), + filtered_out: solution.is_filtered_out(), + score: BigDecimal::from(solution.score()), + trades: solution + .orders() + .iter() + .map(|order| db::ProposedTrade { + order_uid: ByteArray(order.uid.0), + executed_sell: BigDecimal::from(order.executed_sell), + executed_buy: BigDecimal::from(order.executed_buy), + }) + .collect(), + }) + .collect(); + let (price_tokens, price_values) = auction + .native_prices + .iter() + .map(|(token, price)| (token.0.to_vec(), BigDecimal::from(*price))) + .unzip(); + let competition = db::Competition { + auction_id: auction.id, + tip_slot: i64::try_from(*tip).unwrap_or(i64::MAX), + deadline_slot: i64::try_from(deadline).unwrap_or(i64::MAX), + order_uids: auction + .orders + .iter() + .map(|order| order.uid.0.to_vec()) + .collect(), + price_tokens, + price_values, + solutions, + reference_scores: ranking + .reference_scores + .iter() + .map(|(solver, score)| db::ReferenceScore { + solver: ByteArray(solver.0), + score: BigDecimal::from(*score), + }) + .collect(), + }; + db::persist_competition(&self.pool, &competition).await?; tracing::info!( + auction_id = auction.id, tip, deadline, winners = ranking.inner.winners().count(), ranked = ranking.inner.ranked.len(), filtered_out = ranking.inner.filtered_out.len(), - "competition ranked" + "competition persisted" ); Ok(()) } diff --git a/crates/autopilot-svm/src/infra/order_events.rs b/crates/autopilot-svm/src/infra/order_events.rs index 73789f0830..de2c420131 100644 --- a/crates/autopilot-svm/src/infra/order_events.rs +++ b/crates/autopilot-svm/src/infra/order_events.rs @@ -14,6 +14,16 @@ const DEDUP_LOCK: &str = "solana_order_events_dedup"; const DEDUP_LOCK_QUERY: &str = "SELECT pg_advisory_xact_lock(hashtextextended($1::text || $2::text, 0))"; +/// Append the events off the caller's path: a failed write is logged, not +/// returned. +pub fn store_detached(pool: PgPool, uids: Vec, label: OrderEventLabel) { + tokio::spawn(async move { + if let Err(err) = store(&pool, uids, label).await { + tracing::error!(?err, ?label, "failed to store order events"); + } + }); +} + /// Append one event per order, skipping a label the order's latest event /// already carries: a looping order marks each state once, not once per cycle. pub async fn store( diff --git a/crates/autopilot-svm/src/infra/provider.rs b/crates/autopilot-svm/src/infra/provider.rs index 56605bb011..5bc8d16a73 100644 --- a/crates/autopilot-svm/src/infra/provider.rs +++ b/crates/autopilot-svm/src/infra/provider.rs @@ -3,17 +3,18 @@ use { crate::{ domain::{auction::Order, cycle::SolanaCycle}, - infra::{db, inflight::InFlightOrders, prices::NativePrices}, + infra::{db, order_events, prices::NativePrices}, run_loop::AuctionProvider, }, async_trait::async_trait, - chain_types::solana::Pubkey as ChainPubkey, + chain_types::solana::{IntentHash, Pubkey as ChainPubkey}, cow_solana_rpc::SolanaRPC, + database::solana::OrderEventLabel, solana_sdk::{account::Account, program_pack::Pack, pubkey::Pubkey}, spl_token_interface::state::{Account as TokenAccount, AccountState}, sqlx::PgPool, std::{ - sync::atomic::{AtomicI64, Ordering}, + collections::HashSet, time::{SystemTime, UNIX_EPOCH}, }, }; @@ -24,33 +25,16 @@ pub struct DbAuctionProvider { rpc: SolanaRPC, /// Slots the indexer may lag behind the tip before cuts are skipped. max_indexer_lag: u64, - /// Orders with a settlement in flight, excluded from cuts until their - /// submission deadline passes. - inflight: InFlightOrders, prices: NativePrices, - /// 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, - rpc: SolanaRPC, - max_indexer_lag: u64, - inflight: InFlightOrders, - prices: NativePrices, - ) -> Self { + pub fn new(pool: PgPool, rpc: SolanaRPC, max_indexer_lag: u64, prices: NativePrices) -> Self { Self { pool, rpc, max_indexer_lag, - inflight, prices, - last_id: AtomicI64::new(0), } } @@ -97,18 +81,6 @@ impl DbAuctionProvider { }) .collect() } - - /// 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 { @@ -156,31 +128,46 @@ impl AuctionProvider for DbAuctionProvider { None } }; - let mut auction = db::cut(&self.pool, self.next_id(now), now, block_height) + let orders = db::cut(&self.pool, now, block_height) .await .map_err(|err| tracing::warn!(?err, "failed to cut the auction")) .ok()?; // An order with a settlement in flight stays out until the // settlement cannot land any more: a second winner could - // double-settle it. - let held = self.inflight.held_at(*tip); - let before = auction.orders.len(); - auction.orders.retain(|order| !held.contains(&order.uid)); - let held_out = before - auction.orders.len(); - if held_out > 0 { + // double-settle it. A failed read skips the cut rather than cutting + // without the hold. + let tip_slot = i64::try_from(*tip).unwrap_or(i64::MAX); + let held: HashSet = match db::in_flight_orders(&self.pool, tip_slot).await { + Ok(uids) => uids.into_iter().map(|uid| IntentHash(uid.0)).collect(), + Err(err) => { + tracing::warn!(?err, "in-flight order lookup failed, skipping the cut"); + return None; + } + }; + let (orders, held_out): (Vec<_>, Vec<_>) = orders + .into_iter() + .partition(|order| !held.contains(&order.uid)); + if !held_out.is_empty() { metrics() .held_out_orders - .inc_by(u64::try_from(held_out).unwrap_or(u64::MAX)); - tracing::debug!(held_out, "orders held out with settlements in flight"); + .inc_by(u64::try_from(held_out.len()).unwrap_or(u64::MAX)); + tracing::debug!( + held_out = held_out.len(), + "orders held out with settlements in flight" + ); + order_events::store_detached( + self.pool.clone(), + held_out.into_iter().map(|order| order.uid).collect(), + OrderEventLabel::Filtered, + ); } - auction.orders = self.receivable_orders(auction.orders).await; - if auction.orders.is_empty() { + let orders = self.receivable_orders(orders).await; + if orders.is_empty() { return None; } // A cut without prices would rank solutions on incomparable scores, // so a failed lookup skips the cycle instead. - let tokens = auction - .orders + let tokens = orders .iter() .flat_map(|order| [order.sell_token, order.buy_token]) .map(|token| Pubkey::new_from_array(token.0)) @@ -192,14 +179,52 @@ impl AuctionProvider for DbAuctionProvider { return None; } }; - auction.native_prices = prices - .into_iter() - .map(|(token, price)| (ChainPubkey(token.to_bytes()), price)) - .collect(); + let mut auction = crate::domain::auction::Auction { + id: 0, + orders, + native_prices: prices + .into_iter() + .map(|(token, price)| (ChainPubkey(token.to_bytes()), price)) + .collect(), + }; + // The id must be durable before anything references it: windows and + // the competition snapshot key on it, so a failed write skips the + // cycle. + let snapshot = auction_snapshot(*tip, &auction); + auction.id = match db::replace_current_auction(&self.pool, tip_slot, &snapshot).await { + Ok(id) => id, + Err(err) => { + tracing::warn!(?err, "failed to store the auction, skipping the cut"); + return None; + } + }; Some(auction) } } +/// The stored auction body: the solver-facing content without the deadline, +/// which is only known at dispatch. +fn auction_snapshot(tip: u64, auction: &crate::domain::auction::Auction) -> serde_json::Value { + serde_json::json!({ + "tipSlot": tip, + "orders": auction + .orders + .iter() + .map(|order| order.uid.to_string()) + .collect::>(), + "nativePrices": auction + .native_prices + .iter() + .map(|(token, price)| { + ( + Pubkey::new_from_array(token.0).to_string(), + price.to_string(), + ) + }) + .collect::>(), + }) +} + #[derive(prometheus_metric_storage::MetricStorage)] #[metric(subsystem = "auction_provider")] struct Metrics { @@ -273,7 +298,6 @@ mod tests { sqlx::PgPool::connect_lazy("postgresql://").unwrap(), SolanaRPC::new_mock_with_mocks(mocks), 150, - InFlightOrders::default(), NativePrices::seeded([]), ) } diff --git a/crates/autopilot-svm/src/run.rs b/crates/autopilot-svm/src/run.rs index 7a7ce386e5..7492a53c2f 100644 --- a/crates/autopilot-svm/src/run.rs +++ b/crates/autopilot-svm/src/run.rs @@ -9,7 +9,6 @@ use { db, driver::Driver, executor::DriverExecutor, - inflight::InFlightOrders, listen::ListenSession, observation::SettlementWindows, observer::CompetitionObserver, @@ -89,10 +88,7 @@ async fn run(config: Config) { .await .expect("database connection"); - // One shared hold-out: the executor holds into it, the auction cut reads - // it, and the settlement observer releases from it. - let inflight = InFlightOrders::default(); - let windows = SettlementWindows::new(pool.clone(), inflight.clone()); + let windows = SettlementWindows::new(pool.clone()); let listen = ListenSession::spawn( pool.clone(), db::SETTLEMENT_FINALIZED_CHANNEL, @@ -135,7 +131,6 @@ async fn run(config: Config) { CommitmentConfig::confirmed(), ), config.max_indexer_lag_slots, - inflight.clone(), NativePrices::new( &config.native_prices, SolanaRPC::new_with_timeout_and_commitment( @@ -154,12 +149,7 @@ async fn run(config: Config) { config.competition.max_winners.get(), Pubkey(config.contracts.wrapped_native_mint.to_bytes()), )), - Box::new(DriverExecutor::new( - drivers, - windows.clone(), - sponsor, - inflight, - )), + Box::new(DriverExecutor::new(drivers, windows.clone(), sponsor)), Box::new(CompetitionObserver::new(pool, windows)), config.competition.submission_deadline_slots.get(), ); diff --git a/crates/autopilot-svm/src/test_db.rs b/crates/autopilot-svm/src/test_db.rs index 84879c3498..5f7b662f80 100644 --- a/crates/autopilot-svm/src/test_db.rs +++ b/crates/autopilot-svm/src/test_db.rs @@ -11,7 +11,9 @@ pub(crate) async fn pool() -> PgPool { 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, solana.order_events", + solana.order_pda, solana.orders, solana.indexer_state, solana.order_events, \ + solana.auctions, solana.competition_auctions, solana.proposed_solutions, \ + solana.proposed_trade_executions, solana.reference_scores", ) .execute(pool) .await diff --git a/crates/autopilot-svm/src/tests.rs b/crates/autopilot-svm/src/tests.rs index 11f0e62648..9cea0defdd 100644 --- a/crates/autopilot-svm/src/tests.rs +++ b/crates/autopilot-svm/src/tests.rs @@ -8,7 +8,6 @@ use { competition::DriverCompetition, driver::{Driver, dto}, executor::DriverExecutor, - inflight::InFlightOrders, observation::SettlementWindows, observer::CompetitionObserver, prices::NativePrices, @@ -215,7 +214,6 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { pool.clone(), mock_rpc(), 150, - InFlightOrders::default(), NativePrices::seeded(test_prices()), ); let auction = provider.cut_auction(&tip).await.expect("auction cut"); @@ -227,15 +225,13 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { assert_eq!(ranking.winner_count(), 1, "solution won"); } - let inflight = InFlightOrders::default(); - let windows = SettlementWindows::new(pool.clone(), inflight.clone()); + let windows = SettlementWindows::new(pool.clone()); let mut auction_loop = AuctionLoop::new( Box::new(FixedTrigger(tip)), Box::new(DbAuctionProvider::new( pool.clone(), mock_rpc(), 150, - inflight.clone(), NativePrices::seeded(test_prices()), )), Box::new(DriverCompetition::new( @@ -243,12 +239,7 @@ 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], - windows.clone(), - None, - inflight.clone(), - )), + Box::new(DriverExecutor::new(vec![driver], windows.clone(), None)), Box::new(CompetitionObserver::new(pool.clone(), windows.clone())), 25, ); @@ -260,37 +251,88 @@ 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); + // The competition was persisted: the snapshot row, the proposed solution + // under its generated uid, its execution, and the window keyed by the + // same uid. + let snapshots: i64 = sqlx::query_scalar("SELECT count(*) FROM solana.competition_auctions") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(snapshots, 1); + let (solution_uid, solver_id, is_winner, filtered_out): (i64, i64, bool, bool) = + sqlx::query_as("SELECT uid, id, is_winner, filtered_out FROM solana.proposed_solutions") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + (solution_uid, solver_id, is_winner, filtered_out), + (0, 7, true, false) + ); + // The sole winner's reference score: the winners' total without it, zero. + let (reference_solver, reference_score): (Vec, String) = + sqlx::query_as("SELECT solver, reference_score::text FROM solana.reference_scores") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + (reference_solver.as_slice(), reference_score.as_str()), + (&[0xCC; 32][..], "0") + ); + let (executed_sell, executed_buy): (String, String) = sqlx::query_as( + "SELECT executed_sell::text, executed_buy::text FROM solana.proposed_trade_executions \ + WHERE solution_uid = 0", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + (executed_sell.as_str(), executed_buy.as_str()), + ("1000", "600") + ); + let window_uid: i64 = + sqlx::query_scalar("SELECT solution_uid FROM solana.settlement_executions") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(window_uid, 0); // The dispatched order is held out of the next cut until its settlement // transaction cannot land any more: the deadline plus the blockhash - // lifetime. + // lifetime. The hold comes from the persisted competition alone, so a + // fresh provider also stands in for a restart. The lag gate stays out of + // the way, the watermark is still at the dispatch tip. let expired = tip + 25 + solana_sdk::clock::MAX_PROCESSING_AGE as u64 + 1; - // The lag gate stays out of the way: this provider tests the hold, and - // the watermark is still at the dispatch tip. let held_provider = DbAuctionProvider::new( pool.clone(), mock_rpc(), u64::MAX, - inflight.clone(), NativePrices::seeded(test_prices()), ); assert!( - held_provider.cut_auction(&(expired - 1)).await.is_none(), + held_provider.cut_auction(&(tip + 25)).await.is_none(), "in-flight order excluded from the cut" ); + // The deadline sweep closes the window as timed out. The hold outlasts + // it by the blockhash lifetime. + windows.expire_past_deadline(tip + 25).await.unwrap(); + assert!( + held_provider.cut_auction(&(expired - 1)).await.is_none(), + "order stays held while its transaction can still land" + ); assert!( held_provider.cut_auction(&expired).await.is_some(), "order returns once the transaction cannot land" ); - // 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); - // The cycle reported the order's auction progress. The writes are detached - // from the cycle, so they can land after `run_cycle` returns. + // The cycle reported the order's auction progress and the later cuts its + // hold-out. The writes are detached, so they can land after the cuts + // return. let events = tokio::time::timeout(Duration::from_secs(5), async { loop { let mut events: Vec = @@ -298,7 +340,7 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { .fetch_all(&pool) .await .unwrap(); - if events.len() == 2 { + if events.len() == 3 { events.sort(); return events; } @@ -307,5 +349,5 @@ async fn solana_db_mock_cycle_dispatches_the_settlement() { }) .await .expect("order events written before the timeout"); - assert_eq!(events, ["executing", "ready"]); + assert_eq!(events, ["executing", "filtered", "ready"]); } diff --git a/database/sql-solana/V8__competition.sql b/database/sql-solana/V8__competition.sql new file mode 100644 index 0000000000..c913214ecb --- /dev/null +++ b/database/sql-solana/V8__competition.sql @@ -0,0 +1,60 @@ +-- The current auction, replaced on every cut. Its identity column is the +-- auction id sequence, so ids stay sequential across restarts like the EVM +-- auctions table. +CREATE TABLE solana.auctions ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + tip_slot bigint NOT NULL, + json jsonb NOT NULL, + -- Pins the table to one row: a second writer's insert fails instead of + -- leaving two current auctions. + singleton boolean NOT NULL DEFAULT true UNIQUE CHECK (singleton) +); + +-- Auctions that ran a competition, snapshot at ranking time. +CREATE TABLE solana.competition_auctions ( + id bigint PRIMARY KEY, + tip_slot bigint NOT NULL, + deadline_slot bigint NOT NULL, + order_uids bytea[] NOT NULL, + price_tokens bytea[] NOT NULL, + price_values numeric(20,0)[] NOT NULL +); + +-- The in-flight hold-out scans the auctions whose deadline has not passed. +CREATE INDEX solana_competition_auctions_deadline_slot + ON solana.competition_auctions (deadline_slot); + +-- Every solution proposed during a competition. The autopilot generates +-- `uid` per auction, disambiguating the solver-assigned `id` across drivers. +-- No clearing price columns: the EVM twin writes its own empty, and the +-- Solana solve wire carries none. +CREATE TABLE solana.proposed_solutions ( + auction_id bigint NOT NULL, + uid bigint NOT NULL, + id bigint NOT NULL, + solver bytea NOT NULL CHECK (length(solver) = 32), + is_winner boolean NOT NULL, + -- Excluded from the ranking by the fairness filter, never a winner. + filtered_out boolean NOT NULL, + score numeric(20,0) NOT NULL, + PRIMARY KEY (auction_id, uid) +); + +-- The order executions of every proposed solution. +CREATE TABLE solana.proposed_trade_executions ( + auction_id bigint NOT NULL, + solution_uid bigint NOT NULL, + order_uid bytea NOT NULL CHECK (length(order_uid) = 32), + executed_sell numeric(20,0) NOT NULL, + executed_buy numeric(20,0) NOT NULL, + PRIMARY KEY (auction_id, solution_uid, order_uid) +); + +-- Per winning solver, the winners' total score with that solver's solutions +-- removed: the rewards baseline. +CREATE TABLE solana.reference_scores ( + auction_id bigint NOT NULL, + solver bytea NOT NULL CHECK (length(solver) = 32), + reference_score numeric(20,0) NOT NULL, + PRIMARY KEY (auction_id, solver) +);