ca26ce31cb47417eebd0556e6b6b89e5ee3d09c7-PR-43

Committed 07 Sep 2025 09:21PM UTC coverage: 63.935% (+0.3%) from 63.674%

Build # ca26ce31cb47417eebd0556e6b6b89e5ee3d09c7-PR-43

Build Type

Pull #43

github

Committed by

jallum

Commit Message

Add periodic tree sweeping to Resolver for memory management

Implement automatic cleanup of old transaction versions in resolver interval trees:

- Add configurable sweep_interval_ms (1s default) and version_retention_ms (6s default)
- Extend State and Server to track sweep timing and configuration
- Sweep triggers when timeout is :infinity or sweep interval elapsed
- Use optimized bulk tree filtering with single rebalance for performance
- Update tests to handle new resolver initialization signature
- Add property tests validating filter correctness

Pull Request Pull Request #43: Add conflict sharding with async resolver assignment

Run Details

105 of 138 new or added lines in 12 files covered. (76.09%)

6 existing lines in 3 files now uncovered.

3292 of 5149 relevant lines covered (63.93%)

1277.06 hits per line

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

14.29

/lib/bedrock/data_plane/commit_proxy/batch.ex

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

  alias Bedrock.DataPlane.Transaction

  @type reply_fn :: ({:ok, Bedrock.version()} | {:error, :abort} -> :ok)

  @type t :: %__MODULE__{
          started_at: Bedrock.timestamp_in_ms(),
          finalized_at: Bedrock.timestamp_in_ms() | nil,
          last_commit_version: Bedrock.version(),
          commit_version: Bedrock.version(),
          n_transactions: non_neg_integer(),
          buffer: [{index :: non_neg_integer(), reply_fn(), Transaction.encoded(), Task.t()}]
        }
  defstruct started_at: nil,
            finalized_at: nil,
            last_commit_version: nil,
            commit_version: nil,
            n_transactions: 0,
            buffer: []

  @spec new_batch(
          Bedrock.timestamp_in_ms(),
          last_commit_version :: Bedrock.version(),
          commit_version :: Bedrock.version()
        ) :: t()
  def new_batch(started_at, last_commit_version, commit_version) do
    %__MODULE__{
      started_at: started_at,
      last_commit_version: last_commit_version,
      commit_version: commit_version,
      n_transactions: 0,
      buffer: []
    }
  end

  @spec transactions_in_order(t()) :: [
          {index :: non_neg_integer(), reply_fn(), Transaction.encoded(), Task.t()}
        ]
  def transactions_in_order(t), do: Enum.reverse(t.buffer)

  @spec all_callers(t()) :: [reply_fn()]
  def all_callers(t), do: Enum.map(t.buffer, &elem(&1, 1))

  @spec add_transaction(t(), Transaction.encoded(), reply_fn(), Task.t()) :: t()
  def add_transaction(t, transaction, reply_fn, task) when is_binary(transaction) do
    index = t.n_transactions
    %{t | buffer: [{index, reply_fn, transaction, task} | t.buffer], n_transactions: index + 1}
  end

  @spec transaction_count(t()) :: non_neg_integer()
  def transaction_count(t), do: t.n_transactions

  @spec set_finalized_at(t(), Bedrock.timestamp_in_ms()) :: t()
  def set_finalized_at(t, finalized_at), do: %{t | finalized_at: finalized_at}
end

1	defmodule Bedrock.DataPlane.CommitProxy.Batch do
2	@moduledoc false
3
4	alias Bedrock.DataPlane.Transaction
5
6	@type reply_fn :: ({:ok, Bedrock.version()} \| {:error, :abort} -> :ok)
7
8	@type t :: %__MODULE__{
9	started_at: Bedrock.timestamp_in_ms(),
10	finalized_at: Bedrock.timestamp_in_ms() \| nil,
11	last_commit_version: Bedrock.version(),
12	commit_version: Bedrock.version(),
13	n_transactions: non_neg_integer(),
14	buffer: [{index :: non_neg_integer(), reply_fn(), Transaction.encoded(), Task.t()}]
15	}
16	defstruct started_at: nil,
17	finalized_at: nil,
18	last_commit_version: nil,
19	commit_version: nil,
20	n_transactions: 0,
21	buffer: []
22
23	@spec new_batch(
24	Bedrock.timestamp_in_ms(),
25	last_commit_version :: Bedrock.version(),
26	commit_version :: Bedrock.version()
27	) :: t()
28	def new_batch(started_at, last_commit_version, commit_version) do
29	%__MODULE__{	×
30	started_at: started_at,
31	last_commit_version: last_commit_version,
32	commit_version: commit_version,
33	n_transactions: 0,
34	buffer: []
35	}
36	end
37
38	@spec transactions_in_order(t()) :: [
39	{index :: non_neg_integer(), reply_fn(), Transaction.encoded(), Task.t()}
40	]
41	def transactions_in_order(t), do: Enum.reverse(t.buffer)	×
42
43	@spec all_callers(t()) :: [reply_fn()]
44	def all_callers(t), do: Enum.map(t.buffer, &elem(&1, 1))	×
45
46	@spec add_transaction(t(), Transaction.encoded(), reply_fn(), Task.t()) :: t()
47	def add_transaction(t, transaction, reply_fn, task) when is_binary(transaction) do
48	index = t.n_transactions	×
NEW 49	%{t \| buffer: [{index, reply_fn, transaction, task} \| t.buffer], n_transactions: index + 1}	×
50	end
51
52	@spec transaction_count(t()) :: non_neg_integer()
53	def transaction_count(t), do: t.n_transactions	42✔
54
55	@spec set_finalized_at(t(), Bedrock.timestamp_in_ms()) :: t()
56	def set_finalized_at(t, finalized_at), do: %{t \| finalized_at: finalized_at}	×
57	end

jallum / bedrock / ca26ce31cb47417eebd0556e6b6b89e5ee3d09c7-PR-43

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