TCP Transport and RPC Server — Network Layer for a Distributed Log

With the wire protocol defined, the next layer is the network plumbing that actually moves frames between processes: a TCP transport (server and client halves) and an RPC server built on top of it that maps message types to handler functions calling into the topic and consumer managers. The transport lives in api/transport/; the RPC server and its handlers live in broker/rpc/.

Step 1: Build the Transport struct

What Transport is responsible for

Transport owns a TCP listener, a registry of handlers keyed by message type, and the bookkeeping needed to shut down cleanly — a set of active connections and a WaitGroup tracking their goroutines:

// api/transport/transport.go
package transport

import (
    "context"
    "net"
    "sync"

    "github.com/mohitkumar/mlog/api/protocol"
)

// StreamHandler is used for RPCs where the server writes multiple response frames on the same connection.
// After the handler returns, the connection is reused for the next request.
type StreamHandler func(ctx context.Context, msg any, conn net.Conn, codec *protocol.Codec) error

// Transport manages TCP connections with a length-prefixed frame protocol.
// Producer, consumer, and replication use it to Send and Receive raw bytes.
type Transport struct {
    Codec    *protocol.Codec
    handlers map[protocol.MessageType]func(context.Context, any) (any, error)
    ln       net.Listener
    mu       sync.Mutex            // protects conns
    conns    map[net.Conn]struct{} // active connections for graceful shutdown
    wg       sync.WaitGroup        // tracks active connection goroutines
}

func NewTransport() *Transport {
    return &Transport{
        Codec:    &protocol.Codec{},
        handlers: make(map[protocol.MessageType]func(context.Context, any) (any, error)),
        conns:    make(map[net.Conn]struct{}),
    }
}

StreamHandler is declared here for the streaming case (a handler that writes several response frames on one connection instead of a single request/response pair) but the RPC handlers built later in this chapter are all plain request/response, so the transport's main job is dispatching one decoded message to one handler function that returns one response.

Step 2: Register handlers and start listening

RegisterHandler

Handlers are keyed by protocol.MessageType and share one signature — take a context and the decoded message, return a response or an error:

func (t *Transport) RegisterHandler(msgType protocol.MessageType, handler func(context.Context, any) (any, error)) {
    t.handlers[msgType] = handler
}

Listen, Addr, and ListenAndServe

func (t *Transport) Listen(addr string) (net.Listener, error) {
    ln, err := net.Listen("tcp", addr)
    if err != nil {
        return nil, err
    }
    t.ln = ln
    return ln, nil
}

func (t *Transport) Addr() string {
    if t.ln != nil {
        return t.ln.Addr().String()
    }
    return ""
}

func (t *Transport) Serve(ln net.Listener) {
    for {
        conn, err := ln.Accept()
        if err != nil {
            return
        }
        t.wg.Add(1)
        go func() {
            defer t.wg.Done()
            t.handleConn(conn)
        }()
    }
}

func (t *Transport) ListenAndServe(addr string) error {
    ln, err := t.Listen(addr)
    if err != nil {
        return err
    }
    slog.Info("listening", "addr", ln.Addr())
    t.Serve(ln)
    return nil
}

Splitting Listen from Serve matters for one caller in particular: the RPC server (Step 7) needs to know the bound address — including the actual port when addr was :0 — before it hands the listener off to Serve in a goroutine. ListenAndServe is the convenience wrapper for callers (and tests) that don't need that.

Each accepted connection gets its own goroutine running handleConn in a loop, so one slow or stuck client never blocks any other connection.

Step 3: Handle a connection

Timeouts

Two timeouts bound how long the server will wait on a connection: an idle timeout on reads, and a per-handler-invocation timeout on the context passed to the handler.

const (
    // idleTimeout is how long a connection can sit idle before we close it.
    idleTimeout = 5 * time.Minute
    // handlerTimeout is the max time a single handler invocation may take.
    handlerTimeout = 30 * time.Second
)

