feat: transaction isolation level

olavloite · olavloite · commit a03d44e34b7a · 2025-03-30T21:42:18.000+02:00
Add support for :repeatable_read transaction isolation level. This can be
used in the following ways:
- By passing in a value for the :isolation argument when starting a transaction.
- By setting the :isolation_level property of a connection.
- By setting the :isolation_level property in the database config.
diff --git a/.github/workflows/acceptance-tests-on-emulator.yaml b/.github/workflows/acceptance-tests-on-emulator.yaml
@@ -38,7 +38,7 @@ jobs:
     - name: Install dependencies
       run: bundle install
     - name: Run acceptance tests on emulator
-      run: bundle exec rake acceptance
+      run: bundle exec rake acceptance TESTOPTS="-v"
       env:
         SPANNER_EMULATOR_HOST: localhost:9010
         SPANNER_TEST_PROJECT: test-project
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -30,4 +30,4 @@ jobs:
     - name: Install dependencies
       run: bundle install
     - name: Run tests
-      run: bundle exec rake test
+      run: bundle exec rake test TESTOPTS="-v"
diff --git a/acceptance/test_helper.rb b/acceptance/test_helper.rb
@@ -188,6 +188,7 @@ def teardown
         unless @skip_test_table_create
           connection.drop_table :test_models, if_exists: true
         end
+        ActiveRecord::Base.connection_pool.disconnect!
 
         super
       end
@@ -275,6 +276,11 @@ def call(name, start, finish, message_id, values)
 
 Minitest.after_run do
   drop_test_database
+  ActiveRecordSpannerAdapter::Connection.spanners_map.each do |_, spanner|
+    channel = spanner.service.service.spanner_stub.grpc_stub.instance_variable_get(:@ch)
+    channel.close
+  end
+  ActiveRecordSpannerAdapter::Connection.spanners_map.clear
 end
 
 create_test_database
diff --git a/activerecord-spanner-adapter.gemspec b/activerecord-spanner-adapter.gemspec
@@ -24,9 +24,8 @@ Gem::Specification.new do |spec|
 
   spec.required_ruby_version = ">= 3.1"
 
-  spec.add_dependency "google-cloud-spanner", "~> 2.18"
-  # Pin gRPC to 1.64.3, as 1.65 and 1.66 cause test runs to hang randomly.
-  spec.add_dependency "grpc", "1.64.3"
+  spec.add_dependency "google-cloud-spanner", "~> 2.25"
+  spec.add_dependency "google-cloud-spanner-v1", "~> 1.7"
   spec.add_runtime_dependency "activerecord", [">= 7.0", "< 9"]
 
   spec.add_development_dependency "autotest-suffix", "~> 1.1"
diff --git a/lib/active_record/connection_adapters/spanner/database_statements.rb b/lib/active_record/connection_adapters/spanner/database_statements.rb
@@ -292,7 +292,7 @@ def begin_isolated_db_transaction isolation
               if isolation.count != 1
           else
             raise "Unsupported isolation level: #{isolation}" unless
-              [:serializable, :read_only, :buffered_mutations, :pdml].include? isolation
+              [:serializable, :repeatable_read, :read_only, :buffered_mutations, :pdml].include? isolation
           end
 
           log "BEGIN #{isolation}" do
diff --git a/lib/active_record/connection_adapters/spanner_adapter.rb b/lib/active_record/connection_adapters/spanner_adapter.rb
@@ -150,7 +150,8 @@ def spanner_schema_cache
       end
 
       # Spanner Connection API
-      delegate :ddl_batch, :ddl_batch?, :start_batch_ddl, :abort_batch, :run_batch, to: :@connection
+      delegate :ddl_batch, :ddl_batch?, :start_batch_ddl, :abort_batch, :run_batch,
+               :isolation_level, :isolation_level=, to: :@connection
 
       def current_spanner_transaction
         @connection.current_transaction
@@ -170,6 +171,10 @@ def supports_explain?
         false
       end
 
+      def supports_transaction_isolation?
+        true
+      end
+
       def supports_foreign_keys?
         true
       end
diff --git a/lib/activerecord_spanner_adapter/connection.rb b/lib/activerecord_spanner_adapter/connection.rb
@@ -14,10 +14,12 @@ class Connection
     attr_reader :database_id
     attr_reader :spanner
     attr_accessor :current_transaction
