Struct MemoryStateBackend 

pub struct MemoryStateBackend { /* private fields */ }

In-memory state backend using DashMap

This backend stores all state in memory using a concurrent hash map (DashMap). It’s ideal for:

• Testing: Fast, no external dependencies
• Stateless processing: When state doesn’t need to survive restarts
• Small state: When state fits comfortably in memory
• High throughput: Fastest possible state access

Features

• Thread-safe: Uses DashMap for lock-free concurrent access
• TTL support: Automatic expiration of old entries
• Statistics: Tracks access patterns and cache hit rates
• Checkpoint/Restore: Can snapshot state to disk

Example

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

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

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

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

    // Get statistics
    let stats = backend.stats().await;
    println!("Hit rate: {:.2}%",
        100.0 * stats.hit_count as f64 / (stats.hit_count + stats.miss_count) as f64);

    Ok(())
}

Implementations

impl MemoryStateBackend

pub fn new(ttl: Option<Duration>) -> Self

Create a new in-memory state backend

Arguments

• ttl - Optional time-to-live for entries. If None, entries never expire.

Returns

A new MemoryStateBackend instance

Example

use processor::state::MemoryStateBackend;
use std::time::Duration;

// No TTL - entries never expire
let backend1 = MemoryStateBackend::new(None);

// 1 hour TTL
let backend2 = MemoryStateBackend::new(Some(Duration::from_secs(3600)));

pub async fn stats(&self) -> MemoryBackendStats

Get current statistics

Returns

A snapshot of current backend statistics

pub async fn cleanup_expired(&self) -> usize

Clean up expired entries

This method scans all entries and removes those that have exceeded the TTL. Returns the number of entries removed.

Returns

Number of entries removed

Example

let backend = MemoryStateBackend::new(Some(Duration::from_secs(60)));

// Periodically clean up expired entries
let removed = backend.cleanup_expired().await;
println!("Removed {} expired entries", removed);

pub fn start_cleanup_task(&self, interval: Duration) -> JoinHandle<()>

Start automatic cleanup background task

Spawns a background task that periodically cleans up expired entries.

Arguments

• interval - How often to run cleanup

Returns

A handle that cancels the cleanup task when dropped

Example

let backend = MemoryStateBackend::new(Some(Duration::from_secs(3600)));

// Clean up every 5 minutes
let _cleanup_handle = backend.start_cleanup_task(Duration::from_secs(300));

// Cleanup runs in background...

pub async fn checkpoint(&self) -> Vec<(Vec<u8>, Vec<u8>)>

Create a checkpoint of current state

Returns all key-value pairs as a vector for serialization.

Returns

Vector of (key, value) pairs

pub async fn restore( &self, checkpoint: Vec<(Vec<u8>, Vec<u8>)>, ) -> StateResult<()>

Restore state from a checkpoint

Replaces all current state with the provided checkpoint data.

Arguments

• checkpoint - Vector of (key, value) pairs to restore

pub async fn memory_usage(&self) -> usize

Get total memory usage in bytes (approximate)

Returns

Approximate memory usage in bytes

Trait Implementations

impl Clone for MemoryStateBackend

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl StateBackend for MemoryStateBackend

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