Update deps

Author: rukai

Cargo.lock

@@ -21,7 +21,11 @@ inherits = "release"
 debug = true
 
 [workspace.dependencies]
-scylla = { version = "1.0.0", features = ["openssl-010"] }
+# scylla 1.5.0 break our integration tests, requires further investigation.
+# issues relate to:
+# * routing to ports other than 9042 is broken for the await_schema_agreement session.
+# * out of rack routing checks?
+scylla = { version = "=1.4.0", features = ["openssl-010"] }
 bytes = { version = "1.11.1", features = ["serde"] }
 tokio = { version = "1.44.2", features = ["full"] }
 tokio-util = { version = "0.7.7", features = ["codec"] }
@@ -31,7 +35,7 @@ anyhow = "1.0.76"
 serde = { version = "1.0.111", features = ["derive"] }
 serde_yaml = "0.9.17"
 uuid = { version = "1.0.0", features = ["serde", "v4"] }
-reqwest = { version = "0.12.0", default-features = false, features = ["http2"] }
+reqwest = { version = "0.13.0", default-features = false, features = ["http2"] }
 redis = { version = "0.29.2", features = ["tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "cluster"] }
 cdrs-tokio = "8.0"
 cassandra-protocol = "3.0"
@@ -41,8 +45,8 @@ tracing = { version = "0.1.15", features = ["release_max_level_debug"] }
 tracing-subscriber = { version = "0.3.1", features = ["env-filter", "json"] }
 tracing-appender = "0.2.0"
 serde_json = "1.0"
-rcgen = { version = "0.13.0", default-features = false, features = ["pem", "aws_lc_rs"] }
-subprocess = "0.2.7"
+rcgen = { version = "0.14.0", default-features = false, features = ["pem", "aws_lc_rs"] }
+subprocess = "1.0.0"
 chacha20poly1305 = { version = "0.10.0", features = ["std"] }
 csv = "1.2.0"
 redis-protocol = { version = "6.0.0", features = ["bytes"] }

Cargo.toml

@@ -14,5 +14,5 @@ tracing-subscriber.workspace = true
 aws-throwaway.workspace = true
 tracing-appender.workspace = true
 shellfish = { version = "0.10.0", features = ["async"] }
-cargo_metadata = "0.20.0"
+cargo_metadata = "0.23.0"
 shell-quote.workspace = true

ec2-cargo/Cargo.toml

@@ -17,7 +17,7 @@ scylla.workspace = true
 anyhow.workspace = true
 tokio.workspace = true
 tracing.workspace = true
-rstest = "0.25.0"
+rstest = "0.26.0"
 rstest_reuse = "0.7.0"
 test-helpers = { path = "../test-helpers" }
 redis.workspace = true

shotover-proxy/Cargo.toml

@@ -113,7 +113,7 @@ pub enum Operation {
 }
 
 impl Operation {
-    async fn prepare(&self, session: &Arc<CassandraSession>, db: &CassandraDb) {
+    async fn prepare(&self, session: &CassandraSession, db: &CassandraDb) {
         if let CassandraDb::Cassandra = db {
             session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }").await;
             session.await_schema_agreement().await;
@@ -147,7 +147,7 @@ impl Operation {
 
     async fn run(
         &self,
-        session: &Arc<CassandraSession>,
+        session: &CassandraSession,
         reporter: UnboundedSender<Report>,
         parameters: BenchParameters,
     ) {
@@ -229,13 +229,16 @@ pub enum CassandraDriver {
     CdrsTokio,
 }
 
+#[derive(Clone)]
 pub enum CassandraSession {
-    Scylla(ScyllaSession),
+    Scylla(Arc<ScyllaSession>),
     CdrsTokio(
-        CdrsTokioSession<
-            TransportTcp,
-            TcpConnectionManager,
-            RoundRobinLoadBalancingStrategy<TransportTcp, TcpConnectionManager>,
+        Arc<
+            CdrsTokioSession<
+                TransportTcp,
+                TcpConnectionManager,
+                RoundRobinLoadBalancingStrategy<TransportTcp, TcpConnectionManager>,
+            >,
         >,
     ),
 }
@@ -691,7 +694,7 @@ impl Bench for CassandraBench {
         let address = resources;
 
         let session = match (self.protocol, self.driver) {
-            (CassandraProtocol::V4, CassandraDriver::Scylla) => Arc::new(CassandraSession::Scylla(
+            (CassandraProtocol::V4, CassandraDriver::Scylla) => CassandraSession::Scylla(Arc::new(
                 ScyllaSessionBuilder::new()
                     .known_nodes([address])
                     .user("cassandra", "cassandra")
@@ -723,7 +726,7 @@ impl Bench for CassandraBench {
                     .build()
                     .await
                     .unwrap();
-                Arc::new(CassandraSession::CdrsTokio(
+                CassandraSession::CdrsTokio(Arc::new(
                     TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
                         .with_compression(match self.compression {
                             Compression::None => CdrsCompression::None,
@@ -823,7 +826,7 @@ struct AwsNodeInfo {
 
 #[derive(Clone)]
 struct BenchTaskCassandra {
-    session: Arc<CassandraSession>,
+    session: CassandraSession,
     query: PreparedStatement,
     operation: Operation,
 }

shotover-proxy/benches/windsock/cassandra/bench.rs

@@ -2,6 +2,7 @@ use crate::shotover_process;
 use test_helpers::shotover_process::{EventMatcher, Level};
 
 #[tokio::test]
+#[cfg(feature = "cassandra")]
 async fn test_early_shutdown_cassandra_source() {
     shotover_process("tests/test-configs/null-cassandra/topology.yaml")
         .start()
@@ -113,6 +114,7 @@ Caused by:
 }
 
 #[tokio::test]
+#[cfg(all(feature = "valkey", feature = "cassandra"))]
 async fn test_shotover_shutdown_when_protocol_mismatch() {
     shotover_process("tests/test-configs/invalid_protocol_mismatch.yaml")
         .assert_fails_to_start(&[EventMatcher::new()

shotover-proxy/tests/runner/runner_int_tests.rs

@@ -396,7 +396,12 @@ mod test {
         );
         match try_parse_log_directives(&[Some("good=info,bad=blah,warn")]) {
             Ok(_) => panic!(),
-            Err(e) => assert_eq!(e.to_string(), "invalid filter directive: bad=blah"),
+            Err(e) => {
+                assert_eq!(
+                    e.to_string(),
+                    r#"error parsing level filter: expected one of "off", "error", "warn", "info", "debug", "trace", or a number 0-5: bad=blah"#
+                )
+            }
         }
     }
 }

shotover/src/runner.rs

@@ -32,8 +32,9 @@ tracing-subscriber.workspace = true
 anyhow.workspace = true
 rcgen.workspace = true
 rdkafka = { version = "0.37", features = ["cmake-build"], optional = true }
+rdkafka-sys = "=4.8.0"
 docker-compose-runner = "0.3.0"
-j4rs = "0.22.0"
+j4rs = "0.24.0"
 futures-util = "0.3.28"
 http = "1.1.0"
 rustls = { version = "0.23.18", default-features = false, features = [

test-helpers/Cargo.toml

@@ -1,5 +1,7 @@
 use crate::run_command;
-use rcgen::{BasicConstraints, Certificate, CertificateParams, DnType, IsCa, KeyPair, SanType};
+use rcgen::{
+    BasicConstraints, Certificate, CertificateParams, DnType, IsCa, Issuer, KeyPair, SanType,
+};
 use std::path::Path;
 
 pub fn generate_test_certs(path: &Path) {
@@ -32,16 +34,16 @@ pub fn generate_test_certs_with_bad_san(path: &Path) {
 }
 
 pub fn generate_test_certs_with_sans(path: &Path, sans: Vec<SanType>) {
-    let (ca_cert, ca_key) = new_ca();
-    let (cert, cert_key) = new_cert(sans, &ca_cert, &ca_key);
+    let (ca_cert, issuer) = new_ca();
+    let (cert, cert_key) = new_cert(sans, &issuer);
 
     std::fs::create_dir_all(path).unwrap();
     std::fs::write(path.join("localhost_CA.crt"), ca_cert.pem()).unwrap();
     std::fs::write(path.join("localhost.crt"), cert.pem()).unwrap();
     std::fs::write(path.join("localhost.key"), cert_key.serialize_pem()).unwrap();
 }
 
-fn new_ca() -> (Certificate, KeyPair) {
+fn new_ca() -> (Certificate, Issuer<'static, KeyPair>) {
     let mut params = CertificateParams::default();
     params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
     // This must be "Certificate Authority"
@@ -55,10 +57,11 @@ fn new_ca() -> (Certificate, KeyPair) {
 
     let key_pair = KeyPair::generate().unwrap();
     let ca_cert = params.self_signed(&key_pair).unwrap();
-    (ca_cert, key_pair)
+    let issuer = Issuer::new(params, key_pair);
+    (ca_cert, issuer)
 }
 
-fn new_cert(sans: Vec<SanType>, ca_cert: &Certificate, ca_key: &KeyPair) -> (Certificate, KeyPair) {
+fn new_cert(sans: Vec<SanType>, issuer: &Issuer<'_, KeyPair>) -> (Certificate, KeyPair) {
     let mut params = CertificateParams::default();
 
     // This needs to refer to the hosts that certificate will be used by
@@ -72,7 +75,7 @@ fn new_cert(sans: Vec<SanType>, ca_cert: &Certificate, ca_key: &KeyPair) -> (Cer
         .distinguished_name
         .push(DnType::OrganizationName, "ShotoverTestCertificate");
     let cert_key = KeyPair::generate().unwrap();
-    let cert = params.signed_by(&cert_key, ca_cert, ca_key).unwrap();
+    let cert = params.signed_by(&cert_key, issuer).unwrap();
     (cert, cert_key)
 }
 

test-helpers/src/cert.rs

@@ -176,7 +176,7 @@ pub enum CassandraConnection {
     #[cfg(feature = "cassandra-cpp-driver-tests")]
     Cpp(CppConnection),
     Cdrs(CdrsConnection),
-    Scylla(ScyllaConnection),
+    Scylla(Box<ScyllaConnection>),
     Java(JavaConnection),
 }
 
@@ -197,9 +197,9 @@ impl CassandraConnection {
             CassandraDriver::Cdrs => CassandraConnection::Cdrs(
                 CdrsConnection::new(contact_points, port, compression, tls, protocol).await,
             ),
-            CassandraDriver::Scylla => CassandraConnection::Scylla(
+            CassandraDriver::Scylla => CassandraConnection::Scylla(Box::new(
                 ScyllaConnection::new(contact_points, port, compression, tls, protocol).await,
-            ),
+            )),
             CassandraDriver::Java => CassandraConnection::Java(
                 JavaConnection::new(contact_points, port, compression, tls, protocol).await,
             ),
@@ -242,6 +242,10 @@ impl CassandraConnection {
             SessionBuilderScylla::new()
                 .known_node(direct_node)
                 .user("cassandra", "cassandra")
+                .auto_await_schema_agreement(false)
+                // We do not need to refresh metadata as there is nothing else fiddling with the topology or schema.
+                // By default the metadata refreshes every 60s and that can cause performance issues so we disable it by using an absurdly high refresh interval
+                .cluster_metadata_refresh_interval(Duration::from_secs(10000000000))
                 .tls_context(context)
                 .build()
                 .await