connpool

Connection pooling for Rust — blocking & async (tokio), zero deps for the blocking pool.

The Rust member of poolyglot · Pool once. Speak everywhere.

A generic connection pool for Rust. The blocking ConnectionPool is standard-library only (zero runtime dependencies); the async AsyncConnectionPool lives behind the optional tokio feature. It pools anything: DB drivers, HTTP clients, SSH sessions, sockets… because the pool knows nothing about connections except what a small ConnectionFactory tells it. T is tracked by value identity (Clone + Eq + Hash) — e.g. Arc<Conn> — the Rust analog of Go's comparable.

use connpool::{ConnectionPool, FnFactory, PoolConfig, Timeout};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

// A toy "connection": a unique id behind an Arc so clones share identity.
let next = Arc::new(AtomicU64::new(0));
let factory = FnFactory::new({
    let next = next.clone();
    move || Ok::<_, connpool::FactoryError>(Arc::new(next.fetch_add(1, Ordering::SeqCst)))
});

let pool = ConnectionPool::new(factory, PoolConfig { max_size: 20, ..Default::default() }, None)
    .expect("valid config");
pool.open()?;                                 // optional: warm-up + background sweeper

let conn = pool.acquire(Timeout::Unset)?;     // acquire -> use -> release
// ... use conn ...
pool.release(conn)?;

This is the Rust member of poolyglot: one pooling model, implemented per language, kept API-parallel. Field names, event names, defaults, and CloseReason/BreakerState wire values are identical to the Python (connpool), C# (ConnPool), Java (connpool), Go (connpool) and TypeScript (@jeffeek/connpool) implementations — switching languages is a syntax change, not a model change.

Install

The crate is published as connpool-rs (the crates.io name connpool was taken) but is imported as connpool (the library name), so use connpool::... works unchanged:

[dependencies]
connpool-rs = "1"
# with the async pool:
connpool-rs = { version = "1", features = ["tokio"] }

Requires Rust 1.85+ (edition 2024). Only the blocking pool is dependency-free; enabling the tokio feature adds a single dependency (tokio) used by AsyncConnectionPool / AsyncPoolRegistry. With the feature off (the default), the crate pulls in nothing.

Project layout

src/                         the library (dependency-free for the blocking pool; one crate `connpool`)
  core/            errors.rs, clock.rs, connection.rs (PooledConnection/CloseReason), timeouts.rs (Timeout)
  resilience/      breaker.rs (CircuitBreaker + BreakerState/BreakerConfig), retry.rs (RetryPolicy)
  policy/          config.rs (PoolConfig/IdleStrategy), lifecycle.rs (pure "when must a connection retire?" rules)
  factories/       sync_factory.rs, async_factory.rs (tokio-gated)  <- what YOU implement
  observability/   events.rs (PoolListener), stats.rs (PoolStats)
  pools/           sync_pool.rs, async_pool.rs (tokio-gated), registry.rs, async_registry.rs (tokio-gated)
  tests inline     `#[cfg(test)] mod tests` per file (cargo test; 100% line gate via cargo-llvm-cov)
examples/                    client code - NOT part of the published crate

Dependency direction is strictly downward (a layer only uses layers listed above it): core → resilience → policy → factories/observability → pools. Everything lives in the single connpool crate, so one use connpool::... imports the whole public API via src/lib.rs.

📐 See the architecture diagrams — layering, connection lifecycle, and circuit-breaker states (shared across all poolyglot languages).

What you get

Writing a factory

Implement ConnectionFactory<T> (or build one from closures with FnFactory):

use connpool::{ConnectionFactory, FactoryError};
use std::error::Error;

struct MyFactory;

