Bedrock.ControlPlane.Coordinator (bedrock v0.6.0)

View Source

Manages cluster state through Raft consensus and coordinates Director lifecycle.

The Coordinator maintains the authoritative cluster configuration and service directory through distributed consensus. When elected as leader, it manages Director startup with capability-based readiness checking to ensure robust recovery in dynamic environments.

Service Registration Flow

Nodes advertise their services and capabilities to the elected leader Coordinator by calling register_services/2 or register_node_resources/4. The leader then persists this service information through Raft consensus, propagating updates to all Coordinators. The service directory is maintained consistently across the cluster, ensuring the Director receives current service topology during recovery.

Leader Readiness States

Leaders track their readiness state for Director recovery:

  • :not_leader - This node is not the cluster leader
  • :leader_ready - This node is leader and ready to attempt Director recovery
  • :recovery_failed - Director recovery failed; waiting for meaningful capability changes

Upon election, leaders immediately transition to :leader_ready and attempt Director recovery. Config is loaded from object storage at init, and TSL is the output of recovery (not an input), so there's no need to wait for state restoration.

Capability-Based Recovery Retry

Leaders track service capability changes through hashing of recovery-relevant capabilities (coordination, log, storage). Recovery is only retried when meaningful capability changes occur, avoiding unnecessary retry attempts on transient service announcements or time-based intervals.

See Also

Summary

Types

compact_service_info()
ref()
service_info()

Functions

config_key()
deregister_services(coordinator, service_ids, timeout \\ 5000)
fetch_config(coordinator, timeout \\ 5000)
fetch_service_directory(coordinator, timeout \\ 5000)

The current service directory: every worker advertised to the coordinator, by id. The director refreshes its view from this at each recovery attempt — workers register as they come up, and a directory snapshot taken at director start goes stale immediately on a booting node.

fetch_transaction_system_layout(coordinator, timeout \\ 5000)
notify_config(coordinator, config)

Notify the coordinator of a config update.

notify_transaction_system_layout(coordinator, transaction_system_layout)

Notify the coordinator of a new transaction system layout.

register_node_resources(coordinator, client_pid, compact_services, capabilities, timeout \\ 5000)
register_services(coordinator, services, timeout \\ 5000)
start_link(opts)

Types

compact_service_info()

@type compact_service_info() ::
  {service_id :: String.t(), kind :: atom(), name :: atom()}

ref()

@type ref() :: atom() | {atom(), node()}

service_info()

@type service_info() ::
  {service_id :: String.t(), kind :: atom(), worker_ref :: {atom(), node()}}

Functions

config_key()

@spec config_key() :: atom()

deregister_services(coordinator, service_ids, timeout \\ 5000)

@spec deregister_services(
  coordinator_ref :: ref(),
  service_ids :: [String.t()],
  timeout_ms :: timeout_in_ms()
) ::
  {:ok, txn_id :: term()}
  | {:error, :unavailable}
  | {:error, :failed}
  | {:error, :not_leader}

fetch_config(coordinator, timeout \\ 5000)

@spec fetch_config(coordinator_ref :: ref(), timeout_ms :: timeout_in_ms()) ::
  {:ok, config :: Bedrock.ControlPlane.Config.t()}
  | {:error, :unavailable | :timeout}

fetch_service_directory(coordinator, timeout \\ 5000)

@spec fetch_service_directory(coordinator_ref :: ref(), timeout_ms :: timeout_in_ms()) ::
  {:ok, %{required(String.t()) => {atom(), {atom(), node()}}}}
  | {:error, :unavailable | :timeout}

The current service directory: every worker advertised to the coordinator, by id. The director refreshes its view from this at each recovery attempt — workers register as they come up, and a directory snapshot taken at director start goes stale immediately on a booting node.

fetch_transaction_system_layout(coordinator, timeout \\ 5000)

@spec fetch_transaction_system_layout(
  coordinator_ref :: ref(),
  timeout_ms :: timeout_in_ms()
) ::
  {:ok,
   transaction_system_layout ::
     Bedrock.ControlPlane.Config.TransactionSystemLayout.t()}
  | {:error, :unavailable | :timeout}

notify_config(coordinator, config)

@spec notify_config(
  coordinator_ref :: ref(),
  config :: Bedrock.ControlPlane.Config.t()
) :: :ok

Notify the coordinator of a config update.

This is called by the Director during recovery to update the coordinator's cached config (e.g., recovery_attempt state). Config is persisted to object storage by the Director's persistence phase, not via Raft consensus.

notify_transaction_system_layout(coordinator, transaction_system_layout)

@spec notify_transaction_system_layout(
  coordinator_ref :: ref(),
  transaction_system_layout ::
    Bedrock.ControlPlane.Config.TransactionSystemLayout.t()
) :: :ok

Notify the coordinator of a new transaction system layout.

This is called by the Director after recovery completes to update the coordinator's cached TSL and broadcast it to all Link subscribers. Unlike the old Raft-based update, this is a direct notification without consensus - the TSL is persisted to object storage by the Director's persistence phase.

register_node_resources(coordinator, client_pid, compact_services, capabilities, timeout \\ 5000)

@spec register_node_resources(
  coordinator_ref :: ref(),
  client_pid :: pid(),
  compact_services :: [compact_service_info()],
  capabilities :: [Bedrock.Cluster.capability()],
  timeout_ms :: timeout_in_ms()
) ::
  {:ok, txn_id :: term()}
  | {:error, :unavailable}
  | {:error, :failed}
  | {:error, :not_leader}

register_services(coordinator, services, timeout \\ 5000)

@spec register_services(
  coordinator_ref :: ref(),
  services :: [service_info()],
  timeout_ms :: timeout_in_ms()
) ::
  {:ok, txn_id :: term()}
  | {:error, :unavailable}
  | {:error, :failed}
  | {:error, :not_leader}

start_link(opts)