Async Runtime

In the previous sections, we showed how to drive OmniPaxos ourselves by periodically calling tick(), fetching outgoing messages, and polling the log for decided entries. The main benefit of the omnipaxos_runtime crate is that the user can instead await decided entries asynchronously, rather than polling read_decided_suffix. This can be done in two ways:

append_notify(): Append a single entry and await until that entry is decided (returning its log index).
subscribe_decided(): Subscribe to a continuous stream of decided entries as they are committed.

Under the hood, the runtime wraps an OmniPaxos instance in a background actor and returns a cloneable OmniPaxosHandle with async methods. The actor owns the instance and drives tick() on a timer, so the user also does not have to manage an Arc<Mutex<OmniPaxos>> or call tick() themselves.

Similar to the sync API, the runtime does not own the networking. The user still has to drain outgoing messages and deliver incoming ones from their own transport implementation. A ready-to-use TokioRuntime is provided behind the tokio_runtime feature. Other executors can be used by implementing the AsyncRuntime trait.

A full example can be found in examples/kv_store_async.

Dependencies

To use the runtime, add the following to your Cargo.toml:

[dependencies]
omnipaxos = { version = "LATEST_VERSION", features = ["macros"] }
omnipaxos_runtime = { version = "LATEST_VERSION", features = ["tokio_runtime"] }
omnipaxos_storage = "LATEST_VERSION"
tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros"] }

Note: omnipaxos_runtime enables the async_runtime feature on omnipaxos. This feature adds the wire messages used when followers forward append_notify proposals to the leader.

Spawning an Actor

We begin by creating an OmniPaxos instance as usual, and then hand it to spawn_actor:

use omnipaxos::{ClusterConfig, OmniPaxosConfig, ServerConfig};
use omnipaxos_runtime::{spawn_actor, OmniPaxosHandle, RuntimeConfig, TokioRuntime};
use omnipaxos_storage::memory_storage::MemoryStorage;

let op = OmniPaxosConfig {
    cluster_config: ClusterConfig {
        configuration_id: 1,
        nodes: vec![1, 2, 3],
        ..Default::default()
    },
    server_config: ServerConfig {
        pid: 1,
        election_tick_timeout: 5,
        ..Default::default()
    },
}
.build::<KeyValue, MemoryStorage<KeyValue>>(MemoryStorage::default())
.unwrap();

let handle: OmniPaxosHandle<KeyValue> =
    spawn_actor::<_, _, TokioRuntime>(op, RuntimeConfig::default());

The returned OmniPaxosHandle is cloneable and can be shared across async tasks. Dropping all handle clones shuts the actor down.

Incoming and Outgoing

As with the sync API, the user is responsible for networking. Each handle exposes an outgoing message stream via outgoing_messages(). These messages should be forwarded to peers, and received messages should be delivered with handle_incoming:

let out = handle.outgoing_messages();
tokio::spawn(async move {
    while let Ok(msg) = out.recv().await {
        let receiver = msg.get_receiver();
        // send msg to receiver on network layer
    }
});

// handle incoming message from network layer
handle.handle_incoming(msg).await;

Appending Entries

There are two ways to append entries with the handle:

// fire-and-forget append (mirrors the sync API)
handle.append(KeyValue { key: "a".into(), value: 1 }).await.expect("Failed to append");

// wait until the entry is decided and get its log index
let idx = handle
    .append_notify(KeyValue { key: "quota".into(), value: 100 })
    .await
    .expect("Failed to append_notify");

append() behaves like the sync API: the entry is proposed without waiting for it to be decided. This is useful when a decided-log subscriber (described below) applies entries and the caller does not need to wait for a specific index.

append_notify() instead waits until the entry is decided and returns its log index. It can be called on any node: the actor tags the entry, routes it to the current leader, and resolves when that index is decided. If no leader is known yet, the call waits rather than failing immediately (until append_notify_timeout). The timeout for a single call can be overridden with append_notify_with_timeout(entry, duration).

append_notify() can fail with the following AppendError variants:

Timeout: The entry was not decided within the deadline. It is safe to retry.
Propose(ProposeErr): The underlying propose call failed on this node (e.g. a reconfiguration is already pending).
Superseded { log_entry_idx }: A leader change occurred before the entry was decided. This is a heuristic rather than a certainty; check read_decided_suffix at that index before re-appending.
TooManyOutstanding: The cap on concurrent append_notify calls was reached and nothing was proposed. Safe to retry after backing off.
Shutdown: The actor has already shut down.

Events and the Decided Log

The handle can also notify the application when interesting state transitions occur, and push decided entries for state-machine application:

use omnipaxos::util::LogEntry;
use omnipaxos_runtime::OmniPaxosEvent;

// subscribe to leadership / decide / reconfiguration events
let events = handle.subscribe_events();
tokio::spawn(async move {
    while let Ok(event) = events.recv().await {
        match event {
            OmniPaxosEvent::LeaderElected { pid, accepted } => {
                // a new leader was observed
            }
            OmniPaxosEvent::Decided { new_decided_idx } => {
                // the decided index advanced
            }
            OmniPaxosEvent::Reconfigured { stop_sign_idx } => {
                // a StopSign was decided
            }
        }
    }
});

// subscribe to decided log entries from index 0 and apply them
let decided = handle.subscribe_decided(0).await;
while let Ok(entry) = decided.recv().await {
    if let LogEntry::Decided(kv) = entry {
        // apply kv to the application state
    }
}

subscribe_events() returns a shared stream: if the returned receiver is cloned, the clones share the same queue. The stream is also lossy — if events are not drained fast enough, the oldest ones are dropped. Only Decided is self-correcting, since it carries the current decided index. For leadership or reconfiguration, the user can fall back to current_leader() and the decided log (which ends in a decided StopSign after a reconfiguration) if they may have lagged.

subscribe_decided(from) creates an independent subscription that emits decided LogEntrys in log order, including both the backlog already decided at subscription time and entries decided later.

Other Handle Methods

Besides appending and subscribing, the handle also exposes:

current_leader(): Returns the current leader pid and whether this node is in the Accept phase.
decided_idx(): Returns the current decided index.
read_decided_suffix(from): Reads decided entries (and snapshots) from a given index.
reconfigure(ClusterConfig, Option<Vec<u8>>): Proposes a reconfiguration, as described here.
try_become_leader() / reconnected(pid): Manually attempt leadership, or notify that a peer is reachable again.

Configuring the Runtime

The behaviour of the actor can be configured via RuntimeConfig. Using RuntimeConfig::default() is a good starting point. The most relevant fields are:

tick_period (default 10ms): How often the actor calls tick() on the underlying OmniPaxos.
append_notify_timeout (default 5s): Default deadline for append_notify.
max_pending_appends (default 1024): Maximum number of outstanding append_notify calls.
event_capacity (default 1024): Capacity of the event channel. When full, the oldest events are dropped.
outgoing_capacity / incoming_capacity (default 10_000): Capacities of the message channels.
decided_channel_capacity (default 1024): Capacity of each subscribe_decided channel.
max_decided_subscribers (default 128): Maximum number of concurrent decided-log subscribers.

The tick_period should be chosen together with the election_tick_timeout, resend_message_tick_timeout, and flush_batch_tick_timeout in ServerConfig, so that timeouts resolve in the intended wall-clock time. For example, if tick_period is 10ms and election_tick_timeout is 5, the election timeout will be 50ms.

Async Runtime

Dependencies #

Spawning an Actor #

Incoming and Outgoing #

Appending Entries #

Events and the Decided Log #

Other Handle Methods #

Configuring the Runtime #