RUBY-3770 Implement makeTimeoutError semantics in withTransaction

lib/mongo/session.rb

@@ -475,7 +475,7 @@ def with_transaction(options = nil)
           if overload_encountered
             delay = @client.retry_policy.backoff_delay(overload_error_count)
             if backoff_would_exceed_deadline?(deadline, delay)
-              raise Mongo::Error::TimeoutError, 'CSOT timeout expired waiting to retry withTransaction'
+              make_timeout_error_from(last_error, 'CSOT timeout expired waiting to retry withTransaction')
             end
             unless @client.retry_policy.should_retry_overload?(overload_error_count, delay)
               raise(last_error)
@@ -484,7 +484,7 @@ def with_transaction(options = nil)
           else
             backoff = backoff_seconds_for_retry(transaction_attempt)
             if backoff_would_exceed_deadline?(deadline, backoff)
-              raise Mongo::Error::TimeoutError, 'CSOT timeout expired waiting to retry withTransaction'
+              make_timeout_error_from(last_error, 'CSOT timeout expired waiting to retry withTransaction')
             end
             sleep(backoff)
           end
@@ -513,7 +513,7 @@ def with_transaction(options = nil)
 
           if deadline_expired?(deadline)
             transaction_in_progress = false
-            raise
+            make_timeout_error_from(e, 'CSOT timeout expired during withTransaction callback')
           end
 
           if e.is_a?(Mongo::Error) && e.label?('TransientTransactionError')
@@ -554,7 +554,11 @@ def with_transaction(options = nil)
                 e.is_a?(Error::OperationFailure::Family) && e.max_time_ms_expired?
               then
                 transaction_in_progress = false
-                raise
+                if @with_transaction_timeout_ms && deadline_expired?(deadline)
+                  make_timeout_error_from(e, 'CSOT timeout expired during withTransaction commit')
+                else
+                  raise
+                end
               end
 
               if e.label?('SystemOverloadedError')
@@ -569,7 +573,7 @@ def with_transaction(options = nil)
                 delay = @client.retry_policy.backoff_delay(overload_error_count)
                 if backoff_would_exceed_deadline?(deadline, delay)
                   transaction_in_progress = false
-                  raise
+                  make_timeout_error_from(e, 'CSOT timeout expired during withTransaction commit')
                 end
                 unless @client.retry_policy.should_retry_overload?(overload_error_count, delay)
                   transaction_in_progress = false
@@ -591,7 +595,7 @@ def with_transaction(options = nil)
             elsif e.label?('TransientTransactionError')
               if Utils.monotonic_time >= deadline
                 transaction_in_progress = false
-                raise
+                make_timeout_error_from(e, 'CSOT timeout expired during withTransaction commit')
               end
               last_error = e
               if e.label?('SystemOverloadedError')
@@ -1436,5 +1440,17 @@ def backoff_would_exceed_deadline?(deadline, backoff_seconds)
 
       Utils.monotonic_time + backoff_seconds >= deadline
     end
+
+    # Implements makeTimeoutError(lastError) from the transactions-convenient-api spec.
+    # In CSOT mode raises TimeoutError with last_error's message included as a substring.
+    # In non-CSOT mode re-raises last_error directly.
+    def make_timeout_error_from(last_error, timeout_message)
+      if @with_transaction_timeout_ms
+        raise Mongo::Error::TimeoutError, "#{timeout_message}: #{last_error}"
+      else
+        raise last_error
+      end
+    end
+
   end
 end

spec/integration/sdam_prose_spec.rb

@@ -64,4 +64,120 @@
         configureFailPoint: 'failCommand', mode: 'off')
     end
   end
