Introduction

zestors is an actor framework for Rust with Erlang/OTP-style supervision and clustering.

An actor is a tokio task that owns an Inbox and processes the messages sent to it. Messages are plain structs deriving Message, and the set of messages an actor accepts is its Interface. Actors can be supervised: a Supervisor starts, watches and restarts its children according to a restart strategy. Nodes can form a cluster, and an actor on another node is messaged the same way as one on this node.

Ideas behind it

• An actor is just a task. There is no actor system to start: spawn an async closure over an Inbox inside any tokio runtime, and write its receive loop like any other async code — select! over messages, signals, timers and sockets. When you don’t need that control, implement one Handle<M> per message and let Handler run the loop.
• A message means the same thing to every actor. What a message replies with is part of the message, not of the actor that receives it. So one message — GetHealth, say — can be asked of any actor that accepts it, and always answers the same way.
• Address actors by what they accept. An address can be narrowed to part of an actor’s interface: Address<Dyn<(GetHealth, GetChildren)>>. Addresses of unrelated actors that share those messages have the same type and can go in one collection, checked at compile time; or you can send by message type and check at runtime. The supervision tree, the HTTP API and the inspector find their way around a running system this way, without knowing any actor’s type. See Dynamic addresses.
• Supervision in the OTP sense. Supervisors restart actors on the same channel, so a name — and every address to it — stays valid across restarts. Restart strategies, restart budgets and child sets that change at runtime are all plain data.
• Local or remote, the same code. A ClusterAddress reaches an actor on this node or another through the same cast and call. What the network adds — at-most-once delivery, timeouts, lost nodes — is spelled out, not hidden. Clusters can be tested inside one process, on virtual time.

The crates

Most programs depend only on the zestors crate, which re-exports the others as modules.

The API documentation covers every type in detail. This book explains how the pieces fit together.

Getting started

Add zestors and tokio to your Cargo.toml. Actors written with Handler return a rootcause::Report as their error, so most programs need rootcause as well:

[dependencies]
zestors = "0.3"
tokio = { version = "1", features = ["full"] }
rootcause = "0.13"

Clustering — distributed mode — is behind the distr feature, which is off by default. Its messages are usually serialized with serde:

zestors = { version = "0.3", features = ["distr"] }
serde = { version = "1", features = ["derive"] }

A first actor

A counter that can be incremented and asked for its count. Increment expects no reply; GetCount replies with a u32.

use zestors::interface::{Envelope, Interface, Message};
use zestors::prelude::*;
use zestors::runtime::spawn_rand;

#[derive(Message, Debug)]
struct Increment;

#[derive(Message, Debug)]
#[msg(reply = u32)]
struct GetCount;

#[derive(Interface, Debug)]
enum CounterInterface {
    Increment(Envelope<Increment>),
    GetCount(Envelope<GetCount>),
}

#[tokio::main]
async fn main() {
    let child = spawn_rand(|mut inbox: Inbox<CounterInterface>| async move {
        let mut count = 0;
        while let Some(msg) = inbox.recv().await {
            match msg {
                CounterInterface::Increment(_) => count += 1,
                CounterInterface::GetCount(envelope) => {
                    let _ = envelope.reply(count);
                }
            }
        }
        Ok(())
    });

    for _ in 0..5 {
        child.cast(Increment).await.unwrap();
    }
    // One actor's messages are handled in order, so the count includes all
    // five increments.
    assert_eq!(child.call(GetCount).await.unwrap(), 5);

    child.signal_shutdown();
}

The next chapters take this apart:

• Messages and interfaces: the two derives.
• Spawning and messaging actors: spawn_rand, Child, cast and call.
• Lifecycle and signals: what signal_shutdown does, and when.
• Handler actors: the same counter without the hand-written loop.

To run a program under a supervision tree instead of spawning actors by hand, see Supervision. To run it across several machines, see Distributed mode.

The workspace also has runnable examples in crates/zestors/examples:

# A supervision tree, with the HTTP API on :8080
cargo run -p zestors --example supervision

# Two cluster nodes in one process
cargo run -p zestors --features distr --example remote

# One cluster node per terminal
cargo run -p zestors --features distr --example cluster -- node-a 127.0.0.1:7001

Messages and interfaces

Messages

A message is any type that implements Message, which is almost always derived. The derive decides one thing: whether the message expects a reply.

use zestors::interface::Message;

// Fire-and-forget: nothing comes back to the sender.
#[derive(Message, Debug)]
struct Increment;

// Request/reply: calling this message returns a `u32`.
#[derive(Message, Debug)]
#[msg(reply = u32)]
struct GetCount;

Under the hood, Message has two associated types:

• Output is what the sender gets back: u32 for GetCount, () for Increment.
• Kind is Call for a message with a reply and Cast for one without. It decides the pair of types that carry the reply. A Call message travels with a Request<T>, which the receiver answers, and the sender keeps the matching Reply<T> to wait on. A Cast message uses () for both.

Messages can be generic, and any Send + 'static type can be one.

Interfaces

An actor accepts a set of messages, described by an Interface: an enum with one variant per message, each wrapping the message in an Envelope.

use zestors::interface::{Envelope, Interface, Message};
#[derive(Message, Debug)]
struct Increment;
#[derive(Message, Debug)]
#[msg(reply = u32)]
struct GetCount;

#[derive(Interface, Debug)]
enum CounterInterface {
    Increment(Envelope<Increment>),
    GetCount(Envelope<GetCount>),
}

Every variant must be a tuple variant with exactly one Envelope<M> field, and each message type may appear only once. The derive generates:

• conversions between each Envelope<M> and the enum, which is how a sent message becomes the value the actor receives;
• the interface’s Set, the type-level list of messages it accepts, which dynamic addresses are checked against;
• conversions to and from a type-erased AnyEnvelope, used when the sender doesn’t know the actor’s concrete interface.

The derive does not support generic enums.

An Envelope<M> holds the message as envelope.msg and the reply handle as envelope.req. envelope.reply(value) answers a request. For a fire-and-forget message req is (), and the envelope is simply dropped once handled.

The derives also take attributes for crate paths and for distributed mode; they are listed in Derive attributes.

Spawning and messaging actors

Spawning

An actor’s body is an async closure (or function) that takes an Inbox and returns a Result. There are two ways to start one:

• spawn(name, f) registers the actor under a Name you choose. It fails if that name is taken.
• spawn_rand(f) generates a fresh, random name.

Both return a Child, the owning handle to the actor.

use zestors::prelude::*;
use zestors::runtime::{spawn, spawn_rand};

#[tokio::main]
async fn main() {
let child = spawn(Name::new("worker"), |mut inbox: Inbox<()>| async move {
    let mut seen = 0;
    while inbox.recv().await.is_some() {
        seen += 1;
    }
    Ok(seen)
})
.unwrap();

let other = spawn_rand(|mut inbox: Inbox<()>| async move {
    while inbox.recv().await.is_some() {}
    Ok(())
});
child.signal_shutdown();
other.signal_shutdown();
}

Inbox<()> is the simplest interface: the actor accepts only (). Anything more is a #[derive(Interface)] enum, as in Messages and interfaces. The value the closure returns (seen above) is what awaiting the Child gives back.

References to an actor

Messages are sent through the Accepts trait and everything else goes through ActorOps; both are in the prelude. All four references implement them, so any of them can send messages, send signals and read the status.

Child aborts on drop even when it is bound to _ or simply goes out of scope. Keep it, call .detach() on it, or turn it into a plain JoinHandle with .into_handle().

Sending

Each has a *_with(msg, CallOptions) variant. CallOptions can ignore backpressure, or deliver to an actor that is already exiting.

