fix: wake a parked async_http body reader instead of closing under it

gems/dexpace-transport-async_http/lib/dexpace/transport/async_http/exchange.rb

@@ -18,15 +18,17 @@ module AsyncHTTP
       # the Completer whose own hook pushes the reason, and a transient WATCHER task on the
       # caller's task pops the queue -- a scheduler-aware wait from inside the reactor, wakeable
       # from any thread -- and acts ON THE REACTOR'S THREAD: it cancels the exchange task while the
-      # exchange is in flight, and closes the delivered response afterwards, which is what wakes a
-      # consumer blocked in a body read. Closing the native body from the canceller's thread
-      # instead corrupts the reactor's selector (measured: `IOError: stream closed in another
-      # thread` out of the reactor itself). The queue is closed when the exchange ends undelivered
-      # or when the delivered body is released, so the watcher wakes with nil and exits;
-      # `transient: true` keeps a body a caller never closes from holding the caller's `Sync`
-      # block open. A hook can still run after that close -- the source and the completer both
-      # steal their hooks before they notify -- so the push is total over it (#signal) and a cancel
-      # that lost the race against the exchange's own end never raises back into the canceller.
+      # exchange is in flight, and afterwards wakes a consumer blocked in a body read by raising
+      # into that reader's fiber, which then closes the body itself (ResponseBody#cancel_read),
+      # or closes the delivered response when no reader is parked. Closing the native body from
+      # the canceller's thread instead corrupts the reactor's selector (measured: `IOError:
+      # stream closed in another thread` out of the reactor itself). The queue is closed when the
+      # exchange ends undelivered or when the delivered body is released, so the watcher wakes
+      # with nil and exits; `transient: true` keeps a body a caller never closes from holding the
+      # caller's `Sync` block open. A hook can still run after that close -- the source and the
+      # completer both steal their hooks before they notify -- so the push is total over it
+      # (#signal) and a cancel that lost the race against the exchange's own end never raises
+      # back into the canceller.
       #
       # `Async::Cancel` is not a StandardError and leaves this task through the
       # `rescue ::Exception` arm below, which re-raises it unchanged after settling the pivot
@@ -171,8 +173,8 @@ def finish
         # the token, from Future#cancel, or from a pre-dispatch cancellation -- and nil means the
         # exchange ended or the delivered body was released. While the exchange is in flight the
         # task is cancelled, which is what reaches a native call blocked in a read or in the
-        # pool's acquire (TRANSPORT-7); once the response is delivered the response itself is
-        # closed, which is what reaches a consumer blocked in a body read, and never a delivered
+        # pool's acquire (TRANSPORT-7); once the response is delivered a consumer blocked in a
+        # body read is woken and the body closed (#release_delivered), and never a delivered
         # response whose pivot was not cancelled (ASYNC-20). An exchange already in its exit path
         # is left to it: a cancel landing inside a close would leave the pivot to the ensure's net.
         #
@@ -189,12 +191,23 @@ def watch
           return if reason.nil?
 
           if @delivered
-            ::Dexpace.close_quietly(@adapted, logger: @logger)
+            release_delivered
           elsif !@finishing
             @task&.cancel(cause: ::Dexpace::CancelledError.new(reason))
           end
         end
 
+        # A delivered response's body is woken rather than closed under its reader
+        # (ResponseBody#cancel_read); a response delivered with no body has released the watch
+        # already, and anything else is closed as it always was.
+        def release_delivered
+          body = @adapted&.body
+          case body
+          when ResponseBody then body.cancel_read
+          else ::Dexpace.close_quietly(@adapted, logger: @logger)
+          end
+        end
+
         # Idempotent: the queue's close wakes the watcher with nil, and Subscription#detach is a
         # no-op the second time. Reached from the exchange's own exit when nothing was delivered,
         # and from the delivered body's release otherwise -- through the mapper at once when

gems/dexpace-transport-async_http/lib/dexpace/transport/async_http/response_body.rb

@@ -35,6 +35,10 @@ class ResponseBody
         include ::Dexpace::Body
         include ::Dexpace::Closeable
 
+        # The message of the IOError #cancel_read raises into a parked reader: the cause of the
+        # CancelledError that reader surfaces.
+        INTERRUPTED = "the response body was cancelled under a blocked read"
+
         attr_reader :media_type, :content_length
 
         # @param native [Protocol::HTTP::Body::Readable] the native body, owned from here on
