Order Gateway: TCP Sessions

The layout you'll build:

           Tokio runtime (I/O)                                 engine thread (std::thread)
 socket ─▶ reader task: FrameDecoder → Session → decode ─┐
 socket ─▶ reader task ...                                ├─▶ MPSC queue (w05) ─▶ engine v1 (w12)
 socket ─▶ reader task ...                               ─┘                         │ events tagged
                                                                                    ▼ with session id
 socket ◀─ writer task ◀─ per-session bounded queue ◀───────── dispatcher: session id → queue

• Each connection gets a reader task and a writer task. The reader owns the FrameDecoder and the Session. The writer owns the write half of the socket and drains that session's outbound queue. Splitting the socket (TcpStream::into_split) means a slow write never stalls reads, and the reverse.
• The engine runs on a dedicated OS thread, not a Tokio task. It busy-polls one MPSC queue of (session_id, command) pairs. Every Tokio worker thread can push into that queue, so it has to be multi-producer.
• Outbound traffic is routed by session id. The engine's events name exchange order ids. A router on the engine thread maps each id back to the session that owns the order, and a dispatcher looks up that session's outbound queue.

TCP delivers an ordered stream of bytes, with no message boundaries. A 64-byte order you sent with one write can arrive as 64 bytes, as 1 byte and then 63, or glued to the end of the previous order. The sender's Nagle algorithm, segmentation, receive offloads (GRO) and simple timing all reshape it. The frame decoder restores message boundaries:

1. Append whatever the socket returned to a buffer.
2. If at least 2 bytes are buffered, read the length header and validate it immediately.
3. If header plus payload are buffered, return the payload and advance. Otherwise wait for more.
4. Reclaim consumed space so the buffer stays bounded by about one maximum frame, not by the connection's lifetime traffic.

// Frame = 2-byte little-endian payload length, then the payload.
// A payload length of 0 is invalid (every message has at least a type byte).

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameError {
    Empty,
    TooLarge { len: usize, max: usize },
}

pub struct FrameDecoder {
    buf: Vec<u8>,
    start: usize, // first unconsumed byte in buf
    max_payload: usize,
}

impl FrameDecoder {
    pub fn new(max_payload: usize) -> Self {
        FrameDecoder { buf: Vec::with_capacity(4096), start: 0, max_payload }
    }

    /// Append the bytes from one socket read. Reclaim consumed space so
    /// the buffer stays bounded no matter how long the connection lives.
    pub fn feed(&mut self, bytes: &[u8]) {
        todo!()
    }

    /// The next complete payload, borrowed from the internal buffer, or
    /// Ok(None) if more bytes are needed. Reject a bad length as soon as the
    /// header is available. After an error the caller disconnects.
    pub fn next_frame(&mut self) -> Result<Option<&[u8]>, FrameError> {
        todo!()
    }

    /// Bytes received but not yet returned as frames.
    pub fn buffered(&self) -> usize {
        self.buf.len() - self.start
    }

    pub fn capacity(&self) -> usize {
        self.buf.capacity()
    }
}

Every order-entry protocol has a session layer around the business messages. It authenticates once at logon, numbers messages so that loss or duplication is detectable, proves liveness with heartbeats when there's no traffic, and ends cleanly with a logout. Write it as an explicit state machine, as a pure struct with on_message(now, msg, out) and on_tick(now, out), not as if statements scattered through an async handler:

