Struct SyncEngine 

pub struct SyncEngine { /* private fields */ }

Main sync engine coordinator.

Manages the three-tier storage architecture:

• L1: In-memory DashMap with pressure-based eviction
• L2: Redis cache with batch writes
• L3: MySQL/SQLite archive (ground truth)

Thread Safety

The engine is Send + Sync and designed for concurrent access. Internal state uses atomic operations and concurrent data structures.

Implementations

impl SyncEngine

pub async fn contains(&self, id: &str) -> bool

Check if an item exists across all tiers.

Checks in order: L1 cache → Redis EXISTS → L3 Cuckoo filter → SQL query. If found in SQL and Cuckoo filter was untrusted, updates the filter.

Returns

• true → item definitely exists in at least one tier
• false → item does not exist (authoritative)

Example

if engine.contains("user.123").await {
    let item = engine.get("user.123").await;
} else {
    println!("Not found");
}

pub fn contains_fast(&self, id: &str) -> bool

Fast sync check if item might exist (L1 + Cuckoo only).

Use this when you need a quick probabilistic check without async. For authoritative check, use contains() instead.

pub fn len(&self) -> usize

Get the current count of items in L1 cache.

pub fn is_empty(&self) -> bool

Check if L1 cache is empty.

pub async fn status(&self, id: &str) -> ItemStatus

Get the sync status of an item.

Returns detailed state information about where an item exists and its sync status across tiers.

Example

match engine.status("order.456").await {
    ItemStatus::Synced { in_l1, in_l2, in_l3 } => {
        println!("Synced: L1={}, L2={}, L3={}", in_l1, in_l2, in_l3);
    }
    ItemStatus::Pending => println!("Queued for sync"),
    ItemStatus::Missing => println!("Not found"),
}

pub async fn get_many(&self, ids: &[&str]) -> Vec<Option<SyncItem>>

Fetch multiple items in parallel.

Returns a vector of Option<SyncItem> in the same order as input IDs. Missing items are represented as None.

Performance

This method fetches from L1 synchronously, then batches L2/L3 lookups for items not in L1. Much faster than sequential get() calls.

Example

let ids = vec!["user.1", "user.2", "user.3"];
let items = engine.get_many(&ids).await;
for (id, item) in ids.iter().zip(items.iter()) {
    match item {
        Some(item) => println!("{}: found", id),
        None => println!("{}: missing", id),
    }
}

pub async fn submit_many( &self, items: Vec<SyncItem>, ) -> Result<BatchResult, StorageError>

Submit multiple items for sync atomically.

All items are added to L1 and queued for batch persistence. Returns a BatchResult with success/failure counts.

Example

let items = vec![
    SyncItem::from_json("user.1".into(), json!({"name": "Alice"})),
    SyncItem::from_json("user.2".into(), json!({"name": "Bob"})),
];
let result = engine.submit_many(items).await.unwrap();
println!("Submitted: {}, Failed: {}", result.succeeded, result.failed);

pub async fn delete_many( &self, ids: &[&str], ) -> Result<BatchResult, StorageError>

Delete multiple items atomically.

Removes items from all tiers (L1, L2, L3) and updates filters. Returns a BatchResult with counts.

Example

let ids = vec!["user.1", "user.2", "user.3"];
let result = engine.delete_many(&ids).await.unwrap();
println!("Deleted: {}", result.succeeded);

pub async fn get_or_insert_with<F, Fut>( &self, id: &str, factory: F, ) -> Result<SyncItem, StorageError>

where F: FnOnce() -> Fut, Fut: Future<Output = SyncItem>,

Get an item, or compute and insert it if missing.

This is the classic “get or insert” pattern, useful for cache-aside:

1. Check cache (L1 → L2 → L3)
2. If missing, call the async factory function
3. Insert the result and return it

The factory is only called if the item is not found.

Example

let item = engine.get_or_insert_with("user.123", || async {
    // Expensive operation - only runs if not cached
    SyncItem::from_json("user.123".into(), json!({"name": "Fetched from DB"}))
}).await.unwrap();

impl SyncEngine

pub async fn start(&mut self) -> Result<(), StorageError>

Start the engine - connect to backends with proper startup sequence.

Startup flow (trust hierarchy):

