RaftEx.Log (raft_ex v0.1.0)

Persistent replicated Raft log.

This module manages the combination of WAL + segment files + mem tables. It provides the core log operations needed by the Raft consensus algorithm:

Append: Add new entries to the log (via WAL)
Read: Fetch entries by index range or specific indexes
Snapshot: Install and recover snapshots
Checkpoint: Create recovery checkpoints for crash recovery
Configuration: Persist and read cluster configuration

Architecture

The log consists of several layers:

WAL (Write-Ahead Log): Durable, sequential write buffer
Mem Table: In-memory buffer for recent entries
Segments: Immutable on-disk files for older entries
Snapshot: Point-in-time state with metadata

Entry Format

Each log entry is a tuple: {index, term, command}

Where command can be:

{:"$usr", metadata, user_command, reply_mode} - User command
{:"$ra_cluster_change", metadata, new_cluster, reply_mode} - Membership change
{:noop, metadata, next_mac_ver} - No-op entry for leadership
{:"$ra_leave", from, leaving, reply_mode} - Leave command
{:"$ra_cluster", metadata, :delete, reply_type} - Delete cluster

Summary

Types

t()

Functions

append(entry, state)

Append a single entry to the log.

can_write?(log)

Check if the log can accept writes.

checkpoint(idx, cluster, mac_ctx, mac_state, state)

Create a checkpoint at the given index.

close(log)

Close the log and clean up resources.

delete_everything(state)

Delete all log data (used for cleanup).

execute_read_plan(plan, flru, transform_fun, options)

Execute a read plan, fetching entries from disk as needed.

exists(arg, state)

Check if an entry exists in the log.

fetch_term(idx, state)

Fetch the term for a specific index.

fold(from, to, fun, acc, state, strategy \\ :error)

Fold over entries in the range [from, to].

handle_event(evt, state)

Handle a log event (e.g., compaction notification).

has_pending?(log)

Check if there are pending writes.

init(conf)

Initialize a new log instance.

install_snapshot(idx_term, mac_mod, live_indexes, state)

Install a snapshot at the given index and term.

last_index_term(log)

Get the last index and term in the log.

last_written(log)

Get the last written index and term.

next_index(log)

Get the next index that would be assigned.

overview(state)

Get an overview of the log state.

partial_read(indexes, state, transform_fun)

Build a read plan for partial reads.

promote_checkpoint(idx, state)

Promote a checkpoint to be the new recovery point.

read_config(dir)

Read the cluster configuration from disk.

read_plan_info(plan)

Get information about a read plan.

recover_snapshot(log)

Recover snapshot state from disk.

release_resources(max_open, access_pattern, state)

Release resources (file handles, etc.).

set_last_index(idx, state)

Set the last index (used during recovery).

set_snapshot_state(snap_state, state)

Set the snapshot state.

snapshot_index_term(log)

Get the current snapshot index and term.

snapshot_size(state)

Get the size of the current snapshot.

snapshot_state(log)

Get the current snapshot state.

sparse_read(indexes, state)

Read specific entries by index.

tick(now, state)

Process a tick for log maintenance.

truncate(from_index, state)

Truncate the log, removing all entries starting from (and including) from_index. Used when a leader sends conflicting entries at a given index.

update_release_cursor(idx, cluster, arg, mac_state, state)

Update the release cursor for snapshot management.

write(entries, state)

Write multiple entries to the log.

write_config(config, log)

Write the cluster configuration to disk.

Types

t()

@type t() :: %RaftEx.Log{
  cfg: map(),
  current_snapshot: {RaftEx.Types.index(), RaftEx.Types.term_num()} | nil,
  data_dir: Path.t(),
  entries: %{
    required(RaftEx.Types.index()) =>
      {RaftEx.Types.index(), RaftEx.Types.term_num(), term()}
  },
  last_resend_time: non_neg_integer(),
  last_term: RaftEx.Types.term_num(),
  last_wal_write: {pid(), reference(), RaftEx.Types.index()},
  last_written_index_term: {RaftEx.Types.index(), RaftEx.Types.term_num()},
  live_indexes: [RaftEx.Types.index()],
  mem_table: :ets.table() | nil,
  meta_name: atom(),
  next_index: RaftEx.Types.index(),
  pending: [term()],
  range: {RaftEx.Types.index(), RaftEx.Types.index()} | nil,
  reader: term() | nil,
  snapshot_state: term() | nil,
  tx: boolean(),
  wal_name: atom()
}

Functions

append(entry, state)

@spec append({RaftEx.Types.index(), RaftEx.Types.term_num(), term()}, t()) :: t()

Append a single entry to the log.

Returns the updated state. The entry is added to the mem table and scheduled for WAL write.

can_write?(log)

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

Check if the log can accept writes.

checkpoint(idx, cluster, mac_ctx, mac_state, state)

@spec checkpoint(
  RaftEx.Types.index(),
  atom(),
  map(),
  term(),
  t()
) :: {t(), [term()]}

Create a checkpoint at the given index.

close(log)

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

Close the log and clean up resources.

delete_everything(state)

@spec delete_everything(t()) :: :ok

Delete all log data (used for cleanup).

execute_read_plan(plan, flru, transform_fun, options)

@spec execute_read_plan(map(), term(), (term() -> term()), keyword()) ::
  {map(), term()}

