022bf889f44fabf5818028244ac5e8e953e1a798

Committed 18 Aug 2025 02:00AM UTC coverage: 62.424% (+2.5%) from 59.888%

Build # 022bf889f44fabf5818028244ac5e8e953e1a798

Build Type

push

github

Committed by

web-flow

Commit Message

Feature/31 (#35)

# Major Transaction System Overhaul

This branch represents a **fundamental rearchitecture of Bedrock's
transaction processing system** with three main
phases of development spanning performance, unification, reliability,
and architectural improvements.

  ## Phase 1: Transaction Format Unification (fd66e2d)
**Rationale**: Replace the fragmented, inconsistent transaction encoding
system with a unified, robust approach.

  **Key Changes**:
- **Consolidated Transaction Format**: Merged `read_conflicts` and
`read_version` into a single tuple format,
  eliminating redundancy
- **New BedrockTransaction Module**: Replaced the limited
`EncodedTransaction` with a comprehensive 971-line module
  featuring:
    - Tagged binary encoding with self-describing sections
    - CRC validation for data integrity
    - Order-independent sections for better extensibility
    - Efficient partial decoding capabilities
- **Removed Legacy Code**: Eliminated obsolete resolver recovery logic
and transaction fragmentation
- **Comprehensive Testing**: Added extensive test coverage including
binary integration tests

  ## Phase 2: Enhanced Transaction Processing (f96a1ed)
**Rationale**: Address reliability and robustness gaps in transaction
handling.

  **Key Improvements**:
- **Transaction Validation**: Added validation in resolver to ensure
proper transaction format before processing
- **Timeout Management**: Implemented `WaitingList` module for
systematic timeout handling across components
- **Error Resilience**: Enhanced error handling for Director
notifications and unavailable states
- **Standardization**: Moved Logger.require statements to module level
for consistency

  ## Phase 3: Commit Proxy Rework & System-Wide Improvements (dc6a048)
**Rationale**: Optimize transaction distribution and improve
observability across the data plane.

  **Major Architectural Changes**:
- **Single Transaction Per Log**: Reworked commit proxy to produce one
transaction per ... (continued)

Run Details

817 of 1213 new or added lines in 110 files covered. (67.35%)

40 existing lines in 21 files now uncovered.

2369 of 3795 relevant lines covered (62.42%)

871.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

11.76

/lib/bedrock/data_plane/commit_proxy/batching.ex

defmodule Bedrock.DataPlane.CommitProxy.Batching do
  @moduledoc false

  import Bedrock.DataPlane.CommitProxy.Batch,
    only: [new_batch: 3, add_transaction: 3, set_finalized_at: 2]

  import Bedrock.DataPlane.Sequencer, only: [next_commit_version: 1]

  alias Bedrock.DataPlane.CommitProxy.Batch
  alias Bedrock.DataPlane.CommitProxy.State
  alias Bedrock.DataPlane.Transaction

  @spec timestamp() :: Bedrock.timestamp_in_ms()
  defp timestamp, do: :erlang.monotonic_time(:millisecond)

  @spec single_transaction_batch(
          state :: State.t(),
          transaction :: Transaction.encoded(),
          reply_fn :: Batch.reply_fn()
        ) ::
          {:ok, Batch.t()}
          | {:error, :sequencer_unavailable}
  def single_transaction_batch(t, transaction, reply_fn \\ fn _result -> :ok end)

  def single_transaction_batch(%{transaction_system_layout: %{sequencer: nil}}, _transaction, _reply_fn),
    do: {:error, :sequencer_unavailable}

  def single_transaction_batch(state, transaction, reply_fn) when is_binary(transaction) do
    case next_commit_version(state.transaction_system_layout.sequencer) do
      {:ok, last_commit_version, commit_version} ->
        {:ok,
         timestamp()
         |> new_batch(last_commit_version, commit_version)
         |> add_transaction(transaction, reply_fn)
         |> set_finalized_at(timestamp())}

      {:error, :unavailable} ->
        {:error, :sequencer_unavailable}
    end
  end

  @spec start_batch_if_needed(State.t()) :: State.t() | {:error, term()}
  def start_batch_if_needed(%{batch: nil} = t) do
    case next_commit_version(t.transaction_system_layout.sequencer) do
      {:ok, last_commit_version, commit_version} ->
        %{t | batch: new_batch(timestamp(), last_commit_version, commit_version)}

      {:error, reason} ->
        {:error, {:sequencer_unavailable, reason}}
    end
  end

  def start_batch_if_needed(t), do: t

  @spec add_transaction_to_batch(State.t(), Transaction.encoded(), Batch.reply_fn()) ::
          State.t()
  def add_transaction_to_batch(t, transaction, reply_fn) when is_binary(transaction),
    do: %{t | batch: add_transaction(t.batch, transaction, reply_fn)}

  @spec apply_finalization_policy(State.t()) ::
          {State.t(), batch_to_finalize :: Batch.t()} | {State.t(), nil}
  def apply_finalization_policy(t) do
    now = timestamp()

    if max_latency?(t.batch, now, t.max_latency_in_ms) or
         max_transactions?(t.batch, t.max_per_batch) do
      {%{t | batch: nil}, set_finalized_at(t.batch, now)}
    else
      {t, nil}
    end
  end

  @spec max_latency?(
          Batch.t(),
          now :: Bedrock.timestamp_in_ms(),
          max_latency_in_ms :: pos_integer()
        ) :: boolean()
  defp max_latency?(batch, now, max_latency_in_ms), do: batch.started_at + max_latency_in_ms < now

  @spec max_transactions?(Batch.t(), max_per_batch :: pos_integer()) :: boolean()
  defp max_transactions?(batch, max_per_batch), do: batch.n_transactions >= max_per_batch
end

1	defmodule Bedrock.DataPlane.CommitProxy.Batching do
2	@moduledoc false
3
4	import Bedrock.DataPlane.CommitProxy.Batch,
5	only: [new_batch: 3, add_transaction: 3, set_finalized_at: 2]
6
7	import Bedrock.DataPlane.Sequencer, only: [next_commit_version: 1]
8
9	alias Bedrock.DataPlane.CommitProxy.Batch
10	alias Bedrock.DataPlane.CommitProxy.State
11	alias Bedrock.DataPlane.Transaction
12
13	@spec timestamp() :: Bedrock.timestamp_in_ms()
14	defp timestamp, do: :erlang.monotonic_time(:millisecond)	×
15
16	@spec single_transaction_batch(
17	state :: State.t(),
18	transaction :: Transaction.encoded(),
19	reply_fn :: Batch.reply_fn()
20	) ::
21	{:ok, Batch.t()}
22	\| {:error, :sequencer_unavailable}
23	def single_transaction_batch(t, transaction, reply_fn \\ fn _result -> :ok end)	1✔
24
25	def single_transaction_batch(%{transaction_system_layout: %{sequencer: nil}}, _transaction, _reply_fn),	1✔
26	do: {:error, :sequencer_unavailable}
27
28	def single_transaction_batch(state, transaction, reply_fn) when is_binary(transaction) do
29	case next_commit_version(state.transaction_system_layout.sequencer) do	×
30	{:ok, last_commit_version, commit_version} ->	×
31	{:ok,
32	timestamp()
33	\|> new_batch(last_commit_version, commit_version)
34	\|> add_transaction(transaction, reply_fn)
35	\|> set_finalized_at(timestamp())}
36
37	{:error, :unavailable} ->	×
38	{:error, :sequencer_unavailable}
39	end
40	end
41
42	@spec start_batch_if_needed(State.t()) :: State.t() \| {:error, term()}
43	def start_batch_if_needed(%{batch: nil} = t) do
44	case next_commit_version(t.transaction_system_layout.sequencer) do	×
45	{:ok, last_commit_version, commit_version} ->
46	%{t \| batch: new_batch(timestamp(), last_commit_version, commit_version)}	×
47
UNCOV 48	{:error, reason} ->	×
49	{:error, {:sequencer_unavailable, reason}}
50	end
51	end
52
53	def start_batch_if_needed(t), do: t	×
54
55	@spec add_transaction_to_batch(State.t(), Transaction.encoded(), Batch.reply_fn()) ::
56	State.t()
57	def add_transaction_to_batch(t, transaction, reply_fn) when is_binary(transaction),
NEW 58	do: %{t \| batch: add_transaction(t.batch, transaction, reply_fn)}	×
59
60	@spec apply_finalization_policy(State.t()) ::
61	{State.t(), batch_to_finalize :: Batch.t()} \| {State.t(), nil}
62	def apply_finalization_policy(t) do
63	now = timestamp()	×
64
65	if max_latency?(t.batch, now, t.max_latency_in_ms) or	×
66	max_transactions?(t.batch, t.max_per_batch) do	×
NEW 67	{%{t \| batch: nil}, set_finalized_at(t.batch, now)}	×
68	else
69	{t, nil}
70	end
71	end
72
73	@spec max_latency?(
74	Batch.t(),
75	now :: Bedrock.timestamp_in_ms(),
76	max_latency_in_ms :: pos_integer()
77	) :: boolean()
NEW 78	defp max_latency?(batch, now, max_latency_in_ms), do: batch.started_at + max_latency_in_ms < now	×
79
80	@spec max_transactions?(Batch.t(), max_per_batch :: pos_integer()) :: boolean()
NEW 81	defp max_transactions?(batch, max_per_batch), do: batch.n_transactions >= max_per_batch	×
82	end

jallum / bedrock / 022bf889f44fabf5818028244ac5e8e953e1a798

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous