Bedrock.DataPlane.CommitProxy.Finalization (bedrock v0.5.2)

View Source

Transaction 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 sequencer
  • commit_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

Types

abort_reply_fn()

@type abort_reply_fn() :: ([Bedrock.DataPlane.CommitProxy.Batch.reply_fn()] -> :ok)

async_stream_fn()

@type async_stream_fn() :: (enumerable :: Enumerable.t(),
                      fun :: (term() -> term()),
                      opts :: keyword() ->
                        Enumerable.t())

finalization_error()

@type finalization_error() ::
  resolution_error() | storage_coverage_error() | log_push_error()

log_push_batch_fn()

@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()})

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

log_push_single_fn()

@type log_push_single_fn() :: (Bedrock.ControlPlane.Config.ServiceDescriptor.t(),
                         binary(),
                         Bedrock.version() ->
                           :ok | {:error, :unavailable})

metadata_mutations()

@type metadata_mutations() :: [Bedrock.Internal.TransactionBuilder.Tx.mutation()]

resolution_error()

@type resolution_error() :: :timeout | :unavailable | {:resolver_unavailable, term()}

resolver_fn()

sequencer_notify_fn()

@type sequencer_notify_fn() :: (Bedrock.DataPlane.Sequencer.ref(),
                          Bedrock.epoch(),
                          Bedrock.version(),
                          opts :: keyword() ->
                            :ok | {:error, term()})

storage_coverage_error()

@type storage_coverage_error() :: {:storage_team_coverage_error, binary()}

success_reply_fn()

timeout_fn()

@type timeout_fn() :: (non_neg_integer() -> non_neg_integer())

Functions

create_finalization_plan(batch, routing_data)

default_timeout_fn(attempts_used)

@spec default_timeout_fn(non_neg_integer()) :: non_neg_integer()

extract_result_or_handle_error(plan, opts)

finalize_batch(batch, opts)

@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

  1. Conflict Resolution: Calls resolvers to determine which transactions must be aborted
  2. Abort Notification: Immediately notifies clients of aborted transactions
  3. Log Preparation: Distributes successful transaction mutations to appropriate logs
  4. Log Persistence: Pushes transactions to ALL log servers and waits for acknowledgment
  5. Sequencer Notification: Reports successful commit version to the sequencer
  6. 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 sequencer
  • transaction_system_layout: System configuration including resolvers and log servers
  • metadata: 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.

mutation_to_key_or_range(arg)

@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()}

notify_sequencer(plan, sequencer, opts)

notify_successes(plan, opts)

prepare_for_logging(plan)

push_to_logs(plan, opts)

push_transaction_to_logs_direct(last_commit_version, transactions_by_log, commit_version, opts)

@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

  • :ok if 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.

reply_to_all_clients_with_aborted_transactions(aborts)

@spec reply_to_all_clients_with_aborted_transactions([
  Bedrock.DataPlane.CommitProxy.Batch.reply_fn()
]) ::
  :ok

resolve_conflicts(plan, epoch, resolver_layout, opts)

resolve_log_descriptors(log_descriptors, services)

@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()
}

send_reply_with_commit_version(oks, commit_version)

@spec send_reply_with_commit_version(
  [Bedrock.DataPlane.CommitProxy.Batch.reply_fn()],
  Bedrock.version()
) :: :ok

send_reply_with_commit_version_and_index(entries, commit_version)

@spec send_reply_with_commit_version_and_index(
  [
    {Bedrock.DataPlane.CommitProxy.Batch.reply_fn(), non_neg_integer(),
     non_neg_integer()}
  ],
  Bedrock.version()
) :: :ok

try_to_push_transaction_to_log(arg1, transaction, last_commit_version)

@spec try_to_push_transaction_to_log(
  Bedrock.ControlPlane.Config.ServiceDescriptor.t(),
  binary(),
  Bedrock.version()
) :: :ok | {:error, :unavailable}

try_to_push_transaction_to_log_direct(service_ref, transaction, last_commit_version)

@spec try_to_push_transaction_to_log_direct(
  pid() | {atom(), node()},
  binary(),
  Bedrock.version()
) ::
  :ok | {:error, term()}