handleConn

func (t *Transport) handleConn(conn net.Conn) {
    t.trackConn(conn)
    defer func() {
        t.untrackConn(conn)
        conn.Close()
    }()
    for {
        // Set a read deadline so idle connections don't hang forever.
        _ = conn.SetReadDeadline(time.Now().Add(idleTimeout))

        mType, msg, err := t.Codec.Decode(conn)
        if err != nil {
            if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) {
                slog.Warn("read error", "remote", conn.RemoteAddr(), "err", err)
            }
            return
        }
        handler := t.handlers[mType]
        if handler == nil {
            slog.Warn("no handler", "msgType", mType, "remote", conn.RemoteAddr())
            continue
        }

        ctx, cancel := context.WithTimeout(context.Background(), handlerTimeout)
        resp, err := handler(ctx, msg)
        cancel()

        if err != nil {
            slog.Debug("handler error", "msgType", mType, "err", err)
            code := protocol.CodeUnknown
            message := err.Error()
            var rpcErr *protocol.RPCError
            if errors.As(err, &rpcErr) {
                code = rpcErr.Code
                message = rpcErr.Message
            }
            _ = t.Codec.Encode(conn, &protocol.RPCErrorResponse{Code: code, Message: message})
            continue
        }
        if err := t.Codec.Encode(conn, resp); err != nil {
            slog.Warn("encode error", "remote", conn.RemoteAddr(), "err", err)
            return
        }
    }
}

Every error a handler returns is converted into an RPCErrorResponse frame — coded and messaged from the *protocol.RPCError when the handler returned one, or CodeUnknown for a bare error — so the client always gets a well-formed frame back rather than a dropped connection. The connection itself stays open after an error; only a decode failure (EOF, a closed connection) or an encode failure ends the loop.

Step 4: Track connections and shut down gracefully

Every accepted connection is registered in a map so Close can reach it later:

func (t *Transport) trackConn(c net.Conn) {
    t.mu.Lock()
    t.conns[c] = struct{}{}
    t.mu.Unlock()
}

func (t *Transport) untrackConn(c net.Conn) {
    t.mu.Lock()
    delete(t.conns, c)
    t.mu.Unlock()
}

// Close stops accepting new connections, closes all active connections, and waits for goroutines to exit.
func (t *Transport) Close() error {
    var err error
    if t.ln != nil {
        err = t.ln.Close()
        t.ln = nil
    }
    // Close all tracked connections so handleConn goroutines unblock.
    t.mu.Lock()
    for c := range t.conns {
        _ = c.Close()
    }
    t.mu.Unlock()
    t.wg.Wait()
    return err
}

Closing the listener alone only stops new connections from being accepted — any handleConn goroutine blocked in Codec.Decode on an existing connection would otherwise sit there until its idle timeout fired. Close force-closes every tracked connection to unblock those reads immediately, then wg.Wait() blocks until every handleConn goroutine has actually returned, so Close doesn't report success while cleanup is still in flight.

Step 5: Build the TCP client

TransportClient and Dial

The client wraps a single persistent TCP connection plus a codec. Dial bounds the handshake with a timeout and turns on TCP keepalive, since one client connection is typically reused for the lifetime of a peer relationship (a producer talking to a topic leader, a broker replicating from another broker):

type TransportClient struct {
    mu    sync.Mutex
    conn  net.Conn
    codec *protocol.Codec
}

// dialTimeout bounds how long Dial waits for the TCP handshake to complete.
// callTimeout bounds a single request/response round trip (or a single Read/Write) on
// an already-established connection. Without it, a peer that accepts a connection but
// then never sends/reads data (blackholed, half-open) can hang a caller forever —
// nothing else on the client side enforces a deadline. Symmetric with the server's own
// handlerTimeout above.
const (
    dialTimeout = 5 * time.Second
    callTimeout = 30 * time.Second
)

