Remoting
tapio can address actors on another node. Two systems that have configured remoting can send to, ask, and watch each other's actors, using the same calls as if the target were in the same process.
That last claim is the one worth being careful about, so this page starts with exactly how far it goes.
What is and is not transparent
| Local | Remote | |
|---|---|---|
tell never blocks, never raises about the recipient |
yes | yes |
| Ordering, per sender and recipient pair | FIFO | FIFO while the association lives |
| Delivery | at most once | at most once |
| Message identity | received is sent |
received == sent |
| Undeliverable goes to dead letters | yes | yes, on whichever side noticed |
watch fires Terminated when the actor stops |
always right | also fires on unreachable, which can be wrong |
ask failure modes |
timeout, target terminated | plus target unreachable |
Backpressure from offer |
the receiving mailbox | the local outbound buffer only |
| Supervision of a child | yes | never crosses the wire |
ctx.spawn places the actor |
yes | no, placement is a different call |
| Message types | anything a Pydantic field accepts | JSON-representable and registered |
| Latency and failure | in process | a network is in the middle |
Location transparency in tapio means addressing is uniform. An actor holding a ref sends, asks and watches without knowing which node the target is on. It deliberately stops in two places, and both are in bold above.
Failure is not uniform, because a network is in the middle and no API hides that. The unreachability page is about the one entry that can lie to you.
Placement is not uniform either. Starting an actor on another node is a different call, and it is awaited.
Switching it on
Remoting is off unless it is configured, and when it is configured the port is bound while the system is being constructed. That is what settles the canonical address before any ref can write itself down, and it is what makes a configuration that would listen to the world fail to start rather than fail to be secure.
"""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())
A ref that crosses a link is written down as its full address, and resolve
turns that string back into a ref. expect= is how the caller says what it
believes is at the other end, and the claim is checked against the actor's
real message type where every claim about a peer is checked: on the receiving
node.
Asking across a link
ask works unchanged, and the reply comes back to a promise actor under
/system/promises, addressed like anything else.
"""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())
What is new is a third way for it to fail. Locally an ask times out or the
target stops; remotely the peer can also be unreachable, and
AskTargetUnreachable says so at once instead of waiting out a timeout that
was never going to be met. Telling those apart is what lets a caller retry
somewhere else rather than retrying into silence.
Backpressure does not cross a link
await ref.offer(msg) 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, not a worker that is falling behind. The two come apart exactly
when it matters: a worker with a large mailbox reads every frame as 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, so end-to-end 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())
The grant is the backpressure. It is chosen by the worker, obeyed by the producer, and nothing in the transport enforces it.
Starting an actor on another node
There is no placement setting, and no parent-child relationship across a link. An actor is started elsewhere by asking a spawner there to start it locally, and what crosses the wire is a key naming a registered factory plus a model holding its arguments.
"""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())
The reason placement is not transparent is supervision. If a local parent supervised a remote child, every restart, stop and failure report would be a frame on a link that can be quarantined 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 spawned actor is the spawner's child, supervised over there at in-process latency, and the requester holds a ref and watches it.
Both nodes have to be running the same code. A behavior is a closure and a
closure does not cross a socket. A factory key the peer has never heard of
comes back as SpawnFailed(reason="unknown-factory"), which is what a version
skew between two deployments looks like from the requesting side.
What actually goes over the wire
A frame is a four-byte big-endian length followed by a JSON object:
{"v": 1, "to": "/user/checkout/session-7#f3a1c8",
"from": "tapio://web@10.0.0.9:25520",
"t": "orders.protocol.Reserve",
"p": {"sku": "X-1", "qty": 2}}
to omits the address, because a frame arriving on an association is by
definition addressed to the node that received it. from is the sending
system rather than a sending actor: a tell carries no sender, so there is
none to name. It is a diagnostic, so a dead letter can say which node produced
a frame it could not read. Replies go to the reply_to a message carries,
which is a complete ref and the only thing that ever addresses an actor.
v is the wire protocol version, and the handshake pins it. It is
deliberately not the tapio version: a release that does not change the wire
does not change this number, so a fleet can roll from one release to the next
instead of stopping to swap every node at once.
t is a registry key, never an import path. Resolving a dotted name that
arrived on a socket into an importable object is remote code execution.
@register_message() is what puts a class in the registry, and a key nobody
registered becomes a dead letter naming the key. Nothing is imported to find
out what it might have meant.
A message crossing a link is rebuilt from JSON, so it is == to what was sent
and never is it. Locally, a tell delivers the very object that was passed.
Both facts are worth knowing when writing a test.
Delivery, and what it does not promise
Delivery across a link is at most once, FIFO per association, which is the same guarantee as a local send. There are no acknowledgements and no retries.
That is deliberate. A retry is only safe when the receiver can tolerate a repeat, and the library does not know which of your messages those are. Upgrading at-most-once to at-least-once belongs in your protocol, where an idempotency key or a sequence number can be attached by someone who knows what the message means.