fix(storage-cassandra): apply max_connections and override CDRS default reconnect policy

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;
@@ -114,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<
@@ -186,11 +205,36 @@ 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 = 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)
+            // 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)
             .await
@@ -299,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
@@ -314,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);
+    }
+}

crates/storage-cassandra/src/migrations.rs

@@ -241,7 +241,91 @@ async fn execute_migration(session: &Arc<CassandraSession>, 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<CassandraSession>) -> 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<CassandraSession>) -> OpResult<bool> {
+    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<uuid::Uuid> = 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<uuid::Uuid> = 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.

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;
+        }
+    }
+}

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"]

docs/manuals/11-deployment-guide.md

@@ -244,7 +244,30 @@ pool_size = 50  # Increase for higher concurrency
 
 Ensure PostgreSQL `max_connections` accommodates the total pool size plus overhead.
 
-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.
+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