apomesh-state-store-redis

The Redis cache primitive for the production state-store stack. This crate is deliberately not a StateStore implementation — the substrate’s persistence boundary is the StateStore trait, which the Postgres adapter’s HybridStateStore implements while composing this cache inward. HybridStateStore delegates snapshots and the memory tiers to Postgres (durable) and routes hot working-memory reads plus an event-log replay buffer through RedisCache (the hot path). Reach for this crate only when composing or wiring that hybrid stack — never as a standalone store.

Position in the workspace

• Layer: state stores — a typed cache primitive within the deployment-persistence layer, composed by an adapter rather than sitting behind a store port itself. See The event log for the event log and working-memory model it caches.
• Depends on: apomesh-substrate — for the TenantId and SessionId types the keys are built from.
• Consumed by: apomesh-state-store-postgres (HybridStateStore composes it inward) and apomesh-orchestrator-bin (connects the cache at boot and hands it to the hybrid store).

What it owns

• RedisCache — the typed cache handle, wrapping a redis::aio::ConnectionManager (a single multiplexed connection with reconnect-on-drop, not a pool). Clone, and Debug-redacted so an auth-bearing connection string never surfaces in logs.
• The key layout — tenant-prefixed working-memory keys (apomesh:tenant:{tenant}:agent:{agent}:working, a single serialised AgentSnapshot blob) and per-session event-log streams (apomesh:session:{session}:events, Redis Stream entries). Every key is tenant-prefixed so a KEYS apomesh:tenant:X:* sweep maps to all of tenant X’s hot state.
• TTL-bounded working memory — working_memory_put sets an expiry (the HybridStateStore supplies it from operator config) so a cache entry evicts rather than growing indefinitely; an explicit working_memory_invalidate drops an entry when the authoritative Postgres copy is overwritten.
• RedisCacheError — a two-variant error vocabulary (Connect, Io) the HybridStateStore boundary maps into StateStoreError; both variants mean “cache unavailable, fall back to Postgres.”

Public surface

Verified against lib.rs:

• RedisCache::connect(url) -> Result<Self, RedisCacheError> — connect to a Redis URL (e.g. redis://localhost:6379); reconnects transparently on transient failure.
• RedisCache::from_connection(conn) — wrap an already-built ConnectionManager (used by container-backed tests).
• working_memory_get(tenant, agent_uuid) -> Result<Option<Vec<u8>>> — Ok(None) on a miss, so the caller falls back to Postgres and back-populates.
• working_memory_put(tenant, agent_uuid, bytes, ttl) — write-through with an expiry.
• working_memory_invalidate(tenant, agent_uuid) -> Result<u64> — drop a stale entry; returns the number of keys removed.
• event_log_buffer_append(session, envelope) -> Result<String> — append to a session’s hot event-log stream; returns the Redis-assigned entry id.
• ping() -> Result<()> — operator-driven liveness check.

Usage

use std::time::Duration;
use apomesh_state_store_redis::RedisCache;

# async fn demo(tenant: &apomesh_substrate::agent::TenantId) -> anyhow::Result<()> {
let cache = RedisCache::connect("redis://localhost:6379").await?;

// Write-through with a 24h TTL, then read the cached bytes back.
cache
    .working_memory_put(tenant, "agent-uuid", b"snapshot-bytes".to_vec(), Duration::from_secs(86_400))
    .await?;
let cached: Option<Vec<u8>> = cache.working_memory_get(tenant, "agent-uuid").await?;
assert!(cached.is_some());
# Ok(())
# }

In a deployment you do not call RedisCache directly — the HybridStateStore owns every call site. This example shows the primitive in isolation.

Related

• Reference: apomesh-state-store-postgres — the adapter that composes this cache into HybridStateStore.
• Operate: Local stack — where the Redis service comes up alongside Postgres in the Compose stack.