Merge pull request #73 from Lay3rLabs/pool-improvements

Signing client pool wrapper

Author: dakom

Cargo.lock

@@ -3,19 +3,19 @@ members = ["packages/*", "examples/*", "faucet", "integration-test"]
 resolver = "2"
 
 [workspace.package]
-version = "0.4.1"
+version = "0.4.2"
 edition = "2021"
 license = "Apache-2.0"
 repository = "https://github.com/Lay3rLabs/climb"
 
 [workspace.dependencies]
 # Local
-layer-climb = { path = "packages/layer-climb", version = "0.4.0-alpha.6" }
-layer-climb-address = { path = "packages/layer-climb-address", version = "0.4.0-alpha.6" }
-layer-climb-cli = { path = "packages/layer-climb-cli", version = "0.4.0-alpha.6" }
-layer-climb-config = { path = "packages/layer-climb-config", version = "0.4.0-alpha.6" }
-layer-climb-core = { path = "packages/layer-climb-core", version = "0.4.0-alpha.6" }
-layer-climb-proto = { path = "packages/layer-climb-proto", version = "0.4.0-alpha.6" }
+layer-climb = { path = "packages/layer-climb", version = "0.4.2" }
+layer-climb-address = { path = "packages/layer-climb-address", version = "0.4.2" }
+layer-climb-cli = { path = "packages/layer-climb-cli", version = "0.4.2" }
+layer-climb-config = { path = "packages/layer-climb-config", version = "0.4.2" }
+layer-climb-core = { path = "packages/layer-climb-core", version = "0.4.2" }
+layer-climb-proto = { path = "packages/layer-climb-proto", version = "0.4.2" }
 
 # General
 cfg-if = "1.0.0"

Cargo.toml

@@ -101,11 +101,12 @@ impl PoolCommand {
                 )
                 .await?;
 
-                let client_pool: Pool<SigningClientPoolManager> =
+                let client_pool = SigningClientPool::new(
                     Pool::builder(client_pool_manager)
                         .max_size(max_concurrent_accounts)
                         .build()
-                        .context("Failed to create client pool")?;
+                        .context("Failed to create client pool")?,
+                );
 
                 let sends = match sends {
                     None => max_concurrent_accounts * 2,

examples/cli/src/commands/pool.rs

@@ -44,6 +44,7 @@ impl StatusHandler {
     pub async fn new(state: AppState) -> Result<Self> {
         let max_derivation_index = state
             .client_pool
+            .pool
             .manager()
             .derivation_index
             .load(Ordering::SeqCst);

faucet/src/handlers/status.rs

@@ -11,7 +11,7 @@ use layer_climb::{pool::SigningClientPoolManager, prelude::*};
 #[derive(Clone)]
 pub struct AppState {
     pub config: Arc<Config>,
-    pub client_pool: Pool<SigningClientPoolManager>,
+    pub client_pool: SigningClientPool,
     pub query_client: QueryClient,
     pub distributor_addrs: Arc<Mutex<HashMap<u32, Address>>>,
 }
@@ -33,9 +33,11 @@ impl AppState {
         )
         .await?;
 
-        let client_pool: Pool<SigningClientPoolManager> = Pool::builder(client_pool_manager)
-            .max_size(config.concurrency)
-            .build()?;
+        let client_pool = SigningClientPool::new(
+            Pool::builder(client_pool_manager)
+                .max_size(config.concurrency)
+                .build()?,
+        );
 
         let query_client = QueryClient::new(config.chain_config.clone(), None).await?;
 

faucet/src/state.rs

@@ -5,8 +5,8 @@ use crate::{
     querier::{Connection, QueryClient},
     signing::SigningClient,
 };
-use anyhow::{bail, Result};
-use deadpool::managed::{Manager, Metrics, RecycleResult};
+use anyhow::{bail, Error, Result};
+use deadpool::managed::{Manager, Metrics, Object, PoolError, RecycleResult};
 use layer_climb_address::*;
 use layer_climb_config::ChainConfig;
 use tokio::sync::Mutex;
@@ -187,8 +187,27 @@ impl Manager for SigningClientPoolManager {
         _: &Metrics,
     ) -> RecycleResult<anyhow::Error> {
         tracing::debug!("POOL RECYCLING CLIENT {}", client.addr);
-        self.maybe_top_up(client.addr.clone()).await?;
 
         Ok(())
     }
 }
+
+#[derive(Clone)]
+pub struct SigningClientPool {
+    pub pool: deadpool::managed::Pool<SigningClientPoolManager>,
+}
+
+impl SigningClientPool {
+    pub fn new(pool: deadpool::managed::Pool<SigningClientPoolManager>) -> Self {
+        Self { pool }
+    }
+
+    pub async fn get(&self) -> Result<Object<SigningClientPoolManager>, PoolError<Error>> {
+        let client = self.pool.get().await?;
+        self.pool
+            .manager()
+            .maybe_top_up(client.addr.clone())
+            .await?;
+        Ok(client)
+    }
+}