refactor(pool): add pool crate

Cargo.toml

@@ -3,8 +3,10 @@ members = [
     "bin/rundler",
     "bin/tools",
     "crates/rundler",
+    "crates/pool",
     "crates/provider",
     "crates/sim",
+    "crates/task",
     "crates/types",
     "crates/utils"
 ]
@@ -20,15 +22,28 @@ repository = "https://github.com/alchemyplatform/rundler"
 [workspace.dependencies]
 anyhow = "1.0.70"
 async-trait = "0.1.73"
-ethers = "2.0.8"
 cargo-husky = { version = "1", default-features = false, features = ["user-hooks" ] }
+ethers = "2.0.8"
+futures = "0.3.28"
+futures-util = "0.3.28"
+metrics = "0.21.0"
 mockall = "0.11.4"
 parse-display = "0.8.0"
+pin-project = "1.0.12"
+prost = "0.12.0"
 serde = "1.0.160"
 serde_json = "1.0.64"
 rand = "0.8.5"
 reqwest = { version = "0.11.18", default-features = false, features = ["rustls-tls"] }
 thiserror = "1.0.40"
 tokio = { version = "1.27.0", default-features = false }
+tokio-util = "0.7.8"
+tonic = "0.10.0"
+tonic-build = "0.10.0"
+tonic-health = "0.10.0"
+tonic-reflection = "0.10.0"
+tonic-types = "0.10.0"
+tower = "0.4.13"
 tracing = "0.1.37"
+strum = "0.25.0"
 url = "2.3.1"

buf.work.yaml

@@ -1,3 +1,4 @@
 version: v1
 directories:
   - crates/rundler/proto
+  - crates/pool/proto

crates/pool/Cargo.toml

@@ -0,0 +1,47 @@
+[package]
+name = "rundler-pool"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+rundler-provider = { path = "../provider" }
+rundler-sim = { path = "../sim" }
+rundler-task = { path = "../task" }
+rundler-types = { path = "../types" }
+rundler-utils = { path = "../utils" }
+
+anyhow.workspace = true
+async-stream = "0.3.5"
+async-trait.workspace = true
+ethers.workspace = true
+futures.workspace = true
+futures-util.workspace = true
+itertools = "0.11.0"
+metrics.workspace = true
+parking_lot = "0.12.1"
+prost.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tokio-stream = { version = "0.1.12", features = ["sync"] }
+tokio-util.workspace = true
+tonic.workspace = true
+tonic-health.workspace = true
+tonic-reflection.workspace = true
+tracing.workspace = true
+serde.workspace = true
+strum.workspace = true
+url.workspace = true
+
+mockall = {workspace = true, optional = true }
+
+[dev-dependencies]
+mockall.workspace = true
+
+[build-dependencies]
+tonic-build.workspace = true
+
+[features]
+test-utils = [ "mockall" ]

crates/pool/build.rs

@@ -0,0 +1,15 @@
+use std::{env, error, path::PathBuf};
+
+fn main() -> Result<(), Box<dyn error::Error>> {
+    println!("cargo:rerun-if-changed=proto");
+    generate_protos()?;
+    Ok(())
+}
+
+fn generate_protos() -> Result<(), Box<dyn error::Error>> {
+    let out_dir = PathBuf::from(env::var("OUT_DIR")?);
+    tonic_build::configure()
+        .file_descriptor_set_path(out_dir.join("op_pool_descriptor.bin"))
+        .compile(&["proto/op_pool/op_pool.proto"], &["proto"])?;
+    Ok(())
+}

crates/pool/proto/README.md

@@ -0,0 +1 @@
+# TODO

crates/pool/proto/buf.yaml

@@ -0,0 +1,26 @@
+version: v1
+name: ""
+deps: []
+build:
+  excludes: []
+lint:
+  use:
+    - DEFAULT
+  except:
+    - PACKAGE_VERSION_SUFFIX
+    - SERVICE_SUFFIX
+  ignore: []
+  ignore_only: {}
+  allow_comment_ignores: false
+  enum_zero_value_suffix: _UNSPECIFIED
+  rpc_allow_same_request_response: false
+  rpc_allow_google_protobuf_empty_requests: false
+  rpc_allow_google_protobuf_empty_responses: false
+  service_suffix: Service
+breaking:
+  use:
+    - FILE
+  except: []
+  ignore: []
+  ignore_only: {}
+  ignore_unstable_packages: false

crates/rundler/src/op_pool/chain.rs → crates/pool/src/chain.rs

@@ -12,6 +12,7 @@ use ethers::{
 };
 use futures::future;
 use rundler_provider::Provider;
+use rundler_task::block_watcher;
 use rundler_types::contracts::i_entry_point::UserOperationEventFilter;
 use tokio::{
     select,
@@ -21,8 +22,6 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
-use crate::common::block_watcher;
-
 const MAX_LOAD_OPS_CONCURRENCY: usize = 64;
 
 /// A data structure that holds the currently known recent state of the chain,
@@ -31,7 +30,7 @@ const MAX_LOAD_OPS_CONCURRENCY: usize = 64;
 /// Will update itself when `.sync_to_block_number` is called, at which point it
 /// will query a node to determine the new state of the chain.
 #[derive(Debug)]
-pub struct Chain<P: Provider> {
+pub(crate) struct Chain<P: Provider> {
     provider: Arc<P>,
     settings: Settings,
     /// Blocks are stored from earliest to latest, so the oldest block is at the
@@ -62,22 +61,22 @@ pub struct MinedOp {
 }
 
 #[derive(Debug)]
-pub struct Settings {
-    pub history_size: u64,
-    pub poll_interval: Duration,
-    pub entry_point_addresses: Vec<Address>,
+pub(crate) struct Settings {
+    pub(crate) history_size: u64,
+    pub(crate) poll_interval: Duration,
+    pub(crate) entry_point_addresses: Vec<Address>,
 }
 
 #[derive(Debug)]
 struct BlockSummary {
-    pub number: u64,
-    pub hash: H256,
-    pub parent_hash: H256,
-    pub ops: Vec<MinedOp>,
+    number: u64,
+    hash: H256,
+    parent_hash: H256,
+    ops: Vec<MinedOp>,
 }
 
 impl<P: Provider> Chain<P> {
-    pub fn new(provider: Arc<P>, settings: Settings) -> Self {
+    pub(crate) fn new(provider: Arc<P>, settings: Settings) -> Self {
         let history_size = settings.history_size as usize;
         assert!(history_size > 0, "history size should be positive");
         Self {
@@ -88,7 +87,7 @@ impl<P: Provider> Chain<P> {
         }
     }
 
-    pub fn spawn_watcher(
+    pub(crate) fn spawn_watcher(
         mut self,
         sender: broadcast::Sender<Arc<ChainUpdate>>,
         shutdown_token: CancellationToken,
@@ -132,7 +131,10 @@ impl<P: Provider> Chain<P> {
         }
     }
 
-    pub async fn sync_to_block(&mut self, new_head: Block<H256>) -> anyhow::Result<ChainUpdate> {
+    pub(crate) async fn sync_to_block(
+        &mut self,
+        new_head: Block<H256>,
+    ) -> anyhow::Result<ChainUpdate> {
         let new_head = BlockSummary::try_from_block_without_ops(new_head, None)?;
         let Some(current_block) = self.blocks.back() else {
             return self.reset_and_initialize(new_head).await;