fix: coroutine leaks from connection pool

Author: stainless-app[bot]

lib/scrapegraphai/internal/transport/pooled_net_requester.rb

@@ -134,9 +134,9 @@ def execute(request)
 
           # rubocop:disable Metrics/BlockLength
           enum = Enumerator.new do |y|
-            with_pool(url, deadline: deadline) do |conn|
-              next if finished
+            next if finished
 
+            with_pool(url, deadline: deadline) do |conn|
               req, closing = self.class.build_request(request) do
                 self.class.calibrate_socket_timeout(conn, deadline)
               end
@@ -149,7 +149,7 @@ def execute(request)
 
               self.class.calibrate_socket_timeout(conn, deadline)
               conn.request(req) do |rsp|
-                y << [conn, req, rsp]
+                y << [req, rsp]
                 break if finished
 
                 rsp.read_body do |bytes|
@@ -160,6 +160,8 @@ def execute(request)
                 end
                 eof = true
               end
+            ensure
+              conn.finish if !eof && conn&.started?
             end
           rescue Timeout::Error
             raise Scrapegraphai::Errors::APITimeoutError.new(url: url, request: req)
@@ -168,16 +170,11 @@ def execute(request)
           end
           # rubocop:enable Metrics/BlockLength
 
-          conn, _, response = enum.next
+          _, response = enum.next
           body = Scrapegraphai::Internal::Util.fused_enum(enum, external: true) do
             finished = true
-            tap do
-              enum.next
-            rescue StopIteration
-              nil
-            end
+            loop { enum.next }
           ensure
-            conn.finish if !eof && conn&.started?
             closing&.call
           end
           [Integer(response.code), response, body]

test/scrapegraphai/internal/util_test.rb

@@ -310,6 +310,31 @@ def test_copy_write
 end
 
 class Scrapegraphai::Test::UtilFusedEnumTest < Minitest::Test
+  def test_rewind_closing
+    touched = false
+    once = 0
+    steps = 0
+    enum = Enumerator.new do |y|
+      next if touched
+
+      10.times do
+        steps = _1
+        y << _1
+      end
+    ensure
+      once = once.succ
+    end
+
+    fused = Scrapegraphai::Internal::Util.fused_enum(enum, external: true) do
+      touched = true
+      loop { enum.next }
+    end
+    Scrapegraphai::Internal::Util.close_fused!(fused)
+
+    assert_equal(1, once)
+    assert_equal(0, steps)
+  end
+
   def test_closing
     arr = [1, 2, 3]
     once = 0