From e5e326de6b522150d7b4cdd57e7e20c6c406b670 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 10:11:30 -0700 Subject: [PATCH] Scale-out storage: sharded blob, in-memory and lookup stores; configured read replicas reported (ST-1 to ST-4, ST-16 to ST-30) A Sharded blob store places each blob on xxh3(key) mod N, over the whole key; reads fall back to the other members, so blobs placed under an earlier member list stay readable, and deletes find them wherever they are. The member list is recorded in the data store (secrets left out): added or reordered members are a warning, a missing one refuses to open. Blobs are compressed and marked before they reach a member. A Sharded in-memory or lookup store sends each key to its home Redis member, and prefix deletes and purges to all; a node whose member list differs from the recorded one logs an error and runs on. Members are checked for duplicates and must all open. Until read-replica routing is built, each configured replica is reported at startup instead of being silently ignored, and nothing connects to it. The new scaleout_blob_tests covers tests 2 to 7, and the existing blob suite passes against three FileSystem members (BLOB_STORE=Sharded). --- crates/store/src/backend/mod.rs | 2 + crates/store/src/backend/mysql/main.rs | 18 +- crates/store/src/backend/postgres/main.rs | 29 +- crates/store/src/backend/scaleout/blob.rs | 176 ++++++++ crates/store/src/backend/scaleout/layout.rs | 133 ++++++ crates/store/src/backend/scaleout/memory.rs | 128 ++++++ crates/store/src/backend/scaleout/mod.rs | 68 ++++ crates/store/src/build/blob.rs | 14 + crates/store/src/build/lookup.rs | 15 + crates/store/src/build/memory.rs | 15 + crates/store/src/dispatch/blob.rs | 124 +++--- crates/store/src/dispatch/lookup.rs | 63 +++ crates/store/src/lib.rs | 4 + tests/src/store/mod.rs | 1 + tests/src/store/scaleout.rs | 423 ++++++++++++++++++++ tests/src/utils/storage.rs | 15 + 16 files changed, 1144 insertions(+), 84 deletions(-) create mode 100644 crates/store/src/backend/scaleout/blob.rs create mode 100644 crates/store/src/backend/scaleout/layout.rs create mode 100644 crates/store/src/backend/scaleout/memory.rs create mode 100644 crates/store/src/backend/scaleout/mod.rs create mode 100644 tests/src/store/scaleout.rs diff --git a/crates/store/src/backend/mod.rs b/crates/store/src/backend/mod.rs index b3508a3..c94bd86 100644 --- a/crates/store/src/backend/mod.rs +++ b/crates/store/src/backend/mod.rs @@ -26,6 +26,8 @@ pub mod rocksdb; pub mod s3; #[cfg(feature = "sqlite")] pub mod sqlite; +// inbuxa: scale-out storage (sharded stores) +pub mod scaleout; pub const MAX_TOKEN_LENGTH: usize = (u8::MAX >> 1) as usize; diff --git a/crates/store/src/backend/mysql/main.rs b/crates/store/src/backend/mysql/main.rs index 398075f..1206156 100644 --- a/crates/store/src/backend/mysql/main.rs +++ b/crates/store/src/backend/mysql/main.rs @@ -51,18 +51,16 @@ impl MysqlStore { PoolOpts::default().with_constraints(PoolConstraints::new(pool_min, pool_max).unwrap()), ); - let mut replicas = vec![]; + // inbuxa: ST-2: replicas aren't used yet (the scale-out decision), so + // each one is reported rather than silently ignored for replica in config.read_replicas { - replicas.push(Store::MySQL(Arc::new(MysqlStore { - conn_pool: Pool::new( - opts.clone() - .ip_or_hostname(replica.host) - .user(replica.auth_username) - .pass(replica.auth_secret.secret().await?.map(|v| v.into_owned())) - .db_name(Some(replica.database)) - .tcp_port(replica.port as u16), + 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 ), - }))) + ); } let primary = Store::MySQL(Arc::new(MysqlStore { diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index 446b348..c2bbdff 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -57,27 +57,16 @@ impl PostgresStore { .map_err(|e| format!("Failed to create connection pool: {e}"))?; let ts_configs = discover_ts_configs(&primary_pool).await; - let mut replicas = vec![]; + // inbuxa: ST-2: replicas aren't used yet (the scale-out decision), so + // each one is reported rather than silently ignored for replica in config.read_replicas { - let mut cfg = cfg.clone(); - cfg.dbname = replica.database.into(); - cfg.host = replica.host.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; - replicas.push(Store::PostgreSQL(Arc::new(PostgresStore { - conn_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}"))?, - ts_configs: ts_configs.clone(), - }))); + 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 primary = Store::PostgreSQL(Arc::new(PostgresStore { diff --git a/crates/store/src/backend/scaleout/blob.rs b/crates/store/src/backend/scaleout/blob.rs new file mode 100644 index 0000000..bb982c5 --- /dev/null +++ b/crates/store/src/backend/scaleout/blob.rs @@ -0,0 +1,176 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! The sharded blob store (ST-16 to ST-22). A blob lives on its home +//! member; reads fall back to the others so blobs placed under an earlier +//! member list stay readable. Blobs are stored compressed and marked, as +//! on any blob store, before they reach a member. + +use super::{home, layout}; +use crate::{BlobStore, Store, backend::fs::FsStore}; +use registry::schema::structs::{self, BlobStoreBase}; + +pub struct ShardedBlobStore { + pub members: Vec, + pub locations: Vec, +} + +/// A member's kind and location, never its secrets (ST-20, ST-22). +pub fn location(member: &BlobStoreBase) -> String { + match member { + BlobStoreBase::S3(s3) => format!( + "S3 {:?} bucket {} prefix {}", + s3.region, + s3.bucket, + s3.key_prefix.as_deref().unwrap_or_default() + ), + BlobStoreBase::Azure(azure) => format!( + "Azure account {} container {} prefix {}", + azure.storage_account, + azure.container, + azure.key_prefix.as_deref().unwrap_or_default() + ), + BlobStoreBase::FileSystem(fs) => { + format!("FileSystem {}", fs.path.trim_end_matches('/')) + } + BlobStoreBase::FoundationDb(fdb) => format!( + "FoundationDb {}", + fdb.cluster_file.as_deref().unwrap_or("default") + ), + BlobStoreBase::PostgreSql(pg) => { + format!("PostgreSql {}:{} {}", pg.host, pg.port, pg.database) + } + BlobStoreBase::MySql(my) => format!("MySql {}:{} {}", my.host, my.port, my.database), + } +} + +#[allow(unreachable_patterns, unused_variables)] +async fn open_member(member: BlobStoreBase) -> Result { + match member { + #[cfg(feature = "foundation")] + BlobStoreBase::FoundationDb(config) => crate::backend::foundationdb::FdbStore::open(config) + .await + .map(BlobStore::Store), + #[cfg(feature = "postgres")] + BlobStoreBase::PostgreSql(config) => crate::backend::postgres::PostgresStore::open(config) + .await + .map(BlobStore::Store), + #[cfg(feature = "mysql")] + BlobStoreBase::MySql(config) => crate::backend::mysql::MysqlStore::open(config) + .await + .map(BlobStore::Store), + #[cfg(feature = "s3")] + BlobStoreBase::S3(config) => crate::backend::s3::S3Store::open(config).await, + #[cfg(feature = "azure")] + BlobStoreBase::Azure(config) => crate::backend::azure::AzureStore::open(config).await, + BlobStoreBase::FileSystem(config) => FsStore::open(config).await, + _ => Err("Binary was not compiled with this member's blob store backend".to_string()), + } +} + +impl ShardedBlobStore { + /// Opens every member, checks the list (ST-22), and compares it with + /// the recorded one (ST-20). Warnings are returned for the build log. + pub async fn open( + config: structs::ShardedBlobStore, + data: &Store, + warnings: &mut Vec, + ) -> Result { + let members = config.stores.into_iter().collect::>(); + if members.len() < 2 { + return Err("A sharded blob store needs at least two members".to_string()); + } + let locations = members.iter().map(location).collect::>(); + for (index, location) in locations.iter().enumerate() { + if let Some(first) = locations[..index].iter().position(|l| l == location) { + return Err(format!( + "Members {} and {} are the same place: {location}", + first + 1, + index + 1 + )); + } + } + let mut opened = Vec::with_capacity(members.len()); + for (index, member) in members.into_iter().enumerate() { + opened.push( + open_member(member) + .await + .map_err(|err| format!("Member {}: {err}", index + 1))?, + ); + } + match layout::check(data, layout::key(b'b', ""), &locations, true) + .await + .map_err(|err| format!("Failed to read the recorded member list: {err}"))? + { + layout::Comparison::Unchanged => {} + layout::Comparison::Changed(change) => warnings.push(format!( + "The sharded blob store's member list changed ({change}); blobs whose home \ + moved are found by searching the other members" + )), + layout::Comparison::Missing(missing) => { + return Err(format!( + "Members recorded for the sharded blob store are missing, and blobs on \ + them would be unreachable: {}", + missing.join("; ") + )); + } + } + Ok(BlobStore::Sharded(std::sync::Arc::new(ShardedBlobStore { + members: opened, + locations, + }))) + } + + fn home(&self, key: &[u8]) -> usize { + home(key, self.members.len()) + } + + /// The home member, then the others in order (ST-17). An error from + /// the home member is returned without searching (ST-21). + pub async fn get(&self, key: &[u8]) -> trc::Result>> { + let home = self.home(key); + if let Some(data) = Box::pin(self.members[home].raw_get(key)).await? { + return Ok(Some(data)); + } + for (index, member) in self.members.iter().enumerate() { + if index == home { + continue; + } + if let Ok(Some(data)) = Box::pin(member.raw_get(key)).await { + trc::event!( + Store(trc::StoreEvent::UnexpectedError), + Key = key, + Details = format!( + "Misplaced blob: found on member {}, its home is member {}", + index + 1, + home + 1 + ), + ); + return Ok(Some(data)); + } + } + Ok(None) + } + + /// Writes go to the home member only (ST-18, ST-21). + pub async fn put(&self, key: &[u8], data: &[u8]) -> trc::Result<()> { + Box::pin(self.members[self.home(key)].raw_put(key, data)).await + } + + /// The home member first, then the others until one had it (ST-18). + pub async fn delete(&self, key: &[u8]) -> trc::Result { + let home = self.home(key); + if Box::pin(self.members[home].raw_delete(key)).await? { + return Ok(true); + } + for (index, member) in self.members.iter().enumerate() { + if index != home && Box::pin(member.raw_delete(key)).await? { + return Ok(true); + } + } + Ok(false) + } +} diff --git a/crates/store/src/backend/scaleout/layout.rs b/crates/store/src/backend/scaleout/layout.rs new file mode 100644 index 0000000..1b27dba --- /dev/null +++ b/crates/store/src/backend/scaleout/layout.rs @@ -0,0 +1,133 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! The member lists of sharded stores, recorded in the data store (ST-20, +//! ST-26): each member's kind and location, never its secrets. + +use crate::{ + Deserialize, SUBSPACE_INBUXA, Store, ValueKey, + write::{AnyClass, BatchBuilder, ValueClass}, +}; + +/// The fork's feature byte for scale-out storage. +const FEATURE: u8 = b'S'; + +/// Where a member list is recorded: `b` the blob store, `m` the in-memory +/// store, `l` a lookup store by namespace. +pub fn key(kind: u8, name: &str) -> ValueClass { + let mut key = vec![FEATURE, kind]; + key.extend_from_slice(name.as_bytes()); + ValueClass::Any(AnyClass { + subspace: SUBSPACE_INBUXA, + key, + }) +} + +struct Recorded(Vec); + +impl Deserialize for Recorded { + fn deserialize(bytes: &[u8]) -> trc::Result { + serde_json::from_slice(bytes) + .map(Recorded) + .map_err(|err| trc::StoreEvent::DeserializeError.reason(err)) + } +} + +/// How the configured list compares with the recorded one. +#[derive(Debug, PartialEq, Eq)] +pub enum Comparison { + /// No record yet, or the same list. + Unchanged, + /// Members added or reordered: the record was updated. + Changed(String), + /// A recorded member is gone. + Missing(Vec), +} + +pub fn compare(recorded: &[String], current: &[String]) -> Comparison { + let missing = recorded + .iter() + .filter(|member| !current.contains(member)) + .cloned() + .collect::>(); + if !missing.is_empty() { + Comparison::Missing(missing) + } else if recorded == current { + Comparison::Unchanged + } else { + let added = current + .iter() + .filter(|member| !recorded.contains(member)) + .cloned() + .collect::>(); + Comparison::Changed(if added.is_empty() { + format!("members reordered, was {recorded:?}, now {current:?}") + } else { + format!("members added: {added:?}") + }) + } +} + +/// Compares `current` with the record, then records `current` unless a +/// member is missing and `keep_on_missing` is set. +pub async fn check( + data: &Store, + class: ValueClass, + current: &[String], + keep_on_missing: bool, +) -> trc::Result { + if data.is_none() { + return Ok(Comparison::Unchanged); + } + let recorded = data + .get_value::(ValueKey::from(class.clone())) + .await? + .map(|r| r.0); + let comparison = match &recorded { + Some(recorded) => compare(recorded, current), + None => Comparison::Unchanged, + }; + let write = match &comparison { + Comparison::Unchanged => recorded.is_none(), + Comparison::Changed(_) => true, + Comparison::Missing(_) => !keep_on_missing, + }; + if write { + let mut batch = BatchBuilder::new(); + batch.set(class, serde_json::to_vec(current).unwrap_or_default()); + data.write(batch.build_all()).await?; + } + Ok(comparison) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn list(items: &[&str]) -> Vec { + items.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn compares_lists() { + assert_eq!( + compare(&list(&["a", "b"]), &list(&["a", "b"])), + Comparison::Unchanged + ); + assert!(matches!( + compare(&list(&["a", "b"]), &list(&["a", "b", "c"])), + Comparison::Changed(_) + )); + assert!(matches!( + compare(&list(&["a", "b"]), &list(&["b", "a"])), + Comparison::Changed(_) + )); + assert_eq!( + compare(&list(&["a", "b", "c"]), &list(&["a", "c"])), + Comparison::Missing(list(&["b"])) + ); + } +} diff --git a/crates/store/src/backend/scaleout/memory.rs b/crates/store/src/backend/scaleout/memory.rs new file mode 100644 index 0000000..0ccbfb6 --- /dev/null +++ b/crates/store/src/backend/scaleout/memory.rs @@ -0,0 +1,128 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! The sharded in-memory and lookup store (ST-23 to ST-29). Every +//! single-key operation goes to the key's home member, with no fallback; +//! operations over many keys go to every member. + +use super::{home, layout, without_credentials}; +use crate::{InMemoryStore, Store}; +use registry::schema::structs::{self, InMemoryStoreBase}; + +#[derive(Debug)] +pub struct ShardedInMemoryStore { + pub members: Vec, + pub locations: Vec, +} + +/// A member's kind and location, without credentials (ST-26, ST-29). +pub fn location(member: &InMemoryStoreBase) -> String { + let urls = |urls: ®istry::types::map::Map| { + let mut urls = urls + .iter() + .map(|url| without_credentials(url)) + .collect::>(); + urls.sort(); + urls.join(",") + }; + match member { + InMemoryStoreBase::Redis(redis) => format!("Redis {}", without_credentials(&redis.url)), + InMemoryStoreBase::RedisCluster(cluster) => { + format!("RedisCluster {}", urls(&cluster.urls)) + } + InMemoryStoreBase::RedisSentinel(sentinel) => format!( + "RedisSentinel {} {}", + urls(&sentinel.urls), + sentinel.service_name + ), + } +} + +#[allow(unreachable_patterns, unused_variables)] +async fn open_member(member: InMemoryStoreBase) -> Result { + match member { + #[cfg(feature = "redis")] + InMemoryStoreBase::Redis(config) => { + crate::backend::redis::RedisStore::open_single(config).await + } + #[cfg(feature = "redis")] + InMemoryStoreBase::RedisCluster(config) => { + crate::backend::redis::RedisStore::open_cluster(config).await + } + #[cfg(feature = "redis")] + InMemoryStoreBase::RedisSentinel(config) => { + crate::backend::redis::RedisStore::open_sentinel(config).await + } + _ => Err("Binary was not compiled with this member's in-memory backend".to_string()), + } +} + +impl ShardedInMemoryStore { + /// Opens every member and checks the list (ST-29), then compares it with + /// the recorded one (ST-26). `name` tells lookup stores apart. + pub async fn open( + config: structs::ShardedInMemoryStore, + name: &str, + data: &Store, + warnings: &mut Vec, + ) -> Result { + let members = config.stores.into_iter().collect::>(); + if members.len() < 2 { + return Err("A sharded in-memory store needs at least two members".to_string()); + } + let locations = members.iter().map(location).collect::>(); + for (index, location) in locations.iter().enumerate() { + if let Some(first) = locations[..index].iter().position(|l| l == location) { + return Err(format!( + "Members {} and {} are the same server: {location}", + first + 1, + index + 1 + )); + } + } + let mut opened = Vec::with_capacity(members.len()); + for (index, member) in members.into_iter().enumerate() { + opened.push( + open_member(member) + .await + .map_err(|err| format!("Member {}: {err}", index + 1))?, + ); + } + // ST-26: a different list is an error to fix, not a reason to refuse + let kind = if name.is_empty() { b'm' } else { b'l' }; + let difference = match layout::check(data, layout::key(kind, name), &locations, false).await + { + Ok(layout::Comparison::Unchanged) => None, + Ok(layout::Comparison::Changed(change)) => Some(change), + Ok(layout::Comparison::Missing(missing)) => { + Some(format!("members gone: {}", missing.join("; "))) + } + Err(err) => Some(format!("the recorded list couldn't be read: {err}")), + }; + if let Some(difference) = difference { + let message = format!( + "The sharded in-memory store's member list differs from the one recorded \ + ({difference}); every node must use the same list" + ); + trc::event!( + Store(trc::StoreEvent::RedisError), + Details = message.clone() + ); + warnings.push(message); + } + Ok(InMemoryStore::Sharded(std::sync::Arc::new( + ShardedInMemoryStore { + members: opened, + locations, + }, + ))) + } + + /// The member a key lives on (ST-23). + pub fn member(&self, key: &[u8]) -> &InMemoryStore { + &self.members[home(key, self.members.len())] + } +} diff --git a/crates/store/src/backend/scaleout/mod.rs b/crates/store/src/backend/scaleout/mod.rs new file mode 100644 index 0000000..d4cecff --- /dev/null +++ b/crates/store/src/backend/scaleout/mod.rs @@ -0,0 +1,68 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Scale-out storage (`docs/spec/features/scale-out-storage.md`): sharded +//! blob stores (ST-16 to ST-22) and sharded in-memory and lookup stores +//! (ST-23 to ST-29). Each is one more variant of the store enums, whose +//! members are ordinary stores. + +pub mod blob; +pub mod layout; +pub mod memory; + +pub use blob::ShardedBlobStore; +pub use memory::ShardedInMemoryStore; + +/// A key's home: `xxh3_64(key) mod N`, seed 0, over the whole key (ST-16). +/// Fixed forever once shipped. +pub fn home(key: &[u8], members: usize) -> usize { + (xxhash_rust::xxh3::xxh3_64(key) % members.max(1) as u64) as usize +} + +/// A URL without its user information, for records and logs. +pub fn without_credentials(url: &str) -> String { + match url.split_once("://") { + Some((scheme, rest)) => { + let rest = match rest.split_once('/') { + Some((authority, path)) => { + let host = authority.rsplit_once('@').map_or(authority, |(_, h)| h); + format!("{host}/{path}") + } + None => rest.rsplit_once('@').map_or(rest, |(_, h)| h).to_string(), + }; + format!("{scheme}://{rest}") + } + None => url.rsplit_once('@').map_or(url, |(_, h)| h).to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn places_and_hides_credentials() { + // Stable placement: these values must never change (ST-16) + assert_eq!(home(b"", 3), (xxhash_rust::xxh3::xxh3_64(b"") % 3) as usize); + let spread = (0u32..3000) + .map(|n| home(&n.to_be_bytes(), 3)) + .fold([0; 3], |mut acc, h| { + acc[h] += 1; + acc + }); + assert!(spread.iter().all(|n| *n > 800), "{spread:?}"); + + assert_eq!( + without_credentials("redis://user:secret@host:6379/0"), + "redis://host:6379/0" + ); + assert_eq!(without_credentials("rediss://:pw@host"), "rediss://host"); + assert_eq!( + without_credentials("redis://host:6379"), + "redis://host:6379" + ); + } +} diff --git a/crates/store/src/build/blob.rs b/crates/store/src/build/blob.rs index bf6c15e..4fa4169 100644 --- a/crates/store/src/build/blob.rs +++ b/crates/store/src/build/blob.rs @@ -39,6 +39,20 @@ impl BlobStore { structs::BlobStore::FileSystem(file_system_store) => { FsStore::open(file_system_store).await } + // inbuxa: ST-16 to ST-22 + structs::BlobStore::Sharded(sharded) => { + let mut warnings = Vec::new(); + let result = crate::backend::scaleout::ShardedBlobStore::open( + sharded, + &bp.data_store, + &mut warnings, + ) + .await; + for warning in warnings { + bp.build_warning(ObjectType::BlobStore.singleton(), warning); + } + result + } _ => Err("Binary was not compiled with the selected blob store backend".to_string()), }; diff --git a/crates/store/src/build/lookup.rs b/crates/store/src/build/lookup.rs index 495f5d4..f34ac8f 100644 --- a/crates/store/src/build/lookup.rs +++ b/crates/store/src/build/lookup.rs @@ -49,6 +49,21 @@ impl LookupStores { LookupStore::RedisCluster(redis_cluster_store) => { crate::backend::redis::RedisStore::open_cluster(redis_cluster_store).await } + // inbuxa: ST-28 + LookupStore::Sharded(sharded) => { + let mut warnings = Vec::new(); + let result = crate::backend::scaleout::ShardedInMemoryStore::open( + sharded, + store.namespace.as_str(), + &bp.data_store, + &mut warnings, + ) + .await; + for warning in warnings { + bp.build_warning(id, warning); + } + result + } _ => Err( "Binary was not compiled with the selected lookup store backend".to_string(), ), diff --git a/crates/store/src/build/memory.rs b/crates/store/src/build/memory.rs index 754b635..c787743 100644 --- a/crates/store/src/build/memory.rs +++ b/crates/store/src/build/memory.rs @@ -26,6 +26,21 @@ impl InMemoryStore { structs::InMemoryStore::RedisSentinel(redis_sentinel_store) => { crate::backend::redis::RedisStore::open_sentinel(redis_sentinel_store).await } + // inbuxa: ST-23 to ST-29 + structs::InMemoryStore::Sharded(sharded) => { + let mut warnings = Vec::new(); + let result = crate::backend::scaleout::ShardedInMemoryStore::open( + sharded, + "", + &bp.data_store, + &mut warnings, + ) + .await; + for warning in warnings { + bp.build_warning(ObjectType::InMemoryStore.singleton(), warning); + } + result + } _ => Err("Binary was not compiled with the selected in-memory backend".to_string()), }; diff --git a/crates/store/src/dispatch/blob.rs b/crates/store/src/dispatch/blob.rs index 8e01ac8..d850ec4 100644 --- a/crates/store/src/dispatch/blob.rs +++ b/crates/store/src/dispatch/blob.rs @@ -16,28 +16,7 @@ const NONE_MARKER: u8 = 0x00; impl BlobStore { pub async fn get_blob(&self, key: &[u8], range: Range) -> trc::Result>> { let start_time = Instant::now(); - let result = match &self { - BlobStore::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.get_blob(key, 0..usize::MAX).await, - #[cfg(feature = "foundation")] - Store::FoundationDb(store) => store.get_blob(key, 0..usize::MAX).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.get_blob(key, 0..usize::MAX).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.get_blob(key, 0..usize::MAX).await, - #[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, - Store::None => Err(trc::StoreEvent::NotConfigured.into()), - }, - BlobStore::Fs(store) => store.get_blob(key, 0..usize::MAX).await, - #[cfg(feature = "s3")] - BlobStore::S3(store) => store.get_blob(key, 0..usize::MAX).await, - #[cfg(feature = "azure")] - BlobStore::Azure(store) => store.get_blob(key, 0..usize::MAX).await, - } - .caused_by(trc::location!())?; + let result = self.raw_get(key).await.caused_by(trc::location!())?; trc::event!( Store(StoreEvent::BlobRead), @@ -126,28 +105,7 @@ impl BlobStore { }; let start_time = Instant::now(); - let result = match &self { - BlobStore::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.put_blob(key, &data).await, - #[cfg(feature = "foundation")] - Store::FoundationDb(store) => store.put_blob(key, &data).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.put_blob(key, &data).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.put_blob(key, &data).await, - #[cfg(feature = "rocks")] - Store::RocksDb(store) => store.put_blob(key, &data).await, - Store::Ephemeral(store) => store.put_blob(key, &data).await, - Store::None => Err(trc::StoreEvent::NotConfigured.into()), - }, - BlobStore::Fs(store) => store.put_blob(key, &data).await, - #[cfg(feature = "s3")] - BlobStore::S3(store) => store.put_blob(key, &data).await, - #[cfg(feature = "azure")] - BlobStore::Azure(store) => store.put_blob(key, &data).await, - } - .caused_by(trc::location!()); + let result = self.raw_put(key, &data).await.caused_by(trc::location!()); trc::event!( Store(StoreEvent::BlobWrite), @@ -161,7 +119,72 @@ impl BlobStore { pub async fn delete_blob(&self, key: &[u8]) -> trc::Result { let start_time = Instant::now(); - let result = match &self { + let result = self.raw_delete(key).await.caused_by(trc::location!()); + + trc::event!( + Store(StoreEvent::BlobWrite), + Key = key, + Elapsed = start_time.elapsed(), + ); + + result + } + + /// A stored blob as it is on the backend, compression marker included. + pub(crate) async fn raw_get(&self, key: &[u8]) -> trc::Result>> { + match &self { + BlobStore::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.get_blob(key, 0..usize::MAX).await, + #[cfg(feature = "foundation")] + Store::FoundationDb(store) => store.get_blob(key, 0..usize::MAX).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.get_blob(key, 0..usize::MAX).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.get_blob(key, 0..usize::MAX).await, + #[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, + Store::None => Err(trc::StoreEvent::NotConfigured.into()), + }, + BlobStore::Fs(store) => store.get_blob(key, 0..usize::MAX).await, + #[cfg(feature = "s3")] + BlobStore::S3(store) => store.get_blob(key, 0..usize::MAX).await, + #[cfg(feature = "azure")] + BlobStore::Azure(store) => store.get_blob(key, 0..usize::MAX).await, + // inbuxa: ST-17 + BlobStore::Sharded(store) => store.get(key).await, + } + } + + pub(crate) async fn raw_put(&self, key: &[u8], data: &[u8]) -> trc::Result<()> { + match &self { + BlobStore::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.put_blob(key, data).await, + #[cfg(feature = "foundation")] + Store::FoundationDb(store) => store.put_blob(key, data).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.put_blob(key, data).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.put_blob(key, data).await, + #[cfg(feature = "rocks")] + Store::RocksDb(store) => store.put_blob(key, data).await, + Store::Ephemeral(store) => store.put_blob(key, data).await, + Store::None => Err(trc::StoreEvent::NotConfigured.into()), + }, + BlobStore::Fs(store) => store.put_blob(key, data).await, + #[cfg(feature = "s3")] + BlobStore::S3(store) => store.put_blob(key, data).await, + #[cfg(feature = "azure")] + BlobStore::Azure(store) => store.put_blob(key, data).await, + // inbuxa: ST-18 + BlobStore::Sharded(store) => store.put(key, data).await, + } + } + + pub(crate) async fn raw_delete(&self, key: &[u8]) -> trc::Result { + match &self { BlobStore::Store(store) => match store { #[cfg(feature = "sqlite")] Store::SQLite(store) => store.delete_blob(key).await, @@ -181,15 +204,8 @@ impl BlobStore { BlobStore::S3(store) => store.delete_blob(key).await, #[cfg(feature = "azure")] BlobStore::Azure(store) => store.delete_blob(key).await, + // inbuxa: ST-18 + BlobStore::Sharded(store) => store.delete(key).await, } - .caused_by(trc::location!()); - - trc::event!( - Store(StoreEvent::BlobWrite), - Key = key, - Elapsed = start_time.elapsed(), - ); - - result } } diff --git a/crates/store/src/dispatch/lookup.rs b/crates/store/src/dispatch/lookup.rs index 8f2f957..9f015f2 100644 --- a/crates/store/src/dispatch/lookup.rs +++ b/crates/store/src/dispatch/lookup.rs @@ -45,6 +45,11 @@ impl InMemoryStore { }); store.write(batch.build_all()).await.map(|_| ()) } + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let member = store.member(&kv.key); + Box::pin(member.key_set(kv)).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_set(&kv.key, &kv.value, kv.expires).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -90,6 +95,11 @@ impl InMemoryStore { store.write(batch.build_all()).await.map(|_| 0) } } + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let member = store.member(&kv.key); + Box::pin(member.counter_incr(kv, return_value)).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_incr(&kv.key, kv.value, kv.expires).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -109,6 +119,11 @@ impl InMemoryStore { }); store.write(batch.build_all()).await.map(|_| ()) } + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let key = key.into().into_bytes(); + Box::pin(store.member(&key).key_delete(key.clone())).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_delete(key.into().as_bytes()).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -128,6 +143,11 @@ impl InMemoryStore { }); store.write(batch.build_all()).await.map(|_| ()) } + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let key = key.into().into_bytes(); + Box::pin(store.member(&key).counter_delete(key.clone())).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_delete(key.into().as_bytes()).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -167,6 +187,15 @@ impl InMemoryStore { ) .await } + // inbuxa: ST-24: every member + InMemoryStore::Sharded(store) => { + for (index, member) in store.members.iter().enumerate() { + Box::pin(member.key_delete_prefix(prefix)) + .await + .map_err(|err| err.details(format!("Member {}", index + 1)))?; + } + Ok(()) + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_delete_prefix(prefix).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -187,6 +216,11 @@ impl InMemoryStore { ))) .await .map(|value| value.and_then(|v| v.into())), + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let key = key.into().into_bytes(); + Box::pin(store.member(&key).key_get::(key.clone())).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_get(key.into().as_bytes()).await, InMemoryStore::Static(store) => Ok(match store.as_ref() { @@ -217,6 +251,11 @@ impl InMemoryStore { ))) .await } + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let key = key.into().into_bytes(); + Box::pin(store.member(&key).counter_get(key.clone())).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.counter_get(key.into().as_bytes()).await, InMemoryStore::Static(_) | InMemoryStore::Http(_) => { @@ -234,6 +273,11 @@ impl InMemoryStore { ))) .await .map(|value| matches!(value, Some(LookupValue::Value(Empty)))), + // inbuxa: ST-23 + InMemoryStore::Sharded(store) => { + let key = key.into().into_bytes(); + Box::pin(store.member(&key).key_exists(key.clone())).await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => store.key_exists(key.into().as_bytes()).await, InMemoryStore::Static(store) => Ok(match store.as_ref() { @@ -334,6 +378,15 @@ impl InMemoryStore { .caused_by(trc::location!())), } } + // inbuxa: ST-23: a lock and its release meet on one member + InMemoryStore::Sharded(store) => { + Box::pin( + store + .member(&KeyValue::<()>::build_key(prefix, key)) + .try_lock(prefix, key, duration), + ) + .await + } #[cfg(feature = "redis")] InMemoryStore::Redis(store) => { store @@ -431,6 +484,14 @@ impl InMemoryStore { } } } + // inbuxa: ST-24: every member + InMemoryStore::Sharded(store) => { + for (index, member) in store.members.iter().enumerate() { + Box::pin(member.purge_in_memory_store()) + .await + .map_err(|err| err.details(format!("Member {}", index + 1)))?; + } + } #[cfg(feature = "redis")] InMemoryStore::Redis(_) => {} InMemoryStore::Static(_) | InMemoryStore::Http(_) => {} @@ -450,6 +511,8 @@ impl InMemoryStore { match self { #[cfg(feature = "redis")] InMemoryStore::Redis(_) => true, + // inbuxa: ST-3: as its members are + InMemoryStore::Sharded(store) => store.members.iter().all(|m| m.is_redis()), InMemoryStore::Static(_) => false, _ => false, } diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 606c0fd..d30d690 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -171,6 +171,8 @@ pub enum BlobStore { S3(Arc), #[cfg(feature = "azure")] Azure(Arc), + // inbuxa: ST-16 to ST-22 + Sharded(Arc), } #[derive(Clone)] @@ -187,6 +189,8 @@ pub enum InMemoryStore { Redis(Arc), Http(Arc), Static(Arc), + // inbuxa: ST-23 to ST-29 + Sharded(Arc), } #[derive(Clone)] diff --git a/tests/src/store/mod.rs b/tests/src/store/mod.rs index 8919d71..5b10b61 100644 --- a/tests/src/store/mod.rs +++ b/tests/src/store/mod.rs @@ -10,6 +10,7 @@ pub mod lookup; pub mod ops; pub mod query; pub mod registry; +pub mod scaleout; // inbuxa: scale-out storage #[cfg(any(feature = "postgres", feature = "mysql"))] pub mod sql_timeout; diff --git a/tests/src/store/scaleout.rs b/tests/src/store/scaleout.rs new file mode 100644 index 0000000..6ffa715 --- /dev/null +++ b/tests/src/store/scaleout.rs @@ -0,0 +1,423 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Scale-out storage acceptance tests, from +//! `docs/spec/features/scale-out-storage.md`: the sharded blob store over +//! FileSystem members (tests 2 to 8) and, built with `redis`, the sharded +//! in-memory and lookup stores (tests 20, 22 and 23). Each check names its +//! test number or requirement. + +use crate::utils::server::TestServerBuilder; +use registry::{ + schema::{ + enums::CompressionAlgo, + structs::{self, BlobStoreBase, FileSystemStore}, + }, + types::list::List, +}; +use store::{ + BlobStore, Store, + backend::{fs::FsStore, scaleout}, +}; +use types::blob_hash::BlobHash; + +fn fs_member(dir: &std::path::Path) -> BlobStoreBase { + std::fs::create_dir_all(dir).unwrap(); + BlobStoreBase::FileSystem(FileSystemStore { + path: dir.to_str().unwrap().to_string(), + ..Default::default() + }) +} + +async fn open_blob( + data: &Store, + dirs: &[std::path::PathBuf], +) -> (Result, Vec) { + let mut warnings = Vec::new(); + let result = scaleout::ShardedBlobStore::open( + structs::ShardedBlobStore { + stores: List::from_iter(dirs.iter().map(|dir| fs_member(dir))), + }, + data, + &mut warnings, + ) + .await; + (result, warnings) +} + +async fn single(dir: &std::path::Path) -> BlobStore { + FsStore::open(FileSystemStore { + path: dir.to_str().unwrap().to_string(), + ..Default::default() + }) + .await + .unwrap() +} + +/// Which members hold a blob. +async fn holders(dirs: &[std::path::PathBuf], key: &[u8]) -> Vec { + let mut found = Vec::new(); + for (index, dir) in dirs.iter().enumerate() { + if single(dir) + .await + .get_blob(key, 0..usize::MAX) + .await + .unwrap() + .is_some() + { + found.push(index); + } + } + found +} + +/// `cargo test -p tests scaleout_blob_tests -- --ignored`. +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn scaleout_blob_tests() { + let test = TestServerBuilder::new("scaleout_blob_tests") + .await + .build() + .await; + let data = test.server.core.storage.data.clone(); + let base = test.temp_dir.path.join("shards"); + let dirs = (1..=4) + .map(|n| base.join(format!("member-{n}"))) + .collect::>(); + + // Test 2: each blob lands on exactly its home member (ST-16) + let (three, warnings) = open_blob(&data, &dirs[..3]).await; + let three = three.expect("test 8: a sharded blob store builds"); + assert!(warnings.is_empty(), "{warnings:?}"); + let blobs = (0..60u32) + .map(|n| { + let content = format!("blob number {n} with some content").into_bytes(); + (BlobHash::generate(&content), content) + }) + .collect::>(); + for (hash, content) in &blobs { + three + .put_blob(hash.as_slice(), content, CompressionAlgo::Lz4) + .await + .unwrap(); + assert_eq!( + holders(&dirs[..3], hash.as_slice()).await, + vec![scaleout::home(hash.as_slice(), 3)], + "test 2" + ); + assert_eq!( + three + .get_blob(hash.as_slice(), 5..11) + .await + .unwrap() + .as_deref(), + Some(&content[5..11]), + "test 2: ranges" + ); + } + let spread = (0..3) + .map(|m| { + blobs + .iter() + .filter(|(h, _)| scaleout::home(h.as_slice(), 3) == m) + .count() + }) + .collect::>(); + assert!(spread.iter().all(|n| *n > 0), "test 2: spread {spread:?}"); + + // Test 7: two members naming the same directory (ST-22) + let (same, _) = open_blob(&data, &[dirs[0].clone(), dirs[0].clone()]).await; + assert!( + same.err().is_some_and(|e| e.contains("same place")), + "test 7" + ); + let (alone, _) = open_blob(&data, &dirs[..1]).await; + assert!(alone.is_err(), "ST-22: at least two members"); + + // Test 3: a fourth member; every blob still reads, new ones land by the + // new mapping (ST-17, ST-19, ST-20) + let (four, warnings) = open_blob(&data, &dirs).await; + let four = four.unwrap(); + assert!( + warnings.iter().any(|w| w.contains("changed")), + "ST-20: {warnings:?}" + ); + for (hash, content) in &blobs { + assert_eq!( + four.get_blob(hash.as_slice(), 0..usize::MAX) + .await + .unwrap() + .as_deref(), + Some(content.as_slice()), + "test 3" + ); + } + let fresh = BlobHash::generate(b"written after the fourth member"); + four.put_blob( + fresh.as_slice(), + b"written after the fourth member", + CompressionAlgo::None, + ) + .await + .unwrap(); + assert_eq!( + holders(&dirs, fresh.as_slice()).await, + vec![scaleout::home(fresh.as_slice(), 4)], + "test 3" + ); + + // Test 6: a blob whose home moved is deleted from where it is (ST-18) + let moved = blobs + .iter() + .find(|(h, _)| scaleout::home(h.as_slice(), 3) != scaleout::home(h.as_slice(), 4)) + .expect("some blob's home moved"); + assert!( + four.delete_blob(moved.0.as_slice()).await.unwrap(), + "test 6" + ); + assert!( + holders(&dirs, moved.0.as_slice()).await.is_empty(), + "test 6" + ); + assert!( + !four.delete_blob(moved.0.as_slice()).await.unwrap(), + "ST-18" + ); + + // Test 5: one member unreadable; the others still serve (ST-21) + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let broken = 1usize; + std::fs::set_permissions(&dirs[broken], std::fs::Permissions::from_mode(0o000)).unwrap(); + for (hash, content) in &blobs { + if *hash == moved.0 { + continue; + } + let located = holders_unchecked(&dirs, hash.as_slice(), broken).await; + let result = four.get_blob(hash.as_slice(), 0..usize::MAX).await; + match located { + // On a readable member: served, whatever its home + Some(_) => { + assert_eq!( + result.unwrap().as_deref(), + Some(content.as_slice()), + "test 5" + ) + } + // On the unreadable one: a miss or an error, never the blob + // (the FileSystem backend reports an unreadable file as missing) + None => assert!(!matches!(result, Ok(Some(_))), "test 5"), + } + } + let homed_on_broken = (0u32..) + .map(|n| BlobHash::generate(format!("new blob {n}").as_bytes())) + .find(|h| scaleout::home(h.as_slice(), 4) == broken) + .unwrap(); + assert!( + four.put_blob(homed_on_broken.as_slice(), b"x", CompressionAlgo::None) + .await + .is_err(), + "test 5: a write homed on the broken member fails" + ); + std::fs::set_permissions(&dirs[broken], std::fs::Permissions::from_mode(0o755)).unwrap(); + } + + // Test 4: a recorded member removed: refused, named (ST-20) + let (removed, _) = open_blob(&data, &dirs[..2]).await; + let err = removed.err().expect("test 4: refused"); + assert!( + err.contains("member-3") && err.contains("member-4"), + "test 4: {err}" + ); + + // ST-1: nothing changes for a server without one + assert!( + matches!(test.server.core.storage.blob, BlobStore::Store(_)), + "the default blob store is unchanged (ST-1)" + ); + + test.temp_dir.delete(); +} + +/// Where a blob is, skipping a member that can't be read. +async fn holders_unchecked(dirs: &[std::path::PathBuf], key: &[u8], skip: usize) -> Option { + for (index, dir) in dirs.iter().enumerate() { + if index == skip { + continue; + } + if let Ok(Some(_)) = single(dir).await.get_blob(key, 0..usize::MAX).await { + return Some(index); + } + } + None +} + +/// Tests 20, 22 and 23 over two databases of one Redis server, which the +/// store treats as two members. `cargo test -p tests --features redis +/// scaleout_memory_tests -- --ignored`. +#[cfg(feature = "redis")] +#[ignore] +#[tokio::test(flavor = "multi_thread")] +pub async fn scaleout_memory_tests() { + use store::{InMemoryStore, dispatch::lookup::KeyValue}; + + crate::utils::containers::ensure_redis().await; + let test = TestServerBuilder::new("scaleout_memory_tests") + .await + .build() + .await; + let data = test.server.core.storage.data.clone(); + let member = |db: u32| { + structs::InMemoryStoreBase::Redis(structs::RedisStore { + url: format!("redis://127.0.0.1/{db}"), + ..Default::default() + }) + }; + let open = |dbs: Vec, name: &'static str| { + let data = data.clone(); + async move { + let mut warnings = Vec::new(); + let store = scaleout::ShardedInMemoryStore::open( + structs::ShardedInMemoryStore { + stores: List::from_iter(dbs.into_iter().map(member)), + }, + name, + &data, + &mut warnings, + ) + .await; + (store, warnings) + } + }; + let single = |db: u32| async move { + store::backend::redis::RedisStore::open_single(structs::RedisStore { + url: format!("redis://127.0.0.1/{db}"), + ..Default::default() + }) + .await + .unwrap() + }; + + // Test 20: keys live on their home member (ST-23) + let (sharded, warnings) = open(vec![11, 12], "").await; + let sharded = sharded.unwrap(); + assert!(warnings.is_empty(), "{warnings:?}"); + assert!(sharded.is_redis(), "ST-3"); + let members = [single(11).await, single(12).await]; + for n in 0..40u32 { + let key = format!("scaleout-key-{n}").into_bytes(); + sharded + .key_set(KeyValue::new(key.clone(), b"value".to_vec()).expires(60)) + .await + .unwrap(); + let home = scaleout::home(&key, 2); + for (index, member) in members.iter().enumerate() { + assert_eq!( + member.key_exists(key.clone()).await.unwrap(), + index == home, + "test 20: {n}" + ); + } + assert_eq!( + sharded + .key_get::(key.clone()) + .await + .unwrap() + .as_deref(), + Some("value") + ); + } + // Counters, locks and rate limits behave as with one Redis + for n in 0..3 { + assert_eq!( + sharded + .counter_incr(KeyValue::new(b"scaleout-counter".to_vec(), 2), true) + .await + .unwrap(), + 2 * (n + 1) + ); + } + assert!(sharded.try_lock(7, b"scaleout-lock", 30).await.unwrap()); + assert!(!sharded.try_lock(7, b"scaleout-lock", 30).await.unwrap()); + sharded.remove_lock(7, b"scaleout-lock").await.unwrap(); + assert!( + sharded.try_lock(7, b"scaleout-lock", 30).await.unwrap(), + "ST-23" + ); + let rate = registry::schema::structs::Rate { + count: 2, + period: registry::types::duration::Duration::from_millis(60_000), + }; + assert!( + sharded + .is_rate_allowed(9, b"who", &rate, false) + .await + .unwrap() + .is_none() + ); + assert!( + sharded + .is_rate_allowed(9, b"who", &rate, false) + .await + .unwrap() + .is_none() + ); + assert!( + sharded + .is_rate_allowed(9, b"who", &rate, false) + .await + .unwrap() + .is_some() + ); + // A prefix delete clears both members (ST-24) + sharded.key_delete_prefix(b"scaleout-").await.unwrap(); + for member in &members { + for n in 0..40u32 { + assert!( + !member + .key_exists(format!("scaleout-key-{n}").into_bytes()) + .await + .unwrap(), + "ST-24" + ); + } + } + sharded.purge_in_memory_store().await.unwrap(); + + // Test 22: a node with a different member list is told so (ST-26) + let (other, warnings) = open(vec![11, 13], "").await; + assert!(other.is_ok(), "ST-26: it still runs"); + assert!( + warnings.iter().any(|w| w.contains("differs")), + "test 22: {warnings:?}" + ); + + // Test 23: a sharded lookup store (ST-28) + let (lookup, _) = open(vec![11, 12], "scaleout-ns").await; + let lookup: InMemoryStore = lookup.unwrap(); + lookup + .key_set(KeyValue::new(b"scaleout-lookup".to_vec(), b"1".to_vec())) + .await + .unwrap(); + assert!( + lookup + .key_exists(b"scaleout-lookup".to_vec()) + .await + .unwrap() + ); + lookup + .key_delete(b"scaleout-lookup".to_vec()) + .await + .unwrap(); + assert!(lookup.clone().into_store().is_none(), "ST-28"); + + // ST-29: duplicates refused + let (duplicate, _) = open(vec![11, 11], "").await; + assert!(duplicate.is_err(), "ST-29"); + + test.temp_dir.delete(); +} diff --git a/tests/src/utils/storage.rs b/tests/src/utils/storage.rs index 061342d..c5a5434 100644 --- a/tests/src/utils/storage.rs +++ b/tests/src/utils/storage.rs @@ -148,6 +148,21 @@ async fn build_blob_store(typ: BlobStoreType, path: &str) -> BlobStore { path: path.to_string(), ..Default::default() }), + // inbuxa: scale-out storage (ST-16): three FileSystem members + BlobStoreType::Sharded => { + BlobStore::Sharded(registry::schema::structs::ShardedBlobStore { + stores: (1..=3) + .map(|n| { + let dir = format!("{path}/shard-{n}"); + std::fs::create_dir_all(&dir).unwrap(); + registry::schema::structs::BlobStoreBase::FileSystem(FileSystemStore { + path: dir, + ..Default::default() + }) + }) + .collect(), + }) + } _ => unreachable!(), } }