Getting started
An actor system is created inside a coroutine, spawns actors, and is terminated
when you are done with it. Everything below is a live example module: the code
on this page is included from examples/, and CI runs it.
uv add tapio-py
Hello, world
Spawn an actor, send it a message, and let it reply to an address the message carried.
"""The smallest complete actor program: spawn, tell, reply, shut down.
Concepts: starting an `ActorSystem`, spawning a top-level actor, sending it a
message with `tell`, and carrying a return address in the message as an
`ActorRef` field.
There is no `ask` here. A reply is just another message, sent to a ref the
sender put in the request. Seeing that once makes `ask` read as the sugar it
is.
What to watch in the output: the greeter's line comes first, then the
listener's, because the reply is a second message and cannot overtake the
handler that sends it. Both actors stop when the system terminates.
Run it with `uv run python -m tapio_examples.hello_world`.
"""
import asyncio
from collections.abc import Callable
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef
__all__ = ["Greet", "Greeted", "main"]
class Greeted(Message):
"""Sent back to whoever asked for the greeting."""
whom: str
class Greet(Message):
"""A request to greet someone, carrying the address for the reply."""
whom: str
reply_to: ActorRef[Greeted]
def greeter(record: Callable[[str], None]) -> Behavior[Greet]:
"""Build the greeter: it greets, replies, and stays as it is.
Args:
record: Where to write the greeting, so the example can be asserted.
Returns:
The behavior to spawn.
"""
async def on_greet(ctx: ActorContext[Greet], message: Greet) -> Behavior[Greet]:
ctx.log.info("hello, %s!", message.whom)
record(f"greeter: hello, {message.whom}!")
message.reply_to.tell(Greeted(whom=message.whom))
return Behaviors.same()
return Behaviors.receive(on_greet)
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the actors produced, in the order they produced them.
"""
lines: list[str] = []
done = asyncio.get_running_loop().create_future()
async def on_greeted(message: Greeted) -> Behavior[Greeted]:
lines.append(f"listener: {message.whom} has been greeted")
# Handing a result back out to non-actor code. A future is the way to
# do that from inside a handler, and `ask` packages this pattern once
# the request/response shape is familiar.
done.set_result(None)
return Behaviors.same()
async with ActorSystem("hello") as system:
listener = system.spawn(Behaviors.receive_message(on_greeted), name="listener")
hello = system.spawn(greeter(lines.append), name="greeter")
hello.tell(Greet(whom="world", reply_to=listener))
await done
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Run it:
uv run python -m tapio_examples.hello_world
Two actors talking
Neither actor knows the other in advance. Each learns where to answer from the
message it receives, and Behaviors.same() means "keep going, unchanged".
"""Two actors passing a message back and forth a fixed number of times.
Concepts: bidirectional messaging between two live actors, `Behaviors.same()`
as "keep going, unchanged", and stopping an actor from inside its own handler
with `Behaviors.stopped()`.
Each message carries the address of its sender, so neither actor needs to know
the other in advance. `ping` learns about `pong` from the message it receives.
Neither actor holds any state. The hop count travels in the message instead,
which is the cheapest kind of state an actor system has.
What to watch in the output: hops alternate strictly, ping then pong, and the
last line is ping stopping itself once the rally is over.
Run it with `uv run python -m tapio_examples.ping_pong`.
"""
import asyncio
from collections.abc import Callable
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef
__all__ = ["Ping", "Pong", "main"]
ROUNDS = 3
"""How many times the ball crosses the net before ping calls it a day."""
class Ping(Message):
"""A hop towards the ping actor, with the address to answer."""
hop: int
reply_to: ActorRef["Pong"]
class Pong(Message):
"""A hop towards the pong actor, with the address to answer."""
hop: int
reply_to: ActorRef[Ping]
def pong(record: Callable[[str], None]) -> Behavior[Pong]:
"""Build the pong actor: it answers every hop and never stops itself."""
async def on_pong(ctx: ActorContext[Pong], message: Pong) -> Behavior[Pong]:
record(f"pong: hop {message.hop}")
message.reply_to.tell(Ping(hop=message.hop + 1, reply_to=ctx.self_ref))
return Behaviors.same()
return Behaviors.receive(on_pong)
def ping(
partner: ActorRef[Pong],
record: Callable[[str], None],
finished: asyncio.Event,
) -> Behavior[Ping]:
"""Build the ping actor: it answers until the rally is long enough."""
async def on_ping(ctx: ActorContext[Ping], message: Ping) -> Behavior[Ping]:
record(f"ping: hop {message.hop}")
if message.hop >= ROUNDS * 2:
record("ping: that is enough, stopping")
finished.set()
return Behaviors.stopped()
partner.tell(Pong(hop=message.hop + 1, reply_to=ctx.self_ref))
return Behaviors.same()
return Behaviors.receive(on_ping)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per hop, in the order the actors produced them.
"""
lines: list[str] = []
finished = asyncio.Event()
async with ActorSystem("ping-pong") as system:
table = system.spawn(pong(lines.append), name="pong")
player = system.spawn(ping(table, lines.append, finished), name="ping")
player.tell(Ping(hop=1, reply_to=table))
await finished.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
An actor with state
State is an ordinary attribute on a class-based behavior. There is no lock anywhere. One actor handles one message at a time, which gives the mutual exclusion a lock would have.
"""An actor that holds mutable state, written in the class-based style.
Concepts: `AbstractBehavior` for an actor with fields, a union message type,
and answering a query by sending to the `reply_to` address it carried.
The count is an ordinary attribute, changed in place with no lock anywhere.
One actor handles one message at a time, so the mailbox already gives the
mutual exclusion a lock would.
What to watch in the output: the reply reports 3, not 1. Messages sent to one
actor from one place arrive in order, so all three increments are applied
before the query behind them.
Run it with `uv run python -m tapio_examples.counter`.
"""
import asyncio
from tapio import AbstractBehavior, ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef
__all__ = ["Count", "Counter", "GetCount", "Increment", "main"]
class Count(Message):
"""The answer to a `GetCount`."""
value: int
class Increment(Message):
"""Add to the count."""
by: int = 1
class GetCount(Message):
"""Ask for the count, and say where to send it."""
reply_to: ActorRef[Count]
class Counter(AbstractBehavior[Increment | GetCount]):
"""Counts, and reports the count when asked.
The message type is read from the type parameter, so nothing has to repeat
it. A message of any other type is refused at the sender rather than
landing in this mailbox.
"""
def __init__(self, ctx: ActorContext[Increment | GetCount]) -> None:
"""Start at zero."""
super().__init__(ctx)
self._count = 0
async def on_message(
self, message: Increment | GetCount
) -> Behavior[Increment | GetCount]:
"""Apply an increment, or answer a query."""
match message:
case Increment(by=by):
self._count += by
self.ctx.log.debug("count is now %d", self._count)
case GetCount(reply_to=reply_to):
reply_to.tell(Count(value=self._count))
return Behaviors.same()
async def main() -> int:
"""Run the example.
Returns:
The count the actor reported.
"""
answer: asyncio.Future[int] = asyncio.get_running_loop().create_future()
async def on_count(message: Count) -> Behavior[Count]:
answer.set_result(message.value)
return Behaviors.same()
async with ActorSystem("counter") as system:
readout = system.spawn(Behaviors.receive_message(on_count), name="readout")
# Deferred construction. The factory runs when the actor starts, which
# is what gives the behavior its context, and a restart runs it again.
counter = system.spawn(Behaviors.setup(Counter), name="counter")
counter.tell(Increment())
counter.tell(Increment(by=2))
counter.tell(GetCount(reply_to=readout))
value = await answer
print(f"counter: {value}")
return value
if __name__ == "__main__":
asyncio.run(main())
When an actor fails
A failing handler does not raise into the sender. The exception never leaves the actor's own receive loop. It becomes a decision, taken by whoever declared one, and the default is to stop. Ask for a restart when you know a failure is transient:
"""Restarting an actor whose dependency keeps refusing, without thrashing.
Concepts: `Behaviors.supervise(...).on_failure(...)`, `Restart` with
exponential backoff, the restart window, and what happens to messages sent to
an actor that is between incarnations.
The uploader here stands in for anything that talks to a flaky dependency. Its
first two attempts fail and the third works. Restarting immediately would burn
the whole restart window in a millisecond and stop the actor for a fault that
was about to clear, so the strategy waits, and waits longer each time.
While it waits, the actor is absent, not dead. `tell` stays total, its mailbox
keeps filling, and work sent during the window is handled after the new
incarnation starts rather than dropped. On an unbounded mailbox that costs
memory in proportion to the inbound rate times the window, which is why an
actor that backs off usually wants a bounded mailbox.
The second scenario is the other half of the deal. An actor whose failures
never clear uses up its restart window and is stopped. A supervisor that
restarted forever would turn one bug into a busy one.
What to watch in the output: item 1 fails, and items 2 and 3 were sent while
nobody was there to receive them, yet all three are accounted for. The doomed
actor stops itself after its second failure instead of retrying forever.
Run it with `uv run python -m tapio_examples.supervision_backoff`.
"""
import asyncio
from datetime import timedelta
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import (
ActorContext,
Backoff,
PostStop,
Signal,
SupervisorStrategy,
)
__all__ = ["Upload", "main"]
FAILING_ATTEMPTS = 2
"""How many attempts the simulated dependency refuses before it recovers."""
BACKOFF = Backoff(
min_backoff=timedelta(milliseconds=20),
max_backoff=timedelta(milliseconds=80),
# No jitter, so the example is reproducible. Keep the default in
# production. Without jitter, every actor that saw the same dependency
# fail retries at the same moment, over and over.
random_factor=0.0,
)
class Upload(Message):
"""One item to send to the flaky dependency."""
item: int
def uploader(
lines: list[str],
attempts: list[int],
failed: asyncio.Event,
recovered: asyncio.Event,
) -> Behavior[Upload]:
"""An uploader whose dependency refuses the first two attempts.
Args:
lines: Where to record what happened.
attempts: Every item attempted, across incarnations. It lives outside
the behavior on purpose. A restart rebuilds the actor's own state,
and the dependency it talks to does not reset.
failed: Set after the first failure, so the example can send into the
backoff window rather than sleeping and hoping.
recovered: Set once every item has gone through.
Returns:
The supervised behavior.
"""
def build(ctx: ActorContext[Upload]) -> Behavior[Upload]:
lines.append(f"uploader: incarnation {len(attempts) + 1} ready")
async def on_upload(message: Upload) -> Behavior[Upload]:
attempts.append(message.item)
if len(attempts) <= FAILING_ATTEMPTS:
lines.append(f"uploader: item {message.item} failed")
failed.set()
msg = "the dependency refused the connection"
raise ConnectionError(msg)
lines.append(f"uploader: item {message.item} uploaded")
recovered.set()
return Behaviors.same()
return Behaviors.receive_message(on_upload)
return Behaviors.supervise(Behaviors.setup(build)).on_failure(
SupervisorStrategy.restart(
max_restarts=5, window=timedelta(seconds=1), backoff=BACKOFF
),
# Only the failure this actor knows how to survive. Anything else falls
# through to stop, which is what an unsupervised actor already does.
on=ConnectionError,
)
def doomed(lines: list[str], gave_up: asyncio.Event) -> Behavior[Upload]:
"""An actor whose failure never clears, so its restart window runs out."""
def build(ctx: ActorContext[Upload]) -> Behavior[Upload]:
async def on_upload(message: Upload) -> Behavior[Upload]:
lines.append(f"doomed: item {message.item} failed")
msg = "this one is never going to work"
raise ConnectionError(msg)
async def on_signal(
ctx: ActorContext[Upload], signal: Signal
) -> Behavior[Upload]:
if isinstance(signal, PostStop):
lines.append("doomed: restart window exhausted, stopped")
gave_up.set()
return Behaviors.same()
return Behaviors.receive_message(on_upload, on_signal=on_signal)
return Behaviors.supervise(Behaviors.setup(build)).on_failure(
# One restart per second. A second failure inside that window says the
# fault is not transient after all.
SupervisorStrategy.restart(max_restarts=1, window=timedelta(seconds=1)),
on=ConnectionError,
)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing that happened, in order.
"""
lines: list[str] = []
attempts: list[int] = []
failed, recovered, gave_up = (asyncio.Event(), asyncio.Event(), asyncio.Event())
async with ActorSystem("supervision") as system:
flaky = system.spawn(
uploader(lines, attempts, failed, recovered), name="uploader"
)
flaky.tell(Upload(item=1))
await failed.wait()
# Sent into the backoff window, at an actor that does not currently
# exist. Neither send raises, and neither message is lost.
flaky.tell(Upload(item=2))
flaky.tell(Upload(item=3))
await recovered.wait()
unlucky = system.spawn(doomed(lines, gave_up), name="doomed")
unlucky.tell(Upload(item=4))
unlucky.tell(Upload(item=5))
await gave_up.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Three things about a restart are worth knowing before you rely on it. The
behavior the actor was spawned with is evaluated again, so children spawned in
setup come back and children spawned in a message handler do not. The
mailbox survives, both lanes, so work queued behind the failure is still there
afterwards. And watchers hear nothing, because the ref, path and uid are
unchanged and only the incarnation behind them is new.
While an actor backs off it is absent rather than dead. tell stays total and
its mailbox keeps filling. On an unbounded mailbox that costs memory in
proportion to the inbound rate times the window, so an actor that backs off
usually wants a bounded one.
Knowing that an actor has stopped
There is no "is it alive?" call, because the answer would be out of date before the caller could read it. You are told instead:
"""Keeping a registry honest when the things in it can stop on their own.
Concepts: `ctx.watch`, the `Terminated` signal, and evicting an entry without
leaking it.
A registry of live actors is the first thing most people write, and it comes
with the same bug every time: entries for actors that have since stopped. The
obvious fix, asking a ref whether it is still alive, does not exist in tapio
and would not work if it did. The answer would be out of date as soon as the
caller read it, because the actor can stop in between. Watching turns this
around. Instead of asking, you are told, exactly once, on the system lane, and
the entry is removed where that fact arrives.
The sessions here are not the registry's children. Watching is not parenthood.
It is a one-way subscription to "this actor has stopped", which is the most
one actor can know about another it does not supervise.
What to watch in the output: the registry never checks whether a session is
alive. It is told, and the count afterwards shows the entry is gone.
Run it with `uv run python -m tapio_examples.death_watch`.
"""
import asyncio
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef, Signal, Terminated
__all__ = ["Census", "Close", "Register", "main"]
class Close(Message):
"""Ask a session to end itself."""
class Register(Message):
"""Tell the registry about a session it should keep track of."""
who: str
session: ActorRef[Close]
class Census(Message):
"""Ask the registry to report who it still holds."""
def session() -> Behavior[Close]:
"""A session that ends when asked, and says nothing on the way out."""
async def on_close(message: Close) -> Behavior[Close]:
return Behaviors.stopped()
return Behaviors.receive_message(on_close)
def registry(
lines: list[str], evicted: asyncio.Event, counted: asyncio.Event
) -> Behavior[Register | Census]:
"""A registry of live sessions that evicts by being told, not by asking."""
def build(ctx: ActorContext[Register | Census]) -> Behavior[Register | Census]:
live: dict[str, ActorRef[Close]] = {}
async def on_message(
message: Register | Census,
) -> Behavior[Register | Census]:
match message:
case Register(who=who, session=ref):
live[who] = ref
# One call, and from here on this registry cannot hold a
# stale entry for that session.
ctx.watch(ref)
lines.append(f"registry: registered {who}, holding {len(live)}")
case Census():
lines.append(f"registry: holding {sorted(live)}")
counted.set()
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Register | Census], signal: Signal
) -> Behavior[Register | Census]:
if isinstance(signal, Terminated):
# The signal carries the ref, and a ref knows its own path, so
# the entry is found without a second lookup table.
who = signal.ref.path.name
live.pop(who, None)
lines.append(f"registry: {who} stopped, holding {len(live)}")
evicted.set()
return Behaviors.same()
return Behaviors.receive_message(on_message, on_signal=on_signal)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing the registry did, in order.
"""
lines: list[str] = []
evicted, counted = asyncio.Event(), asyncio.Event()
async with ActorSystem("death-watch") as system:
desk = system.spawn(registry(lines, evicted, counted), name="registry")
ada = system.spawn(session(), name="ada")
grace = system.spawn(session(), name="grace")
desk.tell(Register(who="ada", session=ada))
desk.tell(Register(who="grace", session=grace))
# The session ends for its own reasons. This is the case a liveness
# check cannot handle: nobody tells the registry, and it finds out
# anyway.
ada.tell(Close())
await evicted.wait()
desk.tell(Census())
await counted.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Escalating, and shutting down
An actor that cannot fix a failure hands it to its parent, which can. If it
reaches a guardian, nobody has taken responsibility for it, so the system
terminates and re-raises the cause from when_terminated.
"""A failure the actor that hit it cannot fix, handed to the one that can.
Concepts: `SupervisorStrategy.escalate()`, `ChildFailed`, a whole subtree being
rebuilt by its supervisor, and what happens when an escalation runs out of
supervisors.
A worker that cannot parse its input has no way to repair the pipeline it is
part of. Its parent, which built the pipeline, does. Escalating says that:
stop me, and make this your decision. The parent then takes its own decision,
which here is a restart, so the setup runs again and rebuilds every child
rather than only the one that broke.
Escalation is ordinary message flow, not an exception thrown across a task
boundary. The child stops itself and puts a signal on the parent's system
lane, which is why it can be ordered, observed and tested like anything else.
This example shows two more things. An actor outside the restarted subtree is
untouched, because a child failing must never cancel its siblings. That is why
the runtime uses no task group. And an escalation that reaches the guardian
has run out of actors willing to take responsibility, so the system terminates
and re-raises the cause from `when_terminated`. The service embedding tapio
then decides whether to exit or rebuild.
What to watch in the output: the ticker keeps counting across the restart, and
the second scenario ends with the original error, carrying the path it climbed
through.
Run it with `uv run python -m tapio_examples.escalation`.
"""
import asyncio
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import (
ActorContext,
ActorRef,
PostStop,
PreRestart,
Signal,
SupervisorStrategy,
)
__all__ = ["Parse", "Tick", "main"]
class Parse(Message):
"""A line for the worker to parse. An empty one is unparseable."""
line: str
class Tick(Message):
"""A nudge for the ticker, which is here to keep working throughout."""
def worker(lines: list[str], parsed: asyncio.Event | None = None) -> Behavior[Parse]:
"""A parser that escalates rather than pretending it can recover."""
def build(ctx: ActorContext[Parse]) -> Behavior[Parse]:
lines.append("worker: ready")
async def on_parse(message: Parse) -> Behavior[Parse]:
if not message.line:
lines.append("worker: cannot parse an empty line")
msg = "empty input"
raise ValueError(msg)
lines.append(f"worker: parsed {message.line!r}")
if parsed is not None:
parsed.set()
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Parse], signal: Signal
) -> Behavior[Parse]:
if isinstance(signal, PostStop):
lines.append("worker: stopped")
return Behaviors.same()
return Behaviors.receive_message(on_parse, on_signal=on_signal)
return Behaviors.supervise(Behaviors.setup(build)).on_failure(
SupervisorStrategy.escalate(), on=ValueError
)
def pipeline(
lines: list[str],
workers: list[ActorRef[Parse]],
rebuilt: asyncio.Event,
parsed: asyncio.Event,
) -> Behavior[Parse]:
"""A supervisor that builds its subtree in setup, and so rebuilds it on restart."""
def build(ctx: ActorContext[Parse]) -> Behavior[Parse]:
lines.append(f"pipeline: building, incarnation {len(workers) + 1}")
# Spawned in setup, which is what makes this child come back. A child
# spawned from a message handler would be gone until that message
# arrives again.
workers.append(ctx.spawn(worker(lines, parsed), name="worker"))
if len(workers) > 1:
rebuilt.set()
async def on_parse(message: Parse) -> Behavior[Parse]:
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Parse], signal: Signal
) -> Behavior[Parse]:
if isinstance(signal, PreRestart):
lines.append("pipeline: restarting after the worker escalated")
return Behaviors.same()
return Behaviors.receive_message(on_parse, on_signal=on_signal)
return Behaviors.supervise(Behaviors.setup(build)).on_failure(
SupervisorStrategy.restart(max_restarts=3), on=ValueError
)
def ticker(lines: list[str], ticks: list[int]) -> Behavior[Tick]:
"""An actor with no part in any of this, which is the point of it."""
async def on_tick(ctx: ActorContext[Tick], message: Tick) -> Behavior[Tick]:
ticks.append(len(ticks) + 1)
lines.append(f"ticker: tick {len(ticks)}")
return Behaviors.same()
return Behaviors.receive(on_tick)
async def subtree_restarted_by_its_supervisor(lines: list[str]) -> None:
"""Run the first scenario: a worker escalates and its parent rebuilds."""
workers: list[ActorRef[Parse]] = []
ticks: list[int] = []
rebuilt, parsed = asyncio.Event(), asyncio.Event()
async with ActorSystem("escalation") as system:
system.spawn(pipeline(lines, workers, rebuilt, parsed), name="pipeline")
beat = system.spawn(ticker(lines, ticks), name="ticker")
beat.tell(Tick())
workers[0].tell(Parse(line=""))
await rebuilt.wait()
# The sibling never noticed. A failing actor stops only itself and,
# through its supervisor's decision, that supervisor's subtree.
beat.tell(Tick())
workers[1].tell(Parse(line="ok"))
await parsed.wait()
async def escalation_that_nobody_catches(lines: list[str]) -> None:
"""Run the second scenario: the escalation reaches the guardian."""
system = ActorSystem("unsupervised")
lonely = system.spawn(worker(lines), name="worker")
lonely.tell(Parse(line=""))
try:
await system.when_terminated()
except ValueError as error:
lines.append(f"system: terminated by {error}")
for note in getattr(error, "__notes__", []):
lines.append(f"system: {note}")
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing that happened, in order.
"""
lines: list[str] = []
await subtree_restarted_by_its_supervisor(lines)
await escalation_that_nobody_catches(lines)
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Ordinary shutdown drains the tree bottom-up against one deadline for the whole
tree, so the worst case follows shutdown_timeout rather than the depth of
the tree:
"""Shutting a system down on SIGINT, and stopping the tree in the right order.
Concepts: a real signal handler on the event loop, `system.terminate()`, the
bottom-up drain, and `PostStop` as the place a resource is released.
The signal is real. The handler is installed with `add_signal_handler` and the
process sends itself SIGINT, so this exercises the same wiring a deployed
service uses. Nothing is simulated, which matters, because "we handle SIGINT"
is the claim that turns out to be false the first time a container is stopped.
Shutdown is bottom-up, and the whole tree races one deadline rather than one
per actor. Worst-case shutdown therefore follows `shutdown_timeout` instead of
multiplying by the depth of the tree. Each actor sees `PostStop` after its
children have seen theirs, so a connection pool held by a parent outlives the
children still handing work back to it.
`PostStop` is best effort. An actor still stuck in a handler when the deadline
passes is cancelled and may never see it. Release what must be released there,
but do not make correctness depend on it running.
What to watch in the output: the two connections stop before the pool that
owns them, and the pool's own line is last.
Run it with `uv run python -m tapio_examples.graceful_shutdown`.
"""
import asyncio
import os
import signal
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, PostStop, Signal
__all__ = ["Query", "main"]
class Query(Message):
"""A unit of work for a connection to handle."""
sql: str
def connection(name: str, lines: list[str], served: asyncio.Event) -> Behavior[Query]:
"""One pooled connection, which reports when it is closed."""
async def on_query(message: Query) -> Behavior[Query]:
lines.append(f"{name}: ran {message.sql!r}")
if name == "conn-2": # the second of the two, so both have answered
served.set()
return Behaviors.same()
async def on_signal(ctx: ActorContext[Query], sig: Signal) -> Behavior[Query]:
if isinstance(sig, PostStop):
lines.append(f"{name}: closed")
return Behaviors.same()
return Behaviors.receive_message(on_query, on_signal=on_signal)
def pool(lines: list[str], served: asyncio.Event) -> Behavior[Query]:
"""A pool that owns two connections and outlives both of them."""
def build(ctx: ActorContext[Query]) -> Behavior[Query]:
for name in ("conn-1", "conn-2"):
ctx.spawn(connection(name, lines, served), name=name).tell(
Query(sql="select 1")
)
async def on_query(message: Query) -> Behavior[Query]:
return Behaviors.same()
async def on_signal(ctx: ActorContext[Query], sig: Signal) -> Behavior[Query]:
if isinstance(sig, PostStop):
lines.append("pool: closed, after every connection in it")
return Behaviors.same()
return Behaviors.receive_message(on_query, on_signal=on_signal)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing that happened, in order.
"""
lines: list[str] = []
served, interrupted = asyncio.Event(), asyncio.Event()
loop = asyncio.get_running_loop()
def on_sigint() -> None:
# It does nothing but wake the shutdown. A signal handler runs outside
# every actor, so anything it touched directly would be state no
# mailbox is protecting.
lines.append("signal: SIGINT, shutting down")
interrupted.set()
loop.add_signal_handler(signal.SIGINT, on_sigint)
try:
system = ActorSystem("graceful")
system.spawn(pool(lines, served), name="pool")
await served.wait()
os.kill(os.getpid(), signal.SIGINT)
await interrupted.wait()
await system.terminate()
finally:
# Put the interpreter's own handler back, so a second Ctrl-C after
# this example is a KeyboardInterrupt again.
loop.remove_signal_handler(signal.SIGINT)
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Asking for an answer
ask sends one message and awaits one reply. The request still carries a ref
for the answer to come back to, as it does above. What ask adds is that the
ref is a promise rather than an actor, so the reply can be awaited instead of
arranged for.
expect is required, and it matters. A promise has no cell and so no declared
message type of its own. Without expect, request/response would be the only
delivery in the library with no type check on it. A reply of the wrong type
raises AskTypeError in the caller rather than handing back a value whose
static type is a lie.
The failures are the reason to read the example. A timeout is the expensive answer, so it is the last resort. An ask watches its target, and a target that stops fails the ask at once instead of spending the deadline on an answer that is never coming.
"""Asking for an answer, and the three ways of not getting one.
Concepts: `ref.ask`, `AskTimeoutError`, `AskTargetTerminated`, and where a
reply goes once nobody is waiting for it.
`ask` is sugar over the `reply_to` field `hello_world` starts with. The
request still carries a ref for the answer to come back to. What `ask` adds is
that the ref is a promise rather than an actor, so the caller can await the
reply instead of arranging to be told about it later.
The interesting part is what happens when no reply arrives. A timeout is the
expensive answer, so tapio avoids it when it can. An ask watches its target,
and a target that stops fails the ask at once instead of making the caller
wait out the deadline for an answer that cannot come. With the five-second
default, that is the difference between failing now and failing in five
seconds.
What to watch in the output: the third line is the answer the desk produced
for a lookup that had already timed out. It did not vanish, and it resolved
nothing, because there was no future left to resolve. It was recorded as a
dead letter instead. The fourth line is the fast failure: the desk closes
while a reader is waiting, and the reader hears about it immediately even
though it asked for thirty seconds of patience.
Run it with `uv run python -m tapio_examples.ask_timeout`.
"""
import asyncio
from datetime import timedelta
from tapio import (
ActorSystem,
AskTargetTerminated,
AskTimeoutError,
Behavior,
Behaviors,
DeadLetter,
Message,
)
from tapio.actor import ActorContext, ActorRef
__all__ = ["Close", "Lookup", "Shelf", "main"]
TIMEOUT = timedelta(milliseconds=50)
"""Short enough to keep the example quick, and the number the reader prints."""
class Shelf(Message):
"""Where a book is, which is what a lookup is answered with."""
title: str
shelf: int
class Lookup(Message):
"""Ask the desk where a book is, and say where to send the answer."""
title: str
reply_to: ActorRef[Shelf]
class Close(Message):
"""Tell the desk to shut, with whatever it was doing unfinished."""
def desk(catalogue: dict[str, int], stuck: asyncio.Event) -> Behavior[Lookup | Close]:
"""A reference desk that answers lookups, and stalls on one of them.
The stall is an `await` inside the handler, which is the usual way an
actor becomes slow: it is waiting on something outside itself. While it
waits it reads nothing else, so everything behind it in the mailbox waits
too.
"""
def build(ctx: ActorContext[Lookup | Close]) -> Behavior[Lookup | Close]:
async def on_message(message: Lookup | Close) -> Behavior[Lookup | Close]:
match message:
case Close():
ctx.log.info("closing with work outstanding")
return Behaviors.stopped()
case Lookup(title=title, reply_to=reply_to):
if title not in catalogue:
# The slow path. No answer until something else
# happens, which from the asker's side looks the same
# as a desk that has died.
await stuck.wait()
reply_to.tell(Shelf(title=title, shelf=0))
else:
reply_to.tell(Shelf(title=title, shelf=catalogue[title]))
return Behaviors.same()
return Behaviors.receive_message(on_message)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing the reader learned, in order.
"""
lines: list[str] = []
stuck = asyncio.Event()
letters: asyncio.Queue[DeadLetter] = asyncio.Queue()
async with ActorSystem("ask-timeout") as system:
system.dead_letters.subscribe(letters.put_nowait)
reference = system.spawn(desk({"Dune": 3}, stuck), name="desk")
# The happy path: one call, one reply, and a value with a type.
found = await reference.ask(
lambda reply_to: Lookup(title="Dune", reply_to=reply_to),
expect=Shelf,
timeout=TIMEOUT,
)
lines.append(f"reader: '{found.title}' is on shelf {found.shelf}")
# The desk stalls on this one, so the deadline is what ends the wait.
try:
await reference.ask(
lambda reply_to: Lookup(title="Ulysses", reply_to=reply_to),
expect=Shelf,
timeout=TIMEOUT,
)
except AskTimeoutError:
seconds = TIMEOUT.total_seconds()
lines.append(f"reader: gave up on 'Ulysses' after {seconds:g}s")
# The desk gets unstuck and answers a lookup nobody is waiting for any
# more. The answer is not lost. It is recorded as a dead letter.
stuck.set()
letter = await letters.get()
while not isinstance(letter.message, Shelf):
letter = await letters.get()
lines.append(f"dead letter: {type(letter.message).__name__} ({letter.reason})")
# Now the desk closes with a reader still waiting. The ask allowed
# thirty seconds and uses none of them, because it is watching the
# desk and hears that it stopped.
reference.tell(Close())
try:
await reference.ask(
lambda reply_to: Lookup(title="Dune", reply_to=reply_to),
expect=Shelf,
timeout=timedelta(seconds=30),
)
except AskTargetTerminated:
lines.append("reader: the desk closed, so there was no point waiting")
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Awaiting an ask inside a handler stops that actor reading its mailbox until the reply lands. That is sometimes what you want, but usually not: an actor that asks and waits cannot answer anyone else.
Doing something later, and holding what you cannot do yet
A timer sends the actor a message on its own user lane. That is the whole
design. A tick is ordinary traffic, so it queues behind what is already there
and can never re-enter a handler that is still running. Timers belong to the
cell, not to the behavior, which is why they come from Behaviors.with_timers
and why a restart cancels them. A tick scheduled by the incarnation that just
failed must not reach the one replacing it.
start_fixed_delay measures the gap from one send to the next, so an actor
that falls behind simply gets fewer ticks. start_fixed_rate counts ticks off
a fixed schedule and sends the missed ones one after another once a stall is
over. Think twice about the second, because the catch-up burst arrives at an
actor that has just shown it is not keeping up. It is the right choice when
what you promised really is a rate:
"""A token bucket in one actor, with no lock anywhere.
Concepts: `Behaviors.with_timers`, a fixed-rate refill, and the idea that the
mailbox is the mutex.
A rate limiter is shared mutable state under concurrent access, which is the
textbook case for a lock. Written as an actor it needs none. The bucket is an
ordinary variable inside one actor, and an actor handles one message at a
time, so the runtime provides the mutual exclusion. There is no critical
section here because there is no concurrency here. The concurrency is outside,
in the callers, and the mailbox puts it in order.
The refill is a timer, and a timer is not a callback running beside the
receive loop. It puts a `Refill` on this actor's own user lane, so it queues
like everything else and cannot land in the middle of a decision about a
request. That is why the bucket needs no locking even though requests read it
and a clock writes it.
`start_fixed_rate` is used here on purpose, rather than `start_fixed_delay`. A
limiter that let time slip when it was busy would hand out fewer permits than
it promised, and the promise is a rate. The docs warn about that catch-up
burst, and here catching up is the correct behaviour.
What to watch in the output: the first burst of five requests spends a bucket
holding two, and the other three are refused. That is the limiter working, not
failing. Then one tick of the refill puts a permit back and the next request
is allowed.
Run it with `uv run python -m tapio_examples.rate_limiter`.
"""
import asyncio
from datetime import timedelta
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef, TimerScheduler
__all__ = ["Decision", "Refill", "Request", "main"]
CAPACITY = 2
"""How many permits the bucket holds when full."""
REFILL = timedelta(milliseconds=30)
"""How often one permit is put back."""
class Decision(Message):
"""What the limiter decided about one request."""
label: str
allowed: bool
class Request(Message):
"""Ask for a permit."""
label: str
reply_to: ActorRef[Decision]
class Refill(Message):
"""The timer, asking for one permit to be put back."""
Traffic = Request | Refill
def limiter(capacity: int, refill: timedelta) -> Behavior[Traffic]:
"""A token bucket that refuses what it cannot allow.
A limiter should refuse rather than queue. Holding a request until a
permit exists turns a rate limit into unbounded latency, and the caller
usually has something better to do with the answer.
"""
def with_scheduler(timers: TimerScheduler[Traffic]) -> Behavior[Traffic]:
def build(ctx: ActorContext[Traffic]) -> Behavior[Traffic]:
# Plain state on the closure. No lock, and none needed, because
# nothing else in the process can reach it.
tokens = capacity
timers.start_fixed_rate("refill", Refill(), refill)
async def on_message(message: Traffic) -> Behavior[Traffic]:
nonlocal tokens
match message:
case Refill():
tokens = min(capacity, tokens + 1)
case Request(label=label, reply_to=reply_to):
allowed = tokens > 0
if allowed:
tokens -= 1
reply_to.tell(Decision(label=label, allowed=allowed))
return Behaviors.same()
return Behaviors.receive_message(on_message)
return Behaviors.setup(build)
return Behaviors.with_timers(with_scheduler)
def caller(lines: list[str], seen: asyncio.Event, expected: int) -> Behavior[Decision]:
"""Records what the limiter decided, and says when it has heard enough."""
def build(ctx: ActorContext[Decision]) -> Behavior[Decision]:
heard = 0
async def on_decision(message: Decision) -> Behavior[Decision]:
nonlocal heard
verdict = "allowed" if message.allowed else "throttled"
lines.append(f"{message.label}: {verdict}")
heard += 1
if heard == expected:
seen.set()
return Behaviors.same()
return Behaviors.receive_message(on_decision)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per decision, in the order the limiter made them.
"""
lines: list[str] = []
burst_done = asyncio.Event()
after_refill = asyncio.Event()
async with ActorSystem("rate-limiter") as system:
client = system.spawn(caller(lines, burst_done, expected=5), name="client")
gate = system.spawn(limiter(CAPACITY, REFILL), name="limiter")
# Five at once against a bucket of two. Sent in one go, so no refill
# can land in the middle of the burst.
for n in range(1, 6):
gate.tell(Request(label=f"req-{n}", reply_to=client))
await burst_done.wait()
# Wait for the bucket to earn a permit back, then spend it.
later = system.spawn(caller(lines, after_refill, expected=1), name="later")
await asyncio.sleep(REFILL.total_seconds() * 1.5)
gate.tell(Request(label="req-6", reply_to=later))
await after_refill.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
An actor that cannot answer yet has one good option: accept what arrives, put
it aside, and replay it once it can. Behaviors.with_stash gives it a bounded
buffer to do that with, and stash.unstash_all(next_behavior) switches state
and replays the backlog in one call.
The replay goes to the front of the mailbox, ahead of anything that queued up in the meantime, so nothing is reordered. The actor also stays an ordinary actor throughout, which is why the replay is not a loop inside the unstash. A stop arriving mid-replay is honoured rather than queued behind work nobody wants any more.
"""Holding traffic while an actor loads the state it needs to answer it.
Concepts: `Behaviors.with_stash`, `stash.unstash_all`, and behavior-switching
as the way an actor says "I am ready now".
An actor that has to load something before it can work has three options, and
only one is good. Dropping what arrives loses work. Blocking the receive loop
on the load leaves the actor unable to answer anything, including a stop: the
actor is not slow, it is absent. The third option is to accept the messages,
put them aside, and replay them once the state exists.
Replay puts the held messages back at the front of the mailbox, rather than
handing them to the behavior one at a time. Two things follow. The held
messages keep their arrival order and stay ahead of anything that queued up
while the actor was loading, so nothing is reordered. And the actor stays an
ordinary actor throughout: a signal still outranks the backlog, so a stop
arriving mid-replay is honoured instead of queued behind work nobody wants.
What to watch in the output: greetings 1 and 2 arrived before the template and
3 arrived after, and all three are answered in the order they were sent. The
stash is what makes that true. Without it the first two would have been
answered wrongly, or not at all.
Run it with `uv run python -m tapio_examples.stash_on_startup`.
"""
import asyncio
from datetime import timedelta
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, StashBuffer, TimerScheduler
__all__ = ["Greet", "Loaded", "main"]
LOAD_TIME = timedelta(milliseconds=20)
"""How long the pretend store takes to answer."""
class Greet(Message):
"""Ask for someone to be greeted, which needs the template."""
whom: str
class Loaded(Message):
"""The template has arrived, so the actor can start working."""
template: str
Traffic = Greet | Loaded
def greeter(lines: list[str], done: asyncio.Event) -> Behavior[Traffic]:
"""A greeter that cannot greet until its template has loaded.
The stash capacity is required, and it matters. A stash holds traffic the
actor is not keeping up with, so an unbounded one is a memory leak.
Overflow raises in this actor, where the decision about what to drop
belongs.
"""
def ready(template: str) -> Behavior[Traffic]:
"""What the greeter becomes once it has something to greet with."""
async def on_greet(message: Traffic) -> Behavior[Traffic]:
if isinstance(message, Greet):
lines.append("greeter: " + template.format(message.whom))
if message.whom == "carol":
done.set()
return Behaviors.same()
return Behaviors.receive_message(on_greet)
def with_scheduler(timers: TimerScheduler[Traffic]) -> Behavior[Traffic]:
def with_buffer(stash: StashBuffer[Traffic]) -> Behavior[Traffic]:
def build(ctx: ActorContext[Traffic]) -> Behavior[Traffic]:
# Stands in for the ask or the child actor a real load would
# use. All that matters here is that the answer arrives later,
# as a message, like everything else an actor learns.
timers.start_single("load", Loaded(template="hello, {}!"), LOAD_TIME)
lines.append("greeter: loading, holding what arrives")
async def while_loading(message: Traffic) -> Behavior[Traffic]:
if isinstance(message, Loaded):
lines.append(f"greeter: loaded, replaying {stash.size} held")
# One call switches state and replays the backlog.
# Everything held goes back in front of whatever
# arrived while this message was being handled.
return stash.unstash_all(ready(message.template))
stash.stash(message)
lines.append(f"greeter: not ready, stashed {message.whom}")
return Behaviors.same()
return Behaviors.receive_message(while_loading)
return Behaviors.setup(build)
return Behaviors.with_stash(16, with_buffer)
return Behaviors.with_timers(with_scheduler)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing the greeter did, in order.
"""
lines: list[str] = []
done = asyncio.Event()
async with ActorSystem("stash-on-startup") as system:
desk = system.spawn(greeter(lines, done), name="greeter")
# Two arrive before the template does.
desk.tell(Greet(whom="ada"))
desk.tell(Greet(whom="grace"))
# And one after, which must not overtake them.
await asyncio.sleep(LOAD_TIME.total_seconds() * 2)
desk.tell(Greet(whom="carol"))
await done.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
The capacity is required. A stash holds traffic the actor is not keeping up
with, so an unbounded one is a memory leak. Overflow raises
StashOverflowError in the actor that stashed, where the decision about what
to drop belongs. A restart empties the buffer, because messages held by the
state that just failed are not the new state's to answer, and what is
discarded is published as a dead letter rather than dropped.
Talking to someone else's protocol
An actor's declared message type is a contract, which raises a question the first time two actors written by different people have to talk. The service you called replies with its own reply type, and that type does not belong in your protocol. Widening yours to admit it is wrong twice over: it lets anyone send you that message, and it puts a foreign vocabulary inside your handlers.
ctx.message_adapter gives you a ref to hand out instead. It accepts the
other protocol's message, translates it into one of yours, and delivers the
result onto your own user lane. There it is ordinary traffic: it queues where
it arrived, it cannot re-enter a running handler, and it is validated against
your declared type like anything else.
The translation runs in your actor rather than in the sender, which is the reason to prefer this over translating at the call site. The function is your code, so a mistake in it becomes your supervision decision, and a sender that has never heard of the adapter does not have your bug raised into it. An adapter is not an actor, so it cannot be watched or asked. Watch the actor that owns it.
One address, several actors
When the work is uniform and the answer to "too slow" is "more of the same actor", a pool router puts one address in front of several of them:
"""One address in front of several workers, and what happens when one dies.
Concepts: `Routers.pool`, round-robin fan-out, a bounded mailbox, and `offer`
as the way a producer is made to wait.
Reach for a pool router when the work is uniform and the answer to "too slow"
is "more of the same actor". The router is an ordinary actor. It has a
mailbox, it handles one message at a time, and all it does is hand each one to
the next routee in turn. The routees are its children, so their failures are
supervised the ordinary way, and when one stops the pool shrinks.
A router creates no backpressure of its own. Sending to a router never blocks,
just as sending anywhere else never blocks, so a fast producer against slow
workers fills something up. Which thing fills up is the mailbox's business.
That is why the router here has a bounded mailbox and the producer uses
`await router.offer(...)`: the producer is slowed by the pool it is feeding
instead of piling up an unbounded backlog in front of it.
What to watch in the output: six jobs land on three workers in strict
rotation, never twice in a row on the same one. Then a worker is given a job
it does not survive, and the pool carries on with two. The router was told its
routee stopped and stopped routing to it, rather than sending work to an
address nobody reads.
Run it with `uv run python -m tapio_examples.worker_pool`.
"""
import asyncio
from tapio import (
ActorSystem,
Behavior,
Behaviors,
MailboxConfig,
Message,
OverflowStrategy,
Routers,
)
from tapio.actor import ActorContext, ActorRef, PostStop, Signal
__all__ = ["Done", "Job", "Stopped", "main"]
POOL_SIZE = 3
"""How many workers sit behind the one address."""
WORK_TIME = 0.005
"""How long a worker pretends each job takes."""
class Done(Message):
"""What a worker reports when it has finished a job."""
worker: str
item: int
class Stopped(Message):
"""What a worker reports on its way out."""
worker: str
Report = Done | Stopped
class Job(Message):
"""A unit of work, and the address to report it to."""
item: int
reply_to: ActorRef[Report]
poison: bool = False
def worker() -> Behavior[Job]:
"""One member of the pool.
Wrapped in `Behaviors.setup` so each routee gets its own. A pool built
from one already-constructed stateful behavior would share that state
across every member.
"""
def build(ctx: ActorContext[Job]) -> Behavior[Job]:
name = ctx.path.name
# Where to report a stop this worker chose itself. An ordinary
# shutdown is not worth reporting, so it stays empty for those.
dying: list[ActorRef[Report]] = []
async def on_job(message: Job) -> Behavior[Job]:
if message.poison:
# A worker deciding it cannot go on. Its parent is the router,
# which is watching it, so the pool shrinks by one.
dying.append(message.reply_to)
return Behaviors.stopped()
await asyncio.sleep(WORK_TIME)
message.reply_to.tell(Done(worker=name, item=message.item))
return Behaviors.same()
async def on_signal(_: ActorContext[Job], signal: Signal) -> Behavior[Job]:
if isinstance(signal, PostStop) and dying:
dying[0].tell(Stopped(worker=name))
return Behaviors.same()
return Behaviors.receive_message(on_job, on_signal=on_signal)
return Behaviors.setup(build)
def collector(lines: list[str], marks: dict[int, asyncio.Event]) -> Behavior[Report]:
"""Writes down what every worker reported, so the run has one output.
It also signals the points the script below waits for. Only the actor
knows when it has handled something, so saying so is better than sleeping
for a guessed interval.
"""
async def on_report(message: Report) -> Behavior[Report]:
match message:
case Done(worker=name, item=item):
lines.append(f"{name}: job {item}")
case Stopped(worker=name):
lines.append(f"{name}: stopped, and left the pool")
mark = marks.get(len(lines))
if mark is not None:
mark.set()
return Behaviors.same()
return Behaviors.receive_message(on_report)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing a worker reported, in the order it happened.
"""
lines: list[str] = []
marks = {n: asyncio.Event() for n in (6, 7, 11)}
async with ActorSystem("worker-pool") as system:
reports = system.spawn(collector(lines, marks), name="collector")
# The bounded mailbox is what gives `offer` something to wait for.
# Unbounded, the producer below would hand over all six jobs before a
# single one had been started.
workers = system.spawn(
Routers.pool(POOL_SIZE, worker()),
name="workers",
mailbox=MailboxConfig(capacity=2, on_overflow=OverflowStrategy.FAIL),
)
for item in range(1, 7):
# Waits while the router is full, which is backpressure reaching
# the producer from the pool it is feeding.
await workers.offer(Job(item=item, reply_to=reports))
await marks[6].wait()
# Now take a worker out from under the router. The rotation is where
# the first six jobs left it, so this lands on the first worker.
await workers.offer(Job(item=0, reply_to=reports, poison=True))
# A worker reports its own stop before the runtime tells its watchers,
# and the router is one of them. So by the time this line runs, the
# router's `Terminated` is queued ahead of anything sent next.
await marks[7].wait()
# The pool is down to two, and the rotation carries on across what is
# left rather than starting again.
for item in range(7, 11):
await workers.offer(Job(item=item, reply_to=reports))
await marks[11].wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
The router is an ordinary actor whose routees are its children, and that
decides most of its behaviour. Their failures are supervised where they were
declared, so wrapping the routee behavior in Behaviors.supervise(...)
restarts a failed routee in place. A routee that stops leaves the pool, and
when the last one goes the router stops too, because an empty pool is an
address that silently swallows work.
A router creates no backpressure of its own. Sending to one never blocks, just
as sending anywhere never blocks. A routee that cannot take a message gets it
dead-lettered, because the router did not write that message and failing would
take a whole pool down over one busy member. Put backpressure on the router's
own mailbox instead, where a producer can offer into it and wait.
Behaviors as the states of a protocol
The state is the behavior, which is what makes the illegal transitions impossible to write rather than merely checked for:
"""A protocol as a state machine, with one behavior per state.
Concepts: behavior-switching as the states of a protocol, and
`ctx.message_adapter` for talking to a service whose replies are not in your
protocol.
The usual way to write a connection is a field called `state`, an enum, and a
`match` at the top of every handler deciding whether this message is legal
right now. An actor needs none of that. The state is the behavior. A
connection that has not authenticated is a different function from one that
has, so a `Send` arriving too early is not an illegal combination to check
for. It is simply a message that state does not handle.
The gain is that the illegal transitions cannot be written. No branch in
`ready` can accidentally accept a second `Authenticate`, because `ready` never
mentions it. And there is no state variable to leave inconsistent, because
switching state means returning a different behavior and nothing else.
The token service is the second half. It is somebody else's actor and it
answers in its own vocabulary, `Token`, which does not belong in a
connection's protocol. Widening the connection to accept a `Token` would let
anyone send it one. Instead the connection hands out an adapter, which accepts
a `Token` and turns it into the `Authenticated` the connection understands.
The translation runs inside the connection, so a mistake in it is the
connection's failure and not the token service's.
What to watch in the output: the `Send` that arrives while connecting is
refused rather than queued, and the identical `Send` after authentication goes
through. Same message, same actor, different state.
Run it with `uv run python -m tapio_examples.state_machine`.
"""
import asyncio
from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef
__all__ = ["Close", "Connect", "Send", "main"]
class Token(Message):
"""What the token service answers with. Not the connection's protocol."""
value: str
class Issue(Message):
"""What the token service accepts."""
reply_to: ActorRef[Token]
class Connect(Message):
"""Open the connection, which starts authentication."""
class Authenticated(Message):
"""A token has arrived, translated into the connection's own protocol."""
token: str
class Send(Message):
"""Send a payload, which only works once the connection is ready."""
payload: str
class Close(Message):
"""Close the connection, draining what it is already doing."""
Protocol = Connect | Authenticated | Send | Close
def tokens() -> Behavior[Issue]:
"""A service that mints tokens, in its own vocabulary."""
async def on_issue(message: Issue) -> Behavior[Issue]:
message.reply_to.tell(Token(value="t-42"))
return Behaviors.same()
return Behaviors.receive_message(on_issue)
def connection(
lines: list[str], marks: dict[int, asyncio.Event], service: ActorRef[Issue]
) -> Behavior[Protocol]:
"""A connection whose protocol states are its behaviors."""
def say(line: str) -> None:
"""Write a line down, and signal the points the script waits for."""
lines.append(line)
mark = marks.get(len(lines))
if mark is not None:
mark.set()
def build(ctx: ActorContext[Protocol]) -> Behavior[Protocol]:
# Handed to the token service in place of this actor's own ref, so the
# service can answer without the connection accepting `Token` into its
# protocol. The lambda has no annotation, so the type is passed in.
as_authenticated: ActorRef[Token] = ctx.message_adapter(
lambda token: Authenticated(token=token.value), Token
)
def disconnected() -> Behavior[Protocol]:
"""Nothing is open. The only thing that can happen is `Connect`."""
async def on_message(message: Protocol) -> Behavior[Protocol]:
if not isinstance(message, Connect):
say(f"conn: refused {type(message).__name__}, not open")
return Behaviors.same()
say("conn: connecting, asking for a token")
service.tell(Issue(reply_to=as_authenticated))
return authenticating()
return Behaviors.receive_message(on_message)
def authenticating() -> Behavior[Protocol]:
"""Waiting for a token. Still not a state that can send anything."""
async def on_message(message: Protocol) -> Behavior[Protocol]:
if not isinstance(message, Authenticated):
say(f"conn: refused {type(message).__name__}, still connecting")
return Behaviors.same()
say(f"conn: authenticated with {message.token}")
return ready(message.token)
return Behaviors.receive_message(on_message)
def ready(token: str) -> Behavior[Protocol]:
"""Open. This is the only state that mentions `Send` at all."""
async def on_message(message: Protocol) -> Behavior[Protocol]:
match message:
case Send(payload=payload):
say(f"conn: sent {payload!r} with {token}")
case Close():
# A marker message. Everything already queued sits
# ahead of a message sent now, so when this one comes
# back the mailbox is drained and it is safe to stop.
say("conn: closing, draining what is queued")
ctx.self_ref.tell(Close())
return closing()
case _:
say(f"conn: refused {type(message).__name__}, open")
return Behaviors.same()
return Behaviors.receive_message(on_message)
def closing() -> Behavior[Protocol]:
"""Going away. What was already queued is answered, then it stops."""
async def on_message(message: Protocol) -> Behavior[Protocol]:
if isinstance(message, Close):
say("conn: closed")
return Behaviors.stopped()
say(f"conn: dropped {type(message).__name__}, closing")
return Behaviors.same()
return Behaviors.receive_message(on_message)
return disconnected()
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
One line per thing the connection did, in order.
"""
lines: list[str] = []
marks = {n: asyncio.Event() for n in (3, 4, 8)}
async with ActorSystem("state-machine") as system:
service = system.spawn(tokens(), name="tokens")
conn = system.spawn(connection(lines, marks, service), name="conn")
# Too early. The connection is not open, and this state does not
# handle a Send at all.
conn.tell(Send(payload="hello"))
conn.tell(Connect())
# Also too early, and refused by a different state for its own reason.
conn.tell(Send(payload="hello again"))
await marks[3].wait()
# The token arrives through the adapter, and the connection is ready.
await marks[4].wait()
conn.tell(Send(payload="hello"))
conn.tell(Close())
conn.tell(Send(payload="too late"))
await marks[8].wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Two systems on a link
Switch remoting on, turn another system's address into a ref, and send. The
actors do not change at all. Compare this with hello_world above: what is
different is the settings and the one resolve.
"""The same greeting as `hello_world`, with a socket in the middle.
Concepts: switching remoting on with `RemoteSettings`, turning another system's
address into a ref with `resolve`, sending across the association, and getting
the answer back through a `reply_to` that crossed the wire.
Compare it to `hello_world` line by line. The actors are the same: one greets,
one listens, and the request carries the address for the reply. What changed
is outside them, in the settings and in the one `resolve`. That is what
location transparency gives you. It is worth being clear about what it does
not give you. The failure model is different, because a network is in the
middle. Placement is not uniform, because the greeter is started on its own
node. And the message the greeter receives is equal to what was sent rather
than the same object, because it was rebuilt from JSON on the other side.
Both 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 node name in front of every line. The request
is written by `home`, handled by `away`, and the reply arrives back at `home`
without either actor knowing there is a link between them.
Run it with `uv run python -m tapio_examples.two_nodes`.
"""
import asyncio
from collections.abc import Callable
from tapio import (
ActorSystem,
Behavior,
Behaviors,
Message,
RemoteSettings,
TapioSettings,
register_message,
)
from tapio.actor import ActorContext, ActorRef
from tapio.remote.address import format_ref
__all__ = ["Greet", "Greeted", "main"]
@register_message()
class Greeted(Message):
"""Sent back to whoever asked for the greeting."""
whom: str
@register_message()
class Greet(Message):
"""A request to greet someone, carrying the address for the reply.
Both message types are registered, because a type key on a frame is a
registry key and never an import path. A peer decodes what it has been
told about, and imports nothing to find out what an unknown name might
have meant.
"""
whom: str
reply_to: ActorRef[Greeted]
def node() -> TapioSettings:
"""Settings for a system that listens on a loopback port the OS picks.
Port 0 keeps the example free of numbers somebody has to keep unique.
Loopback is the default, and it is the one bind address that needs no
shared secret. A port that accepts frames naming actor paths and message
types is a serious surface, so binding anywhere else without a secret
refuses to start.
Returns:
The settings.
"""
return TapioSettings(remote=RemoteSettings(bind_port=0))
def greeter(record: Callable[[str], None]) -> Behavior[Greet]:
"""Build the greeter, which is exactly the local one.
Args:
record: Where to write the greeting, so the example can be asserted.
Returns:
The behavior to spawn.
"""
async def on_greet(ctx: ActorContext[Greet], message: Greet) -> Behavior[Greet]:
ctx.log.info("hello, %s!", message.whom)
record(f"away: hello, {message.whom}!")
# An ordinary tell. The ref came off the wire, so this reply crosses
# back over the same association, and nothing here had to know that.
message.reply_to.tell(Greeted(whom=message.whom))
return Behaviors.same()
return Behaviors.receive(on_greet)
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the actors produced, in the order they produced them.
"""
lines: list[str] = []
done = asyncio.get_running_loop().create_future()
async def on_greeted(message: Greeted) -> Behavior[Greeted]:
lines.append(f"home: {message.whom} has been greeted")
done.set_result(None)
return Behaviors.same()
async with (
ActorSystem("away", node()) as away,
ActorSystem("home", node()) as home,
):
hello = away.spawn(greeter(lines.append), name="greeter")
listener = home.spawn(Behaviors.receive_message(on_greeted), name="listener")
# In a real deployment this string comes from configuration or from a
# message, not from the other system's own ref, because the two are in
# different processes. Here they are not, so the address is read from
# the ref that spawn returned.
address = format_ref(away.address, hello.path)
lines.append(f"home: resolving {address}")
remote = await home.resolve(address, expect=Greet)
remote.tell(Greet(whom="world", reply_to=listener))
await done
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
The port is bound while the system is being constructed, so the address a ref writes down is settled before any ref is handed out, and a configuration that would listen beyond loopback with no shared secret fails to start rather than failing to be secure.
resolve dials nothing. The first send through the ref creates the
association, and the dial happens behind it, so the call does not wait for a
peer that may be down and a tell to one that never answers dead-letters
instead of hanging. The ref is bound to the peer and not to a link, so it
keeps working after a link fails.
What crosses a link is not quite what crosses a mailbox. Delivery is
at-most-once and FIFO per association, with no acks and no retries. A message
rebuilt from JSON is equal to what was sent and never the same object. And a
type key on a frame is a registry key, so both ends have to
@register_message() what they exchange. An unknown key is a dead letter
naming the key, and nothing is imported to find out what it meant.
When a node stops answering
Watching an actor on another node is the same call as watching one next door, and it is how you depend on something in another process without asking it whether it is still alive:
"""Watching an actor on another node, and rebuilding when that node dies.
Concepts: `ctx.watch` on a ref that came from `resolve`, `Terminated` arriving
because a whole system went away, and a coordinator that reacts by rebuilding
the work somewhere it can still reach.
The watch is the same call as in `death_watch`, on a ref that happens to point
at another process. That is the whole of the API difference. The difference
that matters is in what the signal means. Locally, `Terminated` is a fact: the
actor stopped. Across a link it is a conclusion: this node stopped hearing
from that one. Here the peer really did terminate, so the conclusion is right,
but nothing in the signal says which of the two it was, and nothing can.
Supervision does not cross the wire, so the coordinator is not the remote
worker's supervisor and never sees it fail. It sees it disappear. That is the
supported way to depend on an actor somewhere else: watch it, and have a plan
for the day it stops answering.
What to watch in the output: the third line. The coordinator does not retry
against the node that is gone. It starts a worker it owns and finishes the
job there, which is the difference between a failover and a stall.
Run it with `uv run python -m tapio_examples.node_failure`.
"""
import asyncio
from tapio import (
Behavior,
Behaviors,
Message,
Signal,
Terminated,
register_message,
)
from tapio.actor import ActorContext, ActorRef
from tapio.remote.address import format_ref
from tapio.testkit import two_nodes
__all__ = ["Assign", "Done", "Job", "main"]
@register_message()
class Done(Message):
"""A finished job, and which node finished it."""
item: int
where: str
@register_message()
class Job(Message):
"""A unit of work, carrying the ref its result goes back to."""
item: int
reply_to: ActorRef[Done]
class Assign(Message):
"""Tell the coordinator to get one job done, wherever it can.
Not registered, because it never leaves the node that sends it. Only what
crosses a link needs a wire key.
"""
item: int
def worker(where: str) -> Behavior[Job]:
"""Build an actor that does the work and says where it was done.
Args:
where: The node's name, so the output shows which one ran it.
Returns:
The behavior to spawn.
"""
async def on_job(message: Job) -> Behavior[Job]:
message.reply_to.tell(Done(item=message.item, where=where))
return Behaviors.same()
return Behaviors.receive_message(on_job, msg_type=Job)
def coordinator(
remote: ActorRef[Job],
lines: list[str],
finished: asyncio.Event,
lost: asyncio.Event,
) -> Behavior[Assign | Done]:
"""Build the actor that hands out work and survives losing its worker.
Args:
remote: The worker on the other node, to start with.
lines: Where to record what happened.
finished: Set whenever a job comes back.
lost: Set when the remote worker is gone and a local one has replaced it.
Returns:
The behavior to spawn.
"""
def build(ctx: ActorContext[Assign | Done]) -> Behavior[Assign | Done]:
# One call, and from here on this coordinator cannot be left holding a
# ref to a node that is no longer there without knowing it.
ctx.watch(remote)
current = remote
# The coordinator takes a wider protocol than the worker knows how to
# send, so what goes in `reply_to` is an adapter ref: an `ActorRef[Done]`
# that delivers into this actor. It is addressable like the actor
# behind it, so a result crossing the link finds its way back.
answers: ActorRef[Done] = ctx.message_adapter(lambda done: done, Done)
async def on_message(message: Assign | Done) -> Behavior[Assign | Done]:
nonlocal current
match message:
case Assign(item=item):
# An ordinary tell. Whether it crosses a link is decided
# by which ref this is, and this code does not ask.
current.tell(Job(item=item, reply_to=answers))
case Done(item=item, where=where):
lines.append(f"home: job {item} done by {where}")
finished.set()
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Assign | Done], signal: Signal
) -> Behavior[Assign | Done]:
nonlocal current
if isinstance(signal, Terminated):
lines.append("home: the away node is gone, rebuilding here")
# A child of this actor, so it is supervised by the one that
# depends on it. That is what could never be true of the
# worker on the other node.
current = ctx.spawn(worker("home"), name="worker")
lost.set()
return Behaviors.same()
return Behaviors.receive_message(on_message, on_signal=on_signal)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the coordinator produced, in order.
"""
lines: list[str] = []
finished, lost = asyncio.Event(), asyncio.Event()
async with two_nodes(alpha="home", beta="away") as nodes:
here, there = nodes.alpha, nodes.beta
hand = there.spawn(worker("away"), name="worker")
remote = await here.resolve(format_ref(there.address, hand.path), expect=Job)
boss = here.spawn(coordinator(remote, lines, finished, lost), name="boss")
boss.tell(Assign(item=1))
await finished.wait()
finished.clear()
# The whole node goes away, not just the actor. Every watcher of every
# ref on it hears the same thing.
await there.terminate()
await lost.wait()
boss.tell(Assign(item=2))
await finished.wait()
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
What the signal means is not the same, and this is the one place where the
difference is worth understanding before you need it. Locally, Terminated
says an actor stopped, which is a fact. Across a link it says this node
stopped hearing from that one, which is a conclusion. A partition, a long
pause and a dead process all look identical from a single node.
So tapio does what a single node can do, and says so plainly. Each association
heartbeats. When nothing has arrived for unreachable_after, every local
watcher of an actor over there is told Terminated, a PeerUnreachable event
is published on system.events, and the address is quarantined: sends to
it dead-letter and nothing dials it again.
Recovery is explicit, never automatic. Watchers have already been told that
live actors are gone, so a link coming quietly back would leave two nodes
holding contradictory beliefs with nothing to notice it. clear_quarantine
says this node is willing to talk to that peer again, and remote.reconnect
dials.
The example that shows all of it is the uncomfortable one, with both nodes alive and each convinced the other has died:
"""Two live nodes, a broken network, and both of them wrong about the other.
Concepts: the failure detector, `PeerUnreachable` on the system event stream,
quarantine, and `remote.reconnect` as the explicit repair.
This is the uncomfortable example, and it is here because the behaviour it
shows is the one people meet in production and do not expect. Nothing dies.
Both nodes run the whole time, both keep processing their own work, and the
only thing that breaks is the network between them. Each one then concludes
that the other is gone, tells its watchers so, and stops sending. Both
conclusions are wrong, and both are the best a single node can do.
The fix for this is a quorum: enough nodes to hold a vote, so that a minority
of a partition can find out that it is the minority and stand down. That is
clustering, and it is not in this version. What is here instead is a set of
defaults chosen so that being wrong is recoverable: fail fast, freeze the
address, and let a person or a supervisor decide when to try again.
What to watch in the output: the two blocks of node lines, printed side by
side. `home` says away is gone. `away` says home is gone. Both are still
answering their own callers while they say it. Then note that healing the
network on its own repairs nothing: the last two lines only happen because
somebody asked for them.
Run it with `uv run python -m tapio_examples.partition`.
"""
import asyncio
from datetime import timedelta
from tapio import (
ActorSystem,
Behavior,
Behaviors,
Message,
Signal,
Terminated,
register_message,
)
from tapio.actor import ActorContext, ActorRef
from tapio.actor.events import Subscription
from tapio.remote.address import format_ref
from tapio.remote.failure import PeerUnreachable
from tapio.testkit import two_nodes
__all__ = ["Poke", "Poked", "main"]
@register_message()
class Poked(Message):
"""Proof that the actor on the other node is still working."""
by: str
@register_message()
class Poke(Message):
"""A message with a reply address, to show the link working before it breaks."""
by: str
reply_to: ActorRef[Poked]
def steady(lines: list[str], node: str, answered: asyncio.Event) -> Behavior[Poke]:
"""Build an actor that answers, and keeps answering through everything.
Args:
lines: Where to record what it did.
node: The node it runs on, for the output.
answered: Set once it has answered something.
Returns:
The behavior to spawn.
"""
async def on_poke(message: Poke) -> Behavior[Poke]:
lines.append(f"{node}: poked by {message.by}, still working")
message.reply_to.tell(Poked(by=node))
answered.set()
return Behaviors.same()
return Behaviors.receive_message(on_poke, msg_type=Poke)
def mourner(
remote: ActorRef[Poke], lines: list[str], node: str, bereaved: asyncio.Event
) -> Behavior[Poked]:
"""Build an actor that watches the other node's actor and reports its death.
Args:
remote: The actor on the other node.
lines: Where to record what it was told.
node: The node it runs on, for the output.
bereaved: Set when `Terminated` arrives.
Returns:
The behavior to spawn.
"""
def build(ctx: ActorContext[Poked]) -> Behavior[Poked]:
ctx.watch(remote)
async def on_message(message: Poked) -> Behavior[Poked]:
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Poked], signal: Signal
) -> Behavior[Poked]:
if isinstance(signal, Terminated):
# It is not dead. It is on the other side of a partition,
# answering somebody else. Nothing in this signal says so, and
# nothing could.
lines.append(f"{node}: told that {signal.ref.path} has stopped")
bereaved.set()
return Behaviors.same()
return Behaviors.receive_message(on_message, on_signal=on_signal)
return Behaviors.setup(build)
def watch_events(system: ActorSystem, lines: list[str], node: str) -> Subscription:
"""Report every peer this node gives up on.
Args:
system: The node to subscribe to.
lines: Where to record what it published.
node: The node's name, for the output.
Returns:
The subscription, so the example can stop listening before the two
systems shut down. A node going away makes its peer unreachable too,
and that one is not news.
"""
def record(event: PeerUnreachable) -> None:
verdict = "quarantined" if event.quarantined else "link ended"
lines.append(f"{node}: gave up on {event.peer}, {verdict}")
return system.events.subscribe(PeerUnreachable, record)
async def main() -> list[str]:
"""Run the example.
Returns:
Both nodes' lines, home's first, so the two views can be read side by
side.
"""
home_lines: list[str] = []
away_lines: list[str] = []
home_answered, away_answered = asyncio.Event(), asyncio.Event()
home_bereaved, away_bereaved = asyncio.Event(), asyncio.Event()
# Short enough for an example. In production this is seconds, and it has
# to be comfortably longer than the peer's heartbeat interval or an
# ordinary slow moment reads as a dead node.
async with two_nodes(
alpha="home",
beta="away",
unreachable_after=timedelta(milliseconds=500),
heartbeat_interval=timedelta(milliseconds=20),
) as nodes:
here, there = nodes.alpha, nodes.beta
listening = [
watch_events(here, home_lines, "home"),
watch_events(there, away_lines, "away"),
]
here_actor = here.spawn(steady(home_lines, "home", home_answered), "steady")
there_actor = there.spawn(steady(away_lines, "away", away_answered), "steady")
to_there = await here.resolve(
format_ref(there.address, there_actor.path), expect=Poke
)
to_here = await there.resolve(
format_ref(here.address, here_actor.path), expect=Poke
)
here_watcher = here.spawn(
mourner(to_there, home_lines, "home", home_bereaved), "mourner"
)
there_watcher = there.spawn(
mourner(to_here, away_lines, "away", away_bereaved), "mourner"
)
to_there.tell(Poke(by="home", reply_to=here_watcher))
to_here.tell(Poke(by="away", reply_to=there_watcher))
await away_answered.wait()
await home_answered.wait()
# Nothing dies here. The two nodes simply stop being able to hear each
# other, which is the case no single node can tell from the other one.
nodes.partition()
await home_bereaved.wait()
await away_bereaved.wait()
# Both are still doing their own work, each while believing the other
# is gone. Poking locally proves it.
home_answered.clear()
away_answered.clear()
here_actor.tell(Poke(by="home itself", reply_to=here_watcher))
there_actor.tell(Poke(by="away itself", reply_to=there_watcher))
await home_answered.wait()
await away_answered.wait()
# The network is fine again, and nothing reconnects. Both nodes have
# already told their watchers that live actors are gone, so quietly
# resuming would leave them holding contradictory beliefs with no way
# to notice.
nodes.heal()
home_lines.append("home: network repaired, and still no association")
there.remote.clear_quarantine(here.address) # type: ignore[union-attr]
await here.remote.reconnect(there.address) # type: ignore[union-attr]
home_lines.append("home: reconnected, because somebody decided to")
for subscription in listening:
subscription.unsubscribe()
lines = home_lines + away_lines
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Both nodes are wrong, and both are locally correct. Fixing that needs enough nodes to hold a vote, so that the minority side of a partition can discover that it is the minority. That is clustering, and it is not in this version. What is here instead is a default chosen to be recoverable: fail fast, freeze the address, and let a person or a supervisor decide when to try again. For request/response and work distribution, wrongly deciding a peer is dead costs a retry. Waiting forever costs availability.
ask works across a link too, and it has one failure a local ask does not:
"""Asking a question of an actor on another node, and the two ways it fails.
Concepts: `ask` across an association, the reply finding its way back through
`/system/promises`, and `AskTargetUnreachable` told apart from
`AskTimeoutError`.
The call is the same one `ask_timeout` makes locally. What changes is the
failures. A local ask has two: nobody answered in time, or the actor stopped.
A remote ask has a third, and it is the one worth understanding. The peer can
become unreachable, which means this node stopped hearing from it and decided
that it was gone. The actor over there may be perfectly healthy on the other
side of a partition.
The two failures are different errors on purpose. A timeout says the peer was
there and slow, so waiting longer might help. Unreachable says this node has
given up on the peer, so waiting will not help and retrying somewhere else
might.
What to watch in the output: the third and fourth lines. Both asks failed, in
about the same amount of time, for entirely different reasons. The last line
is the point of the whole example: the actor that "disappeared" answers a
question the moment the network is repaired.
Run it with `uv run python -m tapio_examples.remote_ask`.
"""
import asyncio
from datetime import timedelta
from tapio import (
Behavior,
Behaviors,
Message,
register_message,
)
from tapio.actor import ActorContext, ActorRef
from tapio.errors import AskTargetUnreachable, AskTimeoutError
from tapio.remote.address import format_ref
from tapio.testkit import two_nodes
__all__ = ["Answer", "Ask", "main"]
@register_message()
class Answer(Message):
"""What the oracle says."""
question: str
answer: str
@register_message()
class Ask(Message):
"""A question, and the ref the answer goes back to.
`ask` builds the second field for you: the ref it passes is a promise on
the asking node, addressed under `/system/promises` so that an answer
coming back over the link finds the future somebody is awaiting.
"""
question: str
reply_to: ActorRef[Answer]
def oracle() -> Behavior[Ask]:
"""Build an actor that answers questions, except the ones it ignores.
Returns:
The behavior to spawn.
"""
async def on_ask(ctx: ActorContext[Ask], message: Ask) -> Behavior[Ask]:
if message.question == "the meaning of life":
# Some questions take longer than anyone is prepared to wait.
ctx.log.info("declining to answer %r", message.question)
return Behaviors.same()
message.reply_to.tell(Answer(question=message.question, answer="42"))
return Behaviors.same()
return Behaviors.receive(on_ask)
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the two nodes produced, in the order they produced them.
"""
lines: list[str] = []
# A short window, so the example runs in well under a second. Production
# values are seconds, not milliseconds: the window has to be long enough
# that an ordinary slow moment is not read as a dead node.
async with two_nodes(
alpha="asker",
beta="answers",
unreachable_after=timedelta(milliseconds=500),
heartbeat_interval=timedelta(milliseconds=20),
) as nodes:
here, there = nodes.alpha, nodes.beta
sage = there.spawn(oracle(), name="oracle")
remote = await here.resolve(format_ref(there.address, sage.path), expect=Ask)
answer = await remote.ask(
lambda reply_to: Ask(question="six by seven", reply_to=reply_to),
expect=Answer,
)
lines.append(f"asker: six by seven is {answer.answer}")
try:
await remote.ask(
lambda reply_to: Ask(question="the meaning of life", reply_to=reply_to),
expect=Answer,
timeout=timedelta(milliseconds=100),
)
except AskTimeoutError:
# The peer is there and nobody answered. Asking again later is a
# reasonable thing to do.
lines.append("asker: no answer in time, and the node is still there")
# Now the network, rather than the actor, is the problem. The oracle
# keeps running on its own node throughout.
nodes.partition()
try:
await remote.ask(
lambda reply_to: Ask(question="six by seven", reply_to=reply_to),
expect=Answer,
timeout=timedelta(seconds=30),
)
except AskTargetUnreachable:
# Note the deadline above: thirty seconds, and the ask failed in a
# fraction of one. It failed on the peer, not on the clock.
lines.append("asker: the answering node is unreachable, so no waiting")
nodes.heal()
there.remote.clear_quarantine(here.address) # type: ignore[union-attr]
await here.remote.reconnect(there.address) # type: ignore[union-attr]
# Refs from before the quarantine name a session that is over, so the
# address is resolved again rather than reused.
again = await here.resolve(format_ref(there.address, sage.path), expect=Ask)
repaired = await again.ask(
lambda reply_to: Ask(question="six by seven", reply_to=reply_to),
expect=Answer,
)
lines.append(
f"asker: after reconnecting, six by seven is still {repaired.answer}"
)
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
AskTimeoutError says the peer was there and nobody answered in time, so
asking again may work. AskTargetUnreachable says this node has given up on
the peer, so it will not. Both fail immediately rather than waiting out the
deadline, which is what the death watch under the ask is for.
Testing any of this needs a network that can be broken on purpose, so the
testkit ships one. two_nodes() starts a pair on loopback ports the OS picks,
and partition(), heal(), drop() and delay() lose frames without
breaking anything real.
Starting an actor on another node
Everything above is location transparent: an actor holding a ref sends, asks and watches without knowing which node the target is on. Placement is not, and it is not by design. Starting an actor elsewhere is a different call, and it is awaited, because a round trip is happening.
Both nodes must be running the same code. A behavior is a closure, and a
closure does not cross a socket. What crosses is a key naming a factory and a
model holding its arguments, and the peer looks that key up in its own
registry. Nothing is imported to find out what an unknown key might have meant,
exactly as for a message type. A key the peer has never heard of comes back as
SpawnFailed(reason="unknown-factory"), which is what version skew between two
deployments looks like from the requesting side.
"""Asking another node to start an actor, and letting it supervise the actor.
Concepts: `@remote_behavior` and the factory registry, a spawner actor with an
allowlist, watching what comes back, and seeing the peer restart its own child
while this node is told nothing at all.
Placement is the one part of remoting that is deliberately not transparent.
Sending, asking and watching are the same calls whichever node the target is
on. Starting an actor elsewhere is a different call, and it is awaited, because
a round trip is happening and pretending otherwise would be the kind of
transparency that lies.
The reason is supervision. If the local parent supervised a remote child, then
every restart, stop and failure report would be a frame on a link that can go
silent halfway through the decision. So the tree stays inside one node: the
spawned actor is the spawner's child, supervised over there, and this node
holds a ref and watches it. That is a smaller contract, and it is the largest
one a network can keep.
What to watch in the output: the fourth line. The worker crashed, the peer
restarted it, and the job that was queued behind the crash was answered by the
new incarnation through the very same ref. The count of jobs handled went back
to one, which is the only trace of it here. Nothing was reported, because
nothing had to be: the restart was decided a process boundary away from the
actor rather than a network away.
Run it with `uv run python -m tapio_examples.remote_spawn`.
"""
import asyncio
from tapio import (
Behavior,
Behaviors,
Message,
Spawn,
Spawned,
SpawnFailed,
SpawnReply,
register_message,
remote_behavior,
spawner,
)
from tapio.actor import (
ActorContext,
ActorRef,
Signal,
SupervisorStrategy,
Terminated,
)
from tapio.remote.address import format_ref
from tapio.testkit import two_nodes
__all__ = ["Crash", "Double", "Doubled", "DoublerArgs", "Retire", "doubler", "main"]
class DoublerArgs(Message):
"""What the worker is built with.
An arguments model, not a closure. A behavior is a closure and a closure
does not cross a socket, so what travels is a key naming a factory and a
model naming its arguments. Both nodes have to be running the same code for
the key to mean anything, which is the sentence to remember about all of
this.
"""
factor: int = 2
@register_message()
class Doubled(Message):
"""An answer, and how much work this incarnation has done."""
n: int
handled: int
@register_message()
class Double(Message):
"""A number to multiply, and where the answer goes."""
n: int
reply_to: ActorRef[Doubled]
@register_message()
class Crash(Message):
"""Tells the worker to fail, so that somebody has to decide about it."""
@register_message()
class Retire(Message):
"""Tells the worker to stop, so that its watchers hear about it."""
@remote_behavior("doubler")
def doubler(args: DoublerArgs) -> Behavior[Double | Crash | Retire]:
"""Build a worker that multiplies, fails on request, and can be retired.
The supervision is declared here because here is the only place it can be.
A restart happens entirely on the node that runs the actor, so the strategy
has to be part of what that node builds.
Args:
args: What to multiply by.
Returns:
The behavior the spawner will start.
"""
def build(
ctx: ActorContext[Double | Crash | Retire],
) -> Behavior[Double | Crash | Retire]:
# Rebuilt on every restart, which is what makes the restart visible on
# the other node without anything being reported.
handled = 0
async def on_message(
message: Double | Crash | Retire,
) -> Behavior[Double | Crash | Retire]:
nonlocal handled
if isinstance(message, Crash):
msg = "the doubler fell over"
raise RuntimeError(msg)
if isinstance(message, Retire):
return Behaviors.stopped()
handled += 1
message.reply_to.tell(Doubled(n=message.n * args.factor, handled=handled))
return Behaviors.same()
return Behaviors.receive_message(on_message, msg_type=Double | Crash | Retire)
return Behaviors.supervise(Behaviors.setup(build)).on_failure(
SupervisorStrategy.restart(), on=RuntimeError
)
def overseer(
target: ActorRef[Message], lines: list[str], gone: asyncio.Future[None]
) -> Behavior[Retire]:
"""Build an actor that watches the worker on the other node.
Death watch is what replaces the parent-child link, and it is the whole of
what this node is promised. `Terminated` arrives when the worker stops,
when its node stops, and when the link to that node is given up on. All
three mean the same thing here: that worker is gone, ask for another.
Args:
target: The worker to watch.
lines: Where to write what happened.
gone: Resolved once the worker has been reported gone.
Returns:
The behavior to spawn.
"""
def build(ctx: ActorContext[Retire]) -> Behavior[Retire]:
ctx.watch(target)
async def on_message(message: Retire) -> Behavior[Retire]:
return Behaviors.same()
async def on_signal(
ctx: ActorContext[Retire], signal: Signal
) -> Behavior[Retire]:
if isinstance(signal, Terminated):
lines.append("orders: the worker is gone, so ask for another")
if not gone.done():
gone.set_result(None)
return Behaviors.same()
return Behaviors.receive_message(
on_message, msg_type=Retire, on_signal=on_signal
)
return Behaviors.setup(build)
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the two nodes produced, in the order they produced them.
"""
lines: list[str] = []
gone: asyncio.Future[None] = asyncio.get_running_loop().create_future()
async with two_nodes(alpha="orders", beta="compute") as nodes:
here, there = nodes.alpha, nodes.beta
# The spawner offers one factory and nothing else. An actor that will
# start anything registered, on request, is a capability handed to
# whoever can reach the port.
desk = there.spawn(spawner(offers=["doubler"]), name="spawner")
remote = await here.resolve(format_ref(there.address, desk.path), expect=Spawn)
reply = await remote.ask(
lambda reply_to: Spawn(
factory="doubler",
args=DoublerArgs(factor=2),
name="doubler-1",
reply_to=reply_to,
),
# Both answers, because a refusal is news to act on rather than a
# broken protocol. Asking for `Spawned` alone would turn one into
# an AskTypeError.
expect=SpawnReply,
)
if not isinstance(reply, Spawned):
msg = f"compute refused to start the worker: {reply}"
raise RuntimeError(msg)
worker = reply.ref
lines.append(f"orders: compute started {reply.name} at {worker.path}")
here.spawn(overseer(worker, lines, gone), name="overseer")
first = await worker.ask(
lambda reply_to: Double(n=6, reply_to=reply_to), expect=Doubled
)
lines.append(f"orders: 6 doubled is {first.n}, job {first.handled} for it")
# The crash and the job queue behind each other on the worker's own
# mailbox. The crash is supervised over there, the mailbox survives it,
# and the job is answered by the new incarnation through this same ref.
worker.tell(Crash())
second = await worker.ask(
lambda reply_to: Double(n=7, reply_to=reply_to), expect=Doubled
)
lines.append(f"orders: 7 doubled is {second.n}, job {second.handled} for it")
lines.append("orders: the count restarted, and nothing told me why")
refused = await remote.ask(
lambda reply_to: Spawn(factory="tripler", reply_to=reply_to),
expect=SpawnReply,
)
if isinstance(refused, SpawnFailed):
# What version skew looks like: a key crossed the wire and the peer
# has never heard of it. Nothing is imported to find out what it
# might have meant.
lines.append(f"orders: compute cannot start 'tripler' ({refused.reason})")
worker.tell(Retire())
await gone
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Three things in that example are the whole design.
The factory declares its own supervision, with Behaviors.supervise(...)
around what it returns. That is the only place it can be declared, because the
restart happens entirely on the node that runs the actor. If the local parent
supervised a remote child, every restart, stop and failure report would be a
frame on a link that can go silent halfway through the decision, and
supervision is the one thing in this library that has to be able to answer. So
the tree stays inside one node.
The requester watches instead of parenting. Terminated arrives when the actor
stops, when its node stops, and when the link to that node is given up on, and
all three are indistinguishable on purpose: they mean the same thing to the
requester, which is that this worker is gone and another one should be asked
for. That is a smaller contract than supervision, and it is the largest one a
network can keep.
A spawner offers named factories rather than the registry. An actor that will start anything registered, on request, is a capability handed to whoever can reach the port, and the port's threat model does not assume that is nobody. The allowlist is checked when the spawner is built, so a typo in it fails where it was written.
The one restriction worth knowing before you write an arguments model: it
cannot carry an ActorRef. Arguments are validated on the peer after the
factory key has been checked, deliberately outside the decode that resolves
refs, so a ref in them would have nothing to resolve against. Send the new
actor a message instead. You are holding a ref to it, and refs inside that
message resolve normally.
Backpressure across a link
await ref.offer(msg) on a remote ref waits for room in the sending node's
outbound buffer. That is honest local backpressure against a socket that is not
draining, and it is not backpressure from the receiving actor. A worker with a
large mailbox reads every frame the moment it arrives, so the buffer stays
empty, offer never waits, and the backlog builds up on the other node where
this one cannot see it.
Nothing in a fire-and-forget wire protocol can do better, so flow control is built out of messages, where the receiver is the one who knows:
"""Backpressure across a link, which the transport cannot give you.
Concepts: why `offer` is local backpressure and not end-to-end backpressure,
and the credit-based protocol that is. Workers say how much they will accept,
and the producer sends only what it has been granted.
`await ref.offer(item)` on a remote ref waits for room in *this* node's
outbound buffer. That is a real thing to wait on, and it is a socket that is
not draining rather than a worker that is falling behind. The two come apart
exactly when it matters: a worker with a huge mailbox reads every frame the
moment it arrives, so the buffer stays empty, `offer` never waits, and the
backlog piles up on the other node where this one cannot see it. Nothing in a
fire-and-forget wire protocol can do better, and pretending otherwise would be
the kind of transparency that lies.
So flow control is built out of messages, where the receiver is the one who
knows. Each worker grants the producer a number of items it is willing to have
outstanding. The producer sends that many and no more, and each finished item
grants one back. The grant is the backpressure, it is end to end, and it works
whatever the network is doing.
Compare it with `worker_pool`, which fans out over routees in one process. The
router there needs none of this: a full mailbox pushes back on the sender
directly, because the sender and the mailbox are on the same loop.
What to watch in the output: the third line. Twelve items were done and no
worker ever had more than the three it granted waiting on it. That number is
chosen by the workers and obeyed by the producer, and nothing in the transport
enforces it.
Run it with `uv run python -m tapio_examples.worker_pool_remote`.
"""
import asyncio
from collections.abc import Callable
from tapio import (
Behavior,
Behaviors,
Message,
Spawn,
Spawned,
SpawnReply,
register_message,
remote_behavior,
spawner,
)
from tapio.actor import ActorContext, ActorRef
from tapio.remote.address import format_ref
from tapio.testkit import two_nodes
__all__ = [
"Credit",
"Hello",
"Item",
"WorkerArgs",
"main",
"producer",
"sink",
"spawn_request",
]
ITEMS = 12
"""How much work there is to hand out."""
GRANT = 3
"""How many items a worker will have outstanding at once."""
class WorkerArgs(Message):
"""What a worker is built with: how much it is prepared to have in hand."""
grant: int = GRANT
@register_message()
class Credit(Message):
"""A worker saying how much more it will accept, and how much it has done.
It carries its own ref, so the producer knows which worker is speaking and
has somewhere to send the next item. That ref crosses the link in this
direction, and the items cross back through it in the other.
"""
worker: ActorRef["Hello | Item"]
n: int
done: int
@register_message()
class Hello(Message):
"""Introduces the producer to a worker, which answers with its first grant."""
reply_to: ActorRef[Credit]
@register_message()
class Item(Message):
"""One unit of work, and where to ask for the next one."""
n: int
reply_to: ActorRef[Credit]
@remote_behavior("sink")
def sink(args: WorkerArgs) -> Behavior[Hello | Item]:
"""Build a worker that grants credit and gives it back as it finishes.
Args:
args: How many items it will have outstanding at once.
Returns:
The behavior the spawner will start.
"""
def build(ctx: ActorContext[Hello | Item]) -> Behavior[Hello | Item]:
done = 0
async def on_message(message: Hello | Item) -> Behavior[Hello | Item]:
nonlocal done
if isinstance(message, Hello):
# The opening grant. Until this arrives the producer has been
# told nothing, so it sends nothing.
message.reply_to.tell(
Credit(worker=ctx.self_ref, n=args.grant, done=done)
)
return Behaviors.same()
done += 1
# One item finished, one slot free. This is the whole protocol.
message.reply_to.tell(Credit(worker=ctx.self_ref, n=1, done=done))
return Behaviors.same()
return Behaviors.receive_message(on_message, msg_type=Hello | Item)
return Behaviors.setup(build)
def producer(
workers: list[ActorRef["Hello | Item"]],
items: list[int],
finished: asyncio.Future[tuple[int, list[int]]],
) -> Behavior[Credit]:
"""Build the actor that hands work out, and never outruns its grants.
Args:
workers: The workers on the other node.
items: The work to hand out.
finished: Resolved with the peak outstanding count and how much each
worker did, once every item is done.
Returns:
The behavior to spawn.
"""
def build(ctx: ActorContext[Credit]) -> Behavior[Credit]:
queue = list(items)
credit = {worker.path: 0 for worker in workers}
sent = {worker.path: 0 for worker in workers}
done = {worker.path: 0 for worker in workers}
peak = 0
for worker in workers:
worker.tell(Hello(reply_to=ctx.self_ref))
async def on_credit(message: Credit) -> Behavior[Credit]:
nonlocal peak
at = message.worker.path
credit[at] += message.n
done[at] = message.done
while credit[at] > 0 and queue:
message.worker.tell(Item(n=queue.pop(0), reply_to=ctx.self_ref))
credit[at] -= 1
sent[at] += 1
# What the grant is actually bounding: items sent to a worker that
# it has not reported finishing.
peak = max(peak, max(sent[at] - done[at] for at in sent))
if sum(done.values()) == len(items) and not finished.done():
finished.set_result((peak, [done[worker.path] for worker in workers]))
return Behaviors.same()
return Behaviors.receive_message(on_credit, msg_type=Credit)
return Behaviors.setup(build)
def spawn_request(number: int) -> Callable[[ActorRef[SpawnReply]], Spawn]:
"""Build the request for one worker, given where the answer goes.
A named function rather than a lambda in a loop, because a lambda would
close over the loop variable and every request would ask for the same name.
Args:
number: Which worker this is, which becomes its name on the peer.
Returns:
What `ask` calls with the ref for the reply.
"""
def request(reply_to: ActorRef[SpawnReply]) -> Spawn:
return Spawn(
factory="sink",
args=WorkerArgs(grant=GRANT),
name=f"sink-{number}",
reply_to=reply_to,
)
return request
async def main() -> list[str]:
"""Run the example.
Returns:
The lines the producer wrote, in the order it wrote them.
"""
lines: list[str] = []
finished: asyncio.Future[tuple[int, list[int]]] = (
asyncio.get_running_loop().create_future()
)
async with two_nodes(alpha="orders", beta="compute") as nodes:
here, there = nodes.alpha, nodes.beta
desk = there.spawn(spawner(offers=["sink"]), name="spawner")
remote = await here.resolve(format_ref(there.address, desk.path), expect=Spawn)
workers: list[ActorRef[Hello | Item]] = []
for number in (1, 2):
reply = await remote.ask(spawn_request(number), expect=SpawnReply)
if not isinstance(reply, Spawned):
msg = f"compute refused to start a worker: {reply}"
raise RuntimeError(msg)
workers.append(reply.ref)
lines.append(
f"orders: {len(workers)} workers on compute, each granting {GRANT} "
"items at a time"
)
here.spawn(producer(workers, list(range(ITEMS)), finished), name="producer")
peak, split = await finished
lines.append(f"orders: the work split {split[0]} and {split[1]}")
lines.append(
f"orders: {ITEMS} items done, and never more than {peak} "
"outstanding at one worker"
)
lines.append(
"orders: the grant is the backpressure; offer would have waited on "
"this node's outbound buffer instead"
)
for line in lines:
print(line)
return lines
if __name__ == "__main__":
asyncio.run(main())
Each worker grants the producer a number of items it will have outstanding. The producer sends that many and no more, and each finished item grants one back. The grant is the backpressure, it is end to end, and it holds whatever the network is doing.
What the runtime gives you today
ActorSystem, with a/userguardian above everything you spawn.ctx.spawnandctx.spawn_anonymous, with names unique among live siblings.ref.tell, which never blocks, and validates the message against the recipient's declared type before it goes anywhere.await ref.offer(...)and bounded mailboxes, withFAIL,DROP_NEWandDROP_OLDESToverflow strategies.- Dead letters, subscribable, so a message that went nowhere can be observed rather than guessed at.
Behaviors.supervise(...).on_failure(...): resume, restart with backoff and a restart window, stop, escalate.ctx.watchandTerminated, plusPreRestartandPostStop.await ref.ask(...), with a required reply type, a deadline, and a fast failure when the target stops rather than a wait for the deadline.Behaviors.with_timers(...): single, fixed-delay and fixed-rate timers, cancelled by the cell on restart and on stop.Behaviors.with_stash(...): a bounded buffer andunstash_all, which replays in arrival order ahead of newer traffic.ctx.message_adapter(...): a ref that translates another protocol into yours, in your actor, where a failed translation is your decision.Routers.pool(...): round-robin fan-out over routees that are the router's own children, shrinking as they stop.ctx.log, which tags every record with the actor's path.await system.terminate(), which drains the tree bottom-up against a single deadline and cancels anything still wedged when it passes.@register_message(), which gives a message type the key a frame names it by. A key is a registry lookup and never an import path, so a type name that arrived on a socket can never become an import.system.addressand refs that write themselves down in full, address and incarnation uid included, and resolve back to live refs insidewith system.as_deserialization_context():.system.deliver_frame(...), the receiving half of remoting. Everything a peer can get wrong is decided here and becomes a dead letter naming the peer: an unreadable frame, an unknown type key, a payload that will not validate, an actor that has stopped, a stale incarnation, and a message the recipient does not accept.RemoteSettings, a TCP link with a version check and a shared-secret handshake, optional TLS, and one association per peer with its own bounded outbound buffer.await system.resolve(uri, expect=...)andctx.resolve(...), which turn another system's address into an ordinary ref.ctx.watchon a ref that points at another node, andawait ref.ask(...)across a link, withAskTargetUnreachabletold apart from a timeout.- A heartbeat failure detector, quarantine,
PeerUnreachableonsystem.events, andremote.reconnectas the one way back. tapio.testkit.two_nodes(), with link faults for partitions, dropped frames and delays, so the failure paths are tested rather than described.await ctx.run_blocking(fn, ...), which keeps a blocking call off the loop and out of every other actor's way. See Blocking calls, the page for the mistake that is hardest to see in production.- Supervision, in the detail it deserves: see Supervision for the four decisions and exactly what a restart keeps, and The life of an actor for signals, watching and shutdown.
- The remoting pages: Overview has the table of what is and is not transparent, When a node stops answering has the failure that can be wrong about a live peer, and Security has what the transport is and is not designed for.
tapio.testkit:TestProbe,BehaviorTestKit, pytest fixtures that arrive through an entry point, and the leak assertions. See Testing.@remote_behavior()andspawner(offers=[...]), which start an actor on another node without any supervision crossing the wire. The requester watches what it gets back, a refused request comes back asSpawnFailedwith a reason, and both nodes have to be running the same code.