The Wire Protocol — Framing, Message Types, and a Protobuf Codec

Every request and response in the system — produce, fetch, topic administration, leader discovery — travels over one TCP connection as a length-prefixed frame with a protobuf-encoded payload. In this section you'll define the frame format, the full set of message types, the Go-native request/response structs the rest of the codebase actually programs against, and the Codec that translates between the two. All of this lives in api/protocol/.

A design choice worth calling out up front: this is not gRPC. There's no .proto service definition and no generated client/server stubs. Protobuf here is used purely as a compact, versionable payload format inside a hand-rolled framing protocol — you get schema evolution and fast (de)serialization without pulling in a full RPC framework.

Step 1: Define the frame format

Frame header

Every message on the wire starts with a 6-byte header — a 2-byte message type and a 4-byte payload length — followed by the protobuf-encoded payload:

┌───────────────┬────────────────┬─────────────────────┐
│  Message Type │  Payload Size  │  Payload (protobuf) │
│  2 bytes      │  4 bytes       │  PayloadSize bytes  │
│  (uint16)     │  (uint32)      │                     │
└───────────────┴────────────────┴─────────────────────┘

// api/protocol/frame.go
package protocol

import "encoding/binary"

type MessageType uint16

var byteOrder = binary.BigEndian

const messageTypeSize = 2
const messageSizeSize = 4

// frameHeaderSize is the length prefix size (2 bytes for message type, 4 bytes for message size)
const frameHeaderSize = messageTypeSize + messageSizeSize

// MaxFrameSize is the maximum allowed frame payload size (4MB) to avoid abuse.
const MaxFrameSize = 4 * 1024 * 1024

A fixed-size, big-endian header keeps framing trivial: read 6 bytes, decode a type and a length, then read exactly that many more bytes for the payload. MaxFrameSize caps payloads at 4MB so a corrupt or malicious length field can't make the reader allocate an unbounded buffer.

MessageType enum

Every distinct request or response gets its own constant, generated with iota:

const (
    MsgReplicateStream MessageType = iota
    MsgReplicateResp
    MsgProduce
    MsgProduceResp
    MsgProduceBatch
    MsgProduceBatchResp
    MsgFetch
    MsgFetchResp
    MsgFetchBatch
    MsgFetchBatchResp
    MsgFetchStream
    MsgFetchStreamResp
    MsgCommitOffset
    MsgCommitOffsetResp
    MsgFetchOffset
    MsgFetchOffsetResp
    MsgCreateTopic
    MsgCreateTopicResp
    MsgDeleteTopic
    MsgDeleteTopicResp
    MsgRPCError
    MsgFindTopicLeader
    MsgFindTopicLeaderResp
    MsgFindRaftLeader
    MsgFindRaftLeaderResp
    MsgListTopics
    MsgListTopicsResp
    MsgApplyIsrUpdateEvent
    MsgApplyIsrUpdateEventResp
)

MsgRPCError is special: it isn't paired with a specific request. Any handler can return it in place of the expected response type when something goes wrong (Step 5 covers this).

Step 2: Write the protobuf schema

mlog.proto

Every message type on the wire has a matching protobuf message, defined in one .proto file:

// api/protocol/proto/mlog.proto
syntax = "proto3";

package mlog.protocol;

option go_package = "github.com/mohitkumar/mlog/api/protocol/pb";

message LogEntry {
  uint64 offset = 1;
  bytes value = 2;
}

message ProduceRequest {
  string topic = 1;
  bytes value = 2;
  int32 acks = 3;
}

message ProduceResponse {
  uint64 offset = 1;
}

message FetchBatchRequest {
  string topic = 1;
  string id = 2;
  uint64 offset = 3;
  uint32 max_count = 4;
  string replica_node_id = 5;
}

message FetchBatchResponse {
  repeated LogEntry entries = 1;
}

// ... one message per request/response type, plus topic administration,
// metadata events, and Raft snapshot messages (Step 6).

Generate the Go types from it:

protoc --go_out=. --go_opt=paths=source_relative \
    api/protocol/proto/mlog.proto

