From 0755fad51ec5da188b6c4d7fade54c1b28b401ba Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 14:59:46 -0700 Subject: [PATCH] Scale-out storage: IMAP CONDSTORE raises the mark, and the two-node check (ST-7, test 11) A FETCH with CHANGEDSINCE presents its mod-sequence to the read scope, so a replica must have that change before it answers, as a JMAP sinceState already did. replica_cluster_tests covers test 11: with the replica's replay paused, a write on one node is read back through a second store with its own marks, sharing through Redis. Composite stores nest store futures deeply enough to pass rustc's default query depth once both postgres and redis are compiled in, so the server crates raise their recursion limit. --- crates/common/src/lib.rs | 4 + crates/coordinator/src/lib.rs | 4 + crates/dav/src/lib.rs | 4 + crates/directory/src/lib.rs | 4 + crates/email/src/lib.rs | 4 + crates/groupware/src/lib.rs | 4 + crates/http/src/lib.rs | 4 + crates/imap/src/lib.rs | 4 + crates/imap/src/op/fetch.rs | 2 + crates/jmap/src/lib.rs | 4 + crates/managesieve/src/lib.rs | 4 + crates/pop3/src/lib.rs | 4 + crates/services/src/lib.rs | 4 + crates/smtp/src/lib.rs | 4 + crates/spam-filter/src/lib.rs | 4 + crates/store/src/backend/scaleout/replica.rs | 26 +++- tests/src/lib.rs | 4 + tests/src/store/mod.rs | 2 + tests/src/store/replica_cluster.rs | 135 +++++++++++++++++++ 19 files changed, 223 insertions(+), 2 deletions(-) create mode 100644 tests/src/store/replica_cluster.rs 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(); +}