A message can only be sent to an actor whose interface contains it; anything else is a compile error. To decide at runtime instead, use cast_dyn or call_dyn, which return a NotAccepted error — see Dynamic addresses.

A local call has no timeout: it waits until the actor replies or drops the request. Wrap it in tokio::time::timeout if you need one.

Finding actors by name

Every actor is registered in the process-wide Registry under its Name, for as long as a strong reference (a Child, StrongAddress or Inbox) exists. Code that only knows the name can look it up:

use zestors::prelude::*;
use zestors::runtime::{Registry, spawn};

#[tokio::main]
async fn main() {
let name = Name::new("counter");
let child = spawn(name.clone(), |mut inbox: Inbox<()>| async move {
    while inbox.recv().await.is_some() {}
    Ok(())
})
.unwrap();

// With the exact interface...
let typed: Address<()> = Registry::local().get_typed::<()>(&name).unwrap();
typed.cast(()).await.unwrap();

// ...or untyped, checking at runtime whether it accepts the message.
let untyped = name.address().unwrap();
untyped.cast_dyn(()).await.unwrap();
child.signal_shutdown();
}

The registry covers one process. To reach an actor on another node, see Addressing and sending.

Receiving

The Inbox gives the actor what arrives, in a few ways:

• recv() returns the next message and skips signals. After a shutdown signal it drains the messages still queued, then returns None.
• recv_event() returns an InboxEvent, either a message or a Signal, so the actor can react to signals itself. It returns None once a shutdown has been received and the queue is empty.
• recv_event_always() is like recv_event, but keeps returning events after a shutdown signal, so the actor decides when to stop.
• recv_signal() returns only signals.
• try_recv() doesn’t wait.

Because the actor owns its loop, it can tokio::select! over its inbox and any other future: a timer, a socket, a stream.

Lifecycle and signals

Statuses

An actor is always in one ActorStatus:

Initializing ──► Running ◄──► Suspended
      │             │             │
      └─────────────┴──► Exiting ─┴──► Exited(reason)

• Initializing: spawned, but it hasn’t asked its inbox for anything yet. Messages are accepted and queued.
• Running: it has called a receiving method (recv, recv_event, …) at least once.
• Suspended: it received Signal::Suspend, and stops receiving messages until Signal::Resume.
• Exiting: it received Signal::Shutdown. New messages are refused, and the ones already queued are still delivered.
• Exited: the task has finished. The ExitStatus says how: normally, with an error, by panicking, or by being aborted.

ActorStatusKind is the same list without the exit reason. It is what you name when waiting for a status.

Waiting for a status

ActorOps has a family of monitor_* methods that wait for the actor to reach a status:

Each one first checks the current status, so it returns at once if the actor is already there. The same family exists for actors on other nodes; see Operating on remote actors.

use zestors::prelude::*;
use zestors::runtime::{ActorStatus, ActorStatusKind, spawn_rand};

#[tokio::main]
async fn main() {
let child = spawn_rand(|mut inbox: Inbox<()>| async move {
    while inbox.recv().await.is_some() {}
    Ok(())
});

// Code right after a spawn must not assume the actor is running yet.
child.monitor_init().await.unwrap();

child.signal_suspend();
let status = child
    .monitor_any(&[ActorStatusKind::Suspended, ActorStatusKind::Exited])
    .await;
assert_eq!(status, ActorStatus::Suspended);

child.signal_shutdown();
child.monitor_exit().await.unwrap();
}

Signals

Signals control an actor rather than asking it to do work:

• signal_shutdown() → Signal::Shutdown
• signal_suspend() → Signal::Suspend
• signal_resume() → Signal::Resume

ping() also exists: the runtime answers it by itself, and the actor never sees it.

Two behaviours surprise people:

A signal takes effect later. signal_shutdown() puts the signal in the actor’s queue and returns. status() on the next line can still show the old status. To wait for the effect, use monitor_exit() or another monitor_*.

A signal sent too early is dropped. A name can exist before its actor has been spawned: a ChildSpec reserves it, and a supervisor spawns the actor later. Until then the status is Exited, and a signal is refused (the method returns false). When you might race with a start, call monitor_accepts_messages() first.

Priority

Signals are delivered ahead of queued messages, so a shutdown doesn’t wait behind a backlog. Once a shutdown has been received:

• recv() and recv_event() keep delivering the messages that were already queued, then return None;
• the actor exits as soon as the queue is empty, without looking at any signal still behind the shutdown.

An actor that has to keep handling signals after a shutdown should loop over recv_event_always(), and decide for itself when to stop.

Shutting down with a deadline

Child::shutdown_abort(timeout) sends a shutdown, waits up to timeout for the actor to exit, and aborts it if it doesn’t. Supervisors use the same rule when they stop a child, with the child’s abort_timeout.

Handler actors

Writing the receive loop by hand gives full control. Most actors don’t need it. Handler provides the loop, and asks for one Handle<M> implementation per message.

use zestors::interface::{Envelope, Interface, Message, Request};
use zestors::prelude::*;

#[derive(Message, Debug)]
struct Increment;

#[derive(Message, Debug)]
#[msg(reply = u32)]
struct GetCount;

#[derive(Interface, HandlerInterface, Debug)]
enum CounterInterface {
    Increment(Envelope<Increment>),
    GetCount(Envelope<GetCount>),
}

#[derive(Debug, Clone)]
struct Counter {
    count: u32,
}

impl Handler for Counter {
    type Interface = CounterInterface;
}

impl Handle<Increment> for Counter {
    async fn handle(
        &mut self,
        _ctx: HandlerContext<'_, Self>,
        _msg: Increment,
        _req: (),
    ) -> Result<(), rootcause::Report> {
        self.count += 1;
        Ok(())
    }
}

impl Handle<GetCount> for Counter {
    async fn handle(
        &mut self,
        _ctx: HandlerContext<'_, Self>,
        _msg: GetCount,
        req: Request<u32>,
    ) -> Result<(), rootcause::Report> {
        let _ = req.reply(self.count);
        Ok(())
    }
}

#[tokio::main]
async fn main() {
let child = Counter { count: 0 }.spawn_rand();
for _ in 0..5 {
    child.cast(Increment).await.unwrap();
}
assert_eq!(child.call(GetCount).await.unwrap(), 5);
child.signal_shutdown();
}

• #[derive(HandlerInterface)] on the interface dispatches each variant to the matching Handle<M>. It goes next to #[derive(Interface)].
• The third argument of handle is the message’s reply handle: () for a fire-and-forget message, Request<T> for one with a reply.
• A handler returns Result<(), rootcause::Report>. An error stops the actor.
• Every Handler is an Actor. It is spawned with ActorExt::spawn or spawn_rand, and awaited or messaged like any other actor.
• A Handler that is Clone is also a Blueprint, so it can be supervised as it is.

Lifecycle hooks

All hooks are optional:

exit receives one of these:

• Normal: after a shutdown, or once the inbox has closed;
• InitError: init returned an error;
• InitCancelled: a shutdown arrived while init was still running, so init was cancelled;
• HandlerError: a handler or hook returned an error.

The default exit turns any of these but Normal into an error. exit is not called when the actor panics or is aborted.

init runs alongside the inbox’s signal receiver, so the actor may already report Running while init is still busy. Don’t use monitor_init() as a signal that init has finished.

Reacting to more than messages

Handler::next_event lets the actor react to other futures between messages: a timer, a stream, a background job. Return Some(Ok(event)) to have event handled like a message. event can be any message the handler has a Handle for.