This produces api/protocol/pb/mlog.pb.go — plain generated structs (pb.ProduceRequest, pb.FetchBatchResponse, and so on) with proto.Marshal/proto.Unmarshal support. Nothing outside api/protocol ever imports pb directly; it's an implementation detail of the codec you'll write in Step 4.

Step 3: Define the Go-native request/response types

Why a separate set of types from the generated protobuf structs

Handler signatures, client code, and tests all work with a set of plain Go structs — not the generated pb.* types directly. Keeping them separate means callers never see generated-struct internals (unexported fields, XXX_ bookkeeping) and the wire encoding can change without touching a single call site:

// api/protocol/types.go
package protocol

type LogEntry struct {
    Offset uint64
    Value  []byte
}

type AckMode int32

const (
    AckNone   AckMode = 0
    AckLeader AckMode = 1
    AckAll    AckMode = 2
)

type ProduceRequest struct {
    Topic string
    Value []byte
    Acks  AckMode
}
type ProduceResponse struct {
    Offset uint64
}

type ProduceBatchRequest struct {
    Topic  string
    Values [][]byte
    Acks   AckMode
}
type ProduceBatchResponse struct {
    BaseOffset uint64
    LastOffset uint64
    Count      uint32
}

type FetchRequest struct {
    Topic         string
    Id            string
    Offset        uint64
    ReplicaNodeID string // when set, leader uses ReadUncommitted and stores this offset as replica LEO
}
type FetchResponse struct {
    Entry *LogEntry
}

type FetchBatchRequest struct {
    Topic         string
    Id            string
    Offset        uint64
    MaxCount      uint32
    ReplicaNodeID string
}
type FetchBatchResponse struct {
    Entries []*LogEntry
}

Topic administration and discovery messages follow the same pattern — CreateTopicRequest/CreateTopicResponse, DeleteTopicRequest/DeleteTopicResponse, FindTopicLeaderRequest/FindTopicLeaderResponse, FindRaftLeaderRequest/FindRaftLeaderResponse, ListTopicsRequest/ListTopicsResponse (carrying TopicInfo/ReplicaInfo), and CommitOffsetRequest/FetchOffsetRequest for consumer offset tracking. ApplyIsrUpdateEventRequest/Response is the one message sent to the Raft leader specifically to propose an ISR change.

Converting between Go types and protobuf types

api/protocol/convert.go holds the small helper functions that translate between the two representations — one pair per nested type that needs it (a bare struct with only scalar fields, like ProduceRequest, converts inline in the codec; something with a slice of pointers, like a list of LogEntry, gets its own helper):

// api/protocol/convert.go
package protocol

func logEntryToPB(e *LogEntry) *pb.LogEntry {
    if e == nil {
        return nil
    }
    return &pb.LogEntry{Offset: e.Offset, Value: e.Value}
}

func logEntryFromPB(m *pb.LogEntry) *LogEntry {
    if m == nil {
        return nil
    }
    return &LogEntry{Offset: m.Offset, Value: m.Value}
}

func logEntriesToPB(es []*LogEntry) []*pb.LogEntry {
    if es == nil {
        return nil
    }
    out := make([]*pb.LogEntry, len(es))
    for i, e := range es {
        out[i] = logEntryToPB(e)
    }
    return out
}

topicInfosToPB/topicInfosFromPB and replicaInfosToPB/replicaInfosFromPB follow the identical shape for ListTopicsResponse's nested TopicInfo/ReplicaInfo slices.

Step 4: Build the Codec

Encode: Go struct → protobuf → frame

Codec.Encode type-switches on the concrete Go type passed in, builds the matching pb.* message, marshals it, and writes it as a framed message:

// api/protocol/codec.go
package protocol

type Codec struct{}

