diff --git a/crates/storage-cassandra/src/engine.rs b/crates/storage-cassandra/src/engine.rs index a70db87fb..ce410e752 100644 --- a/crates/storage-cassandra/src/engine.rs +++ b/crates/storage-cassandra/src/engine.rs @@ -6,10 +6,12 @@ use std::sync::Arc; use cdrs_tokio::authenticators::{NoneAuthenticatorProvider, StaticPasswordAuthenticatorProvider}; +use cdrs_tokio::cluster::connection_pool::ConnectionPoolConfigBuilder; use cdrs_tokio::cluster::session::{Session, SessionBuilder, TcpSessionBuilder}; use cdrs_tokio::cluster::{NodeTcpConfigBuilder, TcpConnectionManager}; use cdrs_tokio::load_balancing::RoundRobinLoadBalancingStrategy; use cdrs_tokio::query_values; +use cdrs_tokio::retry::ExponentialReconnectionPolicy; use cdrs_tokio::transport::TransportTcp; use extenddb_storage::error::StorageError; @@ -106,6 +108,21 @@ pub struct CassandraEngine { /// Cap on concurrently in-flight detached TTL claim releases. const TTL_RELEASE_MAX_IN_FLIGHT: usize = 1_024; +// Reconnection backoff bounds for the driver (issue #403). +// +// cdrs-tokio's default policy doubles a 1s base up to 60s, and each outage +// starts a new schedule, so it reaches 60s after about a minute of failed +// attempts. We cap it at 5s: the waits are about 2s, about 4s, then 4.25 to +// 5s, then a fixed 5s with no retry limit. The pool and the control +// connection both use this policy. Declared as constants so they are easy to +// surface as config later if needed. +const RECONNECT_BASE_DELAY: std::time::Duration = std::time::Duration::from_secs(1); +const RECONNECT_MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(5); +/// Attempts after which the exponential delay has saturated at +/// `RECONNECT_MAX_DELAY`. With a 1s base and 5s cap, the backoff delay will hit +/// the cap by the third attempt. +const RECONNECT_MAX_ATTEMPTS: usize = 3; + /// `(account_id, table_name)` → `(fetched_at, config)`; see the field docs. pub(crate) type TtlConfigCache = Arc< std::sync::Mutex< @@ -177,11 +194,35 @@ impl CassandraEngine { StorageError::Connection(format!("Failed to build cluster config: {e}")) })?; - // Create session with round-robin load balancing + // Create session with round-robin load balancing. + // + // Pool size: cdrs-tokio defaults to ONE connection per node + // (ConnectionPoolConfig::default is local_size=1, remote_size=1). One + // connection serializes the whole CQL stream-id space onto a single + // socket and gives no headroom if that socket drops, so we size the pool + // from `max_connections` (default 10) and scatter requests across it. + // This also eases the cdrs-tokio stream-id reuse hazard (issue #403): + // more connections divide the id space. A single-node + // deployment is "local", so local_size is the one that matters; we set + // remote_size the same for multi-node clusters. + let pool_size = pool_size_for(config.max_connections); + let pool_config = ConnectionPoolConfigBuilder::new() + .with_local_size(pool_size) + .with_remote_size(pool_size) + .build(); // Wrap in timeout to prevent indefinite hangs on connection/auth failures const CONNECTION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); let lb = RoundRobinLoadBalancingStrategy::new(); - let session_future = TcpSessionBuilder::new(lb, cluster_config).build(); + let session_future = TcpSessionBuilder::new(lb, cluster_config) + .with_connection_pool_config(pool_config) + // Cap the reconnection delay at 5s instead of the driver's default + // 60s (issue #403). See the RECONNECT_* constants. + .with_reconnection_policy(std::sync::Arc::new(ExponentialReconnectionPolicy::new( + RECONNECT_BASE_DELAY, + RECONNECT_MAX_DELAY, + RECONNECT_MAX_ATTEMPTS, + ))) + .build(); let session = tokio::time::timeout(CONNECTION_TIMEOUT, session_future) .await @@ -290,6 +331,19 @@ impl CassandraEngine { } } +/// Map the configured `max_connections` to the per-node pool size we request +/// from cdrs-tokio. +/// +/// cdrs-tokio defaults to a single connection per node; we instead size the +/// pool from `max_connections`. The driver takes a `usize`, and a pool of zero +/// connections is unusable, so clamp to at least 1. This is the only decision +/// ExtendDB owns here — the driver then opens and manages that many connections. +/// It is a free function so it can be unit-tested without a session, a driver, +/// or a node. +fn pool_size_for(max_connections: u32) -> usize { + (max_connections.max(1)) as usize +} + // TODO: Implement storage traits: // - TableEngine // - DataEngine @@ -305,3 +359,23 @@ impl CassandraEngine { // - RateLimitStore // - AuthorizationStore // - CatalogStore (composite trait) + +#[cfg(test)] +mod pool_size_tests { + use super::pool_size_for; + + #[test] + fn maps_max_connections_to_pool_size() { + // The configured value passes through unchanged for any usable size. + assert_eq!(pool_size_for(1), 1); + assert_eq!(pool_size_for(2), 2); + assert_eq!(pool_size_for(8), 8); + assert_eq!(pool_size_for(10), 10); + } + + #[test] + fn clamps_zero_to_one() { + // A zero-size pool is unusable; we request at least one connection. + assert_eq!(pool_size_for(0), 1); + } +} diff --git a/crates/storage-cassandra/tests/direct/pool_size.rs b/crates/storage-cassandra/tests/direct/pool_size.rs new file mode 100644 index 000000000..f7cf5d606 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/pool_size.rs @@ -0,0 +1,57 @@ +// Copyright 2026 ExtendDB Contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Checks that `max_connections` reaches the driver's per-node pool (#403). +//! +//! The unit test on `pool_size_for` only covers the mapping. It still passes +//! if the engine stops handing the pool config to the session builder, in +//! which case the driver falls back to a pool of one. This test asks the +//! driver itself how many distinct connections the node's pool holds. +//! +//! The pool hands out connections in strict round-robin order, so 16 picks +//! from a pool of 4 visit every connection. The count is of this engine's own +//! pool, not server-side sockets, so other clients cannot change it. + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + use std::sync::Arc; + + use extenddb_storage_cassandra::CassandraEngine; + + #[tokio::test] + async fn max_connections_reaches_the_driver_pool() { + if crate::helpers::skip_without_cassandra() { + return; + } + + const POOL: u32 = 4; + + let mut cfg = crate::helpers::test_config(); + cfg.max_connections = POOL; + cfg.ensure_cached_connection_string(); + let engine = CassandraEngine::new(&cfg, "us-east-1").await.unwrap(); + + // The driver creates a node's pool on first use. + engine + .session() + .query("SELECT release_version FROM system.local") + .await + .unwrap(); + + let nodes = engine.session().cluster_metadata().unignored_nodes(); + assert!(!nodes.is_empty(), "driver reports no nodes"); + + let mut seen = HashSet::new(); + for _ in 0..(POOL * 4) { + let conn = nodes[0].persistent_connection().await.unwrap(); + seen.insert(Arc::as_ptr(&conn) as usize); + } + + assert_eq!( + seen.len(), + POOL as usize, + "the node's pool should hold max_connections connections" + ); + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 1df0244eb..863bbda69 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -39,6 +39,8 @@ mod keyspace_migrations; mod metadata_engine; #[path = "direct/policies.rs"] mod policies; +#[path = "direct/pool_size.rs"] +mod pool_size; #[path = "direct/put_get_item.rs"] mod put_get_item; #[path = "direct/query.rs"] diff --git a/docs/manuals/11-deployment-guide.md b/docs/manuals/11-deployment-guide.md index 419a123d4..6e1f334ab 100755 --- a/docs/manuals/11-deployment-guide.md +++ b/docs/manuals/11-deployment-guide.md @@ -298,6 +298,22 @@ Ensure PostgreSQL `max_connections` accommodates the total pool size plus overhe Backup and restore work is admitted against the pool so it cannot take every connection from ordinary requests. On PostgreSQL and MongoDB one server runs at most 4 `RestoreTableFromBackup` copies per account and `min(16, pool/2)` in total (PostgreSQL: `pool_size`; MongoDB: the engine's driver pool, `maxPoolSize` in the connection string or `max_connections`, default 10, so 5 restores); a restore that waits more than 30 seconds for a slot is refused with `LimitExceededException` and creates nothing. On SQLite one server runs at most `pool_size / 2 - 1` `CreateBackup` copies at once (between 1 and 16; 4 for the default `pool_size` of 10), each holding its snapshot connection for its duration, and refuses a backup that waited 30 seconds with `LimitExceededException`. A MongoDB restore request that is not registered within 60 seconds answers an internal error while the restore goes on in the background; the MongoDB driver does not enforce `waitQueueTimeoutMS`, so the admission bound is what keeps a flood of restores from holding every pooled connection. +### Cassandra Connection Pool + +The Cassandra backend opens `max_connections` connections per node (default 10) +and scatters requests across them: + +```toml +[storage.cassandra] +max_connections = 10 # connections per node +``` + +Keep this above 1. A single connection serializes every request's protocol +stream id onto one socket and leaves no headroom if that socket drops. Spreading +requests over several connections also divides the stream-id space, which eases a +driver-level stream-id reuse hazard seen under sustained load with a slow +statement in flight. + ### PostgreSQL Tuning Key PostgreSQL settings for extenddb workloads: