rabbit_db_exchange: Always use a transaction to bump exchange serials

dumbbell · dumbbell · commit 66abb3ad947d · 2026-01-09T09:43:35.000+01:00
[Why]
Initially, I thought a transaction would be too much for a simple
counter. Therefore I use a regular `get`, plus a `put` with a condition
on the tree node payload version to ensure the put would only succeed of
the read value was still the one in the store.

Unfortunately this technic is very inefficient with a lot of
concurrency. That's because all processes will read the same value and
will all commit a `put`, but only one of them will succeed. After that,
all but on will retry the same read/modify/write with the same poor
result. In the end, all processes are spamming `put` commands and they
make very little progress, loading the store in the process.

[How]
This patch goes back to a simple transaction. This solves the contention
issue.
diff --git a/deps/rabbit/src/rabbit_db_exchange.erl b/deps/rabbit/src/rabbit_db_exchange.erl
@@ -535,36 +535,25 @@ next_serial_in_mnesia_tx(XName) ->
     Serial.
 
 next_serial_in_khepri(XName) ->
-    %% Just storing the serial number is enough, no need to keep #exchange_serial{}
-    Path = khepri_exchange_serial_path(XName),
-    Ret1 = rabbit_khepri:adv_get(Path),
-    case Ret1 of
-        {ok, #{Path := #{data := Serial, payload_version := Vsn}}} ->
-            UpdatePath =
-                khepri_path:combine_with_conditions(
-                  Path, [#if_payload_version{version = Vsn}]),
-            case rabbit_khepri:put(UpdatePath, Serial + 1, #{timeout => infinity}) of
-                ok ->
-                    Serial;
-                {error, {khepri, mismatching_node, _}} ->
-                    next_serial_in_khepri(XName)
-            end;
-        _ ->
-            Serial = 1,
-            ok = rabbit_khepri:put(Path, Serial + 1, #{timeout => infinity}),
-            Serial
-    end.
+    Serial = rabbit_khepri:transaction(
+               fun() ->
+                       next_serial_in_khepri_tx(XName)
+               end, rw),
+    Serial.
 
 -spec next_serial_in_khepri_tx(Exchange) -> Serial when
-      Exchange :: rabbit_types:exchange(),
+      Exchange :: rabbit_types:exchange() | rabbit_exchange:name(),
       Serial :: integer().
 
 next_serial_in_khepri_tx(#exchange{name = XName}) ->
+    next_serial_in_khepri_tx(XName);
+next_serial_in_khepri_tx(XName) ->
     Path = khepri_exchange_serial_path(XName),
     Serial = case khepri_tx:get(Path) of
                  {ok, Serial0} -> Serial0;
                  _ -> 1
              end,
+    %% Just storing the serial number is enough, no need to keep #exchange_serial{}
     ok = khepri_tx:put(Path, Serial + 1),
     Serial.
 