func (c *Codec) Encode(w io.Writer, msg any) error {
    var mType MessageType
    var payload []byte
    var err error
    switch v := msg.(type) {
    case ProduceRequest:
        mType = MsgProduce
        payload, err = proto.Marshal(&pb.ProduceRequest{Topic: v.Topic, Value: v.Value, Acks: int32(v.Acks)})
    case *ProduceRequest:
        mType = MsgProduce
        payload, err = proto.Marshal(&pb.ProduceRequest{Topic: v.Topic, Value: v.Value, Acks: int32(v.Acks)})
    case ProduceResponse:
        mType = MsgProduceResp
        payload, err = proto.Marshal(&pb.ProduceResponse{Offset: v.Offset})
    // ... one case (value and pointer) per message type ...
    case FetchBatchResponse:
        mType = MsgFetchBatchResp
        payload, err = proto.Marshal(&pb.FetchBatchResponse{Entries: logEntriesToPB(v.Entries)})
    default:
        return ErrUnknownMessageType(mType)
    }
    if err != nil {
        return err
    }
    return c.encodeFrame(w, mType, payload)
}

Each case appears twice — once for the value type, once for the pointer type — since callers pass both (tc.Call(req) with a value, but some code builds requests as pointers). That's mechanical repetition, not a design choice worth agonizing over: the switch is written by hand once and rarely touched again as new message types are added at the end.

encodeFrame and decodeFrame

The actual byte-level framing is two small helpers:

func (c *Codec) encodeFrame(w io.Writer, mType MessageType, payload []byte) error {
    length := uint32(len(payload))
    if length > MaxFrameSize {
        return ErrFrameTooLarge
    }
    // Stack-allocated header avoids a heap allocation per frame.
    var header [frameHeaderSize]byte
    byteOrder.PutUint16(header[:], uint16(mType))
    byteOrder.PutUint32(header[messageTypeSize:], length)
    // Single write: combine header + payload to avoid two syscalls.
    buf := make([]byte, frameHeaderSize+len(payload))
    copy(buf, header[:])
    copy(buf[frameHeaderSize:], payload)
    _, err := w.Write(buf)
    return err
}

func (c *Codec) decodeFrame(r io.Reader) (mType MessageType, payload []byte, err error) {
    var header [frameHeaderSize]byte
    if _, err := io.ReadFull(r, header[:]); err != nil {
        return 0, nil, err
    }
    mType = MessageType(byteOrder.Uint16(header[:]))
    length := byteOrder.Uint32(header[messageTypeSize:])
    if length > MaxFrameSize {
        return 0, nil, ErrFrameTooLarge
    }
    payload = make([]byte, length)
    if _, err := io.ReadFull(r, payload); err != nil {
        return 0, nil, err
    }
    return mType, payload, nil
}

Two small performance details worth noting: the frame header is a fixed-size array ([frameHeaderSize]byte), so building it never touches the heap, and the header and payload are copied into one combined buffer before the single Write call — sending them separately would cost an extra syscall per message on a non-buffered connection.

Decode: frame → protobuf → Go struct

Decode is the mirror image — read a frame, look at its message type, unmarshal into the matching pb.* struct, and convert back to the plain Go type:

func (c *Codec) Decode(r io.Reader) (MessageType, any, error) {
    mType, payload, err := c.decodeFrame(r)
    if err != nil {
        return 0, nil, err
    }
    switch mType {
    case MsgProduce:
        var m pb.ProduceRequest
        if err := proto.Unmarshal(payload, &m); err != nil {
            return mType, nil, err
        }
        return mType, ProduceRequest{Topic: m.Topic, Value: m.Value, Acks: AckMode(m.Acks)}, nil
    case MsgFetchBatchResp:
        var m pb.FetchBatchResponse
        if err := proto.Unmarshal(payload, &m); err != nil {
            return mType, nil, err
        }
        return mType, FetchBatchResponse{Entries: logEntriesFromPB(m.Entries)}, nil
    // ... one case per message type ...
    case MsgRPCError:
        var m pb.RPCErrorResponse
        if err := proto.Unmarshal(payload, &m); err != nil {
            return mType, nil, err
        }
        return mType, RPCErrorResponse{Code: m.Code, Message: m.Message}, nil
    default:
        return 0, nil, ErrUnknownMessageType(mType)
    }
}

The caller (the transport layer, next chapter) gets back the MessageType alongside the decoded value so it can dispatch to the right handler without a second type assertion.

Step 5: Define RPC error codes

Error codes clients can act on

