Usage examples

Five small programs, each runnable, each asserting what it shows. The code on this page is included from the files themselves when the book is built, so it cannot drift from what compiles: cargo test builds every example, and CI runs them.

cargo run --example ownership_handover
cargo run --example swim_membership
cargo run --example swim_node
cargo run --example actor_hosting
cargo run --example actor_basic

They are ordered the way the book is: the base first, then what sits on it. Each example is about using a piece, so for why it is shaped that way read the chapter named above it.

Handing an actor from one node to another

Chapter: Ownership.

A owns an actor at epoch 41 and writes. A agrees to hand it to B, which burns the epoch at once, so A’s next write at 41 is refused FENCED. B takes the actor at 42.

Things to notice:

• No call names the acting node. Who is calling is the Context, which a transport fills in. Here it is built by hand (from) because there is no network; over a real connection nothing else changes.
• The refusal comes from the registry, not from A. A still believes it holds 41.
• The floor. MemoryFloor::at(40) seeds the durable record of the highest epoch used, which is what keeps a restart from rewinding the actor.

//! The headline story, in process: a handover that fences the outgoing owner.
//!
//! `A` owns an actor at epoch 41 and writes. `A` agrees to hand it to `B`, which burns
//! the epoch at once. `A`, not yet knowing, writes again at 41 and is **fenced**. `B`
//! takes the actor at 42 and writes.
//!
//! That one refusal is what the protocol exists for. The registry is called directly
//! here, with no network, so the example is the shortest possible account of it. Over a
//! real connection nothing changes except where the caller's identity comes from:
//! [`Context::remote`] is filled in by the transport, never taken from an argument, and
//! the registry maps that address to a node through a roster fixed at construction.
//!
//! Run it with `cargo run --example ownership_handover`.

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;

use jetstream::prelude::jetstream_rpc::context::{Context, RemoteAddr};
use jetstream::prelude::*;
use jetstream_distributed::id::Id;
use jetstream_distributed::ownership::service::{MemoryFloor, Offer, Ownership, Registry};

const ADDR_A: IpAddr = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1));
const ADDR_B: IpAddr = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2));
const A: Id = Id(0);
const B: Id = Id(1);

/// The context a call from `addr` carries: what a transport would fill in.
fn from(addr: IpAddr) -> Context {
    Context::new(Some(RemoteAddr::IpAddr(addr)), None)
}

#[tokio::main]
async fn main() -> Result<()> {
    // Who is which node, by address, and where the epochs start. The floor is a
    // durable record of the highest epoch the registry has used, which is what keeps
    // a restart from rewinding the actor; seeding it at 40 makes the first epoch 41.
    let roster = HashMap::from([(ADDR_A, A), (ADDR_B, B)]);
    let registry = Registry::new(roster, A, Arc::new(MemoryFloor::at(40)))?;

    // The actor is offered to A at 41, and A takes it and writes.
    assert_eq!(registry.poll(from(ADDR_A)).await?, Offer::Grant(41));
    registry.acquire(from(ADDR_A), 41).await?;
    assert_eq!(registry.write(from(ADDR_A), 41).await?, 1);
    println!("A owns the actor at epoch 41 and wrote");

    // A agrees to hand it to B. The epoch is burned from this moment, so A is already
    // fenced out, whether or not it has been told.
    registry.begin_transfer(from(ADDR_A), B).await?;

    // The refusal this protocol exists for. A still believes it holds epoch 41; the
    // registry knows better, and says so in its own word.
    let fenced = registry.write(from(ADDR_A), 41).await.unwrap_err();
    assert_eq!(fenced.code(), Some("FENCED"));
    println!("A's write at the burned epoch 41 was refused: FENCED");

    // A is told to stop, and acknowledges at the transfer's epoch.
    assert_eq!(registry.poll(from(ADDR_A)).await?, Offer::Revoke(42));
    registry.release(from(ADDR_A), 42).await?;

    // Only now is the actor on offer to B, at the new epoch.
    assert_eq!(registry.poll(from(ADDR_B)).await?, Offer::Grant(42));
    registry.acquire(from(ADDR_B), 42).await?;
    assert_eq!(registry.write(from(ADDR_B), 42).await?, 2);
    println!("B took the actor at epoch 42 and wrote");

    // The audit log is what the test oracles read, and it records the refusal.
    for event in registry.log() {
        println!("  {event:?}");
    }
    Ok(())
}

