84f2ec5d5a2517957ec3c4741969ffc8414a4994

Committed 13 Aug 2025 02:57AM UTC coverage: 58.423%. Remained the same

Build # 84f2ec5d5a2517957ec3c4741969ffc8414a4994

Build Type

push

github

Committed by

jallum

Commit Message

0.1.2

- Reworked and simplified documentation
- Increased test coverage
- Project housekeeping

Run Details

1515 of 2481 new or added lines in 105 files covered. (61.06%)

424 existing lines in 63 files now uncovered.

2112 of 3615 relevant lines covered (58.42%)

1870.88 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

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

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

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

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

  @spec single_transaction_batch(
          state :: State.t(),
          transaction :: Bedrock.transaction(),
          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) do
    case next_commit_version(state.transaction_system_layout.sequencer) do
      {:ok, last_commit_version, commit_version} ->
        {:ok,
         new_batch(timestamp(), 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() | no_return()
  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} ->
        # Sequencer not available - this is a system error that should propagate
        exit({:sequencer_unavailable, reason})
    end
  end

  def start_batch_if_needed(t), do: t

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

jallum / bedrock / 84f2ec5d5a2517957ec3c4741969ffc8414a4994

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