Module server_guide

Expand description

§Implementing Custom Storage Engines

d-engine supports pluggable storage through the StorageEngine trait. This storage engine is responsible for persisting Raft log entries and metadata to durable storage.

§Architecture Context

The StorageEngine trait provides a generic interface for storage backends
Physical separation between log storage (LogStore) and metadata storage (MetaStore)
Built-in implementations include:
- File-based: Production-ready file system storage
- RocksDB: High-performance embedded database (requires features = ["rocksdb"])
- Sled: Modern embedded database
- In-memory: Volatile storage for testing
Custom implementations enable support for cloud storage, SQL databases, etc.

§1. Implement the Trait

use d_engine::storage::{StorageEngine, LogStore, MetaStore};
use async_trait::async_trait;

struct CustomLogStore;
struct CustomMetaStore;

#[async_trait]
impl LogStore for CustomLogStore {
    async fn persist_entries(&self, entries: Vec<Entry>) -> Result<(), Error> {
        // Your implementation - batch operations recommended
        Ok(())
    }

    async fn entry(&self, index: u64) -> Result<Option<Entry>, Error> {
        // Retrieve single entry
        Ok(None)
    }

    // Implement all required methods...
}

impl MetaStore for CustomMetaStore {
    fn save_hard_state(&self, state: &HardState) -> Result<(), Error> {
        // Persist hard state atomically
        Ok(())
    }

    fn load_hard_state(&self) -> Result<Option<HardState>, Error> {
        // Load persisted hard state
        Ok(None)
    }
}

// Combine log and metadata stores
struct CustomStorageEngine {
    log_store: Arc<CustomLogStore>,
    meta_store: Arc<CustomMetaStore>,
}

impl StorageEngine for CustomStorageEngine {
    type LogStore = CustomLogStore;
    type MetaStore = CustomMetaStore;

    fn log_store(&self) -> Arc<Self::LogStore> {
        self.log_store.clone()
    }

    fn meta_store(&self) -> Arc<Self::MetaStore> {
        self.meta_store.clone()
    }
}

§2. Key Implementation Notes

Atomicity: Ensure write operations are atomic—use batch operations where possible
Durability: Flush writes to persistent storage—implement flush() properly
Consistency: Maintain exactly-once semantics for log entries
Performance: Target >100k ops/sec for log persistence
Resource Management: Clean up resources in Drop implementation

§3. StorageEngine API Reference

§LogStore Methods

Method	Purpose	Performance Target
`persist_entries()`	Batch persist log entries	>100k entries/sec
`entry()`	Get single entry by index	<1ms latency
`get_entries()`	Get entries in range	<1ms for 10k entries
`purge()`	Remove logs up to index	<100ms for 10k entries
`truncate()`	Remove entries from index	<100ms
`flush()`	Sync writes to disk	Varies by backend
`reset()`	Clear all data	<1s
`last_index()`	Get highest persisted index	<100μs

§MetaStore Methods

Method	Purpose	Criticality
`save_hard_state()`	Persist term/vote state	High - atomic required
`load_hard_state()`	Load persisted state	High
`flush()`	Sync metadata writes	Medium

§4. Testing Your Implementation

Use the standardized test suite to ensure compatibility:

use d_engine::storage_engine_test::{StorageEngineBuilder, StorageEngineTestSuite};
use tempfile::TempDir;

struct CustomStorageEngineBuilder {
    temp_dir: TempDir,
}

#[async_trait]
impl StorageEngineBuilder for CustomStorageEngineBuilder {
    type Engine = CustomStorageEngine;

    async fn build(&self) -> Result<Arc<Self::Engine>, Error> {
        let path = self.temp_dir.path().join("storage");
        let engine = CustomStorageEngine::new(path).await?;
        Ok(Arc::new(engine))
    }

    async fn cleanup(&self) -> Result<(), Error> {
        Ok(()) // TempDir auto-cleans on drop
    }
}

#[tokio::test]
async fn test_custom_storage_engine() -> Result<(), Error> {
    let builder = CustomStorageEngineBuilder::new();
    StorageEngineTestSuite::run_all_tests(builder).await
}

#[tokio::test]
async fn test_performance() -> Result<(), Error> {
    let builder = CustomStorageEngineBuilder::new();
    let engine = builder.build().await?;
    let log_store = engine.log_store();

    // Performance test: persist 10,000 entries
    let start = std::time::Instant::now();
    let entries = (1..=10000)
        .map(|i| create_test_entry(i, i))
        .collect();

    log_store.persist_entries(entries).await?;
    let duration = start.elapsed();

    assert!(duration.as_millis() < 1000, "Should persist 10k entries in <1s");
    Ok(())
}

§5. Register with NodeBuilder

use d_engine::NodeBuilder;

let storage_engine = Arc::new(CustomStorageEngine::new().await?);

NodeBuilder::new(config, shutdown_rx)
    .storage_engine(storage_engine)  // Required component
    .build();

§6. Production Examples

Reference implementations available in:

src/storage/adaptors/rocksdb/ - RocksDB storage engine
src/storage/adaptors/sled/ - Sled storage engine
src/storage/adaptors/file/ - File-based storage
src/storage/adaptors/mem/ - In-memory storage

Enable RocksDB feature in your Cargo.toml:

d-engine = { version = "0.1.3", features = ["rocksdb"] }

§7. Performance Optimization

Batch Operations: Use batch writes for log persistence
Caching: Cache frequently accessed data (e.g., last index)
Concurrency: Use appropriate locking strategies
Compression: Consider compressing log entries for large deployments

See src/storage/adaptors/rocksdb/rocksdb_engine.rs for a production-grade implementation featuring:

Write batches for atomic operations
Log compaction to reclaim disk space
Efficient snapshot storage
Metrics integration

§Implementing Custom State Machines

d-engine supports pluggable state machines through the StateMachine trait. This is where your application’s core logic executes—processing committed log entries to update application state.