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!(), } }