Concept: The Connection Actor — Multiplexing One Socket

Kind: Concept.

The last chapter gave you a MessageStream: a typed, framed channel over one socket. That is the pipe. This chapter is about what sits on top of the pipe, and it is the piece that makes the whole cluster work — one small task that owns the socket and turns it into something the scheduler can treat exactly like a local worker.

Start from the problem, because the shape of the solution falls straight out of it. A worker advertises a capacity — say four — meaning it can run four jobs at once. So the coordinator wants to have four jobs in flight over one connection simultaneously: send Assign for job A, then B, then C, then D, without waiting for A's Result to come back first. The four results will then arrive in whatever order the jobs happen to finish — maybe C, then A, then D, then B. That is the entire difficulty in one sentence: many jobs share one socket, and their answers come back interleaved and out of order. Whoever sent job A has to be handed A's result and not C's.

Why one task must own the socket

The instinct is to let each dispatch call just use the socket directly — write the Assign, then read until the matching Result shows up. That falls apart immediately, and it is worth seeing why, because the failure is the reason the actor pattern exists. A TcpStream (and the MessageStream around it) is not something two tasks can safely share for reading. If job A's task and job C's task both call recv on the same socket, the frames race: A's task might read C's result and C's task might read A's, or the length-prefix reads from the two tasks interleave and corrupt the stream entirely. Sharing a socket across concurrent readers is a data race dressed up as a logic bug.

The fix is a discipline: exactly one task touches the socket. Everyone else talks to that task instead of to the socket. That one task is the connection actor — an actor in the plain sense, a task that owns some state (here, the socket and a table of who is waiting) and is the only thing allowed to mutate it. Other tasks send it requests over a channel; it serializes all socket access through its own single loop. No sharing, no races, and — the payoff — the actor is the one place that sees every frame come back, so it is the natural place to match each result to its waiter.

The three moving parts

The actor needs three things, and each answers one piece of the problem.

An mpsc receiver — the inbox of jobs to send. Many dispatch callers, one actor: that is a multi-producer, single-consumer channel by definition. Each caller sends the actor a job plus a private oneshot::Sender to answer on. The actor pulls jobs from this receiver and writes each as an Assign frame.

A oneshot per job — the private answer line. A oneshot channel carries exactly one value, once. When a caller dispatches a job, it makes a oneshot, keeps the receiver, and hands the sender to the actor. The caller then simply .awaits its receiver — it is parked until its answer arrives, and it is structurally impossible for it to be woken by someone else's result. This is the mechanism that solves out-of-order delivery: the ordering does not matter because each waiter has its own dedicated channel.

A HashMap<JobId, oneshot::Sender> — the table of who is waiting. When the actor sends Assign for a job, it stores that job's oneshot sender under the job's id. When a Result frame comes back, the actor looks up outcome.job_id in the map, removes the sender, and fires the result down it — waking exactly the right waiter. The map is the multiplexer: id in, correct waiter out.

And the loop that ties them together is a select! over two events: a new job arriving on the mpsc, or a frame arriving on the socket. One subtlety makes select! do real work here — a guard on the job arm:

tokio::select! {
    // Only pull new work while we are below the worker's capacity.
    maybe = jobs.recv(), if pending.len() < capacity => { /* send Assign, insert into map */ }
    frame = conn.recv() => { /* match Result back to its waiter, or handle close */ }
}

The if pending.len() < capacity is backpressure, for free. While the worker already holds capacity jobs, the actor stops selecting the job arm entirely — it will not pull another job off the mpsc. The mpsc then fills to its bound and dispatch's send blocks, which means the scheduler blocks trying to hand out more work to this worker. The worker's advertised capacity becomes a real, enforced limit, propagated all the way back to the scheduler through nothing but a select! guard and a bounded channel. No counter to decrement by hand, no semaphore — the pending.len() map and the guard are the whole mechanism.

The toy: a mini actor that matches replies by id

Here is exactly that shape, with the socket replaced by a pair of in-process channels so it runs anywhere. jobs is the mpsc inbox; conn_tx/conn_rx stand in for MessageStream's send and recv; pending is the id→waiter map; and the select! has the capacity guard. Two scenarios run: three jobs whose replies deliberately come back out of order (the worker answers job 2 before job 1), and a worker that dies mid-job so you can watch the cleanup path fire.

