Malachi. Wire
(malachi v0.22.7)
View Source
The binary wire protocol for the NorthGuard log client: a length-prefixed, request/response
framing that replaces the JSON+base64 line protocol (measured ~29% fewer bytes and 9-17x less
serialization CPU in benchmark/protocol_bench.exs).
Frame: <<len::32, body::binary-size(len)>>
Request: <<api_key::16, correlation_id::32, payload::binary>>
Response: <<correlation_id::32, error_code::16, payload::binary>>correlation_id lets a client pipeline (match each response to its request). Records on the wire carry
no offset: the client never sees one; the opaque cursor carries position - so this is a distinct
encoding from Malachi.Log.Record.encode/1 (the on-disk frame, which includes the offset). Keys and cursors are
length-prefixed byte strings with a presence flag (nil vs empty are distinct). Pure, this module
only encodes/decodes binaries; the socket wiring is B1b.
decode_frame/1 is tolerant (returns :incomplete for a partial frame), but the payload decoders
(decode_request/1, decode_produce_req/1, …) assume a well-formed body and raise on a malformed
one. A frame body comes from an untrusted client, so B1b must decode inside a try and answer an error
(or close) on a raise: keeping the malformed-input handling at the connection boundary, not in the codec.
Stability and compatibility
This framing is the compatibility contract with every client: the Node CLI, the Elixir client, and any
future SDK. Two things are stable and must stay so: the byte layout of each frame above, and the
api_key numbers (currently 0..23, @auth through @list_users_with_roles). Clients are compiled against them, so
a running cluster and its clients agree on the wire only as long as both hold.
A change is breaking (every deployed client must update in lockstep, so it cannot ship in a normal release) when it:
- changes the layout or meaning of an existing
api_key's request or response payload, - reuses or renumbers an
api_keythat already shipped, or - changes how
error_codeor its reason string is encoded.
Evolve the protocol additively instead: every new operation, and every extension of an existing one,
takes the next free api_key number. Appending a field to an existing payload is not compatible here,
because the decoders match a payload to its exact end ({value, <<>>} = take_str(rest)): trailing bytes
raise a MatchError rather than being ignored, so an old peer cannot skip a field a newer one appended. A
shipped frame is therefore frozen; a change means a new key. This mirrors the discipline the Apache Iggy
project keeps around its own binary protocol: extend, do not rewrite.
Summary
Functions
Decodes an error response payload (see encode_error/2) back to its reason string.
Peels one frame off a buffer: {:ok, body, rest} or :incomplete if the frame is not all here.
Like decode_frame/1 but bounds the frame: as soon as the 4-byte length prefix is readable, a declared
length over max_size is rejected with {:error, :frame_too_large}, before the body is buffered:
so a hostile length prefix cannot force the server to accumulate unbounded memory.
list_policies (19) takes an empty request.
bind_topic_policy (20): <<put_str(topic), put_str(name)>>, a nil name detaches the topic.
define_policy (17): <<put_str(name), fields>>.
delete_policy (18): <<put_str(name), force::8>>, force 1 deletes a policy topics are bound to.
An error response frame (error_code 1) whose payload is reason as a string.
Wraps a body in a length-prefixed frame.
get_topic_policy (21): put_str(topic).
list_policies response: <<count::32, (put_str(name), fields)*>>, in the order given.
A success response frame (error_code 0) for correlation_id carrying payload.
The records half of a produce request (the count, then each record), which does not depend on the
topic. A sender that reuses one batch across topics encodes it once with this and joins it to each
topic with encode_produce_req_with/2; encode_produce_req(topic, records) is the same bytes as one
binary.
A produce request for topic from records already encoded by encode_produce_records/1, as iodata, so
the records are referenced rather than copied; encode_request_iodata/3 frames it the same way.
Encodes a record for the wire (no offset: the client never sees one).
encode_request/3 for a payload given as iodata, returned as iodata: the frame header and the payload
side by side, so a sender writing it to a socket never copies the payload into a new binary. The bytes
are exactly those of encode_request/3 over the same payload.
get_topic_policy response: <<put_str(topic), put_str(policy), resolution::8, has_definition::8, fields?, count::16, (put_str(name), value, origin::8)*>>. Resolution is 0 none, 1 resolved, 2
unresolved; origin is 0 global, 1 policy, 2 unresolved_backstop. The definition's fields follow only when
has_definition is 1.
Types
@type api_key() :: 0..23
@type error_code() :: non_neg_integer()
@type policy_value() :: non_neg_integer() | String.t() | nil
A policy field as the wire carries it.
@type topic_policy_resp() :: %{ topic: String.t(), policy: String.t() | nil, resolution: :none | :resolved | :unresolved, definition: [{String.t(), policy_value()}] | nil, effective: [ {String.t(), policy_value(), :global | :policy | :unresolved_backstop} ] }
A topic's policy as get_topic_policy answers it.
Functions
@spec auth_key() :: api_key()
@spec decode_define_policy_req(binary()) :: {String.t() | nil, [{String.t(), policy_value()}]}
Decodes an error response payload (see encode_error/2) back to its reason string.
Peels one frame off a buffer: {:ok, body, rest} or :incomplete if the frame is not all here.
@spec decode_frame(binary(), non_neg_integer()) :: {:ok, binary(), binary()} | :incomplete | {:error, :frame_too_large}
Like decode_frame/1 but bounds the frame: as soon as the 4-byte length prefix is readable, a declared
length over max_size is rejected with {:error, :frame_too_large}, before the body is buffered:
so a hostile length prefix cannot force the server to accumulate unbounded memory.
@spec decode_list_policies_req(binary()) :: :ok
list_policies (19) takes an empty request.
@spec decode_list_policies_resp(binary()) :: [ {String.t(), [{String.t(), policy_value()}]} ]
@spec decode_record(binary()) :: {Malachi.Log.Record.t(), binary()}
@spec decode_request(binary()) :: {api_key(), non_neg_integer(), binary()}
@spec decode_response(binary()) :: {non_neg_integer(), error_code(), binary()}
@spec decode_topic_policy_resp(binary()) :: topic_policy_resp()
bind_topic_policy (20): <<put_str(topic), put_str(name)>>, a nil name detaches the topic.
@spec encode_define_policy_req(String.t(), [{String.t(), policy_value()}]) :: binary()
define_policy (17): <<put_str(name), fields>>.
delete_policy (18): <<put_str(name), force::8>>, force 1 deletes a policy topics are bound to.
@spec encode_error(non_neg_integer(), term()) :: binary()
An error response frame (error_code 1) whose payload is reason as a string.
Wraps a body in a length-prefixed frame.
get_topic_policy (21): put_str(topic).
@spec encode_list_policies_resp([{String.t(), [{String.t(), policy_value()}]}]) :: binary()
list_policies response: <<count::32, (put_str(name), fields)*>>, in the order given.
@spec encode_ok(non_neg_integer(), binary()) :: binary()
A success response frame (error_code 0) for correlation_id carrying payload.
@spec encode_produce_records([Malachi.Log.Record.t()]) :: binary()
The records half of a produce request (the count, then each record), which does not depend on the
topic. A sender that reuses one batch across topics encodes it once with this and joins it to each
topic with encode_produce_req_with/2; encode_produce_req(topic, records) is the same bytes as one
binary.
A produce request for topic from records already encoded by encode_produce_records/1, as iodata, so
the records are referenced rather than copied; encode_request_iodata/3 frames it the same way.
@spec encode_record(Malachi.Log.Record.t()) :: binary()
Encodes a record for the wire (no offset: the client never sees one).
@spec encode_request(api_key(), non_neg_integer(), binary()) :: binary()
@spec encode_request_iodata(api_key(), non_neg_integer(), iodata()) :: iodata()
encode_request/3 for a payload given as iodata, returned as iodata: the frame header and the payload
side by side, so a sender writing it to a socket never copies the payload into a new binary. The bytes
are exactly those of encode_request/3 over the same payload.
@spec encode_response(non_neg_integer(), error_code(), binary()) :: binary()
@spec encode_topic_policy_resp(topic_policy_resp()) :: binary()
get_topic_policy response: <<put_str(topic), put_str(policy), resolution::8, has_definition::8, fields?, count::16, (put_str(name), value, origin::8)*>>. Resolution is 0 none, 1 resolved, 2
unresolved; origin is 0 global, 1 policy, 2 unresolved_backstop. The definition's fields follow only when
has_definition is 1.