Beyond a bare error message, the server needs to tell a client what kind of failure occurred, so the client can decide whether to retry, reconnect, or give up. Each error condition gets a numeric code:

// api/protocol/types.go
const (
    CodeUnknown int32 = iota
    CodeTopicRequired
    CodeTopicNameRequired
    CodeTopicNotFound
    CodeNotTopicLeader
    CodeValuesRequired
    CodeReadOffset
    CodeCommitOffset
    CodeRecoverOffsets
    CodeReplicaCountInvalid
    CodeLeaderAddrRequired
    CodeRaftLeaderUnavailable
    CodeTopicExists
    CodeNotEnoughNodes
    CodeCannotReachLeader
    CodeInvalidAckMode
    CodeTimeoutCatchUp
)

// RPCErrorResponse is sent by the server when a handler returns an error.
type RPCErrorResponse struct {
    Code    int32  `json:"code"`
    Message string `json:"message"`
}

// RPCError is returned by the transport client when the server sends an RPCErrorResponse.
type RPCError struct {
    Code    int32
    Message string
}

func (e *RPCError) Error() string { return e.Message }

A handler that hits, say, "this node isn't the leader for this topic" doesn't just return a generic error — it returns (or the RPC layer wraps it into) an RPCErrorResponse{Code: CodeNotTopicLeader, ...}, encoded as MsgRPCError instead of the expected response type. The client's Codec.Decode sees MsgRPCError and hands back an *RPCError, which calling code can type-assert on.

ShouldReconnect: deciding when to abandon a connection

Given an error, ShouldReconnect answers "is this connection worth keeping, or should the caller find a new address and redial?":

func ShouldReconnect(err error) bool {
    if err == nil {
        return false
    }
    // Server returned a structured RPC error code: reconnect on leader/topic/raft changes.
    var rpcErr *RPCError
    if errors.As(err, &rpcErr) {
        switch rpcErr.Code {
        case CodeNotTopicLeader, CodeTopicNotFound, CodeRaftLeaderUnavailable:
            return true
        default:
            return false
        }
    }
    // EOF means the connection was closed by the peer.
    if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
        return true
    }
    // Syscall-level connection errors: reset, broken pipe, refused.
    if errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNREFUSED) {
        return true
    }
    // net.Error covers timeouts and other transient network failures.
    var netErr net.Error
    if errors.As(err, &netErr) {
        return true
    }
    // net.ErrClosed: use of closed network connection.
    if errors.Is(err, net.ErrClosed) {
        return true
    }
    return false
}

This one function is the shared decision point behind every reconnect-on-failure path in the system — the producer and consumer client libraries (covered later) and the broker's own replication engine all call it after an RPC fails, rather than each re-implementing "is this error connection-shaped?" independently.

IsTopicNotReady: a narrower, non-reconnecting retry signal

A related but distinct case: right after CreateTopic returns, a given node may not have opened the topic's local log yet (it's on a poll — see the topic management chapter). That's not a reason to reconnect anywhere; the same node will become ready shortly, so the right response is to retry the same request after a short pause:

func IsTopicNotReady(err error) bool {
    var rpcErr *RPCError
    if errors.As(err, &rpcErr) {
        return rpcErr.Code == CodeTopicNotFound || rpcErr.Code == CodeNotTopicLeader
    }
    return false
}

Notice CodeTopicNotFound/CodeNotTopicLeader appear in both ShouldReconnect and IsTopicNotReady — the two functions answer different questions about the same codes. Whether a caller treats a given error as "redial" or "retry in place" depends on which of the two it checks, and where in the retry stack it's sitting (client libraries check both, in that order — the producer/consumer chapter covers the full retry stack).

Step 6: Encode Raft metadata events

The event envelope

