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_key that already shipped, or
  • changes how error_code or 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

Types

A policy field as the wire carries it.

A topic's policy as get_topic_policy answers it.

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

api_key()

@type api_key() :: 0..23

error_code()

@type error_code() :: non_neg_integer()

policy_value()

@type policy_value() :: non_neg_integer() | String.t() | nil

A policy field as the wire carries it.

topic_policy_resp()

@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

auth_key()

@spec auth_key() :: api_key()

bind_topic_policy_key()

change_password_key()

commit_key()

create_topic_key()

create_user_key()

decode_acl_req(payload)

decode_auth_req(payload)

decode_auth_resp(payload)

decode_bind_topic_policy_req(payload)

@spec decode_bind_topic_policy_req(binary()) :: {String.t() | nil, String.t() | nil}

decode_change_password_req(payload)

decode_commit_req(payload)

decode_create_topic_req(payload)

decode_create_user_req(payload)

decode_define_policy_req(payload)

@spec decode_define_policy_req(binary()) ::
  {String.t() | nil, [{String.t(), policy_value()}]}

decode_delete_policy_req(payload)

@spec decode_delete_policy_req(binary()) :: {String.t() | nil, boolean()}

decode_delete_user_req(payload)

decode_error_reason(payload)

@spec decode_error_reason(binary()) :: String.t() | nil

Decodes an error response payload (see encode_error/2) back to its reason string.

decode_fetch_req(payload)

decode_fetch_resp(payload)

decode_frame(arg1)

@spec decode_frame(binary()) :: {:ok, binary(), binary()} | :incomplete

Peels one frame off a buffer: {:ok, body, rest} or :incomplete if the frame is not all here.

decode_frame(buffer, max_size)

@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.

decode_get_topic_policy_req(payload)

@spec decode_get_topic_policy_req(binary()) :: String.t() | nil

decode_leave_group_req(payload)

decode_list_acls_req(payload)

decode_list_acls_resp(arg)

decode_list_policies_req(arg)

@spec decode_list_policies_req(binary()) :: :ok

list_policies (19) takes an empty request.

decode_list_policies_resp(arg)

@spec decode_list_policies_resp(binary()) :: [
  {String.t(), [{String.t(), policy_value()}]}
]

decode_list_users_resp(arg)

decode_list_users_with_roles_resp(arg)

decode_produce_req(payload)

decode_record(binary)

@spec decode_record(binary()) :: {Malachi.Log.Record.t(), binary()}

decode_request(arg)

@spec decode_request(binary()) :: {api_key(), non_neg_integer(), binary()}

decode_response(arg)

@spec decode_response(binary()) :: {non_neg_integer(), error_code(), binary()}

decode_set_role_req(payload)

decode_stream_ack_req(payload)

decode_subscribe_req(payload)

decode_token_auth_req(payload)

decode_topic_policy_resp(payload)

@spec decode_topic_policy_resp(binary()) :: topic_policy_resp()

define_policy_key()

delete_policy_key()

delete_user_key()

encode_acl_req(username, operation, pattern)

encode_auth_req(username, password)

encode_auth_resp(token)

encode_bind_topic_policy_req(topic, name)

@spec encode_bind_topic_policy_req(String.t(), String.t() | nil) :: binary()

bind_topic_policy (20): <<put_str(topic), put_str(name)>>, a nil name detaches the topic.

encode_change_password_req(username, new_password)

encode_commit_req(topic, group, cursor)

encode_create_topic_req(topic, keyspace_bits)

encode_create_user_req(username, password, permissions)

encode_define_policy_req(name, fields)

@spec encode_define_policy_req(String.t(), [{String.t(), policy_value()}]) :: binary()

define_policy (17): <<put_str(name), fields>>.

encode_delete_policy_req(name, force)

@spec encode_delete_policy_req(String.t(), boolean()) :: binary()

delete_policy (18): <<put_str(name), force::8>>, force 1 deletes a policy topics are bound to.

encode_delete_user_req(username)

encode_error(correlation_id, reason)

@spec encode_error(non_neg_integer(), term()) :: binary()

An error response frame (error_code 1) whose payload is reason as a string.

encode_fetch_req(topic, cursor, group, member, max, wait_ms)

encode_fetch_resp(records, next_cursor)

encode_frame(body)

@spec encode_frame(binary()) :: binary()

Wraps a body in a length-prefixed frame.

encode_get_topic_policy_req(topic)

@spec encode_get_topic_policy_req(String.t()) :: binary()

get_topic_policy (21): put_str(topic).

encode_leave_group_req(topic, group, member)

encode_list_acls_req(username)

encode_list_acls_resp(acls)

encode_list_policies_resp(policies)

@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.

encode_list_users_resp(users)

encode_list_users_with_roles_resp(users)

encode_mtls_auth_req()

encode_ok(correlation_id, payload)

@spec encode_ok(non_neg_integer(), binary()) :: binary()

A success response frame (error_code 0) for correlation_id carrying payload.

encode_produce_records(records)

@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.

encode_produce_req(topic, records)

encode_produce_req_with(topic, encoded_records)

@spec encode_produce_req_with(String.t(), binary()) :: iodata()

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.

encode_record(record)

@spec encode_record(Malachi.Log.Record.t()) :: binary()

Encodes a record for the wire (no offset: the client never sees one).

encode_request(api_key, correlation_id, payload)

@spec encode_request(api_key(), non_neg_integer(), binary()) :: binary()

encode_request_iodata(api_key, correlation_id, payload)

@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.

encode_response(correlation_id, error_code, payload)

@spec encode_response(non_neg_integer(), error_code(), binary()) :: binary()

encode_set_role_req(username, role)

encode_stream_ack_req(topic, group, member, cursor, count)

encode_subscribe_req(topic, group, member, window, max)

encode_token_auth_req(jwt)

encode_topic_policy_resp(resp)

@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.

error_code()

fetch_key()

get_topic_policy_key()

grant_acl_key()

leave_group_key()

list_acls_key()

list_policies_key()

list_users_key()

list_users_with_roles_key()

mtls_auth_key()

ok_code()

produce_key()

revoke_acl_key()

set_role_key()

stream_ack_key()

subscribe_key()

token_auth_key()