feat: introduce op-angstrom, genericize repo for OP

Cargo.toml

@@ -89,6 +89,7 @@ alloy-transport-ws = { version = "1.0.23", default-features = false }
 alloy-trie = { version = "0.9.0", default-features = false }
 angstrom = { path = "./bin/angstrom/" }
 angstrom-amm-quoter = { path = "./crates/amm-quoter/" }
+angstrom-cli = { path = "./crates/cli/" }
 angstrom-eth = { path = "./crates/eth/" }
 angstrom-metrics = { path = "./crates/metrics/" }
 angstrom-network = { path = "./crates/angstrom-net/" }
@@ -98,6 +99,7 @@ angstrom-utils = { path = "./crates/utils/" }
 anyhow = "1"
 aquamarine = "0"
 arbitrary = "1"
+arraydeque = "0"
 assert_matches = "1"
 async-trait = "0"
 auto_impl = "1"
@@ -117,7 +119,9 @@ divan = "0"
 enr = { version = "0", default-features = false, features = ["k256"] }
 eyre = "0"
 futures = "0"
+futures-core = "0"
 futures-util = "0"
+hex = "0.4.3"
 hex-literal = "0"
 homedir = "0"
 hsm-signer = { git = "https://github.com/SorellaLabs/hsm-signer" }
@@ -127,6 +131,7 @@ jsonrpsee-core = "0.25.1"
 jsonrpsee-http-client = "0.25.1"
 jsonrpsee-types = "0.25.1"
 k256 = { version = "0", default-features = false }
+linked_hash_set = "0"
 malachite = "0.4"
 malachite-q = "0.4"
 matching-engine = { path = "./crates/matching-engine/" }
@@ -137,6 +142,7 @@ once_cell = "1"
 order-pool = { path = "./crates/order-pool/" }
 pade = { git = "https://github.com/SorellaLabs/pade", version = "0.1.0" }
 pade-macro = { git = "https://github.com/SorellaLabs/pade", version = "0.1.0" }
+pool-manager = { path = "./crates/pool-manager/" }
 parking_lot = "0"
 pin-project = "1"
 pprof = "0"
@@ -216,3 +222,9 @@ uniswap-v4 = { path = "./crates/uniswap-v4/" }
 uniswap_v3_math = { git = "https://github.com/SorellaLabs/v3-math", branch = "main" }
 url = "2"
 validation = { path = "./crates/validation/" }
+
+# OP-stack support
+reth-optimism-chainspec = { git = "https://github.com/paradigmxyz/reth", version = "1.6.0", tag = "v1.6.0" }
+reth-optimism-cli = { git = "https://github.com/paradigmxyz/reth", version = "1.6.0", tag = "v1.6.0" }
+reth-optimism-node = { git = "https://github.com/paradigmxyz/reth", version = "1.6.0", tag = "v1.6.0" }
+reth-optimism-primitives = { git = "https://github.com/paradigmxyz/reth", version = "1.6.0", tag = "v1.6.0" }

bin/angstrom/Cargo.toml

@@ -11,40 +11,7 @@ Angstrom Sidecar implementation
 """
 
 [dependencies]
-alloy.workspace = true
-alloy-chains.workspace = true
-alloy-primitives.workspace = true
-angstrom-amm-quoter.workspace = true
-angstrom-eth.workspace = true
-angstrom-metrics.workspace = true
-angstrom-network.workspace = true
-angstrom-rpc.workspace = true
-angstrom-types.workspace = true
-clap.workspace = true
-consensus.workspace = true
-eyre.workspace = true
-futures.workspace = true
-hsm-signer.workspace = true
-matching-engine.workspace = true
-order-pool.workspace = true
-parking_lot.workspace = true
-rayon = "1"
-reth.workspace = true
-reth-db.workspace = true
-reth-metrics.workspace = true
-reth-network.workspace = true
-reth-node-builder.workspace = true
-reth-node-ethereum.workspace = true
-reth-node-metrics.workspace = true
-reth-provider.workspace = true
-serde.workspace = true
-telemetry.workspace = true
-tokio.workspace = true
-toml.workspace = true
-tracing.workspace = true
-uniswap-v4.workspace = true
-url.workspace = true
-validation.workspace = true
+angstrom-cli.workspace = true
 
 [target.'cfg(unix)'.dependencies]
 tikv-jemallocator = { version = "0.6.0", optional = true }

bin/angstrom/src/main.rs

@@ -4,7 +4,7 @@
 static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
 fn main() {
-    if let Err(err) = angstrom::run() {
+    if let Err(err) = angstrom_cli::angstrom::run() {
         eprintln!("Error: {err:?}");
         std::process::exit(1);
     }

bin/op-angstrom/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "op-angstrom"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+description = """
+Angstrom Sidecar implementation
+"""
+
+[dependencies]
+angstrom-cli.workspace = true
+
+[target.'cfg(unix)'.dependencies]
+tikv-jemallocator = { version = "0.6.0", optional = true }
+
+[features]
+default = ["jemalloc"]
+jemalloc = ["dep:tikv-jemallocator"]
+jemalloc-prof = ["jemalloc", "tikv-jemallocator?/profiling"]
+
+[[bin]]
+name = "op-angstrom"
+path = "src/main.rs"

bin/op-angstrom/src/main.rs

@@ -0,0 +1,11 @@
+// We use jemalloc for performance reasons
+#[cfg(all(feature = "jemalloc", unix))]
+#[global_allocator]
+static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+
+fn main() {
+    if let Err(err) = angstrom_cli::op_angstrom::run() {
+        eprintln!("Error: {err:?}");
+        std::process::exit(1);
+    }
+}

bin/testnet/Cargo.toml

@@ -13,7 +13,6 @@ alloy.workspace = true
 alloy-primitives.workspace = true
 alloy-rpc-types.workspace = true
 alloy-signer-local.workspace = true
-angstrom.workspace = true
 angstrom-metrics.workspace = true
 angstrom-rpc.workspace = true
 angstrom-types = { workspace = true }

crates/amm-quoter/Cargo.toml

@@ -20,3 +20,7 @@ serde.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 uniswap-v4.workspace = true
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
+alloy-primitives.workspace = true

crates/amm-quoter/src/consensus.rs

@@ -0,0 +1,144 @@
+//! This module implements consensus-based quoter mode.
+use std::{
+    collections::{HashMap, HashSet},
+    pin::Pin,
+    sync::Arc,
+    task::Poll,
+    time::Duration
+};
+
+use angstrom_types::{
+    block_sync::BlockSyncConsumer,
+    consensus::{ConsensusRoundEvent, ConsensusRoundOrderHashes},
+    orders::OrderSet,
+    primitive::PoolId,
+    sol_bindings::{grouped_orders::AllOrders, rpc_orders::TopOfBlockOrder}
+};
+use futures::{Stream, StreamExt, stream::FuturesUnordered};
+use order_pool::order_storage::OrderStorage;
+use rayon::ThreadPool;
+use tokio::{sync::mpsc, time::interval};
+use uniswap_v4::uniswap::pool_manager::SyncedUniswapPools;
+
+use crate::{QuoterManager, Slot0Update, book_snapshots_from_amms};
+
+/// A type alias for the consensus quoter manager.
+pub type ConsensusQuoterManager<BlockSync> = QuoterManager<BlockSync, ConsensusMode>;
+
+/// Mode for consensus-based order book building.
+pub struct ConsensusMode {
+    consensus_stream: Pin<Box<dyn Stream<Item = ConsensusRoundOrderHashes> + Send>>,
+    /// The unique order hashes of the current PreProposalAggregate consensus
+    /// round. Used to build the book for the slot0 stream, so that all
+    /// orders are valid, and the subscription can't be manipulated by orders
+    /// submitted after this round and between the next block
+    active_pre_proposal_aggr_order_hashes: Option<ConsensusRoundOrderHashes>
+}
+
+impl<BlockSync: BlockSyncConsumer> QuoterManager<BlockSync, ConsensusMode> {
+    /// ensure that we haven't registered on the BlockSync.
+    /// We just want to ensure that we don't access during a update period
+    pub fn new(
+        block_sync: BlockSync,
+        orders: Arc<OrderStorage>,
+        recv: mpsc::Receiver<(HashSet<PoolId>, mpsc::Sender<Slot0Update>)>,
+        amms: SyncedUniswapPools,
+        threadpool: ThreadPool,
+        update_interval: Duration,
+        consensus_stream: Pin<Box<dyn Stream<Item = ConsensusRoundOrderHashes> + Send>>
+    ) -> Self {
+        let cur_block = block_sync.current_block_number();
+        let book_snapshots = book_snapshots_from_amms(&amms);
+
+        assert!(
+            update_interval > Duration::from_millis(10),
+            "cannot update quicker than every 10ms"
+        );
+
+        let mode = ConsensusMode { consensus_stream, active_pre_proposal_aggr_order_hashes: None };
+
+        Self {
+            seq_id: 0,
+            block_sync,
+            orders,
+            amms,
+            recv,
+            cur_block,
+            book_snapshots,
+            threadpool,
+            pending_tasks: FuturesUnordered::new(),
+            pool_to_subscribers: HashMap::default(),
+            execution_interval: interval(update_interval),
+            mode
+        }
+    }
+
+    pub(crate) fn all_orders_with_consensus(&self) -> OrderSet<AllOrders, TopOfBlockOrder> {
+        if let Some(hashes) = self.mode.active_pre_proposal_aggr_order_hashes.as_ref() {
+            self.orders
+                .get_all_orders_with_hashes(&hashes.limit, &hashes.searcher)
+        } else {
+            self.orders.get_all_orders()
+        }
+    }
+
+    fn update_consensus_state(&mut self, round: ConsensusRoundOrderHashes) {
+        if matches!(round.round, ConsensusRoundEvent::PropagatePreProposalAgg) {
+            self.mode.active_pre_proposal_aggr_order_hashes = Some(round)
+        }
+    }
+}
+
+impl<BlockSync: BlockSyncConsumer> Future for QuoterManager<BlockSync, ConsensusMode> {
+    type Output = ();
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>
+    ) -> std::task::Poll<Self::Output> {
+        while let Poll::Ready(Some((pools, subscriber))) = self.recv.poll_recv(cx) {
+            self.handle_new_subscription(pools, subscriber);
+        }
+
+        while let Poll::Ready(Some(consensus_update)) =
+            self.mode.consensus_stream.poll_next_unpin(cx)
+        {
+            self.update_consensus_state(consensus_update);
+        }
+
+        while let Poll::Ready(Some(Ok(slot_update))) = self.pending_tasks.poll_next_unpin(cx) {
+            self.send_out_result(slot_update);
+        }
+
+        while self.execution_interval.poll_tick(cx).is_ready() {
+            // cycle through if we can't do any processing
+            if !self.block_sync.can_operate() {
+                cx.waker().wake_by_ref();
+                return Poll::Pending;
+            }
+
+            // update block number, amm snapshot and reset seq id
+            if self.cur_block != self.block_sync.current_block_number() {
+                self.update_book_state();
+                self.cur_block = self.block_sync.current_block_number();
+
+                self.mode.active_pre_proposal_aggr_order_hashes = None;
+
+                self.seq_id = 0;
+            }
+
+            // inc seq_id
+            let seq_id = self.seq_id;
+            // given that we have a max update speed of 10ms, the max
+            // this should reach is 1200 before a new block update
+            // occurs. Becuase of this, there is no need to check for overflow
+            // as 65535 is more than enough
+            self.seq_id += 1;
+
+            let orders = self.all_orders_with_consensus();
+            self.spawn_book_solvers(seq_id, orders);
+        }
+
+        Poll::Pending
+    }
+}