@spec finalize_batch(
  Bedrock.DataPlane.CommitProxy.Batch.t(),
  opts :: [
    epoch: Bedrock.epoch(),
    sequencer: pid(),
    resolver_layout: Bedrock.DataPlane.CommitProxy.ResolverLayout.t(),
    resolver_fn: resolver_fn(),
    resolver_timeout_in_ms: non_neg_integer(),
    proxy_id: pid(),
    metadata_apply_fn: metadata_apply_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()}
  | {: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 every resolver along with the stable proxy identity. Each resolver returns the proxy's exact metadata window. The batch then makes one metadata_apply_fn call: the commit proxy server applies the window - plus, in sharded mode, the batch's own globally-committed metadata - serialized in batch-sequence order, and returns the immutable routing snapshot the batch pushes with.

With SHARDED resolvers no resolver knows the merged global abort set - and none needs to: every resolver receives every transaction's metadata and records it with its LOCAL verdict; the window merge ANDs the verdicts positionally into the exact global verdict (see Bedrock.DataPlane.Resolver).

Parameters

• batch: Transaction batch with commit version details from the sequencer
• opts: Required configuration (epoch, sequencer, resolver_layout, routing_data) plus optional functions for testing and configuration overrides

Returns

• {:ok, n_aborts, n_successes} - Pipeline completed
• {: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 privatized_mutations(term()) :: [{term(), non_neg_integer()}]

The privatized copies a committed mutation must also put on the stream.

A membership CLEAR retires the worker it names, and the worker must learn it IN-BAND — from the stream it already follows — rather than waiting for the next recovery push. So the proxy emits a second copy of the mutation, prefixed past end_of_keyspace/0 and addressed to the shard tag the key itself carries.

The prefix does two jobs, both FDB's (ApplyMetadataMutation.cpp:291-317 privatizes serverKeys with withPrefix(systemKeys.begin) and toCommit->addTag(tag)): it moves the key outside every shard's range so no materializer can ever store it, and it makes the notice unforgeable, because commit ingress rejects that range in BOTH modes — only this function, running after validation, can produce one.

Only removals travel. A worker gaining membership learns it from the distributor that recruited it, so a privatized SET would be a mutation with no reader.

@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() | recovery_required_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 another non-fatal error occurs.
• {:error, recovery_required_error()} if a log reports that the current transaction-system epoch cannot safely continue.