Bedrock.ControlPlane.Director.Recovery.LogRecruitmentPhase (bedrock v0.6.0)

View Source

Transforms abstract vacancy placeholders into concrete service assignments.

Solves the practical challenge of assigning real services to fill log vacancies while balancing efficiency with safety—reusing existing services when possible but avoiding services from the old system that contain recovery data.

Three-Phase Assignment Strategy:

  1. Prefer existing log services that weren't part of the old transaction system
  2. Create new log workers using round-robin distribution across nodes when needed
  3. Lock all recruited services (existing and new) to establish exclusive control

Constraints: Old system services are excluded to preserve committed transaction data in case this recovery fails and another one starts. All recruited services must be successfully locked before proceeding to ensure readiness for transaction processing.

Stalls if insufficient nodes exist for worker creation or if recruited services fail to lock. However, immediately halts with error if any service is locked by a newer epoch (this director has been superseded). Transitions to log replay with complete log service assignments.

Summary

Functions

all_vacancies(logs)
fill_log_vacancies(logs, old_system_log_ids, available_log_ids, available_nodes)
lock_service_for_recovery(service, epoch, context \\ %{})
replace_vacancies_with_log_ids(logs, log_id_for_vacancy)

Functions

all_vacancies(logs)

@spec all_vacancies(%{required(Bedrock.DataPlane.Log.id()) => [term()]}) :: MapSet.t()

fill_log_vacancies(logs, old_system_log_ids, available_log_ids, available_nodes)

@spec fill_log_vacancies(
  logs :: %{required(Bedrock.DataPlane.Log.id()) => any()},
  old_system_log_ids :: MapSet.t(Bedrock.DataPlane.Log.id()),
  available_log_ids :: MapSet.t(Bedrock.DataPlane.Log.id()),
  available_nodes :: [node()]
) ::
  {:ok, %{required(Bedrock.DataPlane.Log.id()) => any()},
   [Bedrock.DataPlane.Log.id()]}
  | {:error,
     {:insufficient_nodes, needed_workers :: pos_integer(),
      available_nodes :: non_neg_integer()}}

lock_service_for_recovery(service, epoch, context \\ %{})

@spec lock_service_for_recovery(
  {atom(), {atom(), node()}},
  Bedrock.epoch(),
  map()
) :: {:ok, pid(), map()} | {:error, term()}

replace_vacancies_with_log_ids(logs, log_id_for_vacancy)

@spec replace_vacancies_with_log_ids(
  logs :: %{required(Bedrock.DataPlane.Log.id()) => any()},
  log_id_for_vacancy :: %{required(any()) => Bedrock.DataPlane.Log.id()}
) :: %{required(Bedrock.DataPlane.Log.id()) => any()}