impl ConnectionFactory<Arc<MyConn>> for MyFactory {
    fn create(&self) -> Result<Arc<MyConn>, FactoryError> {
        Ok(Arc::new(dial()?))                      // enforce your own connect timeout
    }
    fn validate(&self, c: &Arc<MyConn>) -> bool {
        c.ping().is_ok()                            // cheap liveness check
    }
    fn reset(&self, c: &Arc<MyConn>) -> Result<(), FactoryError> {
        c.rollback().map_err(Into::into)            // scrub session state on release
    }
    fn close(&self, c: &Arc<MyConn>) -> Result<(), FactoryError> {
        c.disconnect().map_err(Into::into)          // default: no-op; most types close on Drop
    }
    fn is_fatal_error(&self, err: &(dyn Error + 'static)) -> bool {
        err.to_string().contains("connection reset")
    }
}

Async: implement AsyncConnectionFactory<T> (behind the tokio feature), or build one from closures with AsyncFnFactory.

Configuration (PoolConfig)

PoolConfig is a plain struct with Default — build with struct-update syntax and call validate() (the pool constructor does this for you, returning PoolError::Config on failure). Optional seconds / counts are Option<f64> / Option<usize> / Option<u64>; None means "no limit".

Build, test, run examples

cargo build                         # the blocking library (zero deps)
cargo build --features tokio        # include the async pool
cargo test --all-features           # run all tests
./run_tests.sh                      # tests under coverage (cargo-llvm-cov)
./run_tests.sh -m                   # ...also list uncovered lines
cargo run --example demo            # runnable demo
./lint.sh                           # clippy -D warnings (gate) + cargo fmt --check report
./lint.sh -f                        # ...auto-fix (cargo fmt + clippy --fix)

Linting uses clippy (cargo clippy --all-targets --all-features -- -D warnings), gated; cargo fmt --check is warn-only (the sources are not yet rustfmt-clean). Needs the clippy + rustfmt components. CI runs these as a separate lint job.

Coverage needs the llvm-tools-preview component and cargo-llvm-cov:

rustup component add llvm-tools-preview
cargo install cargo-llvm-cov

Idiom mapping from Python/Go

• Errors are a single PoolError enum (not an exception hierarchy or a marker-interface struct family) — match on the variant, the analog of except PoolTimeoutError / errors.As(&PoolTimeoutError{}).
• Optional seconds/counts are Option<f64> / Option<usize> (Go's *float64/*int, Python's Optional[float]).
• Connections are tracked by value identity, so T: Clone + Eq + Hash — use Arc<Conn> or an id, the direct analog of Go's T comparable.
• No RAII guard type yet — borrowing is explicit acquire() / release(), mirroring Go's Acquire/Release pair (not C#'s using lease = pool.Connection()).

• Async cancellation is future-drop, not a token: dropping the acquire()/use_() future while it's pending frees the reserved slot via a Drop guard, the Rust analog of Python's implicit task-cancellation handling and C#'s CancellationToken.

• A tiny built-in SplitMix64 RNG (Rng/SystemRng) drives jitter so the blocking pool stays zero-dependency instead of pulling in rand.
• Wire values stay identical across languages: CloseReason::as_str(), BreakerState::as_str(), IdleStrategy::as_str(), and PoolStats::to_dictionary() keys are snake_case and match Python, C#, Java, Go and TypeScript exactly.

Limits / extension points

• The breaker guards connection creation. To also trip on request failures, wrap calls with CircuitBreaker::call().
• The breaker counts consecutive failed creations (one per logical acquire, retries included). A real backend failure or a create_timeout firing both count; an acquire-timeout (caller impatience) or cancellation mid-create does not.
• Leak detection is warn-only by design and only runs while the pool is open()ed (the maintenance sweep drives it), so registry pools need an explicit open(), and a warning can lag by up to one maintenance_interval.
• Not fork-safe / not process-shared. max_size is per-process; create a fresh pool per process.
• Pool size is fixed at construction; there is no runtime resize. Use a new pool, or a PoolRegistry keyed by profile.
• Wake-up order of blocked callers is not strictly FIFO (the condition broadcasts on every change).
• A blocking create() can't be interrupted by the pool; give your client a connect timeout. For the async pool, create_timeout is only enforced if your factory honors cancellation.
• Listener hooks are synchronous; push heavy work to a queue.
• A factory's create() must return a distinct object each call (T is pooled by value identity; use Arc<Conn> or similar); returning a shared/singleton connection is rejected with PoolError::ConnectionCreate.
• T is constrained to Clone + Eq + Hash because the pool tracks borrow state by the connection value in a map.
• The async pool requires the tokio feature — Rust's standard library has no async runtime or timer of its own, unlike Go (context) or C# (Task). The blocking pool stays zero-dependency; this mirrors how the TypeScript port is async-only for the opposite reason (no blocking primitives at all).
• Acquire-wait / create-latency percentiles are estimated over the last 1024 samples (bounded memory), not the full history.
• With validate_idle on, idle connections being pinged by the background sweep are briefly reserved and count as in-use (not idle) in stats() for the duration of that check.
• Coverage gate is 100% lines (the LCOV metric Codecov reports), enforced by run_tests.sh. LLVM region/branch coverage is intentionally not gated: async fn state-machine codegen leaves a few region artifacts that aren't real untested paths.

Considered, not implemented

• A PoolGuard/RAII wrapper that releases on Drop (like C#'s using lease = pool.Connection()) — skipped for now; acquire()/release() stays explicit and panics on an unreleased connection are avoided deliberately, since a double-free-style bug in a Drop impl can't return a Result.
• Runtime pool resize — same rationale as the other languages: a PoolRegistry keyed by profile covers the common case without adding mutable shared config.

License

MIT — see LICENSE.