diff --git a/crates/store/src/backend/scaleout/replica_health.rs b/crates/store/src/backend/scaleout/replica_health.rs index 0088890..9ed62e5 100644 --- a/crates/store/src/backend/scaleout/replica_health.rs +++ b/crates/store/src/backend/scaleout/replica_health.rs @@ -410,17 +410,42 @@ async fn replica_lag( let row: Option = match conn.query_first("SHOW REPLICA STATUS").await { Ok(row) => row, - Err(_) => return Ok(None), + Err(err) => { + report( + store.kind, + format!( + "Read replica {} won't report its status: {err}", + replica.label + ), + ); + return Ok(None); + } }; let Some(row) = row else { return Ok(None); }; - match row.get_opt::, _>("Seconds_Behind_Source") { - Some(Ok(Some(seconds))) => Ok(Some(seconds * 1000)), - Some(Ok(None)) => Err(trc::StoreEvent::MysqlError + let behind = ["Seconds_Behind_Source", "Seconds_Behind_Master"] + .into_iter() + .find_map(|column| row.get_opt::(column)) + .and_then(|value| value.ok()); + match behind { + Some(mysql_async::Value::NULL) => Err(trc::StoreEvent::MysqlError .into_err() .details("Replication is stopped")), - _ => Ok(None), + Some(value) => match mysql_async::from_value_opt::(value.clone()) { + Ok(seconds) => Ok(Some(seconds * 1000)), + Err(_) => { + report( + store.kind, + format!( + "Read replica {}: Seconds_Behind_Source isn't a number ({value:?})", + replica.label + ), + ); + Ok(None) + } + }, + None => Ok(None), } } } diff --git a/tests/src/store/mod.rs b/tests/src/store/mod.rs index fdadbac..6b99155 100644 --- a/tests/src/store/mod.rs +++ b/tests/src/store/mod.rs @@ -12,6 +12,8 @@ pub mod query; pub mod registry; #[cfg(feature = "postgres")] pub mod replica; // inbuxa: read replicas +#[cfg(feature = "mysql")] +pub mod replica_mysql; // inbuxa: read replicas on MySQL pub mod scaleout; // inbuxa: scale-out storage #[cfg(any(feature = "postgres", feature = "mysql"))] pub mod sql_timeout; diff --git a/tests/src/store/replica_mysql.rs b/tests/src/store/replica_mysql.rs new file mode 100644 index 0000000..7426c0a --- /dev/null +++ b/tests/src/store/replica_mysql.rs @@ -0,0 +1,264 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Read replicas on MySQL, from `docs/spec/features/scale-out-storage.md` +//! (tests 17, 18 and 19), against a source and a replica in containers: +//! one pair replicating with GTIDs, one by binary log position. Built with +//! `mysql`. + +use crate::utils::{ + containers::{MYSQL_GTID_PORTS, MYSQL_POS_PORTS, mysql_query}, + server::TestServerBuilder, +}; +use registry::schema::structs::{MySqlSettings, MySqlStore, SecretKeyOptional, SecretKeyValue}; +use std::{ + sync::{Arc, atomic::Ordering}, + time::{Duration, Instant}, +}; +use store::{ + Store, + backend::scaleout::replica::{ReplicaState, ReplicatedStore}, +}; + +const GTID_REPLICA: &str = "inbuxa-test-mysql-gtid-replica"; +const POSITION_REPLICA: &str = "inbuxa-test-mysql-pos-replica"; +const SECRET: &str = "mysql replica test passphrase"; + +async fn wait_for(store: &ReplicatedStore, state: ReplicaState, within: Duration, what: &str) { + let deadline = Instant::now() + within; + while store.replicas[0].state() != state { + assert!( + Instant::now() < deadline, + "{what}: the replica is {:?}, not {state:?}", + store.replicas[0].state() + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +fn replicated(store: &Store) -> Arc { + match store { + Store::Replicated(store) => store.clone(), + other => panic!("the data store isn't replicated: {other:?}"), + } +} + +fn message(subject: &str) -> Vec { + format!( + "From: sender@example.org\r\nTo: replica@example.org\r\nSubject: {subject}\r\n\r\nBody.\r\n" + ) + .into_bytes() +} + +fn settings(port: u16, user: &str, secret: &str) -> MySqlSettings { + MySqlSettings { + host: "localhost".into(), + port: port as u64, + database: "stalwart".into(), + auth_username: user.to_string().into(), + auth_secret: SecretKeyOptional::Value(SecretKeyValue { + secret: secret.into(), + }), + } +} + +async fn open(primary: u16, replica: MySqlSettings) -> Store { + store::backend::mysql::MysqlStore::open(MySqlStore { + host: "localhost".into(), + port: primary as u64, + auth_username: "root".to_string().into(), + auth_secret: SecretKeyOptional::Value(SecretKeyValue { + secret: "password".into(), + }), + database: "stalwart".into(), + allow_invalid_certs: true, + read_replicas: registry::types::list::List::from_iter([replica]), + ..Default::default() + }) + .await + .unwrap() +} + +/// Test 17: tests 9, 10 and 12 against a source and replica using GTIDs. +/// `cargo test -p tests --features mysql mysql_replica_tests -- --ignored`, +/// with `STORE=MySqlReplicated`. +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn mysql_replica_tests() { + assert_eq!( + std::env::var("STORE").as_deref(), + Ok("MySqlReplicated"), + "run with STORE=MySqlReplicated" + ); + let test = TestServerBuilder::new("mysql_replica_tests") + .await + .with_default_listeners() + .await + .build() + .await; + let store = replicated(test.server.store()); + wait_for(&store, ReplicaState::Up, Duration::from_secs(90), "ST-15").await; + + let admin = test.account("admin"); + let user = admin + .create_user_account("replica@example.org", SECRET, "Replica", &[], vec![]) + .await; + let client = user.jmap_client().await; + let inbox = client + .mailbox_query( + jmap_client::mailbox::query::Filter::role(jmap_client::mailbox::Role::Inbox).into(), + None::>, + ) + .await + .unwrap() + .take_ids() + .pop() + .unwrap(); + + // Test 9: reads served by the replica (ST-6) + let first = client + .email_import(message("First"), [inbox.clone()], None::>, None) + .await + .unwrap() + .take_id(); + tokio::time::sleep(Duration::from_secs(2)).await; + let before = store.replicas[0].reads.load(Ordering::Relaxed); + assert!( + client + .email_get(&first, None::>) + .await + .unwrap() + .is_some() + ); + assert!( + store.replicas[0].reads.load(Ordering::Relaxed) > before, + "test 17: Email/get on the replica" + ); + + // Test 10: with the replica's applier stopped, a client reads its own + // writes (ST-7) + mysql_query(GTID_REPLICA, "STOP REPLICA SQL_THREAD"); + let paused = client + .email_import( + message("Paused"), + [inbox.clone()], + None::>, + None, + ) + .await + .unwrap() + .take_id(); + assert!( + client + .email_get(&paused, None::>) + .await + .unwrap() + .is_some(), + "test 17: a new message reads back" + ); + + // Test 12: over the lag limit, no reads; back once caught up (ST-10) + let deadline = Instant::now() + Duration::from_secs(30); + while store.replicas[0].state() != ReplicaState::Lagging { + assert!(Instant::now() < deadline, "test 17: never over the limit"); + client + .email_import(message("Lag"), [inbox.clone()], None::>, None) + .await + .unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + } + let before = store.replicas[0].reads.load(Ordering::Relaxed); + assert!( + client + .email_get(&first, None::>) + .await + .unwrap() + .is_some() + ); + assert_eq!( + store.replicas[0].reads.load(Ordering::Relaxed), + before, + "test 17: no reads while lagging" + ); + mysql_query(GTID_REPLICA, "START REPLICA SQL_THREAD"); + wait_for(&store, ReplicaState::Up, Duration::from_secs(30), "test 17").await; + + test.temp_dir.delete(); +} + +/// Tests 18 and 19: replication by binary log position, where lag comes +/// from `Seconds_Behind_Source`. `cargo test -p tests --features mysql +/// mysql_replica_position_tests -- --ignored`, with +/// `STORE=MySqlReplicatedPosition`. +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn mysql_replica_position_tests() { + assert_eq!( + std::env::var("STORE").as_deref(), + Ok("MySqlReplicatedPosition"), + "run with STORE=MySqlReplicatedPosition" + ); + let test = TestServerBuilder::new("mysql_replica_position_tests") + .await + .with_default_listeners() + .await + .build() + .await; + let store = replicated(test.server.store()); + // Test 18: with the privilege, lag is measured and the replica is used + wait_for(&store, ReplicaState::Up, Duration::from_secs(90), "test 18").await; + assert!( + store.replicas[0].lag_ms.load(Ordering::Relaxed) < 5_000, + "test 18: Seconds_Behind_Source" + ); + + // Test 18: without REPLICATION CLIENT, the lag can't be measured, so + // the replica gets no reads + let plain = replicated( + &open( + MYSQL_POS_PORTS.0, + settings(MYSQL_POS_PORTS.1, "plain", "plain"), + ) + .await, + ); + wait_for( + &plain, + ReplicaState::Unmeasurable, + Duration::from_secs(60), + "test 18: no REPLICATION CLIENT", + ) + .await; + + // Test 19: applying in parallel without preserving commit order + mysql_query(POSITION_REPLICA, "STOP REPLICA"); + mysql_query( + POSITION_REPLICA, + "SET GLOBAL replica_preserve_commit_order=OFF; SET GLOBAL replica_parallel_workers=4", + ); + mysql_query(POSITION_REPLICA, "START REPLICA"); + let parallel = replicated( + &open( + MYSQL_POS_PORTS.0, + settings(MYSQL_POS_PORTS.1, "root", "password"), + ) + .await, + ); + wait_for( + ¶llel, + ReplicaState::Excluded, + Duration::from_secs(60), + "test 19", + ) + .await; + mysql_query(POSITION_REPLICA, "STOP REPLICA"); + mysql_query( + POSITION_REPLICA, + "SET GLOBAL replica_parallel_workers=0; SET GLOBAL replica_preserve_commit_order=ON", + ); + mysql_query(POSITION_REPLICA, "START REPLICA"); + + test.temp_dir.delete(); +} diff --git a/tests/src/utils/containers.rs b/tests/src/utils/containers.rs index a8ec8c9..f77031f 100644 --- a/tests/src/utils/containers.rs +++ b/tests/src/utils/containers.rs @@ -40,6 +40,15 @@ pub const PG_PRIMARY_PORT: u16 = 5442; pub const PG_REPLICA_PORT: u16 = 5443; pub const PG_REPLICA_CONTAINER: &str = "inbuxa-test-pg-replica"; pub const PG_PRIMARY_CONTAINER: &str = "inbuxa-test-pg-primary"; +static MYSQL_GTID: OnceCell<(ContainerAsync, ContainerAsync)> = + OnceCell::const_new(); +static MYSQL_POS: OnceCell<(ContainerAsync, ContainerAsync)> = + OnceCell::const_new(); +const MYSQL_REPLICATION_NETWORK: &str = "inbuxa-test-mysql-replication"; +/// The MySQL pair that replicates with GTIDs, and the one that replicates +/// by binary log position. +pub const MYSQL_GTID_PORTS: (u16, u16) = (3317, 3318); +pub const MYSQL_POS_PORTS: (u16, u16) = (3327, 3328); const OPENLDAP_LDAPI_URL: &str = "ldapi://%2Fvar%2Frun%2Fslapd%2Fldapi/"; @@ -623,7 +632,9 @@ pub async fn ensure_postgres_replicated() { /// inbuxa: runs SQL on a test container, through `psql`. pub fn psql(container: &str, sql: &str) -> String { let output = std::process::Command::new("docker") - .args(["exec", container, "psql", "-U", "stalwart", "-d", "stalwart", "-tAc", sql]) + .args([ + "exec", container, "psql", "-U", "stalwart", "-d", "stalwart", "-tAc", sql, + ]) .output() .expect("docker exec"); String::from_utf8_lossy(&output.stdout).trim().to_string() @@ -637,3 +648,133 @@ pub fn docker(action: &str, container: &str) { .expect("docker"); assert!(status.success(), "docker {action} {container}"); } + +fn mysql_container( + name: &str, + port: u16, + gtid: bool, + server_id: &str, +) -> testcontainers::ContainerRequest { + let mut cmd = vec![ + format!("--server-id={server_id}"), + "--log-bin=binlog".to_string(), + "--binlog-format=ROW".to_string(), + "--default-authentication-plugin=mysql_native_password".to_string(), + ]; + if gtid { + cmd.push("--gtid-mode=ON".to_string()); + cmd.push("--enforce-gtid-consistency=ON".to_string()); + } + let _ = name; + GenericImage::new("mysql", "8.0") + .with_wait_for(WaitFor::message_on_stderr("port: 3306 MySQL")) + .with_env_var("MYSQL_ROOT_PASSWORD", "password") + .with_env_var("MYSQL_DATABASE", "stalwart") + .with_cmd(cmd) + .with_network(MYSQL_REPLICATION_NETWORK) + .with_mapped_port(port, 3306.tcp()) + .with_startup_timeout(READY_TIMEOUT) + .with_container_name(name) + .with_reuse(ReuseDirective::Always) +} + +/// inbuxa: runs SQL on a MySQL test container, through the `mysql` client. +pub fn mysql_query(container: &str, sql: &str) -> String { + let output = std::process::Command::new("docker") + .args([ + "exec", + container, + "mysql", + "-uroot", + "-ppassword", + "-N", + "-B", + "-e", + sql, + ]) + .output() + .expect("docker exec"); + if !output.status.success() { + panic!( + "{sql}: {}", + String::from_utf8_lossy(&output.stderr).trim().to_string() + ); + } + String::from_utf8_lossy(&output.stdout).trim().to_string() +} + +/// inbuxa: a MySQL source and replica, replicating with GTIDs or by binary +/// log position (ST-10, ST-15). +pub async fn ensure_mysql_replicated(gtid: bool) { + let (cell, ports, names) = if gtid { + ( + &MYSQL_GTID, + MYSQL_GTID_PORTS, + ( + "inbuxa-test-mysql-gtid-primary", + "inbuxa-test-mysql-gtid-replica", + ), + ) + } else { + ( + &MYSQL_POS, + MYSQL_POS_PORTS, + ( + "inbuxa-test-mysql-pos-primary", + "inbuxa-test-mysql-pos-replica", + ), + ) + }; + cell.get_or_init(|| async { + let primary = mysql_container(names.0, ports.0, gtid, "1") + .start() + .await + .expect("Failed to start the MySQL source"); + let replica = mysql_container(names.1, ports.1, gtid, "2") + .start() + .await + .expect("Failed to start the MySQL replica"); + wait_for_tcp(ports.0).await; + wait_for_tcp(ports.1).await; + + // A replication account, and one without REPLICATION CLIENT (test 18) + mysql_query( + names.0, + concat!( + "CREATE USER IF NOT EXISTS 'repl'@'%' IDENTIFIED WITH mysql_native_password BY 'repl';", + "GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';", + "CREATE USER IF NOT EXISTS 'plain'@'%' IDENTIFIED WITH mysql_native_password BY 'plain';", + "GRANT ALL ON stalwart.* TO 'plain'@'%'; FLUSH PRIVILEGES;" + ), + ); + if mysql_query(names.1, "SELECT SERVICE_STATE FROM performance_schema.replication_applier_status") + .is_empty() + { + let source = if gtid { + format!( + "CHANGE REPLICATION SOURCE TO SOURCE_HOST='{}', SOURCE_PORT=3306, \ + SOURCE_USER='repl', SOURCE_PASSWORD='repl', SOURCE_AUTO_POSITION=1", + names.0 + ) + } else { + let status = mysql_query(names.0, "SHOW MASTER STATUS"); + let mut fields = status.split('\t'); + let file = fields.next().unwrap_or_default().to_string(); + let position = fields.next().unwrap_or("4").to_string(); + format!( + "CHANGE REPLICATION SOURCE TO SOURCE_HOST='{}', SOURCE_PORT=3306, \ + SOURCE_USER='repl', SOURCE_PASSWORD='repl', SOURCE_LOG_FILE='{file}', \ + SOURCE_LOG_POS={position}", + names.0 + ) + }; + mysql_query(names.1, &source); + mysql_query(names.1, "START REPLICA"); + mysql_query(names.1, "SET GLOBAL super_read_only=ON"); + } + (primary, replica) + }) + .await; + wait_for_tcp(ports.0).await; + wait_for_tcp(ports.1).await; +} diff --git a/tests/src/utils/storage.rs b/tests/src/utils/storage.rs index e8919ee..414d04c 100644 --- a/tests/src/utils/storage.rs +++ b/tests/src/utils/storage.rs @@ -91,6 +91,43 @@ pub async fn build_data_store(typ: &str, path: &str) -> DataStore { ..Default::default() }); } + // inbuxa: a MySQL source with a replica, with and without GTIDs + if let Some(gtid) = match typ { + "MySqlReplicated" => Some(true), + "MySqlReplicatedPosition" => Some(false), + _ => None, + } { + crate::utils::containers::ensure_mysql_replicated(gtid).await; + let (primary, replica) = if gtid { + crate::utils::containers::MYSQL_GTID_PORTS + } else { + crate::utils::containers::MYSQL_POS_PORTS + }; + let secret = || { + SecretKeyOptional::Value(SecretKeyValue { + secret: "password".into(), + }) + }; + return DataStore::MySql(MySqlStore { + host: "localhost".into(), + port: primary as u64, + auth_username: "root".to_string().into(), + auth_secret: secret(), + database: "stalwart".into(), + use_tls: false, + allow_invalid_certs: true, + read_replicas: registry::types::list::List::from_iter([ + registry::schema::structs::MySqlSettings { + host: "localhost".into(), + port: replica as u64, + database: "stalwart".into(), + auth_username: "root".to_string().into(), + auth_secret: secret(), + }, + ]), + ..Default::default() + }); + } if typ == "MariaDb" { crate::utils::containers::ensure_mariadb().await; return DataStore::MySql(MySqlStore {