Bedrock.DataPlane.CommitProxy.Finalization (bedrock v0.5.3)
View SourceTransaction finalization pipeline that handles conflict resolution and log persistence.
Version Chain Integrity
CRITICAL: This module maintains the Lamport clock version chain established by the sequencer.
The sequencer provides both last_commit_version and commit_version as a proper chain link:
last_commit_version: The actual last committed version from the sequencercommit_version: The new version assigned to this batch
Always use the exact version values provided by the sequencer through the batch to maintain proper MVCC conflict detection and transaction ordering. Version gaps can exist due to failed transactions, recovery scenarios, or system restarts.
Summary
Functions
Executes the complete transaction finalization pipeline for a batch of transactions.
Pushes transactions directly to logs and waits for acknowledgement from ALL log servers.
Types
@type abort_reply_fn() :: ([Bedrock.DataPlane.CommitProxy.Batch.reply_fn()] -> :ok)
@type async_stream_fn() :: (enumerable :: Enumerable.t(), fun :: (term() -> term()), opts :: keyword() -> Enumerable.t())
@type finalization_error() :: resolution_error() | storage_coverage_error() | log_push_error()
@type log_push_batch_fn() :: (last_commit_version :: Bedrock.version(), transactions_by_log :: %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.DataPlane.Transaction.encoded() }, commit_version :: Bedrock.version(), opts :: [ log_services: %{ required(Bedrock.DataPlane.Log.id()) => pid() | {atom(), node()} }, timeout: Bedrock.timeout_in_ms(), async_stream_fn: async_stream_fn() ] -> :ok | {:error, log_push_error()})
@type log_push_error() :: {:log_failures, [{Bedrock.DataPlane.Log.id(), term()}]} | {:insufficient_acknowledgments, non_neg_integer(), non_neg_integer(), [{Bedrock.DataPlane.Log.id(), term()}]} | :log_push_failed
@type log_push_single_fn() :: (Bedrock.ControlPlane.Config.ServiceDescriptor.t(), binary(), Bedrock.version() -> :ok | {:error, :unavailable})
@type metadata_mutations() :: [Bedrock.Internal.TransactionBuilder.Tx.mutation()]
@type resolution_error() :: :timeout | :unavailable | {:resolver_unavailable, term()}
@type resolver_fn() :: (Bedrock.DataPlane.Resolver.ref(), Bedrock.epoch(), Bedrock.version(), Bedrock.version(), [Bedrock.DataPlane.Transaction.encoded()], [metadata_mutations()], keyword() -> {:ok, [non_neg_integer()], [Bedrock.DataPlane.Resolver.MetadataAccumulator.entry()]} | {:error, term()} | {:failure, :timeout, Bedrock.DataPlane.Resolver.ref()} | {:failure, :unavailable, Bedrock.DataPlane.Resolver.ref()})
@type sequencer_notify_fn() :: (Bedrock.DataPlane.Sequencer.ref(), Bedrock.epoch(), Bedrock.version(), opts :: keyword() -> :ok | {:error, term()})
@type storage_coverage_error() :: {:storage_team_coverage_error, binary()}
@type success_reply_fn() :: ([ {Bedrock.DataPlane.CommitProxy.Batch.reply_fn(), non_neg_integer(), non_neg_integer()} ], Bedrock.version() -> :ok)
@type timeout_fn() :: (non_neg_integer() -> non_neg_integer())
Functions
@spec create_finalization_plan( Bedrock.DataPlane.CommitProxy.Batch.t(), Bedrock.DataPlane.CommitProxy.RoutingData.t() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec default_timeout_fn(non_neg_integer()) :: non_neg_integer()
@spec extract_result_or_handle_error( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t(), keyword() ) :: {:ok, non_neg_integer(), non_neg_integer(), Bedrock.DataPlane.CommitProxy.RoutingData.t()} | {:error, finalization_error()}
@spec finalize_batch( Bedrock.DataPlane.CommitProxy.Batch.t(), opts :: [ epoch: Bedrock.epoch(), sequencer: pid(), resolver_layout: Bedrock.DataPlane.CommitProxy.ResolverLayout.t(), routing_data: Bedrock.DataPlane.CommitProxy.RoutingData.t(), resolver_fn: resolver_fn(), batch_log_push_fn: log_push_batch_fn(), abort_reply_fn: abort_reply_fn(), success_reply_fn: success_reply_fn(), async_stream_fn: async_stream_fn(), log_push_fn: log_push_single_fn(), sequencer_notify_fn: sequencer_notify_fn(), timeout: non_neg_integer() ] ) :: {:ok, n_aborts :: non_neg_integer(), n_oks :: non_neg_integer(), updated_routing_data :: Bedrock.DataPlane.CommitProxy.RoutingData.t()} | {:error, finalization_error()}
Executes the complete transaction finalization pipeline for a batch of transactions.
This function processes a batch through a multi-stage pipeline: conflict resolution, abort notification, log preparation, log persistence, sequencer notification, and success notification. The pipeline maintains transactional consistency by ensuring all operations complete successfully or all pending clients are notified of failure.
Pipeline Stages
- Conflict Resolution: Calls resolvers to determine which transactions must be aborted
- Abort Notification: Immediately notifies clients of aborted transactions
- Log Preparation: Distributes successful transaction mutations to appropriate logs
- Log Persistence: Pushes transactions to ALL log servers and waits for acknowledgment
- Sequencer Notification: Reports successful commit version to the sequencer
- Success Notification: Notifies clients of successful transactions with commit version
Metadata Distribution
During conflict resolution, metadata mutations (keys with \xFF prefix) are extracted from each transaction and sent to the resolver. The resolver returns differential metadata updates that should be merged into the caller's metadata state.
Parameters
batch: Transaction batch with commit version details from the sequencertransaction_system_layout: System configuration including resolvers and log serversmetadata: Current metadata state (list of accumulated metadata entries)opts: Optional functions for testing and configuration overrides
Returns
{:ok, n_aborts, n_successes, updated_routing_data}- Pipeline completed with updated routing{:error, finalization_error()}- Pipeline failed; all pending clients notified of failure
Error Handling
On any pipeline failure, all transactions that haven't been replied to are automatically notified with abort responses before returning the error.
@spec mutation_to_key_or_range( {:set, Bedrock.key(), Bedrock.value()} | {:clear, Bedrock.key()} | {:clear_range, Bedrock.key(), Bedrock.key()} | {:atomic, atom(), Bedrock.key(), Bedrock.value()} ) :: Bedrock.key() | {Bedrock.key(), Bedrock.key()}
@spec notify_sequencer( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t(), Bedrock.DataPlane.Sequencer.ref(), keyword() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec notify_successes( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t(), keyword() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec prepare_for_logging( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec push_to_logs( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t(), keyword() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec push_transaction_to_logs_direct( last_commit_version :: Bedrock.version(), %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.DataPlane.Transaction.encoded() }, commit_version :: Bedrock.version(), opts :: [ log_services: %{ required(Bedrock.DataPlane.Log.id()) => pid() | {atom(), node()} }, async_stream_fn: async_stream_fn(), log_push_fn: (pid() | {atom(), node()}, binary(), Bedrock.version() -> :ok | {:error, term()}), timeout: non_neg_integer() ] ) :: :ok | {:error, log_push_error()}
Pushes transactions directly to logs and waits for acknowledgement from ALL log servers.
This function takes transactions that have already been built per log and pushes them to the appropriate log servers. Each log receives its pre-built transaction. All logs must acknowledge to maintain durability guarantees.
Parameters
last_commit_version: The last known committed version; used to ensure consistency in log ordering.transactions_by_log: Map of log_id to transaction for that log. May be empty transactions if all transactions were aborted.commit_version: The version assigned by the sequencer for this batch.opts: Optional configuration for testing and customization.
Options
:log_services- Map of log_id to service ref (pid or {name, node}) - REQUIRED:async_stream_fn- Function for parallel processing (default: Task.async_stream/3):log_push_fn- Function for pushing to individual logs (default: try_to_push_transaction_to_log_direct/3):timeout- Timeout for log push operations (default: 5_000ms)
Returns
:okif acknowledgements have been received from ALL log servers.{:error, log_push_error()}if any log has not successfully acknowledged the push within the timeout period or other errors occur.
@spec reply_to_all_clients_with_aborted_transactions([ Bedrock.DataPlane.CommitProxy.Batch.reply_fn() ]) :: :ok
@spec resolve_conflicts( Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t(), Bedrock.epoch(), Bedrock.DataPlane.CommitProxy.ResolverLayout.t(), keyword() ) :: Bedrock.DataPlane.CommitProxy.Finalization.FinalizationPlan.t()
@spec resolve_log_descriptors(%{required(Bedrock.DataPlane.Log.id()) => term()}, %{ required(term()) => Bedrock.ControlPlane.Config.ServiceDescriptor.t() }) :: %{ required(Bedrock.DataPlane.Log.id()) => Bedrock.ControlPlane.Config.ServiceDescriptor.t() }
@spec send_reply_with_commit_version( [Bedrock.DataPlane.CommitProxy.Batch.reply_fn()], Bedrock.version() ) :: :ok
@spec send_reply_with_commit_version_and_index( [ {Bedrock.DataPlane.CommitProxy.Batch.reply_fn(), non_neg_integer(), non_neg_integer()} ], Bedrock.version() ) :: :ok
@spec try_to_push_transaction_to_log( Bedrock.ControlPlane.Config.ServiceDescriptor.t(), binary(), Bedrock.version() ) :: :ok | {:error, :unavailable}