RaftEx.LogWal
(raft_ex v0.1.0)
View Source
Write-ahead log process for RaftEx.
Accepts batched append requests from server processes and writes them durably before acknowledging. The WAL provides:
- Batched writes: Multiple entries are grouped and written together
- Configurable fsync: Supports
:datasync,:fsync, or:none - Recovery: Replays WAL entries on startup
- Truncation: Removes entries up to a given index after snapshotting
- Size limits: Rotates when WAL exceeds configured size
Architecture
The WAL maintains an in-memory buffer of pending entries. When a write request arrives, entries are added to the buffer. A flush is triggered when:
- The buffer reaches
wal_max_batch_size - A synchronous write is requested
- A periodic timer fires
Entries are written to a single WAL file with a simple binary format:
<<magic::32, version::8, entry_count::32, entries::binary, crc::32>>Integration
Started by RaftEx.LogWalSupervisor as part of the system supervision tree.
The Log module sends {:append, entries, reply_to} messages and receives
{:written, from_index, to_index} notifications.
Summary
Functions
Append entries to the WAL.
Asynchronously append entries to the WAL.
Returns a specification to start this module under a supervisor.
Get the current WAL state.
Get WAL overview for metrics and monitoring.
Read entries from the WAL buffer (in-memory only).
Recover entries from the WAL file on startup.
Start the WAL GenServer.
Truncate the WAL, removing all entries up to and including index.
Types
@type state() :: %RaftEx.LogWal{ buffer: [{RaftEx.Types.index(), RaftEx.Types.term_num(), term()}], config: map(), file_size: non_neg_integer(), first_index: RaftEx.Types.index(), flush_timer: reference() | nil, last_index: RaftEx.Types.index(), pending_writes: [{pid(), reference()}], wal_fd: :file.io_device() | nil, wal_file: Path.t() }
Functions
@spec append( atom(), [{RaftEx.Types.index(), RaftEx.Types.term_num(), term()}], timeout() ) :: {:ok, RaftEx.Types.index(), RaftEx.Types.index()} | {:error, term()}
Append entries to the WAL.
Returns {:ok, first_index, last_index} when entries are durably written.
@spec append_async(atom(), [{RaftEx.Types.index(), RaftEx.Types.term_num(), term()}]) :: :ok
Asynchronously append entries to the WAL.
The caller will receive {:wal_written, first_index, last_index} when complete.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec info(atom()) :: {RaftEx.Types.index(), RaftEx.Types.index(), non_neg_integer()}
Get the current WAL state.
Returns {first_index, last_index, buffer_size}.
Get WAL overview for metrics and monitoring.
@spec read(atom(), RaftEx.Types.index(), RaftEx.Types.index()) :: [ {RaftEx.Types.index(), RaftEx.Types.term_num(), term()} ]
Read entries from the WAL buffer (in-memory only).
Returns entries in the range [from_index, to_index].
@spec recover(map()) :: {RaftEx.Types.index(), RaftEx.Types.index(), list()}
Recover entries from the WAL file on startup.
Returns {first_index, last_index, entries}.
@spec start_link(map()) :: GenServer.on_start()
Start the WAL GenServer.
Options
:names- Map of process names including:wal:data_dir- Directory to store WAL files:wal_max_size_bytes- Maximum WAL file size before rotation:wal_max_batch_size- Maximum entries per batch write:wal_sync_method-:datasync,:fsync, or:none:wal_compute_checksums- Whether to compute CRC32 checksums
@spec truncate(atom(), RaftEx.Types.index(), timeout()) :: :ok | {:error, term()}
Truncate the WAL, removing all entries up to and including index.
Used after installing a snapshot to discard old entries.