Clustering

Remoting lets two systems that already know about each other exchange messages. Clustering answers the question remoting deliberately does not: who is in this group right now.

"""Three nodes finding each other, agreeing who is in, and one of them leaving.

Concepts: `Cluster`, seed nodes, the statuses a member moves through, the
leader as something every node works out rather than votes on, and a graceful
leave that every node ends up agreeing about.

Remoting (`two_nodes`) lets two systems that already know about each other
exchange messages. Clustering answers the question remoting does not: who is
in this group right now. Nothing here is a vote. Every node merges what it
hears into what it believes, the merge is written so that the order things
arrive in cannot change the result, and the leader is the first member in
address order, which each node computes for itself from the same converged
view.

Note the shape of `join_seed_nodes`: every node is given the same list in the
same order, its own address included. The first seed in that list, and only
the first, may form a new cluster, and only if it hears from nobody. That one
rule is what stops a restarting node from founding a second cluster beside the
one that is already running.

The three systems run in this one process on loopback ports the OS picks, so
the example needs no orchestration and no second machine.

What to watch in the output: the last two lines. The node that leaves is
`node1`, which is both the first seed and the leader, and the other two end up
agreeing it is `removed`. They report that from their own view rather than by
asking anybody, and leadership moved to `node2` on the way without a handover,
because the leader is a function of the membership rather than a post somebody
holds.

They agree in the end rather than at once. `leave` returns when the leaving
node is `removed` in its own view, and the others hear about it on their next
gossip round, so the example waits for each of them instead of reading them
straight away. That wait is the honest shape for anything that watches another
node, and leaving it out is a race that shows up about once in fifteen runs.

Run it with `uv run python -m tapio_examples.cluster_join`.
"""

import asyncio
from datetime import timedelta

from tapio import ActorSystem, RemoteSettings, TapioSettings
from tapio.cluster import Cluster, MemberStatus
from tapio.settings import ClusterSettings

__all__ = ["main"]


def node() -> TapioSettings:
    """Settings for one node: remoting on, on a loopback port the OS picks."""
    return TapioSettings(remote=RemoteSettings(bind_port=0))


def gossiping() -> ClusterSettings:
    """Gossip often enough that an example finishes while you watch it.

    A real deployment leaves these alone. The defaults gossip once a second,
    which is the right rate for a cluster that will be running for weeks.
    """
    return ClusterSettings(
        gossip_interval=timedelta(milliseconds=50),
        join_retry_interval=timedelta(milliseconds=50),
        seed_form_after=timedelta(milliseconds=200),
    )


async def until_removed(cluster: Cluster, address: str) -> MemberStatus:
    """Wait for one node to see another written off.

    `leave` returns as soon as the leaving node is `removed` in its own view.
    Every other node finds out on its next gossip round, so anybody reading
    another node's view waits for it rather than assuming. Reading straight
    after `leave` returns catches a node still at `exiting` often enough to
    matter, which is a race in the reader rather than in the cluster.

    Args:
        cluster: The node doing the watching.
        address: The member it is waiting to see written off.

    Returns:
        The status it settled on, which is `removed`.

    Raises:
        TimeoutError: If the removal has not reached this node in five
            seconds. Gossip here runs every 50ms, so that is long enough to
            mean something is wrong rather than slow.
    """
    async with asyncio.timeout(5.0):
        while True:
            member = cluster.state.member(address)
            if member is not None and member.status is MemberStatus.REMOVED:
                return member.status
            await asyncio.sleep(0.005)


async def main() -> list[str]:
    """Run the example.

    Returns:
        The lines the nodes produced, in the order they produced them.
    """
    lines: list[str] = []

    async with (
        ActorSystem("node1", node()) as first,
        ActorSystem("node2", node()) as second,
        ActorSystem("node3", node()) as third,
    ):
        systems = (first, second, third)
        clusters = [Cluster(system, gossiping()) for system in systems]
        # In a real deployment this list comes from configuration, and it is
        # the same list on every node. Here the addresses are read from the
        # systems, because all three are in this process.
        seeds = [cluster.address for cluster in clusters]

        await asyncio.gather(*(cluster.join_seed_nodes(seeds) for cluster in clusters))

        for system, cluster in zip(systems, clusters, strict=True):
            member = cluster.self_member
            assert member is not None
            lines.append(
                f"{system.name}: {member.status}, member {member.up_number} "
                f"of {len(cluster.members)}"
            )

        leader = {cluster.leader for cluster in clusters}
        lines.append(f"every node agrees the leader is {leader.pop()}")

        # Leaving is not vanishing. The member walks out through Leaving and
        # Exiting, each step waiting for a view every node has seen, which is
        # where a handoff would happen once there is something to hand over.
        leaving, *staying = clusters
        before = leaving.self_member
        assert before is not None
        lines.append(f"node1: leaving, at {before.status}")
        await leaving.leave()

        for system, cluster in zip(systems[1:], staying, strict=True):
            status = await until_removed(cluster, leaving.address)
            lines.append(f"{system.name}: node1 is {status}")

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

