Scale-out storage: read replicas on MySQL, tested (ST-10, ST-15, tests 17 to 19)

Two source-and-replica pairs run in containers: one replicating with
GTIDs, one by binary log position. mysql_replica_tests covers test 17
(tests 9, 10 and 12 with GTIDs) and mysql_replica_position_tests covers
test 18 (lag from Seconds_Behind_Source, and a replica whose account
lacks REPLICATION CLIENT getting no reads) and test 19 (a parallel
replica without replica_preserve_commit_order left out at startup).

The lag reader now takes Seconds_Behind_Source whatever numeric type the
server returns, accepts the older column name, and says in the log why it
gave up measuring.

crates/store/src/backend/scaleout/replica_health.rs

@@ -410,17 +410,42 @@ async fn replica_lag(
                 let row: Option<mysql_async::Row> =
                     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::<Option<u64>, _>("Seconds_Behind_Source") {
+                let behind = ["Seconds_Behind_Source", "Seconds_Behind_Master"]
-                    Some(Ok(Some(seconds))) => Ok(Some(seconds * 1000)),
+                    .into_iter()
-                    Some(Ok(None)) => Err(trc::StoreEvent::MysqlError
+                    .find_map(|column| row.get_opt::<mysql_async::Value, _>(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::<u64>(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),
                 }
             }
         }

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;

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<ReplicatedStore> {
+    match store {
+        Store::Replicated(store) => store.clone(),
+        other => panic!("the data store isn't replicated: {other:?}"),
+    }
+}
+
+fn message(subject: &str) -> Vec<u8> {
+    format!(
+        "From: [email protected]\r\nTo: [email protected]\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("[email protected]", 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::<Vec<_>>,
+        )
+        .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::<Vec<String>>, 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::<Vec<_>>)
+            .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::<Vec<String>>,
+            None,
+        )
+        .await
+        .unwrap()
+        .take_id();
+    assert!(
+        client
+            .email_get(&paused, None::<Vec<_>>)
+            .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::<Vec<String>>, 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::<Vec<_>>)
+            .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(
+        &parallel,
+        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();
+}

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<GenericImage>, ContainerAsync<GenericImage>)> =
+    OnceCell::const_new();
+static MYSQL_POS: OnceCell<(ContainerAsync<GenericImage>, ContainerAsync<GenericImage>)> =
+    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<GenericImage> {
+    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;
+}

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 {