// Dial opens a single TCP connection to addr and enables keepalive so the connection
// stays alive for node-to-node RPC/stream use (one connection per peer).
func Dial(addr string) (*TransportClient, error) {
    conn, err := net.DialTimeout("tcp", addr, dialTimeout)
    if err != nil {
        return nil, err
    }
    if tcp, ok := conn.(*net.TCPConn); ok {
        _ = tcp.SetKeepAlive(true)
        _ = tcp.SetKeepAlivePeriod(30 * time.Second)
    }
    return &TransportClient{conn: conn, codec: &protocol.Codec{}}, nil
}

DialWithFallback

Leader-discovery responses (FindTopicLeaderResponse, FindRaftLeaderResponse — Step 10) carry whatever address the cluster knows a node by. Inside Docker that's a Docker-internal hostname like node1:9092; a client running outside Docker, with the RPC port published to the host, can't resolve that hostname at all. DialWithFallback handles exactly that case, and every RPC client in the system dials through it instead of calling Dial directly, so the fallback logic lives in one place:

// DialWithFallback is like Dial, but when addr is an unresolvable Docker-internal
// hostname (e.g. "node1:9092" returned by FindLeader/FindRaftLeader from inside the
// cluster's network), it falls back to 127.0.0.1:<port>, which works when the caller
// is running outside Docker with the port mapped to the host.
//
// Every RPC client (admin, producer, consumer, and internal replication, which uses
// the consumer client) dials through this one function so the fallback behavior lives
// in a single place instead of being copy-pasted per client.
func DialWithFallback(addr string) (*TransportClient, error) {
    tc, err := Dial(addr)
    if err != nil {
        if strings.Contains(err.Error(), "no such host") {
            if host, port, splitErr := net.SplitHostPort(addr); splitErr == nil && strings.HasPrefix(host, "node") && port != "" {
                fallback := net.JoinHostPort("127.0.0.1", port)
                if tc2, err2 := Dial(fallback); err2 == nil {
                    return tc2, nil
                }
            }
        }
        return nil, err
    }
    return tc, nil
}

The check is narrow on purpose: it only kicks in when the dial error's text contains "no such host" (a DNS resolution failure, not a refused or reset connection) and the hostname starts with node (the cluster's own naming convention for broker containers). Anything else — a real network failure, an address that isn't shaped like a Docker service name — is returned to the caller unchanged rather than silently retried against localhost.

Step 6: Call, Read, and Write

Call: one round trip

Call is the request/response path: encode a message, then read exactly one response frame back. Both Call and ReadResponse reset the connection's deadline before doing I/O, so a peer that stops responding mid-call doesn't hang the caller past callTimeout:

// Call sends a request and reads the response. Safe for concurrent use.
func (c *TransportClient) Call(msg any) (any, error) {
    c.mu.Lock()
    defer c.mu.Unlock()
    _ = c.conn.SetDeadline(time.Now().Add(callTimeout))
    if err := c.codec.Encode(c.conn, msg); err != nil {
        return nil, err
    }
    return c.readResponse()
}

// ReadResponse reads the next frame and returns the decoded value. If the server sent an RPC
// error frame, returns (nil, *protocol.RPCError) so the client can check e.Code.
func (c *TransportClient) ReadResponse() (any, error) {
    c.mu.Lock()
    defer c.mu.Unlock()
    _ = c.conn.SetDeadline(time.Now().Add(callTimeout))
    return c.readResponse()
}

// readResponse is the internal unlocked version.
func (c *TransportClient) readResponse() (any, error) {
    mType, value, err := c.codec.Decode(c.conn)
    if err != nil {
        return nil, err
    }
    if mType == protocol.MsgRPCError {
        if r, ok := value.(protocol.RPCErrorResponse); ok {
            return nil, &protocol.RPCError{Code: r.Code, Message: r.Message}
        }
        return nil, &protocol.RPCError{Code: protocol.CodeUnknown, Message: "rpc error"}
    }
    return value, nil
}

