Malachi.
View Source
Pure-Elixir Malachi.Storage.SegmentStore implementation.
File-per-segment, append-only, with batched writes and an fsync-before-ack durability
contract. append/2 buffers; the buffer is flushed and fsynced on an explicit sync/1
or automatically once it reaches :flush_bytes (default 10MB) or :flush_count records
(default 20k): NorthGuard's size and count triggers. Maintains an in-memory sparse index
({offset, file_position} every :index_interval bytes) for seeking, kept in an :array
sorted by offset so a lookup is an O(log n) binary search; the index is persisted to a
sidecar on seal/1 and rebuilt by scanning on recover/3.
This is deliberately a plain module operating on an immutable handle (no GenServer), so
it is deterministic and trivial to property-test. The time-based flush trigger (~10ms)
and concurrency belong in a higher layer (Malachi.BrokerServer) built on top of this.
Reads via :file.pread/3 and writes via :file.pwrite/3 use explicit positions, so
the single file descriptor serves both append and random read without position races.
Recovery scans the segment in bounded chunks, so it never loads the whole file at once.
Summary
Types
One sparse-index entry: a logical offset and the byte position where it starts.
What the last scan of this segment concluded: :ok, or the first damage it hit, with the byte
position and how much of the file could not be read. Only recover/3 scans, so a freshly opened
or read-only handle is :ok by construction.
One buffered, not-yet-flushed record: its offset, encoded frame, and frame size.
Types
@type index_entry() :: {offset :: non_neg_integer(), file_position :: non_neg_integer()}
One sparse-index entry: a logical offset and the byte position where it starts.
@type integrity_verdict() :: %{ reason: atom(), position: non_neg_integer(), unreadable_bytes: non_neg_integer(), sealed?: boolean() }
What the last scan of this segment concluded: :ok, or the first damage it hit, with the byte
position and how much of the file could not be read. Only recover/3 scans, so a freshly opened
or read-only handle is :ok by construction.
@type pending_frame() :: {offset :: non_neg_integer(), frame :: iodata(), frame_size :: pos_integer()}
One buffered, not-yet-flushed record: its offset, encoded frame, and frame size.
@type t() :: %Malachi.Storage.ElixirStore{ file_descriptor: :file.fd(), flush_bytes: pos_integer(), flush_count: pos_integer(), index: :array.array(), index_interval: pos_integer(), integrity: :ok | integrity_verdict(), last_indexed_position: integer(), next_offset: non_neg_integer(), pending: [pending_frame()], pending_bytes: non_neg_integer(), pending_count: non_neg_integer(), segment: Malachi.Log.Segment.t(), write_position: non_neg_integer() }