1. Initialize WAL (SQLite) - always first, our durability lifeline
2. Connect to SQL (L3) - ground truth, initialize SQL merkle store
3. Drain any pending WAL entries to SQL (blocking)
4. Get SQL merkle root (this is our trusted root)
5. Load CF snapshots from SQLite:
   • If snapshot merkle root matches SQL root → CF is trusted
   • Otherwise → CF must be rebuilt from SQL scan
6. Connect to Redis (L2) - cache layer
7. Compare Redis merkle root with SQL merkle root
   • If match → Redis is synced
   • If mismatch → use branch diff to find stale regions and resync
8. Ready!

pub async fn warm_up(&mut self) -> Result<(), StorageError>

Warm up L1 cache from L2/L3 (call after start)

pub async fn tick(&self)

Perform one tick of maintenance (for manual control instead of run loop).

pub async fn force_flush(&self)

Force flush all pending L2 batches immediately.

pub async fn run(&mut self)

Run the main event loop

pub async fn shutdown(&mut self)

Initiate graceful shutdown

impl SyncEngine

pub fn new( config: SyncEngineConfig, config_rx: Receiver<SyncEngineConfig>, ) -> Self

Create a new sync engine.

The engine starts in Created state. Call start() to connect to backends and transition to Ready.

pub fn state(&self) -> EngineState

Get current engine state.

pub fn state_receiver(&self) -> Receiver<EngineState>

Get a receiver to watch state changes.

pub fn is_ready(&self) -> bool

Check if engine is ready to accept requests.

pub fn memory_pressure(&self) -> f64

Get current memory pressure (0.0 - 1.0+).

pub fn pressure(&self) -> BackpressureLevel

Get current backpressure level.

pub fn should_accept_writes(&self) -> bool

Check if the engine should accept writes (based on pressure).

pub async fn get(&self, id: &str) -> Result<Option<SyncItem>, StorageError>

Get an item by ID.

Checks storage tiers in order: L1 → L2 → L3. Updates access count and promotes to L1 on hit.

pub async fn get_verified( &self, id: &str, ) -> Result<Option<SyncItem>, StorageError>

Get an item with hash verification.

If the item has a non-empty merkle_root, the content hash is verified. Returns StorageError::Corruption if the hash doesn’t match.

pub async fn submit(&self, item: SyncItem) -> Result<(), StorageError>

Submit an item for sync.

The item is immediately stored in L1 and queued for batch write to L2/L3. Uses default options: Redis + SQL (both enabled). Filters are updated only on successful writes in flush_batch_internal().

For custom routing, use submit_with.

pub async fn submit_with( &self, item: SyncItem, options: SubmitOptions, ) -> Result<(), StorageError>

Submit an item with custom routing options.

The item is immediately stored in L1 and queued for batch write. Items are batched by compatible options for efficient pipelined writes.

Example

// Cache-only with 1 minute TTL (no SQL write)
let item = SyncItem::new("cache.key".into(), b"data".to_vec());
engine.submit_with(item, SubmitOptions::cache(CacheTtl::Minute)).await?;

// SQL-only durable storage (no Redis)
let item = SyncItem::new("archive.key".into(), b"data".to_vec());
engine.submit_with(item, SubmitOptions::durable()).await?;

pub async fn delete(&self, id: &str) -> Result<bool, StorageError>

Delete an item from all storage tiers.

Deletes are more complex than writes because the item may exist in:

• L1 (DashMap) - immediate removal
• L2 (Redis) - async removal
• L3 (MySQL) - async removal
• Cuckoo filters (L2/L3) - remove from both
• Merkle trees - update with deletion marker

pub fn l1_stats(&self) -> (usize, usize)

Get L1 cache stats

pub fn l3_filter_stats(&self) -> (usize, usize, FilterTrust)

Get L3 filter stats (entries, capacity, trust_state)

pub fn l3_filter(&self) -> &Arc<FilterManager>

Get access to the L3 filter (for warmup/verification)

pub async fn merkle_roots(&self) -> (Option<String>, Option<String>)

Get merkle root hashes from Redis (L2) and SQL (L3).

Returns (redis_root, sql_root) as hex strings. Returns None for backends that aren’t connected or have empty trees.

pub async fn verify_filter(&self) -> bool

Verify and trust the L3 cuckoo filter.

Compares the filter’s merkle root against L3’s merkle root. If they match, marks the filter as trusted.

Returns true if the filter is now trusted, false otherwise.

pub fn update_gauge_metrics(&self)

Update all gauge metrics with current engine state.

Call this before snapshotting metrics to ensure gauges reflect current state. Useful for OTEL export or monitoring dashboards.