Cluster metadata changes (create a topic, change a topic's leader, update ISR membership) are proposed through Raft as small, typed events. api/protocol/metadata.go defines the envelope and the encode/decode helpers for each event payload:

// api/protocol/metadata.go
package protocol

type MetadataEventType uint16

const (
    MetadataEventTypeCreateTopic MetadataEventType = iota
    MetadataEventTypeDeleteTopic
    MetadataEventTypeLeaderChange
    MetadataEventTypeIsrUpdate
    MetadataEventTypeAddNode
    MetadataEventTypeRemoveNode
    MetadataEventTypeUpdateNode
)

type MetadataEvent struct {
    EventType MetadataEventType
    Data      []byte
}

func EncodeMetadataEvent(ev *MetadataEvent) ([]byte, error) {
    return proto.Marshal(&pb.MetadataEvent{EventType: uint32(ev.EventType), Data: ev.Data})
}

func DecodeMetadataEvent(data []byte) (*MetadataEvent, error) {
    var m pb.MetadataEvent
    if err := proto.Unmarshal(data, &m); err != nil {
        return nil, err
    }
    return &MetadataEvent{EventType: MetadataEventType(m.EventType), Data: m.Data}, nil
}

type CreateTopicEvent struct {
    Topic          string
    ReplicaCount   uint32
    LeaderNodeID   string
    LeaderEpoch    int64
    ReplicaNodeIds []string
}

func EncodeCreateTopicEvent(e CreateTopicEvent) ([]byte, error) {
    return proto.Marshal(&pb.CreateTopicEvent{
        Topic:        e.Topic,
        ReplicaCount: e.ReplicaCount,
        LeaderId:     e.LeaderNodeID,
        LeaderEpoch:  e.LeaderEpoch,
        Replicas:     e.ReplicaNodeIds,
    })
}

DeleteTopicEvent, LeaderChangeEvent, and IsrUpdateEvent follow the same Encode*/Decode* shape — a small Go struct, a matching pb.* message, proto.Marshal/proto.Unmarshal. AddNodeEvent, RemoveNodeEvent, and UpdateNodeEvent are defined here too, for representing node membership as a Raft-replicated event; the cluster chapter covers how membership is actually derived directly from Raft's own voter configuration plus Serf's gossiped tags, so these three types are defined without a caller that applies them.

The Raft-facing package that drives consensus (broker/cluster/raft) defines its own copy of this same envelope and event set locally rather than importing this one — worth knowing if you go looking for where EncodeCreateTopicEvent is actually called from in the cluster chapter and land in this file first instead.

Step 7: Encode replication batches

Why a separate binary format

Client-facing RPCs (produce, fetch) go through the protobuf codec above. But when the replication engine streams raw log data from a leader's segment files, it uses its own compact binary format instead — closer to Kafka's own record-batch format, and avoiding a protobuf message per record when a batch might contain thousands of them:

Batch header (25 bytes):
┌──────────────┬──────────────┬──────────────┬──────────┬────────────┬──────────────────┐
│ Base Offset  │ Batch Length │ Leader Epoch │   CRC    │ Attributes │ Last Offset Delta│
│  8 bytes     │  4 bytes     │  4 bytes     │ 4 bytes  │  1 byte    │  4 bytes         │
└──────────────┴──────────────┴──────────────┴──────────┴────────────┴──────────────────┘

Followed by records:
┌──────────────┬──────────────┬──────────────────────┐
│   Offset     │    Size      │    Value             │
│   8 bytes    │   4 bytes    │   Size bytes         │
├──────────────┼──────────────┼──────────────────────┤
│   Offset     │    Size      │    Value             │
│   ...        │   ...        │   ...                │
└──────────────┴──────────────┴──────────────────────┘

// api/protocol/replication_batch.go
package protocol

const (
    LeaderEpoch                 = 0
    CompressionNone             = 0
    replicationBatchHeaderSize  = 8 + 4 + 4 + 4 + 1 + 4 // 25 bytes
    replicationRecordHeaderSize = 8 + 4                 // offset + size
)

type ReplicationRecord struct {
    Offset int64
    Value  []byte
}

func EncodeReplicationBatch(records []ReplicationRecord) ([]byte, error) {
    if len(records) == 0 {
        return nil, nil
    }
    baseOffset := records[0].Offset
    lastOffset := records[len(records)-1].Offset
    lastOffsetDelta := int32(lastOffset - baseOffset)

    var recordBuf []byte
    for _, r := range records {
        recordBuf = append(recordBuf, encodeRecord(r)...)
    }
    batchLength := int32(len(recordBuf))
    crc := crc32.ChecksumIEEE(recordBuf)

    buf := make([]byte, replicationBatchHeaderSize+len(recordBuf))
    off := 0
    replicationByteOrder.PutUint64(buf[off:off+8], uint64(baseOffset))
    off += 8
    replicationByteOrder.PutUint32(buf[off:off+4], uint32(batchLength))
    off += 4
    replicationByteOrder.PutUint32(buf[off:off+4], uint32(LeaderEpoch))
    off += 4
    replicationByteOrder.PutUint32(buf[off:off+4], crc)
    off += 4
    buf[off] = byte(CompressionNone)
    off++
    replicationByteOrder.PutUint32(buf[off:off+4], uint32(lastOffsetDelta))
    off += 4
    copy(buf[off:], recordBuf)
    return buf, nil
}

Decoding and verifying a batch

DecodeReplicationBatch reverses the encoding and, critically, verifies the CRC before trusting the record bytes:

func DecodeReplicationBatch(data []byte) ([]ReplicationRecord, error) {
    if len(data) < replicationBatchHeaderSize {
        return nil, io.ErrUnexpectedEOF
    }
    baseOffset := int64(replicationByteOrder.Uint64(data[0:8]))
    batchLength := int32(replicationByteOrder.Uint32(data[8:12]))
    crcStored := replicationByteOrder.Uint32(data[16:20])
    recordBuf := data[replicationBatchHeaderSize:]
    if int32(len(recordBuf)) != batchLength {
        return nil, io.ErrUnexpectedEOF
    }
    if crc32.ChecksumIEEE(recordBuf) != crcStored {
        return nil, ErrReplicationBatchCRC
    }
    var records []ReplicationRecord
    p := recordBuf
    for len(p) >= replicationRecordHeaderSize {
        offset := int64(replicationByteOrder.Uint64(p[0:8]))
        size := int32(replicationByteOrder.Uint32(p[8:12]))
        if size < 0 || int(replicationRecordHeaderSize)+int(size) > len(p) {
            break
        }
        value := make([]byte, size)
        copy(value, p[12:12+size])
        records = append(records, ReplicationRecord{Offset: offset, Value: value})
        p = p[12+size:]
    }
    return records, nil
}

The CRC covers only the record bytes, not the header — this catches corruption in the payload (a truncated write, a flipped bit on disk) without needing to checksum fields that are already validated structurally (length, offset ordering).

Step 8: Define protocol-level errors

// api/protocol/error.go
package protocol

var (
    ErrFrameTooLarge       = errors.New("protocol: frame exceeds max size")
    ErrReplicationBatchCRC = errors.New("protocol: replication batch CRC mismatch")
)

func ErrUnknownMessageType(mType MessageType) error {
    return fmt.Errorf("protocol: can not encode type: %d", mType)
}

These three cover the failure modes intrinsic to the wire format itself — as opposed to RPCError's codes, which describe application-level failures (topic not found, not the leader, and so on).

Summary

Component	File	Purpose
Frame format	frame.go	6-byte header (2-byte type + 4-byte length) + payload. MessageType enum.
Go-native types	types.go	Request/response structs every handler, client, and test actually programs against.
Protobuf schema	proto/mlog.proto, generated pb/mlog.pb.go	Wire encoding for every message type, plus Raft metadata events and snapshot messages.
convert.go	convert.go	Small helpers translating nested Go types (LogEntry, TopicInfo) to/from their pb.* counterparts.
Codec	codec.go	Encode/Decode: type-switch each Go struct to/from protobuf, framed with encodeFrame/decodeFrame.
RPCError / ShouldReconnect / IsTopicNotReady	types.go	Structured error codes, and the two decision functions clients use to choose between reconnecting and retrying in place.
Metadata events	metadata.go	Envelope and per-event-type encode/decode for Raft-proposed cluster changes.
Replication batch format	replication_batch.go	Compact binary (non-protobuf) format for bulk record transfer during replication, with a CRC32 check.

With the wire protocol defined, the next page builds the transport and RPC layer — the TCP server and client that actually move these frames over the network, and the handlers that turn decoded requests into calls against the topic manager.