readResponse is where MsgRPCError frames get turned back into Go errors: if the decoded message type is MsgRPCError, the caller gets (nil, *protocol.RPCError) instead of a response value, so calling code can check errors.As(err, &rpcErr) and act on rpcErr.Code — exactly the pattern ShouldReconnect and IsTopicNotReady (previous chapter) are built to consume.

Write and Read: for streaming and batch protocols

Some call patterns don't fit one write/one read — a batched fetch loop, for instance, may write once and then read frames as they arrive. Write and Read expose the two halves separately, each under the same mutex and deadline discipline as Call:

// Write sends a single frame. Safe for concurrent use.
func (c *TransportClient) Write(msg any) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    _ = c.conn.SetDeadline(time.Now().Add(callTimeout))
    return c.codec.Encode(c.conn, msg)
}

// Read reads the next frame. Safe for concurrent use.
func (c *TransportClient) Read() (any, error) {
    c.mu.Lock()
    defer c.mu.Unlock()
    _ = c.conn.SetDeadline(time.Now().Add(callTimeout))
    _, resp, err := c.codec.Decode(c.conn)
    return resp, err
}

func (c *TransportClient) Close() error {
    return c.conn.Close()
}

Note that Write and Read each take the same lock independently rather than one lock spanning both — a caller doing several writes and reads in sequence on one connection (there's exactly one connection here, not a pool) must not interleave with a concurrent caller on the same TransportClient, but the lock is released between each individual operation rather than held across a whole request.

Step 7: Wire up the RPC server

RpcServer struct

RpcServer sits on top of Transport and holds the two pieces of business state its handlers need: the TopicManager, which owns log data and topic/leader metadata, and the ConsumerManager, which tracks committed consumer offsets:

// broker/rpc/server.go
package rpc

import (
    "context"

    "github.com/mohitkumar/mlog/api/protocol"
    "github.com/mohitkumar/mlog/api/transport"
    consumermgr "github.com/mohitkumar/mlog/broker/consumer"
    "github.com/mohitkumar/mlog/broker/topic"
)

// RpcServer holds topic manager and consumer manager for TCP transport RPCs.
type RpcServer struct {
    Addr            string
    topicManager    *topic.TopicManager
    consumerManager *consumermgr.ConsumerManager
    transport       *transport.Transport
}

func NewRpcServer(addr string, topicManager *topic.TopicManager, consumerManager *consumermgr.ConsumerManager) *RpcServer {
    srv := &RpcServer{
        Addr:            addr,
        topicManager:    topicManager,
        consumerManager: consumerManager,
        transport:       transport.NewTransport(),
    }
    srv.RegisterHandlers()
    return srv
}

RegisterHandlers

Every handler is registered as a closure that type-asserts the decoded any down to its concrete request type and calls the matching method:

// RegisterHandlers registers all RPC handlers on tr. Used by Start() and by tests that run the transport themselves.
func (s *RpcServer) RegisterHandlers() {
    // Producer
    s.transport.RegisterHandler(protocol.MsgProduce, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.ProduceRequest)
        return s.Produce(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgProduceBatch, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.ProduceBatchRequest)
        return s.ProduceBatch(ctx, &r)
    })
    // Consumer
    s.transport.RegisterHandler(protocol.MsgFetch, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.FetchRequest)
        return s.Fetch(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgFetchBatch, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.FetchBatchRequest)
        return s.FetchBatch(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgCommitOffset, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.CommitOffsetRequest)
        return s.CommitOffset(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgFetchOffset, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.FetchOffsetRequest)
        return s.FetchOffset(ctx, &r)
    })
    // Topic
    s.transport.RegisterHandler(protocol.MsgCreateTopic, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.CreateTopicRequest)
        return s.CreateTopic(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgDeleteTopic, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.DeleteTopicRequest)
        return s.DeleteTopic(ctx, &r)
    })
    // Discovery
    s.transport.RegisterHandler(protocol.MsgFindTopicLeader, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.FindTopicLeaderRequest)
        return s.FindTopicLeader(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgFindRaftLeader, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.FindRaftLeaderRequest)
        return s.FindRaftLeader(ctx, &r)
    })
    s.transport.RegisterHandler(protocol.MsgListTopics, func(ctx context.Context, req any) (any, error) {
        r := req.(protocol.ListTopicsRequest)
        return s.ListTopics(ctx, &r)
    })
}

