mirror of
https://tangled.org/tranquil.farm/tranquil-pds
synced 2026-08-17 06:46:07 +00:00
787 lines
22 KiB
Rust
787 lines
22 KiB
Rust
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,
|
|
cluster_key: None,
|
|
allow_insecure: false,
|
|
};
|
|
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,
|
|
cluster_key: None,
|
|
allow_insecure: false,
|
|
};
|
|
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();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn two_node_partition_recovery() {
|
|
tracing_subscriber::fmt()
|
|
.with_max_level(tracing_subscriber::filter::LevelFilter::DEBUG)
|
|
.with_test_writer()
|
|
.try_init()
|
|
.ok();
|
|
|
|
let shutdown = CancellationToken::new();
|
|
|
|
let config_a = RippleConfig {
|
|
bind_addr: "127.0.0.1:0".parse().unwrap(),
|
|
seed_peers: vec![],
|
|
machine_id: 100,
|
|
gossip_interval_ms: 100,
|
|
cache_max_bytes: 64 * 1024 * 1024,
|
|
cluster_key: None,
|
|
allow_insecure: false,
|
|
};
|
|
let (cache_a, _rl_a, addr_a) = RippleEngine::start(config_a, shutdown.clone())
|
|
.await
|
|
.expect("node A failed to start");
|
|
|
|
futures::future::join_all((0..50).map(|i| {
|
|
let cache = cache_a.clone();
|
|
async move {
|
|
cache
|
|
.set(
|
|
&format!("pre-{i}"),
|
|
&format!("val-{i}"),
|
|
Duration::from_secs(300),
|
|
)
|
|
.await
|
|
.expect("set on A failed");
|
|
}
|
|
}))
|
|
.await;
|
|
|
|
let config_b = RippleConfig {
|
|
bind_addr: "127.0.0.1:0".parse().unwrap(),
|
|
seed_peers: vec![addr_a],
|
|
machine_id: 200,
|
|
gossip_interval_ms: 100,
|
|
cache_max_bytes: 64 * 1024 * 1024,
|
|
cluster_key: None,
|
|
allow_insecure: false,
|
|
};
|
|
let (cache_b, _rl_b, _addr_b) = RippleEngine::start(config_b, shutdown.clone())
|
|
.await
|
|
.expect("node B failed to start");
|
|
|
|
let b = cache_b.clone();
|
|
poll_until(15_000, 200, move || {
|
|
let b = b.clone();
|
|
async move {
|
|
futures::future::join_all((0..50).map(|i| {
|
|
let b = b.clone();
|
|
async move { b.get(&format!("pre-{i}")).await.is_some() }
|
|
}))
|
|
.await
|
|
.into_iter()
|
|
.all(|present| present)
|
|
}
|
|
})
|
|
.await;
|
|
|
|
futures::future::join_all((0..50).map(|i| {
|
|
let cache = cache_b.clone();
|
|
async move {
|
|
cache
|
|
.set(
|
|
&format!("post-{i}"),
|
|
&format!("bval-{i}"),
|
|
Duration::from_secs(300),
|
|
)
|
|
.await
|
|
.expect("set on B failed");
|
|
}
|
|
}))
|
|
.await;
|
|
|
|
let a = cache_a.clone();
|
|
poll_until(15_000, 200, move || {
|
|
let a = a.clone();
|
|
async move {
|
|
futures::future::join_all((0..50).map(|i| {
|
|
let a = a.clone();
|
|
async move { a.get(&format!("post-{i}")).await.is_some() }
|
|
}))
|
|
.await
|
|
.into_iter()
|
|
.all(|present| present)
|
|
}
|
|
})
|
|
.await;
|
|
|
|
shutdown.cancel();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn two_node_stress_concurrent_load() {
|
|
tracing_subscriber::fmt()
|
|
.with_max_level(tracing_subscriber::filter::LevelFilter::INFO)
|
|
.with_test_writer()
|
|
.try_init()
|
|
.ok();
|
|
|
|
let shutdown = CancellationToken::new();
|
|
let ((cache_a, rl_a), (cache_b, rl_b)) = spawn_pair(shutdown.clone()).await;
|
|
|
|
let tasks: Vec<tokio::task::JoinHandle<()>> = (0u32..8)
|
|
.map(|task_id| {
|
|
let cache = match task_id < 4 {
|
|
true => cache_a.clone(),
|
|
false => cache_b.clone(),
|
|
};
|
|
let rl = match task_id < 4 {
|
|
true => rl_a.clone(),
|
|
false => rl_b.clone(),
|
|
};
|
|
tokio::spawn(async move {
|
|
let value = vec![0xABu8; 1024];
|
|
futures::future::join_all((0u32..500).map(|op| {
|
|
let cache = cache.clone();
|
|
let rl = rl.clone();
|
|
let value = value.clone();
|
|
async move {
|
|
let key_idx = op % 100;
|
|
let key = format!("stress-{task_id}-{key_idx}");
|
|
match op % 4 {
|
|
0 | 1 => {
|
|
cache
|
|
.set_bytes(&key, &value, Duration::from_secs(120))
|
|
.await
|
|
.expect("set_bytes failed");
|
|
}
|
|
2 => {
|
|
let _ = cache.get(&key).await;
|
|
}
|
|
_ => {
|
|
let _ = rl.check_rate_limit(&key, 1000, 60_000).await;
|
|
}
|
|
}
|
|
}
|
|
}))
|
|
.await;
|
|
})
|
|
})
|
|
.collect();
|
|
|
|
let results = tokio::time::timeout(Duration::from_secs(30), futures::future::join_all(tasks))
|
|
.await
|
|
.expect("stress test timed out after 30s");
|
|
|
|
results.into_iter().enumerate().for_each(|(i, r)| {
|
|
r.unwrap_or_else(|e| panic!("task {i} panicked: {e}"));
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_secs(12)).await;
|
|
|
|
shutdown.cancel();
|
|
}
|