Distributed Invalidation

Use a shared invalidation bus when several cache instances should share invalidation intent.

The bus propagates invalidations, not values. Each cache keeps its own local entries. When one cache invalidates a key, tag, or all entries, peers receive the same intent and remove their local copies.

    let bus = Arc::new(InMemoryInvalidationBus::default());
    let first = HydraCache::local()
        .shared_invalidation_bus(bus.clone())
        .invalidation_node_id("first")
        .build();
    let second = HydraCache::local()
        .shared_invalidation_bus(bus)
        .invalidation_node_id("second")
        .build();

    first
        .put("user:42", 42_u64, CacheOptions::new().tag("users"))
        .await?;
    second
        .put("user:42", 42_u64, CacheOptions::new().tag("users"))
        .await?;

    let mut events = second.subscribe_tag("users");
    first.invalidate_tag("users").await?;

    let event = tokio::time::timeout(Duration::from_millis(500), events.recv())
        .await
        .expect("remote invalidation event")
        .expect("subscription stays open");

    assert_eq!(event.origin(), CacheEventOrigin::DistributedBus);
    assert!(!second.contains_key("user:42").await);
    assert_eq!(first.stats().distributed_invalidations_published, 1);
    assert_eq!(second.stats().distributed_invalidations_applied, 1);

Important semantics:

• cached values are never replicated;
• the in-memory bus is best-effort and does not replay messages after restart;
• self-originated messages are ignored to avoid echo loops;
• remote invalidations emit normal events with CacheEventOrigin::DistributedBus;
• diagnostics expose published, received, applied, lagged, decode-error, and closed-receiver counters.

Framed Boundary

InMemoryFramedInvalidationBus serializes each message into a CacheInvalidationFrame before delivery. It is still in-process, but it exercises the binary boundary that external transports can use later.

    let bus = Arc::new(InMemoryFramedInvalidationBus::for_cluster("orders", 128));
    let first = HydraCache::local()
        .shared_invalidation_bus(bus.clone())
        .invalidation_node_id("first")
        .build();
    let second = HydraCache::local()
        .shared_invalidation_bus(bus)
        .invalidation_node_id("second")
        .build();

    let _ = (first, second);

Use this for transport experiments and compatibility tests. It is not a production Redis, NATS, Postgres, or TCP adapter.

Custom Transports

External transports implement CacheInvalidationBus and return a receiver.

#[derive(Debug, Clone)]
struct MyBus;

#[async_trait]
impl CacheInvalidationBus for MyBus {
    async fn publish(&self, message: CacheInvalidationMessage) -> CacheResult<()> {
        // Send `message` through Redis, NATS, Postgres LISTEN/NOTIFY, etc.
        let _ = message;
        Ok(())
    }

    fn subscribe(&self) -> Box<dyn CacheInvalidationReceiver> {
        Box::new(MyReceiver)
    }
}

struct MyReceiver;

#[async_trait]
impl CacheInvalidationReceiver for MyReceiver {
    async fn recv(&mut self) -> CacheInvalidationReceive {
        // Return Message(...) for normal delivery, Lagged(n) for skipped messages,
        // and Closed when the stream is no longer usable.
        CacheInvalidationReceive::Closed
    }
}

Return Message(...) for normal delivery, Lagged(n) when the transport reports skipped messages, and Closed when the stream is no longer usable.