llm_optimizer_processor/state/
mod.rs

1//! State management for stream processing
2//!
3//! This module provides state backend abstractions and implementations for maintaining
4//! window state across the distributed stream processing system. It supports:
5//!
6//! - **Multiple Backends**: In-memory (with TTL), persistent (Sled), and distributed (Redis) storage
7//! - **Checkpointing**: Periodic state snapshots for fault tolerance
8//! - **Distributed Locking**: Coordinate state access across multiple processor instances
9//! - **Async Operations**: All state operations are async-first
10//! - **Type Safety**: Generic state serialization/deserialization
11//!
12//! ## Architecture
13//!
14//! The state module is organized around four core concepts:
15//!
16//! 1. **StateBackend Trait**: Defines the interface for all state backends
17//! 2. **Backend Implementations**: Memory, Sled-based persistent, and Redis distributed storage
18//! 3. **Checkpoint Coordination**: Manages periodic snapshots and recovery
19//! 4. **Distributed Locking**: Redis and PostgreSQL-based distributed locks
20//!
21//! ## Example Usage
22//!
23//! ```rust,no_run
24//! use processor::state::{MemoryStateBackend, StateBackend};
25//! use std::time::Duration;
26//!
27//! #[tokio::main]
28//! async fn main() -> anyhow::Result<()> {
29//!     // Create an in-memory state backend with TTL
30//!     let backend = MemoryStateBackend::new(Some(Duration::from_secs(3600)));
31//!
32//!     // Store state
33//!     backend.put(b"window:123", b"state_data").await?;
34//!
35//!     // Retrieve state
36//!     if let Some(data) = backend.get(b"window:123").await? {
37//!         println!("Retrieved: {:?}", data);
38//!     }
39//!
40//!     // List all windows
41//!     let keys = backend.list_keys(b"window:").await?;
42//!     println!("Found {} windows", keys.len());
43//!
44//!     Ok(())
45//! }
46//! ```
47//!
48//! ## Checkpointing
49//!
50//! The checkpoint coordinator automatically creates state snapshots:
51//!
52//! ```rust,no_run
53//! use processor::state::{CheckpointCoordinator, MemoryStateBackend};
54//! use std::sync::Arc;
55//! use std::time::Duration;
56//!
57//! #[tokio::main]
58//! async fn main() -> anyhow::Result<()> {
59//!     let backend = Arc::new(MemoryStateBackend::new(None));
60//!
61//!     // Create coordinator with 5-minute intervals
62//!     let coordinator = CheckpointCoordinator::new(
63//!         backend.clone(),
64//!         Duration::from_secs(300),
65//!         "/tmp/checkpoints".into(),
66//!     );
67//!
68//!     // Start periodic checkpointing
69//!     coordinator.start().await?;
70//!
71//!     // Work with state...
72//!
73//!     // Checkpoint is created automatically
74//!     coordinator.shutdown().await?;
75//!
76//!     Ok(())
77//! }
78//! ```
79//!
80//! ## Distributed Locking
81//!
82//! Coordinate critical operations across multiple processor instances:
83//!
84//! ```rust,no_run
85//! use processor::state::distributed_lock::{RedisDistributedLock, DistributedLock};
86//! use std::time::Duration;
87//!
88//! #[tokio::main]
89//! async fn main() -> anyhow::Result<()> {
90//!     let lock = RedisDistributedLock::new("redis://localhost:6379").await?;
91//!
92//!     // Acquire lock for checkpoint coordination
93//!     let guard = lock.acquire("checkpoint:state1", Duration::from_secs(30)).await?;
94//!
95//!     // Perform checkpoint - only one instance will execute this
96//!     println!("Creating checkpoint...");
97//!
98//!     // Lock automatically released when guard is dropped
99//!     drop(guard);
100//!
101//!     Ok(())
102//! }
103//! ```
104
105pub mod backend;
106pub mod cached_backend;
107pub mod checkpoint;
108pub mod compression;
109pub mod distributed_lock;
110pub mod memory;
111pub mod postgres_backend;
112pub mod postgres_backend_enhanced;
113#[cfg(test)]
114mod postgres_backend_enhanced_tests;
115pub mod postgres_lock;
116pub mod redis_backend;
117pub mod redis_backend_enterprise;
118pub mod redis_lock;
119pub mod sled_backend;
120
121// Re-export main types
122pub use backend::StateBackend;
123pub use cached_backend::{
124    CacheConfig, CacheStats, CachedStateBackend, EvictionPolicy, WriteStrategy,
125};
126pub use checkpoint::{
127    Checkpoint, CheckpointCoordinator, CheckpointMetadata, CheckpointOptions, CheckpointStats,
128};
129pub use compression::{
130    compress, decompress, CompressionConfig, CompressionStats,
131    CompressionAlgorithm as SledCompressionAlgorithm,
132};
133pub use distributed_lock::{DistributedLock, LockGuard, LockToken};
134pub use memory::MemoryStateBackend;
135pub use postgres_backend::{CheckpointInfo, PostgresConfig, PostgresStateBackend, PostgresStats};
136pub use postgres_backend_enhanced::{
137    EnhancedPostgresBackend, EnhancedPostgresConfig, EnhancedPostgresStats,
138    IsolationLevel, PartitionStrategy, PoolStatistics, PostgresLockGuard,
139    PostgresListener, PostgresPoolConfig, PostgresReplicaConfig, PostgresSslConfig,
140    PostgresStatsSnapshot, PostgresTransaction, ReplicationStatus, RetentionPolicy,
141    TableStatistics,
142};
143pub use postgres_lock::PostgresDistributedLock;
144pub use redis_backend::{RedisConfig, RedisConfigBuilder, RedisStateBackend, RedisStats};
145pub use redis_backend_enterprise::{
146    BackendStats, CompressionAlgorithm, EnterpriseRedisBackend, EnterpriseRedisConfig,
147    EnterpriseRedisConfigBuilder, RedisMetrics, RedisMode, RedlockGuard, SerializationFormat,
148};
149pub use redis_lock::{RedisDistributedLock, RetryConfig};
150pub use sled_backend::{SledBackendStats, SledConfig, SledStateBackend};
151
152/// Type alias for state operation results
153pub type StateResult<T> = Result<T, crate::error::StateError>;
llm_optimizer_processor/state/mod.rs

llm_optimizer_processor/state/
mod.rs
