Leaving gracefully

Most nodes do not crash, they are stopped: for a deployment, a scale down or a restart. A node that stops on purpose can tell the cluster so, which spares everyone the failure detection of the previous step. In this step the device manager stops on request, and its node leaves the cluster.

Stopping the device manager

Command gets one more variant:

#[derive(Serialize, Deserialize)]
enum Command {
    Record {
        group_name: String,
        device_name: String,
        degrees: f64,
    },

    Retire {
        group_name: String,
        device_name: String,
    },

    Average {
        group_name: String,
        reply_to: ReplyTo<Option<f64>>,
    },

    Stop,
}

impl Manifest for Command {
    const MANIFEST: &'static str = "device-manager-command";
}

The device manager answers Stop with Control::Stop. It is the root actor of its node's actor system, so the actor system terminates with it.

Leaving the cluster

async fn manager_node() -> anyhow::Result<()> {
    let (addr, seeds) = node_args()?;
    join_cluster(addr, seeds).await?;
    await_all_up().await?;

    let system = ActorSystem::new(DeviceManager);
    cluster::register(&Key::new(MANAGER_KEY), system.root())
        .context("registering the device manager")?;
    cluster::leave_on_terminated(system)
        .await
        .context("leaving the cluster")?;
    Ok(())
}

leave_on_terminated waits until the actor system has terminated and then leaves. The node announces its departure to every member and tries to drain what it still has queued for them, within its leave_timeout. A member acknowledges once those frames have reached its delivery path, a mailbox or a logged dead letter; that does not prove an actor has handled them. The call returns a Departure, which lists the members that acknowledged and those that did not, for example because their connection ended first or the timeout passed. tellus logs a warning for every member that did not acknowledge. Leaving is final, like being downed: the process exits, and only a restart joins again, as a new node.

Watching a departure

async fn await_left(addr: SocketAddr) -> anyhow::Result<()> {
    let mut state = cluster::cluster_state().context("subscribing to the cluster state")?;
    let left = |state: &ClusterState| {
        state
            .members()
            .iter()
            .any(|member| member.addr() == addr && member.state() == MemberState::Left)
    };
    timeout(LEAVE_TIMEOUT, state.wait_for(left))
        .await
        .context("the node did not leave within the timeout")?
        .context("the cluster state stopped")?;
    Ok(())
}

await_left waits until this node lists the given member as left.

The first node

async fn first_node() -> anyhow::Result<()> {
    let seeds = reserved_addrs(NODES)?;
    let mut manager_nodes = seeds[1..]
        .iter()
        .map(|addr| spawn_node(MANAGER_NODE_ARG, *addr, &seeds))
        .collect::<anyhow::Result<Vec<_>>>()?;
    join_cluster(seeds[0], seeds.clone()).await?;
    await_all_up().await?;
    find_managers(2).await?;

    let manager = receptionist::lookup_at(&Key::new(MANAGER_KEY), seeds[1])
        .context("looking up the device manager on the second node")?
        .into_iter()
        .next()
        .context("no device manager on the second node")?;
    let watcher = ActorSystem::new(Watcher {
        manager: manager.clone(),
    });
    for (device_name, degrees) in [("left", 21.5), ("right", 19.1)] {
        manager.tell(Command::Record {
            group_name: "attic".to_string(),
            device_name: device_name.to_string(),
            degrees,
        });
    }
    print_attic_average(&manager).await?;
    manager.tell(Command::Stop);
    timeout(LEAVE_TIMEOUT, watcher.terminated())
        .await
        .context("no terminated signal within the timeout")?
        .context("awaiting the watcher's termination")?;
    await_left(seeds[1]).await?;
    println!("State of the second node: {}", member_state(seeds[1])?);
    let status = manager_nodes[0]
        .wait()
        .await
        .context("awaiting the second node's exit")?;
    anyhow::ensure!(status.success(), "the second node failed: {status}");

    let manager = find_managers(1).await?.remove(0);
    print_attic_average(&manager).await
}

The first node spawns the watcher before it records the readings, so the watch is on its way long before anything stops. Instead of killing the second node, it then tells the device manager to stop. It waits for the watcher, then for the second node to be listed as left, and for its process to exit.

Run it

From a clone of the tellus repository:

cargo run --quiet -p tellus --features cluster-dev --example clustered_05_leave

It prints:

Average temperature in attic: 20.30
The device manager has terminated
State of the second node: left
No readings in attic

The second node is listed as left, not as down, without any failure detection: that is why this step needs no faster configuration than the defaults.

A real termination

This time the terminated signal is normally real. The device manager's node sends it over the wire as part of the device manager's termination, so it arrives behind every message the device manager sent the watcher, exactly as within one process. The watcher gets a synthesized signal instead, as in the previous step, if the device manager is already gone when the watch arrives, or if the first node cannot finish reading what the second node sent before its leave timeout. The API does not tell the two apart, and tellus offers no way to wait until a remote watch is in place.

What you learned

A node that stops on purpose leaves the cluster: leave_on_terminated waits for its actors to terminate, announces the departure and drains what is queued. The others list the node as left as soon as they receive the departure, and its actors' terminated signals normally arrive as real terminations.

Where to go from here

You have spread the device manager over a cluster: the nodes bootstrap from a list of seeds, messages and references cross the wire, the receptionist finds actors by key, and the cluster handles both ways out of it, a crash and a departure.

For production, the how-to guides show how to run on Kubernetes, shut down gracefully, observe a cluster and secure it with mutual TLS. Membership and downing explains how the nodes agree on who belongs to the cluster, and Message delivery and the death watch ordering guarantee what holds across nodes.