use std::time::Duration;
use zestors::interface::{Envelope, Interface, Message, Request};
use zestors::prelude::*;

#[derive(Message, Debug)]
struct Tick;

#[derive(Message, Debug)]
#[msg(reply = u32)]
struct GetTicks;

#[derive(Interface, HandlerInterface, Debug)]
enum TickerInterface {
    GetTicks(Envelope<GetTicks>),
}

#[derive(Debug)]
struct Ticker {
    ticks: u32,
    interval: tokio::time::Interval,
}

impl Handler for Ticker {
    type Interface = TickerInterface;

    async fn next_event(&mut self) -> Option<Result<impl HandledBy<Self>, rootcause::Report>> {
        self.interval.tick().await;
        Some(Ok(Tick))
    }
}

impl Handle<Tick> for Ticker {
    async fn handle(&mut self, _: HandlerContext<'_, Self>, _: Tick, _: ()) -> Result<(), rootcause::Report> {
        self.ticks += 1;
        Ok(())
    }
}

impl Handle<GetTicks> for Ticker {
    async fn handle(
        &mut self,
        _: HandlerContext<'_, Self>,
        _: GetTicks,
        req: Request<u32>,
    ) -> Result<(), rootcause::Report> {
        let _ = req.reply(self.ticks);
        Ok(())
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
let ticker = Ticker { ticks: 0, interval: tokio::time::interval(Duration::from_secs(1)) };
let child = ticker.spawn_rand();

tokio::time::sleep(Duration::from_millis(3500)).await;
assert!(child.call(GetTicks).await.unwrap() >= 3);
child.signal_shutdown();
}

Tick is not part of the interface, so nobody else can send it. The future returned by next_event is dropped whenever a message or signal arrives first, so it must be cancellation-safe.

For several concurrent futures, keep a BasicScheduler in the handler and return self.scheduler.next().await from next_event. It runs scheduled futures and hands their results to the handler: messages with schedule_msg, closures over the handler’s state with schedule_callback.

Writing the loop yourself

Handler covers most actors. Implement Actor directly — its run takes the Inbox — when the actor needs a receive order Handler doesn’t offer, for example handling signals only between batches of messages. Spawning a closure, as in Spawning and messaging actors, is the same thing without a named type.

Dynamic addresses

Every actor reference has a context: what the reference knows the actor accepts. It is the type parameter in Address<C>, Child<E, C>, StrongAddress<C>, and so on. There are two kinds:

• An interface, such as Address<CounterInterface>. The reference knows the actor’s exact interface.
• A Dyn set, such as Address<Dyn<(Increment, GetCount)>>. The reference only knows that the actor accepts at least these messages, whatever its actual interface is.

Dyn<()>, or just Dyn, accepts nothing. It is what Address defaults to, and what the Registry hands out when it doesn’t know the type.

A Dyn address can send exactly the messages in its set, checked at compile time, just like a typed one. What it hides is the concrete interface. Actors of completely different types that share some messages therefore produce addresses of the same type. That lets them be stored together, passed to the same function, or handed out without revealing what else the actor does.

Converting between contexts

The IntoDyn trait (in the prelude) converts a reference by value, and AsDyn converts it by reference:

The runtime-checked ones return the original reference on failure (Err(self) or None), so nothing is lost. into_context_unchecked skips the check altogether. A wrong context doesn’t cause undefined behaviour, but sending a message the actor doesn’t accept then panics.

Example: many kinds of actor, one list

Two unrelated actors both answer GetHealth. Narrowed to Dyn<(GetHealth,)>, they fit in one Vec:

use zestors::interface::{Envelope, Interface, Message};
use zestors::prelude::*;
use zestors::runtime::{Dyn, spawn_rand};
use zestors::supervision::messages::{GetHealth, Health};

#[derive(Message, Debug)]
struct Ping;

#[derive(Interface, Debug)]
enum WorkerInterface {
    Ping(Envelope<Ping>),
    Health(Envelope<GetHealth>),
}

#[derive(Message, Debug)]
#[msg(reply = "Option<String>")]
struct Get(String);

#[derive(Interface, Debug)]
enum CacheInterface {
    Get(Envelope<Get>),
    Health(Envelope<GetHealth>),
}

#[tokio::main]
async fn main() {
let worker = spawn_rand(|mut inbox: Inbox<WorkerInterface>| async move {
    while let Some(msg) = inbox.recv().await {
        if let WorkerInterface::Health(env) = msg {
            let _ = env.reply(Health::healthy());
        }
    }
    Ok(())
});
let cache = spawn_rand(|mut inbox: Inbox<CacheInterface>| async move {
    while let Some(msg) = inbox.recv().await {
        match msg {
            CacheInterface::Get(env) => { let _ = env.reply(None); }
            CacheInterface::Health(env) => { let _ = env.reply(Health::degraded()); }
        }
    }
    Ok(())
});

// Checked at compile time: both interfaces contain `GetHealth`.
let monitored: Vec<Address<Dyn<(GetHealth,)>>> = vec![
    worker.address().clone().into_dyn(),
    cache.address().clone().into_dyn(),
];

for address in &monitored {
    let health = address.call(GetHealth).await.unwrap();
    println!("{}: {health}", address.name());
}

// The concrete type can be recovered, checked at runtime.
let back = monitored[1].clone().downcast::<CacheInterface>().unwrap();
assert_eq!(back.call(Get("key".into())).await.unwrap(), None);
assert!(monitored[0].clone().downcast::<CacheInterface>().is_err());
worker.signal_shutdown();
cache.signal_shutdown();
}

Without a typed reference at all

Sometimes all you have is an untyped Address (an Address<Dyn>), for example from Name::address() or Registry::get. Then either:

• convert it with into_dyn_checked, or look it up with Registry::local().get_dyn::<S>(&name), which does the same check; or
• send directly with cast_dyn / call_dyn (and their try_/_with variants) from ActorOps. These work on any reference, and return a NotAccepted error, with the message, if the actor doesn’t take it.

How the supervision tree uses this

GetChildren and GetHealth from zestors::supervision::messages are ordinary messages. The supervision tree (SupervisionTree), the HTTP API server and the inspector explore a running system by looking actors up by name and sending them call_dyn(GetChildren) and call_dyn(GetHealth). They never need to know an actor’s type. So:

• a custom supervisor becomes part of the tree by accepting GetChildren;
• any actor can report its health by accepting GetHealth.

Across the cluster

A ClusterAddress has the same kind of context. Cluster::address::<I>() makes one for a whole interface, and Cluster::address_dyn::<(A, B)>() for a set. The set form also lets you reach the remote-capable part of an interface that has local-only messages; see Remote messages.

A ClusterAddress can’t be converted between contexts yet. Pick the context when you create the address.

Supervision

Supervision is optional. Everything so far works without it. It adds Erlang/OTP-style restart trees: a supervisor starts a set of children, notices when one exits, and restarts it according to a policy.

The pieces

• A Blueprint is a recipe for creating an actor, so that it can be created again after a crash. Any Actor that is Clone + Debug is its own blueprint. fn_blueprint(|| …) builds one from a closure, and fn_actor / fn_task turn closures into actors.

• A ChildSpec pairs a blueprint with the Name it runs under and a ChildConfig. blueprint.name("worker")? reserves the name right away; blueprint.rand_name() makes one up.

• A ChildConfig says what a supervisor does with the child:

  • restart_mode: Always, OnError (the default: only after an error, panic or abort) or Never;
  • intensity: an optional per-child restart budget;
  • init_timeout, abort_timeout, start_timeout.

