Struct PostgresStateBackend 

pub struct PostgresStateBackend { /* private fields */ }

PostgreSQL state backend

This backend uses PostgreSQL for persistent, distributed state management. It provides ACID guarantees, connection pooling, automatic migrations, and advanced features like JSONB metadata and checkpointing.

Implementations

impl PostgresStateBackend

pub async fn new(config: PostgresConfig) -> StateResult<Self>

Create a new PostgreSQL state backend

This initializes the connection pool and runs automatic schema migrations.

Arguments

• config - PostgreSQL configuration

Returns

• Ok(PostgresStateBackend) - Successfully initialized backend
• Err(StateError) - Failed to initialize

Example

let config = PostgresConfig::new("postgresql://localhost/test");
let backend = PostgresStateBackend::new(config).await?;

pub async fn put_with_metadata( &self, key: &[u8], value: &[u8], metadata: Value, ) -> StateResult<()>

Put a value with metadata

Arguments

• key - The key to store
• value - The value to store
• metadata - JSON metadata

Example

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

pub async fn put_with_ttl( &self, key: &[u8], value: &[u8], ttl: Duration, ) -> StateResult<()>

Put a value with TTL

Arguments

• key - The key to store
• value - The value to store
• ttl - Time to live duration

pub async fn cleanup_expired(&self) -> StateResult<usize>

Clean up expired entries

This removes all entries where expires_at is in the past.

Returns

Number of entries deleted

pub async fn create_checkpoint(&self) -> StateResult<Uuid>

Create a checkpoint of the current state

This creates a point-in-time snapshot of all state entries.

Returns

Checkpoint UUID

pub async fn restore_checkpoint(&self, checkpoint_id: Uuid) -> StateResult<()>

Restore from a checkpoint

Arguments

• checkpoint_id - UUID of the checkpoint to restore

pub async fn list_checkpoints(&self) -> StateResult<Vec<CheckpointInfo>>

List all checkpoints

pub async fn stats(&self) -> PostgresStats

Get backend statistics

pub async fn vacuum(&self) -> StateResult<()>

Execute VACUUM on the state table

This reclaims storage and updates statistics for better query performance.

pub async fn table_stats(&self) -> StateResult<(i64, i64)>

Get table size statistics

pub async fn batch_put(&self, entries: &[(&[u8], &[u8])]) -> StateResult<()>

Batch put operation

Efficiently insert multiple key-value pairs in a single transaction.

pub async fn health_check(&self) -> StateResult<bool>

Health check - verify database connectivity

pub async fn close(&self)

Close the connection pool gracefully

Trait Implementations

impl StateBackend for PostgresStateBackend

fn get<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = StateResult<Option<Vec<u8>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Retrieve a value for the given key

fn put<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, key: &'life1 [u8], value: &'life2 [u8], ) -> Pin<Box<dyn Future<Output = StateResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Store a value for the given key

fn delete<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = StateResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a key and its associated value

fn list_keys<'life0, 'life1, 'async_trait>( &'life0 self, prefix: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = StateResult<Vec<Vec<u8>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List all keys with the given prefix

fn clear<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = StateResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Clear all state from the backend

fn count<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = StateResult<usize>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get the number of keys in the backend

fn contains<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = StateResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Check if a key exists