jonnifyfsharp

Dive C4.12.2

The actor

The model is pure. To run it for many callers, one writer owns the state and applies the changes in turn.

The transitions of C4.12.1 are pure: each takes a QueueState and returns a new one. A running queue needs one thing more — a single owner that holds the latest state and applies the transitions one at a time. In F# that owner is a MailboxProcessor: an agent with a mailbox and a recursive async receive loop, the language's actor and the analog of Elixir's GenServer. The Broker wraps the model. A private Message union names each request, each carrying its own reply channel; the loop applies the matching transition, replies, and tail-recurses with the new state. Because there is one mailbox, two callers cannot race a claim or mint a colliding id. Every snippet here was run in dotnet fsi.

Watch the mailbox serialize Lab hub
C4.12.2 · part 1

The Message — a request and a reply channel

Every request to the queue becomes a value: a case of a private Message discriminated union. Each case carries the request's arguments and an AsyncReplyChannel — the typed channel the caller waits on for the answer. Because the union is private, only the Broker can construct or read these messages; callers reach it through members. The reply type differs per case: an admission verdict, a job option, a result.

type private Message =
    | Enqueue of id: string * payload: string * AsyncReplyChannel<Admission>
    | Claim of AsyncReplyChannel<Job option>
    | Complete of id: string * token: int * AsyncReplyChannel<Result<unit, string>>
    | PendingSize of AsyncReplyChannel<int>
// type Message =
//   | Enqueue of id: string * payload: string * AsyncReplyChannel<Admission>
//   | Claim of AsyncReplyChannel<Job option>
//   | Complete of id: string * token: int * AsyncReplyChannel<Result<unit,string>>
//   | PendingSize of AsyncReplyChannel<int>
Message
a private union: one case per request, so the loop's match is exhaustive.
AsyncReplyChannel<'T>
the typed channel a caller waits on; the loop calls reply.Reply with a 'T.
private
only the Broker has access to the message type; callers use its members, not the union.
C4.12.2 · part 2

The Broker — one mailbox, one loop

The Broker starts a MailboxProcessor whose body is a recursive async loop carrying the QueueState. Each turn: receive a message, apply the matching pure transition from C4.12.1, reply on the message's channel, and tail-call loop with the new state. The state lives only in the loop's argument — no field, no lock. Members expose the queue: each posts a message and waits for the reply with PostAndReply.

type Broker() =
    let agent =
        MailboxProcessor.Start(fun inbox ->
            let rec loop st = async {
                let! msg = inbox.Receive()
                match msg with
                | Enqueue (id, payload, reply) ->
                    let st', verdict = enqueue id payload st
                    reply.Reply verdict
                    return! loop st'
                | Claim reply ->
                    let st', job = claim st
                    reply.Reply job
                    return! loop st'
                | Complete (id, token, reply) ->
                    let st', result = complete id token st
                    reply.Reply result
                    return! loop st'
                | PendingSize reply ->
                    reply.Reply (pendingSize st)
                    return! loop st }
            loop empty)

    member _.Enqueue(id, payload) = agent.PostAndReply(fun r -> Enqueue(id, payload, r))
    member _.Claim() = agent.PostAndReply(Claim)
    member _.Complete(id, token) = agent.PostAndReply(fun r -> Complete(id, token, r))
    member _.PendingSize() = agent.PostAndReply(PendingSize)
// type Broker =
//   new: unit -> Broker
//   member Claim: unit -> Job option
//   member Complete: id: string * token: int -> Result<unit,string>
//   member Enqueue: id: string * payload: string -> Admission
//   member PendingSize: unit -> int
PostAndReply posts a message and blocks the caller until the loop replies. The real echo_fs broker also exposes EnqueueAsync over PostAndAsyncReply, which yields instead of blocking — so many concurrent callers overlap without holding threads.

Watch the mailbox serialize

Post messages to the broker. They queue in its one mailbox; the receive loop takes them one at a time, applies the pure transition, threads a new QueueState, and replies. Post several before stepping the loop to see them wait in order — the proof that two callers cannot interleave. The readout names the message the loop applied and the reply it sent.

One mailbox · one writer · messages applied in turn

mailbox · FIFO 0 waiting no messages posted loop : QueueState → async owns the state · one at a time Pending = 0 · Completed = 0 idle reply.Reply → return! loop st'
an idle broker with an empty queue. mailbox = 0 · Pending = 0. Post a message, then step the loop.

The state never leaves the loop, and the loop handles one message at a time. That single-writer fact is the whole safety argument: no two transitions interleave, so nothing needs a lock.

The same shape, beside Elixir

Elixir's GenServer is this exact pattern with callbacks: a process with a mailbox, a state threaded through handle_call, and a reply. F#'s MailboxProcessor is the same mailbox and receive loop, hand-written. The shape carries over; the difference is the runtime under it — the Elixir process is supervised, the F# agent is one object on the shared CLR (the subject of C3).

Elixir — a GenServer call and reply

# Elixir — the call posts a message, the server replies
def handle_call({:enqueue, id, payload}, _from, st) do
  {st, verdict} = Store.enqueue(id, payload, st)
  {:reply, verdict, st}
end

F# — the MailboxProcessor loop clause (verified in FSI)

// F# — the loop applies the same pure transition, then replies
| Enqueue (id, payload, reply) ->
    let st', verdict = enqueue id payload st
    reply.Reply verdict
    return! loop st'

Elixir / GenServer

A supervised process with a mailbox; handle_call applies a transition and returns {:reply, value, new_state}.

F# / MailboxProcessor

One agent with a mailbox; the loop clause applies the same transition, calls reply.Reply, and tail-recurses with the new state.

EchoMQ in F#

Status — echo_fs is an in-memory model. The EchoMQ protocol — the Lua transitions executing on a Valkey wire — is not implemented (there is no Lua and no Valkey client in the code); that is the coming-soon transport (EchoMQ.Connector). This block shows how the model uses the structure, not a running bus.

The broker here is not a teaching sketch — it is the bus's real single writer, and it is tested.

runs today echo_fs/EchoMQ/Broker.fs

EchoMQ.Broker is a real MailboxProcessor, built and under test in echo_fs. It takes an optional initial QueueState, starts the agent, and exposes Enqueue, EnqueueAsync, Claim, Complete, and PendingSize as members over PostAndReply / PostAndAsyncReply. The file's own comment states the argument: "there is only ever one writer, so two callers cannot race a claim or mint a colliding id." The lab's Broker is this writer, trimmed to four members.

coming soon The loop applies in-memory Store transitions today. When the Valkey transport EchoMQ.Connector lands, the same broker can route its transitions to the wire — the message type, the loop, and the members do not change, because the store sits behind a port.

The broker serializes transitions over an in-memory Map / Set store, not a CHAMP and not Valkey. The durable backing swaps in behind the same surface; the branded CHAMP and the wire are taught as the target, not claimed as built.

Where this leads

The model and the actor are both built. The last dive runs them as one queue, end to end — construct a broker, enqueue a few jobs, claim the oldest, complete it — the chapter's capstone, and a mini echo_fs.

↑ C4.12 · Lab: build a branded CHAMP store ← C4.12.1 · The model C4.12.3 · Ship it →

References

Sources

Related in this course