This is plain tokio — mpsc, oneshot, select! — so it runs as-is. The output below is real.

use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};

type JobId = u64;

/// The far side's frames, mirrored down to the one that matters here.
enum Frame {
    Result { job_id: JobId, answer: String },
}

/// A job handed to the actor plus the one-shot to answer it on. This pair is the
/// shape of the real `Dispatch = (Job, oneshot::Sender<Result<JobOutcome, _>>)`.
type Dispatch = (JobId, String, oneshot::Sender<Result<String, String>>);

/// One task owns the "connection" and multiplexes many in-flight jobs. `conn_rx`
/// stands in for `MessageStream::recv`; `conn_tx` for `conn.send(Assign)`.
async fn connection_actor(
    mut jobs: mpsc::Receiver<Dispatch>,
    mut conn_rx: mpsc::Receiver<Frame>,
    conn_tx: mpsc::Sender<(JobId, String)>,
    capacity: usize,
) {
    let mut pending: HashMap<JobId, oneshot::Sender<Result<String, String>>> = HashMap::new();

    loop {
        tokio::select! {
            // Only pull new work while below capacity — this is the backpressure.
            maybe = jobs.recv(), if pending.len() < capacity => {
                let Some((job_id, text, reply)) = maybe else { break };
                if conn_tx.send((job_id, text)).await.is_err() {
                    let _ = reply.send(Err("connection lost".into()));
                    break;
                }
                pending.insert(job_id, reply);
            }
            frame = conn_rx.recv() => {
                match frame {
                    Some(Frame::Result { job_id, answer }) => {
                        if let Some(reply) = pending.remove(&job_id) {
                            let _ = reply.send(Ok(answer)); // wake exactly this waiter
                        }
                    }
                    None => break, // the connection closed
                }
            }
        }
    }

    // Connection finished: fail every job still in flight, RETRYABLY, so the
    // caller re-dispatches it elsewhere. This is the crucial line.
    for (job_id, reply) in pending {
        println!("actor: failing in-flight job {job_id} retryably");
        let _ = reply.send(Err("worker connection lost".into()));
    }
}

#[tokio::main]
async fn main() {
    // --- scenario 1: three jobs, replies matched by id (out of order) ---
    {
        let (jobs_tx, jobs_rx) = mpsc::channel::<Dispatch>(8);
        let (assign_tx, mut assign_rx) = mpsc::channel::<(JobId, String)>(8);
        let (result_tx, result_rx) = mpsc::channel::<Frame>(8);

        // A fake worker: collect all three assigns, then answer 2, 1, 3 — proving
        // replies are matched by id, not by the order they come back.
        tokio::spawn(async move {
            let mut seen = Vec::new();
            while let Some((id, text)) = assign_rx.recv().await {
                seen.push((id, text));
                if seen.len() == 3 {
                    for &(id, ref text) in [&seen[1], &seen[0], &seen[2]] {
                        result_tx
                            .send(Frame::Result { job_id: id, answer: format!("re: {text}") })
                            .await
                            .unwrap();
                    }
                }
            }
        });

        tokio::spawn(connection_actor(jobs_rx, result_rx, assign_tx, 4));

        let mut waiters = Vec::new();
        for (id, text) in [(1u64, "alpha"), (2, "beta"), (3, "gamma")] {
            let (reply_tx, reply_rx) = oneshot::channel();
            jobs_tx.send((id, text.into(), reply_tx)).await.unwrap();
            waiters.push((id, reply_rx));
        }
        for (id, rx) in waiters {
            println!("job {id} -> {:?}", rx.await.unwrap());
        }
    }

    println!("---");

    // --- scenario 2: worker dies mid-job -> the pending job fails retryably ---
    {
        let (jobs_tx, jobs_rx) = mpsc::channel::<Dispatch>(8);
        let (assign_tx, mut assign_rx) = mpsc::channel::<(JobId, String)>(8);
        let (result_tx, result_rx) = mpsc::channel::<Frame>(8);

        // Worker takes the assign, then vanishes without answering (a crash).
        tokio::spawn(async move {
            let _ = assign_rx.recv().await;
        });
        drop(result_tx); // the actor's conn_rx now sees the close

        tokio::spawn(connection_actor(jobs_rx, result_rx, assign_tx, 4));

        let (reply_tx, reply_rx) = oneshot::channel();
        jobs_tx.send((99, "doomed".into(), reply_tx)).await.unwrap();
        println!("job 99 -> {:?}", reply_rx.await.unwrap());
    }
}

