Fix transaction reset before pool checkin

lib/db_connection.ex

@@ -1103,7 +1103,7 @@ defmodule DBConnection do
   def transaction(%DBConnection{} = conn, fun, opts) do
     case begin(conn, &run/4, opts) do
       {:ok, _} ->
-        run_transaction(conn, fun, &run/4, opts)
+        run_transaction(conn, fun, &run/4, opts, true)
 
       {:error, error} ->
         rollback_or_raise(error)
@@ -1113,7 +1113,7 @@ defmodule DBConnection do
   def transaction(pool, fun, opts) do
     case begin(pool, &checkout/4, opts) do
       {:ok, conn, _} ->
-        run_transaction(conn, fun, &checkin/4, opts)
+        run_transaction(conn, fun, &transaction_checkin/4, opts, false)
 
       {:error, error} ->
         rollback_or_raise(error)
@@ -1661,12 +1661,6 @@ defmodule DBConnection do
     end
   end
 
-  defp checkin(%DBConnection{} = conn, fun, meter, opts) do
-    return = fun.(conn, meter, opts)
-    checkin(conn)
-    return
-  end
-
   defp meter(opts) do
     case Keyword.get(opts, :log) do
       nil -> nil
@@ -1759,7 +1753,7 @@ defmodule DBConnection do
   defp crash_reason(:throw, value), do: {:nocatch, value}
   defp crash_reason(_, value), do: value
 
-  defp run_transaction(conn, fun, run, opts) do
+  defp run_transaction(conn, fun, run, opts, reset_after?) do
     %DBConnection{conn_ref: conn_ref} = conn
 
     try do
@@ -1801,7 +1795,9 @@ defmodule DBConnection do
             raise err
         end
     after
-      reset(conn)
+      if reset_after? do
+        reset(conn)
+      end
     end
   end
 
@@ -1826,6 +1822,13 @@ defmodule DBConnection do
     end
   end
 
+  defp transaction_checkin(%DBConnection{} = conn, fun, meter, opts) do
+    return = fun.(conn, meter, opts)
+    reset(conn)
+    checkin(conn)
+    return
+  end
+
   defp begin(conn, run, opts) do
     case run.(conn, &run_begin/3, meter(opts), opts) do
       {:ok, conn, {query, result}, meter} ->

test/db_connection_test.exs

@@ -91,4 +91,83 @@ defmodule DBConnectionTest do
       assert :error = DBConnection.connection_module(self())
     end
   end
+
+  test "pool transaction reset after checkin does not clear the next client's aborted status" do
+    stack = [
+      {:ok, :state},
+      {:ok, :begin, :state},
+      {:ok, :commit, :state},
+      {:ok, :begin, :state},
+      {:ok, :rollback, :state}
+    ]
+
+    {:ok, agent} = A.start_link(stack)
+    {:ok, pool} = P.start_link(agent: agent, pool_size: 1)
+
+    parent = self()
+
+    log = fn
+      %DBConnection.LogEntry{call: :commit} ->
+        send(parent, :x_in_commit_log)
+
+        receive do
+          :release_x -> :ok
+        after
+          5000 -> flunk("timed out waiting to release transaction X")
+        end
+
+      %DBConnection.LogEntry{} ->
+        :ok
+    end
+
+    x =
+      Task.async(fn ->
+        Process.put(:agent, agent)
+
+        P.transaction(
+          pool,
+          fn _conn ->
+            :x_done
+          end,
+          log: log
+        )
+      end)
+
+    assert_receive :x_in_commit_log
+
+    y =
+      Task.async(fn ->
+        Process.put(:agent, agent)
+
+        P.transaction(pool, fn conn ->
+          assert {:error, :nested_failed} =
+                   P.transaction(conn, fn conn ->
+                     P.rollback(conn, :nested_failed)
+                   end)
+
+          send(parent, :y_nested_failed)
+
+          receive do
+            :finish_y -> :done
+          after
+            5000 -> flunk("timed out waiting to finish transaction Y")
+          end
+        end)
+      end)
+
+    assert_receive :y_nested_failed
+    send(x.pid, :release_x)
+    assert {:ok, :x_done} = Task.await(x)
+
+    send(y.pid, :finish_y)
+    assert {:error, :rollback} = Task.await(y)
+
+    assert [
+             {:connect, _},
+             {:handle_begin, _},
+             {:handle_commit, _},
+             {:handle_begin, _},
+             {:handle_rollback, _}
+           ] = A.record(agent)
+  end
 end