A node’s view of a failure

Chapter: SWIM.

A node suspects a peer because its own probe failed, then declares it dead when the suspicion times out. A weaker report cannot undo that; a higher incarnation can, which is how a peer refutes an accusation.

Things to notice:

• suspect and declare_dead are methods on the node and not on the wire. What arrives from outside is a batch of Updates, merged by apply.
• The incarnation is an argument. It is the one the node had when the probe began, so a refutation that lands mid-probe cannot be accused after the fact.
• Applying a message twice changes nothing, which is the property that lets gossip be delivered at least once and in any order.

//! A SWIM node's view of a failure, and of being wrongly accused.
//!
//! A [`Node`] keeps one record per member: an incarnation number and a status. Two
//! records are compared by a key, a higher incarnation always winning and, at equal
//! incarnations, the more-dead status winning. Everything below is that one rule
//! applied to whichever record arrived.
//!
//! Suspecting a peer and declaring it dead are things a node does about *its own* view,
//! because its own probe failed, so they are methods on the node and not on the wire.
//! What does arrive from outside is a batch of [`Update`]s, merged by [`Node::apply`].
//!
//! Run it with `cargo run --example swim_membership`.

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr};

use jetstream_distributed::id::Id;
use jetstream_distributed::swim::service::{Liveness, Node, Update};

const ME: Id = Id(1);
const PEER: Id = Id(2);

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Who is a member, by address. A node off the roster cannot be built.
    let roster = HashMap::from([
        (IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), ME),
        (IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)), PEER),
    ]);
    let node = Node::new(roster, ME)?;

    // Everyone starts alive at incarnation 0. Nothing is read from disk.
    assert_eq!(
        node.record(PEER),
        Some(Update {
            node: PEER,
            incarnation: 0,
            status: Liveness::Alive
        })
    );

    // A probe of the peer failed: believe it suspect, at the incarnation the node had
    // recorded when the probe began. That is an argument and not re-read, because a
    // probe is slow and a refutation can land while it is in flight.
    assert!(node.suspect(PEER, 0)?);
    println!("peer is suspect: {:?}", node.record(PEER).unwrap().status);

    // The suspicion timed out. This only takes effect if the peer is still suspect at
    // the incarnation the timer was started for.
    assert!(node.declare_dead(PEER, 0)?);
    println!("peer is dead:    {:?}", node.record(PEER).unwrap().status);

    // A weaker report does not undo a death: alive at the same incarnation loses.
    node.apply(&[Update {
        node: PEER,
        incarnation: 0,
        status: Liveness::Alive,
    }])?;
    assert_eq!(node.record(PEER).unwrap().status, Liveness::Dead);

    // Only a higher incarnation does. This is the refutation the peer makes when it
    // hears it is accused: alive at one incarnation above the accusation.
    node.apply(&[Update {
        node: PEER,
        incarnation: 1,
        status: Liveness::Alive,
    }])?;
    let record = node.record(PEER).unwrap();
    assert_eq!((record.incarnation, record.status), (1, Liveness::Alive));
    println!(
        "peer refuted:    incarnation {} {:?}",
        record.incarnation, record.status
    );

    // Applying the same message twice changes nothing the second time, which is what
    // lets gossip survive being delivered at least once and in any order.
    let again = [Update {
        node: PEER,
        incarnation: 1,
        status: Liveness::Alive,
    }];
    node.apply(&again)?;
    node.apply(&again)?;
    assert_eq!(node.record(PEER).unwrap().incarnation, 1);
    Ok(())
}

A SWIM cluster that members join and leave

Chapter: SWIM.

The previous example calls the node’s methods by hand. This one is the loop around it: a listener that serves ping and sync, a join through a seed, a probe that pings one peer a second and piggybacks its whole view, a suspicion when a ping fails, a timer that turns an unrefuted suspicion into death, and a way to leave. Run three of them; the second and third join through a seed:

cargo run --example swim_node -- run 1
cargo run --example swim_node -- run 2 1
cargo run --example swim_node -- run 3 2

Type leave in one terminal and the others record a departure, not a failure. Press Ctrl-C in another and they suspect it, then declare it dead. Start it again and it comes back at incarnation 1: its first caller tells it the cluster has it as gone, that out-ranks its record, and it answers with a higher incarnation. With no arguments the example does all of that in one process and asserts it, which is what CI runs.

Things to notice:

• A join is a sync with a seed. There is no join call. The seed’s admission policy lets the address in, its reply is the whole table, and the others hear of the newcomer by gossip. Node 3 above never calls node 1 to join.
• Who may join is a policy, and this example’s is open. Every loopback address 127.0.0.1 to 127.0.0.250 is a member. A real deployment writes a stricter one; SWIM does not say who should be admitted. The member number comes from the policy and the connection, never from the request.
• Leaving and dying are different records. leave writes Left, which out-ranks a failure verdict at the same incarnation, so a suspicion timer already running cannot turn a departure into a death. A node that is stopped without leaving is simply dead.
• The timing is this file’s choice and nothing is proved about it. The merge rule, the departure’s precedence, the refutation and the refusal of strangers are the verified and tested parts. When to probe, whom, and how long to wait are plain Rust.
• No indirect ping, and members are never forgotten: a departed member’s record is kept, and counts against the bound on members.
• Identity is the connection’s source address. Each member binds 127.0.0.<id> and connects from it. That works on Linux; on another OS give each member its own host. An address is not a credential, and where a member lives is the deployment’s mapping, not something SWIM gossips.

//! A SWIM node you can run as a process, in a cluster that members join and leave.
//!
//! [`Node`] is a table that obeys the merge rule and answers its peers; it has no
//! clock and no socket. What turns it into a member of a cluster is the part this
//! file writes, in plain Rust that Verus never sees and that nothing here proves:
//!
//! * **join**: start knowing only yourself, and `sync` with a seed. The seed's
//!   admission policy lets your address in as a member, and its reply is the whole
//!   cluster. The seed tells the others by gossip; you never call them to join;
//! * **serve**: accept a connection, hand it to the generated `MembershipService`;
//! * **probe**: every period, `ping` one peer, piggybacking this node's whole view;
//! * **suspect**: a ping that fails or times out makes the node believe the peer
//!   suspect, at the incarnation it had recorded *when the probe began*;
//! * **declare dead**: a suspicion that is not refuted within its timeout becomes
//!   death, but only if the peer is still suspect at that incarnation;
//! * **leave**: record yourself `Left` and tell every peer. A node that is stopped
//!   without doing so is, to the cluster, a node that died.
//!
//! The merge rule, the departure's precedence over a failure verdict, the
//! refutation and the refusal of strangers are the verified and tested parts.
//! *When* to probe, whom, and how long to wait are this file's choices, and
//! `crate::swim::model` says nothing about timing. There is no indirect ping
//! (`ping_req`): a peer that this node alone cannot reach is suspected on this
//! node's word.
//!
//! **Who may join is a policy, and this example's is open**: `127.0.0.<k>` for
//! `k` in `1..=250` is member `k`, and anyone who can reach a seed from such an
//! address is in. A real deployment writes a stricter one. Members are addressed
//! by loopback address because a node knows its caller by address; each listens
//! on port 9100 of its own address and connects from it. That works on Linux,
//! where all of `127.0.0.0/8` is loopback; elsewhere, give each member its own
//! host instead.
//!
//! Run a cluster in three terminals; the second and third join through a seed:
//!
//! ```text
//! cargo run --example swim_node -- run 1
//! cargo run --example swim_node -- run 2 1
//! cargo run --example swim_node -- run 3 2
//! ```
//!
//! Type `leave` in one to see the others record a departure. Press Ctrl-C in
//! another to see them suspect it and declare it dead. Start it again and watch it
//! be told what it was and outrank that. With no arguments the example does all of
//! that inside one process and asserts it: `cargo run --example swim_node`.

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::time::{Duration, Instant};

