Merge pull request rails#44526 from matthewd/restart-transaction

Rollback nested transactions by restarting the parent where possible

Author: matthewd

activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb

@@ -323,7 +323,8 @@ def transaction(requires_new: nil, isolation: nil, joinable: true, &block)
 
       delegate :within_new_transaction, :open_transactions, :current_transaction, :begin_transaction,
                :commit_transaction, :rollback_transaction, :materialize_transactions,
-               :disable_lazy_transactions!, :enable_lazy_transactions!, to: :transaction_manager
+               :disable_lazy_transactions!, :enable_lazy_transactions!, :dirty_current_transaction,
+               to: :transaction_manager
 
       def transaction_open?
         current_transaction.open?
@@ -369,6 +370,12 @@ def rollback_db_transaction
 
       def exec_rollback_db_transaction() end # :nodoc:
 
+      def restart_db_transaction
+        exec_restart_db_transaction
+      end
+
+      def exec_restart_db_transaction() end # :nodoc:
+
       def rollback_to_savepoint(name = nil)
         exec_rollback_to_savepoint(name)
       end

activerecord/lib/active_record/connection_adapters/abstract/query_cache.rb

@@ -8,7 +8,7 @@ module QueryCache
       class << self
         def included(base) # :nodoc:
           dirties_query_cache base, :create, :insert, :update, :delete, :truncate, :truncate_tables,
-            :rollback_to_savepoint, :rollback_db_transaction, :exec_insert_all
+            :rollback_to_savepoint, :rollback_db_transaction, :restart_db_transaction, :exec_insert_all
 
           base.set_callback :checkout, :after, :configure_query_cache!
           base.set_callback :checkin, :after, :disable_query_cache!

activerecord/lib/active_record/connection_adapters/abstract/transaction.rb

@@ -80,6 +80,9 @@ def closed?; true; end
       def open?; false; end
       def joinable?; false; end
       def add_record(record, _ = true); end
+      def restartable?; false; end
+      def dirty?; false; end
+      def dirty!; end
     end
 
     class Transaction # :nodoc:
@@ -94,6 +97,15 @@ def initialize(connection, isolation: nil, joinable: true, run_commit_callbacks:
         @joinable = joinable
         @run_commit_callbacks = run_commit_callbacks
         @lazy_enrollment_records = nil
+        @dirty = false
+      end
+
+      def dirty!
+        @dirty = true
+      end
+
+      def dirty?
+        @dirty
       end
 
       def add_record(record, ensure_finalize = true)
@@ -114,6 +126,12 @@ def records
         @records
       end
 
+      # Can this transaction's current state be recreated by
+      # rollback+begin ?
+      def restartable?
+        joinable? && !dirty?
+      end
+
       def materialize!
         @materialized = true
       end
@@ -167,6 +185,33 @@ def closed?; false; end
       def open?; !closed?; end
     end
 
+    class RestartParentTransaction < Transaction
+      def initialize(connection, parent_transaction, **options)
+        super(connection, **options)
+
+        @parent = parent_transaction
+
+        if isolation_level
+          raise ActiveRecord::TransactionIsolationError, "cannot set transaction isolation in a nested transaction"
+        end
+
+        @parent.state.add_child(@state)
+      end
+
+      delegate :materialize!, :materialized?, :restart, to: :@parent
+
+      def rollback
+        @state.rollback!
+        @parent.restart
+      end
+
+      def commit
+        @state.commit!
+      end
+
+      def full_rollback?; false; end
+    end
+
     class SavepointTransaction < Transaction
       def initialize(connection, savepoint_name, parent_transaction, **options)
         super(connection, **options)
@@ -185,6 +230,10 @@ def materialize!
         super
       end
 
+      def restart
+        connection.rollback_to_savepoint(savepoint_name) if materialized?
+      end
+
       def rollback
         unless @state.invalidated?
           connection.rollback_to_savepoint(savepoint_name) if materialized?
@@ -211,6 +260,17 @@ def materialize!
         super
       end
 
+      def restart
+        return unless materialized?
+
+        if connection.supports_restart_db_transaction?
+          connection.restart_db_transaction
+        else
+          connection.rollback_db_transaction
+          materialize!
+        end
+      end
+
       def rollback
         connection.rollback_db_transaction if materialized?
         @state.full_rollback!
@@ -242,11 +302,19 @@ def begin_transaction(isolation: nil, joinable: true, _lazy: true)
                 joinable: joinable,
                 run_commit_callbacks: run_commit_callbacks
               )