func (s *RpcServer) Start() error {
    ln, err := s.transport.Listen(s.Addr)
    if err != nil {
        return err
    }
    s.Addr = s.transport.Addr()
    go s.transport.Serve(ln)
    return nil
}

func (s *RpcServer) Stop() error {
    return s.transport.Close()
}

Start re-assigns s.Addr from the listener after binding — this is what lets a node start with :0 (any free port) and still know its own real address afterward, to advertise over Serf or hand back from FindTopicLeader.

Step 8: The Produce handler

Produce validates the request, fetches the local log for the topic, confirms this node is the topic's leader (only the leader accepts writes), and then delegates the actual append — including waiting on acks — to the topic manager:

// broker/rpc/producer.go
func (srv *RpcServer) Produce(ctx context.Context, req *protocol.ProduceRequest) (*protocol.ProduceResponse, error) {
    if req.Topic == "" {
        return nil, Err(protocol.CodeTopicRequired, "topic is required")
    }
    if len(req.Value) == 0 {
        return nil, Err(protocol.CodeValuesRequired, "value is required")
    }
    l, err := srv.topicManager.GetLog(req.Topic)
    if err != nil {
        return nil, &protocol.RPCError{Code: protocol.CodeTopicNotFound, Message: fmt.Sprintf("topic %s not found: %v", req.Topic, err)}
    }
    isLeader, _ := srv.topicManager.IsLeader(req.Topic)
    if !isLeader {
        return nil, Err(protocol.CodeNotTopicLeader, "this node is not the topic leader; produce to the topic leader")
    }

    offset, err := srv.topicManager.HandleProduce(ctx, req.Topic, l, &protocol.LogEntry{
        Value: req.Value,
    }, req.Acks)
    if err != nil {
        return nil, FromError(err)
    }
    return &protocol.ProduceResponse{Offset: offset}, nil
}

GetLog returns ErrTopicNotFound both when the topic is genuinely unknown to this node and when its log is still in the middle of opening — either way there's nothing to produce to yet, and that's exactly the code (CodeTopicNotFound) that client-side IsTopicNotReady treats as retriable, so a client that produces immediately after CreateTopic returns can retry in place rather than treating it as fatal.

ProduceBatch follows the identical shape — same validation, same leader check — but calls srv.topicManager.HandleProduceBatch(ctx, req.Topic, l, req.Values, req.Acks) and returns a ProduceBatchResponse{BaseOffset, LastOffset, Count} instead of a single offset.

Step 9: The Fetch handler's dual read path

Two different callers, one endpoint

Fetch is called by two different kinds of caller over the same RPC: a consumer reading committed messages, and a follower broker replicating uncommitted messages from the leader. req.ReplicaNodeID is how the handler tells them apart — set only by the replication path, never by a consumer client:

