diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index abd5e55..5b9ed37 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use crate::auth::{AccessTokenInner, EmailAddress}; diff --git a/crates/coordinator/src/lib.rs b/crates/coordinator/src/lib.rs index 3abbde9..47d3854 100644 --- a/crates/coordinator/src/lib.rs +++ b/crates/coordinator/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] #[allow(unused_imports)] diff --git a/crates/dav/src/lib.rs b/crates/dav/src/lib.rs index dc4e6d3..9ce660b 100644 --- a/crates/dav/src/lib.rs +++ b/crates/dav/src/lib.rs @@ -3,6 +3,10 @@ * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ + +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] #![warn(clippy::large_futures)] pub mod calendar; diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index eb78641..892e50d 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use crate::backend::oidc::OpenIdDirectory; diff --git a/crates/email/src/lib.rs b/crates/email/src/lib.rs index 815803d..6712fd0 100644 --- a/crates/email/src/lib.rs +++ b/crates/email/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] pub mod cache; diff --git a/crates/groupware/src/lib.rs b/crates/groupware/src/lib.rs index 2033549..61d3642 100644 --- a/crates/groupware/src/lib.rs +++ b/crates/groupware/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use calcard::common::timezone::Tz; diff --git a/crates/http/src/lib.rs b/crates/http/src/lib.rs index de1c5d4..3449749 100644 --- a/crates/http/src/lib.rs +++ b/crates/http/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] pub mod api; diff --git a/crates/imap/src/lib.rs b/crates/imap/src/lib.rs index c3e0a02..6e5e97b 100644 --- a/crates/imap/src/lib.rs +++ b/crates/imap/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use std::sync::LazyLock; diff --git a/crates/imap/src/op/fetch.rs b/crates/imap/src/op/fetch.rs index 0bd2720..1f7e2ee 100644 --- a/crates/imap/src/op/fetch.rs +++ b/crates/imap/src/op/fetch.rs @@ -175,6 +175,8 @@ impl SessionData { // Convert state to modseq if let Some(changed_since) = arguments.changed_since { + // inbuxa: ST-7: a replica must have the state the client knows + store::backend::scaleout::replica::present_change(self.account_id, changed_since); // Obtain changes since the modseq. let changelog = self .server diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index dae9311..68e7031 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use jmap_proto::object::JmapObjectId; diff --git a/crates/managesieve/src/lib.rs b/crates/managesieve/src/lib.rs index 2d809aa..2a1d454 100644 --- a/crates/managesieve/src/lib.rs +++ b/crates/managesieve/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] pub mod core; diff --git a/crates/pop3/src/lib.rs b/crates/pop3/src/lib.rs index aeee3a7..4139606 100644 --- a/crates/pop3/src/lib.rs +++ b/crates/pop3/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use std::{net::IpAddr, sync::Arc}; diff --git a/crates/services/src/lib.rs b/crates/services/src/lib.rs index b249b46..b041adc 100644 --- a/crates/services/src/lib.rs +++ b/crates/services/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use broadcast::publisher::spawn_broadcast_publisher; diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index bf1cb5b..6119a7e 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] use common::{ diff --git a/crates/spam-filter/src/lib.rs b/crates/spam-filter/src/lib.rs index f0eedac..e8c9c87 100644 --- a/crates/spam-filter/src/lib.rs +++ b/crates/spam-filter/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores (sharded members, read replicas) nest store +// futures deeply enough to pass rustc's default query depth +#![recursion_limit = "512"] + #![warn(clippy::large_futures)] pub mod analysis; diff --git a/crates/store/src/backend/scaleout/replica.rs b/crates/store/src/backend/scaleout/replica.rs index 5cec3a6..ed433ad 100644 --- a/crates/store/src/backend/scaleout/replica.rs +++ b/crates/store/src/backend/scaleout/replica.rs @@ -141,12 +141,26 @@ pub fn is_replica_subspace(subspace: u8) -> bool { /// the client presented (ST-7, step 4), and the replica it settled on. pub struct ReadScope { accounts: Vec<(u32, u64)>, + /// Change ids the client presented during the scope (ST-7, step 4). + presented: Mutex>, 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 state the client presented raises the mark a replica must have +/// reached before it may answer this scope (ST-7, step 4). +pub fn present_change(account_id: u32, change_id: u64) { + let _ = READ_SCOPE.try_with(|scope| { + scope + .presented + .lock() + .unwrap() + .push((account_id, change_id)) + }); +} + /// A write happened in the current task: a read scope, if any, stops using /// replicas. pub fn note_scope_write() { @@ -167,6 +181,7 @@ pub async fn replica_read( .scope( Arc::new(ReadScope { accounts: accounts.into_iter().collect(), + presented: Mutex::new(Vec::new()), choice: tokio::sync::OnceCell::new(), wrote: std::sync::atomic::AtomicBool::new(false), }), @@ -310,8 +325,15 @@ impl ReplicatedStore { if !replica.usable() { continue; } - for (account_id, presented) in &scope.accounts { - let mark = self.mark(*account_id).await.max(*presented); + let presented = scope.presented.lock().unwrap().clone(); + for (account_id, from_request) in &scope.accounts { + let highest = presented + .iter() + .filter(|(id, _)| id == account_id) + .map(|(_, change_id)| *change_id) + .max() + .unwrap_or_default(); + let mark = self.mark(*account_id).await.max(*from_request).max(highest); if mark == 0 { continue; } diff --git a/tests/src/lib.rs b/tests/src/lib.rs index 3975eff..75f452d 100644 --- a/tests/src/lib.rs +++ b/tests/src/lib.rs @@ -4,6 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +// inbuxa: composite stores nest store futures deeply enough to pass +// rustc's default query depth +#![recursion_limit = "512"] + #[cfg(test)] use ::store::registry::bootstrap::Bootstrap; #[cfg(not(any(target_env = "msvc", target_os = "freebsd")))] diff --git a/tests/src/store/mod.rs b/tests/src/store/mod.rs index 6b99155..c3c31ea 100644 --- a/tests/src/store/mod.rs +++ b/tests/src/store/mod.rs @@ -14,6 +14,8 @@ pub mod registry; pub mod replica; // inbuxa: read replicas #[cfg(feature = "mysql")] pub mod replica_mysql; // inbuxa: read replicas on MySQL +#[cfg(all(feature = "postgres", feature = "redis"))] +pub mod replica_cluster; // inbuxa: read replicas across nodes pub mod scaleout; // inbuxa: scale-out storage #[cfg(any(feature = "postgres", feature = "mysql"))] pub mod sql_timeout; diff --git a/tests/src/store/replica_cluster.rs b/tests/src/store/replica_cluster.rs new file mode 100644 index 0000000..b083dc0 --- /dev/null +++ b/tests/src/store/replica_cluster.rs @@ -0,0 +1,135 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Read replicas across two nodes, from +//! `docs/spec/features/scale-out-storage.md` (test 11): a write on one node +//! is seen by a read on the other straight away, because the high-water +//! mark is shared (ST-7, step 2). Built with `postgres` and `redis`. + +use crate::utils::{ + containers::{PG_REPLICA_CONTAINER, ensure_redis, psql}, + server::TestServerBuilder, +}; +use registry::schema::structs::{Coordinator, InMemoryStore, RedisStore}; +use std::time::{Duration, Instant}; +use store::{ + SerializeInfallible, Store, U64_LEN, ValueKey, + backend::scaleout::replica::{ReplicaState, ReplicatedStore, replica_read}, + write::{AnyClass, BatchBuilder, ValueClass}, +}; +use types::collection::{Collection, SyncCollection}; + +fn replicated(store: &Store) -> std::sync::Arc { + match store { + Store::Replicated(store) => store.clone(), + other => panic!("the data store isn't replicated: {other:?}"), + } +} + +async fn wait_up(store: &ReplicatedStore, what: &str) { + let deadline = Instant::now() + Duration::from_secs(90); + while store.replicas[0].state() != ReplicaState::Up { + assert!( + Instant::now() < deadline, + "{what}: the replica is {:?}", + store.replicas[0].state() + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +fn account_key(account_id: u32) -> ValueClass { + let mut key = b"Rt".to_vec(); + key.extend_from_slice(&account_id.to_be_bytes()); + ValueClass::Any(AnyClass { + subspace: store::SUBSPACE_PROPERTY, + key, + }) +} + +/// Test 11. `cargo test -p tests --features postgres,redis +/// replica_cluster_tests -- --ignored`, with +/// `STORE=PostgreSqlReplicated`. +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn replica_cluster_tests() { + assert_eq!( + std::env::var("STORE").as_deref(), + Ok("PostgreSqlReplicated"), + "run with STORE=PostgreSqlReplicated" + ); + ensure_redis().await; + let redis = || RedisStore { + url: "redis://127.0.0.1".to_string(), + ..Default::default() + }; + + // Node A is a server; node B is a second replicated store with its own + // marks, as another node would have. Both share marks through Redis. + let node_a = TestServerBuilder::new("replica_cluster_a") + .await + .with_default_listeners() + .await + .with_object(Coordinator::Redis(redis())) + .await + .with_object(InMemoryStore::Redis(redis())) + .await + .build() + .await; + let store_a = replicated(node_a.server.store()); + let shared = store::backend::redis::RedisStore::open_single(redis()) + .await + .unwrap(); + store_a.share_marks(shared.clone()); + let node_b = crate::utils::storage::build_data_store( + "PostgreSqlReplicated", + node_a.temp_dir.path.to_str().unwrap(), + ) + .await; + let node_b = Store::build(node_b).await.unwrap(); + let store_b = replicated(&node_b); + store_b.share_marks(shared); + wait_up(&store_a, "node A").await; + wait_up(&store_b, "node B").await; + + // Everything written so far reaches the replica, then replay stops + tokio::time::sleep(Duration::from_secs(2)).await; + psql(PG_REPLICA_CONTAINER, "SELECT pg_wal_replay_pause()"); + + // Node A writes for the account, which assigns a change id + let account_id = 1234u32; + let value = store::rand::random::(); + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .log_container_insert(SyncCollection::Email) + .set(account_key(account_id), value.serialize()); + node_a + .server + .store() + .write(batch.build_all()) + .await + .unwrap(); + + // Node B reads it straight away, in a scope that may use the replica + let read = replica_read([(account_id, 0)], async { + node_b + .get_value::(ValueKey::from(account_key(account_id))) + .await + }) + .await + .unwrap(); + assert_eq!( + read, + Some(value), + "test 11: node B must see node A's write (the replica is paused)" + ); + let _ = U64_LEN; + + psql(PG_REPLICA_CONTAINER, "SELECT pg_wal_replay_resume()"); + node_a.temp_dir.delete(); +}