  Set these with with_mode, with_abort_timeout, and so on.

• A Supervisor is an ordinary actor, built from Supervisor::blueprint().

• A SupervisionStrategy decides what gets restarted when a child exits:

  Strategy	Restarts
  OneForOne (default)	only the child that exited
  OneForAll	every child
  RestForOne	the child that exited, and every child started after it

• A RestartIntensity, for example RestartIntensity::new(5, Duration::from_secs(10)), is the most restarts allowed in a window. A supervisor has one (by default 3 restarts in 5 minutes, set with .intensity(…)), and a child can have its own as well. When either runs out, the supervisor stops its children and exits itself, and its own supervisor takes over.

Running a tree as a program

Node runs a root supervisor as the whole program. It starts the supervisor, shuts it down gracefully on Ctrl+C or SIGTERM, and forces an exit on a second signal. It returns when the root supervisor exits, and never restarts it: restarting the whole program is the job of whatever runs it (systemd, Kubernetes, …).

use zestors::interface::{Envelope, Interface, Message};
use zestors::prelude::*;
use zestors::supervision::messages::GetChildren;
use zestors::supervisor::{Node, Supervisor};

#[derive(Message, Debug)]
struct Ping;

#[derive(Interface, HandlerInterface, Debug)]
enum WorkerInterface {
    Ping(Envelope<Ping>),
}

#[derive(Debug, Clone)]
struct Worker;

impl Handler for Worker {
    type Interface = WorkerInterface;
}

impl Handle<Ping> for Worker {
    async fn handle(
        &mut self,
        _ctx: HandlerContext<'_, Self>,
        _msg: Ping,
        _req: (),
    ) -> Result<(), rootcause::Report> {
        Ok(())
    }
}

#[tokio::main]
async fn main() {
let node = Node::new(
    Supervisor::blueprint()
        .strategy(SupervisionStrategy::OneForOne)
        .child(Worker.name("worker").unwrap())
        .rand_name(),
);

// In a program this is just `node.run().await`. Here the node runs in the
// background so that the example can inspect it and stop it again.
let root = node.root_supervisor().address().clone();
let running = tokio::spawn(node.run());

// A supervisor counts as running once all of its children have initialized.
root.monitor_init().await.unwrap();
assert_eq!(root.call(GetChildren).await.unwrap().len(), 1);

// The root supervisor exiting is a normal end of the program.
root.signal_shutdown();
assert!(running.await.unwrap().is_ok());
}

A supervisor can also be started like any other actor, without a Node, with Supervisor::blueprint().start_rand(), and supervisors can be children of other supervisors.

To run the program as part of a cluster, use ClusterNode instead of Node; see Running a cluster.

Changing children at runtime

A running supervisor accepts RegisterChild(spec) and DeregisterChild(name) messages. For a set of children kept elsewhere, give the blueprint a SupervisorSource with .source(…). InMemorySupervisorSource is the built-in one, and a database-backed source implements the same trait.

A larger example

crates/zestors/examples/supervision.rs builds a tree with:

• nested supervisors with different strategies;
• Handler actors that schedule their own ticks;
• closure actors and tasks;
• an ApiServer;
• a source that keeps adding tasks.

    let source = InMemorySupervisorSource::new_arc();

    let (spec_a, _addr) = fn_blueprint(|| MyActor::new("A"))
        .name("HelloActor")?
        .with_mode(RestartMode::Never)
        .split();

    let (spec_b, _addr) = fn_blueprint(|| MyActor::new("B"))
        .name("HelloActor2")?
        .with_mode(RestartMode::Always)
        .split();

    let (super_spec_a, _addr) = Supervisor::blueprint()
        .children([spec_a, spec_b])
        .name("SupervisorA")?
        .split();

    let (spec_c, _addr) = fn_blueprint(|| MyActor::new("C"))
        .name("HelloActor3")?
        .with_mode(RestartMode::Always)
        .split();

    let (spec_d, _addr) = fn_blueprint(|| MyActor::new("D"))
        .name("HelloActor4")?
        .with_mode(RestartMode::Always)
        .split();

    let (super_spec_b, _addr) = Supervisor::blueprint()
        .children([spec_c, spec_d])
        .source(source.clone())
        .name("SupervisorB")?
        .split();

    let (dyn_actor_spec, _addr) = fn_actor(async |_: Inbox<MyInterface>| Ok(()))
        .name("DynActor")?
        .split();

    let (task_spec, _addr) = fn_task(|mut task_box| async move {
        let mut completed_part1 = false;

        let res = task_box
            .run_until_shutdown(async {
                tokio::time::sleep(Duration::from_secs(2)).await;
                println!("Task completed part 1");
                completed_part1 = true;

                tokio::time::sleep(Duration::from_secs(2)).await;
                println!("Task completed part 2");
            })
            .await;

        if let Err(Cancelled) = res {
            println!("Task was cancelled");
            if completed_part1 {
                // Cleanup part1, to reset for the next time this task is ran.
            }
            return Err(Cancelled.into());
        }

        Ok(())
    })
    .name("TaskActor")?
    .split();

    let app_supervisor = Supervisor::blueprint()
        .strategy(SupervisionStrategy::OneForOne)
        .children([
            super_spec_a,
            super_spec_b,
            dyn_actor_spec,
            task_spec,
            fn_blueprint(|| fn_actor(async |_: Inbox<MyInterface>| Ok(())))
                .name("DynBlueprintActor")?
                .into(),
            fn_blueprint(|| MyActor::new("E"))
                .name("DynBlueprintActor2")?
                .into(),
        ])
        .name("app-supervisor")?;

    let node = Node::new(
        Supervisor::blueprint()
            .strategy(SupervisionStrategy::RestForOne)
            .child(
                ApiServer::blueprint("127.0.0.1:8080".parse().unwrap(), "root-supervisor")
                    .name("ApiServer")?,
            )
            .child(app_supervisor)
            .name("root-supervisor")?,
    );

It runs as follows:

    let root_address = node.root_supervisor().address().clone();
    let node_task = tokio::spawn(node.run());

    // A supervisor is running once all of its children have initialized.
    root_address.monitor_init().await?;
    tracing::info!("All actors started");

    spawn_tasks_in_background(source);