+    attr_accessor :isolation_level
 
     def initialize config
       @instance_id = config[:instance]
       @database_id = config[:database]
+      @isolation_level = config[:isolation_level]
       @spanner = self.class.spanners config
     end
 
@@ -38,6 +40,10 @@ def self.spanners config
       end
     end
 
+    def self.spanners_map
+      @spanners
+    end
+
     # Clears the cached information about the underlying information schemas.
     # Call this method if you drop and recreate a database with the same name
     # to prevent the cached information to be used for the new database.
@@ -271,7 +277,7 @@ def create_transaction_after_failed_first_statement original_error
 
     def begin_transaction isolation = nil
       raise "Nested transactions are not allowed" if current_transaction&.active?
-      self.current_transaction = Transaction.new self, isolation
+      self.current_transaction = Transaction.new self, isolation || @isolation_level
       current_transaction.begin
       current_transaction
     end
diff --git a/lib/activerecord_spanner_adapter/transaction.rb b/lib/activerecord_spanner_adapter/transaction.rb
@@ -55,11 +55,12 @@ def begin
         when :pdml
           @grpc_transaction = @connection.session.create_pdml
         else
+          grpc_isolation = _transaction_isolation_level_to_grpc @isolation
           @begin_transaction_selector = Google::Cloud::Spanner::V1::TransactionSelector.new \
             begin: Google::Cloud::Spanner::V1::TransactionOptions.new(
-              read_write: Google::Cloud::Spanner::V1::TransactionOptions::ReadWrite.new
+              read_write: Google::Cloud::Spanner::V1::TransactionOptions::ReadWrite.new,
+              isolation_level: grpc_isolation
             )
-
         end
         @state = :STARTED
       rescue Google::Cloud::NotFoundError => e
@@ -75,6 +76,15 @@ def begin
       end
     end
 
+    def _transaction_isolation_level_to_grpc isolation
+      case isolation
+      when :serializable
+        Google::Cloud::Spanner::V1::TransactionOptions::IsolationLevel::SERIALIZABLE
+      when :repeatable_read
+        Google::Cloud::Spanner::V1::TransactionOptions::IsolationLevel::REPEATABLE_READ
+      end
+    end
+
     # Forces a BeginTransaction RPC for a read/write transaction. This is used by a
     # connection if the first statement of a transaction failed.
     def force_begin_read_write
diff --git a/test/activerecord_spanner_interleaved_table/interleaved_tables_test.rb b/test/activerecord_spanner_interleaved_table/interleaved_tables_test.rb
@@ -54,6 +54,11 @@ def setup
 
     def teardown
       ActiveRecord::Base.connection_pool.disconnect!
+      ActiveRecordSpannerAdapter::Connection.spanners_map.each do |_, spanner|
+        channel = spanner.service.service.spanner_stub.grpc_stub.instance_variable_get(:@ch)
+        channel.close
+      end
+      ActiveRecordSpannerAdapter::Connection.spanners_map.clear
       @server.stop
       @server_thread.exit
       super
diff --git a/test/activerecord_spanner_interleaved_table_version_7_1_and_higher/interleaved_tables_test.rb b/test/activerecord_spanner_interleaved_table_version_7_1_and_higher/interleaved_tables_test.rb
@@ -58,6 +58,11 @@ def setup
 
     def teardown
       ActiveRecord::Base.connection_pool.disconnect!
+      ActiveRecordSpannerAdapter::Connection.spanners_map.each do |_, spanner|
+        channel = spanner.service.service.spanner_stub.grpc_stub.instance_variable_get(:@ch)
+        channel.close
+      end
+      ActiveRecordSpannerAdapter::Connection.spanners_map.clear
       @server.stop
       @server_thread.exit
       super
diff --git a/test/activerecord_spanner_mock_server/base_spanner_mock_server_test.rb b/test/activerecord_spanner_mock_server/base_spanner_mock_server_test.rb
@@ -86,6 +86,13 @@ def setup
 
     def teardown
       ActiveRecord::Base.connection_pool.disconnect!
