3ad59827436aefbedeccac74876345f233e1cce2

Committed 05 Sep 2025 09:01PM UTC coverage: 63.832% (+5.4%) from 58.423%

Build # 3ad59827436aefbedeccac74876345f233e1cce2

Build Type

push

github

Committed by

jallum

Commit Message

Merge branch 'release/0.2.0'

Run Details

1855 of 2728 new or added lines in 138 files covered. (68.0%)

49 existing lines in 22 files now uncovered.

3228 of 5057 relevant lines covered (63.83%)

597.38 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 / 3ad59827436aefbedeccac74876345f233e1cce2

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