Struct EmbeddedEngine 

pub struct EmbeddedEngine { /* private fields */ }

Embedded d-engine with automatic lifecycle management.

Thread-safe: Clone and share across threads freely. All methods use &self - safe to call from multiple contexts.

Provides high-level KV API for embedded usage:

• start() / start_with() - Initialize and spawn node
• wait_ready() - Wait for leader election
• client() - Get embedded client
• stop() - Graceful shutdown

Example

use d_engine::EmbeddedEngine;
use std::time::Duration;

let engine = EmbeddedEngine::start().await?;  // Returns Arc<EmbeddedEngine>
engine.wait_ready(Duration::from_secs(5)).await?;

let client = engine.client();
client.put(b"key", b"value").await?;

engine.stop().await?;

Implementations

impl EmbeddedEngine

pub async fn start() -> Result<Self>

Start engine with configuration from environment.

Reads CONFIG_PATH environment variable or uses default configuration. Data directory is determined by config’s cluster.db_root_dir setting.

Example

// Set config path via environment variable
std::env::set_var("CONFIG_PATH", "/etc/d-engine/production.toml");

let engine = EmbeddedEngine::start().await?;
engine.wait_ready(Duration::from_secs(5)).await?;

pub async fn start_with(config_path: &str) -> Result<Self>

Start engine with explicit configuration file.

Reads configuration from specified file path. Data directory is determined by config’s cluster.db_root_dir setting.

Arguments

• config_path: Path to configuration file (e.g. “d-engine.toml”)

Example

let engine = EmbeddedEngine::start_with("config/node1.toml").await?;
engine.wait_ready(Duration::from_secs(5)).await?;

pub async fn start_custom<SE, SM>( storage_engine: Arc<SE>, state_machine: Arc<SM>, config_path: Option<&str>, ) -> Result<Self>

where SE: StorageEngine + Debug + 'static, SM: StateMachine + Debug + 'static,

Start engine with custom storage and state machine.

Advanced API for users providing custom storage implementations.

Arguments

• config_path: Optional path to configuration file
• storage_engine: Custom storage engine implementation
• state_machine: Custom state machine implementation

Example

let storage = Arc::new(MyCustomStorage::new()?);
let sm = Arc::new(MyCustomStateMachine::new()?);
let engine = EmbeddedEngine::start_custom(storage, sm, None).await?;

pub async fn wait_ready(&self, timeout: Duration) -> Result<LeaderInfo>

Wait until the cluster is ready to serve requests.

Blocks until a leader has been elected and its no-op entry is committed by a majority of nodes (Raft §8). Only at this point is the leader guaranteed to be aware of all previously committed entries and safe to serve reads and writes.

Event-driven notification (no polling), <1ms latency after noop commit.

Timeout Guidelines

Single-node mode (most common for development):

• Typical: <100ms (near-instant election)
• Recommended: Duration::from_secs(3)

Multi-node HA cluster (production):

• Typical: 1-3s (depends on network latency and general_raft_timeout_duration_in_ms)
• Recommended: Duration::from_secs(10)

Special cases:

• Health checks: Duration::from_secs(3) (fail fast if cluster unhealthy)
• Startup scripts: Duration::from_secs(30) (allow time for cluster stabilization)
• Development/testing: Duration::from_secs(5) (balance between speed and reliability)

Returns

• Ok(LeaderInfo) - Leader elected successfully
• Err(...) - Timeout or cluster unavailable

Example

// Single-node development
let engine = EmbeddedEngine::start().await?;
let leader = engine.wait_ready(Duration::from_secs(3)).await?;

// Multi-node production
let engine = EmbeddedEngine::start_with("cluster.toml").await?;
let leader = engine.wait_ready(Duration::from_secs(10)).await?;
println!("Leader elected: {} (term {})", leader.leader_id, leader.term);

pub fn leader_change_notifier(&self) -> Receiver<Option<LeaderInfo>>

Subscribe to leader change notifications.

Returns a receiver that will be notified whenever:

• First leader is elected
• Leader changes (re-election)
• No leader exists (during election)

Performance

Event-driven notification (no polling), <1ms latency

Example

let mut leader_rx = engine.leader_change_notifier();
tokio::spawn(async move {
    while leader_rx.changed().await.is_ok() {
        match leader_rx.borrow().as_ref() {
            Some(info) => println!("Leader: {} (term {})", info.leader_id, info.term),
            None => println!("No leader"),
        }
    }
});

pub fn is_leader(&self) -> bool

Returns true if the current node is the Raft leader.

Use Cases

• Load balancer health checks (e.g. HAProxy /primary endpoint returning HTTP 200/503)
• Prevent write requests to followers before they fail
• Application-level request routing decisions

Performance

Zero-cost operation (reads cached leader state from watch channel)

Example

// Load balancer health check endpoint
#[get("/primary")]
async fn health_primary(engine: &EmbeddedEngine) -> StatusCode {
    if engine.is_leader() {
        StatusCode::OK  // Routes writes here
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    }
}

// Application request handler
if engine.is_leader() {
    client.put(key, value).await?;
} else {
    return Err("Not leader, write rejected");
}

pub fn leader_info(&self) -> Option<LeaderInfo>

Returns current leader information if available.

Returns

• Some(LeaderInfo) if a leader is elected (includes leader_id and term)
• None if no leader exists (during election or network partition)

Use Cases

• Monitoring dashboards showing cluster state
• Debugging leader election issues
• Logging cluster topology changes

Example

if let Some(info) = engine.leader_info() {
    println!("Leader: {} (term {})", info.leader_id, info.term);
} else {
    println!("No leader elected, cluster unavailable");
}

pub fn client(&self) -> Arc<EmbeddedClient>

Get a reference to the local KV client.

The client is available immediately after start(), but requests will only succeed after wait_ready() completes.

Example

let engine = EmbeddedEngine::start().await?;
engine.wait_ready(Duration::from_secs(5)).await?;
let client = engine.client();
client.put(b"key", b"value").await?;

pub async fn stop(&self) -> Result<()>

Gracefully stop the embedded engine. Stop the embedded d-engine gracefully (idempotent).

This method:

1. Sends shutdown signal to node
2. Waits for node.run() to complete
3. Propagates any errors from node execution

Safe to call multiple times - subsequent calls are no-ops.

Errors

Returns error if node encountered issues during shutdown.

Example

engine.stop().await?;
engine.stop().await?;  // No-op, returns Ok(())

pub async fn is_stopped(&self) -> bool

Check if the engine has been stopped.

Example

if engine.is_stopped() {
    println!("Engine is stopped");
}

pub fn node_id(&self) -> u32

Returns the node ID for testing purposes.

Useful in integration tests that need to identify which node they’re interacting with, especially in multi-node scenarios.

Trait Implementations

impl Clone for EmbeddedEngine

fn clone(&self) -> EmbeddedEngine

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Drop for EmbeddedEngine

fn drop(&mut self)

Executes the destructor for this type.