+      ActiveRecordSpannerAdapter::Connection.spanners_map.each do |_, spanner|
+        if spanner.service
+          channel = spanner.service.service.spanner_stub.grpc_stub.instance_variable_get(:@ch)
+          channel.close
+        end
+      end
+      ActiveRecordSpannerAdapter::Connection.spanners_map.clear
       @server.stop
       @server_thread.exit
       super
diff --git a/test/activerecord_spanner_mock_server/inline_begin_transaction_test.rb b/test/activerecord_spanner_mock_server/inline_begin_transaction_test.rb
@@ -13,43 +13,56 @@ class InlineTransactionTest < BaseSpannerMockServerTest
     def test_read_write_transaction_uses_inlined_begin
       insert_sql = register_insert_singer_result
 
-      ActiveRecord::Base.transaction do
-        Singer.create(first_name: "Dave", last_name: "Allison")
-      end
+      [:serializable, :repeatable_read, nil].each { |isolation|
+        refute ActiveRecord::Base.connection.isolation_level,
+               "Connection should not have a default isolation level"
+        ActiveRecord::Base.transaction isolation: isolation do
+          Singer.create(first_name: "Dave", last_name: "Allison")
+        end
 
-      # There should be no explicit BeginTransaction request.
-      begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
-      assert_empty begin_transaction_requests
-      sql_request = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }.first
-      assert sql_request.transaction&.begin&.read_write
+        # There should be no explicit BeginTransaction request.
+        begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
+        assert_empty begin_transaction_requests
+        sql_request = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }.first
+        assert sql_request.transaction&.begin&.read_write
+        assert_equal _transaction_isolation_level_to_grpc(isolation),
+                     sql_request.transaction&.begin&.isolation_level
 
-      commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
-      assert_equal 1, commit_requests.length
+        commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
+        assert_equal 1, commit_requests.length
+
+        @mock.requests.clear
+      }
     end
 
     def test_read_write_transaction_aborted_dml_is_automatically_retried_with_inline_begin
       insert_sql = register_insert_singer_result
 
-      already_aborted = false
-      ActiveRecord::Base.transaction do
-        already_aborted = @mock.abort_next_transaction unless already_aborted
-        # The following statement will fail with an Aborted error. That will cause the entire
-        # transaction block to be retried.
-        Singer.create(first_name: "Dave", last_name: "Allison")
-      end
-
-      # There should be two transaction attempts, two ExecuteSqlRequests and only one commit.
-      # Both transaction attempts should use an inlined BeginTransaction.
-      begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
-      assert_empty begin_transaction_requests
-      sql_requests = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }
-      assert_equal 2, sql_requests.length
-      sql_requests.each { |req| assert req.transaction&.begin&.read_write }
+      [:serializable, :repeatable_read, nil].each { |isolation|
+        already_aborted = false
+        ActiveRecord::Base.transaction isolation: isolation do
+          already_aborted = @mock.abort_next_transaction unless already_aborted
+          # The following statement will fail with an Aborted error. That will cause the entire
+          # transaction block to be retried.
+          Singer.create(first_name: "Dave", last_name: "Allison")
+        end
 
-      commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
-      assert_equal 1, commit_requests.length
-      rollback_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::RollbackRequest }
-      assert_empty rollback_requests
+        # There should be two transaction attempts, two ExecuteSqlRequests and only one commit.
+        # Both transaction attempts should use an inlined BeginTransaction.
+        begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
+        assert_empty begin_transaction_requests
+        sql_requests = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }
+        assert_equal 2, sql_requests.length
+        sql_requests.each { |req| assert req.transaction&.begin&.read_write }
+        sql_requests.each { |req| assert_equal _transaction_isolation_level_to_grpc(isolation), req.transaction&.begin&.isolation_level }
+
+        commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
+        assert_equal 1, commit_requests.length
+        rollback_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::RollbackRequest }
+        assert_empty rollback_requests
+
+        @mock.requests.clear
+      }
     end
 
     def test_read_write_transaction_aborted_query_is_automatically_retried_with_inline_begin
@@ -270,6 +283,84 @@ def test_error_on_first_statement_and_on_begin_transaction
       assert_empty commit_requests
     end
 