Nothing here is a vote. Every node merges what it hears into what it believes, the merge is written so that the order gossip arrives in cannot change the result, and the leader is the first member in address order, which every node works out for itself. There is no consensus algorithm in tapio and there is not going to be one.

What is in this release

Membership: joining, converging, leading, and leaving. Reachability: every member is watched by a few others, and a member that stops answering is reported unreachable by the nodes watching it. Downing: with a strategy configured, the losing side of a partition is written off rather than blocking the leader for ever, and the losing side downs itself. And the user-facing surface an application reacts to membership through: cluster events delivered to an actor's mailbox, roles, a cluster singleton with handoff, and a group router over the members of a role.

The surface is deliberately small. Clustering earns its keep only when an application reacts to membership, and reacting is easiest when a change is just another message: an event on a mailbox, handled by behaviour switching and supervision like everything else, rather than a callback on some other thread.

Statuses

A member moves up this ladder and never back down it.

Status	Meaning	Set by
joining	contacted a seed, not yet accepted	the joining node
up	a full member	the leader, on convergence
leaving	a graceful exit was asked for	the leaving node
exiting	leaving, and every node has seen it	the leader, on convergence
down	declared dead, and may not return	the leader
removed	gone, and kept as a tombstone	the leader

The order is the merge rule. When two nodes hold different views of one member, the merged view takes the higher status, so a node that learns of a down can never un-learn it however old the gossip that told it. It is also why Member.with_status refuses to move a member backwards: a transition that contradicted the lattice would be undone by the next gossip that arrived.

Akka has a WeaklyUp status between joining and up, which lets a node join while another member is unreachable. tapio does not, deliberately. It buys availability during a partition at the cost of a member that only half the cluster has agreed on, and every feature that places something has to know not to place it there. It is worth adding when somebody has the problem it solves.

Joining

Every node is given the same seed list, in the same order, its own address included:

await cluster.join_seed_nodes(
    [
        "tapio://orders@10.0.0.1:2551",
        "tapio://orders@10.0.0.2:2551",
        "tapio://orders@10.0.0.3:2551",
    ]
)

Every address in that list has to name a host and a port, and so does every address that arrives in a cluster message. tapio://orders on its own parses, since that is how a system with remoting switched off writes its own refs down, but nobody can dial it, and a cluster reaches its members by dialling them. A seed like that is refused where the list is passed, and one that arrives on a socket is refused as a malformed frame.

A node asks every seed to let it in, and keeps asking until it sees itself in the gossip that comes back, because a join is delivered at most once like every other message in tapio. A node that is not a member itself ignores a join request, which is what stops two nodes that started together from admitting each other into two different clusters.

Only the first seed in the list may form a new cluster, and only after seed_form_after in which it has heard from nobody at all. That rule is the whole of the bootstrap safety. A restarted node hears the running cluster's gossip in answer to its own join and joins that instead of founding a second one, so seed_form_after has to stay comfortably longer than the time a running seed takes to answer.

join_seed_nodes returns once this node is up. If it times out, the node is still asking: the error is about how long you were prepared to wait, not about the node giving up. A node that has left, or been downed, cannot join again as itself, and join_seed_nodes raises ClusterError for it. Restart the system to join as a new incarnation.

Convergence, and the leader

A view has converged when every member that is not down or removed is reachable and has seen that exact version of the gossip. Convergence is not consensus. It is the condition under which the leader is allowed to act, and that is all it is used for.

The leader is the first member in address order, among those no live member reports unreachable, whose status is up or leaving. If there is none, for example before anybody is up or while every up member is unreachable, it is the first such member of any live status, joining and exiting included: somebody has to be able to accept the first join. A leader acts only on a converged view, so a fallback leader during a split does nothing until the split is resolved. Every node computes the leader, one node acts on it, and there is no handover protocol because there is no state to hand over.

A leader with an unreachable member converges on nothing and therefore does nothing. Joins wait, leaves wait, and the cluster keeps running. That blocking is what a downing strategy resolves; without one configured, it is where the cluster stays until the member answers again or an operator downs it.

Leaving

await cluster.leave()

The member walks out rather than vanishing: leaving, then exiting once every node has seen it, then removed. Each step needs a converged view, so leaving takes as long as agreement takes, and every node ends up holding the same tombstone. A cluster singleton on the leaving member lets go the moment it hears of its own leave, and no successor starts until the member is removed. Wherever the leave was asked for, the two instances do not normally overlap. An instance still busy in a handler when its host lets go can overlap the successor until that handler returns.

The tombstone is kept rather than pruned. Dropping the record would let a peer holding an older view put the member back, since merging two views unions the members in them. Pruning needs a way to know that every node has seen the removal, and it is not in this release.

Leaving does not terminate the system. Ending the process is the application's decision.

A leave is asked for through leave on the node itself, or through a node's management port. It is never accepted from across a link. Any system that completes a remoting handshake can address a node's cluster daemon, and a leave it could send would bypass the token or certificate the management port asks for.

What a node gossips

One node picks one other member per round and sends it everything it believes. The receiver merges, and answers immediately if its own view turns out to be newer, which is most of why convergence takes rounds rather than seconds. Traffic is therefore linear in the number of nodes, not quadratic.

The state itself is small and every part of it merges the same way:

• Members, merged pairwise by the status lattice above.
• Reachability, one observation per pair of nodes, each carrying the observer's own version so that an unreachability can be retracted by the node that reported it.
• A vector clock, merged by taking the higher count per node, which is how two views are ordered against each other when one is simply newer.
• A seen set, which is what makes convergence observable.

Reachability

Reachability is a separate axis from membership, and conflating the two is the classic mistake. A member can be up and unreachable at the same time. The first is a decision the cluster made about it, the second is an observation one node made about it, and only the second can be wrong.

Every node sorts the member addresses, finds itself, and watches the few that follow it, wrapping round at the end:

cluster = Cluster(system, ClusterSettings(monitored_peers=5))
watched = cluster.monitored  # the members this node watches, in order

So every member is watched by exactly that many others, whether or not anybody has reason to send it anything, and the probing costs one message per watched member per round rather than one per pair. All-to-all monitoring is quadratic, and it is what makes naive implementations fall over at a few dozen nodes.

An unreachable member is watched but not counted. The ring reaches past it to the next member, so a node always watches monitored_peers members it can count on, plus every unreachable member between them.

A watcher sends a heartbeat every heartbeat_interval, the watched member answers, and a member that has not answered for unreachable_after is recorded unreachable by that node. Nothing about that is a decision by the cluster. It is one node saying it cannot get through, it travels in gossip like everything else, and the node that said it is the only one that can take it back, which it does the moment an answer arrives again.

A heartbeat says where to send the answer, and a node answers only an address it has a reason to believe: a member of the view it holds, or a peer it already has a link to. A real watcher always has the second, because its heartbeat came over that link, so a node that is ahead on membership is still answered and does not look dead for the round or two it takes this one to catch up. What the rule refuses is the invented address. Answering by dialling whatever a message names would let any peer that has finished a handshake make this node open a connection to any host and port, as often as it cared to ask.

A member is unreachable to the cluster when any node says so, and reachable again only when every one of them has retracted. That is deliberately pessimistic. One node's bad link is enough to block convergence, which is visible and recoverable, whereas ignoring a minority report is how a half-partitioned node stays a member forever.

The transport's verdict counts as evidence too. When remoting gives up on a link it says so on the event stream, and a watcher takes that as its member going unreachable, because it arrives sooner than a window that has not run out yet. The two sources are retracted separately: an answer to a probe brings back the first, and a link coming up again brings back the second. An answer never retracts what the transport said, since a peer that remoting is refusing to carry frames to cannot answer at all.

The transport's verdict is recorded only for a member on this node's ring. For any other member it is remembered, and the member starts from it if the ring moves onto it. Every record stays in the gossip for good, so recording the transport for every member a node ever had a link to grew with the square of the cluster after one partition, and at a few hundred nodes the gossip no longer fitted in a frame.

The ring reaching past unreachable members is what a downing strategy's safety rests on. A strategy is safe because the two sides of a partition feed it mirror-image views: each side sees the whole of the other as unreachable and names the same losing side without a message crossing the split. Take a node on one side and the run of far-side members that follows it on the ring. None of that run counts, so the node watches all of it, however long it is. Every far-side member is therefore watched by the nearest node on this side before it, and this side sees the whole of the far side as unreachable. A long run is covered a step at a time, one window per step for a member nobody had a link to, and a node does not decide until its whole side has seen the view.

