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

8.7

/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: 4, set_finalized_at: 2]

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

  alias Bedrock.DataPlane.CommitProxy.Batch
  alias Bedrock.DataPlane.CommitProxy.ConflictSharding
  alias Bedrock.DataPlane.CommitProxy.LayoutOptimization
  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} ->
        # Create task for resolver assignment (empty transactions still need processing)
        task = create_resolver_task(transaction, state.precomputed_layout)

        {:ok,
         timestamp()
         |> new_batch(last_commit_version, commit_version)
         |> add_transaction(transaction, reply_fn, task)
         |> set_finalized_at(timestamp())}

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

  @spec create_resolver_task(Transaction.encoded(), LayoutOptimization.precomputed_layout()) :: Task.t()
  def create_resolver_task(transaction, %{resolver_refs: [single_ref], resolver_ends: _ends}) do
    Task.async(fn ->
      sections = Transaction.extract_sections!(transaction, [:read_conflicts, :write_conflicts])
      %{single_ref => sections}
    end)
  end

  def create_resolver_task(transaction, %{resolver_refs: refs, resolver_ends: ends}) do
    Task.async(fn ->
      sections = Transaction.extract_sections!(transaction, [:read_conflicts, :write_conflicts])
      ConflictSharding.shard_conflicts_across_resolvers(sections, ends, refs)
    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(), Task.t()) ::
          State.t()
  def add_transaction_to_batch(t, transaction, reply_fn, task) when is_binary(transaction),
    do: %{t | batch: add_transaction(t.batch, transaction, reply_fn, task)}

  @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: 4, 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.ConflictSharding
11	alias Bedrock.DataPlane.CommitProxy.LayoutOptimization
12	alias Bedrock.DataPlane.CommitProxy.State
13	alias Bedrock.DataPlane.Transaction
14
15	@spec timestamp() :: Bedrock.timestamp_in_ms()
16	defp timestamp, do: :erlang.monotonic_time(:millisecond)	×
17
18	@spec single_transaction_batch(
19	state :: State.t(),
20	transaction :: Transaction.encoded(),
21	reply_fn :: Batch.reply_fn()
22	) ::
23	{:ok, Batch.t()}
24	\| {:error, :sequencer_unavailable}
25	def single_transaction_batch(t, transaction, reply_fn \\ fn _result -> :ok end)	1✔
26
27	def single_transaction_batch(%{transaction_system_layout: %{sequencer: nil}}, _transaction, _reply_fn),	1✔
28	do: {:error, :sequencer_unavailable}
29
30	def single_transaction_batch(state, transaction, reply_fn) when is_binary(transaction) do
31	case next_commit_version(state.transaction_system_layout.sequencer) do	×
32	{:ok, last_commit_version, commit_version} ->
33	# Create task for resolver assignment (empty transactions still need processing)
NEW 34	task = create_resolver_task(transaction, state.precomputed_layout)	×
35
36	{:ok,
37	timestamp()
38	\|> new_batch(last_commit_version, commit_version)
39	\|> add_transaction(transaction, reply_fn, task)
40	\|> set_finalized_at(timestamp())}
41
42	{:error, :unavailable} ->	×
43	{:error, :sequencer_unavailable}
44	end
45	end
46
47	@spec create_resolver_task(Transaction.encoded(), LayoutOptimization.precomputed_layout()) :: Task.t()
48	def create_resolver_task(transaction, %{resolver_refs: [single_ref], resolver_ends: _ends}) do
NEW 49	Task.async(fn ->	×
NEW 50	sections = Transaction.extract_sections!(transaction, [:read_conflicts, :write_conflicts])	×
NEW 51	%{single_ref => sections}	×
52	end)
53	end
54
55	def create_resolver_task(transaction, %{resolver_refs: refs, resolver_ends: ends}) do
NEW 56	Task.async(fn ->	×
NEW 57	sections = Transaction.extract_sections!(transaction, [:read_conflicts, :write_conflicts])	×
NEW 58	ConflictSharding.shard_conflicts_across_resolvers(sections, ends, refs)	×
59	end)
60	end
61
62	@spec start_batch_if_needed(State.t()) :: State.t() \| {:error, term()}
63	def start_batch_if_needed(%{batch: nil} = t) do
64	case next_commit_version(t.transaction_system_layout.sequencer) do	×
65	{:ok, last_commit_version, commit_version} ->
66	%{t \| batch: new_batch(timestamp(), last_commit_version, commit_version)}	×
67
68	{:error, reason} ->	×
69	{:error, {:sequencer_unavailable, reason}}
70	end
71	end
72
73	def start_batch_if_needed(t), do: t	×
74
75	@spec add_transaction_to_batch(State.t(), Transaction.encoded(), Batch.reply_fn(), Task.t()) ::
76	State.t()
77	def add_transaction_to_batch(t, transaction, reply_fn, task) when is_binary(transaction),
NEW 78	do: %{t \| batch: add_transaction(t.batch, transaction, reply_fn, task)}	×
79
80	@spec apply_finalization_policy(State.t()) ::
81	{State.t(), batch_to_finalize :: Batch.t()} \| {State.t(), nil}
82	def apply_finalization_policy(t) do
83	now = timestamp()	×
84
85	if max_latency?(t.batch, now, t.max_latency_in_ms) or	×
86	max_transactions?(t.batch, t.max_per_batch) do	×
87	{%{t \| batch: nil}, set_finalized_at(t.batch, now)}	×
88	else
89	{t, nil}
90	end
91	end
92
93	@spec max_latency?(
94	Batch.t(),
95	now :: Bedrock.timestamp_in_ms(),
96	max_latency_in_ms :: pos_integer()
97	) :: boolean()
98	defp max_latency?(batch, now, max_latency_in_ms), do: batch.started_at + max_latency_in_ms < now	×
99
100	@spec max_transactions?(Batch.t(), max_per_batch :: pos_integer()) :: boolean()
101	defp max_transactions?(batch, max_per_batch), do: batch.n_transactions >= max_per_batch	×
102	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