When a node dies
Watch a device manager on another node, kill its node and turn to a second device manager.
Nodes crash. In this step two nodes run a device manager each, the first node watches the one it uses, and that one's node is killed. The cluster has to notice that the node is gone and agree to remove it, and the watch tells the first node when it has.
Why three nodes
When a node falls silent, the others cannot tell whether it crashed or whether only the network between them failed. So the cluster does not decide by who is silent, but by who is in the majority: the KeepMajority downing provider, the default, keeps the side that sees a strict majority of the members and removes the others. In a cluster of two, every failure is an even split. KeepMajority gives it to the lower address, so the lower node survives a failure of the higher one, but when the lower one is killed, the survivor removes itself. Two nodes cannot guarantee surviving either node's failure, so this step runs three: the first node and two device manager nodes. Whichever of them is killed, the other two are a majority. For the downing rules in full, see Membership and downing.
A majority only means something if all nodes count the same members. So before anything is killed, all three have to agree on the membership.
Agreeing on the members
async fn await_all_up() -> anyhow::Result<()> {
let mut state = cluster::cluster_state().context("subscribing to the cluster state")?;
timeout(
BOOTSTRAP_TIMEOUT,
state.wait_for(|state| state.up().count() == NODES),
)
.await
.context("not all nodes up within the timeout")?
.context("the cluster state stopped")?;
Ok(())
}await_all_up waits until this node lists all three members as up. A device manager node registers its device manager only after that:
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")?;
system
.terminated()
.await
.context("awaiting actor system termination")
}So once the first node sees both registrations and all three members up itself, all three nodes agree. find_managers waits for a given number of registrations, like find_manager of the previous step:
async fn find_managers(count: usize) -> anyhow::Result<Vec<ActorRef<Command>>> {
let mut managers =
receptionist::subscribe(&Key::new(MANAGER_KEY)).context("subscribing to the key")?;
let find = async {
loop {
let current = managers
.current()
.context("resolving the device managers")?;
if current.len() == count {
return anyhow::Ok(current);
}
managers.changed().await;
}
};
timeout(FIND_TIMEOUT, find)
.await
.context("not the expected device managers within the timeout")?
}Detecting a failure faster
async fn join_cluster(addr: SocketAddr, seeds: Vec<SocketAddr>) -> anyhow::Result<()> {
let transport = QuicTransport::dev(addr).context("dev QUIC transport")?;
let failure_deadline = Deadline::new(FAILURE_DEADLINE).context("failure deadline")?;
let mut config = EndpointConfig::new(addr);
config.failure_detector =
Arc::new(move || Box::new(DeadlineFailureDetector::new(failure_deadline)));
config.downing_provider = Arc::new(|| Box::new(KeepMajority::new(DOWNING_DEADLINE)));
cluster::start_endpoint(config, transport).context("remoting endpoint")?;
timeout(
BOOTSTRAP_TIMEOUT,
cluster::bootstrap(FixedSeeds::new(seeds), BootstrapConfig::new()),
)
.await
.context("no cluster within the timeout")?
.context("bootstrapping the cluster")
}By default, a phi accrual failure detector, which adapts to how regular the heartbeats are, decides when a node is unreachable, and KeepMajority decides once the unreachable members have been unreachable for ten seconds. That suits a production network, but makes for a long wait here. So this step configures a deadline failure detector, which counts a node as unreachable once its last heartbeat is three seconds old, and lets KeepMajority decide after two seconds.
Watching across nodes
struct Watcher {
manager: ActorRef<Command>,
}
impl Actor for Watcher {
type Message = ();
type State = ();
type Error = Infallible;
fn init(&self, context: &ActorContext<Self::Message>) -> Result<Self::State, Self::Error> {
context.watch(&self.manager);
Ok(())
}
fn receive(
&self,
_: &ActorContext<Self::Message>,
incoming: Incoming<Self::Message>,
_: Self::State,
) -> Result<Control<Self::State>, Self::Error> {
let Incoming::Terminated(_) = incoming else {
unreachable!("nothing tells the watcher a message")
};
println!("The device manager has terminated");
Ok(Control::Stop)
}
}The watcher watches the device manager on another node with the same context.watch as in the getting started tutorial. When it gets the terminated signal, it prints that and stops.
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")?;
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?;
let watcher = ActorSystem::new(Watcher {
manager: manager.clone(),
});
manager_nodes[0]
.kill()
.await
.context("killing the second node")?;
timeout(DOWNING_TIMEOUT, watcher.terminated())
.await
.context("no terminated signal within the timeout")?
.context("awaiting the watcher's termination")?;
println!("State of the second node: {}", member_state(seeds[1])?);
let manager = find_managers(1).await?.remove(0);
print_attic_average(&manager).await
}The first node records two readings with the device manager on the second node and asks for the average. Then it spawns the watcher and kills the second node. Once the watcher has stopped, the first node prints the second node's member state, finds the remaining device manager and asks it for the average again.
Run it
From a clone of the tellus repository:
cargo run --quiet -p tellus --features cluster-dev --example clustered_04_remote_watchIt prints:
Average temperature in attic: 20.30
The device manager has terminated
State of the second node: down
No readings in atticThe other device manager has no readings: they lived in the memory of the killed node. Making state survive its node is what the event-sourced actors are for.
A synthesized signal
A killed node cannot send a terminated signal, so tellus synthesizes it on the first node, once the cluster has downed the second node. That signal promises less than a real termination. It does not prove that the device manager has stopped: across a network partition it could still be running. It proves that no message from that device manager will ever be delivered on this node again, and that is what the first node needs to stop relying on it.
What you learned
A cluster that must survive a failure needs at least three nodes that agree on the membership. The failure detector and the downing provider turn a node's silence into its removal, and a watch on an actor of that node then fires with a synthesized signal. Next, a node leaves the cluster on its own instead of being killed.