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.
Preallocation
With :prealloc_bytes set, a new segment is sized to that many bytes at creation
(Malachi.Storage.Preallocation) so that appends overwrite an already-allocated region instead
of extending the file. Measured on an ubuntu-latest runner in the regime the pinned ceiling
harness runs (batch 10 x 256B, one sync per produce), that takes the per-flush p50 from 317us to
96us, a 70% cut, against a measured noise floor of 2us. The reason is that a growing file changes
its size on every append, and a size change is metadata the following sync has to journal; a
1-byte sync costs 257us on a growing file and 75us on a sized one.
It is a trade rather than a free win: it also gives up the filesystem's delayed allocation, which
costs more than the journal saves once a flush is large. The crossover is around 170KB per flush
and Malachi.Storage.Preallocation carries the measured curve.
It costs one full-size write at creation (36ms for 64MB on that runner) and it changes what the
unwritten tail looks like, which recovery has to understand: see classify_tail/2. The tail is
truncated away on seal/1 and on close/1, so a segment this store is not actively writing is
byte-exact, and every size the rest of the system reads off disk keeps meaning what it meant.
Failures
Every file operation's error comes back as {:error, posix}, per the failure contract in
Malachi.Storage.SegmentStore. That includes a read that fails while recovery scans the segment:
it is answered as the storage failure it is, never classified as damage to the bytes and recovered
around. The one deliberate exception is preallocation: it is an optimization, so an extension that
fails degrades to an unpreallocated segment rather than failing the open or recovery, and the next
commit reports the condition if it persists. It degrades only if the file can be put back to its
previous size; when it cannot, the handle is refused (see Malachi.Storage.Preallocation.extend_or_restore/4).
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.
What the bytes past the last valid frame are.
Functions
What recovery may do about a classified tail: nothing, discard it, or leave it alone and report.
Classifies the bytes past the last valid frame from the scan's halt and the shape of the tail.
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{ append_refusal: nil | :damaged_tail, 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(), prealloc_bytes: non_neg_integer(), preallocated_to: non_neg_integer() | nil, segment: Malachi.Log.Segment.t(), write_position: non_neg_integer() }
@type tail_classification() :: :clean | :blank | :torn | :rot
What the bytes past the last valid frame are.
:clean- there are none: valid frames consumed the file exactly.:blank- unwritten preallocated space. Not damage, and not a shape a growing segment can produce, since it has no room past its last write.:torn- a write that did not finish. Either the file ended inside a frame (the only shape a growing segment can make) or, in a preallocated segment, a frame whose bytes stop partway and give out into the unwritten zeros behind it. Nothing valid follows it by construction.:rot- a frame that is there and is wrong. Valid frames may well follow it, so these bytes are a peer's problem to fix and never this node's to discard.
Functions
@spec action_for(tail_classification(), boolean()) :: :none | :discard_tail | :preserve
What recovery may do about a classified tail: nothing, discard it, or leave it alone and report.
The state matters here and not in the classification, because it is about permission rather than about what the bytes are. The rule keys on the DAMAGE and not only on the seal marker, because a replica's file carries a marker only when its log rolled locally (by size or age): sealing a segment is a control-plane decision, so a segment the cluster considers immutable usually has no marker on disk, and keying the guard on the marker alone would leave the destructive path wide open in exactly the deployment that matters.
:preserve on an active segment also withdraws permission to append: the recovered handle reads
up to the damage and answers {:error, :damaged_tail} to append/2, since its write position is
where the preserved bytes begin.
@spec classify_tail(:eof | :blank | {:error, atom()}, map()) :: tail_classification()
Classifies the bytes past the last valid frame from the scan's halt and the shape of the tail.
Pure, and separate from what recovery then DOES about it (action_for/2), because the two answer
different questions and only one of them depends on whether the segment is sealed.
Telling :torn from :rot is the whole difficulty, and preallocation is what makes it hard.
While a segment grows, "the file ended inside this frame" is proof the writer died there, because
there is nowhere else for bytes to be. A preallocated segment has no such end: a torn write leaves
a frame header intact and its payload completed by the zeros that were already there, so it comes
back as :bad_crc, exactly like rot.
What still separates them is where the zeros start. A torn write stops partway through the frame and the unwritten region takes over from there, so the zeros begin INSIDE the frame the header describes. Rot flips bytes in a frame that was written whole, so the zeros (if any) begin at or after its end. That is the test, and it is bounded: it looks at one window past the damage rather than scanning the rest of the segment.
It has one known way to be wrong, and it is deliberate: a record whose value legitimately ends in NUL bytes, sitting in the last frame of a segment, rotted, would be read as torn and dropped rather than reported. It is pinned by a test that names it. The trade is that the alternative, calling every interrupted flush corruption, would have every unclean restart report damage.