feat(katana): modify DevApi trait with new endpoints

crates/katana/node/src/lib.rs

@@ -34,6 +34,7 @@ use katana_pool::TxPool;
 use katana_primitives::env::{CfgEnv, FeeTokenAddressses};
 use katana_rpc::dev::DevApi;
 use katana_rpc::metrics::RpcServerMetrics;
+use katana_rpc::proxy_get_request::DevnetProxyLayer;
 use katana_rpc::saya::SayaApi;
 use katana_rpc::starknet::forking::ForkedClient;
 use katana_rpc::starknet::StarknetApi;
@@ -308,6 +309,8 @@ pub async fn spawn<EF: ExecutorFactory>(
     let middleware = tower::ServiceBuilder::new()
         .option_layer(cors)
         .layer(ProxyGetRequestLayer::new("/", "health")?)
+        .layer(DevnetProxyLayer::new("/account_balance", "dev_accountBalance")?)
+        .layer(DevnetProxyLayer::new("/fee_token", "dev_feeToken")?)
         .timeout(Duration::from_secs(20));
 
     let server = ServerBuilder::new()

crates/katana/rpc/rpc-api/src/dev.rs

@@ -24,4 +24,13 @@ pub trait DevApi {
 
     #[method(name = "predeployedAccounts")]
     async fn predeployed_accounts(&self) -> RpcResult<Vec<Account>>;
+
+    #[method(name = "accountBalance")]
+    async fn account_balance(&self, address: String) -> RpcResult<u128>;
+
+    #[method(name = "feeToken")]
+    async fn fee_token(&self) -> RpcResult<String>;
+
+    #[method(name = "mint")]
+    async fn mint(&self) -> RpcResult<()>;
 }

crates/katana/rpc/rpc/Cargo.toml

@@ -11,6 +11,11 @@ anyhow.workspace = true
 dojo-metrics.workspace = true
 futures.workspace = true
 jsonrpsee = { workspace = true, features = [ "server" ] }
+jsonrpsee-core = { version = "0.16.3", features = [ "server", "soketto", "http-helpers" ] }
+jsonrpsee-types = { version = "0.16.3"}
+hyper.workspace = true
+tower = { workspace = true, features = [ "full" ] }
+http = { version = "0.2.7" }
 katana-core.workspace = true
 katana-executor.workspace = true
 katana-pool.workspace = true
@@ -26,6 +31,17 @@ thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+soketto = { version = "0.7.1", features = ["http"] }
+futures-channel = { version = "0.3.14"}
+futures-util = { version = "0.3.14",  features = [
+    "io",
+    "async-await-macro",
+]}
+tokio-stream = { version = "0.1.7" }
+tokio-util = { version = "0.7", features = ["compat"]}
+starknet-crypto.workspace = true
 
 [dev-dependencies]
 alloy = { git = "https://github.com/alloy-rs/alloy", features = [ "contract", "network", "node-bindings", "provider-http", "providers", "signer-local" ] }

crates/katana/rpc/rpc/src/dev.rs

@@ -1,13 +1,19 @@
+use std::str::FromStr;
 use std::sync::Arc;
 
 use jsonrpsee::core::{async_trait, Error};
 use katana_core::backend::Backend;
 use katana_core::service::block_producer::{BlockProducer, BlockProducerMode, PendingExecutor};
 use katana_executor::ExecutorFactory;
-use katana_primitives::Felt;
+use katana_primitives::genesis::constant::{
+    get_fee_token_balance_base_storage_address, ERC20_NAME_STORAGE_SLOT,
+};
+use katana_primitives::ContractAddress;
+use katana_provider::traits::state::StateFactoryProvider;
 use katana_rpc_api::dev::DevApiServer;
 use katana_rpc_types::account::Account;
 use katana_rpc_types::error::dev::DevApiError;
+use starknet_crypto::Felt;
 
 #[allow(missing_debug_implementations)]
 pub struct DevApi<EF: ExecutorFactory> {
@@ -54,7 +60,6 @@
 
         let mut block_context_generator = self.backend.block_context_generator.write();
         block_context_generator.block_timestamp_offset += offset as i64;
-
         Ok(())
     }
 }
