|
| 1 | +//! This module implements consensus-based quoter mode. |
| 2 | +use std::{ |
| 3 | + collections::{HashMap, HashSet}, |
| 4 | + pin::Pin, |
| 5 | + sync::Arc, |
| 6 | + task::Poll, |
| 7 | + time::Duration |
| 8 | +}; |
| 9 | + |
| 10 | +use angstrom_types::{ |
| 11 | + block_sync::BlockSyncConsumer, |
| 12 | + consensus::{ConsensusRoundEvent, ConsensusRoundOrderHashes}, |
| 13 | + orders::OrderSet, |
| 14 | + primitive::PoolId, |
| 15 | + sol_bindings::{grouped_orders::AllOrders, rpc_orders::TopOfBlockOrder} |
| 16 | +}; |
| 17 | +use futures::{Stream, StreamExt, stream::FuturesUnordered}; |
| 18 | +use order_pool::order_storage::OrderStorage; |
| 19 | +use rayon::ThreadPool; |
| 20 | +use tokio::{sync::mpsc, time::interval}; |
| 21 | +use uniswap_v4::uniswap::pool_manager::SyncedUniswapPools; |
| 22 | + |
| 23 | +use crate::{QuoterManager, Slot0Update, book_snapshots_from_amms}; |
| 24 | + |
| 25 | +/// A type alias for the consensus quoter manager. |
| 26 | +pub type ConsensusQuoterManager<BlockSync> = QuoterManager<BlockSync, ConsensusMode>; |
| 27 | + |
| 28 | +/// Mode for consensus-based order book building. |
| 29 | +pub struct ConsensusMode { |
| 30 | + consensus_stream: Pin<Box<dyn Stream<Item = ConsensusRoundOrderHashes> + Send>>, |
| 31 | + /// The unique order hashes of the current PreProposalAggregate consensus |
| 32 | + /// round. Used to build the book for the slot0 stream, so that all |
| 33 | + /// orders are valid, and the subscription can't be manipulated by orders |
| 34 | + /// submitted after this round and between the next block |
| 35 | + active_pre_proposal_aggr_order_hashes: Option<ConsensusRoundOrderHashes> |
| 36 | +} |
| 37 | + |
| 38 | +impl<BlockSync: BlockSyncConsumer> QuoterManager<BlockSync, ConsensusMode> { |
| 39 | + /// ensure that we haven't registered on the BlockSync. |
| 40 | + /// We just want to ensure that we don't access during a update period |
| 41 | + pub fn new( |
| 42 | + block_sync: BlockSync, |
| 43 | + orders: Arc<OrderStorage>, |
| 44 | + recv: mpsc::Receiver<(HashSet<PoolId>, mpsc::Sender<Slot0Update>)>, |
| 45 | + amms: SyncedUniswapPools, |
| 46 | + threadpool: ThreadPool, |
| 47 | + update_interval: Duration, |
| 48 | + consensus_stream: Pin<Box<dyn Stream<Item = ConsensusRoundOrderHashes> + Send>> |
| 49 | + ) -> Self { |
| 50 | + let cur_block = block_sync.current_block_number(); |
| 51 | + let book_snapshots = book_snapshots_from_amms(&amms); |
| 52 | + |
| 53 | + assert!( |
| 54 | + update_interval > Duration::from_millis(10), |
| 55 | + "cannot update quicker than every 10ms" |
| 56 | + ); |
| 57 | + |
| 58 | + let mode = ConsensusMode { consensus_stream, active_pre_proposal_aggr_order_hashes: None }; |
| 59 | + |
| 60 | + Self { |
| 61 | + seq_id: 0, |
| 62 | + block_sync, |
| 63 | + orders, |
| 64 | + amms, |
| 65 | + recv, |
| 66 | + cur_block, |
| 67 | + book_snapshots, |
| 68 | + threadpool, |
| 69 | + pending_tasks: FuturesUnordered::new(), |
| 70 | + pool_to_subscribers: HashMap::default(), |
| 71 | + execution_interval: interval(update_interval), |
| 72 | + mode |
| 73 | + } |
| 74 | + } |
| 75 | + |
| 76 | + pub(crate) fn all_orders_with_consensus(&self) -> OrderSet<AllOrders, TopOfBlockOrder> { |
| 77 | + if let Some(hashes) = self.mode.active_pre_proposal_aggr_order_hashes.as_ref() { |
| 78 | + self.orders |
| 79 | + .get_all_orders_with_hashes(&hashes.limit, &hashes.searcher) |
| 80 | + } else { |
| 81 | + self.orders.get_all_orders() |
| 82 | + } |
| 83 | + } |
| 84 | + |
| 85 | + fn update_consensus_state(&mut self, round: ConsensusRoundOrderHashes) { |
| 86 | + if matches!(round.round, ConsensusRoundEvent::PropagatePreProposalAgg) { |
| 87 | + self.mode.active_pre_proposal_aggr_order_hashes = Some(round) |
| 88 | + } |
| 89 | + } |
| 90 | +} |
| 91 | + |
| 92 | +impl<BlockSync: BlockSyncConsumer> Future for QuoterManager<BlockSync, ConsensusMode> { |
| 93 | + type Output = (); |
| 94 | + |
| 95 | + fn poll( |
| 96 | + mut self: Pin<&mut Self>, |
| 97 | + cx: &mut std::task::Context<'_> |
| 98 | + ) -> std::task::Poll<Self::Output> { |
| 99 | + while let Poll::Ready(Some((pools, subscriber))) = self.recv.poll_recv(cx) { |
| 100 | + self.handle_new_subscription(pools, subscriber); |
| 101 | + } |
| 102 | + |
| 103 | + while let Poll::Ready(Some(consensus_update)) = |
| 104 | + self.mode.consensus_stream.poll_next_unpin(cx) |
| 105 | + { |
| 106 | + self.update_consensus_state(consensus_update); |
| 107 | + } |
| 108 | + |
| 109 | + while let Poll::Ready(Some(Ok(slot_update))) = self.pending_tasks.poll_next_unpin(cx) { |
| 110 | + self.send_out_result(slot_update); |
| 111 | + } |
| 112 | + |
| 113 | + while self.execution_interval.poll_tick(cx).is_ready() { |
| 114 | + // cycle through if we can't do any processing |
| 115 | + if !self.block_sync.can_operate() { |
| 116 | + cx.waker().wake_by_ref(); |
| 117 | + return Poll::Pending; |
| 118 | + } |
| 119 | + |
| 120 | + // update block number, amm snapshot and reset seq id |
| 121 | + if self.cur_block != self.block_sync.current_block_number() { |
| 122 | + self.update_book_state(); |
| 123 | + self.cur_block = self.block_sync.current_block_number(); |
| 124 | + |
| 125 | + self.mode.active_pre_proposal_aggr_order_hashes = None; |
| 126 | + |
| 127 | + self.seq_id = 0; |
| 128 | + } |
| 129 | + |
| 130 | + // inc seq_id |
| 131 | + let seq_id = self.seq_id; |
| 132 | + // given that we have a max update speed of 10ms, the max |
| 133 | + // this should reach is 1200 before a new block update |
| 134 | + // occurs. Becuase of this, there is no need to check for overflow |
| 135 | + // as 65535 is more than enough |
| 136 | + self.seq_id += 1; |
| 137 | + |
| 138 | + let orders = self.all_orders_with_consensus(); |
| 139 | + self.spawn_book_solvers(seq_id, orders); |
| 140 | + } |
| 141 | + |
| 142 | + Poll::Pending |
| 143 | + } |
| 144 | +} |
0 commit comments