// broker/rpc/consumer.go
func (s *RpcServer) Fetch(ctx context.Context, req *protocol.FetchRequest) (*protocol.FetchResponse, error) {
    if req.Topic == "" {
        return nil, Err(protocol.CodeTopicRequired, "topic is required")
    }
    id := req.Id
    if id == "" {
        id = "default"
    }

    off := req.Offset
    if off == 0 && req.ReplicaNodeID == "" {
        if err := s.consumerManager.Recover(); err == nil {
            if cached, err := s.consumerManager.GetOffset(id, req.Topic); err == nil {
                off = cached
            }
        }
    }

    leaderLog, err := s.topicManager.GetLeader(req.Topic)
    if err != nil {
        return nil, &protocol.RPCError{Code: protocol.CodeTopicNotFound, Message: fmt.Sprintf("topic %s not found: %v", req.Topic, err)}
    }

    var raw []byte
    if req.ReplicaNodeID != "" {
        raw, err = leaderLog.ReadUncommitted(off)
    } else {
        raw, err = leaderLog.Read(off)
    }
    if err != nil {
        if req.ReplicaNodeID != "" {
            _ = s.topicManager.RecordReplicaLEOFromFetch(ctx, req.Topic, req.ReplicaNodeID, int64(req.Offset))
        }
        return nil, FromError(err)
    }
    // Segment returns [offset 8 bytes][value]; strip header for response
    const offWidth = 8
    if len(raw) >= offWidth {
        raw = raw[offWidth:]
    }

    if req.ReplicaNodeID != "" {
        _ = s.topicManager.RecordReplicaLEOFromFetch(ctx, req.Topic, req.ReplicaNodeID, int64(off+1))
    }
    return &protocol.FetchResponse{
        Entry: &protocol.LogEntry{Offset: off, Value: raw},
    }, nil
}

GetLeader — as opposed to Produce's GetLog — requires this node to actually be the topic's leader, since only the leader has a complete log to serve reads from; it returns ErrThisNodeNotLeader if not. Reads then branch on req.ReplicaNodeID:

• Consumer fetch (ReplicaNodeID == ""): if the client didn't pass an explicit offset, the server first tries to recover the consumer's last committed offset for (id, topic) from the ConsumerManager and resumes from there. The read itself goes through leaderLog.Read, which only returns data up to the high watermark (committed, replicated data).
• Replication fetch (ReplicaNodeID != ""): the read goes through leaderLog.ReadUncommitted, which allows reading up to the log end offset — data a follower needs to catch up on before it's technically safe for a consumer to see. After every replication read, whether it succeeded or failed, the server calls RecordReplicaLEOFromFetch to tell the topic manager how far this specific replica has fetched, which feeds the in-sync-replica (ISR) computation.

Every record on disk is stored as [offset 8 bytes][value] inside its segment; the handler strips that 8-byte offset prefix before handing the value back over the wire, since the offset is already carried separately in protocol.LogEntry.Offset.

FetchBatch repeats this same ReplicaNodeID branch inside a loop bounded by req.MaxCount, collecting entries until a read fails or the count is reached; if the very first read fails, it returns that error via FromError rather than an empty success, so a real I/O or corruption error can't be mistaken for "no data yet." CommitOffset and FetchOffset are simpler pass-throughs to s.consumerManager.CommitOffset / GetOffset, defaulting req.Id to "default" the same way Fetch does when a client doesn't supply a consumer group id.

Step 10: Topic-management and discovery handlers

These handlers forward to TopicManager, which coordinates topic creation/deletion and leader lookups through Raft (covered in the cluster chapters). Any node can answer FindTopicLeader, FindRaftLeader, and ListTopics — metadata is replicated, so there's no need to route these to a particular node:

// broker/rpc/leader.go
func (s *RpcServer) CreateTopic(ctx context.Context, req *protocol.CreateTopicRequest) (*protocol.CreateTopicResponse, error) {
    if req.Topic == "" {
        return nil, Err(protocol.CodeTopicNameRequired, "topic name is required")
    }
    resp, err := s.topicManager.CreateTopic(ctx, req)
    if err != nil {
        return nil, FromError(err)
    }
    return resp, nil
}

func (s *RpcServer) DeleteTopic(ctx context.Context, req *protocol.DeleteTopicRequest) (*protocol.DeleteTopicResponse, error) {
    if req.Topic == "" {
        return nil, Err(protocol.CodeTopicNameRequired, "topic name is required")
    }
    resp, err := s.topicManager.DeleteTopic(ctx, req)
    if err != nil {
        return nil, FromError(err)
    }
    return resp, nil
}