@@ -92,6 +97,35 @@
         Ok(())
     }
 
+    async fn account_balance(&self, address: String) -> Result<u128, Error> {
+        let account_address: ContractAddress = Felt::from_str(&address).unwrap().into();
+        let provider = self.backend.blockchain.provider();
+        let state = provider.latest().unwrap();
+        // let storage_slot =
+        //     get_storage_var_address("ERC20_balances", &[account_address.into()]).unwrap();
+        let storage_slot = get_fee_token_balance_base_storage_address(account_address);
+        let balance_felt = state
+            .storage(self.backend.chain_spec.fee_contracts.eth, storage_slot)
+            .unwrap()
+            .unwrap();
+        let balance: u128 = balance_felt.to_string().parse().unwrap();
+        Ok(balance)
+    }
+
+    async fn fee_token(&self) -> Result<String, Error> {
+        let provider = self.backend.blockchain.provider();
+        let state = provider.latest().unwrap();
+        let fee_token = state
+            .storage(self.backend.chain_spec.fee_contracts.eth, ERC20_NAME_STORAGE_SLOT)
+            .unwrap()
+            .unwrap();
+        Ok(fee_token.to_string())
+    }
+
+    async fn mint(&self) -> Result<(), Error> {
+        Ok(())
+    }
+
     async fn predeployed_accounts(&self) -> Result<Vec<Account>, Error> {
         Ok(self.backend.chain_spec.genesis.accounts().map(|e| Account::new(*e.0, e.1)).collect())
     }

crates/katana/rpc/rpc/src/future.rs