+    def test_read_write_transaction_uses_default_isolation_level_from_config
+      insert_sql = register_insert_singer_result
+
+      # Close the existing connection pool and create a new one with a different
+      # default isolation level.
+      ActiveRecord::Base.connection_pool.disconnect!
+      ActiveRecord::Base.establish_connection(
+        adapter: "spanner",
+        emulator_host: "localhost:#{@port}",
+        project: "test-project",
+        instance: "test-instance",
+        database: "testdb",
+        default_sequence_kind: "BIT_REVERSED_POSITIVE",
+        isolation_level: :repeatable_read,
+        )
+
+      # Verify the default isolation level.
+      assert_equal :repeatable_read, ActiveRecord::Base.connection.isolation_level
+
+      ActiveRecord::Base.transaction do
+        Singer.create(first_name: "Dave", last_name: "Allison")
+      end
+
+      sql_request = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }.first
+      assert sql_request.transaction&.begin&.read_write
+      assert_equal :REPEATABLE_READ,
+                   sql_request.transaction&.begin&.isolation_level
+    end
+
+    def test_read_write_transaction_uses_default_isolation_level
+      insert_sql = register_insert_singer_result
+
+      [:serializable, :repeatable_read, nil].each { |isolation|
+        ActiveRecord::Base.connection.isolation_level = isolation
+
+        ActiveRecord::Base.transaction do
+          Singer.create(first_name: "Dave", last_name: "Allison")
+        end
+
+        # There should be no explicit BeginTransaction request.
+        begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
+        assert_empty begin_transaction_requests
+        sql_request = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }.first
+        assert sql_request.transaction&.begin&.read_write
+        assert_equal _transaction_isolation_level_to_grpc(isolation),
+                     sql_request.transaction&.begin&.isolation_level
+
+        commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
+        assert_equal 1, commit_requests.length
+
+        @mock.requests.clear
+      }
+    end
+
+    def test_read_write_transaction_overrides_default_isolation_level
+      insert_sql = register_insert_singer_result
+
+      # Set the default isolation level to :repeatable_read
+      ActiveRecord::Base.connection.isolation_level = :repeatable_read
+
+      # Start a transaction with isolation level :serializable.
+      # This should override the default isolation level.
+      ActiveRecord::Base.transaction isolation: :serializable do
+        Singer.create(first_name: "Dave", last_name: "Allison")
+      end
+
+      # There should be no explicit BeginTransaction request.
+      begin_transaction_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::BeginTransactionRequest }
+      assert_empty begin_transaction_requests
+      sql_request = @mock.requests.select { |req| req.is_a?(Google::Cloud::Spanner::V1::ExecuteSqlRequest) && req.sql == insert_sql }.first
+      assert sql_request.transaction&.begin&.read_write
+      assert_equal :SERIALIZABLE,
+                   sql_request.transaction&.begin&.isolation_level
+
+      commit_requests = @mock.requests.select { |req| req.is_a? Google::Cloud::Spanner::V1::CommitRequest }
+      assert_equal 1, commit_requests.length
+    end
+
     def register_insert_singer_result
       sql = "INSERT INTO `singers` (`first_name`, `last_name`, `id`) VALUES (@p1, @p2, @p3)"
       @mock.put_statement_result sql, StatementResult.new(1)
@@ -281,5 +372,16 @@ def register_singer_find_by_id_result
       @mock.put_statement_result sql, MockServerTests::create_random_singers_result(1)
       sql
     end
+
+    def _transaction_isolation_level_to_grpc isolation
+      case isolation
+      when :serializable
+        :SERIALIZABLE
+      when :repeatable_read
+        :REPEATABLE_READ
+      else
+        :ISOLATION_LEVEL_UNSPECIFIED
+      end
+    end
   end
 end
diff --git a/test/migrations_with_mock_server/migrations_with_mock_server_test.rb b/test/migrations_with_mock_server/migrations_with_mock_server_test.rb
@@ -70,6 +70,11 @@ def setup
 
     def teardown
       ActiveRecord::Base.connection_pool.disconnect!
+      ActiveRecordSpannerAdapter::Connection.spanners_map.each do |_, spanner|
+        channel = spanner.service.service.spanner_stub.grpc_stub.instance_variable_get(:@ch)
+        channel.close
+      end
+      ActiveRecordSpannerAdapter::Connection.spanners_map.clear
       @server.stop
       @server_thread.exit
       super
