Bedrock.ControlPlane.Director.Recovery.LockingPhase (bedrock v0.5.2)

View Source

Establishes exclusive director control by selectively locking services from the old system layout.

Service locking serves three purposes: prevent split-brain scenarios where multiple directors attempt concurrent control, halt all transaction processing from the old system, and validate service reachability while collecting recovery state information (transaction versions, durability status).

Only services referenced in the old transaction system layout are locked - these contain data that must be preserved during recovery. Individual service failures (unreachable, timeout) are ignored since recovery gathers as many services as possible from a potentially failed system. However, if any service is already locked with a newer epoch, this director has been superseded and should stop all recovery attempts.

The recovery path is determined by whether the old layout contained logs: no logs means first-time initialization, logs present means recovery from existing data.

Summary

Functions

lock_old_system_services(old_system_services, epoch, context \\ %{})
lock_old_system_services_timeout()
lock_service_for_recovery(service, epoch, context \\ %{})

Functions

lock_old_system_services(old_system_services, epoch, context \\ %{})

@spec lock_old_system_services(
  %{
    required(Bedrock.Service.Worker.id()) => %{
      kind: atom(),
      last_seen: {atom(), node()}
    }
  },
  Bedrock.epoch(),
  map()
) ::
  {:ok, locked_ids :: MapSet.t(Bedrock.Service.Worker.id()),
   new_log_recovery_info_by_id :: %{
     required(Bedrock.DataPlane.Log.id()) =>
       Bedrock.DataPlane.Log.recovery_info()
   },
   new_materializer_recovery_info_by_id :: %{
     required(Bedrock.DataPlane.Materializer.id()) =>
       Bedrock.DataPlane.Materializer.recovery_info()
   },
   transaction_services :: %{
     required(Bedrock.Service.Worker.id()) => %{
       status: {:up, pid()},
       kind: :log | :materializer,
       last_seen: {atom(), node()}
     }
   }, service_pids :: %{required(Bedrock.Service.Worker.id()) => pid()}}
  | {:error, :newer_epoch_exists}

lock_old_system_services_timeout()

@spec lock_old_system_services_timeout() :: Bedrock.timeout_in_ms()

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

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