khepri_tx:api_version()

Author: dumbbell

deps/rabbit/src/rabbit_db_queue.erl

@@ -411,8 +411,13 @@ delete_in_khepri(QueueName, OnlyDurable) ->
     rabbit_khepri:transaction(
       fun () ->
               Path = khepri_queue_path(QueueName),
+              TxApiVersion = rabbit_khepri:tx_api_version(),
               case khepri_tx_adv:delete(Path) of
-                  {ok, #{Path := #{data := _}}} ->
+                  {ok, #{Path := #{data := _}}} when TxApiVersion >= 1 ->
+                      %% we want to execute some things, as decided by rabbit_exchange,
+                      %% after the transaction.
+                      rabbit_db_binding:delete_for_destination_in_khepri(QueueName, OnlyDurable);
+                  {ok, #{data := _}} when TxApiVersion =:= 0 ->
                       %% we want to execute some things, as decided by rabbit_exchange,
                       %% after the transaction.
                       rabbit_db_binding:delete_for_destination_in_khepri(QueueName, OnlyDurable);
@@ -1139,20 +1144,26 @@ do_delete_transient_queues_in_khepri(Qs, FilterFun) ->
 do_delete_transient_queues_in_khepri_tx([], Acc) ->
     {ok, Acc};
 do_delete_transient_queues_in_khepri_tx([{Path, Vsn, QName} | Rest], Acc) ->
-     %% Also see `delete_in_khepri/2'.
-     VersionedPath = khepri_path:combine_with_conditions(
-                       Path, [#if_payload_version{version = Vsn}]),
-     case khepri_tx_adv:delete(VersionedPath) of
-         {ok, #{Path := #{data := _}}} ->
-             Deletions = rabbit_db_binding:delete_for_destination_in_khepri(
-                           QName, false),
-             Acc1 = [{QName, Deletions} | Acc],
-             do_delete_transient_queues_in_khepri_tx(Rest, Acc1);
-         {ok, _} ->
-             do_delete_transient_queues_in_khepri_tx(Rest, Acc);
-         {error, _} = Error ->
-             Error
-     end.
+    %% Also see `delete_in_khepri/2'.
+    VersionedPath = khepri_path:combine_with_conditions(
+                      Path, [#if_payload_version{version = Vsn}]),
+    TxApiVersion = rabbit_khepri:tx_api_version(),
+    case khepri_tx_adv:delete(VersionedPath) of
+        {ok, #{Path := #{data := _}}} when TxApiVersion >= 1 ->
+            Deletions = rabbit_db_binding:delete_for_destination_in_khepri(
+                          QName, false),
+            Acc1 = [{QName, Deletions} | Acc],
+            do_delete_transient_queues_in_khepri_tx(Rest, Acc1);
+        {ok, #{data := _}} when TxApiVersion =:= 0 ->
+            Deletions = rabbit_db_binding:delete_for_destination_in_khepri(
+                          QName, false),
+            Acc1 = [{QName, Deletions} | Acc],
+            do_delete_transient_queues_in_khepri_tx(Rest, Acc1);
+        {ok, _} ->
+            do_delete_transient_queues_in_khepri_tx(Rest, Acc);
+        {error, _} = Error ->
+            Error
+    end.
 
 %% -------------------------------------------------------------------
 %% foreach_transient().

deps/rabbit/src/rabbit_db_rtparams.erl

@@ -113,8 +113,11 @@ set_in_khepri_tx(Key, Term) ->
     Path = khepri_rp_path(Key),
     Record = #runtime_parameters{key   = Key,
                                  value = Term},
+    TxApiVersion = rabbit_khepri:tx_api_version(),
     case khepri_tx_adv:put(Path, Record) of
-        {ok, #{Path := #{data := Params}}} ->
+        {ok, #{Path := #{data := Params}}} when TxApiVersion >= 1 ->
+            {old, Params#runtime_parameters.value};
+        {ok, #{data := Params}} when TxApiVersion =:= 0 ->
             {old, Params#runtime_parameters.value};
         {ok, _} ->
             new

deps/rabbit/src/rabbit_khepri.erl

@@ -145,6 +145,7 @@
          transaction/1,
          transaction/2,
          transaction/3,
+         tx_api_version/0,
 
          clear_store/0,
 
@@ -1057,6 +1058,14 @@ transaction(Fun, ReadWrite, Options) ->
         {error, Reason} -> throw({error, Reason})
     end.
 
+tx_api_version() ->
+    try
+        khepri_tx:api_version()
+    catch
+        error:undef ->
+            0
+    end.
+
 clear_store() ->
     khepri:delete_many(?STORE_ID, "*", ?DEFAULT_COMMAND_OPTIONS).