feat: initial in-house cache distribution

This commit is contained in:
lewis
2026-02-08 10:03:53 +00:00
committed by Tangled
parent 240f24718e
commit 326aadcdb6
42 changed files with 4671 additions and 400 deletions
+19
View File
@@ -5,19 +5,38 @@ dir = "target/nextest"
retries = 0
fail-fast = true
test-threads = "num-cpus"
slow-timeout = { period = "30s", terminate-after = 4 }
[profile.ci]
retries = 2
fail-fast = false
test-threads = "num-cpus"
slow-timeout = { period = "30s", terminate-after = 4 }
[test-groups]
serial-env-tests = { max-threads = 1 }
heavy-load-tests = { max-threads = 4 }
[[profile.default.overrides]]
filter = "test(/import_with_verification/) | test(/plc_migration/)"
test-group = "serial-env-tests"
[[profile.default.overrides]]
filter = "binary(ripple_cluster)"
test-group = "serial-env-tests"
[[profile.default.overrides]]
filter = "binary(whole_story)"
test-group = "heavy-load-tests"
[[profile.ci.overrides]]
filter = "test(/import_with_verification/) | test(/plc_migration/)"
test-group = "serial-env-tests"
[[profile.ci.overrides]]
filter = "binary(ripple_cluster)"
test-group = "serial-env-tests"
[[profile.ci.overrides]]
filter = "binary(whole_story)"
test-group = "heavy-load-tests"
+10 -2
View File
@@ -41,10 +41,18 @@ BACKUP_STORAGE_PATH=/var/lib/tranquil/backups
# BACKUP_RETENTION_COUNT=7
# BACKUP_INTERVAL_SECS=86400
# =============================================================================
# Valkey (for caching and distributed rate limiting)
# Cache & Rate Limiting
# =============================================================================
# If not set, falls back to in-memory caching (single-node only)
# Ripple (in-process CRDT cache) is the default. No config needed for single-node.
# Set VALKEY_URL to use valkey instead (disables ripple).
# VALKEY_URL=redis://localhost:6379
#
# Ripple multi-node settings (only needed when clustering):
# RIPPLE_BIND=0.0.0.0:7890
# RIPPLE_PEERS=10.0.0.2:7890,10.0.0.3:7890
# RIPPLE_MACHINE_ID=1
# RIPPLE_GOSSIP_INTERVAL_MS=200
# RIPPLE_CACHE_MAX_MB=256
# =============================================================================
# Security Secrets
# =============================================================================
Generated
+70
View File
@@ -866,6 +866,26 @@ dependencies = [
"zeroize",
]
[[package]]
name = "bincode"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "36eaf5d7b090263e8150820482d5d93cd964a81e4019913c972f4edcc6edb740"
dependencies = [
"bincode_derive",
"serde",
"unty",
]
[[package]]
name = "bincode_derive"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf95709a440f45e986983918d0e8a1f30a9b1df04918fc828670606804ac3c09"
dependencies = [
"virtue",
]
[[package]]
name = "bitflags"
version = "2.10.0"
@@ -2015,6 +2035,19 @@ version = "1.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
[[package]]
name = "foca"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f59e967f3f675997e4a4a6b99d2a75148d59d64c46211b78b4f34ebb951b273"
dependencies = [
"bincode",
"bytes",
"rand 0.9.2",
"serde",
"tracing",
]
[[package]]
name = "foldhash"
version = "0.1.5"
@@ -5957,8 +5990,10 @@ dependencies = [
"async-trait",
"base64 0.22.1",
"redis",
"tokio-util",
"tracing",
"tranquil-infra",
"tranquil-ripple",
]
[[package]]
@@ -6131,6 +6166,7 @@ dependencies = [
"tranquil-db-traits",
"tranquil-oauth",
"tranquil-repo",
"tranquil-ripple",
"tranquil-scopes",
"tranquil-storage",
"tranquil-types",
@@ -6153,6 +6189,28 @@ dependencies = [
"sqlx",
]
[[package]]
name = "tranquil-ripple"
version = "0.1.0"
dependencies = [
"async-trait",
"backon",
"bincode",
"bytes",
"foca",
"futures",
"parking_lot",
"rand 0.9.2",
"serde",
"thiserror 2.0.17",
"tokio",
"tokio-util",
"tracing",
"tracing-subscriber",
"tranquil-infra",
"uuid",
]
[[package]]
name = "tranquil-scopes"
version = "0.1.0"
@@ -6300,6 +6358,12 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
[[package]]
name = "unty"
version = "0.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae"
[[package]]
name = "ureq"
version = "3.1.4"
@@ -6390,6 +6454,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
[[package]]
name = "virtue"
version = "0.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "051eb1abcf10076295e815102942cc58f9d5e3b4560e46e53c21e8ff6f3af7b1"
[[package]]
name = "vsimd"
version = "0.8.0"
+6 -1
View File
@@ -6,6 +6,7 @@ members = [
"crates/tranquil-crypto",
"crates/tranquil-storage",
"crates/tranquil-cache",
"crates/tranquil-ripple",
"crates/tranquil-repo",
"crates/tranquil-scopes",
"crates/tranquil-auth",
@@ -17,7 +18,7 @@ members = [
]
[workspace.package]
version = "0.1.0"
version = "0.2.0"
edition = "2024"
license = "AGPL-3.0-or-later"
@@ -34,9 +35,11 @@ tranquil-oauth = { path = "crates/tranquil-oauth" }
tranquil-comms = { path = "crates/tranquil-comms" }
tranquil-db-traits = { path = "crates/tranquil-db-traits" }
tranquil-db = { path = "crates/tranquil-db" }
tranquil-ripple = { path = "crates/tranquil-ripple" }
aes-gcm = "0.10"
backon = "1"
bincode = { version = "2", features = ["serde"] }
anyhow = "1.0"
async-trait = "0.1"
aws-config = "1.8"
@@ -51,6 +54,7 @@ chrono = { version = "0.4", features = ["serde"] }
cid = "0.11"
dotenvy = "0.15"
ed25519-dalek = { version = "2.1", features = ["pkcs8"] }
foca = { version = "1", features = ["bincode-codec", "tracing"] }
futures = "0.3"
futures-util = "0.3"
governor = "0.10"
@@ -70,6 +74,7 @@ k256 = { version = "0.13", features = ["ecdsa", "pem", "pkcs8"] }
metrics = "0.24"
metrics-exporter-prometheus = { version = "0.16", default-features = false, features = ["http-listener"] }
multibase = "0.9"
parking_lot = "0.12"
multihash = "0.19"
p256 = { version = "0.13", features = ["ecdsa"] }
p384 = { version = "0.13", features = ["ecdsa"] }
+1 -1
View File
@@ -14,7 +14,7 @@ Another excellent PDS is [Cocoon](https://tangled.org/hailey.at/cocoon), written
It is a superset of the reference PDS, including: passkeys and 2FA (WebAuthn/FIDO2, TOTP, backup codes, trusted devices), SSO login and signup, did:web support (PDS-hosted subdomains or bring-your-own), multi-channel communication (email, discord, telegram, signal) for verification and alerts, granular OAuth scopes with a consent UI showing human-readable descriptions, app passwords with granular permissions (read-only, post-only, or custom scopes), account delegation (letting others manage an account with configurable permission levels), automatic backups (configurable retention and frequency, one-click restore), and a built-in web UI for account management, OAuth consent, repo browsing, and admin.
The PDS itself is a single small binary with no node/npm runtime. It requires postgres and stores blobs on the local filesystem. Valkey is optional (enables distributed rate limiting for multi-node setups).
The PDS itself is a single small binary with no node/npm runtime. It requires postgres. Blobs are stored on the local filesystem by default (S3 optional). Valkey is optional (supported as an alternative to the built-in cache).
## Quick Start
+2
View File
@@ -6,8 +6,10 @@ license.workspace = true
[dependencies]
tranquil-infra = { workspace = true }
tranquil-ripple = { workspace = true }
async-trait = { workspace = true }
base64 = { workspace = true }
redis = { workspace = true }
tokio-util = { workspace = true }
tracing = { workspace = true }
+58 -28
View File
@@ -42,8 +42,8 @@ impl Cache for ValkeyCache {
redis::cmd("SET")
.arg(key)
.arg(value)
.arg("EX")
.arg(ttl.as_secs() as i64)
.arg("PX")
.arg(ttl.as_millis().min(i64::MAX as u128) as i64)
.query_async::<()>(&mut conn)
.await
.map_err(|e| CacheError::Connection(e.to_string()))
@@ -114,25 +114,35 @@ impl DistributedRateLimiter for RedisRateLimiter {
let mut conn = self.conn.clone();
let full_key = format!("rl:{}", key);
let window_secs = window_ms.div_ceil(1000).max(1) as i64;
let count: Result<i64, _> = redis::cmd("INCR")
.arg(&full_key)
.query_async(&mut conn)
.await;
let count = match count {
Ok(c) => c,
let result: Result<i64, _> = redis::Script::new(
r"local c = redis.call('INCR', KEYS[1])
if c == 1 then redis.call('EXPIRE', KEYS[1], ARGV[1]) end
if redis.call('TTL', KEYS[1]) == -1 then redis.call('EXPIRE', KEYS[1], ARGV[1]) end
return c"
)
.key(&full_key)
.arg(window_secs)
.invoke_async(&mut conn)
.await;
match result {
Ok(count) => count <= limit as i64,
Err(e) => {
tracing::warn!("Redis rate limit INCR failed: {}. Allowing request.", e);
return true;
tracing::warn!(error = %e, "redis rate limit script failed, allowing request");
true
}
};
if count == 1 {
let _: Result<bool, redis::RedisError> = redis::cmd("EXPIRE")
.arg(&full_key)
.arg(window_secs)
.query_async(&mut conn)
.await;
}
count <= limit as i64
}
async fn peek_rate_limit_count(&self, key: &str, _window_ms: u64) -> u64 {
let mut conn = self.conn.clone();
let full_key = format!("rl:{}", key);
redis::cmd("GET")
.arg(&full_key)
.query_async::<Option<u64>>(&mut conn)
.await
.ok()
.flatten()
.unwrap_or(0)
}
}
@@ -145,21 +155,41 @@ impl DistributedRateLimiter for NoOpRateLimiter {
}
}
pub async fn create_cache() -> (Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>) {
match std::env::var("VALKEY_URL") {
Ok(url) => match ValkeyCache::new(&url).await {
pub async fn create_cache(
shutdown: tokio_util::sync::CancellationToken,
) -> (Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>) {
if let Ok(url) = std::env::var("VALKEY_URL") {
match ValkeyCache::new(&url).await {
Ok(cache) => {
tracing::info!("Connected to Valkey cache at {}", url);
tracing::info!("using valkey cache at {url}");
let rate_limiter = Arc::new(RedisRateLimiter::new(cache.connection()));
(Arc::new(cache), rate_limiter)
return (Arc::new(cache), rate_limiter);
}
Err(e) => {
tracing::warn!("Failed to connect to Valkey: {}. Running without cache.", e);
(Arc::new(NoOpCache), Arc::new(NoOpRateLimiter))
tracing::warn!("failed to connect to valkey: {e}. falling back to ripple.");
}
},
Err(_) => {
tracing::info!("VALKEY_URL not set. Running without cache.");
}
}
match tranquil_ripple::RippleConfig::from_env() {
Ok(config) => {
let peer_count = config.seed_peers.len();
match tranquil_ripple::RippleEngine::start(config, shutdown).await {
Ok((cache, rate_limiter, _bound_addr)) => {
match peer_count {
0 => tracing::info!("ripple cache started (single-node)"),
n => tracing::info!("ripple cache started ({n} seed peers)"),
}
(cache, rate_limiter)
}
Err(e) => {
tracing::error!("ripple engine failed to start: {e}. running without cache.");
(Arc::new(NoOpCache), Arc::new(NoOpRateLimiter))
}
}
}
Err(e) => {
tracing::error!("ripple config error: {e}. running without cache.");
(Arc::new(NoOpCache), Arc::new(NoOpRateLimiter))
}
}
+3
View File
@@ -81,4 +81,7 @@ pub trait Cache: Send + Sync {
#[async_trait]
pub trait DistributedRateLimiter: Send + Sync {
async fn check_rate_limit(&self, key: &str, limit: u32, window_ms: u64) -> bool;
async fn peek_rate_limit_count(&self, _key: &str, _window_ms: u64) -> u64 {
0
}
}
+1
View File
@@ -89,4 +89,5 @@ ciborium = { workspace = true }
ctor = { workspace = true }
testcontainers = { workspace = true }
testcontainers-modules = { workspace = true }
tranquil-ripple = { workspace = true }
wiremock = { workspace = true }
+23 -16
View File
@@ -30,14 +30,14 @@ fn detect_mime_type(data: &[u8], client_hint: &str) -> String {
);
}
detected
} else if client_hint == "*/*" || client_hint.is_empty() {
warn!(
"Could not detect MIME type and client sent invalid hint: '{}'",
client_hint
);
"application/octet-stream".to_string()
} else {
client_hint.to_string()
match client_hint {
"" | "*/*" => "application/octet-stream".to_string(),
hint if hint.starts_with("text/html") || hint.starts_with("application/xhtml") => {
"application/octet-stream".to_string()
}
hint => hint.to_string(),
}
}
}
@@ -85,8 +85,7 @@ pub async fn upload_blob(
.user_repo
.get_id_by_did(&did)
.await
.ok()
.flatten()
.log_db_err("fetching user id for blob upload")?
.ok_or(ApiError::InternalError(None))?;
let temp_key = format!("temp/{}", uuid::Uuid::new_v4());
@@ -136,7 +135,10 @@ pub async fn upload_blob(
};
let cid = Cid::new_v1(0x55, multihash);
let cid_str = cid.to_string();
let cid_link: CidLink = unsafe { CidLink::new_unchecked(&cid_str) };
let cid_link: CidLink = CidLink::new(&cid_str).map_err(|e| {
error!("Failed to construct CidLink from computed CID: {:?}", e);
ApiError::InternalError(Some("Failed to construct CID".into()))
})?;
let storage_key = cid_str.clone();
info!(
@@ -144,13 +146,12 @@ pub async fn upload_blob(
size, cid_str
);
let was_inserted = match state
match state
.blob_repo
.insert_blob(&cid_link, &mime_type, size as i64, user_id, &storage_key)
.await
{
Ok(Some(_)) => true,
Ok(None) => false,
Ok(_) => {}
Err(e) => {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to insert blob record: {:?}", e);
@@ -158,8 +159,11 @@ pub async fn upload_blob(
}
};
if was_inserted && let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await {
if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await {
let _ = state.blob_store.delete(&temp_key).await;
if let Err(db_err) = state.blob_repo.delete_blob_by_cid(&cid_link).await {
error!("Failed to clean up orphaned blob record after copy failure: {:?}", db_err);
}
error!("Failed to copy blob to final location: {:?}", e);
return Err(ApiError::InternalError(Some("Failed to store blob".into())));
}
@@ -167,7 +171,7 @@ pub async fn upload_blob(
let _ = state.blob_store.delete(&temp_key).await;
if let Some(ref controller) = controller_did {
let _ = state
if let Err(e) = state
.delegation_repo
.log_delegation_action(
&did,
@@ -182,7 +186,10 @@ pub async fn upload_blob(
None,
None,
)
.await;
.await
{
warn!("Failed to log delegation action for blob upload: {:?}", e);
}
}
Ok(Json(json!({
+40 -16
View File
@@ -13,6 +13,7 @@ use crate::util::pds_hostname;
use sqlx::PgPool;
use std::error::Error;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::broadcast;
use tokio_util::sync::CancellationToken;
use tranquil_db::{
@@ -21,6 +22,16 @@ use tranquil_db::{
SsoRepository, UserRepository,
};
static RATE_LIMITING_DISABLED: AtomicBool = AtomicBool::new(false);
pub fn init_rate_limit_override() {
let disabled = std::env::var("DISABLE_RATE_LIMITING").is_ok();
RATE_LIMITING_DISABLED.store(disabled, Ordering::Relaxed);
if disabled {
tracing::warn!("rate limiting is DISABLED via DISABLE_RATE_LIMITING env var");
}
}
#[derive(Clone)]
pub struct AppState {
pub repos: Arc<PostgresRepositories>,
@@ -173,6 +184,7 @@ impl AppState {
pub async fn from_db(db: PgPool, shutdown: CancellationToken) -> Self {
AuthConfig::init();
init_rate_limit_override();
let repos = Arc::new(PostgresRepositories::new(db.clone()));
let block_store = PostgresBlockStore::new(db);
@@ -188,7 +200,7 @@ impl AppState {
let rate_limiters = Arc::new(RateLimiters::new());
let repo_write_locks = Arc::new(RepoWriteLocks::new());
let circuit_breakers = Arc::new(CircuitBreakers::new());
let (cache, distributed_rate_limiter) = create_cache().await;
let (cache, distributed_rate_limiter) = create_cache(shutdown.clone()).await;
let did_resolver = Arc::new(DidResolver::new());
let sso_config = SsoConfig::init();
let sso_manager = SsoManager::from_config(sso_config);
@@ -231,28 +243,27 @@ impl AppState {
self
}
pub fn with_cache(
mut self,
cache: Arc<dyn Cache>,
distributed_rate_limiter: Arc<dyn DistributedRateLimiter>,
) -> Self {
self.cache = cache;
self.distributed_rate_limiter = distributed_rate_limiter;
self
}
pub fn with_circuit_breakers(mut self, circuit_breakers: CircuitBreakers) -> Self {
self.circuit_breakers = Arc::new(circuit_breakers);
self
}
pub async fn check_rate_limit(&self, kind: RateLimitKind, client_ip: &str) -> bool {
if std::env::var("DISABLE_RATE_LIMITING").is_ok() {
if RATE_LIMITING_DISABLED.load(Ordering::Relaxed) {
return true;
}
let key = format!("{}:{}", kind.key_prefix(), client_ip);
let limiter_name = kind.key_prefix();
let (limit, window_ms) = kind.limit_and_window_ms();
if !self
.distributed_rate_limiter
.check_rate_limit(&key, limit, window_ms)
.await
{
crate::metrics::record_rate_limit_rejection(limiter_name);
return false;
}
let limiter = match kind {
RateLimitKind::Login => &self.rate_limiters.login,
@@ -277,10 +288,23 @@ impl AppState {
RateLimitKind::HandleVerification => &self.rate_limiters.handle_verification,
};
let ok = limiter.check_key(&client_ip.to_string()).is_ok();
if !ok {
if limiter.check_key(&client_ip.to_string()).is_err() {
crate::metrics::record_rate_limit_rejection(limiter_name);
return false;
}
ok
let key = format!("{}:{}", kind.key_prefix(), client_ip);
let (limit, window_ms) = kind.limit_and_window_ms();
if !self
.distributed_rate_limiter
.check_rate_limit(&key, limit, window_ms)
.await
{
crate::metrics::record_rate_limit_rejection(limiter_name);
return false;
}
true
}
}
+1 -1
View File
@@ -12,7 +12,7 @@ async fn test_search_accounts_as_admin() {
let (user_did, _) = setup_new_user("search-target").await;
let mut found = false;
let mut cursor: Option<String> = None;
for _ in 0..10 {
for _ in 0..100 {
let url = match &cursor {
Some(c) => format!(
"{}/xrpc/com.atproto.admin.searchAccounts?limit=100&cursor={}",
+272 -74
View File
@@ -9,12 +9,14 @@ use reqwest::{Client, StatusCode, header};
use serde_json::{Value, json};
use sqlx::postgres::PgPoolOptions;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock, RwLock};
#[allow(unused_imports)]
use std::time::Duration;
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tranquil_pds::cache::{Cache, DistributedRateLimiter};
use tranquil_pds::state::AppState;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
@@ -25,6 +27,22 @@ static MOCK_APPVIEW: OnceLock<MockServer> = OnceLock::new();
static MOCK_PLC: OnceLock<MockServer> = OnceLock::new();
static TEST_DB_POOL: OnceLock<sqlx::PgPool> = OnceLock::new();
static TEST_TEMP_DIR: OnceLock<PathBuf> = OnceLock::new();
static CLUSTER: OnceLock<Vec<ServerInstance>> = OnceLock::new();
#[allow(dead_code)]
pub struct ServerConfig {
pub pool: sqlx::PgPool,
pub cache: Option<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>)>,
}
#[allow(dead_code)]
#[derive(Clone)]
pub struct ServerInstance {
pub url: String,
pub port: u16,
pub cache: Option<Arc<dyn Cache>>,
pub distributed_rate_limiter: Option<Arc<dyn DistributedRateLimiter>>,
}
#[cfg(all(not(feature = "external-infra"), feature = "s3-storage"))]
use testcontainers::GenericImage;
@@ -139,45 +157,10 @@ async fn setup_with_external_infra() -> String {
std::env::var("DATABASE_URL").expect("DATABASE_URL must be set when using external infra");
let plc_url = setup_mock_plc_directory().await;
unsafe {
if std::env::var("S3_ENDPOINT").is_ok() {
let s3_endpoint = std::env::var("S3_ENDPOINT").unwrap();
std::env::set_var("BLOB_STORAGE_BACKEND", "s3");
std::env::set_var("BACKUP_STORAGE_BACKEND", "s3");
std::env::set_var("BACKUP_S3_BUCKET", "test-backups");
std::env::set_var(
"S3_BUCKET",
std::env::var("S3_BUCKET").unwrap_or_else(|_| "test-bucket".to_string()),
);
std::env::set_var(
"AWS_ACCESS_KEY_ID",
std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_else(|_| "minioadmin".to_string()),
);
std::env::set_var(
"AWS_SECRET_ACCESS_KEY",
std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_else(|_| "minioadmin".to_string()),
);
std::env::set_var(
"AWS_REGION",
std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string()),
);
std::env::set_var("S3_ENDPOINT", &s3_endpoint);
} else if std::env::var("BLOB_STORAGE_PATH").is_ok() {
std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
} else {
panic!("Either S3_ENDPOINT or BLOB_STORAGE_PATH must be set for external-infra");
}
std::env::set_var("MAX_IMPORT_SIZE", "100000000");
std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
configure_external_storage_env();
std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
}
let mock_server = MockServer::start().await;
setup_mock_appview(&mock_server).await;
let mock_uri = mock_server.uri();
let mock_host = mock_uri.strip_prefix("http://").unwrap_or(&mock_uri);
let mock_did = format!("did:web:{}", mock_host.replace(':', "%3A"));
setup_mock_did_document(&mock_server, &mock_did, &mock_uri).await;
MOCK_APPVIEW.set(mock_server).ok();
register_mock_appview().await;
spawn_app(database_url).await
}
@@ -199,13 +182,7 @@ async fn setup_with_testcontainers() -> String {
std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
}
let mock_server = MockServer::start().await;
setup_mock_appview(&mock_server).await;
let mock_uri = mock_server.uri();
let mock_host = mock_uri.strip_prefix("http://").unwrap_or(&mock_uri);
let mock_did = format!("did:web:{}", mock_host.replace(':', "%3A"));
setup_mock_did_document(&mock_server, &mock_did, &mock_uri).await;
MOCK_APPVIEW.set(mock_server).ok();
register_mock_appview().await;
let container = Postgres::default()
.with_tag("18-alpine")
.with_label("tranquil_pds_test", "true")
@@ -275,13 +252,7 @@ async fn setup_with_testcontainers() -> String {
.bucket("test-backups")
.send()
.await;
let mock_server = MockServer::start().await;
setup_mock_appview(&mock_server).await;
let mock_uri = mock_server.uri();
let mock_host = mock_uri.strip_prefix("http://").unwrap_or(&mock_uri);
let mock_did = format!("did:web:{}", mock_host.replace(':', "%3A"));
setup_mock_did_document(&mock_server, &mock_did, &mock_uri).await;
MOCK_APPVIEW.set(mock_server).ok();
register_mock_appview().await;
S3_CONTAINER.set(s3_container).ok();
let container = Postgres::default()
.with_tag("18-alpine")
@@ -324,6 +295,60 @@ async fn setup_mock_did_document(mock_server: &MockServer, did: &str, service_en
async fn setup_mock_appview(_mock_server: &MockServer) {}
async fn register_mock_appview() {
let mock_server = MockServer::start().await;
setup_mock_appview(&mock_server).await;
let mock_uri = mock_server.uri();
let mock_host = mock_uri.strip_prefix("http://").unwrap_or(&mock_uri);
let mock_did = format!("did:web:{}", mock_host.replace(':', "%3A"));
setup_mock_did_document(&mock_server, &mock_did, &mock_uri).await;
MOCK_APPVIEW.set(mock_server).ok();
}
unsafe fn configure_external_storage_env() {
unsafe {
if std::env::var("S3_ENDPOINT").is_ok() {
let s3_endpoint = std::env::var("S3_ENDPOINT").unwrap();
std::env::set_var("BLOB_STORAGE_BACKEND", "s3");
std::env::set_var("BACKUP_STORAGE_BACKEND", "s3");
std::env::set_var("BACKUP_S3_BUCKET", "test-backups");
std::env::set_var(
"S3_BUCKET",
std::env::var("S3_BUCKET").unwrap_or_else(|_| "test-bucket".to_string()),
);
std::env::set_var(
"AWS_ACCESS_KEY_ID",
std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_else(|_| "minioadmin".to_string()),
);
std::env::set_var(
"AWS_SECRET_ACCESS_KEY",
std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_else(|_| "minioadmin".to_string()),
);
std::env::set_var(
"AWS_REGION",
std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string()),
);
std::env::set_var("S3_ENDPOINT", &s3_endpoint);
} else {
let process_dir = std::env::temp_dir().join(format!(
"tranquil-pds-test-{}",
std::process::id()
));
let blob_path = process_dir.join("blobs");
let backup_path = process_dir.join("backups");
std::fs::create_dir_all(&blob_path).expect("Failed to create blob directory");
std::fs::create_dir_all(&backup_path).expect("Failed to create backup directory");
TEST_TEMP_DIR.set(process_dir).ok();
std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
}
std::env::set_var("MAX_IMPORT_SIZE", "100000000");
std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
}
}
type PlcOperationStore = Arc<RwLock<HashMap<String, Value>>>;
struct PlcPostResponder {
@@ -515,8 +540,44 @@ async fn setup_mock_plc_directory() -> String {
plc_url
}
async fn spawn_app(database_url: String) -> String {
async fn spawn_server(config: ServerConfig) -> ServerInstance {
use tranquil_pds::rate_limit::RateLimiters;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
unsafe {
std::env::set_var("PDS_HOSTNAME", format!("pds.test:{}", addr.port()));
}
let rate_limiters = RateLimiters::new()
.with_login_limit(10000)
.with_account_creation_limit(10000)
.with_password_reset_limit(10000)
.with_email_update_limit(10000)
.with_oauth_authorize_limit(10000)
.with_oauth_token_limit(10000);
let cache_refs = config.cache.as_ref().map(|(c, r)| (c.clone(), r.clone()));
let mut state = AppState::from_db(config.pool, CancellationToken::new())
.await
.with_rate_limiters(rate_limiters);
if let Some((cache, distributed_rate_limiter)) = config.cache {
state = state.with_cache(cache, distributed_rate_limiter);
}
tranquil_pds::sync::listener::start_sequencer_listener(state.clone()).await;
let app = tranquil_pds::app(state);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let (cache, distributed_rate_limiter) = cache_refs
.map(|(c, r)| (Some(c), Some(r)))
.unwrap_or((None, None));
ServerInstance {
url: format!("http://localhost:{}", addr.port()),
port: addr.port(),
cache,
distributed_rate_limiter,
}
}
async fn spawn_app(database_url: String) -> String {
let pool = PgPoolOptions::new()
.max_connections(10)
.acquire_timeout(std::time::Duration::from_secs(30))
@@ -528,34 +589,171 @@ async fn spawn_app(database_url: String) -> String {
.await
.expect("Failed to run migrations");
let test_pool = PgPoolOptions::new()
.max_connections(5)
.max_connections(2)
.acquire_timeout(std::time::Duration::from_secs(30))
.connect(&database_url)
.await
.expect("Failed to create test pool");
TEST_DB_POOL.set(test_pool).ok();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
APP_PORT.set(addr.port()).ok();
unsafe {
std::env::set_var("PDS_HOSTNAME", format!("pds.test:{}", addr.port()));
}
let rate_limiters = RateLimiters::new()
.with_login_limit(10000)
.with_account_creation_limit(10000)
.with_password_reset_limit(10000)
.with_email_update_limit(10000)
.with_oauth_authorize_limit(10000)
.with_oauth_token_limit(10000);
let state = AppState::from_db(pool, CancellationToken::new())
let instance = spawn_server(ServerConfig { pool, cache: None }).await;
APP_PORT.set(instance.port).ok();
instance.url
}
#[allow(dead_code)]
pub async fn spawn_cluster(database_url: String, node_count: usize) -> Vec<ServerInstance> {
use tranquil_ripple::{RippleConfig, RippleEngine};
let pool = PgPoolOptions::new()
.max_connections(10)
.acquire_timeout(std::time::Duration::from_secs(30))
.connect(&database_url)
.await
.with_rate_limiters(rate_limiters);
tranquil_pds::sync::listener::start_sequencer_listener(state.clone()).await;
let app = tranquil_pds::app(state);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
format!("http://localhost:{}", addr.port())
.expect("Failed to connect to Postgres for cluster");
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("Failed to run migrations for cluster");
let test_pool = PgPoolOptions::new()
.max_connections(2)
.acquire_timeout(std::time::Duration::from_secs(30))
.connect(&database_url)
.await
.expect("Failed to create test pool for cluster");
TEST_DB_POOL.set(test_pool).ok();
let shutdown = CancellationToken::new();
let mut ripple_nodes: Vec<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>)> =
Vec::with_capacity(node_count);
let mut bound_addrs: Vec<SocketAddr> = Vec::with_capacity(node_count);
for i in 0..node_count {
let config = RippleConfig {
bind_addr: "127.0.0.1:0".parse().unwrap(),
seed_peers: bound_addrs.clone(),
machine_id: i as u64 + 1,
gossip_interval_ms: 100,
cache_max_bytes: 64 * 1024 * 1024,
};
let (cache, rate_limiter, addr) = RippleEngine::start(config, shutdown.clone())
.await
.expect("failed to start ripple node");
bound_addrs.push(addr);
ripple_nodes.push((cache, rate_limiter));
}
let mut instances: Vec<ServerInstance> = Vec::with_capacity(node_count);
for (cache, rate_limiter) in ripple_nodes {
let server_config = ServerConfig {
pool: pool.clone(),
cache: Some((cache, rate_limiter)),
};
let instance = spawn_server(server_config).await;
instances.push(instance);
}
let first = &instances[0];
APP_PORT.set(first.port).ok();
tokio::time::sleep(Duration::from_millis(2000)).await;
instances
}
#[allow(dead_code)]
pub async fn cluster() -> &'static [ServerInstance] {
CLUSTER.get_or_init(|| {
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
unsafe {
std::env::set_var("TRANQUIL_PDS_ALLOW_INSECURE_SECRETS", "1");
}
if std::env::var("DOCKER_HOST").is_err()
&& let Ok(runtime_dir) = std::env::var("XDG_RUNTIME_DIR")
{
let podman_sock = std::path::Path::new(&runtime_dir).join("podman/podman.sock");
if podman_sock.exists() {
unsafe {
std::env::set_var(
"DOCKER_HOST",
format!("unix://{}", podman_sock.display()),
);
}
}
}
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async move {
unsafe {
std::env::remove_var("DISABLE_RATE_LIMITING");
}
let database_url = if has_external_infra() {
setup_cluster_external_infra().await
} else {
setup_cluster_testcontainers().await
};
let nodes = spawn_cluster(database_url, 3).await;
tx.send(nodes).unwrap();
std::future::pending::<()>().await;
});
});
rx.recv().expect("Failed to start test cluster")
})
}
async fn setup_cluster_external_infra() -> String {
let database_url =
std::env::var("DATABASE_URL").expect("DATABASE_URL must be set when using external infra");
let plc_url = setup_mock_plc_directory().await;
unsafe {
configure_external_storage_env();
std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
}
register_mock_appview().await;
database_url
}
#[cfg(not(feature = "external-infra"))]
async fn setup_cluster_testcontainers() -> String {
let temp_dir = std::env::temp_dir().join(format!("tranquil-pds-cluster-{}", uuid::Uuid::new_v4()));
let blob_path = temp_dir.join("blobs");
let backup_path = temp_dir.join("backups");
std::fs::create_dir_all(&blob_path).expect("Failed to create blob temp directory");
std::fs::create_dir_all(&backup_path).expect("Failed to create backup temp directory");
TEST_TEMP_DIR.set(temp_dir).ok();
let plc_url = setup_mock_plc_directory().await;
unsafe {
std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
std::env::set_var("MAX_IMPORT_SIZE", "100000000");
std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
}
register_mock_appview().await;
let container = Postgres::default()
.with_tag("18-alpine")
.with_label("tranquil_pds_test", "true")
.start()
.await
.expect("Failed to start Postgres for cluster");
let connection_string = format!(
"postgres://postgres:postgres@127.0.0.1:{}",
container
.get_host_port_ipv4(5432)
.await
.expect("Failed to get port")
);
DB_CONTAINER.set(container).ok();
connection_string
}
#[cfg(feature = "external-infra")]
async fn setup_cluster_testcontainers() -> String {
panic!(
"Testcontainers disabled with external-infra feature. Set DATABASE_URL and BLOB_STORAGE_PATH (or S3_ENDPOINT)."
);
}
#[allow(dead_code)]
@@ -800,7 +800,12 @@ async fn test_firehose_outdated_cursor_info() {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let outdated_cursor = 1i64;
let pool = get_test_db_pool().await;
let max_seq: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let outdated_cursor = (max_seq - 100).max(1);
let url = format!(
"ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}",
app_port(),
+4 -2
View File
@@ -25,6 +25,7 @@ async fn test_upload_blob_no_auth() {
async fn test_upload_blob_success() {
let client = client();
let (token, _) = create_account_and_login(&client).await;
let blob_data = format!("blob-{}", uuid::Uuid::new_v4());
let res = client
.post(format!(
"{}/xrpc/com.atproto.repo.uploadBlob",
@@ -32,12 +33,13 @@ async fn test_upload_blob_success() {
))
.header(header::CONTENT_TYPE, "text/plain")
.bearer_auth(token)
.body("This is our blob data")
.body(blob_data)
.send()
.await
.expect("Failed to send request");
assert_eq!(res.status(), StatusCode::OK);
let status = res.status();
let body: Value = res.json().await.expect("Response was not valid JSON");
assert_eq!(status, StatusCode::OK, "uploadBlob failed: {body}");
assert!(body["blob"]["ref"]["$link"].as_str().is_some());
}
File diff suppressed because it is too large Load Diff
+27 -14
View File
@@ -160,24 +160,37 @@ async fn test_list_repos_shows_status_field() {
set_account_takedown(&did, Some("test-takedown-ref")).await;
let res = client
.get(format!(
let mut cursor: Option<String> = None;
let mut takendown_repo: Option<Value> = None;
loop {
let mut url = format!(
"{}/xrpc/com.atproto.sync.listRepos?limit=1000",
base_url().await
))
.send()
.await
.expect("Failed to send request");
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Response was not valid JSON");
let repos = body["repos"].as_array().unwrap();
let takendown_repo = repos.iter().find(|r| r["did"] == did);
);
if let Some(ref c) = cursor {
url.push_str(&format!("&cursor={}", c));
}
let res = client
.get(&url)
.send()
.await
.expect("Failed to send request");
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Response was not valid JSON");
let repos = body["repos"].as_array().unwrap();
if let Some(found) = repos.iter().find(|r| r["did"] == did) {
takendown_repo = Some(found.clone());
break;
}
match body["cursor"].as_str() {
Some(c) => cursor = Some(c.to_string()),
None => break,
}
}
assert!(takendown_repo.is_some(), "Takendown repo should be in list");
let repo = takendown_repo.unwrap();
assert_eq!(repo["active"], false);
assert_eq!(repo["status"], "takendown");
assert_eq!(repo["active"], false, "repo should be inactive: {:?}", repo);
assert_eq!(repo["status"], "takendown", "repo status should be takendown: {:?}", repo);
}
#[tokio::test]
+64 -64
View File
@@ -486,7 +486,8 @@ async fn test_blob_lifecycle_upload_use_remove() {
let base = base_url().await;
let (did, jwt) = setup_new_user("blob-lifecycle").await;
let blob1_data = b"First blob for testing lifecycle";
let blob1_data = format!("First blob for testing lifecycle {}", uuid::Uuid::new_v4());
let blob1_data = blob1_data.as_bytes();
let upload1_res = client
.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", base))
.header(header::CONTENT_TYPE, "text/plain")
@@ -500,7 +501,8 @@ async fn test_blob_lifecycle_upload_use_remove() {
let blob1 = upload1_body["blob"].clone();
let blob1_cid = blob1["ref"]["$link"].as_str().unwrap();
let blob2_data = b"Second blob for testing lifecycle";
let blob2_data = format!("Second blob for testing lifecycle {}", uuid::Uuid::new_v4());
let blob2_data = blob2_data.as_bytes();
let upload2_res = client
.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", base))
.header(header::CONTENT_TYPE, "text/plain")
@@ -1278,7 +1280,7 @@ async fn test_scale_100_posts_with_pagination() {
let (did, jwt) = setup_new_user("scale-posts").await;
let post_count = 1000;
let post_futures: Vec<_> = (0..post_count)
futures::stream::iter(0..post_count)
.map(|i| {
let client = client.clone();
let base = base.to_string();
@@ -1311,9 +1313,9 @@ async fn test_scale_100_posts_with_pagination() {
);
}
})
.collect();
join_all(post_futures).await;
.buffer_unordered(50)
.collect::<Vec<()>>()
.await;
let count_res = client
.get(format!("{}/xrpc/com.atproto.repo.listRecords", base))
@@ -1349,9 +1351,7 @@ async fn test_scale_100_posts_with_pagination() {
"All posts should have unique URIs"
);
let delete_futures: Vec<_> = all_uris
.iter()
.take(500)
futures::stream::iter(all_uris.iter().take(500))
.map(|uri| {
let client = client.clone();
let base = base.to_string();
@@ -1373,9 +1373,9 @@ async fn test_scale_100_posts_with_pagination() {
assert_eq!(res.status(), StatusCode::OK);
}
})
.collect();
join_all(delete_futures).await;
.buffer_unordered(50)
.collect::<Vec<()>>()
.await;
let final_count = count_records(&client, base, &jwt, &did, "app.bsky.feed.post").await;
assert_eq!(
@@ -1396,54 +1396,51 @@ async fn test_scale_many_users_social_graph() {
let users: Vec<(String, String)> = join_all(user_futures).await;
let follow_futures: Vec<_> = users
let follow_pairs: Vec<(String, String, String)> = users
.iter()
.enumerate()
.flat_map(|(i, (follower_did, follower_jwt))| {
let client = client.clone();
let base = base.to_string();
users.iter().enumerate().filter(move |(j, _)| *j != i).map({
let client = client.clone();
let base = base.clone();
let follower_did = follower_did.clone();
let follower_jwt = follower_jwt.clone();
move |(_, (followee_did, _))| {
let client = client.clone();
let base = base.clone();
let follower_did = follower_did.clone();
let follower_jwt = follower_jwt.clone();
let followee_did = followee_did.clone();
async move {
let rkey = format!(
"follow_{}",
&uuid::Uuid::new_v4().simple().to_string()[..12]
);
let res = client
.post(format!("{}/xrpc/com.atproto.repo.putRecord", base))
.bearer_auth(&follower_jwt)
.json(&json!({
"repo": follower_did,
"collection": "app.bsky.graph.follow",
"rkey": rkey,
"record": {
"$type": "app.bsky.graph.follow",
"subject": followee_did,
"createdAt": Utc::now().to_rfc3339()
}
}))
.send()
.await
.expect("Follow failed");
let status = res.status();
let body: Value = res.json().await.unwrap_or_default();
assert_eq!(status, StatusCode::OK, "Follow failed: {:?}", body);
}
}
})
users.iter().enumerate()
.filter(move |(j, _)| *j != i)
.map(|(_, (followee_did, _))| {
(follower_did.clone(), follower_jwt.clone(), followee_did.clone())
})
.collect::<Vec<_>>()
})
.collect();
join_all(follow_futures).await;
futures::stream::iter(follow_pairs)
.map(|(follower_did, follower_jwt, followee_did)| {
let client = client.clone();
let base = base.to_string();
async move {
let rkey = format!(
"follow_{}",
&uuid::Uuid::new_v4().simple().to_string()[..12]
);
let res = client
.post(format!("{}/xrpc/com.atproto.repo.putRecord", base))
.bearer_auth(&follower_jwt)
.json(&json!({
"repo": follower_did,
"collection": "app.bsky.graph.follow",
"rkey": rkey,
"record": {
"$type": "app.bsky.graph.follow",
"subject": followee_did,
"createdAt": Utc::now().to_rfc3339()
}
}))
.send()
.await
.expect("Follow failed");
let status = res.status();
let body: Value = res.json().await.unwrap_or_default();
assert_eq!(status, StatusCode::OK, "Follow failed: {:?}", body);
}
})
.buffer_unordered(50)
.collect::<Vec<()>>()
.await;
let expected_follows_per_user = user_count - 1;
let verify_futures: Vec<_> = users
@@ -1529,13 +1526,13 @@ async fn test_scale_many_blobs_in_repo() {
let (did, jwt) = setup_new_user("scale-blobs").await;
let blob_count = 300;
let blob_futures: Vec<_> = (0..blob_count)
let blobs: Vec<Value> = futures::stream::iter(0..blob_count)
.map(|i| {
let client = client.clone();
let base = base.to_string();
let jwt = jwt.clone();
async move {
let blob_data = format!("Blob data number {} with some padding to make it realistic size for testing purposes", i);
let blob_data = format!("Blob data number {} {} with some padding to make it realistic size for testing purposes", i, uuid::Uuid::new_v4());
let res = client
.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", base))
.header(header::CONTENT_TYPE, "text/plain")
@@ -1549,13 +1546,16 @@ async fn test_scale_many_blobs_in_repo() {
body["blob"].clone()
}
})
.collect();
.buffer_unordered(50)
.collect::<Vec<Value>>()
.await;
let blobs: Vec<Value> = join_all(blob_futures).await;
let post_futures: Vec<_> = blobs
let blob_chunks: Vec<(usize, Vec<Value>)> = blobs
.chunks(3)
.enumerate()
.map(|(i, chunk)| (i, chunk.to_vec()))
.collect();
futures::stream::iter(blob_chunks)
.map(|(i, blob_chunk)| {
let client = client.clone();
let base = base.to_string();
@@ -1596,9 +1596,9 @@ async fn test_scale_many_blobs_in_repo() {
assert_eq!(status, StatusCode::OK, "Post with blobs failed: {:?}", body);
}
})
.collect();
join_all(post_futures).await;
.buffer_unordered(50)
.collect::<Vec<()>>()
.await;
let list_blobs_res = client
.get(format!("{}/xrpc/com.atproto.sync.listBlobs", base))
+27
View File
@@ -0,0 +1,27 @@
[package]
name = "tranquil-ripple"
version.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
tranquil-infra = { workspace = true }
async-trait = { workspace = true }
backon = { workspace = true }
bincode = { workspace = true }
bytes = { workspace = true }
foca = { workspace = true }
parking_lot = { workspace = true }
rand = "0.9"
serde = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "io-util", "sync", "time"] }
tokio-util = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
futures = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
tracing-subscriber = { workspace = true }
uuid = { workspace = true }
+89
View File
@@ -0,0 +1,89 @@
use crate::crdt::CrdtStore;
use async_trait::async_trait;
use parking_lot::RwLock;
use std::sync::Arc;
use std::time::Duration;
use tranquil_infra::{Cache, CacheError};
pub struct RippleCache {
store: Arc<RwLock<CrdtStore>>,
}
impl RippleCache {
pub fn new(store: Arc<RwLock<CrdtStore>>) -> Self {
Self { store }
}
}
#[async_trait]
impl Cache for RippleCache {
async fn get(&self, key: &str) -> Option<String> {
self.store
.read()
.cache_get(key)
.and_then(|bytes| String::from_utf8(bytes).ok())
}
async fn set(&self, key: &str, value: &str, ttl: Duration) -> Result<(), CacheError> {
self.store
.write()
.cache_set(key.to_string(), value.as_bytes().to_vec(), ttl.as_millis() as u64);
Ok(())
}
async fn delete(&self, key: &str) -> Result<(), CacheError> {
self.store.write().cache_delete(key);
Ok(())
}
async fn get_bytes(&self, key: &str) -> Option<Vec<u8>> {
self.store.read().cache_get(key)
}
async fn set_bytes(&self, key: &str, value: &[u8], ttl: Duration) -> Result<(), CacheError> {
self.store
.write()
.cache_set(key.to_string(), value.to_vec(), ttl.as_millis() as u64);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn cache_trait_roundtrip() {
let store = Arc::new(RwLock::new(CrdtStore::new(1)));
let cache = RippleCache::new(store);
cache
.set("test", "value", Duration::from_secs(60))
.await
.unwrap();
assert_eq!(cache.get("test").await, Some("value".to_string()));
}
#[tokio::test]
async fn cache_trait_bytes() {
let store = Arc::new(RwLock::new(CrdtStore::new(1)));
let cache = RippleCache::new(store);
let data = vec![0xDE, 0xAD, 0xBE, 0xEF];
cache
.set_bytes("bin", &data, Duration::from_secs(60))
.await
.unwrap();
assert_eq!(cache.get_bytes("bin").await, Some(data));
}
#[tokio::test]
async fn cache_trait_delete() {
let store = Arc::new(RwLock::new(CrdtStore::new(1)));
let cache = RippleCache::new(store);
cache
.set("del", "x", Duration::from_secs(60))
.await
.unwrap();
cache.delete("del").await.unwrap();
assert_eq!(cache.get("del").await, None);
}
}
+85
View File
@@ -0,0 +1,85 @@
use std::net::SocketAddr;
fn fnv1a(data: &[u8]) -> u64 {
data.iter().fold(0xcbf29ce484222325u64, |hash, &byte| {
(hash ^ byte as u64).wrapping_mul(0x100000001b3)
})
}
#[derive(Debug, Clone)]
pub struct RippleConfig {
pub bind_addr: SocketAddr,
pub seed_peers: Vec<SocketAddr>,
pub machine_id: u64,
pub gossip_interval_ms: u64,
pub cache_max_bytes: usize,
}
fn parse_env_with_warning<T: std::str::FromStr>(var_name: &str, raw: &str) -> Option<T> {
match raw.parse::<T>() {
Ok(v) => Some(v),
Err(_) => {
tracing::warn!(var = var_name, value = raw, "invalid env var value, using default");
None
}
}
}
impl RippleConfig {
pub fn from_env() -> Result<Self, RippleConfigError> {
let bind_addr: SocketAddr = std::env::var("RIPPLE_BIND")
.unwrap_or_else(|_| "0.0.0.0:0".into())
.parse()
.map_err(|e| RippleConfigError::InvalidAddr(format!("{e}")))?;
let seed_peers: Vec<SocketAddr> = std::env::var("RIPPLE_PEERS")
.unwrap_or_default()
.split(',')
.filter(|s| !s.trim().is_empty())
.map(|s| {
s.trim()
.parse()
.map_err(|e| RippleConfigError::InvalidAddr(format!("{s}: {e}")))
})
.collect::<Result<Vec<_>, _>>()?;
let machine_id: u64 = std::env::var("RIPPLE_MACHINE_ID")
.ok()
.and_then(|v| parse_env_with_warning::<u64>("RIPPLE_MACHINE_ID", &v))
.unwrap_or_else(|| {
let host_str = std::fs::read_to_string("/etc/hostname")
.map(|s| s.trim().to_string())
.unwrap_or_else(|_| format!("pid-{}", std::process::id()));
let input = format!("{host_str}:{bind_addr}:{}", std::process::id());
fnv1a(input.as_bytes())
});
let gossip_interval_ms: u64 = std::env::var("RIPPLE_GOSSIP_INTERVAL_MS")
.ok()
.and_then(|v| parse_env_with_warning::<u64>("RIPPLE_GOSSIP_INTERVAL_MS", &v))
.unwrap_or(200)
.max(50);
let cache_max_mb: usize = std::env::var("RIPPLE_CACHE_MAX_MB")
.ok()
.and_then(|v| parse_env_with_warning::<usize>("RIPPLE_CACHE_MAX_MB", &v))
.unwrap_or(256)
.clamp(1, 16_384);
let cache_max_bytes = cache_max_mb.saturating_mul(1024).saturating_mul(1024);
Ok(Self {
bind_addr,
seed_peers,
machine_id,
gossip_interval_ms,
cache_max_bytes,
})
}
}
#[derive(Debug, thiserror::Error)]
pub enum RippleConfigError {
#[error("invalid address: {0}")]
InvalidAddr(String),
}
+31
View File
@@ -0,0 +1,31 @@
use super::lww_map::LwwDelta;
use super::g_counter::GCounterDelta;
use serde::{Deserialize, Serialize};
const SCHEMA_VERSION: u8 = 1;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CrdtDelta {
#[serde(default = "default_version")]
pub version: u8,
pub source_node: u64,
pub cache_delta: Option<LwwDelta>,
pub rate_limit_deltas: Vec<GCounterDelta>,
}
fn default_version() -> u8 {
1
}
impl CrdtDelta {
pub fn is_empty(&self) -> bool {
self.cache_delta
.as_ref()
.map_or(true, |d| d.entries.is_empty())
&& self.rate_limit_deltas.is_empty()
}
pub fn is_compatible(&self) -> bool {
self.version == SCHEMA_VERSION
}
}
@@ -0,0 +1,305 @@
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GCounter {
pub increments: HashMap<u64, u64>,
pub window_start_ms: u64,
pub window_duration_ms: u64,
}
impl GCounter {
pub fn new(window_start_ms: u64, window_duration_ms: u64) -> Self {
Self {
increments: HashMap::new(),
window_start_ms,
window_duration_ms,
}
}
pub fn total(&self) -> u64 {
self.increments
.values()
.copied()
.fold(0u64, u64::saturating_add)
}
pub fn increment(&mut self, node_id: u64) {
let slot = self.increments.entry(node_id).or_insert(0);
*slot = slot.saturating_add(1);
}
pub fn merge(&mut self, other: &GCounter) -> bool {
let mut changed = false;
other.increments.iter().for_each(|(&node, &count)| {
let slot = self.increments.entry(node).or_insert(0);
let new_val = (*slot).max(count);
if new_val != *slot {
*slot = new_val;
changed = true;
}
});
changed
}
pub fn is_expired(&self, now_wall_ms: u64) -> bool {
now_wall_ms.saturating_sub(self.window_start_ms) >= self.window_duration_ms
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GCounterDelta {
pub key: String,
pub counter: GCounter,
}
pub struct RateLimitStore {
counters: HashMap<String, GCounter>,
node_id: u64,
dirty: HashSet<String>,
}
impl RateLimitStore {
pub fn new(node_id: u64) -> Self {
Self {
counters: HashMap::new(),
node_id,
dirty: HashSet::new(),
}
}
fn aligned_window_start(now_wall_ms: u64, window_ms: u64) -> u64 {
(now_wall_ms / window_ms.max(1)) * window_ms.max(1)
}
pub fn check_and_increment(
&mut self,
key: &str,
limit: u32,
window_ms: u64,
now_wall_ms: u64,
) -> bool {
if window_ms == 0 {
return false;
}
let window_start = Self::aligned_window_start(now_wall_ms, window_ms);
let counter = self
.counters
.entry(key.to_string())
.and_modify(|c| {
if c.window_start_ms != window_start {
*c = GCounter::new(window_start, window_ms);
}
})
.or_insert_with(|| GCounter::new(window_start, window_ms));
let current = counter.total();
if current >= limit as u64 {
return false;
}
counter.increment(self.node_id);
self.dirty.insert(key.to_string());
true
}
pub fn merge_counter(&mut self, key: String, remote: &GCounter) -> bool {
if remote.window_duration_ms == 0 {
return false;
}
match self.counters.get_mut(&key) {
Some(local) if local.window_start_ms == remote.window_start_ms => {
if local.window_duration_ms != remote.window_duration_ms {
tracing::warn!(
key = %key,
local_window = local.window_duration_ms,
remote_window = remote.window_duration_ms,
"window_duration_ms mismatch, rejecting merge"
);
return false;
}
let changed = local.merge(remote);
if changed {
self.dirty.insert(key);
}
changed
}
Some(local) if remote.window_start_ms > local.window_start_ms => {
self.counters.insert(key.clone(), remote.clone());
self.dirty.insert(key);
true
}
None => {
self.counters.insert(key.clone(), remote.clone());
self.dirty.insert(key);
true
}
_ => false,
}
}
pub fn extract_dirty_deltas(&self) -> Vec<GCounterDelta> {
self.dirty
.iter()
.filter_map(|key| {
self.counters
.get(key)
.map(|counter| GCounterDelta {
key: key.clone(),
counter: counter.clone(),
})
})
.collect()
}
pub fn clear_dirty(&mut self) {
self.dirty.clear();
}
pub fn clear_dirty_keys(&mut self, keys: impl Iterator<Item = impl AsRef<str>>) {
keys.for_each(|k| {
self.dirty.remove(k.as_ref());
});
}
pub fn peek_count(&self, key: &str, window_ms: u64, now_wall_ms: u64) -> u64 {
match self.counters.get(key) {
Some(counter) if counter.window_start_ms == Self::aligned_window_start(now_wall_ms, window_ms) => {
counter.total()
}
_ => 0,
}
}
pub fn peek_dirty_counter(&self, key: &str) -> Option<&GCounter> {
match self.dirty.contains(key) {
true => self.counters.get(key),
false => None,
}
}
pub fn clear_single_dirty(&mut self, key: &str) {
self.dirty.remove(key);
}
pub fn estimated_bytes(&self) -> usize {
const PER_COUNTER_OVERHEAD: usize = 128;
self.counters
.iter()
.map(|(key, counter)| {
key.len()
+ std::mem::size_of::<GCounter>()
+ counter.increments.len() * (std::mem::size_of::<u64>() * 2)
+ PER_COUNTER_OVERHEAD
})
.fold(0usize, usize::saturating_add)
}
pub fn gc_expired(&mut self, now_wall_ms: u64) {
let expired: Vec<String> = self
.counters
.iter()
.filter(|(_, c)| c.is_expired(now_wall_ms))
.map(|(k, _)| k.clone())
.collect();
expired.iter().for_each(|key| {
self.counters.remove(key);
self.dirty.remove(key);
});
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn increment_and_total() {
let mut counter = GCounter::new(0, 60_000);
counter.increment(1);
counter.increment(1);
counter.increment(2);
assert_eq!(counter.total(), 3);
}
#[test]
fn merge_per_node_max() {
let mut a = GCounter::new(0, 60_000);
a.increment(1);
a.increment(1);
a.increment(2);
let mut b = GCounter::new(0, 60_000);
b.increment(1);
b.increment(2);
b.increment(2);
b.increment(2);
a.merge(&b);
assert_eq!(*a.increments.get(&1).unwrap(), 2);
assert_eq!(*a.increments.get(&2).unwrap(), 3);
assert_eq!(a.total(), 5);
}
#[test]
fn merge_commutativity() {
let mut a = GCounter::new(0, 60_000);
a.increments.insert(1, 5);
a.increments.insert(2, 3);
let mut b = GCounter::new(0, 60_000);
b.increments.insert(1, 3);
b.increments.insert(2, 7);
let mut ab = a.clone();
ab.merge(&b);
let mut ba = b.clone();
ba.merge(&a);
assert_eq!(ab.total(), ba.total());
}
#[test]
fn window_rollover() {
let mut store = RateLimitStore::new(1);
assert!(store.check_and_increment("k", 2, 1000, 500));
assert!(store.check_and_increment("k", 2, 1000, 600));
assert!(!store.check_and_increment("k", 2, 1000, 700));
assert!(store.check_and_increment("k", 2, 1000, 1500));
}
#[test]
fn rate_limit_enforcement() {
let mut store = RateLimitStore::new(1);
assert!(store.check_and_increment("k", 3, 60_000, 100));
assert!(store.check_and_increment("k", 3, 60_000, 200));
assert!(store.check_and_increment("k", 3, 60_000, 300));
assert!(!store.check_and_increment("k", 3, 60_000, 400));
}
#[test]
fn gc_expired_windows() {
let mut store = RateLimitStore::new(1);
store.check_and_increment("k", 10, 1000, 0);
assert_eq!(store.counters.len(), 1);
assert_eq!(store.dirty.len(), 1);
store.gc_expired(2000);
assert_eq!(store.counters.len(), 0);
assert_eq!(store.dirty.len(), 0);
}
#[test]
fn dirty_tracking() {
let mut store = RateLimitStore::new(1);
assert!(store.extract_dirty_deltas().is_empty());
store.check_and_increment("k1", 10, 60_000, 100);
store.check_and_increment("k2", 10, 60_000, 100);
assert_eq!(store.extract_dirty_deltas().len(), 2);
store.clear_dirty();
assert!(store.extract_dirty_deltas().is_empty());
store.check_and_increment("k1", 10, 60_000, 200);
assert_eq!(store.extract_dirty_deltas().len(), 1);
}
}
+206
View File
@@ -0,0 +1,206 @@
use serde::{Deserialize, Serialize};
use std::time::{SystemTime, UNIX_EPOCH};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct HlcTimestamp {
pub wall_ms: u64,
pub counter: u32,
pub node_id: u64,
}
impl HlcTimestamp {
pub const ZERO: Self = Self {
wall_ms: 0,
counter: 0,
node_id: 0,
};
}
impl PartialOrd for HlcTimestamp {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for HlcTimestamp {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.wall_ms
.cmp(&other.wall_ms)
.then(self.counter.cmp(&other.counter))
.then(self.node_id.cmp(&other.node_id))
}
}
fn advance_counter(wall: u64, counter: u32) -> (u64, u32) {
match counter == u32::MAX {
true => (wall.saturating_add(1), 0),
false => (wall, counter + 1),
}
}
pub struct Hlc {
node_id: u64,
last_wall_ms: u64,
last_counter: u32,
}
impl Hlc {
pub fn new(node_id: u64) -> Self {
Self {
node_id,
last_wall_ms: 0,
last_counter: 0,
}
}
fn physical_now() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
pub fn now(&mut self) -> HlcTimestamp {
let phys = Self::physical_now();
let (wall, counter) = match phys > self.last_wall_ms {
true => (phys, 0u32),
false => advance_counter(self.last_wall_ms, self.last_counter),
};
self.last_wall_ms = wall;
self.last_counter = counter;
HlcTimestamp {
wall_ms: wall,
counter,
node_id: self.node_id,
}
}
pub fn receive(&mut self, remote: HlcTimestamp) -> HlcTimestamp {
let phys = Self::physical_now();
let max_allowed = phys + 60_000;
let capped_remote_wall = remote.wall_ms.min(max_allowed);
if remote.wall_ms > max_allowed {
tracing::warn!(
remote_wall_ms = remote.wall_ms,
local_wall_ms = phys,
drift_ms = remote.wall_ms.saturating_sub(phys),
capped_to = max_allowed,
"remote HLC wall clock >60s ahead, capping"
);
}
let remote_counter = match capped_remote_wall == remote.wall_ms {
true => remote.counter,
false => 0u32,
};
let max_wall = phys.max(self.last_wall_ms).max(capped_remote_wall);
let (wall, counter) = match max_wall {
w if w == phys && w > self.last_wall_ms && w > capped_remote_wall => (w, 0u32),
w if w == self.last_wall_ms && w == capped_remote_wall => {
advance_counter(w, self.last_counter.max(remote_counter))
}
w if w == self.last_wall_ms => advance_counter(w, self.last_counter),
w if w == capped_remote_wall => advance_counter(w, remote_counter),
w => (w, 0u32),
};
self.last_wall_ms = wall;
self.last_counter = counter;
HlcTimestamp {
wall_ms: wall,
counter,
node_id: self.node_id,
}
}
pub fn node_id(&self) -> u64 {
self.node_id
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn monotonicity() {
let mut hlc = Hlc::new(1);
let timestamps: Vec<HlcTimestamp> = (0..100).map(|_| hlc.now()).collect();
timestamps.windows(2).for_each(|w| {
assert!(w[1] > w[0], "timestamps must be strictly increasing");
});
}
#[test]
fn merge_takes_max_within_drift_cap() {
let mut hlc = Hlc::new(1);
let now = Hlc::physical_now();
let remote = HlcTimestamp {
wall_ms: now + 5000,
counter: 10,
node_id: 2,
};
let merged = hlc.receive(remote);
assert!(merged.wall_ms >= remote.wall_ms);
let after = hlc.now();
assert!(after > merged);
}
#[test]
fn drift_cap_limits_remote_wall() {
let mut hlc = Hlc::new(1);
let now = Hlc::physical_now();
let remote = HlcTimestamp {
wall_ms: now + 120_000,
counter: 50,
node_id: 2,
};
let merged = hlc.receive(remote);
assert!(merged.wall_ms <= now + 60_000 + 1);
let after = hlc.now();
assert!(after > merged);
}
#[test]
fn total_order_across_nodes() {
let a = HlcTimestamp {
wall_ms: 100,
counter: 0,
node_id: 1,
};
let b = HlcTimestamp {
wall_ms: 100,
counter: 0,
node_id: 2,
};
assert!(a < b);
assert_ne!(a, b);
}
#[test]
fn counter_overflow_bumps_wall() {
let mut hlc = Hlc::new(1);
let future_wall = u64::MAX / 2;
hlc.last_wall_ms = future_wall;
hlc.last_counter = u32::MAX;
let ts = hlc.now();
assert_eq!(ts.wall_ms, future_wall + 1);
assert_eq!(ts.counter, 0);
let ts2 = hlc.now();
assert!(ts2 > ts);
}
#[test]
fn receive_counter_overflow_bumps_wall() {
let mut hlc = Hlc::new(1);
let future_wall = u64::MAX / 2;
hlc.last_wall_ms = future_wall;
hlc.last_counter = u32::MAX;
let remote = HlcTimestamp {
wall_ms: future_wall,
counter: u32::MAX,
node_id: 2,
};
let merged = hlc.receive(remote);
assert_eq!(merged.wall_ms, future_wall + 1);
assert_eq!(merged.counter, 0);
}
}
+388
View File
@@ -0,0 +1,388 @@
use super::hlc::HlcTimestamp;
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LwwEntry {
pub value: Option<Vec<u8>>,
pub timestamp: HlcTimestamp,
pub ttl_ms: u64,
pub created_at_wall_ms: u64,
}
impl LwwEntry {
fn is_expired(&self, now_wall_ms: u64) -> bool {
self.ttl_ms > 0 && now_wall_ms.saturating_sub(self.created_at_wall_ms) >= self.ttl_ms
}
fn is_tombstone(&self) -> bool {
self.value.is_none()
}
fn tombstone_expired(&self, now_wall_ms: u64) -> bool {
self.is_tombstone()
&& self.ttl_ms > 0
&& now_wall_ms.saturating_sub(self.created_at_wall_ms) >= self.ttl_ms.saturating_mul(2)
}
fn entry_byte_size(&self, key: &str) -> usize {
const OVERHEAD: usize = 128;
key.len()
+ self.value.as_ref().map_or(0, Vec::len)
+ std::mem::size_of::<Self>()
+ OVERHEAD
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LwwDelta {
pub entries: Vec<(String, LwwEntry)>,
}
struct LruTracker {
counter: u64,
counter_to_key: BTreeMap<u64, String>,
key_to_counter: HashMap<String, u64>,
}
impl LruTracker {
fn new() -> Self {
Self {
counter: 0,
counter_to_key: BTreeMap::new(),
key_to_counter: HashMap::new(),
}
}
fn promote(&mut self, key: &str) {
if let Some(old_counter) = self.key_to_counter.remove(key) {
self.counter_to_key.remove(&old_counter);
}
self.counter = self.counter.saturating_add(1);
self.counter_to_key.insert(self.counter, key.to_string());
self.key_to_counter.insert(key.to_string(), self.counter);
}
fn remove(&mut self, key: &str) {
if let Some(counter) = self.key_to_counter.remove(key) {
self.counter_to_key.remove(&counter);
}
}
fn pop_least_recent(&mut self) -> Option<String> {
let (&counter, _) = self.counter_to_key.iter().next()?;
let key = self.counter_to_key.remove(&counter)?;
self.key_to_counter.remove(&key);
Some(key)
}
}
pub struct LwwMap {
entries: HashMap<String, LwwEntry>,
lru: Mutex<LruTracker>,
estimated_bytes: usize,
}
impl LwwMap {
pub fn new() -> Self {
Self {
entries: HashMap::new(),
lru: Mutex::new(LruTracker::new()),
estimated_bytes: 0,
}
}
pub fn get(&self, key: &str, now_wall_ms: u64) -> Option<Vec<u8>> {
let entry = self.entries.get(key)?;
if entry.is_expired(now_wall_ms) || entry.is_tombstone() {
return None;
}
let value = entry.value.clone();
self.lru.lock().promote(key);
value
}
pub fn set(&mut self, key: String, value: Vec<u8>, timestamp: HlcTimestamp, ttl_ms: u64, wall_ms_now: u64) {
let entry = LwwEntry {
created_at_wall_ms: wall_ms_now,
value: Some(value),
timestamp,
ttl_ms,
};
self.remove_estimated_bytes(&key);
self.estimated_bytes += entry.entry_byte_size(&key);
self.entries.insert(key.clone(), entry);
self.lru.lock().promote(&key);
}
pub fn delete(&mut self, key: &str, timestamp: HlcTimestamp, wall_ms_now: u64) {
match self.entries.get(key) {
Some(existing) if existing.timestamp >= timestamp => return,
_ => {}
}
let ttl_ms = self
.entries
.get(key)
.map_or(60_000, |e| e.ttl_ms.max(60_000));
let entry = LwwEntry {
value: None,
timestamp,
ttl_ms,
created_at_wall_ms: wall_ms_now,
};
self.remove_estimated_bytes(key);
self.estimated_bytes += entry.entry_byte_size(key);
self.entries.insert(key.to_string(), entry);
self.lru.lock().remove(key);
}
pub fn merge_entry(&mut self, key: String, remote: LwwEntry) -> bool {
match self.entries.get(&key) {
Some(existing) if existing.timestamp >= remote.timestamp => false,
_ => {
let is_tombstone = remote.is_tombstone();
self.remove_estimated_bytes(&key);
self.estimated_bytes += remote.entry_byte_size(&key);
self.entries.insert(key.clone(), remote);
let mut lru = self.lru.lock();
match is_tombstone {
true => lru.remove(&key),
false => lru.promote(&key),
}
true
}
}
}
pub fn extract_delta_since(&self, watermark: HlcTimestamp) -> LwwDelta {
let entries = self
.entries
.iter()
.filter(|(_, entry)| entry.timestamp > watermark)
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
LwwDelta { entries }
}
pub fn gc_tombstones(&mut self, now_wall_ms: u64) {
let expired_keys: Vec<String> = self
.entries
.iter()
.filter(|(_, entry)| entry.tombstone_expired(now_wall_ms))
.map(|(k, _)| k.clone())
.collect();
expired_keys.iter().for_each(|key| {
self.remove_estimated_bytes(key);
self.entries.remove(key);
});
let mut lru = self.lru.lock();
expired_keys.iter().for_each(|key| {
lru.remove(key);
});
}
pub fn gc_expired(&mut self, now_wall_ms: u64) {
let expired_keys: Vec<String> = self
.entries
.iter()
.filter(|(_, entry)| entry.is_expired(now_wall_ms) && !entry.is_tombstone())
.map(|(k, _)| k.clone())
.collect();
expired_keys.iter().for_each(|key| {
self.remove_estimated_bytes(key);
self.entries.remove(key);
});
let mut lru = self.lru.lock();
expired_keys.iter().for_each(|key| {
lru.remove(key);
});
}
pub fn evict_lru(&mut self) -> Option<String> {
let key = self.lru.lock().pop_least_recent()?;
self.remove_estimated_bytes(&key);
self.entries.remove(&key);
Some(key)
}
pub fn estimated_bytes(&self) -> usize {
self.estimated_bytes
}
pub fn len(&self) -> usize {
self.entries.len()
}
fn remove_estimated_bytes(&mut self, key: &str) {
if let Some(existing) = self.entries.get(key) {
let size = existing.entry_byte_size(key);
if size > self.estimated_bytes {
tracing::warn!(
entry_size = size,
estimated_bytes = self.estimated_bytes,
key = key,
"estimated_bytes underflow detected, resetting to 0"
);
}
self.estimated_bytes = self.estimated_bytes.saturating_sub(size);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn ts(wall: u64, counter: u32, node: u64) -> HlcTimestamp {
HlcTimestamp {
wall_ms: wall,
counter,
node_id: node,
}
}
#[test]
fn set_and_get() {
let mut map = LwwMap::new();
map.set("k1".into(), b"hello".to_vec(), ts(100, 0, 1), 60_000, 100);
assert_eq!(map.get("k1", 100), Some(b"hello".to_vec()));
}
#[test]
fn ttl_expiry() {
let mut map = LwwMap::new();
map.set("k1".into(), b"hello".to_vec(), ts(100, 0, 1), 1000, 100);
assert_eq!(map.get("k1", 100), Some(b"hello".to_vec()));
assert_eq!(map.get("k1", 1200), None);
}
#[test]
fn merge_higher_timestamp_wins() {
let mut map = LwwMap::new();
map.set("k1".into(), b"old".to_vec(), ts(100, 0, 1), 60_000, 100);
let merged = map.merge_entry(
"k1".into(),
LwwEntry {
value: Some(b"new".to_vec()),
timestamp: ts(200, 0, 2),
ttl_ms: 60_000,
created_at_wall_ms: 200,
},
);
assert!(merged);
assert_eq!(map.get("k1", 200), Some(b"new".to_vec()));
}
#[test]
fn merge_lower_timestamp_rejected() {
let mut map = LwwMap::new();
map.set("k1".into(), b"current".to_vec(), ts(200, 0, 1), 60_000, 200);
let merged = map.merge_entry(
"k1".into(),
LwwEntry {
value: Some(b"stale".to_vec()),
timestamp: ts(100, 0, 2),
ttl_ms: 60_000,
created_at_wall_ms: 100,
},
);
assert!(!merged);
assert_eq!(map.get("k1", 200), Some(b"current".to_vec()));
}
#[test]
fn merge_commutativity() {
let e1 = LwwEntry {
value: Some(b"a".to_vec()),
timestamp: ts(100, 0, 1),
ttl_ms: 60_000,
created_at_wall_ms: 100,
};
let e2 = LwwEntry {
value: Some(b"b".to_vec()),
timestamp: ts(200, 0, 2),
ttl_ms: 60_000,
created_at_wall_ms: 200,
};
let mut map_ab = LwwMap::new();
map_ab.merge_entry("k".into(), e1.clone());
map_ab.merge_entry("k".into(), e2.clone());
let mut map_ba = LwwMap::new();
map_ba.merge_entry("k".into(), e2);
map_ba.merge_entry("k".into(), e1);
assert_eq!(map_ab.get("k", 200), map_ba.get("k", 200));
}
#[test]
fn merge_idempotency() {
let e = LwwEntry {
value: Some(b"a".to_vec()),
timestamp: ts(100, 0, 1),
ttl_ms: 60_000,
created_at_wall_ms: 100,
};
let mut map = LwwMap::new();
assert!(map.merge_entry("k".into(), e.clone()));
assert!(!map.merge_entry("k".into(), e));
}
#[test]
fn delete_creates_tombstone() {
let mut map = LwwMap::new();
map.set("k1".into(), b"val".to_vec(), ts(100, 0, 1), 60_000, 100);
map.delete("k1", ts(200, 0, 1), 200);
assert_eq!(map.get("k1", 200), None);
}
#[test]
fn tombstone_gc() {
let mut map = LwwMap::new();
map.set("k1".into(), b"val".to_vec(), ts(100, 0, 1), 60_000, 100);
map.delete("k1", ts(100, 1, 1), 100);
assert_eq!(map.len(), 1);
map.gc_tombstones(100 + 120_001);
assert_eq!(map.len(), 0);
}
#[test]
fn delta_extraction() {
let mut map = LwwMap::new();
map.set("k1".into(), b"a".to_vec(), ts(100, 0, 1), 60_000, 100);
map.set("k2".into(), b"b".to_vec(), ts(200, 0, 1), 60_000, 200);
let delta = map.extract_delta_since(ts(150, 0, 0));
assert_eq!(delta.entries.len(), 1);
assert_eq!(delta.entries[0].0, "k2");
}
#[test]
fn lru_eviction() {
let mut map = LwwMap::new();
map.set("k1".into(), b"a".to_vec(), ts(100, 0, 1), 60_000, 100);
map.set("k2".into(), b"b".to_vec(), ts(101, 0, 1), 60_000, 101);
map.set("k3".into(), b"c".to_vec(), ts(102, 0, 1), 60_000, 102);
let _ = map.get("k1", 102);
let evicted = map.evict_lru();
assert_eq!(evicted.as_deref(), Some("k2"));
}
#[test]
fn merged_entries_are_evictable() {
let mut map = LwwMap::new();
map.merge_entry(
"remote_key".into(),
LwwEntry {
value: Some(b"remote_val".to_vec()),
timestamp: ts(100, 0, 2),
ttl_ms: 60_000,
created_at_wall_ms: 100,
},
);
let evicted = map.evict_lru();
assert_eq!(evicted.as_deref(), Some("remote_key"));
assert_eq!(map.len(), 0);
}
}
+214
View File
@@ -0,0 +1,214 @@
pub mod delta;
pub mod hlc;
pub mod lww_map;
pub mod g_counter;
use delta::CrdtDelta;
use hlc::{Hlc, HlcTimestamp};
use lww_map::LwwMap;
use g_counter::RateLimitStore;
use std::time::{SystemTime, UNIX_EPOCH};
pub struct CrdtStore {
hlc: Hlc,
cache: LwwMap,
rate_limits: RateLimitStore,
last_broadcast_ts: HlcTimestamp,
}
impl CrdtStore {
pub fn new(node_id: u64) -> Self {
Self {
hlc: Hlc::new(node_id),
cache: LwwMap::new(),
rate_limits: RateLimitStore::new(node_id),
last_broadcast_ts: HlcTimestamp::ZERO,
}
}
fn wall_ms_now() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
pub fn cache_get(&self, key: &str) -> Option<Vec<u8>> {
self.cache.get(key, Self::wall_ms_now())
}
pub fn cache_set(&mut self, key: String, value: Vec<u8>, ttl_ms: u64) {
let ts = self.hlc.now();
self.cache.set(key, value, ts, ttl_ms, Self::wall_ms_now());
}
pub fn cache_delete(&mut self, key: &str) {
let ts = self.hlc.now();
self.cache.delete(key, ts, Self::wall_ms_now());
}
pub fn rate_limit_peek(&self, key: &str, window_ms: u64) -> u64 {
self.rate_limits
.peek_count(key, window_ms, Self::wall_ms_now())
}
pub fn rate_limit_check(&mut self, key: &str, limit: u32, window_ms: u64) -> bool {
self.rate_limits
.check_and_increment(key, limit, window_ms, Self::wall_ms_now())
}
pub fn peek_broadcast_delta(&self) -> CrdtDelta {
let cache_delta = {
let d = self.cache.extract_delta_since(self.last_broadcast_ts);
match d.entries.is_empty() {
true => None,
false => Some(d),
}
};
let rate_limit_deltas = self.rate_limits.extract_dirty_deltas();
CrdtDelta {
version: 1,
source_node: self.hlc.node_id(),
cache_delta,
rate_limit_deltas,
}
}
pub fn commit_broadcast(&mut self, delta: &CrdtDelta) {
let max_ts = delta
.cache_delta
.as_ref()
.and_then(|d| d.entries.iter().map(|(_, e)| e.timestamp).max())
.unwrap_or(self.last_broadcast_ts);
self.last_broadcast_ts = max_ts;
let committed_keys: std::collections::HashSet<&str> = delta
.rate_limit_deltas
.iter()
.map(|d| d.key.as_str())
.collect();
committed_keys.iter().for_each(|&key| {
let still_matches = self
.rate_limits
.peek_dirty_counter(key)
.zip(delta.rate_limit_deltas.iter().find(|d| d.key == key))
.is_some_and(|(current, committed)| {
current.window_start_ms == committed.counter.window_start_ms
&& current.total() == committed.counter.total()
});
if still_matches {
self.rate_limits.clear_single_dirty(key);
}
});
}
pub fn merge_delta(&mut self, delta: &CrdtDelta) -> bool {
if !delta.is_compatible() {
tracing::warn!(
version = delta.version,
"dropping incompatible CRDT delta version"
);
return false;
}
let mut changed = false;
if let Some(ref cache_delta) = delta.cache_delta {
cache_delta.entries.iter().for_each(|(key, entry)| {
let _ = self.hlc.receive(entry.timestamp);
if self.cache.merge_entry(key.clone(), entry.clone()) {
changed = true;
}
});
}
delta.rate_limit_deltas.iter().for_each(|rd| {
if self
.rate_limits
.merge_counter(rd.key.clone(), &rd.counter)
{
changed = true;
}
});
changed
}
pub fn run_maintenance(&mut self) {
let now = Self::wall_ms_now();
self.cache.gc_tombstones(now);
self.cache.gc_expired(now);
self.rate_limits.gc_expired(now);
}
pub fn cache_estimated_bytes(&self) -> usize {
self.cache.estimated_bytes()
}
pub fn rate_limit_estimated_bytes(&self) -> usize {
self.rate_limits.estimated_bytes()
}
pub fn evict_lru(&mut self) -> Option<String> {
self.cache.evict_lru()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn roundtrip_cache() {
let mut store = CrdtStore::new(1);
store.cache_set("key".into(), b"value".to_vec(), 60_000);
assert_eq!(store.cache_get("key"), Some(b"value".to_vec()));
}
#[test]
fn delta_merge_convergence() {
let mut store_a = CrdtStore::new(1);
let mut store_b = CrdtStore::new(2);
store_a.cache_set("x".into(), b"from_a".to_vec(), 60_000);
store_b.cache_set("y".into(), b"from_b".to_vec(), 60_000);
let delta_a = store_a.peek_broadcast_delta();
store_a.commit_broadcast(&delta_a);
let delta_b = store_b.peek_broadcast_delta();
store_b.commit_broadcast(&delta_b);
store_b.merge_delta(&delta_a);
store_a.merge_delta(&delta_b);
assert_eq!(store_a.cache_get("x"), Some(b"from_a".to_vec()));
assert_eq!(store_a.cache_get("y"), Some(b"from_b".to_vec()));
assert_eq!(store_b.cache_get("x"), Some(b"from_a".to_vec()));
assert_eq!(store_b.cache_get("y"), Some(b"from_b".to_vec()));
}
#[test]
fn rate_limit_across_stores() {
let mut store_a = CrdtStore::new(1);
let mut store_b = CrdtStore::new(2);
store_a.rate_limit_check("rl:test", 5, 60_000);
store_a.rate_limit_check("rl:test", 5, 60_000);
store_b.rate_limit_check("rl:test", 5, 60_000);
let delta_a = store_a.peek_broadcast_delta();
store_a.commit_broadcast(&delta_a);
store_b.merge_delta(&delta_a);
let delta_b = store_b.peek_broadcast_delta();
store_b.commit_broadcast(&delta_b);
store_a.merge_delta(&delta_b);
}
#[test]
fn incompatible_version_rejected() {
let mut store = CrdtStore::new(1);
let delta = CrdtDelta {
version: 255,
source_node: 99,
cache_delta: None,
rate_limit_deltas: vec![],
};
assert!(!store.merge_delta(&delta));
}
}
+94
View File
@@ -0,0 +1,94 @@
use crate::cache::RippleCache;
use crate::config::RippleConfig;
use crate::crdt::CrdtStore;
use crate::eviction::MemoryBudget;
use crate::gossip::{GossipEngine, PeerId};
use crate::rate_limiter::RippleRateLimiter;
use crate::transport::Transport;
use parking_lot::RwLock;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tranquil_infra::{Cache, DistributedRateLimiter};
pub struct RippleEngine;
impl RippleEngine {
pub async fn start(
config: RippleConfig,
shutdown: CancellationToken,
) -> Result<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>, SocketAddr), RippleStartError> {
let store = Arc::new(RwLock::new(CrdtStore::new(config.machine_id)));
let (transport, incoming_rx) = Transport::bind(config.bind_addr, config.machine_id, shutdown.clone())
.await
.map_err(|e| RippleStartError::Bind(e.to_string()))?;
let transport = Arc::new(transport);
let bound_addr = transport.local_addr();
let local_id = PeerId {
addr: bound_addr,
machine_id: config.machine_id,
generation: 0,
};
let gossip = GossipEngine::new(transport, store.clone(), local_id);
let gossip_handle = gossip.spawn(
config.seed_peers,
config.gossip_interval_ms,
incoming_rx,
shutdown.clone(),
);
let budget = MemoryBudget::new(config.cache_max_bytes);
let store_for_eviction = store.clone();
let eviction_shutdown = shutdown.clone();
let eviction_handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
loop {
tokio::select! {
_ = eviction_shutdown.cancelled() => break,
_ = interval.tick() => {
budget.enforce(&mut store_for_eviction.write());
}
}
}
});
let shutdown_for_monitor = shutdown.clone();
tokio::spawn(async move {
shutdown_for_monitor.cancelled().await;
let gossip_result = gossip_handle.await;
let eviction_result = eviction_handle.await;
if let Err(e) = gossip_result {
tracing::error!(error = %e, "gossip task panicked");
}
if let Err(e) = eviction_result {
tracing::error!(error = %e, "eviction task panicked");
}
});
let cache: Arc<dyn Cache> = Arc::new(RippleCache::new(store.clone()));
let rate_limiter: Arc<dyn DistributedRateLimiter> =
Arc::new(RippleRateLimiter::new(store));
tracing::info!(
bind = %bound_addr,
machine_id = config.machine_id,
max_cache_mb = config.cache_max_bytes / (1024 * 1024),
"ripple engine started"
);
Ok((cache, rate_limiter, bound_addr))
}
}
#[derive(Debug, thiserror::Error)]
pub enum RippleStartError {
#[error("failed to bind transport: {0}")]
Bind(String),
#[error("configuration error: {0}")]
Config(String),
}
+78
View File
@@ -0,0 +1,78 @@
use crate::crdt::CrdtStore;
pub struct MemoryBudget {
max_bytes: usize,
}
impl MemoryBudget {
pub fn new(max_bytes: usize) -> Self {
Self { max_bytes }
}
pub fn enforce(&self, store: &mut CrdtStore) {
store.run_maintenance();
let max_bytes = self.max_bytes;
let total_bytes = store.cache_estimated_bytes().saturating_add(store.rate_limit_estimated_bytes());
let overshoot_ratio = match total_bytes > max_bytes && max_bytes > 0 {
true => total_bytes / max_bytes,
false => 0,
};
const BASE_BATCH: usize = 256;
let batch_size = match overshoot_ratio {
0..=1 => BASE_BATCH,
2..=4 => BASE_BATCH * 4,
_ => BASE_BATCH * 8,
};
let evicted = std::iter::from_fn(|| {
let current = store.cache_estimated_bytes().saturating_add(store.rate_limit_estimated_bytes());
match current > max_bytes {
true => store.evict_lru(),
false => None,
}
})
.take(batch_size)
.count();
if evicted > 0 {
tracing::info!(
evicted_entries = evicted,
cache_bytes = store.cache_estimated_bytes(),
rate_limit_bytes = store.rate_limit_estimated_bytes(),
max_bytes = self.max_bytes,
"memory budget eviction"
);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn eviction_under_budget() {
let mut store = CrdtStore::new(1);
let budget = MemoryBudget::new(1024 * 1024);
store.cache_set("k".into(), vec![1, 2, 3], 60_000);
budget.enforce(&mut store);
assert!(store.cache_get("k").is_some());
}
#[test]
fn eviction_over_budget() {
let mut store = CrdtStore::new(1);
let budget = MemoryBudget::new(100);
(0..50).for_each(|i| {
store.cache_set(
format!("key-{i}"),
vec![0u8; 64],
60_000,
);
});
budget.enforce(&mut store);
let total = store.cache_estimated_bytes().saturating_add(store.rate_limit_estimated_bytes());
assert!(total <= 100);
}
}
+390
View File
@@ -0,0 +1,390 @@
use crate::crdt::delta::CrdtDelta;
use crate::crdt::CrdtStore;
use crate::transport::{ChannelTag, IncomingFrame, Transport};
use foca::{Config, Foca, Notification, Runtime, Timer};
use parking_lot::RwLock;
use rand::rngs::StdRng;
use rand::SeedableRng;
use std::collections::HashSet;
use std::fmt;
use std::net::SocketAddr;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
const MAX_GCOUNTER_NODES: usize = 256;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct PeerId {
pub addr: SocketAddr,
pub machine_id: u64,
pub generation: u32,
}
impl fmt::Display for PeerId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}@{}(g{})", self.machine_id, self.addr, self.generation)
}
}
impl foca::Identity for PeerId {
type Addr = SocketAddr;
fn addr(&self) -> SocketAddr {
self.addr
}
fn renew(&self) -> Option<Self> {
Some(Self {
addr: self.addr,
machine_id: self.machine_id,
generation: self.generation.saturating_add(1),
})
}
fn win_addr_conflict(&self, adversary: &Self) -> bool {
self.generation > adversary.generation
}
}
impl serde::Serialize for PeerId {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeTuple;
let mut tup = serializer.serialize_tuple(3)?;
tup.serialize_element(&self.addr.to_string())?;
tup.serialize_element(&self.machine_id)?;
tup.serialize_element(&self.generation)?;
tup.end()
}
}
impl<'de> serde::Deserialize<'de> for PeerId {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let (addr_str, machine_id, generation): (String, u64, u32) =
serde::Deserialize::deserialize(deserializer)?;
let addr: SocketAddr = addr_str.parse().map_err(serde::de::Error::custom)?;
Ok(Self {
addr,
machine_id,
generation,
})
}
}
enum RuntimeAction {
SendTo(PeerId, Vec<u8>),
ScheduleTimer(Timer<PeerId>, Duration),
MemberUp(SocketAddr),
MemberDown(SocketAddr),
}
struct BufferedRuntime {
actions: Vec<RuntimeAction>,
}
impl BufferedRuntime {
fn new() -> Self {
Self {
actions: Vec::new(),
}
}
}
struct MemberTracker {
active_addrs: HashSet<SocketAddr>,
}
impl MemberTracker {
fn new() -> Self {
Self {
active_addrs: HashSet::new(),
}
}
fn member_up(&mut self, addr: SocketAddr) {
self.active_addrs.insert(addr);
}
fn member_down(&mut self, addr: SocketAddr) {
self.active_addrs.remove(&addr);
}
fn active_peers(&self) -> impl Iterator<Item = SocketAddr> + '_ {
self.active_addrs.iter().copied()
}
}
impl Runtime<PeerId> for &mut BufferedRuntime {
fn notify(&mut self, notification: Notification<'_, PeerId>) {
match notification {
Notification::MemberUp(peer) => {
self.actions.push(RuntimeAction::MemberUp(peer.addr));
}
Notification::MemberDown(peer) => {
self.actions.push(RuntimeAction::MemberDown(peer.addr));
}
_ => {}
}
}
fn send_to(&mut self, to: PeerId, data: &[u8]) {
self.actions
.push(RuntimeAction::SendTo(to, data.to_vec()));
}
fn submit_after(&mut self, event: Timer<PeerId>, after: Duration) {
self.actions.push(RuntimeAction::ScheduleTimer(event, after));
}
}
pub struct GossipEngine {
transport: Arc<Transport>,
store: Arc<RwLock<CrdtStore>>,
local_id: PeerId,
}
impl GossipEngine {
pub fn new(
transport: Arc<Transport>,
store: Arc<RwLock<CrdtStore>>,
local_id: PeerId,
) -> Self {
Self {
transport,
store,
local_id,
}
}
pub fn spawn(
self,
seed_peers: Vec<SocketAddr>,
gossip_interval_ms: u64,
mut incoming_rx: mpsc::Receiver<IncomingFrame>,
shutdown: CancellationToken,
) -> tokio::task::JoinHandle<()> {
let mut config = Config::simple();
config.max_packet_size = NonZeroUsize::new(2 * 1024 * 1024).expect("nonzero");
config.periodic_gossip = Some(foca::PeriodicParams {
frequency: Duration::from_millis(gossip_interval_ms),
num_members: NonZeroUsize::new(3).expect("nonzero"),
});
config.periodic_announce = Some(foca::PeriodicParams {
frequency: Duration::from_secs(30),
num_members: NonZeroUsize::new(3).expect("nonzero"),
});
let rng = StdRng::from_os_rng();
let codec = foca::BincodeCodec(bincode::config::standard());
let mut foca: Foca<PeerId, _, _, _> = Foca::new(self.local_id.clone(), config, rng, codec);
let transport = self.transport.clone();
let store = self.store.clone();
let (timer_tx, mut timer_rx) = mpsc::channel::<(Timer<PeerId>, Duration)>(256);
const WATERMARK_STALE_SECS: u64 = 30;
tokio::spawn(async move {
let mut runtime = BufferedRuntime::new();
let mut members = MemberTracker::new();
let mut last_commit = tokio::time::Instant::now();
seed_peers.iter().for_each(|&addr| {
let seed_id = PeerId {
addr,
machine_id: 0,
generation: 0,
};
if let Err(e) = foca.announce(seed_id, &mut runtime) {
tracing::warn!(error = %e, "failed to announce to seed peer");
}
});
drain_runtime_actions(&mut runtime, &transport, &timer_tx, &mut members, &shutdown);
let mut gossip_tick =
tokio::time::interval(Duration::from_millis(gossip_interval_ms));
let mut maintenance_tick = tokio::time::interval(Duration::from_secs(10));
loop {
tokio::select! {
_ = shutdown.cancelled() => {
tracing::info!("gossip engine shutting down, flushing final delta");
flush_final_delta(&store, &transport, &members);
break;
}
Some(frame) = incoming_rx.recv() => {
match frame.tag {
ChannelTag::Gossip => {
if let Err(e) = foca.handle_data(&frame.data, &mut runtime) {
tracing::warn!(error = %e, "foca handle_data error");
}
drain_runtime_actions(&mut runtime, &transport, &timer_tx, &mut members, &shutdown);
}
ChannelTag::CrdtSync => {
const MAX_DELTA_ENTRIES: usize = 10_000;
const MAX_DELTA_RATE_LIMITS: usize = 10_000;
match bincode::serde::decode_from_slice::<CrdtDelta, _>(&frame.data, bincode::config::standard()) {
Ok((delta, _)) => {
let cache_len = delta.cache_delta.as_ref().map_or(0, |d| d.entries.len());
let rl_len = delta.rate_limit_deltas.len();
let gcounter_oversize = delta.rate_limit_deltas.iter().any(|rd| rd.counter.increments.len() > MAX_GCOUNTER_NODES);
let window_mismatch = delta.rate_limit_deltas.iter().any(|rd| rd.counter.window_duration_ms == 0);
match cache_len > MAX_DELTA_ENTRIES || rl_len > MAX_DELTA_RATE_LIMITS || gcounter_oversize || window_mismatch {
true => {
tracing::warn!(
cache_entries = cache_len,
rate_limit_entries = rl_len,
gcounter_oversize = gcounter_oversize,
"dropping invalid CRDT delta"
);
}
false => {
store.write().merge_delta(&delta);
}
}
}
Err(e) => {
tracing::warn!(error = %e, "failed to decode crdt sync delta");
}
}
}
_ => {}
}
}
_ = gossip_tick.tick() => {
let pending = {
let s = store.read();
let delta = s.peek_broadcast_delta();
match delta.is_empty() {
true => None,
false => {
match bincode::serde::encode_to_vec(&delta, bincode::config::standard()) {
Ok(bytes) => Some((bytes, delta)),
Err(e) => {
tracing::warn!(error = %e, "failed to serialize broadcast delta");
None
}
}
},
}
};
if let Some((ref data, ref delta)) = pending {
let peers: Vec<SocketAddr> = members.active_peers().collect();
let mut all_queued = true;
let cancel = shutdown.clone();
peers.iter().for_each(|&addr| {
match transport.try_queue(addr, ChannelTag::CrdtSync, data) {
true => {}
false => {
all_queued = false;
let t = transport.clone();
let d = data.clone();
let c = cancel.clone();
tokio::spawn(async move {
tokio::select! {
_ = c.cancelled() => {}
_ = t.send(addr, ChannelTag::CrdtSync, &d) => {}
}
});
}
}
});
let stale = last_commit.elapsed() > Duration::from_secs(WATERMARK_STALE_SECS);
if all_queued || peers.is_empty() || stale {
if stale && !all_queued {
tracing::warn!(
elapsed_secs = last_commit.elapsed().as_secs(),
"force-advancing broadcast watermark (staleness cap)"
);
}
store.write().commit_broadcast(delta);
last_commit = tokio::time::Instant::now();
}
}
if let Err(e) = foca.gossip(&mut runtime) {
tracing::warn!(error = %e, "foca gossip error");
}
drain_runtime_actions(&mut runtime, &transport, &timer_tx, &mut members, &shutdown);
}
Some((timer, _)) = timer_rx.recv() => {
if let Err(e) = foca.handle_timer(timer, &mut runtime) {
tracing::warn!(error = %e, "foca handle_timer error");
}
drain_runtime_actions(&mut runtime, &transport, &timer_tx, &mut members, &shutdown);
}
_ = maintenance_tick.tick() => {
store.write().run_maintenance();
tracing::trace!(
members = foca.num_members(),
cache_bytes = store.read().cache_estimated_bytes(),
"maintenance cycle"
);
}
}
}
})
}
}
fn flush_final_delta(
store: &Arc<RwLock<CrdtStore>>,
transport: &Arc<Transport>,
members: &MemberTracker,
) {
let s = store.read();
let delta = s.peek_broadcast_delta();
if delta.is_empty() {
return;
}
match bincode::serde::encode_to_vec(&delta, bincode::config::standard()) {
Ok(bytes) => {
members.active_peers().for_each(|addr| {
let _ = transport.try_queue(addr, ChannelTag::CrdtSync, &bytes);
});
}
Err(e) => {
tracing::warn!(error = %e, "failed to serialize final delta on shutdown");
}
}
}
fn drain_runtime_actions(
runtime: &mut BufferedRuntime,
transport: &Arc<Transport>,
timer_tx: &mpsc::Sender<(Timer<PeerId>, Duration)>,
members: &mut MemberTracker,
shutdown: &CancellationToken,
) {
let actions: Vec<RuntimeAction> = runtime.actions.drain(..).collect();
actions.into_iter().for_each(|action| match action {
RuntimeAction::SendTo(peer, data) => {
let t = transport.clone();
let c = shutdown.clone();
tokio::spawn(async move {
tokio::select! {
_ = c.cancelled() => {}
_ = t.send(peer.addr, ChannelTag::Gossip, &data) => {}
}
});
}
RuntimeAction::ScheduleTimer(timer, duration) => {
let tx = timer_tx.clone();
tokio::spawn(async move {
tokio::time::sleep(duration).await;
let _ = tx.send((timer, duration)).await;
});
}
RuntimeAction::MemberUp(addr) => {
tracing::info!(peer = %addr, "member up");
members.member_up(addr);
}
RuntimeAction::MemberDown(addr) => {
tracing::info!(peer = %addr, "member down");
members.member_down(addr);
}
});
}
+11
View File
@@ -0,0 +1,11 @@
pub mod cache;
pub mod config;
pub mod crdt;
pub mod engine;
pub mod eviction;
pub mod gossip;
pub mod rate_limiter;
pub mod transport;
pub use config::RippleConfig;
pub use engine::{RippleEngine, RippleStartError};
@@ -0,0 +1,49 @@
use crate::crdt::CrdtStore;
use async_trait::async_trait;
use parking_lot::RwLock;
use std::sync::Arc;
use tranquil_infra::DistributedRateLimiter;
pub struct RippleRateLimiter {
store: Arc<RwLock<CrdtStore>>,
}
impl RippleRateLimiter {
pub fn new(store: Arc<RwLock<CrdtStore>>) -> Self {
Self { store }
}
}
#[async_trait]
impl DistributedRateLimiter for RippleRateLimiter {
async fn check_rate_limit(&self, key: &str, limit: u32, window_ms: u64) -> bool {
self.store.write().rate_limit_check(key, limit, window_ms)
}
async fn peek_rate_limit_count(&self, key: &str, window_ms: u64) -> u64 {
self.store.read().rate_limit_peek(key, window_ms)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn rate_limiter_trait_allows_within_limit() {
let store = Arc::new(RwLock::new(CrdtStore::new(1)));
let rl = RippleRateLimiter::new(store);
assert!(rl.check_rate_limit("test", 5, 60_000).await);
assert!(rl.check_rate_limit("test", 5, 60_000).await);
}
#[tokio::test]
async fn rate_limiter_trait_blocks_over_limit() {
let store = Arc::new(RwLock::new(CrdtStore::new(1)));
let rl = RippleRateLimiter::new(store);
assert!(rl.check_rate_limit("k", 3, 60_000).await);
assert!(rl.check_rate_limit("k", 3, 60_000).await);
assert!(rl.check_rate_limit("k", 3, 60_000).await);
assert!(!rl.check_rate_limit("k", 3, 60_000).await);
}
}
+442
View File
@@ -0,0 +1,442 @@
use backon::{ExponentialBuilder, Retryable};
use bytes::{Buf, BufMut, BytesMut};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024;
const MAX_INBOUND_CONNECTIONS: usize = 512;
const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum ChannelTag {
Gossip = 0x01,
CrdtSync = 0x02,
Raft = 0x03,
Direct = 0x04,
}
impl ChannelTag {
fn from_u8(v: u8) -> Option<Self> {
match v {
0x01 => Some(Self::Gossip),
0x02 => Some(Self::CrdtSync),
0x03 => Some(Self::Raft),
0x04 => Some(Self::Direct),
_ => None,
}
}
}
#[derive(Debug)]
pub struct IncomingFrame {
pub from: SocketAddr,
pub tag: ChannelTag,
pub data: Vec<u8>,
}
struct ConnectionWriter {
tx: mpsc::Sender<Vec<u8>>,
}
pub struct Transport {
local_addr: SocketAddr,
_machine_id: u64,
connections: Arc<parking_lot::Mutex<HashMap<SocketAddr, ConnectionWriter>>>,
connecting: Arc<parking_lot::Mutex<std::collections::HashSet<SocketAddr>>>,
#[allow(dead_code)]
inbound_count: Arc<AtomicUsize>,
shutdown: CancellationToken,
incoming_tx: mpsc::Sender<IncomingFrame>,
}
impl Transport {
pub async fn bind(
addr: SocketAddr,
machine_id: u64,
shutdown: CancellationToken,
) -> Result<(Self, mpsc::Receiver<IncomingFrame>), std::io::Error> {
let listener = TcpListener::bind(addr).await?;
let local_addr = listener.local_addr()?;
let (incoming_tx, incoming_rx) = mpsc::channel(4096);
let inbound_count = Arc::new(AtomicUsize::new(0));
let transport = Self {
local_addr,
_machine_id: machine_id,
connections: Arc::new(parking_lot::Mutex::new(HashMap::new())),
connecting: Arc::new(parking_lot::Mutex::new(std::collections::HashSet::new())),
inbound_count: inbound_count.clone(),
shutdown: shutdown.clone(),
incoming_tx: incoming_tx.clone(),
};
let cancel = shutdown.clone();
let inbound_counter = inbound_count.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = cancel.cancelled() => break,
result = listener.accept() => {
match result {
Ok((stream, peer_addr)) => {
let current = inbound_counter.load(Ordering::Relaxed);
if current >= MAX_INBOUND_CONNECTIONS {
tracing::warn!(
peer = %peer_addr,
count = current,
max = MAX_INBOUND_CONNECTIONS,
"rejecting inbound connection: limit reached"
);
drop(stream);
continue;
}
inbound_counter.fetch_add(1, Ordering::Relaxed);
Self::spawn_reader(
stream,
peer_addr,
incoming_tx.clone(),
cancel.clone(),
inbound_counter.clone(),
);
tracing::debug!(peer = %peer_addr, "accepted inbound connection");
}
Err(e) => {
tracing::warn!(error = %e, "accept failed");
}
}
}
}
}
});
tracing::info!(addr = %local_addr, "ripple transport bound");
Ok((transport, incoming_rx))
}
pub fn local_addr(&self) -> SocketAddr {
self.local_addr
}
pub fn try_queue(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) -> bool {
let frame = match encode_frame(tag, data) {
Some(f) => f,
None => return false,
};
let conns = self.connections.lock();
match conns.get(&target) {
Some(writer) => writer.tx.try_send(frame).is_ok(),
None => false,
}
}
pub async fn send(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) {
let frame = match encode_frame(tag, data) {
Some(f) => f,
None => return,
};
let writer = {
let conns = self.connections.lock();
conns.get(&target).map(|w| w.tx.clone())
};
match writer {
Some(tx) => {
if tx.send(frame).await.is_err() {
self.connections.lock().remove(&target);
self.connect_and_send(target, tag, data).await;
}
}
None => {
self.connect_and_send(target, tag, data).await;
}
}
}
async fn connect_and_send(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) {
{
let mut connecting = self.connecting.lock();
if connecting.contains(&target) {
tracing::debug!(peer = %target, "connection already in-flight, dropping frame");
return;
}
connecting.insert(target);
}
let result = self.connect_and_send_inner(target, tag, data).await;
self.connecting.lock().remove(&target);
result
}
async fn connect_and_send_inner(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) {
let shutdown = self.shutdown.clone();
let stream = (|| async {
tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(target))
.await
.map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "connect timeout"))?
})
.retry(
ExponentialBuilder::default()
.with_min_delay(Duration::from_millis(50))
.with_max_delay(Duration::from_secs(2))
.with_max_times(3),
)
.when(|_| !shutdown.is_cancelled())
.await;
match stream {
Ok(stream) => {
let (read_half, write_half) = stream.into_split();
let (write_tx, mut write_rx) = mpsc::channel::<Vec<u8>>(1024);
let cancel = self.shutdown.clone();
let connections = self.connections.clone();
let peer = target;
tokio::spawn(async move {
let mut writer = write_half;
loop {
tokio::select! {
_ = cancel.cancelled() => break,
msg = write_rx.recv() => {
match msg {
Some(buf) => {
let write_result = tokio::time::timeout(
WRITE_TIMEOUT,
writer.write_all(&buf),
).await;
match write_result {
Ok(Ok(())) => {}
Ok(Err(e)) => {
tracing::warn!(peer = %peer, error = %e, "write failed, closing connection");
break;
}
Err(_) => {
tracing::warn!(peer = %peer, "write timed out, closing connection");
break;
}
}
}
None => break,
}
}
}
}
connections.lock().remove(&peer);
});
Self::spawn_reader_half(read_half, target, self.incoming_tx.clone(), self.shutdown.clone());
let frame = match encode_frame(tag, data) {
Some(f) => f,
None => return,
};
let _ = write_tx.send(frame).await;
self.connections.lock().insert(
target,
ConnectionWriter { tx: write_tx },
);
tracing::debug!(peer = %target, "established outbound connection");
}
Err(e) => {
tracing::warn!(peer = %target, error = %e, "failed to connect after retries");
}
}
}
fn spawn_reader(
stream: TcpStream,
peer_addr: SocketAddr,
incoming_tx: mpsc::Sender<IncomingFrame>,
cancel: CancellationToken,
inbound_counter: Arc<AtomicUsize>,
) {
tokio::spawn(async move {
let mut buf = BytesMut::with_capacity(8192);
let mut stream = stream;
loop {
if buf.len() > MAX_FRAME_SIZE * 2 {
tracing::warn!(peer = %peer_addr, buf_len = buf.len(), "read buffer exceeded limit, closing connection");
break;
}
tokio::select! {
_ = cancel.cancelled() => break,
n = stream.read_buf(&mut buf) => {
match n {
Ok(0) | Err(_) => break,
Ok(_) => {
if !Self::process_frames(&mut buf, peer_addr, &incoming_tx) {
break;
}
}
}
}
}
}
inbound_counter.fetch_sub(1, Ordering::Relaxed);
});
}
fn spawn_reader_half(
read_half: tokio::net::tcp::OwnedReadHalf,
peer_addr: SocketAddr,
incoming_tx: mpsc::Sender<IncomingFrame>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
let mut buf = BytesMut::with_capacity(8192);
let mut reader = read_half;
loop {
if buf.len() > MAX_FRAME_SIZE * 2 {
tracing::warn!(peer = %peer_addr, buf_len = buf.len(), "read buffer exceeded limit, closing connection");
break;
}
tokio::select! {
_ = cancel.cancelled() => break,
n = reader.read_buf(&mut buf) => {
match n {
Ok(0) | Err(_) => break,
Ok(_) => {
if !Self::process_frames(&mut buf, peer_addr, &incoming_tx) {
break;
}
}
}
}
}
}
});
}
fn process_frames(
buf: &mut BytesMut,
peer_addr: SocketAddr,
incoming_tx: &mpsc::Sender<IncomingFrame>,
) -> bool {
loop {
match decode_frame(buf) {
DecodeResult::Frame(tag, data) => {
if let Err(e) = incoming_tx.try_send(IncomingFrame {
from: peer_addr,
tag,
data,
}) {
tracing::warn!(peer = %peer_addr, error = %e, "incoming frame channel full, dropping frame");
}
}
DecodeResult::NeedMoreData => return true,
DecodeResult::Corrupt => return false,
}
}
}
}
fn encode_frame(tag: ChannelTag, data: &[u8]) -> Option<Vec<u8>> {
match data.len() > MAX_FRAME_SIZE {
true => {
tracing::warn!(
frame_len = data.len(),
max = MAX_FRAME_SIZE,
"refusing to encode oversized frame"
);
None
}
false => {
let len = u32::try_from(data.len()).ok()?;
let mut buf = Vec::with_capacity(5 + data.len());
buf.put_u32(len);
buf.put_u8(tag as u8);
buf.extend_from_slice(data);
Some(buf)
}
}
}
enum DecodeResult {
Frame(ChannelTag, Vec<u8>),
NeedMoreData,
Corrupt,
}
fn decode_frame(buf: &mut BytesMut) -> DecodeResult {
loop {
if buf.len() < 5 {
return DecodeResult::NeedMoreData;
}
let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
if len > MAX_FRAME_SIZE {
tracing::warn!(frame_len = len, max = MAX_FRAME_SIZE, "oversized frame, closing connection");
buf.clear();
return DecodeResult::Corrupt;
}
if buf.len() < 5 + len {
return DecodeResult::NeedMoreData;
}
buf.advance(4);
let tag_byte = buf[0];
buf.advance(1);
let data = buf.split_to(len).to_vec();
match ChannelTag::from_u8(tag_byte) {
Some(tag) => return DecodeResult::Frame(tag, data),
None => {
tracing::debug!(tag = tag_byte, "skipping frame with unknown channel tag");
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn frame_roundtrip() {
let original = b"hello world";
let encoded = encode_frame(ChannelTag::Gossip, original).expect("should encode");
let mut buf = BytesMut::from(&encoded[..]);
match decode_frame(&mut buf) {
DecodeResult::Frame(tag, data) => {
assert_eq!(tag, ChannelTag::Gossip);
assert_eq!(data, original);
}
_ => panic!("expected frame"),
}
assert!(buf.is_empty());
}
#[test]
fn partial_frame_returns_need_more() {
let encoded = encode_frame(ChannelTag::CrdtSync, b"test data").expect("should encode");
let mut buf = BytesMut::from(&encoded[..3]);
assert!(matches!(decode_frame(&mut buf), DecodeResult::NeedMoreData));
}
#[test]
fn multiple_frames() {
let f1 = encode_frame(ChannelTag::Gossip, b"first").expect("should encode");
let f2 = encode_frame(ChannelTag::Direct, b"second").expect("should encode");
let mut buf = BytesMut::new();
buf.extend_from_slice(&f1);
buf.extend_from_slice(&f2);
match decode_frame(&mut buf) {
DecodeResult::Frame(tag1, data1) => {
assert_eq!(tag1, ChannelTag::Gossip);
assert_eq!(data1, b"first");
}
_ => panic!("expected frame"),
}
match decode_frame(&mut buf) {
DecodeResult::Frame(tag2, data2) => {
assert_eq!(tag2, ChannelTag::Direct);
assert_eq!(data2, b"second");
}
_ => panic!("expected frame"),
}
}
}
@@ -0,0 +1,610 @@
use std::sync::Arc;
use std::time::Duration;
use tokio_util::sync::CancellationToken;
use tranquil_infra::{Cache, DistributedRateLimiter};
use tranquil_ripple::{RippleConfig, RippleEngine};
async fn spawn_pair(
shutdown: CancellationToken,
) -> (
(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>),
(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>),
) {
let config_a = RippleConfig {
bind_addr: "127.0.0.1:0".parse().unwrap(),
seed_peers: vec![],
machine_id: 1,
gossip_interval_ms: 100,
cache_max_bytes: 64 * 1024 * 1024,
};
let (cache_a, rl_a, addr_a) = RippleEngine::start(config_a, shutdown.clone())
.await
.expect("node A failed to start");
let config_b = RippleConfig {
bind_addr: "127.0.0.1:0".parse().unwrap(),
seed_peers: vec![addr_a],
machine_id: 2,
gossip_interval_ms: 100,
cache_max_bytes: 64 * 1024 * 1024,
};
let (cache_b, rl_b, _addr_b) = RippleEngine::start(config_b, shutdown.clone())
.await
.expect("node B failed to start");
tokio::time::sleep(Duration::from_millis(2000)).await;
((cache_a, rl_a), (cache_b, rl_b))
}
async fn poll_until<F, Fut>(max_ms: u64, interval_ms: u64, check_fn: F)
where
F: Fn() -> Fut,
Fut: std::future::Future<Output = bool>,
{
let deadline = tokio::time::Instant::now() + Duration::from_millis(max_ms);
let interval = Duration::from_millis(interval_ms);
loop {
if check_fn().await {
return;
}
if tokio::time::Instant::now() + interval > deadline {
panic!("poll_until timed out after {max_ms}ms");
}
tokio::time::sleep(interval).await;
}
}
#[tokio::test]
async fn two_node_cache_convergence() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _rl_a), (cache_b, _rl_b)) = spawn_pair(shutdown.clone()).await;
cache_a
.set("test-key", "hello-from-a", Duration::from_secs(300))
.await
.expect("set on A failed");
assert_eq!(
cache_a.get("test-key").await.as_deref(),
Some("hello-from-a"),
);
let b = cache_b.clone();
poll_until(10_000, 200, || {
let b = b.clone();
async move { b.get("test-key").await.as_deref() == Some("hello-from-a") }
})
.await;
shutdown.cancel();
}
#[tokio::test]
async fn two_node_delete_convergence() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let key = format!("del-{}", uuid::Uuid::new_v4());
cache_a
.set(&key, "to-be-deleted", Duration::from_secs(300))
.await
.expect("set on A failed");
let b = cache_b.clone();
let k = key.clone();
poll_until(10_000, 200, move || {
let b = b.clone();
let k = k.clone();
async move { b.get(&k).await.is_some() }
})
.await;
cache_a.delete(&key).await.expect("delete on A failed");
let b = cache_b.clone();
let k = key.clone();
poll_until(10_000, 200, move || {
let b = b.clone();
let k = k.clone();
async move { b.get(&k).await.is_none() }
})
.await;
shutdown.cancel();
}
#[tokio::test]
async fn two_node_lww_conflict_resolution() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let key = format!("lww-{}", uuid::Uuid::new_v4());
cache_a
.set(&key, "value-from-a", Duration::from_secs(300))
.await
.expect("set on A failed");
cache_b
.set(&key, "value-from-b", Duration::from_secs(300))
.await
.expect("set on B failed");
let a = cache_a.clone();
let b = cache_b.clone();
let k = key.clone();
poll_until(15_000, 200, move || {
let a = a.clone();
let b = b.clone();
let k = k.clone();
async move {
let (va, vb) = tokio::join!(a.get(&k), b.get(&k));
matches!((va, vb), (Some(a), Some(b)) if a == b)
}
})
.await;
let val_a = cache_a.get(&key).await.expect("A should have the key");
let val_b = cache_b.get(&key).await.expect("B should have the key");
assert_eq!(val_a, val_b, "both nodes must agree on the same value after LWW resolution");
shutdown.cancel();
}
#[tokio::test]
async fn two_node_binary_data_convergence() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let key = format!("bin-{}", uuid::Uuid::new_v4());
let payload: Vec<u8> = (0..=255u8).collect();
cache_a
.set_bytes(&key, &payload, Duration::from_secs(300))
.await
.expect("set_bytes on A failed");
let b = cache_b.clone();
let k = key.clone();
let expected = payload.clone();
poll_until(10_000, 200, move || {
let b = b.clone();
let k = k.clone();
let expected = expected.clone();
async move {
b.get_bytes(&k)
.await
.map(|v| v == expected)
.unwrap_or(false)
}
})
.await;
shutdown.cancel();
}
#[tokio::test]
async fn two_node_ttl_expiration() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let key = format!("ttl-{}", uuid::Uuid::new_v4());
cache_a
.set(&key, "ephemeral", Duration::from_secs(2))
.await
.expect("set on A failed");
let b = cache_b.clone();
let k = key.clone();
poll_until(10_000, 200, move || {
let b = b.clone();
let k = k.clone();
async move { b.get(&k).await.is_some() }
})
.await;
tokio::time::sleep(Duration::from_secs(3)).await;
assert!(cache_a.get(&key).await.is_none(), "A should have expired the key");
assert!(cache_b.get(&key).await.is_none(), "B should have expired the key");
shutdown.cancel();
}
#[tokio::test]
async fn two_node_rapid_overwrite_convergence() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let key = format!("rapid-{}", uuid::Uuid::new_v4());
futures::future::join_all((0..50).map(|i| {
let cache = cache_a.clone();
let k = key.clone();
async move {
cache
.set(&k, &format!("value-{i}"), Duration::from_secs(300))
.await
.expect("set failed");
}
}))
.await;
let b = cache_b.clone();
let k = key.clone();
poll_until(10_000, 200, move || {
let b = b.clone();
let k = k.clone();
async move { b.get(&k).await.as_deref() == Some("value-49") }
})
.await;
shutdown.cancel();
}
#[tokio::test]
async fn two_node_many_keys_convergence() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let prefix = format!("many-{}", uuid::Uuid::new_v4());
futures::future::join_all((0..200).map(|i| {
let cache = cache_a.clone();
let p = prefix.clone();
async move {
cache
.set(
&format!("{p}-{i}"),
&format!("val-{i}"),
Duration::from_secs(300),
)
.await
.expect("set failed");
}
}))
.await;
let b = cache_b.clone();
let p = prefix.clone();
poll_until(30_000, 500, move || {
let b = b.clone();
let p = p.clone();
async move {
futures::future::join_all((0..200).map(|i| {
let b = b.clone();
let p = p.clone();
async move { b.get(&format!("{p}-{i}")).await.is_some() }
}))
.await
.into_iter()
.all(|present| present)
}
})
.await;
let results: Vec<Option<String>> = futures::future::join_all((0..200).map(|i| {
let b = cache_b.clone();
let p = prefix.clone();
async move { b.get(&format!("{p}-{i}")).await }
}))
.await;
results.into_iter().enumerate().for_each(|(i, val)| {
assert_eq!(
val.as_deref(),
Some(format!("val-{i}").as_str()),
"key {i} mismatch on B"
);
});
shutdown.cancel();
}
#[tokio::test]
async fn two_node_concurrent_disjoint_writes() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let prefix = format!("disj-{}", uuid::Uuid::new_v4());
let write_a = {
let cache = cache_a.clone();
let p = prefix.clone();
async move {
futures::future::join_all((0..100).map(|i| {
let cache = cache.clone();
let p = p.clone();
async move {
cache
.set(
&format!("{p}-a-{i}"),
&format!("a-{i}"),
Duration::from_secs(300),
)
.await
.expect("set failed");
}
}))
.await;
}
};
let write_b = {
let cache = cache_b.clone();
let p = prefix.clone();
async move {
futures::future::join_all((0..100).map(|i| {
let cache = cache.clone();
let p = p.clone();
async move {
cache
.set(
&format!("{p}-b-{i}"),
&format!("b-{i}"),
Duration::from_secs(300),
)
.await
.expect("set failed");
}
}))
.await;
}
};
tokio::join!(write_a, write_b);
let a = cache_a.clone();
let b = cache_b.clone();
let p = prefix.clone();
poll_until(30_000, 500, move || {
let a = a.clone();
let b = b.clone();
let p = p.clone();
async move {
let a_has_b_keys = futures::future::join_all((0..100).map(|i| {
let a = a.clone();
let p = p.clone();
async move { a.get(&format!("{p}-b-{i}")).await.is_some() }
}))
.await
.into_iter()
.all(|v| v);
let b_has_a_keys = futures::future::join_all((0..100).map(|i| {
let b = b.clone();
let p = p.clone();
async move { b.get(&format!("{p}-a-{i}")).await.is_some() }
}))
.await
.into_iter()
.all(|v| v);
a_has_b_keys && b_has_a_keys
}
})
.await;
shutdown.cancel();
}
#[tokio::test]
async fn two_node_concurrent_same_key_writes() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((cache_a, _), (cache_b, _)) = spawn_pair(shutdown.clone()).await;
let prefix = format!("same-{}", uuid::Uuid::new_v4());
let write_a = {
let cache = cache_a.clone();
let p = prefix.clone();
async move {
futures::future::join_all((0..50).map(|i| {
let cache = cache.clone();
let p = p.clone();
async move {
cache
.set(
&format!("{p}-{i}"),
&format!("a-{i}"),
Duration::from_secs(300),
)
.await
.expect("set failed");
}
}))
.await;
}
};
let write_b = {
let cache = cache_b.clone();
let p = prefix.clone();
async move {
futures::future::join_all((0..50).map(|i| {
let cache = cache.clone();
let p = p.clone();
async move {
cache
.set(
&format!("{p}-{i}"),
&format!("b-{i}"),
Duration::from_secs(300),
)
.await
.expect("set failed");
}
}))
.await;
}
};
tokio::join!(write_a, write_b);
let a = cache_a.clone();
let b = cache_b.clone();
let p = prefix.clone();
poll_until(15_000, 200, move || {
let a = a.clone();
let b = b.clone();
let p = p.clone();
async move {
futures::future::join_all((0..50).map(|i| {
let a = a.clone();
let b = b.clone();
let p = p.clone();
async move {
let va = a.get(&format!("{p}-{i}")).await.unwrap_or_default();
let vb = b.get(&format!("{p}-{i}")).await.unwrap_or_default();
!va.is_empty() && va == vb
}
}))
.await
.into_iter()
.all(|v| v)
}
})
.await;
let results: Vec<(String, String)> = futures::future::join_all((0..50).map(|i| {
let a = cache_a.clone();
let b = cache_b.clone();
let p = prefix.clone();
async move {
let va = a.get(&format!("{p}-{i}")).await.unwrap_or_default();
let vb = b.get(&format!("{p}-{i}")).await.unwrap_or_default();
(va, vb)
}
}))
.await;
results.into_iter().enumerate().for_each(|(i, (va, vb))| {
assert_eq!(va, vb, "key {i}: nodes disagree (A={va}, B={vb})");
});
shutdown.cancel();
}
#[tokio::test]
async fn two_node_rate_limit_split_increment() {
tracing_subscriber::fmt()
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
.with_test_writer()
.try_init()
.ok();
let shutdown = CancellationToken::new();
let ((_, rl_a), (_, rl_b)) = spawn_pair(shutdown.clone()).await;
let key = format!("rl-split-{}", uuid::Uuid::new_v4());
let limit: u32 = 200;
let window_ms: u64 = 600_000;
futures::future::join_all((0..40).map(|_| {
let rl = rl_a.clone();
let k = key.clone();
async move {
let allowed = rl.check_rate_limit(&k, limit, window_ms).await;
assert!(allowed, "should be allowed within limit");
}
}))
.await;
futures::future::join_all((0..30).map(|_| {
let rl = rl_b.clone();
let k = key.clone();
async move {
let allowed = rl.check_rate_limit(&k, limit, window_ms).await;
assert!(allowed, "should be allowed within limit");
}
}))
.await;
let rl_peek = rl_a.clone();
let k = key.clone();
poll_until(15_000, 200, move || {
let rl = rl_peek.clone();
let k = k.clone();
async move { rl.peek_rate_limit_count(&k, window_ms).await >= 70 }
})
.await;
let mut remaining = 0u32;
loop {
if !rl_a.check_rate_limit(&key, limit, window_ms).await {
break;
}
remaining += 1;
if remaining > limit {
panic!("rate limiter never denied - convergence failed");
}
}
let expected_remaining = limit - 70;
let margin = 15;
assert!(
remaining.abs_diff(expected_remaining) <= margin,
"expected ~{expected_remaining} remaining hits, got {remaining} (margin={margin})"
);
shutdown.cancel();
}
+3 -5
View File
@@ -1,6 +1,6 @@
[Unit]
Description=Tranquil PDS AT Protocol PDS
After=tranquil-pds-db.service tranquil-pds-minio.service tranquil-pds-valkey.service
After=tranquil-pds-db.service
[Container]
ContainerName=tranquil-pds-app
Image=localhost/tranquil-pds:latest
@@ -8,10 +8,8 @@ Pod=tranquil-pds.pod
EnvironmentFile=/srv/tranquil-pds/config/tranquil-pds.env
Environment=SERVER_HOST=0.0.0.0
Environment=SERVER_PORT=3000
Environment=S3_ENDPOINT=http://localhost:9000
Environment=AWS_REGION=us-east-1
Environment=S3_BUCKET=pds-blobs
Environment=VALKEY_URL=redis://localhost:6379
Volume=/srv/tranquil-pds/blobs:/var/lib/tranquil/blobs:Z
Volume=/srv/tranquil-pds/backups:/var/lib/tranquil/backups:Z
HealthCmd=wget -q --spider http://localhost:3000/xrpc/_health
HealthInterval=30s
HealthTimeout=10s
+7 -66
View File
@@ -10,23 +10,18 @@ services:
SERVER_PORT: "3000"
PDS_HOSTNAME: "${PDS_HOSTNAME:?PDS_HOSTNAME is required}"
DATABASE_URL: "postgres://tranquil_pds:${DB_PASSWORD:?DB_PASSWORD is required}@db:5432/pds"
S3_ENDPOINT: "http://minio:9000"
AWS_REGION: "us-east-1"
S3_BUCKET: "pds-blobs"
AWS_ACCESS_KEY_ID: "${MINIO_ROOT_USER:-minioadmin}"
AWS_SECRET_ACCESS_KEY: "${MINIO_ROOT_PASSWORD:?MINIO_ROOT_PASSWORD is required}"
VALKEY_URL: "redis://valkey:6379"
BLOB_STORAGE_PATH: "/var/lib/tranquil/blobs"
BACKUP_STORAGE_PATH: "/var/lib/tranquil/backups"
JWT_SECRET: "${JWT_SECRET:?JWT_SECRET is required (min 32 chars)}"
DPOP_SECRET: "${DPOP_SECRET:?DPOP_SECRET is required (min 32 chars)}"
MASTER_KEY: "${MASTER_KEY:?MASTER_KEY is required (min 32 chars)}"
CRAWLERS: "${CRAWLERS:-https://bsky.network}"
volumes:
- blob_data:/var/lib/tranquil/blobs
- backup_data:/var/lib/tranquil/backups
depends_on:
db:
condition: service_healthy
minio:
condition: service_healthy
valkey:
condition: service_healthy
healthcheck:
test: ["CMD", "wget", "-q", "--spider", "http://localhost:3000/xrpc/_health"]
interval: 30s
@@ -81,60 +76,6 @@ services:
reservations:
memory: 128M
minio:
image: cgr.dev/chainguard/minio:latest
restart: unless-stopped
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: "${MINIO_ROOT_USER:-minioadmin}"
MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD:?MINIO_ROOT_PASSWORD is required}"
volumes:
- minio_data:/data
deploy:
resources:
limits:
memory: 512M
reservations:
memory: 128M
minio-init:
image: cgr.dev/chainguard/minio-client:latest-dev
depends_on:
- minio
entrypoint: >
/bin/sh -c "
for i in 1 2 3 4 5 6 7 8 9 10; do
mc alias set local http://minio:9000 $${MINIO_ROOT_USER} $${MINIO_ROOT_PASSWORD} && break;
echo 'Waiting for minio...'; sleep 2;
done;
mc mb --ignore-existing local/pds-blobs;
mc mb --ignore-existing local/pds-backups;
mc anonymous set none local/pds-blobs;
exit 0;
"
environment:
MINIO_ROOT_USER: "${MINIO_ROOT_USER:-minioadmin}"
MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD:?MINIO_ROOT_PASSWORD is required}"
valkey:
image: valkey/valkey:9-alpine
restart: unless-stopped
command: valkey-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru
volumes:
- valkey_data:/data
healthcheck:
test: ["CMD", "valkey-cli", "ping"]
interval: 10s
timeout: 5s
retries: 3
start_period: 5s
deploy:
resources:
limits:
memory: 300M
reservations:
memory: 64M
nginx:
image: nginx:1.29-alpine
restart: unless-stopped
@@ -180,7 +121,7 @@ services:
volumes:
postgres_data:
minio_data:
valkey_data:
blob_data:
backup_data:
prometheus_data:
acme_challenge:
+5 -25
View File
@@ -10,12 +10,11 @@ services:
- ./.env
environment:
DATABASE_URL: postgres://postgres:postgres@db:5432/pds
S3_ENDPOINT: http://objsto:9000
VALKEY_URL: redis://cache:6379
volumes:
- blob_data:/var/lib/tranquil/blobs
- backup_data:/var/lib/tranquil/backups
depends_on:
- db
- objsto
- cache
frontend:
build:
@@ -38,25 +37,6 @@ services:
volumes:
- postgres_data:/var/lib/postgresql
objsto:
image: cgr.dev/chainguard/minio:latest
ports:
- "9000:9000"
- "9001:9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
volumes:
- minio_data:/data
command: server /data --console-address ":9001"
cache:
image: valkey/valkey:9-alpine
ports:
- "6379:6379"
volumes:
- valkey_data:/data
prometheus:
image: prom/prometheus:v3.8.0
ports:
@@ -72,6 +52,6 @@ services:
volumes:
postgres_data:
minio_data:
valkey_data:
blob_data:
backup_data:
prometheus_data:
+16 -11
View File
@@ -43,7 +43,7 @@ For production setups with proper service management, continue to either the Deb
## Standalone Containers (No Compose)
If you already have postgres and valkey running on the host (eg., from the [Debian install guide](install-debian.md)), you can run just the app containers.
If you already have postgres running on the host (eg. from the [Debian install guide](install-debian.md)), you can run just the app containers.
Build the images:
```sh
@@ -51,7 +51,7 @@ podman build -t tranquil-pds:latest .
podman build -t tranquil-pds-frontend:latest ./frontend
```
Run the backend with host networking (so it can access postgres/valkey on localhost) and mount the blob storage:
Run the backend with host networking (so it can access postgres on localhost) and mount the blob storage:
```sh
podman run -d --name tranquil-pds \
--network=host \
@@ -106,7 +106,7 @@ apt install -y podman
```bash
mkdir -p /etc/containers/systemd
mkdir -p /srv/tranquil-pds/{postgres,valkey,blobs,backups,certs,acme,config}
mkdir -p /srv/tranquil-pds/{postgres,blobs,backups,certs,acme,config}
```
## Create Environment File
@@ -127,10 +127,15 @@ For quadlets, also add `DATABASE_URL` with the full connection string (systemd d
Copy the quadlet files from the repository:
```bash
cp /opt/tranquil-pds/deploy/quadlets/*.pod /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/*.container /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/tranquil-pds.pod /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/tranquil-pds-db.container /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/tranquil-pds-app.container /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/tranquil-pds-frontend.container /etc/containers/systemd/
cp /opt/tranquil-pds/deploy/quadlets/tranquil-pds-nginx.container /etc/containers/systemd/
```
Optional quadlets for valkey and minio are also available in `deploy/quadlets/` if you need them.
Note: Systemd doesn't support shell-style variable expansion in `Environment=` lines. The quadlet files expect DATABASE_URL to be set in the environment file.
## Create nginx Configuration
@@ -160,7 +165,7 @@ echo "$DB_PASSWORD" | podman secret create tranquil-pds-db-password -
```bash
systemctl daemon-reload
systemctl start tranquil-pds-db tranquil-pds-valkey
systemctl start tranquil-pds-db
sleep 10
```
@@ -172,7 +177,7 @@ DATABASE_URL="postgres://tranquil_pds:your-db-password@localhost:5432/pds" sqlx
## Obtain Wildcard SSL Certificate
User handles are served as subdomains (eg., `alice.pds.example.com`), so you need a wildcard certificate. Wildcard certs require DNS-01 validation.
User handles are served as subdomains (eg. `alice.pds.example.com`), so you need a wildcard certificate. Wildcard certs require DNS-01 validation.
Create temporary self-signed cert to start services:
```bash
@@ -195,7 +200,7 @@ podman run --rm -it \
Follow the prompts to add TXT records to your DNS. Note: manual mode doesn't auto-renew.
For automated renewal, use a DNS provider plugin (eg., cloudflare, route53).
For automated renewal, use a DNS provider plugin (eg. cloudflare, route53).
Link certificates and restart:
```bash
@@ -207,7 +212,7 @@ systemctl restart tranquil-pds-nginx
## Enable All Services
```bash
systemctl enable tranquil-pds-db tranquil-pds-valkey tranquil-pds-app tranquil-pds-frontend tranquil-pds-nginx
systemctl enable tranquil-pds-db tranquil-pds-app tranquil-pds-frontend tranquil-pds-nginx
```
## Configure Firewall
@@ -252,7 +257,7 @@ rc-service podman start
```sh
mkdir -p /srv/tranquil-pds/{data,config}
mkdir -p /srv/tranquil-pds/data/{postgres,valkey,blobs,backups,certs,acme}
mkdir -p /srv/tranquil-pds/data/{postgres,blobs,backups,certs,acme}
```
## Clone Repository and Build Images
@@ -346,7 +351,7 @@ DATABASE_URL="postgres://tranquil_pds:$DB_PASSWORD@$DB_IP:5432/pds" sqlx migrate
## Obtain Wildcard SSL Certificate
User handles are served as subdomains (eg., `alice.pds.example.com`), so you need a wildcard certificate. Wildcard certs require DNS-01 validation.
User handles are served as subdomains (eg. `alice.pds.example.com`), so you need a wildcard certificate. Wildcard certs require DNS-01 validation.
Create temporary self-signed cert to start services:
```sh
-8
View File
@@ -46,14 +46,6 @@ mkdir -p /var/lib/tranquil/blobs /var/lib/tranquil/backups
We'll set ownership after creating the service user.
## Install valkey
```bash
apt install -y valkey
systemctl enable valkey-server
systemctl start valkey-server
```
## Install deno (for frontend build)
```bash
-3
View File
@@ -3,7 +3,6 @@
If you're reaching for kubernetes for this app, you're experienced enough to know how to spin up:
- cloudnativepg (or your preferred postgres operator)
- valkey
- a PersistentVolume for blob storage
- the app itself (it's just a container with some env vars)
@@ -13,7 +12,6 @@ The container image expects:
- `DATABASE_URL` - postgres connection string
- `BLOB_STORAGE_PATH` - path to blob storage (mount a PV here)
- `BACKUP_STORAGE_PATH` - path for repo backups (optional but recommended)
- `VALKEY_URL` - redis:// connection string
- `PDS_HOSTNAME` - your PDS hostname (without protocol)
- `JWT_SECRET`, `DPOP_SECRET`, `MASTER_KEY` - generate with `openssl rand -base64 48`
- `CRAWLERS` - typically `https://bsky.network`
@@ -41,4 +39,3 @@ data:
</body>
</html>
```
-11
View File
@@ -194,16 +194,6 @@ sudo -u postgres psql -c "CREATE DATABASE pds OWNER tranquil_pds;" 2>/dev/null |
sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE pds TO tranquil_pds;"
log_success "postgres configured"
log_info "Installing valkey..."
apt install -y valkey 2>/dev/null || {
log_warn "valkey not in repos, installing redis..."
apt install -y redis-server
systemctl enable redis-server
systemctl start redis-server
}
systemctl enable valkey-server 2>/dev/null || true
systemctl start valkey-server 2>/dev/null || true
log_info "Creating blob storage directories..."
mkdir -p /var/lib/tranquil/blobs /var/lib/tranquil/backups
log_success "Blob storage directories created"
@@ -313,7 +303,6 @@ DATABASE_MAX_CONNECTIONS=100
DATABASE_MIN_CONNECTIONS=10
BLOB_STORAGE_PATH=/var/lib/tranquil/blobs
BACKUP_STORAGE_PATH=/var/lib/tranquil/backups
VALKEY_URL=redis://localhost:6379
JWT_SECRET=${JWT_SECRET}
DPOP_SECRET=${DPOP_SECRET}
MASTER_KEY=${MASTER_KEY}
+8 -1
View File
@@ -16,7 +16,14 @@ echo "Running database migrations..."
sqlx database create 2>/dev/null || true
sqlx migrate run --source "$PROJECT_DIR/migrations"
echo ""
ulimit -n 65536
echo "Building test binaries..."
cargo test --no-run 2>&1 | tail -1
echo "Running tests..."
echo ""
ulimit -n 65536
cargo nextest run "$@"
echo ""
echo "All tests passed."
+5 -50
View File
@@ -32,7 +32,7 @@ start_infra() {
echo "Stale infra file found, cleaning up..."
rm -f "$INFRA_FILE"
fi
$CONTAINER_CMD rm -f "${CONTAINER_PREFIX}-postgres" "${CONTAINER_PREFIX}-minio" "${CONTAINER_PREFIX}-valkey" 2>/dev/null || true
$CONTAINER_CMD rm -f "${CONTAINER_PREFIX}-postgres" 2>/dev/null || true
echo "Starting PostgreSQL..."
$CONTAINER_CMD run -d \
--name "${CONTAINER_PREFIX}-postgres" \
@@ -43,25 +43,7 @@ start_infra() {
--label tranquil_pds_test=true \
postgres:18-alpine \
-c max_connections=500 >/dev/null
echo "Starting MinIO..."
$CONTAINER_CMD run -d \
--name "${CONTAINER_PREFIX}-minio" \
-e MINIO_ROOT_USER=minioadmin \
-e MINIO_ROOT_PASSWORD=minioadmin \
-p 9000 \
--label tranquil_pds_test=true \
cgr.dev/chainguard/minio:latest server /data >/dev/null
echo "Starting Valkey..."
$CONTAINER_CMD run -d \
--name "${CONTAINER_PREFIX}-valkey" \
-P \
--label tranquil_pds_test=true \
valkey/valkey:9-alpine >/dev/null
echo "Waiting for services to be ready..."
sleep 2
PG_PORT=$($CONTAINER_CMD port "${CONTAINER_PREFIX}-postgres" 5432 | head -1 | cut -d: -f2)
MINIO_PORT=$($CONTAINER_CMD port "${CONTAINER_PREFIX}-minio" 9000 | head -1 | cut -d: -f2)
VALKEY_PORT=$($CONTAINER_CMD port "${CONTAINER_PREFIX}-valkey" 6379 | head -1 | cut -d: -f2)
for i in {1..30}; do
if $CONTAINER_CMD exec "${CONTAINER_PREFIX}-postgres" pg_isready -U postgres >/dev/null 2>&1; then
break
@@ -69,37 +51,10 @@ start_infra() {
echo "Waiting for PostgreSQL... ($i/30)"
sleep 1
done
for i in {1..30}; do
if curl -s "http://127.0.0.1:${MINIO_PORT}/minio/health/live" >/dev/null 2>&1; then
break
fi
echo "Waiting for MinIO... ($i/30)"
sleep 1
done
for i in {1..30}; do
if $CONTAINER_CMD exec "${CONTAINER_PREFIX}-valkey" valkey-cli ping 2>/dev/null | grep -q PONG; then
break
fi
echo "Waiting for Valkey... ($i/30)"
sleep 1
done
echo "Creating MinIO buckets..."
$CONTAINER_CMD run --rm --network host \
-e MC_HOST_minio="http://minioadmin:minioadmin@127.0.0.1:${MINIO_PORT}" \
cgr.dev/chainguard/minio-client:latest-dev mb minio/test-bucket --ignore-existing >/dev/null 2>&1 || true
$CONTAINER_CMD run --rm --network host \
-e MC_HOST_minio="http://minioadmin:minioadmin@127.0.0.1:${MINIO_PORT}" \
cgr.dev/chainguard/minio-client:latest-dev mb minio/test-backups --ignore-existing >/dev/null 2>&1 || true
PG_PORT=$($CONTAINER_CMD port "${CONTAINER_PREFIX}-postgres" 5432 | head -1 | cut -d: -f2)
cat > "$INFRA_FILE" << EOF
export DATABASE_URL="postgres://postgres:postgres@127.0.0.1:${PG_PORT}/postgres"
export TEST_DB_PORT="${PG_PORT}"
export S3_ENDPOINT="http://127.0.0.1:${MINIO_PORT}"
export S3_BUCKET="test-bucket"
export BACKUP_S3_BUCKET="test-backups"
export AWS_ACCESS_KEY_ID="minioadmin"
export AWS_SECRET_ACCESS_KEY="minioadmin"
export AWS_REGION="us-east-1"
export VALKEY_URL="redis://127.0.0.1:${VALKEY_PORT}"
export TRANQUIL_PDS_TEST_INFRA_READY="1"
export TRANQUIL_PDS_ALLOW_INSECURE_SECRETS="1"
export SKIP_IMPORT_VERIFICATION="true"
@@ -113,8 +68,9 @@ EOF
}
stop_infra() {
echo "Stopping test infrastructure..."
$CONTAINER_CMD rm -f "${CONTAINER_PREFIX}-postgres" "${CONTAINER_PREFIX}-minio" "${CONTAINER_PREFIX}-valkey" 2>/dev/null || true
$CONTAINER_CMD rm -f "${CONTAINER_PREFIX}-postgres" 2>/dev/null || true
rm -f "$INFRA_FILE"
rm -rf "${TMPDIR:-/tmp}"/tranquil-pds-test-* 2>/dev/null || true
echo "Infrastructure stopped."
}
status_infra() {
@@ -124,7 +80,6 @@ status_infra() {
echo "Config file: $INFRA_FILE"
source "$INFRA_FILE"
echo "Database URL: $DATABASE_URL"
echo "S3 Endpoint: $S3_ENDPOINT"
else
echo "Config file: NOT FOUND"
fi
@@ -158,7 +113,7 @@ case "${1:-}" in
echo "Usage: $0 {start|stop|restart|status|env}"
echo ""
echo "Commands:"
echo " start - Start test infrastructure (Postgres, MinIO, Valkey)"
echo " start - Start test infrastructure (Postgres)"
echo " stop - Stop and remove test containers"
echo " restart - Stop then start infrastructure"
echo " status - Show infrastructure status"