    // Runs until Ctrl+C/SIGTERM, or until the root supervisor exits.
    node_task.await??;
    Ok(())

cargo run -p zestors --example supervision

Limits

Supervision is local: a supervisor supervises children in its own process. A ChildSpec holds a local address, and the tree walk uses the local registry. To watch an actor on another node, use the cross-node monitor_* operations; see Operating on remote actors.

Observability

Inspecting from code

Every reference can report on its actor without disturbing it:

• status(), msg_len(), signal_len() and reached_backpressure() read single values;
• snapshot() returns a ChannelSnapshot with the status, queue lengths, and the recent spawn and exit history.

A running tree can be walked with SupervisionTree. It starts from a supervisor’s ChildDescription and asks each supervisor for its children with GetChildren; see Dynamic addresses. Any actor can take part in health reporting by accepting GetHealth and replying with a Health: healthy, degraded or unhealthy, with an optional message and details.

The HTTP API server

ApiServer is an actor that serves the tree over HTTP. Add it as a child next to the rest of the tree, and give it the name of the root supervisor:

Supervisor::blueprint()
    .child(ApiServer::blueprint("127.0.0.1:8080".parse()?, "root-supervisor").name("api-server")?)
    .child(app_supervisor)
    .name("root-supervisor")?

The last two take a JSON array of names as the request body.

The routes are not stable yet and may change in any minor release.

The inspector

zestors-inspector is a desktop GUI, still a proof of concept, that draws the tree served by the API server and polls it for changes. It connects to http://localhost:8080.

cargo run -p zestors --example supervision   # in one terminal
just inspector run                           # in another; or: cargo run --release -p zestors-inspector

The API server and the inspector see one process. In a cluster, each node runs its own.

Distributed mode (experimental)

Several zestors programs — nodes — can form a cluster. An actor on one node can message an actor on another, using the same cast and call as for a local actor.

Not production ready. Distributed mode is new. Its APIs are bound to change, and there will be bugs.

The moving parts

• A node is one program in the cluster. ClusterNode runs it: it does everything Node does (runs the root supervisor, and shuts down on Ctrl+C/SIGTERM), and also joins the cluster.
• Membership is tracked with SWIM, a gossip protocol (via foca). Each node learns who else is up, and notices when a node leaves or crashes. Cluster is the handle to this view.
• A backend carries the bytes between nodes. Quic from zestors-distr-quic is the one to use: QUIC with mutual TLS, so only nodes with a certificate from your CA can join. Other transports implement the Backend trait.
• A GlobalName names an actor anywhere in the cluster: an actor’s Name plus the node it is on, written name@node.
• A ClusterAddress is a reference to an actor in the cluster, on this node or another. It sends messages with ClusterAccepts (cast, call, …) and operates on the actor with ClusterActorOps (signals, status, monitor_*).
• A remote message is a message that can cross the network. It has a StableId that names it the same way on every node, and it can be encoded, usually with serde.

From a local program to a distributed one

1. Enable the distr feature of zestors, which adds the distr and distr_quic modules and their items in the prelude: zestors = { version = "0.3", features = ["distr"] }.
2. Make the messages remote. Derive StableId and Serialize/Deserialize on each message that crosses the network, and give it an id: #[msg(id = "<uuid>")]. Its reply type must be serializable too. See Remote messages.
3. Replace Node with ClusterNode, configured with a ClusterConfig: the node’s name, the backend, and the seeds to join through. See Running a cluster.
4. Register what each node accepts. A node that hosts an actor registers the messages other nodes may send to it: config.register::<Msg>(). Sending needs no registration.
5. Address actors by GlobalName. cluster.address::<I>(GlobalName::new("name", "node")).await gives a ClusterAddress<I>. Every actor registered under a Name is reachable. See Addressing and sending.
6. Handle the network’s failures. A remote call can fail in ways a local one can’t: the node is gone, a timeout expires, the message is too large. See Delivery and failure.

A complete example

Two nodes on the simulated network, which runs in one process on virtual time (see Testing with sim). With QUIC, only the backend passed to ClusterConfig::new changes.

use serde::{Deserialize, Serialize};
use zestors::{
    distr::sim::SimNetwork,
    interface::{Envelope, Interface, Message},
    prelude::*,
    runtime::spawn,
    supervisor::Supervisor,
};

#[derive(Message, StableId, Serialize, Deserialize, Debug)]
#[msg(reply = u32, id = "b5a4c0de-0000-4000-8000-000000000001")]
struct Double(u32);

#[derive(Interface, Debug)]
enum CalcInterface {
    Double(Envelope<Double>),
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
let net = SimNetwork::new(1);

// node-b hosts the actor, so it registers the message it accepts.
let a = ClusterNode::new(
    Supervisor::blueprint().rand_name(),
    ClusterConfig::new("node-a", net.backend("10.0.0.1:7000")),
);
let b = ClusterNode::new(
    Supervisor::blueprint().rand_name(),
    ClusterConfig::new("node-b", net.backend("10.0.0.2:7000"))
        .seed(Seed::new("node-a", "10.0.0.1:7000"))
        .register::<Double>(),
);

// Nodes on one simulated network share a process, so this actor is spawned
// once and served by node-b.
let _calc = spawn(Name::new_static("calc"), |mut inbox: Inbox<CalcInterface>| async move {
    while let Some(CalcInterface::Double(envelope)) = inbox.recv().await {
        let n = envelope.msg.0;
        let _ = envelope.reply(n * 2);
    }
    Ok(())
})
.unwrap();

let cluster = a.cluster();
tokio::spawn(a.run());
tokio::spawn(b.run());
cluster.wait_for_members(1).await;

let calc = cluster
    .address::<CalcInterface>(GlobalName::new("calc", "node-b"))
    .await
    .unwrap();
assert!(calc.is_remote());
assert_eq!(calc.call(Double(21)).await.unwrap(), 42);
}

crates/zestors/examples/remote.rs is the same over real QUIC, and crates/zestors/examples/cluster.rs runs one node per terminal so you can watch nodes join and leave.

Remote messages

A message can be sent to another node when it is a RemoteMessage. There is nothing to implement: every message that

1. has a StableId,
2. can be encoded and decoded, and
3. has a reply type that can be encoded and decoded too

is one. With serde, that is one line of derives:

use serde::{Deserialize, Serialize};
use zestors::interface::Message;
use zestors::prelude::*;

#[derive(Message, StableId, Serialize, Deserialize, Debug)]
#[msg(reply = u64, id = "0e1c8a6e-6f7a-4b8e-9a53-3f2d1c0b9a87")]
struct Fibonacci(u32);

Message ids

A node receives bytes, and needs to know which type to decode them as. Rust type names aren’t stable across builds, so each remote message carries a MessageId: a UUID you choose once and never change. Leave the id out and the compiler error suggests a freshly generated one to paste in.

• Never change an id once nodes running different builds may talk to each other: a node that doesn’t know an id answers RemoteError::UnknownMessage.
• Never reuse an id for two types. Registering two types under one id panics when the node is built, because the other nodes would otherwise decode the bytes as whichever type won.
• Changing a message’s fields changes its wire format. Nodes that disagree fail to decode it (RemoteError::Decode). Treat a message like any other wire format: add fields compatibly, or introduce a new message with a new id.

Encoding

The wire format comes from the Encode and Decode traits. Every serde type implements them, using postcard. To use another format, implement both traits by hand on a type that doesn’t implement Serialize; for a type that does, wrap it in a newtype first.

A message to an actor on the same node is never encoded: it is delivered as the value itself. A type whose encoding is broken therefore works locally and fails only once it crosses the network.

Messages are limited to 4 MiB once encoded; a larger one fails with CastFailure::TooLarge.

Registering what a node accepts

A node accepts a remote message only if it was registered when the node was configured:

let config = ClusterConfig::new("node-b", backend)
    .register::<Fibonacci>()
    .register::<Greet>();

• Registering is only needed on the receiving node. Sending, and addressing an actor with Cluster::address, need nothing.
• Registration is fixed when the node is built. A running node can’t start accepting a new message type, so there is never a moment where it serves requests it can’t yet handle.
• A registered message can be sent to any actor on the node that accepts it.

Registering everything at once

With the auto-register feature, ClusterConfig::auto_register() registers every remote message in the binary: every non-generic type that derives StableId. To leave one out, add #[msg(no_auto_register)]. Generic messages always have to be registered by hand, once per concrete type.

auto_register silently skips a type that derives StableId but isn’t a RemoteMessage, for example because it doesn’t derive Serialize. A sender then gets RemoteError::UnknownMessage back. If that happens, check the derives.

Interfaces with local-only messages

Cluster::address::<I>() addresses an actor by its whole interface, which requires every message in I to be remote. An interface that mixes remote and local-only messages doesn’t compile there. Address the part that can cross the network instead:

let remote_part = cluster
    .address_dyn::<(Fibonacci, Greet)>(GlobalName::new("worker", "node-b"))
    .await?;

The same is true for sending through a ClusterAddress: only remote messages can be sent through one, even when the actor happens to be local. For a local actor, ClusterAddress::local_address() gives back the ordinary Address, which can send anything.

Replies inside messages

A reply type is sent back automatically. When a message needs to carry a reply channel, for example so the actor can hand it on to another actor, use a RemoteRequest field; see Replies inside messages.

Running a cluster

ClusterNode

ClusterNode is Node with cluster membership added. It runs a root supervisor the same way, shuts down on Ctrl+C/SIGTERM the same way, and additionally:

• starts the backend and joins the cluster before the supervisor starts;
• serves the messages registered in its ClusterConfig;
• announces its departure on the way out, so the other nodes see it leave at once instead of timing it out.

