Concept: TCP Is a Byte Stream — Framing with LengthDelimitedCodec

Kind: Concept. New crate: tokio-util — this chapter shows it working before you build with it.

For seven arcs the coordinator has run every job in-process. A LocalWorker is a WorkerHandle that awaits run_eval on the same machine; the scheduler dispatches to it and gets a JobOutcome back. That is the whole system so far, and it is complete — it schedules, retries, persists, and reports. What it cannot do is spread work across machines. The last arc adds that, and it adds it, as promised in Part I, as a new impl WorkerHandle — the scheduler never changes. But a RemoteWorker has to actually get a Job to another process and a JobOutcome back, and between here and there is a TCP socket. This chapter is about the single most surprising thing about that socket, and the crate that tames it.

The surprising thing: TCP has no messages

You think of the wire in terms of messages: the coordinator sends an Assign, the worker sends back a Result. That is the mental model, and it is the model the rest of the arc is built on. But TCP does not have that model. TCP is a stream of bytes. It guarantees that the bytes you write come out the other end in order and without gaps — and that is all it guarantees. It does not remember where one write ended and the next began. It is free to glue two of your writes into one read, or split one of your writes across two reads, however the kernel and the network happen to buffer things.

That is not a bug or an edge case; it is the definition of a stream protocol, and it bites the moment you send two things in a row. Let us watch it bite. Here is a client that does two entirely separate write_all calls — five bytes, then five more — and a server that reads until the connection closes and reports what it got:

// scratch Cargo.toml deps:
//   tokio = { version = "1", features = ["full"] }
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    // Server: read every byte until the peer closes, then report what arrived.
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        let mut buf = Vec::new();
        sock.read_to_end(&mut buf).await.unwrap();
        buf
    });

    // Client: two *separate* writes, each a whole "message" as far as we mean it.
    let mut client = TcpStream::connect(addr).await.unwrap();
    client.write_all(b"hello").await.unwrap();
    client.write_all(b"world").await.unwrap();
    client.shutdown().await.unwrap(); // close so read_to_end returns

    let got = server.await.unwrap();
    println!("two writes of 5 bytes each; the reader saw {} bytes:", got.len());
    println!("{:?}", String::from_utf8(got).unwrap());
}

Predict the output before you read on. You wrote "hello" and "world" as two calls — how many reads does the other side see, and what is in them?

two writes of 5 bytes each; the reader saw 10 bytes:
"helloworld"

One blob. "helloworld", ten bytes, no seam. The reader has no way to know you meant two messages — the boundary between them existed only in your source code, and TCP threw it away. If those had been two JSON objects, {"a":1}{"b":2}, they would have arrived as {"a":1}{"b":2} with nothing marking where the first ends, and a naive serde_json::from_slice over the whole buffer would choke on the trailing content after the first object. Worse, on a busy network the split can land inside an object — {"a": in one read, 1}{"b":2} in the next — and now you are hand-writing a buffering parser that stitches partial reads back together. That job, "turn a byte stream back into the messages someone meant," is called framing, and you do not want to write it by hand.

The fix: prefix each frame with its length

There are a few classic ways to put boundaries back into a byte stream. You could pick a delimiter byte (like a newline) and agree that it never appears inside a message — but JSON contains newlines, so you would have to escape them, and now you have two encodings. The robust, standard answer is simpler: before each message, write its length. The reader reads a fixed-size length header first — say, four bytes — learns "the next N bytes are one message," reads exactly N more bytes (buffering across as many TCP reads as that takes), and hands you those N bytes as one complete frame. Then it does it again for the next message. Length-prefixing turns an undelimited stream back into a sequence of discrete, whole messages.

You will not write that reader. tokio-util ships it as LengthDelimitedCodec. A codec is a small object that knows how to turn bytes into frames and frames into bytes; LengthDelimitedCodec is the one that does exactly the length-prefix scheme just described. You wrap it around a TcpStream with Framed, and the pair gives you two superpowers: Framed is both a Stream of incoming frames (.next().await yields the next complete Bytes blob) and a Sink for outgoing frames (.send(bytes).await length-prefixes and writes one blob). The four-byte length header is added on send and stripped on receive; you never see it.

The toy: a message channel over a real socket

Here is the whole scheme working, in a domain with none of the coordinator's machinery. We define a tiny two-variant enum, Note, and a wrapper called NoteStream that owns a Framed<TcpStream, LengthDelimitedCodec> and offers exactly two methods: send(&Note) and recv() -> Option<Note>. send serializes the note to JSON bytes and pushes one frame; recv pulls one frame and deserializes it. This is a scaled-down MessageStream — the very type you build in the next chapter — and it is deliberately the same shape, method for method.

It needs tokio-util, futures, bytes, and serde, none of which are on the playground, so it is marked ignore. It is a real, runnable program — the output below is from actually running it against a loopback socket.

// scratch Cargo.toml deps:
//   tokio       = { version = "1", features = ["full"] }
//   tokio-util  = { version = "0.7", features = ["codec"] }
//   futures     = "0.3"
//   bytes       = "1"
//   serde       = { version = "1", features = ["derive"] }
//   serde_json  = "1"
use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{Framed, LengthDelimitedCodec};

/// The toy message — a two-variant enum, self-describing on the wire via
/// serde's internal `"type"` tag (exactly how the real `Message` is tagged).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum Note {
    Hello { who: String },
    Ping { seq: u32 },
}

/// A bidirectional channel of `Note`s over one TCP connection. This is a
/// scaled-down `MessageStream`: `Framed` + `LengthDelimitedCodec` do the framing,
/// serde_json does the encoding, one method each way.
struct NoteStream {
    framed: Framed<TcpStream, LengthDelimitedCodec>,
}

impl NoteStream {
    fn new(stream: TcpStream) -> Self {
        Self {
            framed: Framed::new(stream, LengthDelimitedCodec::new()),
        }
    }

    async fn send(&mut self, note: &Note) -> std::io::Result<()> {
        let bytes = serde_json::to_vec(note).expect("serialize");
        self.framed.send(Bytes::from(bytes)).await
    }

    /// `None` means the peer closed the connection cleanly.
    async fn recv(&mut self) -> Option<Note> {
        match self.framed.next().await {
            Some(Ok(frame)) => Some(serde_json::from_slice(&frame).expect("deserialize")),
            Some(Err(e)) => panic!("frame error: {e}"),
            None => None,
        }
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    // Echo server: read each framed message, send it straight back.
    tokio::spawn(async move {
        let (sock, _) = listener.accept().await.unwrap();
        let mut server = NoteStream::new(sock);
        while let Some(note) = server.recv().await {
            server.send(&note).await.unwrap();
        }
    });

    let sock = TcpStream::connect(addr).await.unwrap();
    let mut client = NoteStream::new(sock);

    // Two messages, sent back-to-back with no gap between them on the wire.
    let a = Note::Hello { who: "w1".into() };
    let b = Note::Ping { seq: 7 };
    client.send(&a).await.unwrap();
    client.send(&b).await.unwrap();

    // The codec still hands us exactly two whole messages, in order.
    let ra = client.recv().await.unwrap();
    let rb = client.recv().await.unwrap();
    println!("sent:    {a:?}");
    println!("sent:    {b:?}");
    println!("got back: {ra:?}");
    println!("got back: {rb:?}");
    assert_eq!(a, ra);
    assert_eq!(b, rb);
}

The two sends go out back-to-back — exactly the situation that produced one glued blob in the first program. Predict what recv returns this time, and how many times.

sent:    Hello { who: "w1" }
sent:    Ping { seq: 7 }
got back: Hello { who: "w1" }
got back: Ping { seq: 7 }

Two calls to recv, two whole messages, in order, each a fully-typed Note — even though the same two writes over a bare socket arrived as "helloworld". Nothing in this program buffers partial reads or hunts for boundaries; LengthDelimitedCodec does all of it under Framed. The length prefix it wrote before each JSON blob is what let the reader carve the stream back into Hello { who: "w1" } and Ping { seq: 7 }. That is the entire idea of the arc's wire: frame on the outside, JSON on the inside, one typed message per recv.

Where the errors go

In the toy, recv panics on a malformed frame — fine for a demo. The real MessageStream cannot panic a connection task, so it does the disciplined thing: both a serde failure and a framing/IO failure map to ControlError::Protocol(...), the retryable protocol-error variant you built in Part II. That choice is deliberate, and it is the same reasoning as the whole error taxonomy: a garbled frame or a dropped connection is a transient fault about this worker's link, not a claim that the job is bad — so it should be retryable, and the scheduler should get the chance to hand the job to someone else. A clean end-of-stream is different: recv returns Ok(None), meaning "the peer closed, no error," which the connection actor in the next concept chapter reads as "this worker is gone." Malformed is an error; closed is a None. Hold that distinction — the actor leans on it.

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

Everything here maps straight onto the control-core codec you build in the very next chapter, piece for piece:

• NoteStream (owns a Framed, offers send/recv) ↔ MessageStream — the real wrapper, Framed<TcpStream, LengthDelimitedCodec> inside, send(&Message) and recv() -> Option<Message> outside.
• Note (a small tagged enum) ↔ Message — the Register/Assign/Result/Heartbeat enum you already defined in Part II, tagged with the same #[serde(tag = "type")], finally traveling over a socket instead of only round-tripping in a unit test.
• serde_json::to_vec + from_slice per frame ↔ the same calls in MessageStream — one JSON document per length-delimited frame, both directions.
• the toy's recv panic on bad bytes ↔ ControlError::Protocol — the real code turns a serde or framing failure into the retryable protocol error instead of panicking.
• recv returning None on clean close ↔ Ok(None) from MessageStream::recv — the "peer closed cleanly" signal the connection actor watches for.

That last row is the seam between this chapter and the next: framing gives you whole messages, and "no more messages" (None) is how the socket tells you the worker left. The next build chapter turns NoteStream into MessageStream and proves it with a Message round-trip over a real loopback socket — the twelfth control-core test, and the first one that touches the network.

Questions to lock

Stop on each; the whole arc's wire rests here.

1. TCP guarantees your bytes arrive in order and intact. What does it specifically not guarantee, and why does that force you to add framing before you can send two messages in a row?
2. What are the two separate jobs that LengthDelimitedCodec and serde_json each do, and why is it worth keeping them as two layers rather than folding them into one?
3. Which two futures traits must you import to call .send() and .next() on a Framed, and what does the compiler error look like when you forget?
4. MessageStream::recv maps a malformed frame to ControlError::Protocol but returns Ok(None) on a clean close. Why is one an error and the other not — and what will the connection actor do with each?

Next: build MessageStream for real, and watch a Message survive a round trip over an actual socket.