@@ -0,0 +1,216 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+//! Utilities for handling async code.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use futures_util::future::FutureExt;
+use jsonrpsee_core::Error;
+use tokio::sync::{watch, OwnedSemaphorePermit, Semaphore, TryAcquireError};
+use tokio::time::{self, Duration, Interval};
+
+/// Polling for server stop monitor interval in milliseconds.
+const STOP_MONITOR_POLLING_INTERVAL: Duration = Duration::from_millis(1000);
+
+/// This is a flexible collection of futures that need to be driven to completion
+/// alongside some other future, such as connection handlers that need to be
+/// handled along with a listener for new connections.
+///
+/// In order to `.await` on these futures and drive them to completion, call
+/// `select_with` providing some other future, the result of which you need.
+pub(crate) struct FutureDriver<F> {
+    futures: Vec<F>,
+    stop_monitor_heartbeat: Interval,
+}
+
+impl<F> Default for FutureDriver<F> {
+    fn default() -> Self {
+        let mut heartbeat = time::interval(STOP_MONITOR_POLLING_INTERVAL);
+
+        heartbeat.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+
+        FutureDriver { futures: Vec::new(), stop_monitor_heartbeat: heartbeat }
+    }
+}
+
+impl<F> FutureDriver<F> {
+    /// Add a new future to this driver
+    pub(crate) fn add(&mut self, future: F) {
+        self.futures.push(future);
+    }
+}
+
+impl<F> FutureDriver<F>
+where
+    F: Future + Unpin,
+{
+    pub(crate) async fn select_with<S: Future>(&mut self, selector: S) -> S::Output {
+        tokio::pin!(selector);
+
+        DriverSelect { selector, driver: self }.await
+    }
+
+    fn drive(&mut self, cx: &mut Context) {
+        let mut i = 0;
+
+        while i < self.futures.len() {
+            if self.futures[i].poll_unpin(cx).is_ready() {
+                // Using `swap_remove` since we don't care about ordering
+                // but we do care about removing being `O(1)`.
+                //
+                // We don't increment `i` in this branch, since we now
+                // have a shorter length, and potentially a new value at
+                // current index
+                self.futures.swap_remove(i);
+            } else {
+                i += 1;
+            }
+        }
+    }
+
+    fn poll_stop_monitor_heartbeat(&mut self, cx: &mut Context) {
+        // We don't care about the ticks of the heartbeat, it's here only
+        // to periodically wake the `Waker` on `cx`.
+        let _ = self.stop_monitor_heartbeat.poll_tick(cx);
+    }
+}
+
+impl<F> Future for FutureDriver<F>
+where
+    F: Future + Unpin,
+{
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        let this = Pin::into_inner(self);
+
+        this.drive(cx);
+
+        if this.futures.is_empty() { Poll::Ready(()) } else { Poll::Pending }
+    }
+}
+
+/// This is a glorified select `Future` that will attempt to drive all
+/// connection futures `F` to completion on each `poll`, while also
+/// handling incoming connections.
+struct DriverSelect<'a, S, F> {
+    selector: S,
+    driver: &'a mut FutureDriver<F>,
+}
+
+impl<'a, R, F> Future for DriverSelect<'a, R, F>
+where
+    R: Future + Unpin,
+    F: Future + Unpin,
+{
+    type Output = R::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        let this = Pin::into_inner(self);
+
+        this.driver.drive(cx);
+        this.driver.poll_stop_monitor_heartbeat(cx);
+
+        this.selector.poll_unpin(cx)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct StopHandle(watch::Receiver<()>);
+
+impl StopHandle {
+    pub(crate) fn new(rx: watch::Receiver<()>) -> Self {
+        Self(rx)
+    }
+
+    pub(crate) fn shutdown_requested(&self) -> bool {
+        // if a message has been seen, it means that `stop` has been called.
+        self.0.has_changed().unwrap_or(true)
+    }
+
+    pub(crate) async fn shutdown(&mut self) {
+        // Err(_) implies that the `sender` has been dropped.
+        // Ok(_) implies that `stop` has been called.
+        let _ = self.0.changed().await;
+    }
+}
+
+/// Server handle.
+///
+/// When all [`StopHandle`]'s have been `dropped` or `stop` has been called
+/// the server will be stopped.
+#[derive(Debug, Clone)]
+pub struct ServerHandle(Arc<watch::Sender<()>>);
+
+impl ServerHandle {
+    /// Create a new server handle.
+    pub fn new(tx: watch::Sender<()>) -> Self {
+        Self(Arc::new(tx))
+    }
+
+    /// Tell the server to stop without waiting for the server to stop.
+    pub fn stop(&self) -> Result<(), Error> {
+        self.0.send(()).map_err(|_| Error::AlreadyStopped)
+    }
+
+    /// Wait for the server to stop.
+    pub async fn stopped(self) {
+        self.0.closed().await
+    }
+
+    /// Check if the server has been stopped.
+    pub fn is_stopped(&self) -> bool {
+        self.0.is_closed()
+    }
+}
+
+/// Limits the number of connections.
+#[derive(Debug)]
+pub(crate) struct ConnectionGuard(Arc<Semaphore>);
+
+impl ConnectionGuard {
+    pub(crate) fn new(limit: usize) -> Self {
+        Self(Arc::new(Semaphore::new(limit)))
+    }
+
+    pub(crate) fn try_acquire(&self) -> Option<OwnedSemaphorePermit> {
+        match self.0.clone().try_acquire_owned() {
+            Ok(guard) => Some(guard),
+            Err(TryAcquireError::Closed) => {
+                unreachable!("Semaphore::Close is never called and can't be closed; qed")
+            }
+            Err(TryAcquireError::NoPermits) => None,
+        }
+    }
+
+    pub(crate) fn available_connections(&self) -> usize {
+        self.0.available_permits()
+    }
+}

crates/katana/rpc/rpc/src/lib.rs

@@ -5,8 +5,13 @@
 
 pub mod dev;
 pub mod metrics;
+pub mod proxy_get_request;
 pub mod saya;
 pub mod starknet;
 pub mod torii;
 
+mod future;
+mod logger;
+mod server;
+mod transport;
 mod utils;