    // Development only: use `Tls::from_pem` to authenticate cluster members.
    let mut config = ClusterConfig::new(name, Quic::new(bind, Tls::insecure_dev().unwrap()));
    for seed in args {
        let (seed_name, seed_addr) = seed.split_once('=').expect("seed must be name=addr");
        config = config.seed(Seed::new(
            seed_name,
            seed_addr.parse::<SocketAddr>().unwrap(),
        ));
    }

    let node = ClusterNode::new(Supervisor::blueprint().rand_name(), config);

    let mut events = node.cluster().subscribe();
    tokio::spawn(async move {
        while let Ok(event) = events.recv().await {
            tracing::info!("{event:?}");
        }
    });

    node.run().await

Take node.cluster() before calling run(), which consumes the node. The Cluster handle is cheap to clone, and works before the node has started: it then reports no members, and sends fail with CastFailure::NotRunning.

To add clustering to a Node you have already configured (for example with a custom exit watcher), use ClusterNode::from_node(node, config).

ClusterConfig

ClusterConfig::new(name, backend) takes the node’s name and its transport. The rest is optional:

Nodes don’t have to agree on any of these.

Node names

A node’s name (NodeName) identifies it in the cluster, and is the node part of every GlobalName that points at it. With QUIC, it is also the TLS server name that peers dial, so it must be a valid DNS name — node-a or worker-3.cluster.internal — and it must match the node’s certificate.

QUIC and TLS

Quic::new(bind, tls) listens on bind (a UDP address) with the identity tls. For production, use mutual TLS with your own CA:

use zestors::prelude::*;
fn main() -> Result<(), Box<dyn std::error::Error>> {
let tls = Tls::from_pem(
    &std::fs::read("ca.pem")?,     // the cluster CA that peers must chain to
    &std::fs::read("node-a.pem")?, // this node's certificate (chain)
    &std::fs::read("node-a.key")?, // this node's private key
)?;
let config = ClusterConfig::new("node-a", Quic::new("0.0.0.0:7000".parse()?, tls));
let _ = config; Ok(())
}

• Every node presents a certificate signed by the CA, and verifies its peers’ certificates. Holding a certificate from the CA is what lets a node join.
• A node’s certificate must carry exactly one DNS name, the node’s name. What a peer is called in the cluster is what its certificate says.

For example, with openssl:

# The cluster CA, once.
openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:prime256v1 -nodes \
  -keyout ca.key -out ca.pem -days 3650 -subj "/CN=my-cluster-ca"

# A certificate for the node named node-a.
openssl req -newkey ec -pkeyopt ec_paramgen_curve:prime256v1 -nodes \
  -keyout node-a.key -out node-a.csr -subj "/CN=node-a"
openssl x509 -req -in node-a.csr -CA ca.pem -CAkey ca.key -CAcreateserial \
  -out node-a.pem -days 365 -extfile <(printf "subjectAltName=DNS:node-a")

Tls::insecure_dev() skips all verification: each node makes a self-signed certificate, and anyone who can reach the port can join as any node. The traffic is still encrypted. Use it only for local development and examples.

Starting and stopping

• NodeStatus (from cluster.status(), or wait_for_status) moves from Starting to Up, then to Leaving on the way out.
• A node that the rest of the cluster declared down, and that can’t rejoin, is Defunct. It keeps running its supervisor, but no longer takes part in the cluster. Watch for it and restart the process.
• To stop a node from inside the program, signal its root supervisor, as with Node. run() then leaves the cluster and returns.

Addressing and sending

Getting a ClusterAddress

Cluster makes addresses:

For an actor on another node, address and address_dyn ask that node whether an actor by that name is there, and whether it accepts the messages. They fail with an AddressError if not. For a name on this node they look in the local registry without touching the network.

    // Addressing an actor on another node asks that node, so first wait until
    // the caller has joined the host.
    caller_cluster.wait_for_members(1).await;
    let greeter = caller_cluster
        .address::<GreeterInterface>(GlobalName::new("greeter", "host"))
        .await
        .unwrap();
    println!("{}", greeter.call(Greet("world".into())).await.unwrap());

    // The same works for an actor on this node: the host's own address for its
    // greeter delivers locally, without encoding the message.
    let local_greeter = host_cluster
        .address::<GreeterInterface>(GlobalName::new("greeter", "host"))
        .await
        .unwrap();
    println!(
        "{}",
        local_greeter.call(Greet("host".into())).await.unwrap()
    );