+            elsif current_transaction.restartable?
+              RestartParentTransaction.new(
+                @connection,
+                current_transaction,
+                isolation: isolation,
+                joinable: joinable,
+                run_commit_callbacks: run_commit_callbacks
+              )
             else
               SavepointTransaction.new(
                 @connection,
                 "active_record_#{@stack.size}",
-                @stack.last,
+                current_transaction,
                 isolation: isolation,
                 joinable: joinable,
                 run_commit_callbacks: run_commit_callbacks
@@ -276,8 +344,20 @@ def lazy_transactions_enabled?
         @lazy_transactions_enabled
       end
 
+      def dirty_current_transaction
+        current_transaction.dirty!
+      end
+
       def materialize_transactions
         return if @materializing_transactions
+
+        # As a logical simplification for now, we assume anything that requests
+        # materialization is about to dirty the transaction. Note this is just
+        # an assumption about the caller, not a direct property of this method.
+        # It can go away later when callers are able to handle dirtiness for
+        # themselves.
+        dirty_current_transaction
+
         return unless @has_unmaterialized_transactions
 
         @connection.lock.synchronize do
@@ -301,6 +381,8 @@ def commit_transaction
             @stack.pop
           end
 
+          dirty_current_transaction if transaction.dirty?
+
           transaction.commit
           transaction.commit_records
         end

activerecord/lib/active_record/connection_adapters/abstract_adapter.rb

@@ -335,6 +335,10 @@ def savepoint_errors_invalidate_transactions?
         false
       end
 
+      def supports_restart_db_transaction?
+        false
+      end
+
       # Does this adapter support application-enforced advisory locking?
       def supports_advisory_locks?
         false

activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb

@@ -81,6 +81,10 @@ def supports_transaction_isolation?
         true
       end
 
+      def supports_restart_db_transaction?
+        true
+      end
+
       def supports_explain?
         true
       end
@@ -224,6 +228,10 @@ def exec_rollback_db_transaction # :nodoc:
         execute("ROLLBACK", "TRANSACTION")
       end
 
+      def exec_restart_db_transaction # :nodoc:
+        execute("ROLLBACK AND CHAIN", "TRANSACTION")
+      end
+
       def empty_insert_statement_value(primary_key = nil) # :nodoc:
         "VALUES ()"
       end

activerecord/lib/active_record/connection_adapters/postgresql/database_statements.rb

@@ -109,8 +109,7 @@ def begin_db_transaction # :nodoc:
         end
 
         def begin_isolated_db_transaction(isolation) # :nodoc:
-          begin_db_transaction
-          execute "SET TRANSACTION ISOLATION LEVEL #{transaction_isolation_levels.fetch(isolation)}"
+          execute("BEGIN ISOLATION LEVEL #{transaction_isolation_levels.fetch(isolation)}", "TRANSACTION")
         end
 
         # Commits a transaction.
@@ -125,6 +124,12 @@ def exec_rollback_db_transaction # :nodoc:
           execute("ROLLBACK", "TRANSACTION")
         end
 
+        def exec_restart_db_transaction # :nodoc:
+          @raw_connection.cancel unless @raw_connection.transaction_status == PG::PQTRANS_IDLE
+          @raw_connection.block
+          execute("ROLLBACK AND CHAIN", "TRANSACTION")
+        end
+
         # From https://www.postgresql.org/docs/current/functions-datetime.html#FUNCTIONS-DATETIME-CURRENT
         HIGH_PRECISION_CURRENT_TIMESTAMP = Arel.sql("CURRENT_TIMESTAMP").freeze # :nodoc:
         private_constant :HIGH_PRECISION_CURRENT_TIMESTAMP

activerecord/lib/active_record/connection_adapters/postgresql/schema_statements.rb

@@ -288,7 +288,7 @@ def reset_pk_sequence!(table, pk = nil, sequence = nil) # :nodoc:
             quoted_sequence = quote_table_name(sequence)
             max_pk = query_value("SELECT MAX(#{quote_column_name pk}) FROM #{quote_table_name(table)}", "SCHEMA")
             if max_pk.nil?
-              if database_version >= 100000
+              if database_version >= 10_00_00
                 minvalue = query_value("SELECT seqmin FROM pg_sequence WHERE seqrelid = #{quote(quoted_sequence)}::regclass", "SCHEMA")
               else
                 minvalue = query_value("SELECT min_value FROM #{quoted_sequence}", "SCHEMA")

activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb

@@ -183,7 +183,7 @@ def supports_index_sort_order?
       end
 
       def supports_partitioned_indexes?
-        database_version >= 110_000 # >= 11.0
+        database_version >= 11_00_00 # >= 11.0
       end
 
       def supports_partial_index?
@@ -234,19 +234,23 @@ def supports_savepoints?
         true
       end
 
+      def supports_restart_db_transaction?
+        database_version >= 12_00_00 # >= 12.0
+      end
+
       def supports_insert_returning?
         true
       end
 
       def supports_insert_on_conflict?
-        database_version >= 90500 # >= 9.5
+        database_version >= 9_05_00 # >= 9.5
       end
       alias supports_insert_on_duplicate_skip? supports_insert_on_conflict?
       alias supports_insert_on_duplicate_update? supports_insert_on_conflict?
       alias supports_insert_conflict_target? supports_insert_on_conflict?
 
       def supports_virtual_columns?
-        database_version >= 120_000 # >= 12.0
+        database_version >= 12_00_00 # >= 12.0
       end
 
       def index_algorithms
@@ -398,7 +402,7 @@ def supports_foreign_tables?
       end
 
       def supports_pgcrypto_uuid?
-        database_version >= 90400 # >= 9.4
+        database_version >= 9_04_00 # >= 9.4
       end
 
       def supports_optimizer_hints?
@@ -531,7 +535,7 @@ def build_insert_sql(insert) # :nodoc:
       end
 
       def check_version # :nodoc:
-        if database_version < 90300 # < 9.3
+        if database_version < 9_03_00 # < 9.3
           raise "Your version of PostgreSQL (#{database_version}) is too old. Active Record supports PostgreSQL >= 9.3."
         end
       end

activerecord/test/cases/transactions_test.rb

@@ -697,10 +697,14 @@ def test_releasing_named_savepoints
 
   def test_savepoints_name
     Topic.transaction do
+      Topic.delete_all # Dirty the transaction to force a savepoint below
+
       assert_nil Topic.connection.current_savepoint_name
       assert_nil Topic.connection.current_transaction.savepoint_name
 
       Topic.transaction(requires_new: true) do
+        Topic.delete_all # Dirty the transaction to force a savepoint below
+
         assert_equal "active_record_1", Topic.connection.current_savepoint_name
         assert_equal "active_record_1", Topic.connection.current_transaction.savepoint_name
 
@@ -1102,6 +1106,43 @@ def test_unprepared_statement_materializes_transaction
     end
   end
 
+  def test_nested_transactions_skip_excess_savepoints
+    capture_sql do
+      # RealTransaction (begin..commit)
+      Topic.transaction(requires_new: true) do
+        # ResetParentTransaction (no queries)
+        Topic.transaction(requires_new: true) do
+          Topic.delete_all
+          # SavepointTransaction (savepoint..release)
+          Topic.transaction(requires_new: true) do
+            # ResetParentTransaction (no queries)
+            Topic.transaction(requires_new: true) do
+              Topic.delete_all
+            end
+          end
+        end
+        Topic.delete_all
+      end
+    end
+
+    actual_queries = ActiveRecord::SQLCounter.log_all
+
+    expected_queries = [
+      /BEGIN/i,
+      /DELETE/i,
+      /^SAVEPOINT/i,
+      /DELETE/i,
+      /^RELEASE/i,
+      /DELETE/i,
+      /COMMIT/i,
+    ]
+
+    assert_equal expected_queries.size, actual_queries.size
+    expected_queries.zip(actual_queries) do |expected, actual|
+      assert_match expected, actual
+    end
+  end
+
   if ActiveRecord::Base.connection.prepared_statements
     def test_prepared_statement_materializes_transaction
       Topic.first