+
+  describe 'Connection Pool Backpressure' do
+    min_server_fcv '8.2'
+    require_topology :single
+
+    let(:subscriber) { Mrss::EventSubscriber.new }
+
+    let(:client) do
+      new_local_client(
+        SpecConfig.instance.addresses,
+        SpecConfig.instance.all_test_options.merge(
+          max_connecting: 100,
+          max_pool_size: 100,
+        ),
+      ).tap do |client|
+        client.subscribe(Mongo::Monitoring::CONNECTION_POOL, subscriber)
+      end
+    end
+
+    after do
+      sleep 1
+      admin_db = root_authorized_client.use('admin').database
+
+      if defined?(@prev_ingressConnectionEstablishmentRateLimiterEnabled) &&
+         defined?(@prev_ingressConnectionEstablishmentRatePerSec) &&
+         defined?(@prev_ingressConnectionEstablishmentBurstCapacitySecs) &&
+         defined?(@prev_ingressConnectionEstablishmentMaxQueueDepth)
+        admin_db.command(
+          setParameter: 1,
+          ingressConnectionEstablishmentRateLimiterEnabled: @prev_ingressConnectionEstablishmentRateLimiterEnabled,
+        )
+        admin_db.command(
+          setParameter: 1,
+          ingressConnectionEstablishmentRatePerSec: @prev_ingressConnectionEstablishmentRatePerSec,
+        )
+        admin_db.command(
+          setParameter: 1,
+          ingressConnectionEstablishmentBurstCapacitySecs: @prev_ingressConnectionEstablishmentBurstCapacitySecs,
+        )
+        admin_db.command(
+          setParameter: 1,
+          ingressConnectionEstablishmentMaxQueueDepth: @prev_ingressConnectionEstablishmentMaxQueueDepth,
+        )
+      else
+        # Fallback: at least disable the limiter if previous values were not captured.
+        admin_db.command(
+          setParameter: 1,
+          ingressConnectionEstablishmentRateLimiterEnabled: false,
+        )
+      end
+    end
+
+    it 'generates checkout failures when the ingress connection rate limiter is active' do
+      admin_db = root_authorized_client.use('admin').database
+
+      # Capture current ingress connection establishment parameters so they can be restored.
+      current_params = admin_db.command(
+        getParameter: 1,
+        ingressConnectionEstablishmentRateLimiterEnabled: 1,
+        ingressConnectionEstablishmentRatePerSec: 1,
+        ingressConnectionEstablishmentBurstCapacitySecs: 1,
+        ingressConnectionEstablishmentMaxQueueDepth: 1,
+      ).first
+
+      @prev_ingressConnectionEstablishmentRateLimiterEnabled =
+        current_params['ingressConnectionEstablishmentRateLimiterEnabled']
+      @prev_ingressConnectionEstablishmentRatePerSec =
+        current_params['ingressConnectionEstablishmentRatePerSec']
+      @prev_ingressConnectionEstablishmentBurstCapacitySecs =
+        current_params['ingressConnectionEstablishmentBurstCapacitySecs']
+      @prev_ingressConnectionEstablishmentMaxQueueDepth =
+        current_params['ingressConnectionEstablishmentMaxQueueDepth']
+
+      # Enable the ingress rate limiter with test-specific values.
+      admin_db.command(
+        setParameter: 1,
+        ingressConnectionEstablishmentRateLimiterEnabled: true,
+      )
+      admin_db.command(
+        setParameter: 1,
+        ingressConnectionEstablishmentRatePerSec: 20,
+      )
+      admin_db.command(
+        setParameter: 1,
+        ingressConnectionEstablishmentBurstCapacitySecs: 1,
+      )
+      admin_db.command(
+        setParameter: 1,
+        ingressConnectionEstablishmentMaxQueueDepth: 1,
+      )
+
+      # Add a document so $where has something to process.
+      client.use('test')['test'].delete_many
+      client.use('test')['test'].insert_one({})
+
+      # Run 100 parallel find_one operations that contend for connections.
+      threads = 100.times.map do
+        Thread.new do
+          begin
+            client.use('test')['test'].find(
+              '$where' => 'function() { sleep(2000); return true; }'
+            ).first
+          rescue StandardError
+            # Ignore connection errors (including checkout timeouts).
+          end
+        end
+      end
+      threads.each(&:join)
+
+      checkout_failed = subscriber.select_published_events(
+        Mongo::Monitoring::Event::Cmap::ConnectionCheckOutFailed
+      )
+
+      expect(checkout_failed.length).to be >= 10
+    end
+  end
 end

spec/integration/secondary_reads_spec.rb

@@ -28,7 +28,7 @@
 
           end_stats = get_read_counters
 
-          end_stats[:secondary].should be_within(10).of(start_stats[:secondary])
+          end_stats[:secondary].should be_within(50).of(start_stats[:secondary])
           end_stats[:primary].should >= start_stats[:primary] + 30
         end
       end
@@ -50,7 +50,7 @@
 
           end_stats = get_read_counters
 
-          end_stats[:primary].should be_within(10).of(start_stats[:primary])
+          end_stats[:primary].should be_within(50).of(start_stats[:primary])
           end_stats[:secondary].should >= start_stats[:secondary] + 30
         end
       end

spec/mongo/retryable/token_bucket_spec.rb

@@ -67,7 +67,16 @@
   end
 
   describe 'thread safety' do
-    let(:bucket) { described_class.new(capacity: 1000) }
+    # Use capacity 2000, start at 1000 tokens.
+    # With 500 consumes and 500 deposits, floor/ceiling cannot be hit:
+    #   min possible = 1000 - 500 = 500 > 0 (all consumes succeed)
+    #   max possible = 1000 + 500 = 1500 < 2000 (all deposits effective)
+    # So the net change is guaranteed to be 0, making the assertion reliable.
+    let(:bucket) do
+      b = described_class.new(capacity: 2000)
+      b.consume(1000)
+      b
+    end
 
     def run_concurrent_operations(bucket)
       threads = []