    // A message with a reply channel in it: send the `RemoteRequest` inside the
    // message, and wait on the `Reply`.
    let (request, reply) = RemoteRequest::new();
    greeter
        .cast(CountLetters {
            text: "world".into(),
            reply: request,
        })
        .await
        .unwrap();
    println!("{} letters", reply.await.unwrap());

A few things to know:

• Wait for the node to be a member. Addressing or sending to a node this node doesn’t know yet fails with CastFailure::NotAMember. After starting, use cluster.wait_for_members(n) or wait_until.
• An address follows the name, not the actor. A remote ClusterAddress is resolved by name on the other node for every message. If the actor is restarted under the same name, for example by its supervisor, the address reaches the new one. If nothing holds the name any more, sends fail with RemoteError::NoSuchActor.
• Every named actor is reachable. Any registered actor on a node can be addressed by any other member, and sent any registered message it accepts. Access is controlled by cluster membership (the TLS certificates), not per actor.

Sending

ClusterAccepts (in the prelude) is the counterpart of Accepts:

Each has a *_with(msg, ClusterCallOptions) variant.

Returning from cast means the message was queued for sending, not that it arrived. Only a reply confirms that the actor got it.

Timeouts

A call to an actor on another node waits at most:

1. the timeout in its ClusterCallOptions, if set; otherwise
2. the address’s timeout, set with ClusterAddress::with_timeout; otherwise
3. the node’s ClusterConfig::call_timeout, 30 seconds by default.

It then fails with ClusterReplyError::Timeout.

A call to an actor on this node only has a timeout if ClusterCallOptions::timeout sets one. Otherwise it waits as long as a local call would.

Local and remote, one type

A ClusterAddress can point to either side. Code that holds one works the same wherever the actor turns out to be, with these differences:

• a local actor gets the message itself, without encoding;
• a local actor’s queue applies backpressure the usual way, while a remote node refuses messages when overloaded (see Delivery and failure);
• the default timeout only applies to remote actors.

is_local(), is_remote() and node() tell you which it is. local_address() returns the plain Address for a local actor, which you need for messages that aren’t remote.

Wrapping a ClusterAddress

Implement ClusterActorRef for a type of your own that holds a ClusterAddress, and it gets ClusterAccepts and ClusterActorOps too. This is useful for a typed client struct around a remote service.

Replies inside messages

A message’s reply type is sent back to the sender automatically. Sometimes the reply channel has to be part of the message itself: the actor forwards the job to another actor, which answers directly, or the message is a cast whose answer comes later.

Locally that channel is a Request<T>. Across the network it is a RemoteRequest<T>:

use serde::{Deserialize, Serialize};
use zestors::interface::Message;
use zestors::prelude::*;

#[derive(Message, StableId, Serialize, Deserialize, Debug)]
#[msg(id = "7c6f8a0e-2b1d-4e3f-9a8b-0c1d2e3f4a5b")]
struct CountLetters {
    text: String,
    reply: RemoteRequest<usize>,
}

#[tokio::main]
async fn main() {
// Keep the `Reply`, send the `RemoteRequest` in the message.
let (request, reply) = RemoteRequest::new();
let msg = CountLetters { text: "world".into(), reply: request };

// ...the actor that receives `msg`, on whichever node, answers it:
let CountLetters { text, reply: request } = msg;
request.reply(text.chars().count()).unwrap();

assert_eq!(reply.await.unwrap(), 5);
}

When the message goes to another node, the request stays on the sending node, and the receiving actor gets a stand-in. Whatever the actor answers is sent back and resolves the original. Locally it is just a Request.

• There is no timeout. Like a local request, the Reply waits until it is answered, or fails when the request is dropped or the node it went to is lost.
• If the message is never sent, the request is dropped, and the Reply fails.
• A RemoteRequest only works inside a message. It can’t be serialized any other way, and can’t be part of a reply.
• T has to be encodable, like any message.

Operating on remote actors

ClusterActorOps (in the prelude) is the counterpart of ActorOps for a ClusterAddress. Everything that doesn’t send a message lives here:

For an actor on another node, each of these is a round trip. So they are all async, and all can fail with a ClusterOpError: the request couldn’t be sent, or no answer came. They are not queued behind the actor’s messages, so they answer even when the actor has a backlog.

Monitoring

The monitor_* family works as it does locally, and waits across nodes. Their results are nested:

• the outer Result is about the network;
• the inner one is what the actor did.

For example, monitor_exit() gives Ok(Ok(())) for a normal exit and Ok(Err(e)) for a failed one.

use serde::{Deserialize, Serialize};
use zestors::{
    distr::{ClusterReplyError, ClusterOpError, sim::SimNetwork},
    interface::{Envelope, Interface, Message},
    prelude::*,
    runtime::spawn,
    supervisor::Supervisor,
};
#[derive(Message, StableId, Serialize, Deserialize, Debug)]
#[msg(id = "b5a4c0de-0000-4000-8000-000000000002")]
struct Work;
#[derive(Interface, Debug)]
enum WorkerInterface { Work(Envelope<Work>) }
#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
let net = SimNetwork::new(1);
let a = ClusterNode::new(Supervisor::blueprint().rand_name(),
    ClusterConfig::new("node-a", net.backend("10.0.0.1:7000")));
let b = ClusterNode::new(Supervisor::blueprint().rand_name(),
    ClusterConfig::new("node-b", net.backend("10.0.0.2:7000"))
        .seed(Seed::new("node-a", "10.0.0.1:7000"))
        .register::<Work>());
let _worker = spawn(Name::new_static("worker"), |mut inbox: Inbox<WorkerInterface>| async move {
    while inbox.recv().await.is_some() {}
    Ok(())
}).unwrap();
let cluster = a.cluster();
tokio::spawn(a.run());
tokio::spawn(b.run());
cluster.wait_for_members(1).await;
// On node-a, for an actor on node-b:
let worker = cluster
    .address::<WorkerInterface>(GlobalName::new("worker", "node-b"))
    .await
    .unwrap();

let exited = tokio::spawn({
    let worker = worker.clone();
    async move { worker.monitor_exit().await }
});

worker.signal_shutdown().await.unwrap();

match exited.await.unwrap() {
    Ok(Ok(())) => println!("exited normally"),
    Ok(Err(error)) => println!("exited with an error: {error}"),
    // The actor's node was lost: there is no telling what happened to it.
    Err(ClusterOpError::Reply(ClusterReplyError::Disconnected)) => println!("node lost"),
    Err(other) => println!("couldn't monitor: {other}"),
}
}

A monitor on another node:

• is held by that node until the status is reached. It has no timeout, not even the call timeout, so it can wait for as long as the actor lives;
• ends with ClusterReplyError::Disconnected if the connection to that node is lost, or the node leaves or fails. This is Erlang’s noconnection: the actor may still be running;
• is cancelled on the other node when you drop the future;
• is cleaned up by the other node if your node goes away.

monitor_init differs slightly from the local one: an actor whose name is reserved but has never been spawned counts as Exited.

Supervising across nodes

A supervisor supervises actors in its own process only. To react to an actor on another node, monitor it, for example from a Handler’s next_event, and decide there what to do when it exits or its node disconnects.

Delivery and failure

A local message can only fail because the actor is gone. A remote one crosses a network, and can fail on the way. This chapter lists what is and isn’t guaranteed.

At most once

A message is delivered at most once. Nothing is ever resent. If no reply comes, the message may or may not have been handled. Retrying is up to you, so make messages you retry safe to handle twice.

Ordering

Messages from one node to one actor arrive in the order they were sent. This holds per sending node. Two nodes messaging the same actor have no order between their messages.

Each node spreads its traffic to a peer over several independent streams, or lanes. ClusterConfig::lanes, 4 by default, sets how many. An actor always uses the same lane, so its messages stay in order, while a large message to one actor doesn’t hold up messages to actors on other lanes. With one lane, everything to a peer is in order.

Operations — signals, status reads, monitors — skip the actor’s message queue, as they do locally. A signal_shutdown can therefore overtake messages sent before it.

When messages are lost

A remote message can disappear without an error at the sender:

• The receiving actor is overloaded. A node takes in bursts, but once about 100 000 messages or 16 MiB are waiting for one actor, it refuses more with RemoteError::Overloaded. For a call that error comes back as the reply. A cast has no reply, so a refused cast is dropped silently. The only trace is a warning logged on the receiving node. (A local cast waits for room instead.)
• The connection is down. While a node reconnects to a peer, messages for that peer are dropped. Calls among them fail when their timeout runs out.
• The node leaves or fails after the message was sent. Pending calls fail with ClusterReplyError::Disconnected, and so do monitors.

If a message must not be lost, call it, and treat an error as “unknown outcome”.

Timeouts

Remote calls time out after ClusterConfig::call_timeout (30 s) unless the address or the call sets another timeout. Local calls through a ClusterAddress have no default timeout. Monitors never time out. See Addressing and sending.

Which error means what

Sending returns one of three errors:

CastFailure, the reasons the message never left:

• NotRunning: this node hasn’t started, or has stopped;
• NotAMember: the target node isn’t a known member;
• Unreachable: the target node can’t be connected to right now;
• Full: too much is queued for that node (only from try_cast);
• TooLarge: the encoded message is over 4 MiB;
• Encode: encoding failed;
• Closed, NotAccepted: the actor is on this node, and is closed or doesn’t accept the message.

ClusterReplyError, the reasons the message was sent but no reply came:

• Remote(RemoteError): the other node answered with an error (below);
• Disconnected: the node was lost first; outcome unknown;
• Timeout: no reply in time; outcome unknown;
• Decode: the reply couldn’t be decoded.

RemoteError, what the other node answered:

• UnknownMessage: it doesn’t have this message registered;
• NoSuchActor: no actor holds that name;
• NotAccepted: the actor doesn’t accept this message;
• Closed: the actor is shutting down;
• Overloaded: see above;
• NoReply: the actor dropped the request without answering. A local actor that drops a request gives the same error;
• Decode, Encode, TooLarge: the message or reply didn’t survive the wire;
• Unknown: a newer node sent an error this version doesn’t know.