A record names its observer by address, not by incarnation. A node that restarts at the same address therefore finds its predecessor's claims in the gossip, counting again now that its address is live. It takes back every claim under its address that it did not make itself, because only the observer can retract a claim and it is the observer at that address now.

The two views are mirror images only up to the membership changes that were in flight when the split came. A member admitted moments before may be known on one side only, and a member the leader has just removed may still be exiting on the other. The counting strategies allow for that. A side counts itself over the up and leaving members the leader has accepted, and counts the far side over every live member, so a change that crossed to one side cannot make both sides the winner. A split during a join or a leave can therefore down both sides, which is the safe direction. A down made on one side only is the case no count covers: an operator downing a member while the cluster is split can make the two views disagree about who is left.

A node also waits, before it decides, until every member it can still hear has seen its current view. A far-side member that nobody on this side observes looks reachable from here, but it cannot have seen anything this side has gossiped since the split. Waiting for it keeps a node from counting it on this side. A strategy that raises is logged and asked again on the next turn, and LeaseMajority counts a lease that does not answer within its acquire_timeout, or that fails, as lost.

A link coming up is not an answer either. It proves a process is accepting connections, and what this node is asking is whether the daemon behind it is still replying, so the next probe settles that one round later. Reading a handshake as a reply would let a peer whose links churn faster than unreachable_after look healthy forever without ever answering, which is the failure the watching exists to catch.

Silence is judged behind an interface. The default is a fixed window, which has no opinion about how variable a network is, so unreachable_after has to sit well above heartbeat_interval or a slow moment reads as a dead node. Setting phi_accrual switches to a phi-accrual detector, which learns the spread of a peer's timings instead of being told a number and suspects it on a scale that means the same confidence whether the link is fast and steady or slow and jittery. It reads the same interface, so nothing else about the monitor changes, and phi_threshold and phi_acceptable_pause tune it.

One rule this contradicts

Remoting says recovery from a quarantine is never automatic: a system that gave up on a peer stays given up on until somebody calls reconnect. A clustered system does not follow that rule for its own members. Each round, a node clears the quarantine on every alive member, so nothing it might have to talk to is left refused.

The rule is right for a single system and wrong here, and the difference is membership. One node alone cannot tell a false alarm from a dead peer, so it refuses to guess twice. A cluster does not have to guess: an unreachable member has not been downed, so it is still a member, and the cluster's job is to keep trying to reach it until a downing strategy says to stop. Only members are forgiven, and only while nobody has decided otherwise: a peer this system talks to but has not clustered with follows remoting's rule as before.

It covers every member and not only the ones this node watches, because the two sets are different jobs. The ring decides who is judged, and it is a few nodes per member so that heartbeat traffic stays linear. Gossip goes to any member at all. A node therefore has to be able to dial members it does not watch, and if only the watching node forgave a quarantine then every other pair would stay refused for good. In a cluster larger than monitored_peers that is most pairs, and a partition that healed would never converge again.

Clearing a quarantine dials nothing. It says only that this node is willing to be associated again, so doing it every round for a member that is perfectly reachable costs a set lookup and changes nothing.

Cluster events

Membership changes reach an application as messages, delivered to an ordinary actor's mailbox. An actor subscribes, and from then on it is told what changed:

cluster.subscribe(worker, MemberUp, MemberRemoved, UnreachableMember)

There are seven events, all carrying the member they are about: MemberUp, MemberLeaving, MemberRemoved, UnreachableMember, ReachableMember, LeaderChanged, and SelfDown. Subscribing to none of them in particular means all of them. The subscriber has to accept the events it asks for as part of its declared message type, because they arrive on its own mailbox and are type-checked like anything else. A subscriber that cannot take an event it asked for is dropped, with a warning in the log. An actor that does not want the events in its own type can subscribe an adapter from message_adapter instead.

A subscriber hears the current membership the moment it subscribes, as the events that would have carried it: an actor that starts after the cluster has formed is told who is up before it is told what changes next. So there is no separate "give me a snapshot" call, and no window in which a late subscriber has missed something it can never learn.

An event is this node's view, not the truth. It is emitted when this node's own membership state moves, so two nodes may see the same change a gossip round apart. That is the same guarantee everything else in clustering gives.

A subscriber that stops is forgotten, because the daemon watches it. For an adapter, the daemon watches the actor that owns it. A ref with nothing to watch, such as a dead-letter ref, is refused with a warning. So is a ref to an actor on another node: events are not registered on the wire, so a subscriber must be a local actor, and each node tells its own subscribers. Call unsubscribe only for an actor that wants to keep running and stop listening.

Roles

A role is what a node says it is for, fixed when it joins and part of what the cluster agreed on when it accepted the node:

cluster = Cluster(system, ClusterSettings(roles=frozenset({"worker"})))
workers = cluster.members_with_role("worker")  # oldest first

Every member carries its roles in the gossip, so every node can filter on them without asking anyone. The two features below are the filtering: a singleton runs on the oldest member of a role, and a group router spreads work over the members of one.

Cluster singletons

Some work has to happen in exactly one place: a scheduler that must not fire twice, a coordinator that owns a piece of state. A ClusterSingleton places one such actor and moves it when its host goes away.

"""One actor across a whole cluster, moved to a new node when its host leaves.

Concepts: `ClusterSingleton`, the oldest member of a role as where a singleton
runs, and the handoff that happens when that member goes away.

Some work has to happen in exactly one place: a scheduler that must not fire
twice, a coordinator that owns a piece of state. Spawn the same
`ClusterSingleton` manager on every node, and the one on the oldest member runs
the instance while every other manager waits. "Oldest" is the member the leader
accepted first, an order every node computes the same way, so there is no
election and no lock, only a function of the membership every node already
agrees on.

When the host leaves, the next oldest member starts the instance. It is a fresh
start, not a move of live state: what mattered on the old host does not cross to
the new one, which is the honest shape of a thing that has to survive its host
going away. Here `node1` is both the first seed and the oldest member, so the
coordinator starts there; when `node1` leaves, it reappears on `node2`.

The three systems run in this one process on loopback ports the OS picks, so the
example needs no orchestration and no second machine.

What to watch in the output: the coordinator names the node it is running on
when it starts. It says `node1` first, and after `node1` leaves it says `node2`,
without anything having told the second manager to take over.

Run it with `uv run python -m tapio_examples.cluster_singleton`.
"""

import asyncio
from datetime import timedelta

from tapio import (
    ActorSystem,
    Behavior,
    Behaviors,
    Message,
    RemoteSettings,
    TapioSettings,
)
from tapio.actor import ActorContext, Signal
from tapio.actor.signals import PostStop
from tapio.cluster import Cluster, ClusterSingleton
from tapio.settings import ClusterSettings

__all__ = ["Tick", "main"]


class Tick(Message):
    """A message the coordinator does not need to act on, only to exist for."""


class Coordinators:
    """Where every coordinator instance reports which node it runs on."""

    def __init__(self) -> None:
        """Start with nobody running the coordinator."""
        self.running: set[str] = set()

    def started(self, node: str) -> None:
        """Record that the coordinator started on a node."""
        self.running.add(node)

    def stopped(self, node: str) -> None:
        """Record that the coordinator on a node stopped."""
        self.running.discard(node)


def coordinator(registry: Coordinators, node: str) -> Behavior[Tick]:
    """The singleton instance: it reports its life to the shared registry.

    Args:
        registry: Where the instance says which node it runs on.
        node: The name of the node it is running on.

    Returns:
        The behavior.
    """

    def build(ctx: ActorContext[Tick]) -> Behavior[Tick]:
        registry.started(node)

        async def on_message(message: Tick) -> Behavior[Tick]:
            return Behaviors.same()

        async def on_signal(ctx: ActorContext[Tick], signal: Signal) -> Behavior[Tick]:
            if isinstance(signal, PostStop):
                registry.stopped(node)
            return Behaviors.same()

        return Behaviors.receive_message(on_message, msg_type=Tick, on_signal=on_signal)

    return Behaviors.setup(build)


def node() -> TapioSettings:
    """Settings for one node: remoting on, on a loopback port the OS picks."""
    return TapioSettings(remote=RemoteSettings(bind_port=0))


def gossiping() -> ClusterSettings:
    """Gossip often enough that the example finishes while you watch it."""
    return ClusterSettings(
        gossip_interval=timedelta(milliseconds=50),
        join_retry_interval=timedelta(milliseconds=50),
        seed_form_after=timedelta(milliseconds=200),
    )


async def until_one_on(registry: Coordinators, *, not_on: str | None = None) -> str:
    """Wait until the coordinator runs on exactly one node, and say which.

    Args:
        registry: Where the coordinators report where they run.
        not_on: A node the coordinator must have moved off, when waiting for a
            handoff rather than the first placement.

    Returns:
        The node the coordinator now runs on.

    Raises:
        TimeoutError: If it has not settled within five seconds.
    """
    async with asyncio.timeout(5.0):
        while True:
            running = set(registry.running)
            if len(running) == 1 and not_on not in running:
                return next(iter(running))
            await asyncio.sleep(0.005)


async def main() -> list[str]:
    """Run the example.

    Returns:
        The lines the nodes produced, in the order they produced them.
    """
    lines: list[str] = []
    registry = Coordinators()

    async with (
        ActorSystem("node1", node()) as first,
        ActorSystem("node2", node()) as second,
        ActorSystem("node3", node()) as third,
    ):
        systems = (first, second, third)
        clusters = [Cluster(system, gossiping()) for system in systems]
        seeds = [cluster.address for cluster in clusters]

        await asyncio.gather(*(cluster.join_seed_nodes(seeds) for cluster in clusters))

        # The same manager on every node. Only the one on the oldest member
        # runs the instance; the rest wait for a handoff that may never come.
        for system in systems:
            system.spawn(
                ClusterSingleton(
                    coordinator(registry, system.name), name="coordinator"
                ),
                name="singleton",
            )

        first_host = await until_one_on(registry)
        lines.append(f"the coordinator runs on {first_host}, the oldest member")

        # The oldest member leaves. Its instance stops with it, and the next
        # oldest starts one, without anything telling it to.
        await clusters[0].leave()
        second_host = await until_one_on(registry, not_on="node1")
        lines.append(f"node1 left, so the coordinator moved to {second_host}")

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

Spawn the same manager on every node. Each watches membership, and the manager on the oldest member of the singleton's role, and only that one, runs the instance. "Oldest" is the member with the lowest up_number, the order the leader accepted members in, which every node computes the same way from the same gossip. So at a converged view exactly one manager runs the instance, with no election and no lock.

Handoff is triggered by a host going away, and a successor starts only once the old host is removed. A crash is only ever seen as removal: every manager hears MemberRemoved, recomputes the oldest, and the new oldest starts the instance.

A graceful leave is seen earlier, as MemberLeaving. The leaving host lets its instance go when it hears that about itself. Every other manager goes on counting the leaving host as the oldest, so no successor starts before the removal. The removal needs every member, the leaving host included, to have seen the leave, so by then the host has asked its instance to stop. That holds wherever the leave was asked for: on the host, or with tapio-cluster leave pointed at any other node. The host does not wait for its instance to finish stopping, though. An instance still busy in a handler when the successor starts overlaps it until that handler returns.

A host on the losing side of a split is downed rather than removed, and it never sees its own removal. Its manager lets the instance go on SelfDown and never hosts again. The majority starts a successor once it has removed the old host, which can be before the downed side has noticed, so two instances can overlap for up to the difference between the two sides' down_after. Set terminate_on_down or act on when_downed to end the downed process as well.

The instance is a fresh start wherever it runs, not a move of live state. What mattered on the old host does not cross to the new one, which is the honest shape of a thing that has to survive its host going away. If it owns state that has to outlive a node, that state belongs somewhere the next host can read it, not in the actor's memory.

Group routers

A pool router owns its routees, spawning them as its own children. A group router owns none of them. It routes to whatever actor each member of a role publishes at an agreed path, and the pool follows membership:

def front_desk() -> Behavior[Job]:
    """Pass every job to a group router over the `worker` members."""

    def start(ctx: ActorContext[Job]) -> Behavior[Job]:
        proxy = ctx.spawn(
            Routers.group(Job, role="worker", path="/user/worker"), "workers"
        )

        async def on_job(job: Job) -> Behavior[Job]:
            proxy.tell(job)
            return Behaviors.same()

        return Behaviors.receive_message(on_job, msg_type=Job)

    return Behaviors.setup(start)

A member that joins is added, and one that is removed or goes unreachable is dropped within a convergence. An empty group is not the end of the router, the way an empty pool is: members come and go, so instead of stopping it holds and dead-letters what it is handed until a routee appears.

The routee on each node is reached by its bare path, the way the cluster daemon itself is, so it has to be published as a well-known name there with system.refs.register_well_known(ref). That is the one piece of setup a group router needs beyond a pool, and it is what lets a router on any node address the routee without knowing which incarnation is answering over there. Selection reuses the same RoutingStrategy as a pool, so round-robin and anything written for a pool works here without a change.

The message type is named rather than read off a routee, because the routees are on other nodes and there is no spawned child here to read it from. That is the only difference in how the two routers are built.

Managing a cluster

Reading a cluster's membership and downing a stuck member are things an operator does from outside the application, so they happen over a small HTTP port a node opens rather than through a code path the application has to carry. A node given ManagementSettings answers a few requests: one reads what it believes, and the others ask it to let a member leave or to down one.

"""An operator reading a cluster and downing a member, over the management port.

Concepts: `ManagementSettings`, the small HTTP surface a node opens for an
operator, and the `tapio-cluster` command that speaks to it. A node given
management settings answers a few HTTP requests: one reads what it believes
about the cluster, and the others ask it to let a member leave or to down one.

The point of the port is that it is out of band. Reading membership and downing
a stuck member are things an operator does from outside the application, without
a code path in the application for either. Here the requests are made with the
standard library so the example needs nothing extra, but they are the same
requests the `tapio-cluster` command makes:

```bash
tapio-cluster --port 25530 status
tapio-cluster --port 25530 down tapio://cluster@127.0.0.1:...

Downing is ordinarily a strategy's decision about which side of a split lives. This is the operator's version of the same move, for a member no strategy will reach: it goes to down exactly as a strategy would put it there, and the decision travels to every node as gossip. The node the operator asked answers the moment it has been asked, not once the member has gone, which is why the example waits for the membership to settle rather than reading it straight away.

Three systems run in this one process on loopback ports the OS picks, so the example needs no orchestration and no second machine.

Run it with uv run python -m tapio_examples.cluster_management. """

import asyncio import json from datetime import timedelta

from tapio import ActorSystem, RemoteSettings, TapioSettings from tapio.cluster import Cluster, MemberStatus from tapio.settings import ClusterSettings, ManagementSettings

all = ["main"]

def node() -> TapioSettings: """Settings for one node: remoting on, on a loopback port the OS picks.""" return TapioSettings(remote=RemoteSettings(bind_port=0))

def gossiping() -> ClusterSettings: """Gossip often enough that an example finishes while you watch it.""" return ClusterSettings( gossip_interval=timedelta(milliseconds=50), join_retry_interval=timedelta(milliseconds=50), seed_form_after=timedelta(milliseconds=200), )

async def request( port: int, method: str, path: str, body: dict[str, str] | None = None ) -> dict[str, object]: """Make one HTTP request to a management port and read its JSON answer.

The management surface is plain HTTP, so this is a raw request written by
hand rather than a client library, to show there is nothing more to it.

Args:
    port: The management port to reach.
    method: The HTTP method.
    path: The path to request.
    body: The JSON body to send, or `None` for a request with no body.

Returns:
    The parsed JSON answer.
"""
reader, writer = await asyncio.open_connection("127.0.0.1", port)
lines = [f"{method} {path} HTTP/1.1", "Host: 127.0.0.1"]
payload = b""
if body is not None:
    payload = json.dumps(body).encode("utf-8")
    lines.append("Content-Type: application/json")
    lines.append(f"Content-Length: {len(payload)}")
lines.append("Connection: close")
writer.write(("\r\n".join(lines) + "\r\n\r\n").encode("latin-1") + payload)
await writer.drain()
raw = await reader.read()
writer.close()
await writer.wait_closed()
_, _, tail = raw.partition(b"\r\n\r\n")
parsed: dict[str, object] = json.loads(tail) if tail else {}
return parsed

async def until_gone(cluster: Cluster, address: str) -> None: """Wait until a member has left a node's live membership.

A down travels as gossip, so a node the operator did not ask learns about
it a round later. Reading straight away would catch it before the news
arrived, which is a race in the reader rather than in the cluster.

Args:
    cluster: The node doing the watching.
    address: The member it is waiting to see leave.

Raises:
    TimeoutError: If the member is still listed after five seconds, which
        with gossip every 50ms means something is wrong rather than slow.
"""
async with asyncio.timeout(5.0):
    while True:
        if all(member.address != address for member in cluster.members):
            return
        await asyncio.sleep(0.005)

async def main() -> list[str]: """Run the example.

Returns:
    The lines the operator's view produced, in order.
"""
lines: list[str] = []

async with (
    ActorSystem("node1", node()) as first,
    ActorSystem("node2", node()) as second,
    ActorSystem("node3", node()) as third,
):
    systems = (first, second, third)
    # Only the first node opens a management port. An operator reaches the
    # cluster through any one node, since every node holds the whole view.
    clusters = [
        Cluster(
            system,
            gossiping(),
            management=ManagementSettings(bind_port=0) if system is first else None,
        )
        for system in systems
    ]
    seeds = [cluster.address for cluster in clusters]
    await asyncio.gather(*(cluster.join_seed_nodes(seeds) for cluster in clusters))

    address = clusters[0].management_address
    assert address is not None
    port = int(address.rsplit(":", 1)[1])

    status = await request(port, "GET", "/status")
    members = status["members"]
    assert isinstance(members, list)
    lines.append(f"operator sees {len(members)} members, leader {status['leader']}")

    # Down node3 through node1's port. The operator never touches node3.
    victim = clusters[2].address
    answer = await request(port, "POST", "/down", {"address": victim})
    lines.append(f"asked node1 to {answer['accepted']} node3")

    # The decision reaches node2, which the operator did not ask, as gossip.
    await until_gone(clusters[1], victim)
    remaining = await request(port, "GET", "/status")
    members = remaining["members"]
    assert isinstance(members, list)
    alive = [m["address"] for m in members if m["status"] != MemberStatus.DOWN]
    lines.append(f"after downing node3, {len(alive)} members remain live")

for line in lines:
    print(line)
return lines

if name == "main": asyncio.run(main())

The `tapio-cluster` command is the client, and the same requests are a `curl`
away for anyone who would rather script them. It ships in the `cli` extra
rather than in the library, so a service that embeds tapio does not carry the
command's own dependencies to run the runtime:

```bash
uv add 'tapio-py[cli]'

tapio-cluster --port 25530 status
tapio-cluster --port 25530 leave tapio://orders@10.0.0.2:2551
tapio-cluster --port 25530 down  tapio://orders@10.0.0.3:2551

An operator reaches the cluster through any one node, since every node holds the whole view. What a status reports is that node's view, which is the truth once the cluster has converged and its best guess until then, the same caveat every gossip-based answer carries. A leave or a down is answered the moment the node has been asked, not once the member has gone: the decision travels as gossip like every other one, so the node replies 202 Accepted and the next status shows it taking effect.

Downing over this port is the operator's version of the move a downing strategy makes on its own. It is for a member no strategy will reach: one that is unreachable to everyone, so no split fires a strategy, or a cluster running with no strategy configured at all. The member goes to down exactly as a strategy would put it there, and a down cannot be taken back. The downed member hears the decision as gossip, and when_downed returns on it. It shuts itself down if it was clustered with terminate_on_down, which works with no strategy configured. In a small cluster the downed member often hears of its downing only after the leader has removed it, since nobody gossips to a downed member. That counts as being downed too.

The port can down a member, so it is a serious surface, and it is off unless it is configured, like remoting. When it is on it binds loopback by default. Binding it anywhere another host can reach requires a token, the same refusal remoting makes about a secret:

Cluster(
    system,
    management=ManagementSettings(
        bind_host="0.0.0.0",
        token=SecretStr("..."),  # required beyond loopback, or it will not start
    ),
)

The token is presented as Authorization: Bearer <token> and compared in constant time. On loopback it is optional, since reaching the port at all already means being on the machine.

This port is not an API to put a crowd in front of. It expects loopback or a sidecar. It holds at most thirty-two connections at once and refuses the rest with a 503. The cap is there because this port answers on the same event loop the cluster daemon replies to heartbeats on. A flood that starves that loop stops the node answering probes, and the rest of the cluster then calls it unreachable. A cluster with a downing strategy removes such a node. See the security page for what the port expects of a deployment.

The port also speaks TLS, using the same TLSSettings remoting does:

Cluster(
    system,
    management=ManagementSettings(
        bind_host="0.0.0.0",
        tls=TLSSettings(
            certfile="server.pem", keyfile="server.key", cafile="ca.pem"
        ),
    ),
)

With certfile and keyfile the port answers HTTPS, and the command reaches it with --tls (or a --cafile that trusts the server's certificate). Add cafile on the node and the port also requires a client certificate signed by that authority, which authenticates the operator the way a token does. Mutual TLS is therefore the second way to satisfy the bind-beyond-loopback rule: a bearer token alone travels in a header, so a plaintext bind beyond loopback leaks it, while a token over TLS or a client certificate does not. The command presents its certificate with --client-cert and --client-key.

The certificate and key are loaded while the cluster is being constructed. A file that is missing or cannot be read makes Cluster(...) raise OSError and release the port, rather than leave a port bound that never answers. Over TLS, a connection refused by the cap is closed without a 503, so the command reports that it could not reach the node.

Addressing, and the one uid rule this bends

A cluster daemon publishes itself as a well-known name at /system/cluster, so it can be addressed by a bare path with no incarnation uid. That is the opposite of the rule refs normally follow, and it is opt-in for exactly one reason: a seed is named by an address in a configuration file, and a joining node has no way to know which incarnation is answering over there. Every other ref still carries its uid and still addresses one incarnation only.