func (srv *RpcServer) FindTopicLeader(ctx context.Context, req *protocol.FindTopicLeaderRequest) (*protocol.FindTopicLeaderResponse, error) {
    if req.Topic == "" {
        return nil, Err(protocol.CodeTopicRequired, "topic is required")
    }

    leaderAddr, err := srv.topicManager.GetTopicLeaderRPCAddr(req.Topic)
    if err != nil {
        return nil, &protocol.RPCError{Code: protocol.CodeTopicNotFound, Message: fmt.Sprintf("topic %s not found: %v", req.Topic, err)}
    }

    return &protocol.FindTopicLeaderResponse{
        LeaderAddr: leaderAddr,
    }, nil
}

// FindRaftLeader returns the RPC address of the current Raft (metadata) leader.
// Any node can answer; clients should send create-topic and other metadata ops to this address.
func (srv *RpcServer) FindRaftLeader(ctx context.Context, req *protocol.FindRaftLeaderRequest) (*protocol.FindRaftLeaderResponse, error) {
    addr, err := srv.topicManager.GetRaftLeaderRPCAddr()
    if err != nil {
        return nil, &protocol.RPCError{Code: protocol.CodeRaftLeaderUnavailable, Message: err.Error()}
    }
    return &protocol.FindRaftLeaderResponse{RaftLeaderAddr: addr}, nil
}

// ListTopics returns all topics with leader and replica info. Any node can answer.
func (srv *RpcServer) ListTopics(ctx context.Context, req *protocol.ListTopicsRequest) (*protocol.ListTopicsResponse, error) {
    return srv.topicManager.ListTopics(), nil
}

Note the addresses returned by FindTopicLeader and FindRaftLeader are exactly what a caller then passes to transport.DialWithFallback (Step 5) — this is the pairing that lets a client running outside Docker still reach a leader whose address is a Docker-internal hostname.

Step 11: Map domain errors to RPC codes

Handlers throughout broker/rpc construct errors two ways: Err(code, message) for a condition the handler itself detects (bad input, wrong node), and FromError(err) for an error bubbling up from deeper layers (broker/topic, broker/log, broker/segment, broker/cluster/raft) that needs translating into a wire-level code:

// broker/rpc/error.go
package rpc

import (
    "errors"

    "github.com/mohitkumar/mlog/api/protocol"
    raft "github.com/mohitkumar/mlog/broker/cluster/raft"
    "github.com/mohitkumar/mlog/broker/log"
    "github.com/mohitkumar/mlog/broker/segment"
    "github.com/mohitkumar/mlog/broker/topic"
)

// Err returns an RPCError with the given code and message; the transport sends it to the client.
func Err(code int32, message string) error {
    return &protocol.RPCError{Code: code, Message: message}
}

// CodeFor returns the protocol RPC code for the given error.
func CodeFor(err error) int32 {
    if err == nil {
        return 0
    }
    switch {
    case errors.Is(err, topic.ErrTopicNotFound):
        return protocol.CodeTopicNotFound
    case errors.Is(err, topic.ErrTopicExists):
        return protocol.CodeTopicExists
    case errors.Is(err, topic.ErrNotEnoughNodes):
        return protocol.CodeNotEnoughNodes
    case errors.Is(err, topic.ErrCannotReachLeader):
        return protocol.CodeCannotReachLeader
    case errors.Is(err, topic.ErrThisNodeNotLeader):
        return protocol.CodeNotTopicLeader
    case errors.Is(err, topic.ErrInvalidAckMode):
        return protocol.CodeInvalidAckMode
    case errors.Is(err, topic.ErrTimeoutCatchUp):
        return protocol.CodeTimeoutCatchUp
    case errors.Is(err, topic.ErrValuesEmpty):
        return protocol.CodeValuesRequired
    case errors.Is(err, log.ErrLogOffsetOutOfRange), errors.Is(err, segment.ErrSegmentOffsetNotFound), errors.Is(err, log.ErrLogOffsetBeyondHW):
        return protocol.CodeReadOffset
    case errors.Is(err, raft.ErrRaftNoLeader), errors.Is(err, raft.ErrRaftNodeNotFound):
        return protocol.CodeRaftLeaderUnavailable
    default:
        return protocol.CodeUnknown
    }
}