Execute a read plan, fetching entries from disk as needed.

exists(arg, state)

@spec exists(
  {RaftEx.Types.index(), RaftEx.Types.term_num()},
  t()
) :: {boolean(), t()}

Check if an entry exists in the log.

fetch_term(idx, state)

@spec fetch_term(RaftEx.Types.index(), t()) :: {RaftEx.Types.term_num() | nil, t()}

Fetch the term for a specific index.

fold(from, to, fun, acc, state, strategy \\ :error)

@spec fold(
  RaftEx.Types.index(),
  RaftEx.Types.index(),
  (term(), term() -> term()),
  term(),
  t(),
  atom()
) :: {term(), t()}

@spec fold(
  RaftEx.Types.index(),
  RaftEx.Types.index(),
  (term(), term() -> term()),
  term(),
  t(),
  atom()
) :: {term(), t()}

Fold over entries in the range [from, to].

Applies fun to each entry, accumulating acc. Returns {final_acc, state}.

handle_event(evt, state)

@spec handle_event(term(), t()) :: {t(), [term()]}

Handle a log event (e.g., compaction notification).

has_pending?(log)

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

Check if there are pending writes.

init(conf)

@spec init(map()) :: t()

Initialize a new log instance.

Sets up the data directory, WAL, ETS table for mem table, and recovers from existing state if present.

install_snapshot(idx_term, mac_mod, live_indexes, state)

@spec install_snapshot(
  {RaftEx.Types.index(), RaftEx.Types.term_num()},
  module(),
  [RaftEx.Types.index()],
  t()
) :: {:ok, t(), [term()]}

Install a snapshot at the given index and term.

This clears all log entries up to and including the snapshot index.

last_index_term(log)

@spec last_index_term(t()) :: {RaftEx.Types.index(), RaftEx.Types.term_num()}

Get the last index and term in the log.

last_written(log)

@spec last_written(t()) :: {RaftEx.Types.index(), RaftEx.Types.term_num()}

Get the last written index and term.

next_index(log)

@spec next_index(t()) :: RaftEx.Types.index()

Get the next index that would be assigned.

overview(state)

@spec overview(t()) :: map()

Get an overview of the log state.

partial_read(indexes, state, transform_fun)

@spec partial_read([RaftEx.Types.index()], t(), (RaftEx.Types.index(),
                                           RaftEx.Types.term_num(),
                                           term() ->
                                             term())) :: %{
  dir: Path.t(),
  read: map(),
  plan: list()
}

Build a read plan for partial reads.

Returns a map with :dir, :read, and :plan keys indicating which entries are in memory vs need to be read from disk.

promote_checkpoint(idx, state)

@spec promote_checkpoint(RaftEx.Types.index(), t()) :: {t(), [term()]}

Promote a checkpoint to be the new recovery point.

read_config(dir)

@spec read_config(t() | Path.t()) :: {:ok, map()} | {:error, term()}

Read the cluster configuration from disk.

read_plan_info(plan)

@spec read_plan_info(map()) :: %{
  num_read: non_neg_integer(),
  num_in_segments: non_neg_integer(),
  num_segments: non_neg_integer()
}

Get information about a read plan.

recover_snapshot(log)

@spec recover_snapshot(t()) :: term() | nil

Recover snapshot state from disk.

release_resources(max_open, access_pattern, state)

@spec release_resources(non_neg_integer(), atom(), t()) :: t()

Release resources (file handles, etc.).

set_last_index(idx, state)

@spec set_last_index(RaftEx.Types.index(), t()) :: {:ok, t()}

Set the last index (used during recovery).

set_snapshot_state(snap_state, state)

@spec set_snapshot_state(term(), t()) :: t()

Set the snapshot state.

snapshot_index_term(log)

@spec snapshot_index_term(t()) ::
  {RaftEx.Types.index(), RaftEx.Types.term_num()} | nil

Get the current snapshot index and term.

snapshot_size(state)

@spec snapshot_size(t()) :: non_neg_integer() | nil

Get the size of the current snapshot.

snapshot_state(log)

@spec snapshot_state(t()) :: term() | nil

Get the current snapshot state.

sparse_read(indexes, state)

@spec sparse_read([RaftEx.Types.index()], t()) :: {[term()], t()}

Read specific entries by index.

Returns {entries, state} where entries is a list of {index, term, command} tuples.

tick(now, state)

@spec tick(non_neg_integer(), t()) :: t()

Process a tick for log maintenance.

truncate(from_index, state)

@spec truncate(RaftEx.Types.index(), t()) :: t()

Truncate the log, removing all entries starting from (and including) from_index. Used when a leader sends conflicting entries at a given index.

update_release_cursor(idx, cluster, arg, mac_state, state)

@spec update_release_cursor(
  RaftEx.Types.index(),
  atom(),
  {module(), term()},
  term(),
  t()
) :: {t(), [term()]}

Update the release cursor for snapshot management.

write(entries, state)

@spec write([{RaftEx.Types.index(), RaftEx.Types.term_num(), term()}], t()) ::
  {:ok, t()} | {:error, term()}

Write multiple entries to the log.

Appends all entries and then flushes to WAL. Returns {:ok, updated_state}.

write_config(config, log)

@spec write_config(map(), t()) :: :ok

Write the cluster configuration to disk.