From 1518c69033e024258f932e761b05255582acac0c Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 14:05:59 -0700 Subject: [PATCH] Scale-out storage: PostgreSQL and MySQL read replicas (ST-5 to ST-15) A data store with readReplicas becomes a replicated store. Writes, operator-written SQL and everything outside a read scope go to the primary. JMAP reads before a request's first write, IMAP LIST, STATUS, SEARCH, SORT and FETCH, POP3 RETR and TOP, DAV GET, PROPFIND and REPORT, and blob downloads run in a read scope. Only account data (properties, indexes, change logs, counters, ACLs, blobs, the search index) is read from a replica; the registry, in-memory values, the task queue and the rest stay on the primary. In a scope, the first read picks a replica round-robin among those up and under the lag limit, and only if it has every change this node has written or heard of for the scope's accounts: marks come from write results, the cluster's state-change broadcasts, a sinceState the client presents, and, with more than one node, Redis. A write inside the scope sends the rest of it to the primary. A miss on a replica is looked up on the primary, and a replica error retries the read there and marks the replica down. Each node samples lag every second (WAL positions on PostgreSQL; GTID sets or Seconds_Behind_Source on MySQL), stops reading from a replica over 5 s and starts again under 2.5 s, and probes a down replica every 10 s. At startup a replica is left out if it's the primary, isn't read-only, applies out of commit order, or doesn't show a marker written to the primary within six tries. replica_tests (postgres, STORE=PostgreSqlReplicated) runs a primary and a streaming hot standby in containers: tests 9, 10, 12, 13, 14 and 15 pass. --- crates/common/src/config/storage.rs | 9 +- crates/dav/src/request.rs | 37 ++ crates/http/src/request.rs | 8 +- crates/imap/src/core/client.rs | 25 +- crates/imap/src/op/mod.rs | 6 +- crates/jmap/src/api/request.rs | 60 ++- crates/pop3/src/client.rs | 34 +- crates/services/src/broadcast/subscriber.rs | 17 + crates/store/src/backend/mysql/main.rs | 43 +- crates/store/src/backend/mysql/mod.rs | 2 +- crates/store/src/backend/postgres/main.rs | 53 ++- crates/store/src/backend/postgres/mod.rs | 4 +- crates/store/src/backend/scaleout/mod.rs | 21 + crates/store/src/backend/scaleout/replica.rs | 369 +++++++++++++++ .../src/backend/scaleout/replica_health.rs | 429 ++++++++++++++++++ crates/store/src/dispatch/blob.rs | 33 ++ crates/store/src/dispatch/mod.rs | 2 + crates/store/src/dispatch/search.rs | 45 ++ crates/store/src/dispatch/store.rs | 103 +++++ crates/store/src/lib.rs | 11 +- tests/src/store/mod.rs | 2 + tests/src/store/replica.rs | 329 ++++++++++++++ tests/src/utils/containers.rs | 96 ++++ tests/src/utils/storage.rs | 29 ++ 24 files changed, 1722 insertions(+), 45 deletions(-) create mode 100644 crates/store/src/backend/scaleout/replica.rs create mode 100644 crates/store/src/backend/scaleout/replica_health.rs create mode 100644 tests/src/store/replica.rs diff --git a/crates/common/src/config/storage.rs b/crates/common/src/config/storage.rs index 6bc87b4..235b84e 100644 --- a/crates/common/src/config/storage.rs +++ b/crates/common/src/config/storage.rs @@ -41,12 +41,19 @@ impl Storage { ); } + let coordinator = Coordinator::build(bp, &memory).await.unwrap_or_default(); + // inbuxa: ST-7: with more than one node, read replicas share + // high-water marks through the in-memory store + if !matches!(coordinator, Coordinator::None) { + bp.data_store.share_marks(&memory); + } + Storage { registry: bp.registry.clone(), data: bp.data_store.clone(), blob: BlobStore::build(bp).await.unwrap_or_default(), search, - coordinator: Coordinator::build(bp, &memory).await.unwrap_or_default(), + coordinator, memory, tracing: Store::build_tracing(bp).await.unwrap_or_default(), metrics: Store::build_metrics(bp).await.unwrap_or_default(), diff --git a/crates/dav/src/request.rs b/crates/dav/src/request.rs index 2aaa3cb..0c3f590 100644 --- a/crates/dav/src/request.rs +++ b/crates/dav/src/request.rs @@ -74,9 +74,19 @@ pub(crate) trait DavRequestDispatcher: Sync + Send { method: DavMethod, body: Vec, ) -> impl Future> + Send; + + fn dispatch_dav_inner( + &self, + headers: &RequestHeaders<'_>, + access_token: AccessToken, + resource: DavResourceName, + method: DavMethod, + body: Vec, + ) -> impl Future> + Send; } impl DavRequestDispatcher for Server { + // inbuxa: ST-6: GET, PROPFIND and REPORT may be served by a read replica async fn dispatch_dav_request( &self, headers: &RequestHeaders<'_>, @@ -84,6 +94,33 @@ impl DavRequestDispatcher for Server { resource: DavResourceName, method: DavMethod, body: Vec, + ) -> crate::Result { + if matches!( + method, + DavMethod::GET | DavMethod::HEAD | DavMethod::PROPFIND | DavMethod::REPORT + ) { + let accounts = access_token + .all_ids() + .map(|account_id| (account_id, 0)) + .collect::>(); + store::backend::scaleout::replica::replica_read( + accounts, + self.dispatch_dav_inner(headers, access_token, resource, method, body), + ) + .await + } else { + self.dispatch_dav_inner(headers, access_token, resource, method, body) + .await + } + } + + async fn dispatch_dav_inner( + &self, + headers: &RequestHeaders<'_>, + access_token: AccessToken, + resource: DavResourceName, + method: DavMethod, + body: Vec, ) -> crate::Result { // Dispatch match method { diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index 11f9d7c..a3bdadd 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -151,7 +151,13 @@ impl ParseHttp for Server { path.next().and_then(BlobId::from_base32), path.next(), ) { - return match self.blob_download(&blob_id, &access_token).await? { + // inbuxa: ST-6: a download may be served by a read replica + let blob = store::backend::scaleout::replica::replica_read( + access_token.all_ids().map(|account_id| (account_id, 0)), + self.blob_download(&blob_id, &access_token), + ) + .await?; + return match blob { Some(blob) => Ok(DownloadResponse { filename: name.to_string(), content_type: req diff --git a/crates/imap/src/core/client.rs b/crates/imap/src/core/client.rs index 2de60d8..cba267c 100644 --- a/crates/imap/src/core/client.rs +++ b/crates/imap/src/core/client.rs @@ -93,7 +93,25 @@ impl Session { let mut requests = requests.into_iter().peekable(); while let Some(request) = requests.next() { - let result = match request.command { + // inbuxa: ST-6: commands that only read may be served by a read + // replica; any write they make still goes to the primary + let replica_accounts = match (&request.command, &self.state) { + ( + Command::List + | Command::Lsub + | Command::Status + | Command::Search(_) + | Command::Sort(_) + | Command::Fetch(_), + State::Authenticated { data } | State::Selected { data, .. }, + ) => data + .access_token + .all_ids() + .map(|account_id| (account_id, 0)) + .collect::>(), + _ => Vec::new(), + }; + let dispatch = async { match request.command { Command::List | Command::Lsub => self .handle_list(request) .await @@ -256,6 +274,11 @@ impl Session { .handle_uidbatches(request) .await .map(|_| SessionResult::Continue), + } }; + let result = if replica_accounts.is_empty() { + dispatch.await + } else { + store::backend::scaleout::replica::replica_read(replica_accounts, dispatch).await }; match result { diff --git a/crates/imap/src/op/mod.rs b/crates/imap/src/op/mod.rs index 70cdee3..de67d8b 100644 --- a/crates/imap/src/op/mod.rs +++ b/crates/imap/src/op/mod.rs @@ -62,8 +62,8 @@ impl ToModSeq for u64 { macro_rules! spawn_op { ($data:expr, $($code:tt)*) => { { - - tokio::spawn(async move { + // inbuxa: ST-6: the operation keeps its read scope + tokio::spawn(store::backend::scaleout::replica::carry(async move { let data = &($data); if let Err(err) = (async { @@ -73,7 +73,7 @@ macro_rules! spawn_op { { let _ = data.write_error(err).await; } - }); + })); Ok(())} }; diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index 064895c..e7aa6cf 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -94,6 +94,10 @@ impl RequestHandler for Server { request.method_calls.len(), ); + // inbuxa: ST-6: reads before the request's first write may go to a + // read replica + let mut has_written = false; + for mut call in request.method_calls { // Resolve result and id references if let Err(error) = response.resolve_references(&mut call.method) { @@ -126,15 +130,59 @@ impl RequestHandler for Server { // Add response let method_name = call.name.as_str(); - match self - .handle_method_call( + + // inbuxa: ST-6, ST-7: a read before the first write may use a + // replica that has every change the client has seen + let eligible = !has_written + && matches!( call.method, - call.name, - access_token, - &mut next_call, - session, + RequestMethod::Get(_) + | RequestMethod::Query(_) + | RequestMethod::Changes(_) + | RequestMethod::QueryChanges(_) + ); + if matches!( + call.method, + RequestMethod::Set(_) + | RequestMethod::Copy(_) + | RequestMethod::ImportEmail(_) + | RequestMethod::UploadBlob(_) + ) { + has_written = true; + } + let presented = match &call.method { + RequestMethod::Changes(changes) => match &changes.since_state { + jmap_proto::types::state::State::Exact(change_id) => { + Some((changes.account_id.document_id(), *change_id)) + } + _ => None, + }, + _ => None, + }; + let method_call = self.handle_method_call( + call.method, + call.name, + access_token, + &mut next_call, + session, + ); + let result = if eligible { + store::backend::scaleout::replica::replica_read( + access_token.all_ids().map(|account_id| { + ( + account_id, + presented + .filter(|(id, _)| *id == account_id) + .map_or(0, |(_, change_id)| change_id), + ) + }), + method_call, ) .await + } else { + method_call.await + }; + match result { Ok(mut method_response) => { match &mut method_response { diff --git a/crates/pop3/src/client.rs b/crates/pop3/src/client.rs index ab171bd..183724d 100644 --- a/crates/pop3/src/client.rs +++ b/crates/pop3/src/client.rs @@ -113,10 +113,21 @@ impl Session { Command::List { msg } => { self.handle_list(msg).await.map(|_| SessionResult::Continue) } - Command::Retr { msg } => self - .handle_fetch(msg, None) + Command::Retr { msg } => { + // inbuxa: ST-6: may be served by a read replica + let accounts = self + .state + .access_token() + .all_ids() + .map(|account_id| (account_id, 0)) + .collect::>(); + store::backend::scaleout::replica::replica_read( + accounts, + self.handle_fetch(msg, None), + ) .await - .map(|_| SessionResult::Continue), + .map(|_| SessionResult::Continue) + } Command::Dele { msg } => self .handle_dele(vec![msg]) .await @@ -125,10 +136,21 @@ impl Session { .handle_dele(msgs) .await .map(|_| SessionResult::Continue), - Command::Top { msg, n } => self - .handle_fetch(msg, n.into()) + Command::Top { msg, n } => { + // inbuxa: ST-6: may be served by a read replica + let accounts = self + .state + .access_token() + .all_ids() + .map(|account_id| (account_id, 0)) + .collect::>(); + store::backend::scaleout::replica::replica_read( + accounts, + self.handle_fetch(msg, n.into()), + ) .await - .map(|_| SessionResult::Continue), + .map(|_| SessionResult::Continue) + } Command::Uidl { msg } => { self.handle_uidl(msg).await.map(|_| SessionResult::Continue) } diff --git a/crates/services/src/broadcast/subscriber.rs b/crates/services/src/broadcast/subscriber.rs index d12f24b..55d6efd 100644 --- a/crates/services/src/broadcast/subscriber.rs +++ b/crates/services/src/broadcast/subscriber.rs @@ -105,6 +105,23 @@ pub fn spawn_broadcast_subscriber(inner: Arc, mut shutdown_rx: watch::Rec ); match event { BroadcastEvent::PushNotification(notification) => { + // inbuxa: ST-7: another node's change raises this + // node's high-water mark for the account + match ¬ification { + common::ipc::PushNotification::StateChange(change) => inner + .build_server() + .core + .storage + .data + .note_change(change.account_id, change.change_id), + common::ipc::PushNotification::EmailPush(push) => inner + .build_server() + .core + .storage + .data + .note_change(push.account_id, push.change_id), + _ => {} + } if inner .ipc .push_tx diff --git a/crates/store/src/backend/mysql/main.rs b/crates/store/src/backend/mysql/main.rs index 1206156..0655bca 100644 --- a/crates/store/src/backend/mysql/main.rs +++ b/crates/store/src/backend/mysql/main.rs @@ -20,6 +20,8 @@ use mysql_async::{ impl MysqlStore { pub async fn open(config: structs::MySqlStore) -> Result { + // inbuxa: ST-15: where the primary is, to tell a replica from it + let primary_location = (config.host.clone(), config.port as u16, config.database.clone()); let mut opts = OptsBuilder::default() .ip_or_hostname(config.host) .user(config.auth_username) @@ -51,24 +53,43 @@ impl MysqlStore { PoolOpts::default().with_constraints(PoolConstraints::new(pool_min, pool_max).unwrap()), ); - // inbuxa: ST-2: replicas aren't used yet (the scale-out decision), so - // each one is reported rather than silently ignored + // inbuxa: ST-5 to ST-15: each replica inherits the primary's settings + // except where it is and how to sign in + let mut replicas = vec![]; for replica in config.read_replicas { - trc::event!( - Store(trc::StoreEvent::MysqlError), - Details = format!( - "Read replica {}:{} {} isn't used yet: every operation goes to the primary", - replica.host, replica.port, replica.database - ), - ); + replicas.push(crate::backend::scaleout::replica::Replica::new( + Store::MySQL(Arc::new(MysqlStore { + conn_pool: Pool::new( + opts.clone() + .ip_or_hostname(replica.host.clone()) + .user(replica.auth_username) + .pass(replica.auth_secret.secret().await?.map(|v| v.into_owned())) + .db_name(Some(replica.database.clone())) + .tcp_port(replica.port as u16), + ), + })), + replica.host, + replica.port as u16, + replica.database, + )); } let primary = Store::MySQL(Arc::new(MysqlStore { conn_pool: Pool::new(opts), })); - - Ok(primary) + // ST-1: no replicas, no change + if replicas.is_empty() { + return Ok(primary); + } + Ok(Store::Replicated( + crate::backend::scaleout::replica::ReplicatedStore::new( + primary, + primary_location, + replicas, + crate::backend::scaleout::replica::ReplicaKind::MySql, + ), + )) } pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> { diff --git a/crates/store/src/backend/mysql/mod.rs b/crates/store/src/backend/mysql/mod.rs index 9cf087b..c98838f 100644 --- a/crates/store/src/backend/mysql/mod.rs +++ b/crates/store/src/backend/mysql/mod.rs @@ -26,7 +26,7 @@ pub struct MysqlStore { } #[inline(always)] -fn into_error(err: impl Display) -> trc::Error { +pub(crate) fn into_error(err: impl Display) -> trc::Error { trc::StoreEvent::MysqlError.reason(err) } diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index c2bbdff..348f5e4 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -27,6 +27,8 @@ use utils::tls::rustls_client_config; impl PostgresStore { pub async fn open(config: structs::PostgreSqlStore) -> Result { + // inbuxa: ST-15: where the primary is, to tell a replica from it + let primary_location = (config.host.clone(), config.port as u16, config.database.clone()); let mut cfg = Config::new(); cfg.dbname = config.database.into(); cfg.host = config.host.into(); @@ -57,16 +59,35 @@ impl PostgresStore { .map_err(|e| format!("Failed to create connection pool: {e}"))?; let ts_configs = discover_ts_configs(&primary_pool).await; - // inbuxa: ST-2: replicas aren't used yet (the scale-out decision), so - // each one is reported rather than silently ignored + // inbuxa: ST-5 to ST-15: each replica inherits the primary's settings + // except where it is and how to sign in + let mut replicas = vec![]; for replica in config.read_replicas { - trc::event!( - Store(trc::StoreEvent::PostgresqlError), - Details = format!( - "Read replica {}:{} {} isn't used yet: every operation goes to the primary", - replica.host, replica.port, replica.database - ), - ); + let mut cfg = cfg.clone(); + cfg.dbname = replica.database.clone().into(); + cfg.host = replica.host.clone().into(); + cfg.user = replica.auth_username; + cfg.password = replica.auth_secret.secret().await?.map(|v| v.into_owned()); + cfg.port = (replica.port as u16).into(); + cfg.options = replica.options; + let pool = if config.use_tls { + cfg.create_pool( + Some(Runtime::Tokio1), + MakeRustlsConnect::new(rustls_client_config(config.allow_invalid_certs)?), + ) + } else { + cfg.create_pool(Some(Runtime::Tokio1), NoTls) + } + .map_err(|e| format!("Failed to create connection pool: {e}"))?; + replicas.push(crate::backend::scaleout::replica::Replica::new( + Store::PostgreSQL(Arc::new(PostgresStore { + conn_pool: pool, + ts_configs: ts_configs.clone(), + })), + replica.host, + replica.port as u16, + replica.database, + )); } let primary = Store::PostgreSQL(Arc::new(PostgresStore { @@ -74,8 +95,18 @@ impl PostgresStore { ts_configs, })); - - Ok(primary) + // ST-1: no replicas, no change + if replicas.is_empty() { + return Ok(primary); + } + Ok(Store::Replicated( + crate::backend::scaleout::replica::ReplicatedStore::new( + primary, + primary_location, + replicas, + crate::backend::scaleout::replica::ReplicaKind::PostgreSql, + ), + )) } pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> { diff --git a/crates/store/src/backend/postgres/mod.rs b/crates/store/src/backend/postgres/mod.rs index 39d5040..299115f 100644 --- a/crates/store/src/backend/postgres/mod.rs +++ b/crates/store/src/backend/postgres/mod.rs @@ -29,7 +29,7 @@ pub struct PostgresStore { } #[inline(always)] -fn into_error(err: tokio_postgres::error::Error) -> trc::Error { +pub(crate) fn into_error(err: tokio_postgres::error::Error) -> trc::Error { let mut local_err = trc::StoreEvent::PostgresqlError.reason(error_chain(&err)); if let Some(db_err) = err.as_db_error() { local_err = local_err.code(db_err.code().code().to_string()); @@ -71,7 +71,7 @@ pub(crate) fn is_timeout_error(err: &tokio_postgres::Error) -> bool { } #[inline(always)] -fn into_pool_error(err: deadpool_postgres::PoolError) -> trc::Error { +pub(crate) fn into_pool_error(err: deadpool_postgres::PoolError) -> trc::Error { match err { deadpool_postgres::PoolError::Backend(err) => into_error(err), err => trc::StoreEvent::PostgresqlError.reason(error_chain(&err)), diff --git a/crates/store/src/backend/scaleout/mod.rs b/crates/store/src/backend/scaleout/mod.rs index d4cecff..7d2bd7e 100644 --- a/crates/store/src/backend/scaleout/mod.rs +++ b/crates/store/src/backend/scaleout/mod.rs @@ -12,6 +12,27 @@ pub mod blob; pub mod layout; pub mod memory; +pub mod replica; +pub mod replica_health; + +/// Calls a PostgreSQL or MySQL backend directly. Replicated stores use it +/// instead of going back through `Store`, whose futures would then contain +/// themselves. +#[macro_export] +#[doc(hidden)] +macro_rules! sql_backend { + ($store:expr, $backend:ident => $call:expr) => { + match $store { + #[cfg(feature = "postgres")] + $crate::Store::PostgreSQL($backend) => $call, + #[cfg(feature = "mysql")] + $crate::Store::MySQL($backend) => $call, + _ => Err(trc::StoreEvent::NotConfigured + .into_err() + .details("A replicated store's member isn't PostgreSQL or MySQL")), + } + }; +} pub use blob::ShardedBlobStore; pub use memory::ShardedInMemoryStore; diff --git a/crates/store/src/backend/scaleout/replica.rs b/crates/store/src/backend/scaleout/replica.rs new file mode 100644 index 0000000..5cec3a6 --- /dev/null +++ b/crates/store/src/backend/scaleout/replica.rs @@ -0,0 +1,369 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! SQL read replicas (ST-5 to ST-15). The primary answers everything unless +//! a read is made inside a [`replica_read`] scope, opened by the call sites +//! ST-6 names, for account data only. Inside one, the first read picks a +//! replica that is healthy, within the lag limit, and has caught up with +//! every change this node knows of for the scope's accounts (ST-7); +//! otherwise the scope stays on the primary. A miss or an error on a +//! replica is answered from the primary (ST-8, ST-12). + +use crate::{ + InMemoryStore, SUBSPACE_ACL, SUBSPACE_BLOB_LINK, SUBSPACE_BLOBS, SUBSPACE_COUNTER, + SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_PROPERTY, SUBSPACE_SEARCH_INDEX, Store, ValueKey, + write::{AssignedId, AssignedIds, ValueClass}, +}; +use ahash::AHashMap; +use std::{ + future::Future, + sync::{ + Arc, Mutex, OnceLock, + atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering}, + }, +}; + +/// A replica more than this far behind gets no reads (ST-11). +pub const LAG_LIMIT_MS: u64 = 5_000; +/// ... and gets them again under this (ST-11). +pub const LAG_READMIT_MS: u64 = 2_500; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum ReplicaState { + /// Not yet checked at startup (ST-15): no reads. + Unvalidated = 0, + Up = 1, + /// Failed a read or a probe (ST-12): no reads until a probe succeeds. + Down = 2, + /// Over the lag limit (ST-11). + Lagging = 3, + /// Failed a startup check (ST-15): never used. + Excluded = 4, + /// Its lag can't be measured (ST-11): never used. + Unmeasurable = 5, +} + +impl ReplicaState { + fn from_u8(value: u8) -> Self { + match value { + 1 => ReplicaState::Up, + 2 => ReplicaState::Down, + 3 => ReplicaState::Lagging, + 4 => ReplicaState::Excluded, + 5 => ReplicaState::Unmeasurable, + _ => ReplicaState::Unvalidated, + } + } +} + +pub struct Replica { + pub store: Store, + /// `host:port database`, for logs. + pub label: String, + /// The connection settings, for ST-15's "is it the primary" check. + pub location: (String, u16, String), + state: AtomicU8, + pub lag_ms: AtomicU64, + /// Reads served, for observability and tests. + pub reads: AtomicU64, + /// Startup marker checks that timed out (ST-15). + pub marker_misses: AtomicU8, +} + +impl Replica { + pub fn new(store: Store, host: String, port: u16, database: String) -> Self { + Replica { + store, + label: format!("{host}:{port} {database}"), + location: (host, port, database), + state: AtomicU8::new(ReplicaState::Unvalidated as u8), + lag_ms: AtomicU64::new(u64::MAX), + reads: AtomicU64::new(0), + marker_misses: AtomicU8::new(0), + } + } + + pub fn state(&self) -> ReplicaState { + ReplicaState::from_u8(self.state.load(Ordering::Relaxed)) + } + + pub fn set_state(&self, state: ReplicaState) -> ReplicaState { + ReplicaState::from_u8(self.state.swap(state as u8, Ordering::Relaxed)) + } + + fn usable(&self) -> bool { + self.state() == ReplicaState::Up + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicaKind { + PostgreSql, + MySql, +} + +pub struct ReplicatedStore { + pub primary: Store, + /// The primary's connection settings, for ST-15. + pub primary_location: (String, u16, String), + pub replicas: Vec, + pub kind: ReplicaKind, + next: AtomicUsize, + /// The highest change id this node has written or heard of, per account + /// (ST-7). + marks: Mutex>, + /// Shared marks, when more than one node runs (ST-7, step 2). + shared: OnceLock, +} + +/// Account data a replica may serve. Everything else (the registry, +/// in-memory values, the task queue, telemetry, the fork's own records) +/// is always read from the primary (ST-5). +pub fn is_replica_subspace(subspace: u8) -> bool { + matches!( + subspace, + SUBSPACE_PROPERTY + | SUBSPACE_INDEXES + | SUBSPACE_LOGS + | SUBSPACE_COUNTER + | SUBSPACE_ACL + | SUBSPACE_BLOB_LINK + | SUBSPACE_BLOBS + | SUBSPACE_SEARCH_INDEX + ) +} + +/// A read scope: the accounts whose data it reads, each with any change id +/// the client presented (ST-7, step 4), and the replica it settled on. +pub struct ReadScope { + accounts: Vec<(u32, u64)>, + choice: tokio::sync::OnceCell>, + /// Set by any write made inside the scope: from then on it reads from + /// the primary (ST-6: a read in a request that writes). + wrote: std::sync::atomic::AtomicBool, +} + +/// A write happened in the current task: a read scope, if any, stops using +/// replicas. +pub fn note_scope_write() { + let _ = READ_SCOPE.try_with(|scope| scope.wrote.store(true, Ordering::Relaxed)); +} + +tokio::task_local! { + static READ_SCOPE: Arc; +} + +/// Runs `fut` with replica-eligible reads for `accounts`: `(account, the +/// change id the client presented, or 0)` (ST-6). +pub async fn replica_read( + accounts: impl IntoIterator, + fut: F, +) -> F::Output { + READ_SCOPE + .scope( + Arc::new(ReadScope { + accounts: accounts.into_iter().collect(), + choice: tokio::sync::OnceCell::new(), + wrote: std::sync::atomic::AtomicBool::new(false), + }), + fut, + ) + .await +} + +/// Carries the current read scope, if any, into a future about to be +/// spawned as its own task (ST-6). +pub fn carry(fut: F) -> impl Future { + let scope = READ_SCOPE.try_with(|scope| scope.clone()).ok(); + async move { + match scope { + Some(scope) => READ_SCOPE.scope(scope, fut).await, + None => fut.await, + } + } +} + +#[allow(dead_code)] // used with a PostgreSQL or MySQL backend +fn change_id_key(account_id: u32) -> ValueKey { + ValueKey { + account_id, + collection: 0, + document_id: 0, + class: ValueClass::ChangeId, + } +} + +impl ReplicatedStore { + pub fn new( + primary: Store, + primary_location: (String, u16, String), + replicas: Vec, + kind: ReplicaKind, + ) -> Arc { + let store = Arc::new(ReplicatedStore { + primary, + primary_location, + replicas, + kind, + next: AtomicUsize::new(0), + marks: Mutex::new(AHashMap::new()), + shared: OnceLock::new(), + }); + super::replica_health::spawn(Arc::downgrade(&store)); + store + } + + /// Shares high-water marks through the in-memory store, for a cluster + /// (ST-7, step 2). + pub fn share_marks(&self, memory: InMemoryStore) { + let _ = self.shared.set(memory); + } + + #[allow(dead_code)] // used with the redis feature + fn shared_key(account_id: u32) -> Vec { + let mut key = b"_rm".to_vec(); + key.extend_from_slice(&account_id.to_be_bytes()); + key + } + + /// Records a change this node made or heard of (ST-7, step 1). + pub fn note_change(&self, account_id: u32, change_id: u64) { + let mut marks = self.marks.lock().unwrap(); + let mark = marks.entry(account_id).or_default(); + if change_id > *mark { + *mark = change_id; + } + } + + /// Records the change ids a write produced, locally and, in a cluster, + /// in the shared store before the write returns (ST-7, steps 1 and 2). + pub async fn note_write(&self, ids: &AssignedIds) { + let mut changes = Vec::new(); + for id in &ids.ids { + if let AssignedId::ChangeId(change) = id { + self.note_change(change.account_id, change.change_id); + changes.push((change.account_id, change.change_id)); + } + } + #[cfg(feature = "redis")] + if let Some(InMemoryStore::Redis(redis)) = self.shared.get() { + for (account_id, change_id) in changes { + let _ = redis + .key_set( + &Self::shared_key(account_id), + change_id.to_string().as_bytes(), + Some(2 * LAG_LIMIT_MS / 1000), + ) + .await; + } + } + #[cfg(not(feature = "redis"))] + let _ = changes; + } + + async fn mark(&self, account_id: u32) -> u64 { + let local = self + .marks + .lock() + .unwrap() + .get(&account_id) + .copied() + .unwrap_or_default(); + #[cfg(feature = "redis")] + let shared = match self.shared.get() { + Some(InMemoryStore::Redis(redis)) => redis + .key_get::(&Self::shared_key(account_id)) + .await + .ok() + .flatten() + .and_then(|mark| mark.parse::().ok()) + .unwrap_or_default(), + _ => 0, + }; + #[cfg(not(feature = "redis"))] + let shared = 0; + local.max(shared) + } + + /// The replica the current scope reads from, if any (ST-7, ST-13). + pub async fn read_target(&self, subspace: u8) -> Option { + if !is_replica_subspace(subspace) { + return None; + } + let scope = READ_SCOPE.try_with(|scope| scope.clone()).ok()?; + if scope.wrote.load(Ordering::Relaxed) { + return None; + } + *scope.choice.get_or_init(|| self.choose(&scope)).await + } + + async fn choose(&self, scope: &ReadScope) -> Option { + let count = self.replicas.len(); + let start = self.next.fetch_add(1, Ordering::Relaxed); + 'replicas: for offset in 0..count { + let index = (start + offset) % count; + let replica = &self.replicas[index]; + if !replica.usable() { + continue; + } + for (account_id, presented) in &scope.accounts { + let mark = self.mark(*account_id).await.max(*presented); + if mark == 0 { + continue; + } + let seen: trc::Result = crate::sql_backend!( + &replica.store, + db => db.get_counter(change_id_key(*account_id)).await + ); + match seen { + Ok(seen) if seen as u64 >= mark => {} + // Behind for this account: the primary answers (ST-7) + Ok(_) => return None, + Err(err) => { + self.failed(index, err); + continue 'replicas; + } + } + } + return Some(index); + } + None + } + + /// A replica read failed: it's down until a probe succeeds (ST-12). + pub fn failed(&self, index: usize, err: trc::Error) { + let replica = &self.replicas[index]; + if replica.set_state(ReplicaState::Down) != ReplicaState::Down { + super::replica_health::report( + self.kind, + format!("Read replica {} is down: {err}", replica.label), + ); + } + } + + pub fn served(&self, index: usize) { + self.replicas[index].reads.fetch_add(1, Ordering::Relaxed); + } +} + +impl Store { + /// Records a change heard from another node (ST-7, step 1). Nothing to + /// do without replicas. + pub fn note_change(&self, account_id: u32, change_id: u64) { + if let Store::Replicated(store) = self { + store.note_change(account_id, change_id); + } + } + + /// Shares high-water marks through `memory`, when more than one node + /// runs (ST-7, step 2). + pub fn share_marks(&self, memory: &InMemoryStore) { + if let Store::Replicated(store) = self { + store.share_marks(memory.clone()); + } + } +} diff --git a/crates/store/src/backend/scaleout/replica_health.rs b/crates/store/src/backend/scaleout/replica_health.rs new file mode 100644 index 0000000..0088890 --- /dev/null +++ b/crates/store/src/backend/scaleout/replica_health.rs @@ -0,0 +1,429 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Each node watches its replicas (ST-10 to ST-15): startup checks, then a +//! lag sample every second, and a probe every ten seconds while one is +//! down. Changes are logged with the backend's existing error event +//! (ST-30). + +use super::replica::{ + LAG_LIMIT_MS, LAG_READMIT_MS, Replica, ReplicaKind, ReplicaState, ReplicatedStore, +}; +use crate::{ + SUBSPACE_INBUXA, SerializeInfallible, Store, ValueKey, + write::{AnyClass, BatchBuilder, ValueClass}, +}; +use std::{ + collections::VecDeque, + sync::Weak, + time::{Duration, Instant}, +}; + +const SAMPLE_EVERY: Duration = Duration::from_secs(1); +const PROBE_EVERY: Duration = Duration::from_secs(10); +/// Startup marker checks, one a probe, before a replica is left out. +const MARKER_TRIES: u8 = 6; +/// Primary positions kept, one a second. +const SAMPLES_KEPT: usize = 120; + +pub fn report(kind: ReplicaKind, message: String) { + match kind { + ReplicaKind::PostgreSql => { + trc::event!(Store(trc::StoreEvent::PostgresqlError), Details = message) + } + ReplicaKind::MySql => trc::event!(Store(trc::StoreEvent::MysqlError), Details = message), + } +} + +/// Where the primary is replicated up to, in whatever unit the backend +/// counts (a PostgreSQL LSN, or a MySQL GTID set). +#[allow(dead_code)] // built with a PostgreSQL or MySQL backend +#[derive(Debug, Clone, PartialEq, Eq)] +enum Position { + Lsn(u64), + Gtid(String), +} + +pub fn spawn(store: Weak) { + tokio::spawn(async move { + let mut samples: VecDeque<(Instant, Position)> = VecDeque::new(); + let mut last_probe: Vec> = Vec::new(); + loop { + { + let Some(store) = store.upgrade() else { + return; + }; + if last_probe.len() != store.replicas.len() { + last_probe = vec![None; store.replicas.len()]; + } + tick(&store, &mut samples, &mut last_probe).await; + } + tokio::time::sleep(SAMPLE_EVERY).await; + } + }); +} + +async fn tick( + store: &ReplicatedStore, + samples: &mut VecDeque<(Instant, Position)>, + last_probe: &mut [Option], +) { + // The primary's position now (ST-10) + match primary_position(store).await { + Ok(Some(position)) => { + if samples.back().is_none_or(|(_, last)| *last != position) { + samples.push_back((Instant::now(), position)); + } else if let Some(last) = samples.back_mut() { + // Unchanged: an idle primary; nothing new to wait for + last.0 = last.0.min(Instant::now()); + } + while samples.len() > SAMPLES_KEPT { + samples.pop_front(); + } + } + Ok(None) => {} + Err(err) => report( + store.kind, + format!("Failed to read the primary's replication position: {err}"), + ), + } + + for (index, replica) in store.replicas.iter().enumerate() { + let state = replica.state(); + match state { + ReplicaState::Excluded | ReplicaState::Unmeasurable => continue, + ReplicaState::Unvalidated | ReplicaState::Down => { + if last_probe[index].is_some_and(|at| at.elapsed() < PROBE_EVERY) { + continue; + } + last_probe[index] = Some(Instant::now()); + if state == ReplicaState::Unvalidated { + validate(store, index, replica).await; + continue; + } + } + ReplicaState::Up | ReplicaState::Lagging => {} + } + + match replica_lag(store, replica, samples).await { + Ok(Some(lag)) => { + replica + .lag_ms + .store(lag, std::sync::atomic::Ordering::Relaxed); + let next = match state { + ReplicaState::Up if lag > LAG_LIMIT_MS => ReplicaState::Lagging, + ReplicaState::Lagging if lag < LAG_READMIT_MS => ReplicaState::Up, + ReplicaState::Down if lag < LAG_READMIT_MS => ReplicaState::Up, + ReplicaState::Down => ReplicaState::Lagging, + other => other, + }; + if next != state { + replica.set_state(next); + report( + store.kind, + format!( + "Read replica {} is {} (lag {lag} ms)", + replica.label, + match next { + ReplicaState::Up => "up", + _ => "over the lag limit", + } + ), + ); + } + } + Ok(None) => { + if replica.set_state(ReplicaState::Unmeasurable) != ReplicaState::Unmeasurable { + report( + store.kind, + format!( + "Read replica {} gets no reads: its lag can't be measured \ + (the REPLICATION CLIENT privilege may be missing)", + replica.label + ), + ); + } + } + Err(err) if state != ReplicaState::Down => store.failed(index, err), + Err(_) => {} + } + } +} + +/// The startup checks (ST-15). A replica that can't be reached yet stays +/// unchecked, and is tried again at the next probe. +async fn validate(store: &ReplicatedStore, index: usize, replica: &Replica) { + let exclude = |why: String| { + replica.set_state(ReplicaState::Excluded); + report( + store.kind, + format!("Read replica {} is left out: {why}", replica.label), + ); + }; + if replica.location == store.primary_location { + return exclude("it's the primary itself".to_string()); + } + match read_only(store.kind, &replica.store).await { + Ok(Ok(())) => {} + Ok(Err(why)) => return exclude(why), + Err(err) => { + report( + store.kind, + format!("Read replica {} can't be checked yet: {err}", replica.label), + ); + return; + } + } + + // It must be a copy of this primary: a marker written there shows up + let marker = rand::random::() | 1; + let class = ValueClass::Any(AnyClass { + subspace: SUBSPACE_INBUXA, + key: [b'S', b'r', index as u8].to_vec(), + }); + let mut batch = BatchBuilder::new(); + batch.set(class.clone(), marker.serialize()); + if let Err(err) = store.primary.write(batch.build_all()).await { + report( + store.kind, + format!("Failed to write the replica check marker: {err}"), + ); + return; + } + let deadline = Instant::now() + Duration::from_millis(LAG_LIMIT_MS); + loop { + match replica + .store + .get_value::(ValueKey::from(class.clone())) + .await + { + Ok(Some(seen)) if seen == marker => { + replica.set_state(ReplicaState::Up); + report(store.kind, format!("Read replica {} is up", replica.label)); + return; + } + Ok(_) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(100)).await; + } + // A replica still catching up (after a burst of writes at + // startup, say) gets a few more tries before it's left out + Ok(_) => { + let misses = replica + .marker_misses + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + + 1; + return if misses >= MARKER_TRIES { + exclude("it isn't a copy of this primary".to_string()) + } else { + report( + store.kind, + format!( + "Read replica {} hasn't shown the check marker yet; trying again", + replica.label + ), + ) + }; + } + Err(err) => { + report( + store.kind, + format!("Read replica {} can't be checked yet: {err}", replica.label), + ); + return; + } + } + } +} + +#[allow(unused_variables)] +async fn read_only(kind: ReplicaKind, replica: &Store) -> trc::Result> { + match (kind, replica) { + #[cfg(feature = "postgres")] + (ReplicaKind::PostgreSql, Store::PostgreSQL(pg)) => { + let conn = pg + .conn_pool + .get() + .await + .map_err(crate::backend::postgres::into_pool_error)?; + let row = conn + .query_one("SELECT pg_is_in_recovery()", &[]) + .await + .map_err(crate::backend::postgres::into_error)?; + Ok(if row.get::<_, bool>(0) { + Ok(()) + } else { + Err("it isn't read-only (pg_is_in_recovery() is false)".to_string()) + }) + } + #[cfg(feature = "mysql")] + (ReplicaKind::MySql, Store::MySQL(my)) => { + use mysql_async::prelude::Queryable; + let mut conn = my + .conn_pool + .get_conn() + .await + .map_err(crate::backend::mysql::into_error)?; + let (read_only, super_read_only): (i64, i64) = conn + .query_first("SELECT @@global.read_only, @@global.super_read_only") + .await + .map_err(crate::backend::mysql::into_error)? + .unwrap_or((0, 0)); + if read_only == 0 && super_read_only == 0 { + return Ok(Err( + "it isn't read-only (neither read_only nor super_read_only is on)".to_string(), + )); + } + let parallel: Option<(i64, i64)> = conn + .query_first( + "SELECT @@global.replica_parallel_workers, @@global.replica_preserve_commit_order", + ) + .await + .ok() + .flatten(); + if let Some((workers, preserve)) = parallel + && workers > 0 + && preserve == 0 + { + return Ok(Err( + "it applies in parallel without preserving commit order".to_string(), + )); + } + Ok(Ok(())) + } + _ => Ok(Err("the backend isn't compiled in".to_string())), + } +} + +#[cfg(feature = "postgres")] +fn parse_lsn(text: &str) -> Option { + let (high, low) = text.split_once('/')?; + Some((u64::from_str_radix(high, 16).ok()? << 32) | u64::from_str_radix(low, 16).ok()?) +} + +#[allow(unused_variables)] +async fn primary_position(store: &ReplicatedStore) -> trc::Result> { + match &store.primary { + #[cfg(feature = "postgres")] + Store::PostgreSQL(pg) => { + let conn = pg + .conn_pool + .get() + .await + .map_err(crate::backend::postgres::into_pool_error)?; + let row = conn + .query_one("SELECT pg_current_wal_lsn()::text", &[]) + .await + .map_err(crate::backend::postgres::into_error)?; + Ok(parse_lsn(row.get::<_, &str>(0)).map(Position::Lsn)) + } + #[cfg(feature = "mysql")] + Store::MySQL(my) => { + use mysql_async::prelude::Queryable; + let mut conn = my + .conn_pool + .get_conn() + .await + .map_err(crate::backend::mysql::into_error)?; + let mode: Option = conn + .query_first("SELECT @@global.gtid_mode") + .await + .map_err(crate::backend::mysql::into_error)?; + if mode.as_deref() != Some("ON") { + return Ok(None); + } + let executed: Option = conn + .query_first("SELECT @@global.gtid_executed") + .await + .map_err(crate::backend::mysql::into_error)?; + Ok(executed.map(Position::Gtid)) + } + _ => Ok(None), + } +} + +/// The replica's lag in milliseconds (ST-10): the age of the oldest +/// primary position it hasn't applied yet. `None` when it can't be +/// measured. +#[allow(unused_variables)] +async fn replica_lag( + store: &ReplicatedStore, + replica: &Replica, + samples: &VecDeque<(Instant, Position)>, +) -> trc::Result> { + let age = |at: Instant| at.elapsed().as_millis() as u64; + match &replica.store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(pg) => { + let conn = pg + .conn_pool + .get() + .await + .map_err(crate::backend::postgres::into_pool_error)?; + let row = conn + .query_one("SELECT pg_last_wal_replay_lsn()::text", &[]) + .await + .map_err(crate::backend::postgres::into_error)?; + let Some(replayed) = row.get::<_, Option<&str>>(0).and_then(parse_lsn) else { + return Err(trc::StoreEvent::PostgresqlError + .into_err() + .details("The replica isn't replaying")); + }; + Ok(Some( + samples + .iter() + .find(|(_, position)| matches!(position, Position::Lsn(lsn) if *lsn > replayed)) + .map(|(at, _)| age(*at)) + .unwrap_or(0), + )) + } + #[cfg(feature = "mysql")] + Store::MySQL(my) => { + use mysql_async::prelude::Queryable; + let mut conn = my + .conn_pool + .get_conn() + .await + .map_err(crate::backend::mysql::into_error)?; + if samples.iter().any(|(_, p)| matches!(p, Position::Gtid(_))) { + // With GTIDs: the oldest primary set the replica lacks + for (at, position) in samples { + let Position::Gtid(set) = position else { + continue; + }; + let applied: Option = conn + .exec_first( + "SELECT GTID_SUBSET(?, @@global.gtid_executed)", + (set.as_str(),), + ) + .await + .map_err(crate::backend::mysql::into_error)?; + if applied != Some(1) { + return Ok(Some(age(*at))); + } + } + Ok(Some(0)) + } else { + // Without: Seconds_Behind_Source, which needs REPLICATION CLIENT + let row: Option = + match conn.query_first("SHOW REPLICA STATUS").await { + Ok(row) => row, + Err(_) => 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 + .into_err() + .details("Replication is stopped")), + _ => Ok(None), + } + } + } + _ => Ok(None), + } +} diff --git a/crates/store/src/dispatch/blob.rs b/crates/store/src/dispatch/blob.rs index d850ec4..40ef517 100644 --- a/crates/store/src/dispatch/blob.rs +++ b/crates/store/src/dispatch/blob.rs @@ -145,6 +145,33 @@ impl BlobStore { #[cfg(feature = "rocks")] Store::RocksDb(store) => store.get_blob(key, 0..usize::MAX).await, Store::Ephemeral(store) => store.get_blob(key, 0..usize::MAX).await, + // inbuxa: ST-9: a replica, then the primary for what it lacks + Store::Replicated(store) => match store.read_target(crate::SUBSPACE_BLOBS).await { + Some(index) => match crate::sql_backend!( + &store.replicas[index].store, + db => db.get_blob(key, 0..usize::MAX).await + ) { + Ok(Some(data)) => { + store.served(index); + Ok(Some(data)) + } + Ok(None) => crate::sql_backend!( + &store.primary, + db => db.get_blob(key, 0..usize::MAX).await + ), + Err(err) => { + store.failed(index, err); + crate::sql_backend!( + &store.primary, + db => db.get_blob(key, 0..usize::MAX).await + ) + } + }, + None => crate::sql_backend!( + &store.primary, + db => db.get_blob(key, 0..usize::MAX).await + ), + }, Store::None => Err(trc::StoreEvent::NotConfigured.into()), }, BlobStore::Fs(store) => store.get_blob(key, 0..usize::MAX).await, @@ -171,6 +198,9 @@ impl BlobStore { #[cfg(feature = "rocks")] Store::RocksDb(store) => store.put_blob(key, data).await, Store::Ephemeral(store) => store.put_blob(key, data).await, + Store::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.put_blob(key, data).await) + } Store::None => Err(trc::StoreEvent::NotConfigured.into()), }, BlobStore::Fs(store) => store.put_blob(key, data).await, @@ -197,6 +227,9 @@ impl BlobStore { #[cfg(feature = "rocks")] Store::RocksDb(store) => store.delete_blob(key).await, Store::Ephemeral(store) => store.delete_blob(key).await, + Store::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.delete_blob(key).await) + } Store::None => Err(trc::StoreEvent::NotConfigured.into()), }, BlobStore::Fs(store) => store.delete_blob(key).await, diff --git a/crates/store/src/dispatch/mod.rs b/crates/store/src/dispatch/mod.rs index 205affd..1d287e4 100644 --- a/crates/store/src/dispatch/mod.rs +++ b/crates/store/src/dispatch/mod.rs @@ -26,6 +26,8 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(_) => "rocksdb", Self::Ephemeral(_) => "ephemeral", + // inbuxa: ST-3: as its primary + Self::Replicated(store) => store.primary.id(), Self::None => "none", } } diff --git a/crates/store/src/dispatch/search.rs b/crates/store/src/dispatch/search.rs index 5e01844..5d9d3b3 100644 --- a/crates/store/src/dispatch/search.rs +++ b/crates/store/src/dispatch/search.rs @@ -201,6 +201,25 @@ impl SearchStore { filters: &[SearchFilter], sort: &[SearchComparator], ) -> trc::Result> { + // inbuxa: ST-6, ST-12: a replica answers search queries in a read scope + if let SearchStore::Store(Store::Replicated(store)) = self { + return match store.read_target(crate::SUBSPACE_SEARCH_INDEX).await { + Some(replica) => match crate::sql_backend!( + &store.replicas[replica].store, + db => db.query(index, filters, sort).await + ) { + Ok(ids) => { + store.served(replica); + Ok(ids) + } + Err(err) => { + store.failed(replica, err); + crate::sql_backend!(&store.primary, db => db.query(index, filters, sort).await) + } + }, + None => crate::sql_backend!(&store.primary, db => db.query(index, filters, sort).await), + }; + } match self { SearchStore::Store(store) => match store { #[cfg(feature = "postgres")] @@ -215,6 +234,13 @@ impl SearchStore { } pub async fn query_global(&self, query: SearchQuery) -> trc::Result> { + // inbuxa: ST-5: global queries are maintenance, on the primary + if let SearchStore::Store(Store::Replicated(store)) = self { + return crate::sql_backend!( + &store.primary, + db => db.query(query.index, &query.filters, &query.comparators).await + ); + } match self { SearchStore::Store(store) => match store { #[cfg(feature = "postgres")] @@ -245,6 +271,9 @@ impl SearchStore { } pub async fn index(&self, documents: Vec) -> trc::Result<()> { + if let SearchStore::Store(Store::Replicated(store)) = self { + return crate::sql_backend!(&store.primary, db => db.index(documents).await); + } match self { SearchStore::Store(store) => match store { #[cfg(feature = "postgres")] @@ -259,6 +288,9 @@ impl SearchStore { } pub async fn unindex(&self, query: SearchQuery) -> trc::Result { + if let SearchStore::Store(Store::Replicated(store)) = self { + return crate::sql_backend!(&store.primary, db => db.unindex(query).await); + } match self { SearchStore::Store(store) => match store { #[cfg(feature = "postgres")] @@ -279,6 +311,8 @@ impl SearchStore { Store::PostgreSQL(_) => None, #[cfg(feature = "mysql")] Store::MySQL(_) => None, + // inbuxa: ST-3: as its primary + Store::Replicated(replicated) if replicated.primary.is_pg_or_mysql() => None, store => Some(store), }, _ => None, @@ -289,6 +323,10 @@ impl SearchStore { match self { #[cfg(feature = "mysql")] SearchStore::Store(Store::MySQL(_)) => true, + #[cfg(feature = "mysql")] + SearchStore::Store(Store::Replicated(store)) => { + matches!(store.primary, Store::MySQL(_)) + } _ => false, } } @@ -297,6 +335,10 @@ impl SearchStore { match self { #[cfg(feature = "postgres")] SearchStore::Store(Store::PostgreSQL(_)) => true, + #[cfg(feature = "postgres")] + SearchStore::Store(Store::Replicated(store)) => { + matches!(store.primary, Store::PostgreSQL(_)) + } _ => false, } } @@ -316,6 +358,9 @@ impl SearchStore { Store::PostgreSQL(store) => store.create_search_tables().await, #[cfg(feature = "mysql")] Store::MySQL(store) => store.create_search_tables().await, + Store::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.create_search_tables().await) + } _ => Ok(()), }, SearchStore::ElasticSearch(store) => store.create_indexes().await, diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index af693bf..a863f75 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -35,6 +35,23 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.get_value(key).await, Self::Ephemeral(store) => store.get_value(key).await, + // inbuxa: ST-6 to ST-8, ST-12 + Self::Replicated(store) => match store.read_target(key.subspace()).await { + Some(index) => { + match crate::sql_backend!(&store.replicas[index].store, db => db.get_value::(key.clone()).await) { + Ok(Some(value)) => { + store.served(index); + Ok(Some(value)) + } + Ok(None) => crate::sql_backend!(&store.primary, db => db.get_value(key).await), + Err(err) => { + store.failed(index, err); + crate::sql_backend!(&store.primary, db => db.get_value(key).await) + } + } + } + None => crate::sql_backend!(&store.primary, db => db.get_value(key).await), + }, Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()) @@ -53,6 +70,21 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.key_exists(key).await, Self::Ephemeral(store) => store.key_exists(key).await, + // inbuxa: ST-6 to ST-8, ST-12 + Self::Replicated(store) => match store.read_target(key.subspace()).await { + Some(index) => match crate::sql_backend!(&store.replicas[index].store, db => db.key_exists(key.clone()).await) { + Ok(true) => { + store.served(index); + Ok(true) + } + Ok(false) => crate::sql_backend!(&store.primary, db => db.key_exists(key).await), + Err(err) => { + store.failed(index, err); + crate::sql_backend!(&store.primary, db => db.key_exists(key).await) + } + }, + None => crate::sql_backend!(&store.primary, db => db.key_exists(key).await), + }, Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()) @@ -76,6 +108,38 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.iterate(params, cb).await, Self::Ephemeral(store) => store.iterate(params, cb).await, + // inbuxa: ST-6, ST-12: a failed replica iteration is repeated on + // the primary when nothing was handed to the callback yet + #[allow(unused_mut, unused_variables)] + Self::Replicated(store) => { + let mut cb = cb; + match store.read_target(params.begin.subspace()).await { + Some(index) => { + let mut called = false; + let result = crate::sql_backend!(&store.replicas[index].store, db => db + .iterate(params.clone(), |key, value| { + called = true; + cb(key, value) + }) + .await); + match result { + Ok(()) => { + store.served(index); + Ok(()) + } + Err(err) if !called => { + store.failed(index, err); + crate::sql_backend!(&store.primary, db => db.iterate(params, cb).await) + } + Err(err) => { + store.failed(index, err.clone()); + Err(err) + } + } + } + None => crate::sql_backend!(&store.primary, db => db.iterate(params, cb).await), + } + } Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()); @@ -104,6 +168,25 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.get_counter(key).await, Self::Ephemeral(store) => store.get_counter(key).await, + // inbuxa: ST-6, ST-12 + Self::Replicated(store) => { + let key: ValueKey = key.into(); + match store.read_target(crate::Key::subspace(&key)).await { + Some(index) => { + match crate::sql_backend!(&store.replicas[index].store, db => db.get_counter(key.clone()).await) { + Ok(value) => { + store.served(index); + Ok(value) + } + Err(err) => { + store.failed(index, err); + crate::sql_backend!(&store.primary, db => db.get_counter(key).await) + } + } + } + None => crate::sql_backend!(&store.primary, db => db.get_counter(key).await), + } + } Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()) @@ -123,6 +206,10 @@ impl Store { Self::PostgreSQL(store) => store.sql_query(query, ¶ms).await, #[cfg(feature = "mysql")] Self::MySQL(store) => store.sql_query(query, ¶ms).await, + // inbuxa: ST-5: operator-written statements always go to the primary + Self::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.sql_query(query, ¶ms).await) + } _ => Err(trc::StoreEvent::NotSupported.into_err()), }; @@ -152,6 +239,15 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.write(batch).await, Self::Ephemeral(store) => store.write(batch).await, + // inbuxa: ST-5, ST-7: writes go to the primary, and record marks + Self::Replicated(store) => { + crate::backend::scaleout::replica::note_scope_write(); + let result = crate::sql_backend!(&store.primary, db => db.write(batch).await); + if let Ok(ids) = &result { + store.note_write(ids).await; + } + result + } Self::None => Err(trc::StoreEvent::NotConfigured.into()), }; @@ -197,6 +293,7 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.purge_store().await, Self::Ephemeral(store) => store.purge_store().await, + Self::Replicated(store) => crate::sql_backend!(&store.primary, db => db.purge_store().await), Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()) @@ -215,6 +312,9 @@ impl Store { #[cfg(feature = "rocks")] Self::RocksDb(store) => store.delete_range(from, to).await, Self::Ephemeral(store) => store.delete_range(from, to).await, + Self::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.delete_range(from, to).await) + } Self::None => Err(trc::StoreEvent::NotConfigured.into()), } .caused_by(trc::location!()) @@ -360,6 +460,9 @@ impl Store { Self::PostgreSQL(store) => store.create_storage_tables().await, #[cfg(feature = "mysql")] Self::MySQL(store) => store.create_storage_tables().await, + Self::Replicated(store) => { + crate::sql_backend!(&store.primary, db => db.create_storage_tables().await) + } _ => Ok(()), } } diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index d30d690..90744c7 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -159,6 +159,8 @@ pub enum Store { #[cfg(feature = "rocks")] RocksDb(Arc), Ephemeral(Arc), + // inbuxa: ST-5 to ST-15: a PostgreSQL or MySQL primary with read replicas + Replicated(Arc), #[default] None, } @@ -654,8 +656,8 @@ impl Store { #[cfg(feature = "rocks")] (Store::RocksDb(a), Store::RocksDb(b)) => Arc::ptr_eq(a, b), (Store::Ephemeral(a), Store::Ephemeral(b)) => Arc::ptr_eq(a, b), - #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] - (Store::SQLReadReplica(a), Store::SQLReadReplica(b)) => Arc::ptr_eq(a, b), + // inbuxa: ST-3 + (Store::Replicated(a), Store::Replicated(b)) => Arc::ptr_eq(a, b), (Store::None, Store::None) => true, _ => false, } @@ -664,6 +666,8 @@ impl Store { #[inline(always)] pub fn is_sql(&self) -> bool { match self { + // inbuxa: ST-3: as its primary + Store::Replicated(store) => store.primary.is_sql(), #[cfg(feature = "sqlite")] Store::SQLite(_) => true, #[cfg(feature = "postgres")] @@ -677,6 +681,8 @@ impl Store { #[inline(always)] pub fn is_pg_or_mysql(&self) -> bool { match self { + // inbuxa: ST-3: as its primary + Store::Replicated(store) => store.primary.is_pg_or_mysql(), #[cfg(feature = "mysql")] Store::MySQL(_) => true, #[cfg(feature = "postgres")] @@ -715,6 +721,7 @@ impl std::fmt::Debug for Store { #[cfg(feature = "rocks")] Self::RocksDb(_) => f.debug_tuple("RocksDb").finish(), Self::Ephemeral(_) => f.debug_tuple("Ephemeral").finish(), + Self::Replicated(store) => f.debug_tuple("Replicated").field(&store.primary).finish(), Self::None => f.debug_tuple("None").finish(), } diff --git a/tests/src/store/mod.rs b/tests/src/store/mod.rs index 5b10b61..fdadbac 100644 --- a/tests/src/store/mod.rs +++ b/tests/src/store/mod.rs @@ -10,6 +10,8 @@ pub mod lookup; pub mod ops; pub mod query; pub mod registry; +#[cfg(feature = "postgres")] +pub mod replica; // inbuxa: read replicas pub mod scaleout; // inbuxa: scale-out storage #[cfg(any(feature = "postgres", feature = "mysql"))] pub mod sql_timeout; diff --git a/tests/src/store/replica.rs b/tests/src/store/replica.rs new file mode 100644 index 0000000..5a54358 --- /dev/null +++ b/tests/src/store/replica.rs @@ -0,0 +1,329 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Read-replica acceptance tests, from +//! `docs/spec/features/scale-out-storage.md` (tests 9, 10, 12, 13, 14 and +//! 15), against a PostgreSQL primary with a streaming hot standby in +//! containers. Built with `postgres`. + +use crate::utils::{ + containers::{ + PG_PRIMARY_CONTAINER, PG_REPLICA_CONTAINER, PG_REPLICA_PORT, docker, ensure_postgres, psql, + }, + server::TestServerBuilder, +}; +use jmap_client::email; +use registry::schema::structs::{ + Imap, PostgreSqlSettings, PostgreSqlStore, SecretKeyOptional, SecretKeyValue, +}; +use std::{ + sync::{Arc, atomic::Ordering}, + time::{Duration, Instant}, +}; +use store::{ + Store, + backend::scaleout::replica::{ReplicaState, ReplicatedStore}, +}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + +const SECRET: &str = "replica test user 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 reads(store: &ReplicatedStore) -> u64 { + store.replicas[0].reads.load(Ordering::Relaxed) +} + +fn message(subject: &str) -> Vec { + format!( + "From: sender@example.org\r\nTo: replica@example.org\r\nSubject: {subject}\r\n\r\nBody of {subject}, with the word zebrafish.\r\n" + ) + .into_bytes() +} + +async fn imap_fetch() -> String { + let stream = tokio::net::TcpStream::connect("127.0.0.1:9991") + .await + .unwrap(); + let (reader, mut writer) = tokio::io::split(stream); + let mut lines = BufReader::new(reader).lines(); + lines.next_line().await.unwrap(); + let mut transcript = String::new(); + for (tag, command) in [ + ("a", format!("LOGIN \"replica@example.org\" \"{SECRET}\"")), + ("b", "SELECT INBOX".to_string()), + ("c", "FETCH 1 BODY.PEEK[]".to_string()), + ("d", "LOGOUT".to_string()), + ] { + writer + .write_all(format!("{tag} {command}\r\n").as_bytes()) + .await + .unwrap(); + while let Ok(Ok(Some(line))) = + tokio::time::timeout(Duration::from_secs(10), lines.next_line()).await + { + transcript.push_str(&line); + transcript.push('\n'); + if line.starts_with(&format!("{tag} ")) { + break; + } + } + } + transcript +} + +/// `cargo test -p tests --features postgres replica_tests -- --ignored`, +/// with `STORE=PostgreSqlReplicated`. +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn replica_tests() { + assert_eq!( + std::env::var("STORE").as_deref(), + Ok("PostgreSqlReplicated"), + "run with STORE=PostgreSqlReplicated" + ); + let test = TestServerBuilder::new("replica_tests") + .await + .with_default_listeners() + .await + .with_object(Imap { + allow_plain_text_auth: true, + ..Default::default() + }) + .await + .build() + .await; + let replicated: Arc = match test.server.store() { + Store::Replicated(store) => store.clone(), + other => panic!("test 9: the data store isn't replicated: {other:?}"), + }; + // ST-15: checked and in use + wait_for( + &replicated, + ReplicaState::Up, + Duration::from_secs(60), + "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 are 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 = reads(&replicated); + assert!( + client + .email_get(&first, None::>) + .await + .unwrap() + .is_some() + ); + assert!( + reads(&replicated) > before, + "test 9: Email/get on the replica" + ); + let before = reads(&replicated); + let transcript = imap_fetch().await; + assert!( + transcript.contains("Subject: First"), + "test 9: {transcript}" + ); + assert!(reads(&replicated) > before, "test 9: FETCH on the replica"); + + // Test 15: full-text search through the replicated store (ST-3, ST-6) + tokio::time::sleep(Duration::from_secs(2)).await; + let found = client + .email_query( + email::query::Filter::text("zebrafish").into(), + None::>, + ) + .await + .unwrap() + .take_ids(); + assert!(found.contains(&first), "test 15: {found:?}"); + + // Test 10: with replay paused, a client still reads its own writes (ST-7, ST-8) + let state = client + .email_changes("n", None) + .await + .unwrap() + .new_state() + .to_string(); + psql(PG_REPLICA_CONTAINER, "SELECT pg_wal_replay_pause()"); + 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 10: a new message reads back" + ); + let changes = client.email_changes(state, None).await.unwrap(); + assert!( + changes.created().contains(&paused), + "test 10: Email/changes" + ); + let renamed = client + .mailbox_create("Before", None::, jmap_client::mailbox::Role::None) + .await + .unwrap() + .take_id(); + client.mailbox_rename(&renamed, "After").await.unwrap(); + assert_eq!( + client + .mailbox_get(&renamed, None::>) + .await + .unwrap() + .unwrap() + .name(), + Some("After"), + "test 10: a renamed mailbox" + ); + + // Test 12: over the lag limit, no reads; back once caught up (ST-10, ST-11) + let deadline = Instant::now() + Duration::from_secs(20); + while replicated.replicas[0].state() != ReplicaState::Lagging { + assert!(Instant::now() < deadline, "test 12: 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 = reads(&replicated); + assert!( + client + .email_get(&first, None::>) + .await + .unwrap() + .is_some() + ); + assert_eq!( + reads(&replicated), + before, + "test 12: no reads while lagging" + ); + psql(PG_REPLICA_CONTAINER, "SELECT pg_wal_replay_resume()"); + wait_for( + &replicated, + ReplicaState::Up, + Duration::from_secs(20), + "test 12", + ) + .await; + + // Test 13: the replica stopped; requests still succeed (ST-12) + docker("stop", PG_REPLICA_CONTAINER); + for _ in 0..3 { + assert!( + client + .email_get(&first, None::>) + .await + .unwrap() + .is_some(), + "test 13: served while the replica is down" + ); + } + wait_for( + &replicated, + ReplicaState::Down, + Duration::from_secs(15), + "test 13", + ) + .await; + docker("start", PG_REPLICA_CONTAINER); + wait_for( + &replicated, + ReplicaState::Up, + Duration::from_secs(60), + "test 13: back", + ) + .await; + + // Test 14: a replica that is writable and unrelated, or the primary + // itself, is left out (ST-15) + ensure_postgres().await; + let setting = |port: u16| PostgreSqlSettings { + host: "localhost".into(), + port: port as u64, + database: "stalwart".into(), + auth_username: "stalwart".to_string().into(), + auth_secret: SecretKeyOptional::Value(SecretKeyValue { + secret: "stalwart".into(), + }), + options: None, + }; + for (port, what) in [ + (5432, "an unrelated database"), + (5442, "the primary itself"), + ] { + let store = store::backend::postgres::PostgresStore::open(PostgreSqlStore { + host: "localhost".into(), + port: 5442, + auth_username: "stalwart".to_string().into(), + auth_secret: SecretKeyOptional::Value(SecretKeyValue { + secret: "stalwart".into(), + }), + database: "stalwart".into(), + read_replicas: registry::types::list::List::from_iter([setting(port)]), + ..Default::default() + }) + .await + .unwrap(); + let Store::Replicated(store) = store else { + panic!("test 14") + }; + wait_for( + &store, + ReplicaState::Excluded, + Duration::from_secs(20), + &format!("test 14: {what}"), + ) + .await; + } + let _ = (PG_PRIMARY_CONTAINER, PG_REPLICA_PORT); + + test.temp_dir.delete(); +} diff --git a/tests/src/utils/containers.rs b/tests/src/utils/containers.rs index 3ddce4d..a8ec8c9 100644 --- a/tests/src/utils/containers.rs +++ b/tests/src/utils/containers.rs @@ -32,6 +32,14 @@ static CHALLTESTSRV: OnceCell> = OnceCell::const_ne static PEBBLE: OnceCell> = OnceCell::const_new(); static POWERDNS: OnceCell> = OnceCell::const_new(); static SCIM_TESTER: OnceCell> = OnceCell::const_new(); +// inbuxa: scale-out storage (ST-5 to ST-15): a primary and a streaming replica +static PG_PRIMARY: OnceCell> = OnceCell::const_new(); +static PG_REPLICA: OnceCell> = OnceCell::const_new(); +const PG_REPLICATION_NETWORK: &str = "inbuxa-test-pg-replication"; +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"; const OPENLDAP_LDAPI_URL: &str = "ldapi://%2Fvar%2Frun%2Fslapd%2Fldapi/"; @@ -541,3 +549,91 @@ async fn wait_for_http(url: &str) { tokio::time::sleep(Duration::from_millis(500)).await; } } + +/// inbuxa: a PostgreSQL primary on port 5442 with a hot-standby streaming +/// replica on 5443, for the read-replica tests (ST-5 to ST-15). +pub async fn ensure_postgres_replicated() { + PG_PRIMARY + .get_or_init(|| async { + GenericImage::new("postgres", "16-alpine") + .with_wait_for(WaitFor::message_on_stderr( + "database system is ready to accept connections", + )) + .with_env_var("POSTGRES_USER", "stalwart") + .with_env_var("POSTGRES_PASSWORD", "stalwart") + .with_env_var("POSTGRES_DB", "stalwart") + .with_copy_to( + "/docker-entrypoint-initdb.d/replication.sh", + b"#!/bin/sh\necho 'host replication all all scram-sha-256' >> \"$PGDATA/pg_hba.conf\"\n" + .to_vec(), + ) + .with_cmd([ + "postgres", + "-c", + "wal_level=replica", + "-c", + "max_wal_senders=10", + "-c", + "hot_standby=on", + ]) + .with_network(PG_REPLICATION_NETWORK) + .with_mapped_port(PG_PRIMARY_PORT, 5432.tcp()) + .with_startup_timeout(READY_TIMEOUT) + .with_container_name(PG_PRIMARY_CONTAINER) + .with_reuse(ReuseDirective::Always) + .start() + .await + .expect("Failed to start the PostgreSQL primary") + }) + .await; + wait_for_tcp(PG_PRIMARY_PORT).await; + PG_REPLICA + .get_or_init(|| async { + GenericImage::new("postgres", "16-alpine") + .with_wait_for(WaitFor::message_on_stderr( + "database system is ready to accept read-only connections", + )) + .with_env_var("PGPASSWORD", "stalwart") + .with_cmd([ + "sh", + "-c", + concat!( + "if [ ! -s /var/lib/postgresql/replica/PG_VERSION ]; then ", + "until pg_basebackup -h inbuxa-test-pg-primary -U stalwart ", + "-D /var/lib/postgresql/replica -R -X stream; do sleep 1; done; fi; ", + "chown -R postgres:postgres /var/lib/postgresql/replica; ", + "chmod 700 /var/lib/postgresql/replica; ", + "exec su-exec postgres postgres -D /var/lib/postgresql/replica ", + "-c hot_standby=on" + ), + ]) + .with_network(PG_REPLICATION_NETWORK) + .with_mapped_port(PG_REPLICA_PORT, 5432.tcp()) + .with_startup_timeout(READY_TIMEOUT) + .with_container_name(PG_REPLICA_CONTAINER) + .with_reuse(ReuseDirective::Always) + .start() + .await + .expect("Failed to start the PostgreSQL replica") + }) + .await; + wait_for_tcp(PG_REPLICA_PORT).await; +} + +/// 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]) + .output() + .expect("docker exec"); + String::from_utf8_lossy(&output.stdout).trim().to_string() +} + +/// inbuxa: stops or starts a test container. +pub fn docker(action: &str, container: &str) { + let status = std::process::Command::new("docker") + .args([action, container]) + .status() + .expect("docker"); + assert!(status.success(), "docker {action} {container}"); +} diff --git a/tests/src/utils/storage.rs b/tests/src/utils/storage.rs index c5a5434..e8919ee 100644 --- a/tests/src/utils/storage.rs +++ b/tests/src/utils/storage.rs @@ -62,6 +62,35 @@ impl RegistryEnvStores for RegistryStore { } pub async fn build_data_store(typ: &str, path: &str) -> DataStore { + // inbuxa: scale-out storage: a primary with a streaming read replica + if typ == "PostgreSqlReplicated" { + crate::utils::containers::ensure_postgres_replicated().await; + let secret = || { + SecretKeyOptional::Value(SecretKeyValue { + secret: "stalwart".into(), + }) + }; + return DataStore::PostgreSql(PostgreSqlStore { + host: "localhost".into(), + port: crate::utils::containers::PG_PRIMARY_PORT as u64, + auth_username: "stalwart".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::PostgreSqlSettings { + host: "localhost".into(), + port: crate::utils::containers::PG_REPLICA_PORT as u64, + database: "stalwart".into(), + auth_username: "stalwart".to_string().into(), + auth_secret: secret(), + options: None, + }, + ]), + ..Default::default() + }); + } if typ == "MariaDb" { crate::utils::containers::ensure_mariadb().await; return DataStore::MySql(MySqlStore {