Malachi.Log.Record (malachi v0.22.7)

The most granular unit of data in Malachi's log storage, mirroring NorthGuard's record: a key, a value, and user-defined headers, all opaque bytes - plus a logical offset (assigned on append) and a timestamp.

On-disk frame format

Each record is persisted as a self-describing frame so the log can be scanned and recovered after a crash, and so corruption can be detected:

<<magic::16, payload_length::32, crc32::32, payload::binary-size(payload_length)>>

where payload is:

<<offset::64, timestamp::64, flags::8,
  key_length::32, key::binary, value_length::32, value::binary,
  header_count::32, (key_length::32, key, value_length::32, value)... >>

flags bit 0 distinguishes a nil key (absent) from an empty-binary key. The leading magic/payload_length/crc32 header lets recovery (a) detect a partial trailing write (truncated frame) and stop cleanly, and (b) detect bit-rot via CRC.

Where a scan stops

A scan hits one of four things, and telling them apart is what recovery is built on:

a valid frame, and it continues;
:incomplete, meaning the bytes run out inside a frame;
:blank, meaning the frame header is all zeros. Zeros are not a frame that went wrong, they are space that was never written, which is what the unwritten region of a preallocated segment reads back as (Malachi.Storage.Preallocation). A segment that grows never produces this, because it has no space past its last write;
{:error, reason}, meaning a frame is there and is wrong.

The magic is 0x4D51, so a zero magic can never be a real frame, and a payload shorter than the fixed fields every record carries can never be one either.

Summary

Types

t()

Functions

check_one(binary)

Verifies the frame at the front of binary without deserializing its payload: the framing and the CRC are checked, then the frame is skipped. Same return shapes as decode_one/1 minus the record itself, so {:error, :bad_payload} cannot occur here (the payload is never parsed).

decode_all(binary)

Decodes every complete, valid frame from the front of binary.

decode_one(binary)

Decodes a single frame from the front of binary.

encode(record)

Encodes a record (with its offset already assigned) into a binary frame.

encoded_size(record)

The exact on-disk frame size, in bytes, this record will occupy, matching encode/1 byte-for-byte. The offset need not be assigned, since it is always a fixed 8 bytes. Used by the broker to drive size-based segment rollover with the same accounting the log writes.

new(value, opts \\ [])

Builds a record. offset is left nil and assigned by the store on append.

Types

t()

@type t() :: %Malachi.Log.Record{
  headers: [{binary(), binary()}],
  key: binary() | nil,
  offset: non_neg_integer() | nil,
  timestamp: non_neg_integer(),
  value: binary()
}

Functions

check_one(binary)

@spec check_one(binary()) ::
  {:ok, pos_integer(), binary()} | :incomplete | :blank | {:error, atom()}

This is the integrity-scan path (Malachi.Log.verify/2): a scrub walks whole segments only to confirm that every frame still matches its checksum, and building a Record struct per frame would dominate that cost for no benefit.

decode_all(binary)

@spec decode_all(binary()) :: {[{t(), non_neg_integer()}], non_neg_integer()}

Decodes every complete, valid frame from the front of binary.

Returns {records_with_positions, valid_bytes} where records_with_positions is a list of {record, byte_position_in_binary} and valid_bytes is the number of bytes consumed by valid frames. Decoding stops at the first incomplete or corrupt frame, so valid_bytes is exactly the safe truncation point for crash recovery.

decode_one(binary)

@spec decode_one(binary()) ::
  {:ok, t(), pos_integer(), binary()} | :incomplete | :blank | {:error, atom()}

Decodes a single frame from the front of binary.

Returns {:ok, record, frame_size, rest} on success, :incomplete if binary does not yet contain a full frame (partial/trailing write), :blank if the frame header is all zeros (unwritten space, never a damaged frame), or {:error, reason} if the framing is corrupt.

encode(record)

@spec encode(t()) :: binary()

Encodes a record (with its offset already assigned) into a binary frame.

encoded_size(record)

@spec encoded_size(t()) :: pos_integer()

new(value, opts \\ [])

@spec new(
  binary(),
  keyword()
) :: t()

Builds a record. offset is left nil and assigned by the store on append.

Options

:key - binary key, or nil (default nil)
:headers - list of {binary, binary} tuples (default [])
:timestamp - epoch milliseconds (default: now)