• Pure and synchronous. No sockets, no clock reads. Time comes in as an argument, so a test can play a scripted hour of session life in microseconds, which is exactly what the harness does.
• Actions out, not side effects. The session pushes Forward, LogonAck or Disconnect into a caller-owned Vec. The async shell performs them. One reused Vec per connection means no allocation per message.
• Fail closed. Anything unexpected (a message before logon, a sequence gap, a second logon) disconnects. A session that tries to "recover" from a confused client tends to forward orders the client never meant to send.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
    AwaitingLogon,
    Active,
    Closed,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Inbound {
    Logon { seq: u64, token: u64 },
    NewOrder { seq: u64, order_id: u64 },
    Heartbeat { seq: u64 },
    Logout { seq: u64 },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reason {
    BadSeq { expected: u64, got: u64 },
    BadToken,
    NotLoggedOn,
    DuplicateLogon,
    Timeout,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Action {
    LogonAck,
    Forward { order_id: u64 },
    LogoutAck,
    Disconnect(Reason),
}

pub struct Session {
    state: State,
    token: u64,
    next_seq: u64,
    last_rx_ns: u64,
    timeout_ns: u64,
}

impl Session {
    pub fn new(now_ns: u64, token: u64, timeout_ns: u64) -> Self {
        Session { state: State::AwaitingLogon, token, next_seq: 1, last_rx_ns: now_ns, timeout_ns }
    }

    pub fn state(&self) -> State {
        self.state
    }

    /// Push actions into `out` (do not clear it). Rules, in order:
    ///  - Closed: ignore everything.
    ///  - seq != expected (first expected is 1): Disconnect(BadSeq).
    ///  - otherwise expected += 1, last receive time = now_ns, then
    ///    AwaitingLogon: Logon with the right token -> LogonAck, Active;
    ///                   wrong token -> Disconnect(BadToken);
    ///                   any other message -> Disconnect(NotLoggedOn).
    ///    Active: Logon -> Disconnect(DuplicateLogon); NewOrder -> Forward;
    ///            Heartbeat -> nothing; Logout -> LogoutAck, Closed.
    ///  - every Disconnect moves the session to Closed.
    pub fn on_message(&mut self, now_ns: u64, msg: Inbound, out: &mut Vec<Action>) {
        todo!()
    }

    /// Timer. If not Closed and now - last receive > timeout_ns:
    /// Disconnect(Timeout). This also applies while awaiting logon.
    pub fn on_tick(&mut self, now_ns: u64, out: &mut Vec<Action>) {
        todo!()
    }
}

Inbound (sessions → engine): many reader tasks on several Tokio worker threads push (session_id, command) into one queue the engine thread drains. That's the bounded MPSC from week 5. When it's full, the reader task, not the engine, has to wait. Make it wait by yielding (tokio::task::yield_now, or a short async sleep with backoff), not by blocking the worker thread. A full inbound queue means the engine is saturated, so applying backpressure to clients is correct. The TCP window will eventually push back on them.

The explainer below runs this pipeline as it looks by week 24, with a journal and standby behind the engine. Choose Ack after match for this week's shape, then drag the load past the engine's capacity: the bounded inbound queue fills and the push-back reaches the gateway (here as busy rejects, the week 24 policy; this week the reader waits and TCP pushes back).

Outbound (engine → sessions): the engine produces execution reports for many sessions. It must never block on any of them, because one stalled client socket would freeze matching for everyone. Give each session a bounded outbound queue and have the engine use a non-blocking push. tokio::sync::mpsc::Sender::try_send is synchronous and safe to call from a plain thread, and it wakes the writer task if it's parked. If the push fails because the queue is full, the client isn't keeping up. Mark the session for disconnect rather than buffering without bound.

Two costs you'll see in the round-trip benchmark: a wakeup of a parked Tokio worker from another thread is a futex syscall plus a scheduler decision, which costs microseconds rather than nanoseconds. An engine that sleeps when idle pays a similar wakeup on every order. Busy-polling the engine thread removes the second cost. Week 30 measures both properly.

Set TCP_NODELAY on every accepted socket. Nagle's algorithm holds back a small write while earlier data is unacknowledged, hoping to coalesce it with the next one. Combined with the receiver's delayed ACK, that can add tens of milliseconds to an order. On Linux the delayed-ACK timer starts at 40 ms. For a protocol of small request/response messages, Nagle is pure harm. Do your own batching instead: when several execution reports for one session are queued, write them in one syscall.

Socket buffer sizes (SO_RCVBUF / SO_SNDBUF) set how much the kernel will absorb before applying TCP flow control. Large buffers hide a slow consumer for longer, so they help throughput and hurt your ability to detect lag. Moderate sizes plus an application-level outbound bound give you control.

Expect a small-message ping-pong over loopback to land in roughly the 10–30 µs range with ordinary blocking sockets and an async runtime on a modern Linux machine, and lower when both ends busy-poll. Most of that is syscalls and wakeups, not the TCP protocol. The exact figure depends on CPU, kernel version and mitigations, which is why week 31 measures it on your own machine.

List the behaviours your gateway must survive, and decide what it does about each:

• A slow reader: the client stops reading. Its outbound queue fills, and the gateway disconnects it.
• A flood: a client sends faster than its share. Week 17 adds throttles. For now, the bounded inbound queue applies backpressure.
• Half-open connections: the client vanished without a FIN. A heartbeat timeout disconnects it. The repo's binary protocol has no heartbeats, so there the backstop is TCP keepalive, which week 31's socket tuning turns on.
• Garbage: bad lengths, unknown message types, sequence gaps. Fail closed, disconnect, log the session id and the offending bytes.
• A connect storm at the open: thousands of logons within a second. Logon validation runs in the reader task, off the engine thread. The engine thread only registers the session and queues its LogonAck.