@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.

@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.