@@ -53,6 +57,8 @@ def initialize(native:, media_type:, content_length:, cancellation:,
           @logger = logger
           @on_release = on_release
           @source = nil
+          @reader = nil
+          @reader_scheduler = nil
           initialize_closeable(owned: true)
           initialize_single_use
         end
@@ -91,6 +97,38 @@ def write_to(sink)
           written
         end
 
+        # The exchange watcher's entry for a cancellation that lands after delivery (TRANSPORT-7's
+        # body path), run on the reactor's thread. A consumer parked inside the native read is
+        # WOKEN, never closed under: the watcher raises an IOError into the reader's fiber through
+        # the scheduler -- a scheduler checkpoint, the fiber provably suspended inside #pull's own
+        # read, which is the mechanism Async::Task#cancel and #with_timeout use -- and the reader's
+        # own #pull closes this body on its own fiber once the native wait has unwound; with no
+        # reader parked the body is closed here. Closing the connection under a parked reader
+        # wakes it only by accident of the selector: io_uring's poll sees the close, but epoll
+        # silently drops a closed descriptor from its interest set, so on the EPoll selector
+        # (io-event's choice wherever liburing is absent -- every hosted CI runner) the reader
+        # stayed parked until something else fired, and on Ruby 4.0 IO#close's own interrupt of
+        # the parked fiber is deferred through Fiber::Scheduler#fiber_interrupt and lands in the
+        # caller's fiber at whatever suspension point it reaches next. A reader on another thread
+        # or another reactor is not this scheduler's to interrupt, and gets the close.
+        #
+        # @return [nil]
+        def cancel_read
+          reader = @reader
+          scheduler = ::Fiber.scheduler
+          if reader&.alive? && !scheduler.nil? && scheduler.equal?(@reader_scheduler)
+            scheduler.raise(reader, ::IOError, INTERRUPTED)
+          else
+            ::Dexpace.close_quietly(self, logger: @logger)
+          end
+          nil
+        rescue ::FiberError
+          # The reader was resumed between the check and the raise; there is nothing parked to
+          # wake, and the close is what a read after this one meets.
+          ::Dexpace.close_quietly(self, logger: @logger)
+          nil
+        end
+
         # The same handle every call (BODY-14), built with `.over`, never `.wrapping`, so no byte
         # is read ahead of demand and no one-byte-per-read defect is inherited (3a's residue).
         #
@@ -104,7 +142,7 @@ def source
         def pull
           raise ::Dexpace::ClosedError, "the response body is closed" if closed?
 
-          chunk = @native.read
+          chunk = parked { @native.read }
           if @cancellation.cancelled?
             close
             raise ::Dexpace::CancelledError, @cancellation.reason
@@ -117,6 +155,17 @@ def pull
           raise Errors.classify_read(error, cancellation: @cancellation)
         end
 
+        # Records the fiber blocked in the native read, and the scheduler it is blocked under, for
+        # #cancel_read; cleared on every exit, so a recorded reader is always one inside the read.
+        def parked
+          @reader_scheduler = ::Fiber.scheduler
+          @reader = ::Fiber.current
+          yield
+        ensure
+          @reader = nil
+          @reader_scheduler = nil
+        end
+
         # BODY-15: releases the native body -- which returns its connection to the pool, or resets
         # the stream when the body was not read to the end -- idempotent through Closeable's latch,
         # quietly (see the class comment), and then the exchange's own release hook.

gems/dexpace-transport-async_http/sig/dexpace/transport/async_http/exchange.rbs

@@ -41,6 +41,7 @@ module Dexpace
         def net: () -> nil
         def finish: () { () -> untyped } -> untyped
         def watch: () -> untyped
+        def release_delivered: () -> untyped
         def release_watch: () -> nil
       end
     end

gems/dexpace-transport-async_http/sig/dexpace/transport/async_http/response_body.rbs

@@ -18,13 +18,17 @@ module Dexpace
         include Dexpace::Body
         include Dexpace::Closeable
 
+        INTERRUPTED: String
+
         @native: untyped
         @media_type: Dexpace::MediaType?
         @content_length: Integer
         @cancellation: Dexpace::Cancellation
         @logger: Instrumentation::Logger
         @on_release: _Release?
         @source: Dexpace::IO::BufferedSource?
+        @reader: untyped
+        @reader_scheduler: untyped
 
         attr_reader media_type: Dexpace::MediaType?
         attr_reader content_length: Integer
@@ -36,10 +40,12 @@ module Dexpace
                 | () -> Enumerator[String, nil]
         def write_to: (Dexpace::IO::_Sink sink) -> Integer
         def source: () -> Dexpace::IO::BufferedSource
+        def cancel_read: () -> nil
 
         private
 
         def pull: () -> String?
+        def parked: [T] () { () -> T } -> T
         def release: () -> untyped
       end
     end

gems/dexpace-transport-async_http/test/dexpace/transport/async_http/cancellation_test.rb

@@ -234,52 +234,87 @@ class DexpaceTransportAsyncHTTPCancellationDeliveryTest < DexpaceTestCase
     end
 
     # A cancel that lands after delivery but through the TOKEN reaches a consumer blocked in a
-    # body read: the watcher closes the response from inside the reactor, the blocked read wakes,
-    # and the token is asked first, so the reader sees the cancellation and not a stream failure.
+    # body read: the watcher wakes the parked reader from inside the reactor, the reader closes
+    # the body, and the token is asked first, so the reader sees the cancellation and not a
+    # stream failure.
     test "TRANSPORT-7 on the body path: a token cancelled under a blocked body read wakes the " \
          "reader with CancelledError and releases the body" do
+      assert_body_read_woken
+    end
+
+    # The same, on every selector this host has. The watcher used to CLOSE the delivered
+    # response under the parked reader and rely on the close to wake it, which io_uring does and
+    # epoll does not: green on a developer machine, red on every hosted runner (whose io-event
+    # has no liburing), and on Ruby 4.0 IO#close's own deferred interrupt landed later in the
+    # fixture's Thread#join. Naming the selector keeps the property host-independent.
+    AsyncHTTPReactor::SELECTORS.each do |selector|
+      test "TRANSPORT-7 on the body path, on the #{selector} selector: the parked reader is " \
+           "woken by the watcher, never by the test's bound" do
+        assert_body_read_woken(selector: selector)
+      end
+    end
+
+    private
+
+    def assert_body_read_woken(selector: nil)
       server = AsyncHTTPHoldingServer.new(hold: :body)
       adapter = AsyncHTTP.build
       source = Dexpace::Cancellation.source
       arrived = ::Thread::Queue.new
-      canceller = ::Thread.new do
-        arrived.pop # the first chunk was yielded: the reader is now blocked on the held second one
-        source.cancel(:reader_cancelled)
-      end
+      canceller = cancel_on(arrived, source)
 
-      reactor_over(server) do |task|
-        response = value_within(adapter.call(request("http://127.0.0.1:#{server.port}/"), nil,
-                                             source.token,))
-        server.wait_for_accept
-        chunks = []
-        # Bounded: a watcher that never closed the delivered response would leave the read blocked
-        # for good. The bound's Async::TimeoutError is a StandardError the token-first classifier
-        # turns into the SAME CancelledError, so the cause is what tells the two wakes apart.
-        error = assert_raises(Dexpace::CancelledError) do
-          task.with_timeout(BOUND) do
-            response.body.each do |chunk|
-              chunks << chunk
-              arrived.push(true)
-            end
-          end
-        end
+      reactor_over(server, selector: selector) do |task|
+        response = held_response(adapter, server, source)
 
-        assert_equal(:reader_cancelled, error.reason)
-        # The wake must be the WATCHER's close and never this test's own bound: with the watcher's
-        # close of the delivered response deleted (the reviewer's mutation 37), the bound fires
-        # inside the native read five seconds later and the classifier still answers
-        # CancelledError(:reader_cancelled) with the body closed -- every assertion around this
-        # one passes. A close under a blocked read surfaces from the library as a bare IOError,
-        # which the classifier saw and Ruby attached; the bound would have attached its own.
-        refute_kind_of(::Async::TimeoutError, error.cause,
-                       "the reader was woken by the test's bound, not by the watcher's close",)
-        assert_equal(["first"], chunks)
-        assert_predicate(response.body, :closed?)
+        assert_woken_by_watcher(response, *read_until_cancelled(task, response, arrived))
       end
       canceller.join
     ensure
       adapter&.close
       server&.close
     end
+
+    def cancel_on(arrived, source)
+      ::Thread.new do
+        arrived.pop # the first chunk was yielded: the reader is now blocked on the held second one
+        source.cancel(:reader_cancelled)
+      end
+    end
+
+    # The delivered response whose body the holding server keeps open after the first chunk.
+    def held_response(adapter, server, source)
+      url = "http://127.0.0.1:#{server.port}/"
+      value_within(adapter.call(request(url), nil, source.token)).tap { server.wait_for_accept }
+    end
+
+    # Bounded: a watcher that never woke the parked reader would leave the read blocked for good.
+    # The bound's Async::TimeoutError is a StandardError the token-first classifier turns into
+    # the SAME CancelledError, so the cause is what tells the two wakes apart.
+    def read_until_cancelled(task, response, arrived)
+      chunks = []
+      error = assert_raises(Dexpace::CancelledError) do
+        task.with_timeout(BOUND) do
+          response.body.each do |chunk|
+            chunks << chunk
+            arrived.push(true)
+          end
+        end
+      end
+      [error, chunks]
+    end
+
+    def assert_woken_by_watcher(response, error, chunks)
+      assert_equal(:reader_cancelled, error.reason)
+      # The wake must be the WATCHER's and never this test's own bound: with the watcher's wake
+      # of the delivered response deleted (the reviewer's mutation 37), the bound fires inside
+      # the native read five seconds later and the classifier still answers
+      # CancelledError(:reader_cancelled) with the body closed -- every assertion around this one
+      # passes. The watcher's wake is an IOError raised into the parked reader.
+      refute_kind_of(::Async::TimeoutError, error.cause,
+                     "the reader was woken by the test's bound, not by the watcher",)
+      assert_kind_of(::IOError, error.cause)
+      assert_equal(["first"], chunks)
+      assert_predicate(response.body, :closed?)
+    end
   end
 end