ClusterActorOps methods fail with ClusterOpError, which is NotSent or Reply in the same way. Cluster::address fails with an AddressError: NoSuchActor, TypeMismatch (the actor doesn’t accept the messages), or Remote (the node couldn’t be asked).

Membership and events

Cluster shows which other nodes are up:

• members() lists them, and member(name) looks one up;
• is_reachable(name) tells whether this node can currently connect to it;
• wait_for_members(n) waits until exactly n other nodes are up;
• wait_until(condition) waits for any condition on the member list.

Events

subscribe() returns a stream of ClusterEvents:

A node restarted before anyone noticed shows up as Failed for the old incarnation, followed by Up for the new one.

The stream is a broadcast channel: a receiver that falls far behind misses events. To know the members right now and every change after that, use subscribe_with_snapshot(). Each change is then in either the snapshot or the stream, never both and never neither. Calling members() and subscribe() separately can miss a change in between.

    // Development only: use `Tls::from_pem` to authenticate cluster members.
    let mut config = ClusterConfig::new(name, Quic::new(bind, Tls::insecure_dev().unwrap()));
    for seed in args {
        let (seed_name, seed_addr) = seed.split_once('=').expect("seed must be name=addr");
        config = config.seed(Seed::new(
            seed_name,
            seed_addr.parse::<SocketAddr>().unwrap(),
        ));
    }

    let node = ClusterNode::new(Supervisor::blueprint().rand_name(), config);

    let mut events = node.cluster().subscribe();
    tokio::spawn(async move {
        while let Ok(event) = events.recv().await {
            tracing::info!("{event:?}");
        }
    });

    node.run().await

Run it in a few terminals to watch nodes join, and press Ctrl+C or kill -9 one of them to see the others notice it leave or fail:

cargo run -p zestors --features distr --example cluster -- node-a 127.0.0.1:7001
cargo run -p zestors --features distr --example cluster -- node-b 127.0.0.1:7002 node-a=127.0.0.1:7001
cargo run -p zestors --features distr --example cluster -- node-c 127.0.0.1:7003 node-a=127.0.0.1:7001

Failure detection

Membership uses SWIM: nodes probe each other at random, suspect a node that stops answering, and declare it down once the suspicion isn’t refuted in time. How quickly a crash is noticed depends on the membership settings. ClusterConfig::expected_size tunes the defaults for a cluster of that size, and foca_config replaces them entirely.

A node declared down that can’t rejoin becomes NodeStatus::Defunct. It keeps running, but no longer takes part in the cluster; restart it.

A node’s incarnation grows with every start, so peers can tell a restarted node from the old one. It is based on the clock. If the clock can be set back between restarts, use ClusterConfig::generation_store(path) to persist it.

Testing with sim

The sim feature of zestors-distr runs whole clusters inside one test, on a simulated network. Enable it for tests only:

[dev-dependencies]
zestors-distr = { version = "0.3", features = ["sim"] }
tokio = { version = "1", features = ["test-util"] }

Together with #[tokio::test(start_paused = true)] the cluster runs on virtual time. Timeouts and failure detection that take seconds for real take no time, and a run with the same seed repeats exactly.

The simulation runs the real membership protocol and messaging; only the transport is replaced. It doesn’t model TLS, connection setup or reconnect backoff, which the QUIC backend’s own tests cover.

Real nodes on a simulated network

SimNetwork::backend(addr) is a Backend, so a normal ClusterNode runs on it unchanged. Any address works, as long as each node has its own.

use zestors::{distr::sim::SimNetwork, prelude::*, supervisor::Supervisor};

// In a test: #[tokio::test(start_paused = true)]
#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
let net = SimNetwork::new(7);
let node = |name: &str, addr: &str| {
    ClusterConfig::new(name, net.backend(addr))
};

let a = ClusterNode::new(Supervisor::blueprint().rand_name(), node("node-a", "10.0.0.1:1"));
let b = ClusterNode::new(
    Supervisor::blueprint().rand_name(),
    node("node-b", "10.0.0.2:1").seed(Seed::new("node-a", "10.0.0.1:1")),
);
let (a_cluster, b_cluster) = (a.cluster(), b.cluster());
tokio::spawn(a.run());
tokio::spawn(b.run());
a_cluster.wait_for_members(1).await;

// Cut the network in two: each side declares the other down.
net.partition(&["node-a"], &["node-b"]);
a_cluster.wait_for_members(0).await;
b_cluster.wait_for_members(0).await;
}

The network can also slow down (set_latency) and heal (heal). foca_config_mut and timings_mut tune the nodes it starts.

Membership-only nodes

SimNetwork::start(name, addr, seeds) starts a lighter node that takes part in membership only, without a supervisor or messaging. SimNode::leave() and SimNode::crash() stop it cleanly or abruptly. These are useful for testing code that reacts to ClusterEvents.

Things to know

• All simulated nodes share the process, and therefore one Registry. An actor spawned in the test is visible to every node, so GlobalName::new("x", "node-b") finds it through node-b. Spawn each actor once, and address it through the node that should serve it.
• Actor names must be unique in the process. Each test runs in its own process under cargo nextest, but with cargo test the tests in one binary share one registry.

Custom backends

A backend is the network a cluster runs on. Quic is the one provided; the Backend trait in zestors-distr-backend lets a cluster run over anything else. The simulated network in sim is itself a backend.

A backend is deliberately small. It only has to:

• connect and accept with verified identity. Connection::peer() is the node on the other end, established by the backend from a certificate, a key or credentials, and never taken from the peer’s word. The cluster trusts it completely.
• offer independent, ordered, reliable streams. A stall on one stream must not hold up the others.
• offer unreliable datagrams, if it can. A backend without them returns DatagramError::Unsupported, and the cluster uses streams instead.

Everything else is done by the cluster, the same for every backend: one connection per peer, reconnecting with backoff, telling unreachable from down, framing, ordering and routing messages.

The three traits:

NodeName and NodeAddr are opaque to the cluster. The backend decides what a name must look like (for QUIC, a DNS name) and how an address is resolved (for QUIC, host:port).

See the zestors-distr-backend API docs for the full contract, and zestors-distr-quic for a complete implementation.

Feature flags

zestors

Without distr, the cluster crates and the QUIC stack (quinn, rustls) aren’t compiled at all.

zestors-distr

_ra

Several crates define a _ra feature. It is only there so that rust-analyzer in this workspace (see .vscode/settings.json) checks feature-gated code. Don’t enable it in your own project.

Derive attributes

zestors has four derives, all in zestors::prelude:

Each derive reads its options from #[msg(...)] and #[zestors(...)] attributes. The two are interchangeable.

A remote message usually combines both keys in one attribute:

use serde::{Deserialize, Serialize};
use zestors::interface::Message;
use zestors::prelude::*;

#[derive(Message, StableId, Serialize, Deserialize, Debug)]
#[msg(reply = "Vec<u8>", id = "4f9e5a8b-7c6d-4e3f-8a1b-2c3d4e5f6a7b")]
struct Read {
    path: String,
}

A few details:

• Quote reply types that contain <: reply = "Option<String>". A plain path such as reply = u32 needs no quotes.
• The path keys are only needed when you depend on the sub-crates directly instead of on zestors. For example, with zestors-interface on its own: #[zestors(interface_path = "zestors_interface")].
• HandlerInterface generates code that names rootcause::Report, so the crate using it must depend on rootcause.
• Interface only works on enums whose variants each hold exactly one Envelope<M>, and doesn’t support generics. Message does support generic types.
• StableId on a generic type is never auto-registered. Register each concrete type with ClusterConfig::register.