From dc20cafc41417bd4f9adb3e0c23a0b04619898b6 Mon Sep 17 00:00:00 2001 From: Joel Shepherd Date: Thu, 22 Oct 2026 21:39:16 +0000 Subject: [PATCH 1/3] =?UTF-8?q?=E2=80=A2=20fix(storage-cassandra):=20apply?= =?UTF-8?q?=20max=5Fconnections=20and=20advertise=20a=20reachable=20broadc?= =?UTF-8?q?ast=20address=20(#403)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two connection-resilience problems on the Cassandra backend: 1. create_session never applied config.max_connections, so the driver ran one connection per node regardless of config. Now wired through with_connection_pool_config, so the pool opens max_connections connections and scatters requests across them. 2. Containerized nodes advertise their internal bridge IP as broadcast_rpc_address. The driver reconnects to the advertised address, which was not reachable outside the container. devtools/cassandra-node and the CI service containers now advertise 127.0.0.1; the deployment guide documents the requirement for production nodes. Test: connection_pool_size.rs asserts the pool size follows max_connections (red when it was ignored, green now). Recovery-after-drop verified manually via nodetool disablebinary/enablebinary. Partially addresses #403: the underlying cdrs-tokio stream-id reuse is a driver defect, mitigated here (more connections divide the id space) and tracked upstream. --- .github/workflows/integration-cassandra.yml | 15 +++ crates/storage-cassandra/src/engine.rs | 22 ++++- .../tests/direct/connection_pool_size.rs | 95 +++++++++++++++++++ .../tests/direct_integration.rs | 2 + docs/manuals/11-deployment-guide.md | 48 ++++++++++ 5 files changed, 180 insertions(+), 2 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/connection_pool_size.rs diff --git a/.github/workflows/integration-cassandra.yml b/.github/workflows/integration-cassandra.yml index 08ad150a8..4874f539c 100644 --- a/.github/workflows/integration-cassandra.yml +++ b/.github/workflows/integration-cassandra.yml @@ -51,6 +51,11 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci + # Advertise the address the runner reaches the service on (the mapped + # port on localhost). Otherwise the node advertises its container IP, + # which the Cassandra driver would reconnect to and never reach, + # hanging the pool after any connection drop (issue #403). + CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- @@ -107,6 +112,11 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci + # Advertise the address the runner reaches the service on (the mapped + # port on localhost). Otherwise the node advertises its container IP, + # which the Cassandra driver would reconnect to and never reach, + # hanging the pool after any connection drop (issue #403). + CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- @@ -158,6 +168,11 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci + # Advertise the address the runner reaches the service on (the mapped + # port on localhost). Otherwise the node advertises its container IP, + # which the Cassandra driver would reconnect to and never reach, + # hanging the pool after any connection drop (issue #403). + CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- diff --git a/crates/storage-cassandra/src/engine.rs b/crates/storage-cassandra/src/engine.rs index 23f9bc36e..1b5448298 100644 --- a/crates/storage-cassandra/src/engine.rs +++ b/crates/storage-cassandra/src/engine.rs @@ -6,6 +6,7 @@ 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; @@ -166,11 +167,28 @@ 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 is also the mitigation for 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 = config.max_connections.max(1) as usize; + 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) + .build(); let session = tokio::time::timeout(CONNECTION_TIMEOUT, session_future) .await diff --git a/crates/storage-cassandra/tests/direct/connection_pool_size.rs b/crates/storage-cassandra/tests/direct/connection_pool_size.rs new file mode 100644 index 000000000..1db011868 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/connection_pool_size.rs @@ -0,0 +1,95 @@ +// Copyright 2026 ExtendDB Contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Regression test for issue #403 (part 1): the connection pool size must +//! follow `max_connections`. +//! +//! cdrs-tokio defaults to ONE connection per node. ExtendDB's config carries +//! `max_connections` (default 10) and advertises it through +//! `ConnectionConfig::max_connections`, but `create_session` never applied it, +//! so every deployment ran a single connection regardless of config — the +//! precondition for the stream-id-exhaustion failure in #403, and a plain +//! false promise (config says 10, driver opens 1). +//! +//! This test builds two engines with different `max_connections` and checks, +//! via the node's `system_views.clients`, that the larger pool opens strictly +//! more client connections than the smaller one, and at least as many as its +//! configured size. It asserts a comparison rather than an exact count because +//! the driver also opens a control connection and other clients may be present; +//! the comparison isolates the effect of `max_connections` without depending on +//! that overhead. + +#[cfg(test)] +mod tests { + use crate::helpers::test_config; + use extenddb_storage_cassandra::CassandraEngine; + + /// Count client connections the node currently sees from the loopback + /// address, queried through the given engine's own session. Counts rows + /// client-side rather than using CQL `COUNT(*)`, whose aggregate column type + /// on a virtual table is awkward to deserialize. + async fn client_connection_count(engine: &CassandraEngine) -> i64 { + let body = engine + .session() + .query("SELECT address, port FROM system_views.clients") + .await + .expect("query system_views.clients") + .response_body() + .expect("response body"); + let rows = body.into_rows().unwrap_or_default(); + i64::try_from(rows.len()).unwrap_or(i64::MAX) + } + + #[tokio::test] + async fn pool_size_follows_max_connections() { + if crate::helpers::skip_without_cassandra() { + return; + } + + let small_size: u32 = 2; + let large_size: u32 = 8; + + // Build the small-pool engine and measure how many connections the node + // sees from us. Each engine opens its own pool, so we measure each + // against its own session. + let mut small_cfg = test_config(); + small_cfg.max_connections = small_size; + small_cfg.ensure_cached_connection_string(); + let small_engine = CassandraEngine::new(&small_cfg, "us-east-1") + .await + .expect("build small-pool engine"); + // Give the pool a moment to finish opening every connection. + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + let small_count = client_connection_count(&small_engine).await; + + let mut large_cfg = test_config(); + large_cfg.max_connections = large_size; + large_cfg.ensure_cached_connection_string(); + let large_engine = CassandraEngine::new(&large_cfg, "us-east-1") + .await + .expect("build large-pool engine"); + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + // Count AFTER the large engine is up: this reflects both pools plus any + // other clients. The delta attributable to the large engine is + // large_total - small_count; asserting it carries large_size proves the + // config was applied and scales. + let both_count = client_connection_count(&large_engine).await; + + // Before the fix, create_session ignored max_connections and opened one + // connection per engine, so both engines together added ~2 and this + // assertion fails. With the fix, the large engine alone adds >= 8. + let large_delta = both_count - small_count; + assert!( + large_delta >= i64::from(large_size), + "large pool (max_connections={large_size}) added only {large_delta} connections \ + (small_count={small_count}, both_count={both_count}); expected at least \ + {large_size} — max_connections is not being applied to the pool (issue #403)" + ); + assert!( + large_delta > i64::from(small_size), + "large pool did not open more connections than the small pool \ + (delta={large_delta}, small_size={small_size}) — pool size does not follow \ + max_connections (issue #403)" + ); + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b81..71b84b6c8 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -25,6 +25,8 @@ mod authorization_store; mod backup_engine; #[path = "direct/cassandra_engine.rs"] mod cassandra_engine; +#[path = "direct/connection_pool_size.rs"] +mod connection_pool_size; #[path = "direct/delete_item.rs"] mod delete_item; #[path = "direct/groups.rs"] diff --git a/docs/manuals/11-deployment-guide.md b/docs/manuals/11-deployment-guide.md index fb03c271b..58a7d7529 100755 --- a/docs/manuals/11-deployment-guide.md +++ b/docs/manuals/11-deployment-guide.md @@ -244,6 +244,54 @@ pool_size = 50 # Increase for higher concurrency Ensure PostgreSQL `max_connections` accommodates the total pool size plus overhead. +### Cassandra Connection Pool and Broadcast Address + +Two Cassandra-specific settings affect connection resilience. Both matter for +any production deployment. + +**Pool size.** 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. More +connections also divide the stream-id space, which avoids a driver-level +stream-id reuse hazard under sustained load with a slow statement in flight. + +**Broadcast RPC address (important).** The Cassandra driver reconnects to the +address each node advertises in `system.local`/`system.peers` +(`broadcast_rpc_address`), *not* to the contact point ExtendDB was configured +with. If a node advertises an address the ExtendDB host cannot reach, the +connection works at first but the pool never recovers from any later drop — a +node restart, a network blip, an idle timeout — and every request fails with +`No nodes available in query plan!` until ExtendDB is restarted. + +Configure each node's `broadcast_rpc_address` to an address every ExtendDB +instance can reach: + +```yaml +# cassandra.yaml +broadcast_rpc_address:
+``` + +This is the usual trap for **containerized** Cassandra: a container advertises +its internal bridge IP (for example `240.x` or `172.x`), which a client +connecting through a published host port cannot reach. For the official +`cassandra` image, set it through the environment: + +```bash +docker run ... -e CASSANDRA_BROADCAST_RPC_ADDRESS= cassandra:4.1 +``` + +For local development and CI, `devtools/cassandra-node` sets +`CASSANDRA_BROADCAST_RPC_ADDRESS=127.0.0.1` by default (override with +`--broadcast-rpc-address`), and the `integration-cassandra` CI workflow sets the +same on its service container. + ### PostgreSQL Tuning Key PostgreSQL settings for extenddb workloads: From fb55676da0d88560ffdae89687ac3312b547fdfd Mon Sep 17 00:00:00 2001 From: Joel Shepherd Date: Sat, 24 Oct 2026 19:51:20 +0000 Subject: [PATCH 2/3] fix(storage-cassandra): wait for schema (DDL) changes to settle before completing migration process --- crates/storage-cassandra/src/migrations.rs | 86 ++++++++++++++++++- .../direct/migration_schema_agreement.rs | 82 ++++++++++++++++++ .../tests/direct_integration.rs | 2 + 3 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 crates/storage-cassandra/tests/direct/migration_schema_agreement.rs diff --git a/crates/storage-cassandra/src/migrations.rs b/crates/storage-cassandra/src/migrations.rs index b1bc4fe20..b8ede0c64 100644 --- a/crates/storage-cassandra/src/migrations.rs +++ b/crates/storage-cassandra/src/migrations.rs @@ -213,7 +213,91 @@ async fn execute_migration(session: &Arc, sql: &str) -> OpResu .await .map_err(|e| OpError::Internal(format!("Migration failed: {e}")))?; } - Ok(()) + + // A migration is DDL (CREATE KEYSPACE/TABLE, etc.). The driver runs queries + // over a pool of connections and selects one per query, so the statement + // that records or checks a migration can land on a different connection — + // and a different coordinator view — from the one that just ran the DDL. + // cdrs-tokio 9.0.2 exposes no schema-agreement wait, and learns of schema + // changes asynchronously via SCHEMA_CHANGE events, so a read issued right + // after a CREATE can hit a coordinator that has not yet applied it and fail + // with "table ... does not exist". With a single connection the statements + // serialized on one coordinator and the race was effectively impossible; + // the multi-connection pool (issue #403) exposes it. Wait for the cluster + // schema to agree before the next statement so the new object is visible + // everywhere. This is startup/DDL-only, so the poll cost is irrelevant. + wait_for_schema_agreement(session).await +} + +/// Poll until the cluster agrees on a single schema version. +/// +/// Reads `schema_version` from `system.local` (this node) and `system.peers` +/// (every other node); when they are all equal the schema change has +/// propagated and newly created objects are visible through any connection in +/// the pool. On a single-node cluster `system.peers` is empty, so this returns +/// as soon as the local version is readable — which still gives the driver's +/// event loop a beat to apply the change before the next query. +async fn wait_for_schema_agreement(session: &Arc) -> OpResult<()> { + use std::time::{Duration, Instant}; + + const TIMEOUT: Duration = Duration::from_secs(30); + const INTERVAL: Duration = Duration::from_millis(200); + + let deadline = Instant::now() + TIMEOUT; + loop { + if schema_versions_agree(session).await? { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(OpError::Internal( + "Timed out waiting for Cassandra schema agreement after a migration".to_owned(), + )); + } + tokio::time::sleep(INTERVAL).await; + } +} + +/// Return true when `system.local` and every `system.peers` row report the same +/// `schema_version`. +async fn schema_versions_agree(session: &Arc) -> OpResult { + use cdrs_tokio::types::IntoRustByName as _; + + let local = session + .query("SELECT schema_version FROM system.local") + .await + .map_err(|e| OpError::Internal(format!("schema agreement (local): {e}")))? + .response_body() + .map_err(|e| OpError::Internal(format!("schema agreement (local): {e}")))? + .into_rows() + .unwrap_or_default(); + + let Some(local_version) = local.first().and_then(|r| { + let v: Option = r.get_r_by_name("schema_version").ok(); + v + }) else { + // No readable local version yet; treat as not-yet-agreed and retry. + return Ok(false); + }; + + let peers = session + .query("SELECT schema_version FROM system.peers") + .await + .map_err(|e| OpError::Internal(format!("schema agreement (peers): {e}")))? + .response_body() + .map_err(|e| OpError::Internal(format!("schema agreement (peers): {e}")))? + .into_rows() + .unwrap_or_default(); + + for peer in &peers { + let peer_version: Option = peer.get_r_by_name("schema_version").ok(); + match peer_version { + Some(v) if v == local_version => {} + // A peer that disagrees, or whose version is not yet readable, means + // the change has not propagated. Retry. + _ => return Ok(false), + } + } + Ok(true) } /// Record a migration in the `schema_history` table. diff --git a/crates/storage-cassandra/tests/direct/migration_schema_agreement.rs b/crates/storage-cassandra/tests/direct/migration_schema_agreement.rs new file mode 100644 index 000000000..a674769aa --- /dev/null +++ b/crates/storage-cassandra/tests/direct/migration_schema_agreement.rs @@ -0,0 +1,82 @@ +// Copyright 2026 ExtendDB Contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Regression test for the migration schema-agreement race (issue #403). +//! +//! The Cassandra backend opens a multi-connection pool (`max_connections`). +//! The driver round-robins queries across those connections, so a read issued +//! right after a DDL `CREATE TABLE` can land on a connection whose coordinator +//! view has not yet applied the change and fail with "table ... does not +//! exist". `run_data_migrations` creates `schema_history` and then immediately +//! queries it, so it hit this race once the pool held more than one connection. +//! +//! `run_data_migrations` now waits for cluster schema agreement after each DDL +//! statement. This test concentrates the race: it creates many fresh account +//! keyspaces concurrently, runs data migrations on each, and then reads +//! `schema_history` back from each. Before the fix this surfaced the race +//! intermittently; with the fix every keyspace migrates and reads back cleanly. + +#[cfg(test)] +mod tests { + use crate::helpers::{setup_engine, unique_test_account}; + use extenddb_storage_cassandra::migrations::run_data_migrations; + + #[tokio::test] + async fn migrations_survive_concurrent_keyspace_creation() { + if crate::helpers::skip_without_cassandra() { + return; + } + + // Enough concurrent keyspaces to exercise pool round-robin; each one is + // an independent CREATE KEYSPACE + CREATE TABLE schema_history + read. + const KEYSPACES: usize = 12; + + let engine = setup_engine().await; + let session = engine.session_arc(); + + let mut keyspaces = Vec::with_capacity(KEYSPACES); + let mut handles = Vec::with_capacity(KEYSPACES); + + for _ in 0..KEYSPACES { + let keyspace = format!("extenddb_sa_test_account_{}", unique_test_account()); + keyspaces.push(keyspace.clone()); + + let engine = engine.clone(); + let session = session.clone(); + handles.push(tokio::spawn(async move { + engine + .create_keyspace(&keyspace) + .await + .unwrap_or_else(|e| panic!("create_keyspace {keyspace}: {e:?}")); + + // The call under test. Internally: USE, then for the first + // migration CREATE TABLE schema_history, then SELECT/INSERT + // against it. Without the schema-agreement wait, the read can + // race the create across pool connections. + run_data_migrations(&session, &keyspace) + .await + .unwrap_or_else(|e| panic!("run_data_migrations {keyspace}: {e:?}")); + + // Read schema_history straight back, fully qualified, which is + // routed through the pool and would fail if the table were not + // yet visible on the chosen coordinator. + let check = format!("SELECT version FROM {keyspace}.schema_history"); + session + .query(check) + .await + .unwrap_or_else(|e| panic!("read schema_history {keyspace}: {e:?}")); + })); + } + + for handle in handles { + handle.await.expect("migration task panicked"); + } + + // Clean up the keyspaces this test created. + for keyspace in keyspaces { + let _ = session + .query(format!("DROP KEYSPACE IF EXISTS {keyspace}")) + .await; + } + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 71b84b6c8..b6ed71936 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -35,6 +35,8 @@ mod groups; mod index; #[path = "direct/metadata_engine.rs"] mod metadata_engine; +#[path = "direct/migration_schema_agreement.rs"] +mod migration_schema_agreement; #[path = "direct/policies.rs"] mod policies; #[path = "direct/put_get_item.rs"] From b4cf251c27b2d71923563571e924dca743dcb90b Mon Sep 17 00:00:00 2001 From: Joel Shepherd Date: Mon, 26 Oct 2026 18:39:01 +0000 Subject: [PATCH 3/3] Responding to feedback on PR 457. * Broadcast address was a red herring and doesn't play into the client connection process: removed from code and docs. * Retaining change to enable configurable connection pool size, rather than letting CDRS default to 1. * CDRS defaulted to a connection retry policy that would exponentially back off to 60s delay, and then try forever, once a minute. An ill-timed reconnection attempt could experience a delay of up to 2 minutes, giving an appearance of a permanent connection failure, even with a pool size > 1. Reconnection policy is now explicitly configured with a maximum delay of 5s. The "try forever" behavior is not configurable. The "max_attempts" configuration specifies how many attempts are required before the backoff meets/exceeds the maximum delay, not the total number of reconnection attempts. Testing: I'm not including a test to connect/disconnect but in local testing, repeated connection failures (via rapidly bouncing the node) were able to recover in < 1.5s over 15 consecutive restarts. With the default reconnection policy, reconnect attempts appeated to stall after 12 consecutive restarts (as the driver side backoff delay had reached the max of 60s, causing long delays between attempts). --- .github/workflows/integration-cassandra.yml | 15 --- crates/storage-cassandra/src/engine.rs | 61 +++++++++++- .../tests/direct/connection_pool_size.rs | 95 ------------------- .../tests/direct_integration.rs | 2 - docs/manuals/11-deployment-guide.md | 46 ++------- 5 files changed, 67 insertions(+), 152 deletions(-) delete mode 100644 crates/storage-cassandra/tests/direct/connection_pool_size.rs diff --git a/.github/workflows/integration-cassandra.yml b/.github/workflows/integration-cassandra.yml index 4874f539c..08ad150a8 100644 --- a/.github/workflows/integration-cassandra.yml +++ b/.github/workflows/integration-cassandra.yml @@ -51,11 +51,6 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci - # Advertise the address the runner reaches the service on (the mapped - # port on localhost). Otherwise the node advertises its container IP, - # which the Cassandra driver would reconnect to and never reach, - # hanging the pool after any connection drop (issue #403). - CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- @@ -112,11 +107,6 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci - # Advertise the address the runner reaches the service on (the mapped - # port on localhost). Otherwise the node advertises its container IP, - # which the Cassandra driver would reconnect to and never reach, - # hanging the pool after any connection drop (issue #403). - CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- @@ -168,11 +158,6 @@ jobs: - 9042:9042 env: CASSANDRA_CLUSTER_NAME: extenddb-ci - # Advertise the address the runner reaches the service on (the mapped - # port on localhost). Otherwise the node advertises its container IP, - # which the Cassandra driver would reconnect to and never reach, - # hanging the pool after any connection drop (issue #403). - CASSANDRA_BROADCAST_RPC_ADDRESS: 127.0.0.1 HEAP_NEWSIZE: 128M MAX_HEAP_SIZE: 1024M options: >- diff --git a/crates/storage-cassandra/src/engine.rs b/crates/storage-cassandra/src/engine.rs index ac0d08ad8..9496f2bb1 100644 --- a/crates/storage-cassandra/src/engine.rs +++ b/crates/storage-cassandra/src/engine.rs @@ -11,6 +11,7 @@ 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; @@ -115,6 +116,23 @@ 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's connection pool (issue #403). +// +// cdrs-tokio's default `ExponentialReconnectionPolicy` climbs to a 60s max +// delay, so after repeated connection drops the pool can wait up to ~60s +// (plus jitter) before retrying — long enough to look like a permanent failure +// under sustained load. We keep the exponential shape but cap the max delay far +// lower: the per-node delay goes `BASE` (2s), then 4s, then holds at `MAX` (5s) +// ±15% jitter on every attempt thereafter, and the policy never stops retrying. +// 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 doubling (2s, 4s) hits +/// the cap by the third attempt; this is a saturation point, not a retry limit +/// (the policy retries indefinitely). +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< @@ -198,7 +216,7 @@ impl CassandraEngine { // (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 = config.max_connections.max(1) as usize; + let pool_size = pool_size_for(config.max_connections); let pool_config = ConnectionPoolConfigBuilder::new() .with_local_size(pool_size) .with_remote_size(pool_size) @@ -208,6 +226,14 @@ impl CassandraEngine { let lb = RoundRobinLoadBalancingStrategy::new(); let session_future = TcpSessionBuilder::new(lb, cluster_config) .with_connection_pool_config(pool_config) + // Bound the reconnection backoff so the pool recovers promptly under + // sustained connection drops instead of climbing to the driver's + // default 60s ceiling (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) @@ -317,6 +343,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 @@ -332,3 +371,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/connection_pool_size.rs b/crates/storage-cassandra/tests/direct/connection_pool_size.rs deleted file mode 100644 index 1db011868..000000000 --- a/crates/storage-cassandra/tests/direct/connection_pool_size.rs +++ /dev/null @@ -1,95 +0,0 @@ -// Copyright 2026 ExtendDB Contributors -// SPDX-License-Identifier: Apache-2.0 - -//! Regression test for issue #403 (part 1): the connection pool size must -//! follow `max_connections`. -//! -//! cdrs-tokio defaults to ONE connection per node. ExtendDB's config carries -//! `max_connections` (default 10) and advertises it through -//! `ConnectionConfig::max_connections`, but `create_session` never applied it, -//! so every deployment ran a single connection regardless of config — the -//! precondition for the stream-id-exhaustion failure in #403, and a plain -//! false promise (config says 10, driver opens 1). -//! -//! This test builds two engines with different `max_connections` and checks, -//! via the node's `system_views.clients`, that the larger pool opens strictly -//! more client connections than the smaller one, and at least as many as its -//! configured size. It asserts a comparison rather than an exact count because -//! the driver also opens a control connection and other clients may be present; -//! the comparison isolates the effect of `max_connections` without depending on -//! that overhead. - -#[cfg(test)] -mod tests { - use crate::helpers::test_config; - use extenddb_storage_cassandra::CassandraEngine; - - /// Count client connections the node currently sees from the loopback - /// address, queried through the given engine's own session. Counts rows - /// client-side rather than using CQL `COUNT(*)`, whose aggregate column type - /// on a virtual table is awkward to deserialize. - async fn client_connection_count(engine: &CassandraEngine) -> i64 { - let body = engine - .session() - .query("SELECT address, port FROM system_views.clients") - .await - .expect("query system_views.clients") - .response_body() - .expect("response body"); - let rows = body.into_rows().unwrap_or_default(); - i64::try_from(rows.len()).unwrap_or(i64::MAX) - } - - #[tokio::test] - async fn pool_size_follows_max_connections() { - if crate::helpers::skip_without_cassandra() { - return; - } - - let small_size: u32 = 2; - let large_size: u32 = 8; - - // Build the small-pool engine and measure how many connections the node - // sees from us. Each engine opens its own pool, so we measure each - // against its own session. - let mut small_cfg = test_config(); - small_cfg.max_connections = small_size; - small_cfg.ensure_cached_connection_string(); - let small_engine = CassandraEngine::new(&small_cfg, "us-east-1") - .await - .expect("build small-pool engine"); - // Give the pool a moment to finish opening every connection. - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - let small_count = client_connection_count(&small_engine).await; - - let mut large_cfg = test_config(); - large_cfg.max_connections = large_size; - large_cfg.ensure_cached_connection_string(); - let large_engine = CassandraEngine::new(&large_cfg, "us-east-1") - .await - .expect("build large-pool engine"); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - // Count AFTER the large engine is up: this reflects both pools plus any - // other clients. The delta attributable to the large engine is - // large_total - small_count; asserting it carries large_size proves the - // config was applied and scales. - let both_count = client_connection_count(&large_engine).await; - - // Before the fix, create_session ignored max_connections and opened one - // connection per engine, so both engines together added ~2 and this - // assertion fails. With the fix, the large engine alone adds >= 8. - let large_delta = both_count - small_count; - assert!( - large_delta >= i64::from(large_size), - "large pool (max_connections={large_size}) added only {large_delta} connections \ - (small_count={small_count}, both_count={both_count}); expected at least \ - {large_size} — max_connections is not being applied to the pool (issue #403)" - ); - assert!( - large_delta > i64::from(small_size), - "large pool did not open more connections than the small pool \ - (delta={large_delta}, small_size={small_size}) — pool size does not follow \ - max_connections (issue #403)" - ); - } -} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index b6ed71936..a9ad516df 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -25,8 +25,6 @@ mod authorization_store; mod backup_engine; #[path = "direct/cassandra_engine.rs"] mod cassandra_engine; -#[path = "direct/connection_pool_size.rs"] -mod connection_pool_size; #[path = "direct/delete_item.rs"] mod delete_item; #[path = "direct/groups.rs"] diff --git a/docs/manuals/11-deployment-guide.md b/docs/manuals/11-deployment-guide.md index b2be935ec..5bb8ab69e 100755 --- a/docs/manuals/11-deployment-guide.md +++ b/docs/manuals/11-deployment-guide.md @@ -253,13 +253,10 @@ and refuses a backup that waited 30 seconds with `LimitExceededException`. A Mon 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 and Broadcast Address +### Cassandra Connection Pool -Two Cassandra-specific settings affect connection resilience. Both matter for -any production deployment. - -**Pool size.** The Cassandra backend opens `max_connections` connections per -node (default 10) and scatters requests across them: +The Cassandra backend opens `max_connections` connections per node (default 10) +and scatters requests across them: ```toml [storage.cassandra] @@ -267,39 +264,10 @@ 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. More -connections also divide the stream-id space, which avoids a driver-level -stream-id reuse hazard under sustained load with a slow statement in flight. - -**Broadcast RPC address (important).** The Cassandra driver reconnects to the -address each node advertises in `system.local`/`system.peers` -(`broadcast_rpc_address`), *not* to the contact point ExtendDB was configured -with. If a node advertises an address the ExtendDB host cannot reach, the -connection works at first but the pool never recovers from any later drop — a -node restart, a network blip, an idle timeout — and every request fails with -`No nodes available in query plan!` until ExtendDB is restarted. - -Configure each node's `broadcast_rpc_address` to an address every ExtendDB -instance can reach: - -```yaml -# cassandra.yaml -broadcast_rpc_address:
-``` - -This is the usual trap for **containerized** Cassandra: a container advertises -its internal bridge IP (for example `240.x` or `172.x`), which a client -connecting through a published host port cannot reach. For the official -`cassandra` image, set it through the environment: - -```bash -docker run ... -e CASSANDRA_BROADCAST_RPC_ADDRESS= cassandra:4.1 -``` - -For local development and CI, `devtools/cassandra-node` sets -`CASSANDRA_BROADCAST_RPC_ADDRESS=127.0.0.1` by default (override with -`--broadcast-rpc-address`), and the `integration-cassandra` CI workflow sets the -same on its service container. +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