Predict two things before you read the output. First: in scenario 1, does job 1 print re: alpha even though the worker answered job 2 first? Second: in scenario 2, what does job 99 receive when the worker dies without answering?

job 1 -> Ok("re: alpha")
job 2 -> Ok("re: beta")
job 3 -> Ok("re: gamma")
---
actor: failing in-flight job 99 retryably
job 99 -> Err("worker connection lost")

Every waiter got its own answer despite the scrambled reply order — because each waited on a private oneshot and the actor routed by id through the map. And the doomed job did not hang forever: when conn_rx returned None, the loop broke, and the cleanup pass fired an error into every still-pending waiter. job 99 unblocked with an error rather than deadlocking.

The most important line: failing pending jobs on close

That cleanup loop is the pivot of the entire cluster arc, so slow down on it. When the socket closes — the worker crashed, the network dropped, the process was killed — the actor's recv returns None (a clean close) or an Err (a broken frame), the loop breaks, and control reaches the final for over pending. Every job the worker accepted but never answered is sitting in that map. The actor sends each one an error.

The kind of error is everything. In the real RemoteWorker, that error is ControlError::Worker(...) — a retryable error. Recall from Part II that dispatch returning a retryable error is precisely the signal the scheduler acts on: it hands the job to a different worker. So "the socket closed" becomes, through this one loop, "redeliver every in-flight job elsewhere." A worker dying mid-job does not lose the job — it releases it back to the scheduler. That is at-least-once delivery, and it is born right here, in the difference between failing a pending job retryably and letting it hang.

One-for-one: the toy ↔ the real thing

Every part of the toy maps onto the real connection_actor and RemoteWorker you build next:

• connection_actor (toy) ↔ connection_actor (real) — same loop, same select! with the capacity guard; the real one selects over a MessageStream instead of a stand-in mpsc, and also grows a heartbeat-timeout arm (second half of the arc).
• Dispatch = (JobId, String, oneshot::Sender<...>) ↔ Dispatch = (Job, oneshot::Sender<Result<JobOutcome, ControlError>>) — the job plus its private answer line, posted to the actor's inbox.
• the mpsc job inbox ↔ RemoteWorker's tx: mpsc::Sender<Dispatch> — RemoteWorker::dispatch makes a oneshot, sends (job, sender) to the actor, and awaits the receiver. The whole round-trip hides behind one WorkerHandle::dispatch.
• pending: HashMap<JobId, oneshot::Sender> ↔ the identical map in the real actor — id in, correct waiter out; the multiplexer itself.
• conn_tx.send(Assign) / conn_rx.recv() ↔ conn.send(&Message::Assign { job }) / conn.recv() — the real socket, via MessageStream.
• if pending.len() < capacity ↔ the same guard — backpressure that throttles the scheduler through the bounded mpsc.
• the final for failing pending jobs ↔ the real cleanup firing ControlError::Worker into each pending oneshot — retryable, so the scheduler redelivers. The at-least-once seam.

Hold that last row against the promise the whole course opened with. The scheduler still just calls dispatch on a dyn WorkerHandle and gets a Result — it never learns there is a socket, an actor, a map, or a redelivery underneath. The connection actor is the machine that makes a remote worker indistinguishable from a local one, and it fits behind the same seam because the seam was built for exactly this.

Questions to lock

Stop on each; the next build assembles precisely this.

1. Why can't two dispatch calls just share the socket and each read until their own result shows up? What specifically goes wrong, and how does routing everything through one actor task fix it?
2. What job does the per-job oneshot do that the shared mpsc inbox cannot? Why does giving each waiter its own channel make out-of-order results a non-problem?
3. Trace the backpressure: how does if pending.len() < capacity on the select! job arm end up throttling the scheduler? Name every link in the chain.
4. When the socket closes, the actor fails every pending job with a retryable error. Why retryable specifically, and what does the scheduler do as a result? What property of the whole system is that the beginning of?

Next: build it for real — RemoteWorker, the connection_actor over a MessageStream, the SharedPool, the accept loop, and the worker binary on the far end.