use jetstream::prelude::jetstream_rpc::session::{Session, SingleLaneSession};
use jetstream::prelude::*;
use jetstream_distributed::id::Id;
use jetstream_distributed::swim::service::{
    Admission, Liveness, Membership, Node, Update,
    membership_protocol::{MembershipChannel, MembershipService},
};
use tokio::net::{TcpListener, TcpSocket};
use tokio::task::JoinHandle;

const PORT: u16 = 9100;

/// The timings. Nothing proved depends on them.
#[derive(Clone, Copy)]
struct Timing {
    /// How often this node pings one peer.
    probe_every: Duration,
    /// How long a call may take before the peer is suspected.
    probe_timeout: Duration,
    /// How long a suspicion stands before it becomes death.
    suspicion: Duration,
}

const LIVE: Timing = Timing {
    probe_every: Duration::from_secs(1),
    probe_timeout: Duration::from_millis(500),
    suspicion: Duration::from_secs(5),
};

const FAST: Timing = Timing {
    probe_every: Duration::from_millis(40),
    probe_timeout: Duration::from_millis(100),
    suspicion: Duration::from_millis(200),
};

fn addr_of(member: u128) -> IpAddr {
    IpAddr::V4(Ipv4Addr::new(127, 0, 0, member as u8))
}

/// Who may join: this example lets in any loopback address `127.0.0.1` to
/// `127.0.0.250`, as the member with that last number.
fn policy() -> Admission {
    std::sync::Arc::new(|a| match a {
        IpAddr::V4(v4) => match v4.octets() {
            [127, 0, 0, k @ 1..=250] => Some(Id(k as u128)),
            _ => None,
        },
        IpAddr::V6(_) => None,
    })
}

/// Answer pings and syncs on `127.0.0.<me>:9100` for as long as the task lives.
async fn serve(node: Node, me: u128) -> std::io::Result<()> {
    let listener = TcpListener::bind(SocketAddr::new(addr_of(me), PORT)).await?;
    loop {
        let (stream, _) = listener.accept().await?;
        let node = node.clone();
        tokio::spawn(async move {
            // The session reads the peer's address off the stream, and `Node`
            // admits or refuses it: that is where "who is calling" comes from.
            let session = SingleLaneSession::<MembershipChannel, _, _>::service_io(stream);
            let Ok(lane) = Session::<MembershipChannel>::accept_lane(&session).await else {
                return;
            };
            let mut service = MembershipService { inner: node };
            let _ = jetstream_rpc::server::run(&mut service, lane).await;
        });
    }
}

#[derive(Clone, Copy)]
enum Call {
    /// SWIM's ping, with the view piggybacked.
    Ping,
    /// Push/pull: send the view, get the peer's whole table back. How a node joins.
    Sync,
}

/// One call to `peer`, made from this member's own address, carrying `view`.
async fn call(
    me: u128,
    peer: u128,
    kind: Call,
    view: Vec<Update>,
    within: Duration,
) -> Result<Vec<Update>> {
    let call = async {
        // Bind the source so the peer sees `127.0.0.<me>` and applies its policy to it.
        let socket = TcpSocket::new_v4()?;
        socket.bind(SocketAddr::new(addr_of(me), 0))?;
        let stream = socket.connect(SocketAddr::new(addr_of(peer), PORT)).await?;
        // The session owns the lane's I/O, so it outlives the channel.
        let session = SingleLaneSession::<MembershipChannel, _, _>::client_io(stream);
        let lane = Session::<MembershipChannel>::open_lane(&session).await?;
        let channel = MembershipChannel::new(64, Box::new(lane));
        channel.negotiate_version(u32::MAX).await?;
        match kind {
            Call::Ping => channel.ping(Context::default(), view).await,
            Call::Sync => channel.sync(Context::default(), view).await,
        }
    };
    match tokio::time::timeout(within, call).await {
        Ok(reply) => reply,
        Err(_) => Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "call timed out").into()),
    }
}

