jonnifyfsharp

Dive C4.09.2

The MailboxProcessor

The registry has a writer now. In F# the writer is an agent with a mailbox and a recursive loop.

F#'s actor is the MailboxProcessor — the analog of Elixir's GenServer. It is an agent with a mailbox and a recursive async receive loop: it takes one message at a time, computes a new state, replies if the message asked for an answer, and tail-calls itself with the new state. The bus this chapter cites already has one — EchoMQ.Broker, built and tested in the echo_fs port (BrokerTests). This dive reads the real broker: its message type, its receive loop, and the two ways a caller posts to it.

Step the receive loop Module hub
C4.09.2 · part 1

The message type — a DU with reply channels

Every command to the broker is a case of one discriminated union. Each case names its arguments and carries an AsyncReplyChannel — the typed channel the loop uses to hand a value back to the caller. The verb decides what the channel returns: an Enqueue returns an Admission, a Claim returns a Job option, a Complete returns a Result, and PendingSize returns an int. The type is private: callers never construct a message directly — they go through the broker's members, which is the next part.

// echo_fs/EchoMQ/Broker.fs — the command type
type private Message =
    | Enqueue of id: string * payload: string * AsyncReplyChannel<Admission>
    | Claim of now: int64 * leaseMs: int64 * AsyncReplyChannel<Job option>
    | Complete of id: string * token: int * AsyncReplyChannel<Result<unit, string>>
    | PendingSize of AsyncReplyChannel<int>
Message
a private DU; one case per broker verb. Exhaustiveness in the loop's match means a new verb cannot be forgotten.
AsyncReplyChannel<'T>
the typed reply channel a caller threads in; the loop calls reply.Reply with the answer of type 'T.
private
only the broker's own members build a Message; callers cannot post a malformed command.
C4.09.2 · part 2

The receive loop — one message, one transition

MailboxProcessor.Start takes a function of the inbox and runs it on the thread pool. Inside, a recursive loop holds the queue's state. Each turn awaits the next message with inbox.Receive(), matches the verb, applies the matching pure Store transition to get a new state and a result, sends the result down the reply channel, and tail-calls loop with the new state. The state is a value threaded from turn to turn — never a shared mutable cell.

// echo_fs/EchoMQ/Broker.fs — the agent and its loop
let agent =
    MailboxProcessor.Start(fun inbox ->
        let rec loop (st: QueueState) =
            async {
                let! msg = inbox.Receive()

                match msg with
                | Enqueue (id, payload, reply) ->
                    let st', verdict = Store.enqueue id payload st
                    reply.Reply verdict
                    return! loop st'
                | Claim (now, leaseMs, reply) ->
                    let st', job = Store.claim now leaseMs st
                    reply.Reply job
                    return! loop st'
                | Complete (id, token, reply) ->
                    let st', result = Store.complete id token st
                    reply.Reply result
                    return! loop st'
                | PendingSize reply ->
                    reply.Reply(Store.pendingSize st)
                    return! loop st
            }

        loop (defaultArg initial Store.empty))
return! loop st' is a tail call inside the async workflow, so the loop runs in constant stack — it can receive forever. Each branch reuses a pure Store function (C4.09.1's transitions), so the agent is the only place state changes, and the change is always a value swap.

Step the broker through its mailbox

A fixed queue of three messages sits in the mailbox. Step the broker forward: it takes the front message, applies the matching Store transition, replies on the channel, and advances. The readout names the message, the reply, and the pending depth after the turn — the real verdicts (Enqueued, Duplicate, a claimed job, a size). One message is handled per step, in order; nothing interleaves.

EchoMQ.Broker · the mailbox, one message per step

mailbox · FIFO Enqueue JOB…aZxKq9P Enqueue JOB…aZxKq9P (dup) Claim (oldest pending) EchoMQ.Broker · loop st handling Enqueue …aZxKq9P pending 0 reply —
The mailbox holds three messages. Press receive next to handle the front one — the broker applies the matching Store transition and replies, then advances.

The loop is the whole concurrency story: one message in, one transition, one reply, one new state. Because there is a single loop, the order of handling is the order of arrival — the property the last dive turns into safety.

C4.09.2 · part 3

Posting — blocking and non-blocking

Callers do not build a Message; they call the broker's members, which post one and wait for its reply. PostAndReply blocks the calling thread until the loop answers — the synchronous call. PostAndAsyncReply returns an Async that yields when the reply arrives, so many callers can have a request in flight without each holding a thread. Both thread the reply channel in via a lambda; the broker fills it from inside the loop.

// echo_fs/EchoMQ/Broker.fs — the public surface
member _.Enqueue(id: string, payload: string) : Admission =
    agent.PostAndReply(fun r -> Enqueue(id, payload, r))

// non-blocking: yields until the mailbox replies, so callers overlap freely
member _.EnqueueAsync(id: string, payload: string) : Async<Admission> =
    agent.PostAndAsyncReply(fun r -> Enqueue(id, payload, r))

member _.Claim(now: int64, leaseMs: int64) : Job option =
    agent.PostAndReply(fun r -> Claim(now, leaseMs, r))

member _.PendingSize() : int = agent.PostAndReply PendingSize

A reduced broker built and run in dotnet fsi shows the verdicts the members return — the same results the real EchoMQ.Broker gives:

let b = Broker()
b.Enqueue("JOB0Nc1aZxKq9P", "build #42")   // val it: Admission = Enqueued
b.Enqueue("JOB0Nc1aZxKq9P", "build #42")   // val it: Admission = Duplicate
b.PendingSize()                            // val it: int = 1

Elixir — GenServer.call posts and waits

# Elixir — the call posts a message, the server replies
GenServer.call(broker, {:enqueue, id, payload})   # => :enqueued | :duplicate

F# — PostAndReply is the same shape

// F# — PostAndReply posts the message, the loop replies
broker.Enqueue(id, payload)   // => Enqueued | Duplicate

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 above 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.Tests/BrokerTests.fs

EchoMQ.Broker is a real MailboxProcessor — built and under test in echo_fs. The Broker type takes an optional initial QueueState, starts the agent, and exposes Enqueue, EnqueueAsync, Claim, Complete, and PendingSize as members over PostAndReply / PostAndAsyncReply. The BrokerTests suite asserts the loop serializes correctly: one test enqueues a job, checks the pending depth, and claims it back; another posts many ids at once and confirms none are lost — the subject of the next dive.

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 is behind a port.

Where this leads

The broker exists, holds the registry, and handles one message at a time. That single-loop fact is the chapter's payoff, and the next dive states it directly: because every mint and claim queues through one mailbox, two callers cannot race a claim or mint a colliding id — proved by the parallel-posts test.

↑ C4.09 · Branded CHAMP maps & MailboxProcessor ← C4.09.1 · The branded CHAMP C4.09.3 · The single writer →

References

Sources

Related in this course