use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use hydracache_core::{
CacheError, CacheEventKind, CacheEventOptions, CacheEventOrigin, CacheOptions, Result,
};
use tokio::sync::watch;
use crate::tests::common::{user, User};
use crate::{
CacheInvalidation, CacheInvalidationBus, CacheInvalidationMessage, CacheInvalidationReceive,
CacheInvalidationReceiver, HydraCache, InMemoryInvalidationBus,
};
#[derive(Debug, Clone)]
struct ClosingBus;
#[async_trait]
impl CacheInvalidationBus for ClosingBus {
async fn publish(&self, _message: CacheInvalidationMessage) -> Result<()> {
Ok(())
}
fn subscribe(&self) -> Box<dyn CacheInvalidationReceiver> {
Box::new(ClosingReceiver)
}
}
struct ClosingReceiver;
#[async_trait]
impl CacheInvalidationReceiver for ClosingReceiver {
async fn recv(&mut self) -> CacheInvalidationReceive {
CacheInvalidationReceive::Closed
}
}
#[derive(Debug, Clone)]
struct FailingPublishBus;
#[async_trait]
impl CacheInvalidationBus for FailingPublishBus {
async fn publish(&self, _message: CacheInvalidationMessage) -> Result<()> {
Err(CacheError::Backend("publish failed".to_owned()))
}
fn subscribe(&self) -> Box<dyn CacheInvalidationReceiver> {
Box::new(ClosingReceiver)
}
}
#[derive(Debug, Clone)]
struct LagThenCloseBus;
#[async_trait]
impl CacheInvalidationBus for LagThenCloseBus {
async fn publish(&self, _message: CacheInvalidationMessage) -> Result<()> {
Ok(())
}
fn subscribe(&self) -> Box<dyn CacheInvalidationReceiver> {
Box::new(LagThenCloseReceiver { polled: false })
}
}
struct LagThenCloseReceiver {
polled: bool,
}
#[async_trait]
impl CacheInvalidationReceiver for LagThenCloseReceiver {
async fn recv(&mut self) -> CacheInvalidationReceive {
if self.polled {
CacheInvalidationReceive::Closed
} else {
self.polled = true;
CacheInvalidationReceive::Lagged(3)
}
}
}
async fn wait_until<F>(mut condition: F)
where
F: FnMut() -> bool,
{
tokio::time::timeout(Duration::from_millis(500), async {
loop {
if condition() {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("condition should become true before timeout");
}
async fn wait_until_absent(cache: &HydraCache, key: &str) {
tokio::time::timeout(Duration::from_millis(500), async {
loop {
if !cache.contains_key(key).await {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("key should become absent before timeout");
}
#[tokio::test]
async fn shared_bus_propagates_tag_invalidations_between_cache_instances() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let source = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("source")
.build();
let target = HydraCache::local()
.shared_invalidation_bus(bus)
.invalidation_node_id("target")
.build();
let mut target_events =
target.subscribe(CacheEventOptions::mutations().origin(CacheEventOrigin::DistributedBus));
source
.put("user:42", user(42), CacheOptions::new().tag("users"))
.await
.unwrap();
target
.put("user:42", user(42), CacheOptions::new().tag("users"))
.await
.unwrap();
assert_eq!(source.invalidate_tag("users").await.unwrap(), 1);
wait_until_absent(&target, "user:42").await;
let event = tokio::time::timeout(Duration::from_millis(500), target_events.recv())
.await
.expect("remote event should arrive")
.expect("subscription should stay open");
assert_eq!(event.kind(), CacheEventKind::TagInvalidated);
assert_eq!(event.origin(), CacheEventOrigin::DistributedBus);
assert_eq!(event.tag(), Some("users"));
assert_eq!(event.affected_keys(), Some(1));
assert_eq!(source.stats().distributed_invalidations_published, 1);
assert_eq!(source.stats().distributed_invalidations_received, 0);
assert_eq!(target.stats().distributed_invalidations_received, 1);
assert_eq!(target.stats().distributed_invalidations_applied, 1);
}
#[tokio::test]
async fn key_invalidation_is_published_even_when_source_does_not_hold_key() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let source = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("source")
.build();
let target = HydraCache::local()
.shared_invalidation_bus(bus)
.invalidation_node_id("target")
.build();
let mut target_events = target.subscribe(
CacheEventOptions::mutations()
.include_kind(CacheEventKind::KeyInvalidated)
.origin(CacheEventOrigin::DistributedBus),
);
target
.put("user:7", user(7), CacheOptions::new())
.await
.unwrap();
assert!(!source.invalidate_key("user:7").await.unwrap());
wait_until_absent(&target, "user:7").await;
let event = tokio::time::timeout(Duration::from_millis(500), target_events.recv())
.await
.expect("remote key invalidation should arrive")
.expect("subscription should stay open");
assert_eq!(event.key(), Some("user:7"));
assert_eq!(source.stats().distributed_invalidations_published, 1);
assert_eq!(target.stats().distributed_invalidations_applied, 1);
}
#[tokio::test]
async fn shared_bus_propagates_flush_without_echoing_back_to_source() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let source = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("source")
.build();
let target = HydraCache::local()
.shared_invalidation_bus(bus)
.invalidation_node_id("target")
.build();
source
.put("source-only", user(1), CacheOptions::new())
.await
.unwrap();
target
.put("target-only", user(2), CacheOptions::new())
.await
.unwrap();
source.flush().await.unwrap();
wait_until_absent(&target, "target-only").await;
wait_until(|| source.stats().distributed_invalidations_published == 1).await;
assert!(!source.contains_key("source-only").await);
assert_eq!(source.stats().distributed_invalidations_received, 0);
assert_eq!(target.stats().distributed_invalidations_received, 1);
assert_eq!(target.stats().distributed_invalidations_applied, 1);
}
#[tokio::test]
async fn typed_cache_invalidation_uses_physical_keys_on_the_shared_bus() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let source = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("source")
.build();
let target = HydraCache::local()
.shared_invalidation_bus(bus)
.invalidation_node_id("target")
.build();
let source_users = source.typed::<User>("users");
let target_users = target.typed::<User>("users");
target_users
.put("42", user(42), CacheOptions::new())
.await
.unwrap();
assert!(!source_users.invalidate_key("42").await.unwrap());
wait_until_absent(&target, "users:42").await;
assert_eq!(target_users.get("42").await.unwrap(), None);
assert_eq!(source.stats().distributed_invalidations_published, 1);
assert_eq!(target.stats().distributed_invalidations_applied, 1);
}
#[tokio::test]
async fn owned_bus_builder_generates_observable_node_id_and_handles_closed_receiver() {
let cache = HydraCache::local().invalidation_bus(ClosingBus).build();
assert!(cache.invalidation_node_id().starts_with("hydracache-node-"));
wait_until(|| cache.stats().distributed_invalidation_receiver_closed == 1).await;
assert_eq!(cache.stats().distributed_invalidations_received, 0);
assert_eq!(cache.stats().distributed_invalidations_applied, 0);
assert_eq!(cache.stats().distributed_invalidation_receiver_closed, 1);
assert!(cache.stats().has_distributed_invalidation_bus_issues());
}
#[tokio::test]
async fn listener_spawn_without_bus_is_a_noop() {
let cache = HydraCache::local().build();
let (_shutdown, receiver) = watch::channel(false);
cache.spawn_invalidation_listener(receiver);
assert!(cache.invalidation_node_id().starts_with("hydracache-node-"));
}
#[tokio::test]
async fn manual_listener_can_shutdown_without_processing_messages() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let cache = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("manual")
.build();
let (shutdown, receiver) = watch::channel(false);
cache.spawn_invalidation_listener(receiver);
tokio::task::yield_now().await;
shutdown.send(true).unwrap();
tokio::task::yield_now().await;
assert_eq!(cache.stats().distributed_invalidations_received, 0);
}
#[tokio::test]
async fn listener_exits_when_cache_inner_is_dropped() {
let bus = Arc::new(InMemoryInvalidationBus::new(16));
let cache = HydraCache::local()
.shared_invalidation_bus(bus.clone())
.invalidation_node_id("transient")
.build();
let (_shutdown, receiver) = watch::channel(false);
cache.spawn_invalidation_listener(receiver);
drop(cache);
bus.publish(CacheInvalidationMessage::new(
"remote",
CacheInvalidation::flush(),
))
.await
.unwrap();
tokio::task::yield_now().await;
}
#[tokio::test]
async fn self_originated_bus_messages_are_ignored_by_listener() {
let cache = HydraCache::local()
.invalidation_bus(InMemoryInvalidationBus::new(16))
.invalidation_node_id("self")
.build();
cache
.put("self-key", user(9), CacheOptions::new())
.await
.unwrap();
assert!(cache.remove("self-key").await.unwrap());
tokio::time::sleep(Duration::from_millis(20)).await;
assert_eq!(cache.stats().distributed_invalidations_published, 1);
assert_eq!(cache.stats().distributed_invalidations_received, 0);
assert_eq!(cache.stats().distributed_invalidations_applied, 0);
}
#[tokio::test]
async fn publish_failures_are_reported_in_stats_and_returned_to_caller() {
let cache = HydraCache::local()
.invalidation_bus(FailingPublishBus)
.invalidation_node_id("failing")
.build();
let error = cache.invalidate_tag("users").await.unwrap_err();
assert!(error.to_string().contains("publish failed"));
assert_eq!(cache.stats().distributed_invalidations_published, 0);
assert_eq!(cache.stats().distributed_invalidation_publish_failures, 1);
assert!(cache.stats().has_distributed_invalidation_bus_issues());
}
#[tokio::test]
async fn receiver_lag_and_close_are_reported_in_stats() {
let cache = HydraCache::local()
.invalidation_bus(LagThenCloseBus)
.invalidation_node_id("lagging")
.build();
wait_until(|| {
let stats = cache.stats();
stats.distributed_invalidation_lagged == 3
&& stats.distributed_invalidation_receiver_closed == 1
})
.await;
let stats = cache.stats();
assert_eq!(stats.distributed_invalidation_lagged, 3);
assert_eq!(stats.distributed_invalidation_receiver_closed, 1);
assert!(stats.has_distributed_invalidation_bus_issues());
}