Recached

fb3defeb951c4b0ad9552be01797f0e4e04d962f-PR-43

Committed 03 Sep 2025 03:45AM UTC coverage: 63.433% (+0.005%) from 63.428%

Build # fb3defeb951c4b0ad9552be01797f0e4e04d962f-PR-43

Build Type

Pull #43

github

Committed by

jallum

Commit Message

Add conflict sharding with async resolver assignment

Move expensive conflict distribution off critical path by computing resolver
assignments asynchronously during batching instead of blocking transaction
acceptance. Improves throughput and reduces latency.

- Add ConflictSharding module to distribute conflicts by key range
- Implement async Task-based resolver assignment in batching
- Add LayoutOptimization for precomputed static structures
- Update Batch and finalization pipeline for Task integration
- Extract KeyRange.overlap? utility, remove ResolutionPlan

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

Run Details

63 of 95 new or added lines in 10 files covered. (66.32%)

2 existing lines in 2 files now uncovered.

3027 of 4772 relevant lines covered (63.43%)

615.91 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	21✔
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 / fb3defeb951c4b0ad9552be01797f0e4e04d962f-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