CU28 — A client talks to a broker

from somatize import Broker, Graph, Worker

g = Graph.somatize(Encode() >> Classify().at("gpu-box"))
g.forward(x, broker=Broker.embedded({"gpu-box": Worker.at("gpu-box:7000")}))

Status: closed. A crate in soma-fabric, soma-fabric-broker, with 45 tests — 16 of the thread, 13 of the handle, 11 of the protocol, 5 of the paths — plus test_remote.py (41) and four in soma-core/tests/unit/placement.rs.

The question: what changes for somebody who has no platform?

The answer this use case exists to give is which broker, and that is a URL. Not a degraded mode with its own code, not a branch on whether there is an account: the same call, the same protocol, the same handle.

deployment	what it is	what it adds
embedded	in the client’s own process	nothing. It is what makes soma work alone
local	a process on a head node	reachable by more than one client
platform	ours	authentication, pairing, leases, metering

Only the first exists today, and there is deliberately no Broker trait: a trait with one implementor is the shape this project was started to stop writing. It arrives with the second.

A `Worker` stops being a connection and becomes a declaration

That is the change a caller can see, and it has a consequence worth stating rather than discovering: an unreachable host now fails when it is needed rather than when it is named. Worker.at("bad:7000") used to fail in the constructor; now it fails inside the run, from the slice that wanted it. Better behaviour — a graph names hosts a run may never reach, and a branch not taken is a worker not needed — and a change, not a side effect.

Ask eagerly, connect lazily

The two costs pull in opposite directions and both are real, so the split is between them:

Asking where a host is costs tens of bytes, and has to happen before the first node runs, because what gets packed for a host depends on which hosts turn out to be the same place.
Connecting costs a socket, a process, or both.

So a rendezvous is asked for once and remembered by the session, and the wire it describes is opened the first time somebody actually sends work down it. The ask happens once however it was triggered: a client that resolved every host up front to decide what to pack finds the answers already there when the run reaches them.

Two names for one place are one wire, and it is not a nicety

A worker has one catalog, and half of one is a different catalog. Provisioning the same process twice, once per host name, replaces what it had live and takes every activation over there with it — a run that quietly loses its state, not an extra socket.

Only the session sees more than one host at a time, so only the session can know. But what to pack is Python’s, because it is the half that knows what a cloudpickle is. The two are reconciled without either learning the other’s business: wire_token(host) answers with opaque bytes that are equal exactly when two hosts are one wire. Python groups by equality and never finds out what a path is, nor when two of them count as one.

What decides it is Path::shared, and the asymmetry in it is the point: an address is an identity — the same host and port is the same process — while a command is a thing to run, and running it twice gives two of them. Today’s suite stands up two hosts from an identical argv and requires two processes.

The ladder of four, of which two can be answered

This is the one place the crate builds ahead of its consumer, on purpose:

same process — nothing is transferred
shared mount — a path is written and read; free, and a cluster has one
direct socket — one crossing, lowest latency. The broker steps out
relayed — streamed through the broker, no disk and no durability

Two of them can be answered today. All four are in the message, because the alternative is that adding a rung later changes Reply::Met — and a message that changes is a version that changes for everybody. The ladder is the design; what arrives later is the probing that chooses, not the vocabulary.

An object store is not one of the rungs. It is where durable things live, and renting eleven nines for an activation that lives forty milliseconds is the wrong shape before it is a bill.

The rung that transfers nothing answers with a SlotId and never a handle, which looks like a needless indirection and is not: every message here has to survive a round trip through bytes, including the ones an embedded broker answers without leaving the process. It costs nothing and buys a conformance suite that can round-trip every message there is.

The embedded broker is a thread, and it really serializes

Both of those look like waste and neither is, because control and cargo go by different routes:

	what crosses	how much	how often
the rendezvous	a path	tens of bytes	once per host per session
the wire next door	an activation	megabytes	once per forward, fifty thousand forwards

A run across four workers is nine messages — one greeting, four rendezvous, four goodbyes — some tens of microseconds, once, outside the loop. The broker is in the first row and steps out of the second, so being honest here is not measurable there. And being honest buys the thing that matters: the protocol is exercised for real from the first day, by a round trip that actually happens, before any broker exists outside a process. A protocol whose only implementation never serialized anything would be a protocol nobody had tested.

The failure that type exists not to have is a client blocked forever on an answer that is never coming. Every channel operation maps to Unanswered::Gone and never to an unwrap, and that is pinned by a test rather than by care — which is why Embedded::served_by is public: without a way to stand up a desk that fails, the one failure mode worth testing is the one that cannot happen.

Which hosts does this graph name?

Placement::hosts(), the half of host_of that reads the other way, and it exists because of who asks. A client handed a dictionary of workers already knew the names — they were its keys. A client that talks to a broker does not.

Once each, because a host named by ten nodes is one rendezvous and not ten. And sorted, which is not tidiness: they come out of a HashMap, and iterating one gives a different order every run. That would make the order rendezvous are asked for — and so the order failures happen in — irreproducible, and this project has already paid for a nondeterministic order once, when an artifact’s id changed because the caller reordered a dictionary.

What is not honoured yet, said out loud

A Reply::Met can carry a good_for, and nothing enforces it. No broker issues one today — the embedded one has no policy — so enforcing it would be a mechanism with no tenant. The day one does, the enforcement belongs to the handle and not to the engine: it is the only thing that knows when the rendezvous was granted.

Questionnaire

The protocol survives meeting a binary that disagrees with it (broker/tests/unit/protocol.rs)

every question and every answer goes and comes back equal
a greeting from a version we do not speak is still readable
and is refused naming both numbers
a greeting that grew a field could not be read, which is why it must not grow
leftovers are as suspicious as missing bytes, and a truncated message is refused
an answer is not a question

The four paths, of which two can be answered (broker/tests/unit/path.rs)

all four cross, including the one that transfers nothing
a pipe and a socket are the same path
a command keeps its arguments in order
how long it is good for crosses as a duration

A broker on a thread, which must never become a hang (broker/tests/unit/embedded.rs)

the session stays open across rendezvous, and is greeted once
a desk that panics is reported and not waited on
and one that has fallen over stays fallen over instead of hanging
one thread for the broker, and not one per ask
dropping it ends its thread rather than leaving it behind
two threads reach one broker at once
bytes that are not a message are refused by the desk and are not fatal
a host it does not know is told what it does know

One host, standing in the engine’s hole (broker/tests/unit/reaching.rs)

building a handle asks the broker nothing
an unreachable host fails when it is needed and not when it is named
four hosts share one greeting, and a host is asked about once however many times it is wanted
a rendezvous nobody took is not let go of, and one that was taken is
a slice reaches a real worker through the broker, and the second reuses the wire without asking again
the paths the negotiation has not arrived for are refused by name

Two names for one place (broker/tests/unit/reaching.rs)

two hosts at one address share one wire
two hosts with the same command are two processes
and what is packed for them is packed once
two names for one place declared with different packing are refused, by name, saying what each of them asked for

Which hosts a placement names (soma-core/tests/unit/placement.rs)

they come back once each
a placement that sends nothing away names no hosts
the order does not depend on the order they were placed in
moving a node elsewhere leaves no ghost behind

What a client writes (soma-python/tests/test_remote.py)

a worker is declared with an address or a command, and declaring it starts nothing
a broker takes a dict from host to Worker, and says which host was wrong
two workers are two processes, and the artifact is sent only once
a worker that gets nothing is told nothing
a graph run in pieces keeps the worker it had
provision says out loud what forward says on its own

Two names for one place, which is one catalog (soma-python/tests/test_remote.py)

two names for one place are told once each, about one artifact holding both halves
while two addresses are two catalogs with half each
two names for one place packed differently is refused, naming both
a host the broker never heard of is left out and not raised over

Proved against an artifact and not against a wire, which is the half this side owns. The grouping was mutated — keyed by host name instead of by wire — and two of the four fail; the two that survive are the contrast rows, which is what they are for.

What is pending

session.rs has no test module, though it holds the rules that matter most — the ask remembered, the wire shared, the token. They are reached through reaching.rs, which is real coverage and not a file of its own.
The local and platform brokers, the path negotiation, the agent and the queue. They arrive with a consumer and not before.