/// The members this node would probe: everyone it has heard of except itself and
/// those that left. A member declared dead is still probed, because it may come back.
fn peers(node: &Node, me: u128) -> Vec<u128> {
    node.view()
        .into_iter()
        .filter(|u| u.node.0 != me && u.status != Liveness::Left)
        .map(|u| u.node.0)
        .collect()
}

/// Join through `seed`, then probe whoever the cluster turns out to hold, forever.
async fn probe(node: Node, me: u128, seed: Option<u128>, timing: Timing, verbose: bool) {
    // A join is a sync with the seed; its reply is the cluster. The seed may not be
    // up yet, so keep asking.
    if let Some(seed) = seed {
        loop {
            match call(me, seed, Call::Sync, node.view(), timing.probe_timeout).await {
                Ok(table) => {
                    if let Err(e) = node.apply(&table) {
                        eprintln!("[{me}] refused the seed's table: {e}");
                    }
                    break;
                }
                Err(e) => {
                    if verbose {
                        eprintln!("[{me}] cannot reach seed {seed} yet ({e})");
                    }
                    tokio::time::sleep(timing.probe_every).await;
                }
            }
        }
    }
    let mut seen = node.log().len();
    let mut turn = 0usize;
    loop {
        tokio::time::sleep(timing.probe_every).await;
        let peers = peers(&node, me);
        if peers.is_empty() {
            continue;
        }
        let peer = peers[turn % peers.len()];
        turn += 1;
        // The incarnation this node holds *now*, before the ping: a refutation
        // may land while the ping is in flight, and the accusation must not
        // outlive it.
        let held = node.record(Id(peer)).expect("a member").incarnation;
        match call(me, peer, Call::Ping, node.view(), timing.probe_timeout).await {
            // The ack is the peer's own word about everyone the ping mentioned,
            // including itself: the evidence that it is up.
            Ok(ack) => {
                if let Err(e) = node.apply(&ack) {
                    eprintln!("[{me}] refused a reply from {peer}: {e}");
                }
            }
            Err(e) => {
                if node.suspect(Id(peer), held).unwrap_or(false) && verbose {
                    eprintln!("[{me}] ping to {peer} failed ({e}); suspecting it");
                }
                let node = node.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(timing.suspicion).await;
                    // A no-op unless the peer is still suspect at `held`.
                    let _ = node.declare_dead(Id(peer), held);
                });
            }
        }
        if verbose {
            let log = node.log();
            for event in &log[seen..] {
                eprintln!("[{me}] {event:?}");
            }
            seen = log.len();
        }
    }
}

/// Start member `me`: a fresh node that knows only itself, serving and probing,
/// and joining through `seed` if there is one.
fn start(
    me: u128,
    seed: Option<u128>,
    timing: Timing,
    verbose: bool,
) -> (Node, Vec<JoinHandle<()>>) {
    // A node starts at incarnation 0 and keeps nothing on disk. A restart is
    // corrected by the first peer that tells it what the cluster thinks of it.
    let node = Node::admitting(HashMap::from([(addr_of(me), Id(me))]), Id(me), policy())
        .expect("on its own roster");
    let server = {
        let node = node.clone();
        tokio::spawn(async move {
            if let Err(e) = serve(node, me).await {
                eprintln!("[{me}] cannot serve on 127.0.0.{me}:{PORT}: {e}");
            }
        })
    };
    let prober = tokio::spawn(probe(node.clone(), me, seed, timing, verbose));
    (node, vec![server, prober])
}

/// Leave: record the departure and tell every peer this node knows of.
async fn leave(node: &Node, me: u128, timing: Timing) {
    let said = node.leave();
    for peer in peers(node, me) {
        // A peer that cannot be reached is told by gossip from one that can.
        let _ = call(me, peer, Call::Ping, vec![said], timing.probe_timeout).await;
    }
}

fn status(node: &Node, of: u128) -> Option<Liveness> {
    node.record(Id(of)).map(|u| u.status)
}

