Bedrock.ControlPlane.Director.Recovery.LogReplayPhase (bedrock v0.5.2)
View SourceMigrates the active transaction window from old logs to newly recruited log configuration.
Solves the critical data migration challenge of transferring the current window of committed transactions to known-good storage. Since recovery must read old logs anyway to verify what's recoverable, it duplicates that data to new logs rather than trusting potentially corrupted or failing storage. This ensures storage servers can continue processing from reliable storage.
Migration Strategy (Consistent Hashing): With consistent hashing, shard→log mapping is computed at runtime. Each new log receives the full list of survivors (old logs that were locked) and can pull transactions from any of them. The log's personalized transaction stream is reconstructed by pulling from survivors and filtering by shard index.
For initial recovery (no old logs), each new log is initialized with an empty transaction stream at version zero.
Service Access: All log services (old and new) are locked by this point in recovery,
so the phase simply retrieves their PIDs from the service_pids map using Map.fetch!
to enforce that all required services must be available.
Data Integrity: Only copies committed transactions within the established version vector (the active window), discarding uncommitted transactions that lack durability guarantees. Maintains transaction ordering and ensures storage servers can continue processing seamlessly in the new configuration.
Stalls with detailed failure information if log copying fails due to service issues.
However, immediately halts with error if any log reports :newer_epoch_exists (this
director has been superseded). Transitions to sequencer startup with complete migration.
Summary
Functions
Copies transaction data from survivor logs to a new log.
Replays transactions from survivor logs into new logs.
Functions
@spec copy_log_data( new_log_id :: Bedrock.DataPlane.Log.id(), survivor_pids :: [pid()], first_version :: Bedrock.version(), last_version :: Bedrock.version(), service_pids :: %{required(Bedrock.DataPlane.Log.id()) => pid()} ) :: {:ok, pid()} | {:error, term()}
Copies transaction data from survivor logs to a new log.
The new log receives a list of survivor PIDs and can pull from any of them. For initial recovery (empty survivor list), the log is initialized at the version vector.
@spec pair_with_old_log_ids([Bedrock.DataPlane.Log.id()], [Bedrock.DataPlane.Log.id()]) :: Enumerable.t({Bedrock.DataPlane.Log.id(), Bedrock.DataPlane.Log.id() | :none})
@spec replay_into_new_logs( survivor_log_ids :: [Bedrock.DataPlane.Log.id()], new_log_ids :: [Bedrock.DataPlane.Log.id()], version_vector :: Bedrock.version_vector(), recovery_attempt :: map(), context :: map() ) :: :ok | {:error, {:failed_to_copy_some_logs, %{required(Bedrock.DataPlane.Log.id()) => term()}}}
Replays transactions from survivor logs into new logs.
With consistent hashing, each new log receives the full list of survivors and can pull from any of them. The Shale recovery logic handles trying multiple sources if the first is unavailable.
For initial recovery (no survivors), new logs are initialized at the version vector.
@spec replay_old_logs_into_new_logs( old_log_ids :: [Bedrock.DataPlane.Log.id()], new_log_ids :: [Bedrock.DataPlane.Log.id()], version_vector :: Bedrock.version_vector(), recovery_attempt :: map(), context :: map() ) :: :ok | {:error, {:failed_to_copy_some_logs, %{required(Bedrock.DataPlane.Log.id()) => term()}}}