Bedrock.ControlPlane.Coordinator.DiskRaftLog (bedrock v0.5.2)

View Source

A DETS-based implementation of the Raft log using transaction chaining.

This module provides persistent storage for Raft consensus operations, ensuring that log entries survive process and node restarts.

Design

  • Uses DETS for key-value storage with transaction chaining
  • Chain links use forward pointers for O(1) truncation
  • No in-memory state - DETS provides all storage
  • Atomic batch operations for consistency

DETS Schema

  • Transaction records: {transaction_id, data}

  • Chain links: {{:chain, transaction_id}, next_transaction_id | nil}

  • Well-known keys: {:tail, transaction_id}, {:last_commit, transaction_id}, {:current_term, election_term}

File Layout

Coordinator follows standard Bedrock working directory pattern:

/data/coordinator/      # Base path from config[:coordinator][:path]
└── raft/               # Coordinator working directory
    └── raft_log.dets   # DETS file

Summary

Types

chain_link_record()
dets_error()
dets_operation_result()
dets_record()
input_transaction()
metadata_record()
open_result()
stored_transaction_record()
t()

Functions

append_transactions(t, prev_id, transactions)

Append the given block of transactions to the log.

build_chain_links(prev_id, transactions)

Helper function to build chain link records.

close(disk_raft_log)

Close the DETS table.

commit_up_to(t, transaction_id)

Mark all transactions up to and including the given transaction as committed.

current_term(t)

Get the current election term from persistent storage. Returns 0 if no term has been persisted yet (initial state).

has_transaction_id?(t, transaction_id)

Does the log contain the given transaction?

initial_transaction_id(t)

Get the initial transaction for the log.

new(opts \\ [])

Create a new DETS-based raft log.

new_id(t, term, sequence)

Create a new log with the given term and sequence number.

newest_safe_transaction_id(t)

Get the newest safe transaction in the log.

newest_transaction_id(t)

Get the newest transaction in the log.

open(log)

Open the DETS table for reading and writing.

purge_transactions_after(t, transaction_id)

Purge the log of all transactions after the given id.

save_current_term(t, term)

Save the current election term to persistent storage. This must be called before responding to RPCs to ensure Raft safety.

sync(disk_raft_log)

Sync the DETS table to disk to ensure durability.

transactions_from(t, from, to)

Get a list of transactions from the given starting point.

transactions_to(t, to)

Get a list of transactions that have occurred up to the given transaction.

walk_chain_inclusive(log, current_id, to_id)

Helper function to walk chain inclusively from current to target.

Types

dets_error()

@type dets_error() ::
  {:error,
   :file_not_found | :permission_denied | :badarg | :table_not_open | term()}

dets_operation_result()

@type dets_operation_result() :: {:ok, t()} | dets_error()

dets_record()

@type dets_record() ::
  stored_transaction_record() | chain_link_record() | metadata_record()

input_transaction()

@type input_transaction() :: {term :: Bedrock.Raft.election_term(), data :: term()}

metadata_record()

@type metadata_record() ::
  {:tail, Bedrock.Raft.transaction_id()}
  | {:last_commit, Bedrock.Raft.transaction_id()}
  | {:current_term, Bedrock.Raft.election_term()}

open_result()

@type open_result() :: {:ok, t()} | dets_error()

stored_transaction_record()

@type stored_transaction_record() ::
  {Bedrock.Raft.transaction_id(), input_transaction()}

t()

@type t() :: %Bedrock.ControlPlane.Coordinator.DiskRaftLog{
  is_open: boolean(),
  table_file: String.t(),
  table_name: atom()
}

Functions

append_transactions(t, prev_id, transactions)

@spec append_transactions(t(), Bedrock.Raft.transaction_id(), [input_transaction()]) ::
  dets_operation_result() | {:error, :prev_transaction_not_found}

Append the given block of transactions to the log.

build_chain_links(prev_id, transactions)

@spec build_chain_links(Bedrock.Raft.transaction_id(), [stored_transaction_record()]) ::
  [
    chain_link_record()
  ]

Helper function to build chain link records.

close(disk_raft_log)

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

Close the DETS table.

commit_up_to(t, transaction_id)

@spec commit_up_to(t(), Bedrock.Raft.transaction_id()) ::
  dets_operation_result() | :unchanged

Mark all transactions up to and including the given transaction as committed.

current_term(t)

@spec current_term(t()) :: Bedrock.Raft.election_term()

Get the current election term from persistent storage. Returns 0 if no term has been persisted yet (initial state).

has_transaction_id?(t, transaction_id)

@spec has_transaction_id?(t(), Bedrock.Raft.transaction_id()) :: boolean()

Does the log contain the given transaction?

initial_transaction_id(t)

@spec initial_transaction_id(t()) :: Bedrock.Raft.tuple_transaction_id()

Get the initial transaction for the log.

new(opts \\ [])

@spec new(keyword()) :: t()

Create a new DETS-based raft log.

Options

  • :log_dir - Directory to store DETS file (required)
  • :table_name - Name for the DETS table (default: :raft_log)

Examples

iex> log = DiskRaftLog.new(log_dir: "/tmp/raft")
iex> is_struct(log, DiskRaftLog)
true

new_id(t, term, sequence)

@spec new_id(t(), Bedrock.Raft.election_term(), Bedrock.Raft.index()) ::
  Bedrock.Raft.tuple_transaction_id()

Create a new log with the given term and sequence number.

newest_safe_transaction_id(t)

@spec newest_safe_transaction_id(t()) :: Bedrock.Raft.transaction_id()

Get the newest safe transaction in the log.

newest_transaction_id(t)

@spec newest_transaction_id(t()) :: Bedrock.Raft.transaction_id()

Get the newest transaction in the log.

open(log)

@spec open(t()) :: open_result()

Open the DETS table for reading and writing.

This must be called before any other operations.

purge_transactions_after(t, transaction_id)

@spec purge_transactions_after(t(), Bedrock.Raft.transaction_id()) ::
  dets_operation_result()

Purge the log of all transactions after the given id.

save_current_term(t, term)

@spec save_current_term(t(), Bedrock.Raft.election_term()) :: dets_operation_result()

Save the current election term to persistent storage. This must be called before responding to RPCs to ensure Raft safety.

sync(disk_raft_log)

@spec sync(t()) :: :ok | {:error, :table_not_open | term()}

Sync the DETS table to disk to ensure durability.

transactions_from(t, from, to)

@spec transactions_from(
  t(),
  Bedrock.Raft.transaction_id(),
  Bedrock.Raft.transaction_id() | :newest | :newest_safe
) :: [stored_transaction_record()]

Get a list of transactions from the given starting point.

transactions_to(t, to)

@spec transactions_to(t(), Bedrock.Raft.transaction_id() | :newest | :newest_safe) ::
  [
    stored_transaction_record()
  ]

Get a list of transactions that have occurred up to the given transaction.

walk_chain_inclusive(log, current_id, to_id)

@spec walk_chain_inclusive(
  t(),
  Bedrock.Raft.transaction_id(),
  Bedrock.Raft.transaction_id()
) :: [
  stored_transaction_record()
]
@spec walk_chain_inclusive(
  t(),
  Bedrock.Raft.transaction_id(),
  Bedrock.Raft.transaction_id()
) :: []

Helper function to walk chain inclusively from current to target.