Module state 

State management for stream processing

This module provides state backend abstractions and implementations for maintaining window state across the distributed stream processing system. It supports:

• Multiple Backends: In-memory (with TTL), persistent (Sled), and distributed (Redis) storage
• Checkpointing: Periodic state snapshots for fault tolerance
• Distributed Locking: Coordinate state access across multiple processor instances
• Async Operations: All state operations are async-first
• Type Safety: Generic state serialization/deserialization

Architecture

The state module is organized around four core concepts:

1. StateBackend Trait: Defines the interface for all state backends
2. Backend Implementations: Memory, Sled-based persistent, and Redis distributed storage
3. Checkpoint Coordination: Manages periodic snapshots and recovery
4. Distributed Locking: Redis and PostgreSQL-based distributed locks

Example Usage

use processor::state::{MemoryStateBackend, StateBackend};
use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create an in-memory state backend with TTL
    let backend = MemoryStateBackend::new(Some(Duration::from_secs(3600)));

    // Store state
    backend.put(b"window:123", b"state_data").await?;

    // Retrieve state
    if let Some(data) = backend.get(b"window:123").await? {
        println!("Retrieved: {:?}", data);
    }

    // List all windows
    let keys = backend.list_keys(b"window:").await?;
    println!("Found {} windows", keys.len());

    Ok(())
}

Checkpointing

The checkpoint coordinator automatically creates state snapshots:

use processor::state::{CheckpointCoordinator, MemoryStateBackend};
use std::sync::Arc;
use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let backend = Arc::new(MemoryStateBackend::new(None));

    // Create coordinator with 5-minute intervals
    let coordinator = CheckpointCoordinator::new(
        backend.clone(),
        Duration::from_secs(300),
        "/tmp/checkpoints".into(),
    );

    // Start periodic checkpointing
    coordinator.start().await?;

    // Work with state...

    // Checkpoint is created automatically
    coordinator.shutdown().await?;

    Ok(())
}

Distributed Locking

Coordinate critical operations across multiple processor instances:

use processor::state::distributed_lock::{RedisDistributedLock, DistributedLock};
use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let lock = RedisDistributedLock::new("redis://localhost:6379").await?;

    // Acquire lock for checkpoint coordination
    let guard = lock.acquire("checkpoint:state1", Duration::from_secs(30)).await?;

    // Perform checkpoint - only one instance will execute this
    println!("Creating checkpoint...");

    // Lock automatically released when guard is dropped
    drop(guard);

    Ok(())
}

Re-exports

pub use backend::StateBackend;

pub use cached_backend::CacheConfig;

pub use cached_backend::CacheStats;

pub use cached_backend::CachedStateBackend;

pub use cached_backend::EvictionPolicy;

pub use cached_backend::WriteStrategy;

pub use checkpoint::Checkpoint;

pub use checkpoint::CheckpointCoordinator;

pub use checkpoint::CheckpointMetadata;

pub use checkpoint::CheckpointOptions;

pub use checkpoint::CheckpointStats;

pub use compression::compress;

pub use compression::decompress;

pub use compression::CompressionConfig;

pub use compression::CompressionStats;

pub use compression::CompressionAlgorithm as SledCompressionAlgorithm;

pub use distributed_lock::DistributedLock;

pub use distributed_lock::LockGuard;

pub use distributed_lock::LockToken;

pub use memory::MemoryStateBackend;

pub use postgres_backend::CheckpointInfo;

pub use postgres_backend::PostgresConfig;

pub use postgres_backend::PostgresStateBackend;

pub use postgres_backend::PostgresStats;

pub use postgres_backend_enhanced::EnhancedPostgresBackend;

pub use postgres_backend_enhanced::EnhancedPostgresConfig;

pub use postgres_backend_enhanced::EnhancedPostgresStats;

pub use postgres_backend_enhanced::IsolationLevel;

pub use postgres_backend_enhanced::PartitionStrategy;

pub use postgres_backend_enhanced::PoolStatistics;

pub use postgres_backend_enhanced::PostgresLockGuard;

pub use postgres_backend_enhanced::PostgresListener;

pub use postgres_backend_enhanced::PostgresPoolConfig;

pub use postgres_backend_enhanced::PostgresReplicaConfig;

pub use postgres_backend_enhanced::PostgresSslConfig;

pub use postgres_backend_enhanced::PostgresStatsSnapshot;

pub use postgres_backend_enhanced::PostgresTransaction;

pub use postgres_backend_enhanced::ReplicationStatus;

pub use postgres_backend_enhanced::RetentionPolicy;

pub use postgres_backend_enhanced::TableStatistics;

pub use postgres_lock::PostgresDistributedLock;

pub use redis_backend::RedisConfig;

pub use redis_backend::RedisConfigBuilder;

pub use redis_backend::RedisStateBackend;

pub use redis_backend::RedisStats;

pub use redis_backend_enterprise::BackendStats;

pub use redis_backend_enterprise::CompressionAlgorithm;

pub use redis_backend_enterprise::EnterpriseRedisBackend;

pub use redis_backend_enterprise::EnterpriseRedisConfig;

pub use redis_backend_enterprise::EnterpriseRedisConfigBuilder;

pub use redis_backend_enterprise::RedisMetrics;

pub use redis_backend_enterprise::RedisMode;

pub use redis_backend_enterprise::RedlockGuard;

pub use redis_backend_enterprise::SerializationFormat;

pub use redis_lock::RedisDistributedLock;

pub use redis_lock::RetryConfig;

pub use sled_backend::SledBackendStats;

pub use sled_backend::SledConfig;

pub use sled_backend::SledStateBackend;

Modules

backend

State backend trait definition

cached_backend

Tiered caching layer for optimal performance with distributed state

checkpoint

Checkpointing and recovery for state backends

compression

Compression algorithms for state backends

distributed_lock

Distributed locking for coordinating state access across processor instances

memory

In-memory state backend implementation

postgres_backend

PostgreSQL-based persistent state backend

postgres_backend_enhanced

Enterprise-grade PostgreSQL backend with advanced features

postgres_lock

PostgreSQL-based distributed locking implementation

redis_backend

Redis-based distributed state backend implementation

redis_backend_enterprise

Enterprise-grade Redis state backend with advanced features

redis_lock

Redis-based distributed locking implementation

sled_backend

Sled-based persistent state backend

Type Aliases

StateResult

Type alias for state operation results