Module postgres_backend 

PostgreSQL-based persistent state backend

This module provides a production-ready PostgreSQL state backend for distributed state management. It supports connection pooling, transactions, JSONB metadata, automatic schema migrations, and advanced features like checkpointing and TTL.

Features

• Connection Pooling: Managed by sqlx PgPool for optimal performance
• Async Operations: Full async/await support throughout
• JSONB Metadata: Store additional context with state entries
• Automatic Migrations: Schema versioning and automatic upgrades
• Transactions: ACID guarantees with BEGIN/COMMIT/ROLLBACK
• TTL Support: Automatic expiration and cleanup of old entries
• Checkpointing: Point-in-time snapshots for backup and recovery
• Query Optimization: Indexes, prepared statements, and hints
• Production Ready: SSL, retries, timeouts, and health checks

Example

use processor::state::{PostgresStateBackend, PostgresConfig, StateBackend};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Configure PostgreSQL backend
    let config = PostgresConfig::new("postgresql://user:pass@localhost/dbname")
        .with_pool_size(10, 50)
        .with_ssl_mode("require")
        .with_timeout(std::time::Duration::from_secs(30));

    // Initialize backend (runs migrations automatically)
    let backend = PostgresStateBackend::new(config).await?;

    // Use like any other state backend
    backend.put(b"window:123", b"aggregated_data").await?;

    // Store with metadata
    backend.put_with_metadata(
        b"window:123",
        b"data",
        serde_json::json!({"source": "kafka", "partition": 5})
    ).await?;

    // Create checkpoint
    let checkpoint_id = backend.create_checkpoint().await?;

    // Cleanup expired entries
    let deleted = backend.cleanup_expired().await?;
    println!("Deleted {} expired entries", deleted);

    Ok(())
}

Structs

CheckpointInfo

Checkpoint metadata

PostgresConfig

PostgreSQL backend configuration

PostgresStateBackend

PostgreSQL state backend

PostgresStats

Statistics for the PostgreSQL backend