// FromError converts an error to an RPCError with the appropriate code for the client.
func FromError(err error) error {
    if err == nil {
        return nil
    }
    return &protocol.RPCError{Code: CodeFor(err), Message: err.Error()}
}

CodeFor is a flat errors.Is switch over sentinel errors owned by the packages that actually detect these conditions — broker/topic, broker/log, broker/segment, broker/cluster/raft — so broker/rpc doesn't need to know how those errors are constructed, only what code each one maps to on the wire. Anything not matched falls through to protocol.CodeUnknown, which the client treats as a non-retriable, non-reconnectable failure (see ShouldReconnect in the wire protocol chapter).

Step 12: The produce request/response flow

Client                          Server
  │                                │
  │  [MsgType][Size][protobuf]  →  │  1. Codec.Decode() reads frame
  │                                │  2. Look up handler for message type
  │                                │  3. Type-assert → ProduceRequest
  │                                │  4. Call TopicManager.HandleProduce()
  │                                │  5. Encode ProduceResponse → protobuf
  │  ← [MsgType][Size][protobuf]   │  6. Codec.Encode() writes response frame
  │                                │

Step 13: The error response flow

  │                                │  4. Handler returns *RPCError
  │  ← [MsgRPCError][Size][...]   │  5. Encode RPCErrorResponse frame
  │                                │
  │  Client checks ShouldReconnect │
  │  If true: rediscover leader    │
  │  via FindTopicLeader/          │
  │  FindRaftLeader, then          │
  │  DialWithFallback              │

Summary

Component	File	Purpose
Transport (server)	api/transport/transport.go	TCP listener, accept connections, decode frames via Codec, dispatch to handlers with a per-call context timeout, encode response. Tracks connections for graceful shutdown.
Transport (client)	api/transport/transport.go	Dial opens a keepalive TCP connection; DialWithFallback retries against 127.0.0.1 when a Docker-internal nodeN hostname fails to resolve. Call sends a request and reads one response, auto-converting MsgRPCError frames to *RPCError. Write/Read expose the two halves separately for streaming/batch use.
RPC server	broker/rpc/server.go	RpcServer holds *topic.TopicManager and *consumer.ConsumerManager, maps message types to handler closures, Start/Stop wrap the transport's listen/serve/close lifecycle.
Produce handler	broker/rpc/producer.go	Validates request, confirms this node is the topic leader via IsLeader, calls TopicManager.HandleProduce/HandleProduceBatch with the requested ack mode.
Fetch handler	broker/rpc/consumer.go	Dual-path on ReplicaNodeID: empty means a consumer reading committed data (Read, with offset recovery from ConsumerManager); set means a replica reading uncommitted data (ReadUncommitted) and reporting its LEO via RecordReplicaLEOFromFetch. Strips the 8-byte segment offset header before returning.
Topic/leader discovery	broker/rpc/leader.go	CreateTopic/DeleteTopic forward to TopicManager; FindTopicLeader/FindRaftLeader/ListTopics are answerable by any node since metadata is Raft-replicated.
Error mapping	broker/rpc/error.go	Err() builds an *RPCError directly; CodeFor()/FromError() map sentinel errors from topic/log/segment/raft to protocol error codes for the wire.

The next page covers Serf-based cluster discovery (broker/cluster/discovery) — how nodes find each other, gossip liveness, and exchange the tags used to build the RPC addresses this chapter's handlers hand back to clients.