Malachi.Log (malachi v0.8.13)

View Source

An append-only log made of a sequence of segments: the building block beneath NorthGuard's Range.

A log owns a directory of segment files. At most one segment is active (appendable); the rest are sealed (immutable). When the active segment hits a seal threshold (size or age), the log rolls: it seals the active segment and a subsequent append opens a new one. New brokers/segments are created cheaply, which is what gives NorthGuard its "balanced by design" property, modelled here at the single-node level.

Offsets and segment naming

Offsets are globally monotonic across the whole log. Each segment is named by its zero-padded base_offset (Kafka-style), so segments sort by base offset and each sealed segment's end offset is simply the next segment's base offset: no manifest and no scan of sealed segments is needed to route reads.

Reads

read/3 serves committed records from the single segment containing offset. To read across a roll boundary, the caller advances offset and calls again (the normal consumer paging pattern). Sealed segments are opened read-only on demand using their persisted sparse index, then closed.

Like Malachi.Storage.ElixirStore, this is a functional module over an immutable handle (no GenServer). Time-based flushing and concurrency belong in a layer on top.

Summary

Types

t()

Functions

append(log, records)

Buffers records, assigning each a globally monotonic offset. Records are not durable until sync/1. Returns the updated log and the first/last offsets assigned.

close(log)

Closes the active segment's file handle, if any.

delete(log)

Closes the log and removes its directory with every segment file, irreversible. The log owns its directory, so this drops all of its data (used by retention to expire a whole segment). Best-effort on the file removal: a leftover file without control-plane metadata is harmless.

open(directory, opts \\ [])

Opens a fresh, empty log in directory.

pending?(log)

Whether the log has buffered records not yet flushed.

read(log, offset, max_records)

Reads up to max_records committed records from the segment containing offset. Returns :eof past the end of the log, or {:error, :out_of_range} below its start.

rebuild_index(directory, opts \\ [])

Rebuilds the sparse index of every segment in directory from the segments themselves.

recover(directory, opts \\ [])

Reopens an existing log, recovering all segments. Only the last (active) segment is scanned; sealed segments are trusted via their file names.

roll(log)

Forces the active segment to seal (no-op if there is no active segment).

sync(log)

Flushes and fsyncs the active segment, then rolls it if a seal threshold is hit.

verify(directory, opts \\ [])

Verifies every segment file in directory read-only, checking each record's checksum, and aggregates the result. Unlike recover/2 it opens nothing for writing, repairs nothing, and trusts nothing by filename: this is what a background scrub uses to catch corruption at rest in segments that are sealed and therefore never re-scanned by the normal open path.

Types

t()

@type t() :: %Malachi.Log{
  active: term() | nil,
  active_base_offset: non_neg_integer() | nil,
  directory: Path.t(),
  integrity: :ok | map(),
  next_offset: non_neg_integer(),
  sealed_base_offsets: [non_neg_integer()],
  segment_opts: keyword(),
  store: module()
}

Functions

append(log, records)

@spec append(t(), [Malachi.Log.Record.t()]) ::
  {:ok, t(), non_neg_integer(), non_neg_integer()} | {:error, term()}

Buffers records, assigning each a globally monotonic offset. Records are not durable until sync/1. Returns the updated log and the first/last offsets assigned.

close(log)

@spec close(t()) :: :ok

Closes the active segment's file handle, if any.

delete(log)

@spec delete(t()) :: :ok

Closes the log and removes its directory with every segment file, irreversible. The log owns its directory, so this drops all of its data (used by retention to expire a whole segment). Best-effort on the file removal: a leftover file without control-plane metadata is harmless.

open(directory, opts \\ [])

@spec open(
  Path.t(),
  keyword()
) :: {:ok, t()}

Opens a fresh, empty log in directory.

Options

  • :base_offset - first offset of the log (default 0)
  • :store - SegmentStore implementation (default ElixirStore)
  • any remaining options (:max_bytes, :max_age_ms, :index_interval, :flush_bytes) are passed through to each segment.

pending?(log)

@spec pending?(t()) :: boolean()

Whether the log has buffered records not yet flushed.

read(log, offset, max_records)

@spec read(t(), non_neg_integer(), pos_integer()) ::
  {:ok, [Malachi.Log.Record.t()]} | :eof | {:error, term()}

Reads up to max_records committed records from the segment containing offset. Returns :eof past the end of the log, or {:error, :out_of_range} below its start.

rebuild_index(directory, opts \\ [])

@spec rebuild_index(
  Path.t(),
  keyword()
) :: :ok | {:error, term()}

Rebuilds the sparse index of every segment in directory from the segments themselves.

The repair for a damaged index: it is derived data, so no replica is involved, unlike a damaged segment. Only call it after verify/2 reports the records intact, or the rebuilt index will faithfully describe the damage. An absent index file is not an error to begin with (reads just scan from the start of the segment), so a rebuild simply puts the fast path back.

recover(directory, opts \\ [])

@spec recover(
  Path.t(),
  keyword()
) :: {:ok, t()}

Reopens an existing log, recovering all segments. Only the last (active) segment is scanned; sealed segments are trusted via their file names.

roll(log)

@spec roll(t()) :: {:ok, t()}

Forces the active segment to seal (no-op if there is no active segment).

sync(log)

@spec sync(t()) :: {:ok, t()}

Flushes and fsyncs the active segment, then rolls it if a seal threshold is hit.

verify(directory, opts \\ [])

@spec verify(
  Path.t(),
  keyword()
) ::
  {:ok,
   %{
     records: non_neg_integer(),
     bytes: non_neg_integer(),
     files: non_neg_integer()
   }}
  | {:error, map() | :enoent}

Verifies every segment file in directory read-only, checking each record's checksum, and aggregates the result. Unlike recover/2 it opens nothing for writing, repairs nothing, and trusts nothing by filename: this is what a background scrub uses to catch corruption at rest in segments that are sealed and therefore never re-scanned by the normal open path.

Returns {:ok, %{records: n, bytes: b, files: k}} when every file is intact, {:error, details} for the first damaged file (with its path, byte position and reason), or {:error, :enoent} when the directory holds no segment files at all (nothing stored here, or retention removed it).

Options