async fn until(what: &str, mut ready: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(10);
    while !ready() {
        assert!(Instant::now() < deadline, "timed out waiting for {what}");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

fn stop(tasks: &[JoinHandle<()>]) {
    for task in tasks {
        task.abort();
    }
}

/// Members join through different seeds, one leaves, one is stopped, and the
/// one that left comes back.
async fn demonstrate() {
    // 1 starts alone; 2 joins through 1; 3 joins through 2, whose reply is how 3
    // learns of 1. 1 learns of 3 by gossip from 2.
    let (n1, _t1) = start(1, None, FAST, false);
    let (n2, t2) = start(2, Some(1), FAST, false);
    let (n3, t3) = start(3, Some(2), FAST, false);
    until("every member to know every other, alive", || {
        [(&n1, 2), (&n1, 3), (&n2, 3), (&n3, 1)]
            .iter()
            .all(|(n, of)| status(n, *of) == Some(Liveness::Alive))
    })
    .await;
    println!("three members joined through two seeds, and all know each other");

    // 3 leaves, and says so. The others record a departure, not a failure.
    leave(&n3, 3, FAST).await;
    until("1 and 2 to record 3 as having left", || {
        status(&n1, 3) == Some(Liveness::Left) && status(&n2, 3) == Some(Liveness::Left)
    })
    .await;
    stop(&t3);
    // A suspicion that was already running for 3 would have no effect: wait it out.
    tokio::time::sleep(FAST.suspicion * 2).await;
    assert_eq!(status(&n1, 3), Some(Liveness::Left), "left, and not dead");
    println!("member 3 left: members 1 and 2 recorded a departure, not a failure");

    // 2 is stopped without leaving. To the cluster that is a failure.
    stop(&t2);
    until("1 to declare 2 dead", || {
        status(&n1, 2) == Some(Liveness::Dead)
    })
    .await;
    println!("member 2 stopped without leaving: member 1 declared it dead");

    // 3 starts again, as a new process would: incarnation 0, no memory. It joins
    // through 1, which holds it as having left, and says so in its reply; that
    // out-ranks 3's own record, so 3 answers with a higher incarnation.
    let (n3, _t3) = start(3, Some(1), FAST, false);
    until("1 to see 3 alive again", || {
        status(&n1, 3) == Some(Liveness::Alive)
    })
    .await;
    let back = n1.record(Id(3)).unwrap();
    assert!(back.incarnation >= 1, "back only at a higher incarnation");
    assert!(n3.record(Id(3)).unwrap().incarnation >= 1);
    println!(
        "member 3 rejoined: alive again at incarnation {}",
        back.incarnation
    );
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let args: Vec<String> = std::env::args().skip(1).collect();
    let args: Vec<&str> = args.iter().map(String::as_str).collect();
    let (me, seed) = match args[..] {
        [] => return demonstrate().await,
        ["run", me] => (me, None),
        ["run", me, seed] => (me, Some(seed)),
        _ => {
            eprintln!("usage: swim_node [run <id> [<seed id>]]");
            std::process::exit(2);
        }
    };
    let me: u128 = me.parse().expect("a member number");
    let seed: Option<u128> = seed.map(|s| s.parse().expect("a member number"));
    assert!((1..=250).contains(&me), "1 <= id <= 250");

    let (node, tasks) = start(me, seed, LIVE, true);
    eprintln!("[{me}] member {me} on 127.0.0.{me}:{PORT}; type `leave` to leave");

    // Typing `leave` leaves. A closed stdin does nothing: only a person saying so
    // is a departure.
    let (told, mut leaving) = tokio::sync::oneshot::channel::<()>();
    std::thread::spawn(move || {
        let mut line = String::new();
        while std::io::stdin().read_line(&mut line).is_ok_and(|n| n > 0) {
            if line.trim() == "leave" {
                let _ = told.send(());
                return;
            }
            line.clear();
        }
    });

    let mut shown = Vec::new();
    loop {
        tokio::select! {
            _ = tokio::time::sleep(LIVE.probe_every * 2) => {}
            Ok(()) = &mut leaving => {
                leave(&node, me, LIVE).await;
                eprintln!("[{me}] left, and told the cluster");
                return;
            }
        }
        let view = node.view();
        if view != shown {
            let line: Vec<String> = view
                .iter()
                .map(|u| format!("{}:{:?}@{}", u.node.0, u.status, u.incarnation))
                .collect();
            println!("[{me}] view: {}", line.join(" "));
            shown = view;
        }
        if tasks.iter().any(|t| t.is_finished()) {
            std::process::exit(1);
        }
    }
}

Hosting an actor under a fence

Chapters: Services over Jetstream and the actor.

A supervisor hosts an ordinary service behind a mailbox. The fence is the question “may this host still act for its actor?”. When it closes, the next request is answered FENCED without reaching the actor. Draining then drops the service and leaves the mailbox answering.

Things to notice:

• The actor is a plain trait implementation. It has no idea it has a mailbox, a fence or a supervisor.
• spawn requires a fence. There is no way to host an actor that cannot be told it lost.
• The caller gets an ordinary generated channel. It cannot tell a local actor from a remote one.
• In a deployment the fence is the ownership registry’s may_write(host, epoch). That wiring is not built; the example uses a flag in its place.

//! Hosting an actor under a fence, calling it, and draining it.
//!
//! A [`LocalSupervisor`] hosts actors on one node, each behind its own mailbox. The
//! actor is an ordinary service; it is handed a [`Fence`](jetstream_distributed::actor::service::Fence), which is the question "may
//! this host still act for its actor?". In a deployment the fence is the ownership
//! registry's `may_write(host, epoch)`. Here it is a flag, so the example can close it.
//!
//! What it shows, in order: an actor is hosted and called; the fence closes and the
//! next request is refused `FENCED` without reaching the actor; the actor is drained
//! and its service dropped; and discovery stops pointing at it.
//!
//! Run it with `cargo run --example actor_hosting`.

use std::num::NonZeroU32;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use jetstream::prelude::*;
use jetstream_distributed::actor::service::{
    ActorRef, Capacity, LocalSupervisor, Status, Supervisor,
};
use jetstream_distributed::id::Id;
use jetstream_distributed::scheduling::discovery::{Discovery, DiscoveryError};

use counter_protocol::{CounterChannel, CounterService};

/// How many requests the actor's mailbox holds before a sender has to wait.
const ROOM: Capacity = Capacity::Bounded(NonZeroU32::new(16).unwrap());

/// This node, and the actor it hosts. Both are plain [`Id`]s.
const ME: Id = Id(1);
const COUNTER: Id = Id(42);

#[service]
trait Counter {
    /// How many calls the counter has handled, counting this one.
    async fn next(&self, ctx: Context) -> Result<u64>;
}

#[derive(Default)]
struct CounterImpl(AtomicU64);

impl Counter for CounterImpl {
    async fn next(&self, _ctx: Context) -> Result<u64> {
        Ok(self.0.fetch_add(1, Ordering::SeqCst) + 1)
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let supervisor = LocalSupervisor::new(ME);

    // The fence starts open. Anything `Fn() -> bool` is a fence.
    let owner = Arc::new(AtomicBool::new(true));
    let fence = {
        let owner = owner.clone();
        move || owner.load(Ordering::SeqCst)
    };

    // Host the actor. `spawn` requires a fence: a host with no way to be told it
    // lost the actor is not something the supervisor will run.
    supervisor
        .spawn::<CounterChannel, _, _>(
            COUNTER,
            ROOM,
            CounterService {
                inner: CounterImpl::default(),
            },
            fence,
        )
        .await?;
    assert_eq!(supervisor.status(COUNTER), Status::Running);

    // Call it. A transport onto the actor's mailbox becomes an ordinary generated
    // channel, so the caller cannot tell it from a remote one.
    let transport = supervisor
        .transport::<CounterChannel>(COUNTER, Context::default())
        .await?;
    let counter = CounterChannel::new(64, transport);
    counter.negotiate_version(u32::MAX).await?;
    assert_eq!(counter.next(Context::default()).await?, 1);
    assert_eq!(counter.next(Context::default()).await?, 2);
    println!("hosted and called: the counter is at 2");

    // Discovery, asked about this actor, names this node.
    assert_eq!(supervisor.discover(ActorRef(COUNTER)).await, Ok(vec![ME]));

    // The actor's epoch moves on: ownership passed to another host. The fence
    // closes, and the next request is refused without reaching the actor.
    owner.store(false, Ordering::SeqCst);
    let refused = counter.next(Context::default()).await.unwrap_err();
    assert_eq!(refused.code(), Some("FENCED"));
    println!("fence closed: the next call was answered FENCED");

    // Drain it. A handler that was running would be finished, not cancelled; here
    // there is none. The service is dropped and the mailbox stays, answering FENCED.
    supervisor.drain(COUNTER).await?;
    assert_eq!(supervisor.status(COUNTER), Status::Retired);
    assert_eq!(
        supervisor.discover(ActorRef(COUNTER)).await,
        Err(DiscoveryError::NotFound),
        "a retired actor is not where a caller should go"
    );
    println!("drained: the actor is retired and no longer discoverable here");
    Ok(())
}

Two actors, one calling the other

Chapter: The mailbox.

A holds a channel to B and calls it. A is generic over what B is, so it cannot tell a mailbox from anything else that speaks the protocol.

//! Two actors, each behind its own mailbox.
//!
//! `ActorAImpl` calls `ActorB` as `self.b.b(ctx).await`. The `B` it holds is a
//! generated channel whose transport is a [`Session`] on `B`'s mailbox, and `B`'s
//! implementation is served from the other end by [`serve`], one request at a
//! time. `ActorAImpl` is generic over `B: ActorB` and cannot tell. `main` reaches
//! `A` the same way, through `A`'s mailbox.

use std::num::NonZeroU32;

use jetstream::prelude::*;
use jetstream_distributed::actor::service::{Capacity, Delivery, Mailbox, Session, serve};

use actora_protocol::{ActorAChannel, ActorAService};
use actorb_protocol::{ActorBChannel, ActorBService};

/// How many messages each mailbox holds before a sender has to wait. Always the
/// caller's to say.
const MAILBOX: Capacity = Capacity::Bounded(NonZeroU32::new(16).unwrap());

#[service]
trait ActorA {
    async fn a(&self, ctx: Context) -> Result<()>;
}

#[service]
trait ActorB {
    async fn b(&self, ctx: Context) -> Result<()>;
}

struct ActorAImpl<B> {
    b: B,
}

impl<B> ActorA for ActorAImpl<B>
where
    B: ActorB + Send + Sync,
{
    async fn a(&self, ctx: Context) -> Result<()> {
        eprintln!("A: calling B");
        self.b.b(ctx).await?;
        eprintln!("A: B answered");
        Ok(())
    }
}

struct ActorBImpl;

impl ActorB for ActorBImpl {
    async fn b(&self, _ctx: Context) -> Result<()> {
        eprintln!("B: handling a call");
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // B, behind its mailbox.
    let b_mailbox = Mailbox::<Delivery<ActorBChannel>>::new(MAILBOX);
    let mut b = ActorBService { inner: ActorBImpl };
    tokio::spawn({
        let mailbox = b_mailbox.clone();
        async move { serve(mailbox, &mut b).await }
    });
    let b_channel = ActorBChannel::new(64, Box::new(Session::new(b_mailbox, Context::default())));
    b_channel.negotiate_version(u32::MAX).await?;

    // A, holding a channel to B, behind its own mailbox.
    let a_mailbox = Mailbox::<Delivery<ActorAChannel>>::new(MAILBOX);
    let mut a = ActorAService {
        inner: ActorAImpl { b: b_channel },
    };
    tokio::spawn({
        let mailbox = a_mailbox.clone();
        async move { serve(mailbox, &mut a).await }
    });
    let a_channel = ActorAChannel::new(64, Box::new(Session::new(a_mailbox, Context::default())));
    a_channel.negotiate_version(u32::MAX).await?;

    for _ in 0..2 {
        a_channel.a(Context::default()).await?;
    }
    Ok(())
}

Longer examples are the tests

The tests are real programs too, and a good deal longer. They are the better reading once the examples above make sense, because each one is a story with an expected ending:

They are plain Rust that Verus never sees. What they check is tested, not proved; see Evidence.