From debe774b1a37bee11fa3df249be7376046c0d853 Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 14:54:30 -0700
Subject: [PATCH 1/8] Read large values without 8 KB chunking in
 TimeoutRecoveryReader

Values with at least CHUNK_SIZE bytes still owed are read straight from the socket, up to LARGE_READ_SIZE (64 KiB) per read and never past the value's end, instead of being copied through the 8 KB line buffer. A 2 MB get drops from ~259 to ~37 syscalls on the exp_async_timeout_recovery path.
---
 CHANGELOG.md                                  |  3 ++
 bin/benchmark                                 | 17 ++++++++
 lib/dalli/protocol/timeout_recovery_reader.rb | 38 ++++++++++++++++-
 test/protocol/test_timeout_recovery_reader.rb | 41 +++++++++++++++++++
 4 files changed, 98 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 603559f0..298a4c9d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ Unreleased
 - Add experimental per-request `exp_async_timeout_recovery` option: reads wait with explicit deadlines and poll the
   socket once more before timing out, so replies delayed only by a busy fiber scheduler are not discarded. (nherson)
 
+- `exp_async_timeout_recovery` reads values of 8 KB or more straight from the socket in reads of up to 64 KiB
+  instead of through the 8 KB line buffer, cutting a 2 MB get from ~259 to ~37 syscalls. (nherson)
+
 - Add per-request opaque-correlation controls and a client `opaque_correlation_request_only` mode
   for opt-in rollouts, with eager PRNG initialization and unchanged unflagged reads. (mrattle)
 
diff --git a/bin/benchmark b/bin/benchmark
index 8a9ba396..a9556071 100755
--- a/bin/benchmark
+++ b/bin/benchmark
@@ -7,11 +7,13 @@
 # run with:
 # bundle exec bin/benchmark
 # RUBY_YJIT_ENABLE=1 BENCH_TARGET=get bundle exec bin/benchmark
+# BENCH_TARGET=async_get bundle exec bin/benchmark
 require 'bundler/inline'
 require 'json'
 
 gemfile do
   source 'https://rubygems.org'
+  gem 'async'
   gem 'benchmark-ips'
   gem 'logger'
   gem 'opentelemetry-sdk'
@@ -184,6 +186,21 @@ if %w[all get].include?(bench_target)
   end
 end
 
+# Large-value gets under a fiber scheduler, through the default read path and the
+# exp_async_timeout_recovery reader. GC stays enabled: with the GC suite, the
+# discarded values exhaust memory within a few seconds at the default payload size.
+if %w[all async_get].include?(bench_target)
+  require 'async'
+  Sync do
+    Benchmark.ips do |x|
+      x.config(warmup: bench_warmup, time: bench_time)
+      x.report('get dalli async') { client.get('key') }
+      x.report('get dalli async timeout recovery') { client.get('key', exp_async_timeout_recovery: true) }
+      x.compare!
+    end
+  end
+end
+
 if %w[all get_multi].include?(bench_target)
   Benchmark.ips do |x|
     x.config(warmup: bench_warmup, time: bench_time, suite: suite)
diff --git a/lib/dalli/protocol/timeout_recovery_reader.rb b/lib/dalli/protocol/timeout_recovery_reader.rb
index e5d7b7d9..1601fc90 100644
--- a/lib/dalli/protocol/timeout_recovery_reader.rb
+++ b/lib/dalli/protocol/timeout_recovery_reader.rb
@@ -8,6 +8,9 @@ module Protocol
     ##
     class TimeoutRecoveryReader
       CHUNK_SIZE = 8192
+      # Most a single large-value read asks for. Bounds the per-connection scratch
+      # buffer at 64 KiB while cutting a 2 MB value from ~257 reads to ~35.
+      LARGE_READ_SIZE = 65_536
 
       def initialize(sock, timeout)
         @sock = sock
@@ -36,7 +39,13 @@ def read(count)
         data = String.new(capacity: count, encoding: Encoding::BINARY)
         data << @buffer.byteslice(@offset, available) if available.positive?
         @offset = @buffer.bytesize
-        append_from_socket(data, count - available)
+        remaining = count - available
+        # Large values bypass the 8 KB line buffer: each read asks the socket for at
+        # most what is still owed, so a value costs a few syscalls rather than one per
+        # CHUNK_SIZE bytes plus a copy out of @buffer, and never reads past its end.
+        return append_directly(data, remaining) if remaining >= CHUNK_SIZE
+
+        append_from_socket(data, remaining)
       end
 
       private
@@ -59,6 +68,33 @@ def append_from_socket(data, remaining)
         data
       end
 
+      def append_directly(data, remaining)
+        while remaining.positive?
+          chunk = if data.empty?
+                    read_partial(remaining, data)
+                  else
+                    read_partial([remaining, LARGE_READ_SIZE].min, scratch)
+                  end
+          return data.empty? ? nil : data unless chunk
+
+          data << chunk unless chunk.equal?(data)
+          remaining -= chunk.bytesize
+        end
+        data
+      end
+
+      def scratch
+        @scratch ||= String.new(encoding: Encoding::BINARY)
+      end
+
+      # Waits like read_chunk, but reads at most maxlen bytes into outbuf.
+      def read_partial(maxlen, outbuf)
+        while (result = @sock.read_nonblock(maxlen, outbuf, exception: false)).is_a?(Symbol)
+          wait_for(result)
+        end
+        result
+      end
+
       # read_nonblock overwrites the buffer, so unconsumed bytes are set aside and restored
       # after the read; @offset only moves once the read returns.
       def fill
diff --git a/test/protocol/test_timeout_recovery_reader.rb b/test/protocol/test_timeout_recovery_reader.rb
index 9e96d7e0..6bf88386 100644
--- a/test/protocol/test_timeout_recovery_reader.rb
+++ b/test/protocol/test_timeout_recovery_reader.rb
@@ -95,6 +95,47 @@ def write_in_background(peer, *chunks)
     end
   end
 
+  it 'reads a large value in a few socket reads and stops at its end' do
+    chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE
+    value = SecureRandom.random_bytes((chunk * 40) + 3)
+    with_reader do |reader, peer, sock|
+      reads = 0
+      counter = Module.new do
+        define_method(:read_nonblock) do |*args, **kwargs|
+          reads += 1
+          super(*args, **kwargs)
+        end
+      end
+      sock.singleton_class.prepend(counter)
+      # Default UNIX socket buffers (8 KB on macOS) would cap each read regardless of
+      # the reader, so make room for the whole response before writing it.
+      peer.setsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF, 1 << 20)
+      sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 1 << 20)
+      half = value.bytesize / 2
+      write_in_background(peer, "VA #{value.bytesize}\r\n", value.byteslice(0, half),
+                          value.byteslice(half..), "\r\nEN\r\n").join
+
+      assert_equal "VA #{value.bytesize}\r\n", reader.gets("\r\n")
+      assert_equal value, reader.read(value.bytesize)
+      assert_operator reads, :<, 20
+      assert_equal "\r\n", reader.read(2)
+      assert_equal "EN\r\n", reader.gets("\r\n")
+      assert_nil reader.gets("\r\n")
+    end
+  end
+
+  it 'returns the bytes received when EOF cuts a large value short' do
+    chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE
+    value = SecureRandom.random_bytes(chunk * 3)
+    with_reader do |reader, peer|
+      writer = write_in_background(peer, value)
+
+      assert_equal value, reader.read(chunk * 5)
+      assert_nil reader.read(chunk * 5)
+      writer.join
+    end
+  end
+
   it 'allocates one string per buffered line' do
     with_reader do |reader, peer|
       peer.write("HD\r\n" * 101)

From 2908c24cdf696f8fcfcde575e54fc503d137572e Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 15:45:03 -0700
Subject: [PATCH 2/8] Fix TimeoutRecoveryReader replaying old bytes at TLS EOF

At EOF a socket's read_nonblock empties the buffer it was given, but OpenSSL's leaves it untouched. fill then reset the read offset, so bytes already returned were returned again: after a TLS server closed the connection cleanly, the next request could receive the previous reply. fill now clears the buffer when the read hits EOF.

Also adds reader tests for the large-value path's timeout behaviour: recovery from a wait that reports a timeout mid-value, failing within one timeout once a value stops arriving, and a full timeout per wait while a value arrives in pauses.
---
 CHANGELOG.md                                  |  3 +
 lib/dalli/protocol/timeout_recovery_reader.rb |  8 +-
 test/protocol/test_timeout_recovery_reader.rb | 83 ++++++++++++++++++-
 test/utils/ssl_socket_pair.rb                 | 34 ++++++++
 4 files changed, 120 insertions(+), 8 deletions(-)
 create mode 100644 test/utils/ssl_socket_pair.rb

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 298a4c9d..d80bdc4a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,9 @@ Unreleased
 - `exp_async_timeout_recovery` reads values of 8 KB or more straight from the socket in reads of up to 64 KiB
   instead of through the 8 KB line buffer, cutting a 2 MB get from ~259 to ~37 syscalls. (nherson)
 
+- Fix `exp_async_timeout_recovery` replaying the previous reply instead of reporting EOF after a TLS server
+  closes the connection cleanly. (nherson)
+
 - Add per-request opaque-correlation controls and a client `opaque_correlation_request_only` mode
   for opt-in rollouts, with eager PRNG initialization and unchanged unflagged reads. (mrattle)
 
diff --git a/lib/dalli/protocol/timeout_recovery_reader.rb b/lib/dalli/protocol/timeout_recovery_reader.rb
index 1601fc90..88d597b6 100644
--- a/lib/dalli/protocol/timeout_recovery_reader.rb
+++ b/lib/dalli/protocol/timeout_recovery_reader.rb
@@ -96,14 +96,14 @@ def read_partial(maxlen, outbuf)
       end
 
       # read_nonblock overwrites the buffer, so unconsumed bytes are set aside and restored
-      # after the read; @offset only moves once the read returns.
+      # after the read; @offset only moves once the read returns. At EOF a socket empties the
+      # buffer but OpenSSL leaves it untouched, so it is cleared here to avoid replaying old bytes.
       def fill
         partial = @buffer.byteslice(@offset, @buffer.bytesize - @offset) if pending?
         filled = read_chunk
+        @buffer.clear unless filled
         @offset = 0
-        if partial
-          filled ? @buffer.prepend(partial) : @buffer.replace(partial)
-        end
+        @buffer.prepend(partial) if partial
         filled
       end
 
diff --git a/test/protocol/test_timeout_recovery_reader.rb b/test/protocol/test_timeout_recovery_reader.rb
index 6bf88386..8d48c0e9 100644
--- a/test/protocol/test_timeout_recovery_reader.rb
+++ b/test/protocol/test_timeout_recovery_reader.rb
@@ -2,6 +2,7 @@
 
 require_relative '../helper'
 require 'objspace'
+require_relative '../utils/ssl_socket_pair'
 
 describe Dalli::Protocol::TimeoutRecoveryReader do
   def with_reader(timeout: 0.05)
@@ -12,6 +13,18 @@ def with_reader(timeout: 0.05)
     peer&.close unless peer&.closed?
   end
 
+  def elapsed
+    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    yield
+    Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
+  end
+
+  # Default UNIX socket buffers (8 KB on macOS) cap each read regardless of the reader.
+  def enlarge_buffers(sock, peer)
+    peer.setsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF, 1 << 20)
+    sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 1 << 20)
+  end
+
   def write_in_background(peer, *chunks)
     Thread.new do
       chunks.each { |chunk| peer.write(chunk) }
@@ -107,10 +120,7 @@ def write_in_background(peer, *chunks)
         end
       end
       sock.singleton_class.prepend(counter)
-      # Default UNIX socket buffers (8 KB on macOS) would cap each read regardless of
-      # the reader, so make room for the whole response before writing it.
-      peer.setsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF, 1 << 20)
-      sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 1 << 20)
+      enlarge_buffers(sock, peer)
       half = value.bytesize / 2
       write_in_background(peer, "VA #{value.bytesize}\r\n", value.byteslice(0, half),
                           value.byteslice(half..), "\r\nEN\r\n").join
@@ -136,6 +146,71 @@ def write_in_background(peer, *chunks)
     end
   end
 
+  it 'keeps reading a large value when a wait in the middle of it reports a timeout while data is ready' do
+    chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE
+    value = SecureRandom.random_bytes(chunk * 8)
+    with_reader(timeout: 0.05) do |reader, peer, sock|
+      enlarge_buffers(sock, peer)
+      peer.write(value.byteslice(0, chunk * 3))
+      timed_out = []
+      # A scheduler that resumes the reader after the deadline reports a timeout even though the
+      # rest of the value has arrived by then.
+      sock.define_singleton_method(:wait_readable) do |timeout|
+        return super(timeout) if timeout.zero?
+
+        timed_out << timeout
+        peer.write(value.byteslice((chunk * 3)..)) if timed_out.one?
+        nil
+      end
+
+      assert_equal value, reader.read(value.bytesize)
+      assert_equal [0.05], timed_out
+    end
+  end
+
+  it 'fails within one timeout once a large value stops arriving' do
+    chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE
+    with_reader(timeout: 0.2) do |reader, peer, sock|
+      enlarge_buffers(sock, peer)
+      peer.write(SecureRandom.random_bytes(chunk * 3))
+      duration = elapsed { Timeout.timeout(5) { assert_raises(IO::TimeoutError) { reader.read(chunk * 8) } } }
+
+      # One wait for the timeout, then a poll that does not wait.
+      assert_operator duration, :>=, 0.2
+      assert_operator duration, :<, 0.3
+    end
+  end
+
+  it 'gives each wait a full timeout while a large value arrives in pauses' do
+    chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE
+    parts = Array.new(6) { SecureRandom.random_bytes(chunk * 2) }
+    with_reader(timeout: 0.1) do |reader, peer|
+      writer = Thread.new do
+        parts.each do |part|
+          peer.write(part)
+          sleep(0.03)
+        end
+      end
+      duration = elapsed { assert_equal parts.join, reader.read(parts.sum(&:bytesize)) }
+
+      assert_operator duration, :>, 0.1
+      writer.join
+    end
+  end
+
+  it 'returns nil at EOF instead of replaying earlier bytes when a TLS peer closes' do
+    sock, peer = SSLSocketPair.open
+    reader = Dalli::Protocol::TimeoutRecoveryReader.new(sock, 0.05)
+    peer.write("HD\r\n")
+    peer.close
+
+    assert_equal "HD\r\n", reader.gets("\r\n")
+    assert_nil reader.gets("\r\n")
+    assert_nil reader.read(1)
+  ensure
+    sock&.close
+  end
+
   it 'allocates one string per buffered line' do
     with_reader do |reader, peer|
       peer.write("HD\r\n" * 101)
diff --git a/test/utils/ssl_socket_pair.rb b/test/utils/ssl_socket_pair.rb
new file mode 100644
index 00000000..134e426c
--- /dev/null
+++ b/test/utils/ssl_socket_pair.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+require 'openssl'
+require 'socket'
+require_relative 'certificate_generator'
+
+##
+# A connected TLS client/server pair over loopback TCP, using the test certificates. The client is a
+# Dalli::Socket::SSLSocket, as Dalli would create.
+##
+module SSLSocketPair
+  def self.open
+    listener = TCPServer.new('127.0.0.1', 0)
+    client_tcp = TCPSocket.new('127.0.0.1', listener.local_address.ip_port)
+    server = OpenSSL::SSL::SSLSocket.new(listener.accept, server_context)
+    server.sync_close = true
+    accepting = Thread.new { server.accept }
+    client = Dalli::Socket::SSLSocket.new(client_tcp, CertificateGenerator.ssl_context)
+    client.hostname = 'localhost'
+    client.sync_close = true
+    client.connect
+    accepting.join
+    [client, server]
+  ensure
+    listener&.close
+  end
+
+  def self.server_context
+    context = OpenSSL::SSL::SSLContext.new
+    context.cert = OpenSSL::X509::Certificate.new(File.read(CertificateGenerator::MEMCACHED_CERT_PATH))
+    context.key = OpenSSL::PKey.read(File.read(CertificateGenerator::MEMCACHED_PK_PATH))
+    context
+  end
+end

From ef466cb39476ca89bf34164d3d6c474751bbce9d Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 15:45:04 -0700
Subject: [PATCH 3/8] Add randomized TimeoutRecoveryReader tests and
 large-value recovery integration tests

The fuzz test compares the reader with IO#gets/IO#read semantics over UNIX, TCP, and TLS sockets, with and without each Async selector, while injecting short reads, spurious waits, and waits that report a timeout although data is ready. It checks pending?, that large reads stop at the value's end, that every wait is bounded by the socket timeout, and that a stalled stream fails within one timeout of its last byte. FUZZ_SEED, FUZZ_ITERATIONS, and FUZZ_STATS control it.

The integration tests serve large values in parts with pauses from a separate process, covering recovery while another fiber holds the thread inside a value, pauses each shorter than the timeout, and failure within one timeout once a value stops arriving.
---
 .../test_async_timeout_recovery.rb            | 178 +++++-
 .../test_timeout_recovery_reader_fuzz.rb      | 556 ++++++++++++++++++
 2 files changed, 731 insertions(+), 3 deletions(-)
 create mode 100644 test/protocol/test_timeout_recovery_reader_fuzz.rb

diff --git a/test/integration/test_async_timeout_recovery.rb b/test/integration/test_async_timeout_recovery.rb
index 01ae7a0c..7618b73b 100644
--- a/test/integration/test_async_timeout_recovery.rb
+++ b/test/integration/test_async_timeout_recovery.rb
@@ -7,8 +7,22 @@
 
 # Replies from a separate process, so they arrive while a fiber in this process holds the thread.
 # A key named "delay<ms>-..." delays its reply; every get writes one byte to #signal on receipt.
+# A key containing "size<n>" gets an n-byte value; with "parts<n>" and "gap<ms>" the value is
+# written in n parts with a pause before each later part. Halfway through each pause, once the
+# client has drained the previous part and is waiting, a byte is written to #signal.
 class DelayedReplyServer
-  SCRIPT = <<~'RUBY'
+  VALUE_OF = <<~'RUBY'
+    def value_of(key)
+      size = key[/size(\d+)/, 1]&.to_i
+      return "value-of-#{key}" unless size
+
+      pattern = "EN\r\nVA 3 k\r\n\0\xFF".b
+      (pattern * ((size / pattern.bytesize) + 1)).byteslice(0, size)
+    end
+  RUBY
+  singleton_class.class_eval(VALUE_OF)
+
+  SCRIPT = VALUE_OF + <<~'RUBY'
     require 'socket'
     listener = UNIXServer.new(ARGV.fetch(0))
     signal = IO.for_fd(3, autoclose: false)
@@ -24,8 +38,19 @@ class DelayedReplyServer
           when 'mg'
             signal.write('.')
             sleep(key[/\Adelay(\d+)ms/, 1].to_i / 1000.0)
-            value = "value-of-#{key}"
-            connection.write("VA #{value.bytesize}#{" k#{key}" if flags.include?('k')}\r\n#{value}\r\n")
+            value = value_of(key)
+            parts = key[/parts(\d+)/, 1]&.to_i || 1
+            part_size = (value.bytesize + parts - 1) / parts
+            first, *rest = Array.new(parts) { |i| value.byteslice(i * part_size, part_size) }
+            connection.write("VA #{value.bytesize}#{" k#{key}" if flags.include?('k')}\r\n", first,
+                             rest.empty? ? "\r\n" : '')
+            pause = key[/gap(\d+)ms/, 1].to_i / 2000.0
+            rest.each_with_index do |part, index|
+              sleep(pause)
+              signal.write('.')
+              sleep(pause)
+              connection.write(part, index == rest.size - 1 ? "\r\n" : '')
+            end
           else raise "Unexpected command: #{command}"
           end
         end
@@ -144,6 +169,99 @@ def elapsed
   end
 end
 
+module LargeValueHelpers
+  # More than a UNIX socket buffer holds, so the server's write of a first half returns only once the
+  # client is reading inside the value.
+  LARGE_SIZE = 900_000
+
+  ##
+  # Records reads and waits on a connection's socket.
+  ##
+  class SocketLog
+    def self.now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+
+    def initialize(sock)
+      events = @events = []
+      sock.singleton_class.prepend(Module.new do
+        define_method(:read_nonblock) do |*args, **kwargs|
+          result = super(*args, **kwargs)
+          events << [:read, result.is_a?(String) ? result.bytesize : 0, SocketLog.now]
+          result
+        end
+
+        define_method(:wait_readable) do |timeout = nil|
+          result = super(timeout)
+          events << [:wait, timeout, result ? true : false]
+          result
+        end
+      end)
+    end
+
+    def last_data_at
+      @events.reverse_each.find { |kind, bytes| kind == :read && bytes.positive? }&.last
+    end
+
+    # Bytes received before each wait that reported a timeout and was followed by a poll that found data.
+    def recoveries(timeout)
+      received = 0
+      @events.each_cons(2).filter_map do |event, following|
+        received += event[1] if event.first == :read
+        received if event == [:wait, timeout, false] && following == [:wait, 0, true]
+      end
+    end
+  end
+
+  def in_scheduler(selector, &)
+    return yield unless selector
+
+    with_selector(selector, &)
+  end
+
+  def assert_reads_through_pauses(client, selector)
+    key = "size#{LARGE_SIZE}-parts6-gap30ms"
+    value = nil
+    duration = in_scheduler(selector) do
+      client.alive!
+      elapsed { value = client.get(key, recover) }
+    end
+
+    assert_equal DelayedReplyServer.value_of(key), value
+    # Five 30ms pauses: none outlasts the timeout, together they do.
+    assert_operator duration, :>, 2 * socket_timeout
+  end
+
+  def assert_fails_within_timeout_of_last_byte(client, selector)
+    error = hang = nil
+    in_scheduler(selector) do
+      client.alive!
+      log = SocketLog.new(connection_manager(client).sock)
+      client.get("size#{LARGE_SIZE}-parts2-gap500ms", recover)
+    rescue Dalli::NetworkError => e
+      error = e
+      hang = SocketLog.now - log.last_data_at
+    end
+
+    assert_kind_of Dalli::NetworkError, error
+    assert_match(/IO::TimeoutError/, error.message)
+    assert_operator hang, :>=, socket_timeout
+    assert_operator hang, :<, socket_timeout + 0.1
+    key = "size#{LARGE_SIZE}-after"
+
+    assert_equal DelayedReplyServer.value_of(key), in_scheduler(selector) { client.get(key, recover) }
+  end
+
+  # A fiber that wakes on the server's second signal, sent while the reader waits inside the value
+  # for its second half, and holds the thread until after that half arrives.
+  def start_busy_fiber_inside_value(task, signal)
+    task.async do
+      2.times { await_signal(signal) }
+      busy
+    end
+  end
+end
+
 module RecoveryParityHelpers
   MULTI_KEYS = (Array.new(40) { |i| "multi-#{i}" } << 'missing').freeze
 
@@ -188,6 +306,7 @@ def assert_same_results(client)
 describe 'exp_async_timeout_recovery' do
   include AsyncTimeoutRecoveryHelpers
   include RecoveryParityHelpers
+  include LargeValueHelpers
 
   describe 'without a fiber scheduler' do
     it 'returns the same results as default reads over TCP' do
@@ -247,6 +366,14 @@ def assert_same_results(client)
         assert_equal 'value-of-b', client.get('b', recover)
       end
     end
+
+    it 'reads a large value delivered in pauses that are each shorter than the timeout' do
+      with_delayed_reply_client { |client| assert_reads_through_pauses(client, nil) }
+    end
+
+    it 'fails a large value within one timeout once it stops arriving' do
+      with_delayed_reply_client { |client| assert_fails_within_timeout_of_last_byte(client, nil) }
+    end
   end
 
   AsyncTimeoutRecoveryHelpers::SELECTORS.each do |selector|
@@ -297,6 +424,51 @@ def assert_same_results(client)
         end
       end
 
+      it 'keeps reading a large value whose rest arrives while another fiber holds the thread' do
+        with_delayed_reply_client do |client, server|
+          key = "size#{LargeValueHelpers::LARGE_SIZE}-parts2-gap40ms"
+          value = sock = log = nil
+          duration = with_selector(selector) do |task|
+            client.alive!
+            sock = connection_manager(client).sock
+            log = LargeValueHelpers::SocketLog.new(sock)
+            start_busy_fiber_inside_value(task, server.signal)
+            elapsed { value = client.get(key, recover) }
+          end
+          recoveries = log.recoveries(socket_timeout)
+
+          assert_equal DelayedReplyServer.value_of(key), value
+          assert_operator duration, :>, socket_timeout
+          assert_same sock, connection_manager(client).sock
+          assert recoveries.any? { |received| received.between?(100, LargeValueHelpers::LARGE_SIZE) },
+                 "expected a wait inside the value to time out and find data; recovered after #{recoveries} bytes"
+        end
+      end
+
+      it 'fails a large value without the option when another fiber holds the thread inside it' do
+        with_delayed_reply_client do |client, server|
+          error = with_selector(selector) do |task|
+            client.alive!
+            start_busy_fiber_inside_value(task, server.signal)
+            client.get("size#{LargeValueHelpers::LARGE_SIZE}-parts2-gap40ms")
+            nil
+          rescue Dalli::NetworkError => e
+            e
+          end
+
+          assert_kind_of Dalli::NetworkError, error
+          assert_match(/IO::TimeoutError/, error.message)
+        end
+      end
+
+      it 'reads a large value delivered in pauses that are each shorter than the timeout' do
+        with_delayed_reply_client { |client| assert_reads_through_pauses(client, selector) }
+      end
+
+      it 'fails a large value within one timeout once it stops arriving' do
+        with_delayed_reply_client { |client| assert_fails_within_timeout_of_last_byte(client, selector) }
+      end
+
       it 'still fails a read when no reply arrives before the timeout' do
         with_delayed_reply_client do |client|
           error = nil
diff --git a/test/protocol/test_timeout_recovery_reader_fuzz.rb b/test/protocol/test_timeout_recovery_reader_fuzz.rb
new file mode 100644
index 00000000..433e665e
--- /dev/null
+++ b/test/protocol/test_timeout_recovery_reader_fuzz.rb
@@ -0,0 +1,556 @@
+# frozen_string_literal: true
+
+require_relative '../helper'
+require 'objspace'
+require 'zlib'
+require_relative '../utils/ssl_socket_pair'
+require 'async' if Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('3.3')
+
+# Randomized checks of TimeoutRecoveryReader. Each case streams generated bytes from a thread in
+# random pieces while the reader's socket injects short reads, spurious waits, and waits that report
+# a timeout although data is ready (what a fiber scheduler does when it resumes the reader late).
+# Results are compared with IO#gets/IO#read semantics over the same bytes.
+#
+# FUZZ_SEED=<n> replays a run, FUZZ_ITERATIONS=<n> scales it, and FUZZ_STATS=1 prints coverage.
+module TimeoutRecoveryReaderFuzz
+  Reader = Dalli::Protocol::TimeoutRecoveryReader
+  CHUNK = Reader::CHUNK_SIZE
+  LARGE = Reader::LARGE_READ_SIZE
+  SEED = Integer(ENV.fetch('FUZZ_SEED') { Random.new_seed.to_s })
+  ITERATIONS = Integer(ENV.fetch('FUZZ_ITERATIONS', '8'))
+  STATS = Hash.new(0)
+
+  TRANSPORTS = %i[unix tcp ssl].freeze
+  SELECTORS =
+    if defined?(IO::Event::Selector)
+      %i[URing EPoll KQueue Select].select { |name| IO::Event::Selector.const_defined?(name) }
+    else
+      []
+    end
+  SCHEDULERS = [nil, *SELECTORS].freeze
+
+  # :exact never stalls, :stall stops the stream for good, :pauses stops it for under a timeout at a time.
+  TIMEOUTS = { exact: 1.0, stall: 0.1, pauses: 0.2 }.freeze
+  PAUSES = 3
+  PAUSE_FRACTION = 0.4
+  # How long past the timeout a stalled read may take to fail.
+  TIMEOUT_SLACK = 0.1
+  CASE_LIMIT = 30
+
+  SIZES = [0, 1, 2, CHUNK - 1, CHUNK, CHUNK + 1, (2 * CHUNK) + 1, LARGE - 1, LARGE, LARGE + 1,
+           (3 * LARGE) + CHUNK - 1].freeze
+  PATTERNS = ["\r\n", "\r", "\n", "EN\r\n", "VA 12 k\r\n", "MN\r\n", "\0", "\xFF\xFE", 'é😀'].map(&:b).freeze
+  SEPARATORS = ["\r\n", "\n"].freeze
+
+  def self.now
+    Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  end
+
+  ##
+  # IO#gets / IO#read semantics over a fixed byte string.
+  ##
+  class Model
+    attr_reader :pos
+
+    def initialize(stream)
+      @stream = stream
+      @pos = 0
+    end
+
+    def remaining
+      @stream.bytesize - @pos
+    end
+
+    def rest
+      @stream.byteslice(@pos..)
+    end
+
+    # Stream offset an operation must see before it can return, or nil when it needs EOF.
+    def need(kind, arg)
+      if kind == :gets
+        index = @stream.byteindex(arg, @pos)
+        index && (index + arg.bytesize)
+      elsif arg.zero?
+        @pos
+      elsif arg <= remaining
+        @pos + arg
+      end
+    end
+
+    def apply(kind, arg)
+      if kind == :gets
+        return nil if remaining.zero?
+
+        index = @stream.byteindex(arg, @pos)
+        take(index ? index + arg.bytesize - @pos : remaining)
+      else
+        return take(0) if arg.zero?
+        return nil if remaining.zero?
+
+        take([arg, remaining].min)
+      end
+    end
+
+    private
+
+    def take(length)
+      data = @stream.byteslice(@pos, length)
+      @pos += length
+      data
+    end
+  end
+
+  ##
+  # Fault injection and bookkeeping for one reader socket.
+  ##
+  class Faults
+    attr_reader :waits, :delivered, :last_data_at
+
+    def initialize(rng)
+      @rng = rng
+      @spurious = [0.0, 0.05, 0.2].sample(random: rng)
+      @writable = [0.0, 0.02].sample(random: rng)
+      @short = [0.0, 0.2, 0.5].sample(random: rng)
+      @starve = [0.0, 0.3, 1.0].sample(random: rng)
+      @waits = []
+      @delivered = 0
+      mark
+    end
+
+    def attach(sock)
+      sock.singleton_class.prepend(Hooks)
+      sock.instance_variable_set(:@fuzz_faults, self)
+    end
+
+    def mark
+      @last_data_at = TimeoutRecoveryReaderFuzz.now
+    end
+
+    # The length to request, or the Symbol a spurious EAGAIN would return.
+    def read_length(maxlen)
+      STATS[:large_reads] += 1 if maxlen > CHUNK
+      roll = @rng.rand
+      return :wait_readable if roll < @spurious
+      return :wait_writable if roll < @spurious + @writable
+
+      roll > 1 - @short ? @rng.rand(1..maxlen) : maxlen
+    end
+
+    def received(result)
+      return result unless result.is_a?(String)
+
+      @delivered += result.bytesize
+      mark
+      result
+    end
+
+    # Records the wait and decides whether to report a timeout although the socket is ready.
+    def starve?(sock, event, timeout)
+      @waits << [timeout, Fiber.blocking?, !Fiber.scheduler.nil?]
+      return false if timeout.nil? || timeout.zero? || @rng.rand >= @starve
+      return false unless Fiber.blocking { IO.instance_method(event).bind_call(sock.to_io, 0) }
+
+      STATS[:starved_waits] += 1
+      true
+    end
+  end
+
+  ##
+  # Prepended to the reader's socket.
+  ##
+  module Hooks
+    def read_nonblock(maxlen, buf = nil, exception: true)
+      length = @fuzz_faults.read_length(maxlen)
+      return length if length.is_a?(Symbol)
+
+      @fuzz_faults.received(super(length, buf, exception: exception))
+    end
+
+    def wait_readable(timeout = nil)
+      @fuzz_faults.starve?(self, :wait_readable, timeout) ? nil : super
+    end
+
+    def wait_writable(timeout = nil)
+      @fuzz_faults.starve?(self, :wait_writable, timeout) ? nil : super
+    end
+  end
+
+  ##
+  # Byte streams and the reads to run against them.
+  ##
+  module Scripts
+    module_function
+
+    def payload(rng)
+      size = case rng.rand(6)
+             when 0, 1 then SIZES.sample(random: rng)
+             when 2 then rng.rand(1..(4 * CHUNK))
+             when 3, 4 then rng.rand(1..(4 * LARGE))
+             else rng.rand(1..1_500_000)
+             end
+      return rng.bytes(size) if rng.rand(3).zero?
+
+      unit = PATTERNS.sample(rng.rand(1..3), random: rng).join.b
+      (unit * ((size / unit.bytesize) + 1)).byteslice(0, size)
+    end
+
+    # meta-protocol-shaped replies: header line, value, terminator, then EN and EOF.
+    def protocol_script(rng, with_large_value: false)
+      stream = String.new(encoding: Encoding::BINARY)
+      ops = protocol_values(rng, with_large_value).flat_map do |value|
+        stream << "VA #{value.bytesize} t#{rng.rand(1000)}\r\n" << value << "\r\n"
+        [[:gets, SEPARATORS.sample(random: rng)], [:read, value.bytesize], [:read, 2]]
+      end
+      stream << "EN\r\n"
+      [stream, ops + [[:gets, "\r\n"], [:gets, "\r\n"], [:read, 1]]]
+    end
+
+    def protocol_values(rng, with_large_value)
+      values = Array.new(rng.rand(1..4)) { payload(rng) }
+      return values unless with_large_value && values.none? { |value| value.bytesize >= 3 * CHUNK }
+
+      values << rng.bytes(rng.rand((3 * CHUNK)..(3 * LARGE)))
+    end
+
+    # Arbitrary bytes and operations, including reads of exactly the rest and past EOF.
+    def random_script(rng)
+      stream = Array.new(rng.rand(1..4)) { payload(rng) }.join.b
+      model = Model.new(stream)
+      ops = Array.new(rng.rand(4..24)) do
+        op = rng.rand(3).zero? ? [:gets, SEPARATORS.sample(random: rng)] : [:read, read_size(rng, model.remaining)]
+        model.apply(*op)
+        op
+      end
+      [stream, ops]
+    end
+
+    def read_size(rng, remaining)
+      case rng.rand(4)
+      when 0 then remaining
+      when 1 then remaining + rng.rand(1..CHUNK)
+      when 2 then SIZES.sample(random: rng)
+      else rng.rand(0..(3 * LARGE))
+      end
+    end
+  end
+
+  ##
+  # Writer steps: [:write, bytes], [:pause, seconds], or [:stall] until released.
+  ##
+  module Plans
+    module_function
+
+    # The steps, plus the offset where a :stall stream stops.
+    def build(mode, rng, stream, ops, index)
+      case mode
+      when :exact then [exact(rng, stream), nil]
+      when :pauses then [pauses(rng, stream), nil]
+      when :stall
+        stall_at = index.even? ? stall_inside_large_read(rng, stream, ops) : rng.rand(0..stream.bytesize)
+        [pieces(rng, stream.byteslice(0, stall_at)) + [[:stall]] + pieces(rng, stream.byteslice(stall_at..)), stall_at]
+      end
+    end
+
+    def exact(rng, stream)
+      pieces(rng, stream).flat_map { |step| rng.rand(8).zero? ? [[:pause, rng.rand * 0.002], step] : [step] }
+    end
+
+    # Pauses at distinct offsets, each shorter than the timeout.
+    def pauses(rng, stream)
+      cuts = Set.new
+      cuts << rng.rand(1...stream.bytesize) while cuts.size < PAUSES
+      pause = [:pause, TIMEOUTS[:pauses] * PAUSE_FRACTION]
+      segments = [0, *cuts.sort, stream.bytesize].each_cons(2)
+      segments.flat_map { |from, to| pieces(rng, stream.byteslice(from...to)) << pause }[0...-1]
+    end
+
+    # An offset at least two chunks into a value of three or more, so the reader is past its line
+    # buffer and reading the value straight from the socket when the stream stops.
+    def stall_inside_large_read(rng, stream, ops)
+      model = Model.new(stream)
+      reads = ops.filter_map do |kind, arg|
+        start = model.pos
+        model.apply(kind, arg)
+        [start, arg] if kind == :read && arg >= 3 * CHUNK
+      end
+      start, length = reads.sample(random: rng)
+      start + rng.rand((2 * CHUNK)...length)
+    end
+
+    def pieces(rng, bytes)
+      steps = []
+      offset = 0
+      while offset < bytes.bytesize
+        length = case rng.rand(5)
+                 when 0 then rng.rand(1..16)
+                 when 1 then [CHUNK - 1, CHUNK, CHUNK + 1, LARGE - 1, LARGE + 1].sample(random: rng)
+                 when 2 then rng.rand(1..(2 * LARGE))
+                 else rng.rand(1..400_000)
+                 end
+        steps << [:write, bytes.byteslice(offset, length)]
+        offset += length
+      end
+      steps
+    end
+  end
+
+  ##
+  # Socket pairs and the thread that writes to the peer.
+  ##
+  module Wire
+    module_function
+
+    def socket_pair(transport)
+      case transport
+      when :unix then UNIXSocket.pair
+      when :tcp
+        server = TCPServer.new('127.0.0.1', 0)
+        client = TCPSocket.new('127.0.0.1', server.local_address.ip_port)
+        [client, server.accept]
+      when :ssl then SSLSocketPair.open
+      end
+    ensure
+      server&.close
+    end
+
+    def start_writer(peer, steps, release)
+      Thread.new do
+        Thread.current.report_on_exception = false
+        steps.each do |step, arg|
+          case step
+          when :write then peer.write(arg)
+          when :pause then sleep(arg)
+          when :stall then release.pop
+          end
+        end
+      rescue IOError, SystemCallError, OpenSSL::SSL::SSLError
+        nil # the reader gave up and closed its end
+      ensure
+        close_quietly(peer)
+      end
+    end
+
+    def close_quietly(io)
+      io&.close
+    rescue StandardError
+      nil
+    end
+  end
+
+  ##
+  # One case: the reader under test, its faults, the model, and where the stream stalls.
+  ##
+  Case = Struct.new(:label, :reader, :faults, :model, :ops, :stall_at, :timeout) do
+    def returned
+      @returned ||= 0
+    end
+
+    def returned!(bytes)
+      @returned = returned + (bytes ? bytes.bytesize : 0)
+    end
+
+    def buffered
+      faults.delivered - returned
+    end
+
+    def stalls_before?(kind, arg)
+      return false unless stall_at
+
+      need = model.need(kind, arg)
+      need.nil? || need > stall_at
+    end
+  end
+
+  ##
+  # Assertions shared by every case.
+  ##
+  module Checks
+    def assert_same_bytes(expected, actual, context)
+      return assert_nil(actual, context) if expected.nil?
+
+      refute_nil actual, "#{context}: expected #{expected.bytesize} bytes, got nil"
+      assert_equal Encoding::BINARY, actual.encoding, context
+      return pass if expected == actual
+
+      shorter = [expected.bytesize, actual.bytesize].min
+      index = (0...shorter).find { |i| expected.getbyte(i) != actual.getbyte(i) } || shorter
+
+      flunk "#{context}: expected #{expected.bytesize} bytes, got #{actual.bytesize}; first difference at byte #{index}"
+    end
+
+    # finish_request! discards the connection when pending? is true, so it must reflect exactly the
+    # bytes taken from the socket but not yet returned. A read that owed CHUNK_SIZE or more after
+    # draining the buffer must not take bytes past its end.
+    def assert_buffer_consistent(test_case, large_read, context)
+      buffered = test_case.buffered
+
+      assert_equal buffered.positive?, test_case.reader.pending?, "#{context}: pending? with #{buffered} bytes buffered"
+      assert_equal 0, buffered, "#{context}: large read took #{buffered} bytes past its end" if large_read
+    end
+
+    def assert_fails_within_timeout(test_case, kind, arg, context)
+      assert_raises(IO::TimeoutError, context) { test_case.reader.public_send(kind, arg) }
+      hang = TimeoutRecoveryReaderFuzz.now - test_case.faults.last_data_at
+      timeout = test_case.timeout
+      STATS[:timeouts] += 1
+      STATS[:max_hang_past_timeout] = [STATS[:max_hang_past_timeout], hang - timeout].max
+
+      assert_operator hang, :>=, timeout * 0.9, "#{context}: failed #{hang.round(3)}s after the last byte"
+      assert_operator hang, :<=, timeout + TIMEOUT_SLACK,
+                      "#{context}: failed #{hang.round(3)}s after the last byte with a #{timeout}s timeout"
+    end
+
+    # Every pause is shorter than the timeout, but together they outlast it.
+    def assert_paused_through(started, label)
+      paused = PAUSES * TIMEOUTS[:pauses] * PAUSE_FRACTION
+
+      assert_operator TimeoutRecoveryReaderFuzz.now - started, :>=, paused * 0.9, "#{label}: finished before its pauses"
+    end
+
+    # Every wait is bounded by the reader's timeout and, under a fiber scheduler, lets other fibers
+    # run. The only other wait is the zero-timeout poll that bypasses the scheduler after a wait
+    # reports a timeout.
+    def assert_bounded_waits(test_case)
+      test_case.faults.waits.each do |wait_timeout, blocking, scheduled|
+        bounded = wait_timeout&.zero? ? blocking : wait_timeout == test_case.timeout && !(scheduled && blocking)
+
+        assert bounded, "#{test_case.label}: waited with timeout #{wait_timeout.inspect} " \
+                        "(blocking: #{blocking.inspect}, scheduler: #{scheduled})"
+      end
+    end
+
+    def assert_bounded_scratch(test_case)
+      scratch = test_case.reader.instance_variable_get(:@scratch)
+      return unless scratch
+
+      assert_operator ObjectSpace.memsize_of(scratch), :<=, LARGE + 256, "#{test_case.label}: scratch buffer grew"
+    end
+  end
+
+  ##
+  # Runs cases; included into the spec.
+  ##
+  module Cases
+    include Checks
+
+    def fuzz(transport, scheduler, mode, count)
+      with_scheduler(scheduler) do
+        count.times do |index|
+          label = "FUZZ_SEED=#{SEED} #{transport}/#{scheduler || 'no scheduler'}/#{mode} case #{index}"
+          rng = Random.new(Zlib.crc32(label))
+          Timeout.timeout(CASE_LIMIT, nil, "#{label} still running after #{CASE_LIMIT}s") do
+            fuzz_case(transport, mode, index, rng, label)
+          end
+        end
+      end
+    end
+
+    private
+
+    def with_scheduler(selector)
+      return yield unless selector
+
+      reactor = Async::Reactor.new(selector: IO::Event::Selector.const_get(selector).new(Fiber.current))
+      reactor.run(finished: false) do |task|
+        yield
+      ensure
+        task.children&.each(&:stop)
+      end.wait
+    ensure
+      Fiber.set_scheduler(nil) if selector
+    end
+
+    def fuzz_case(transport, mode, index, rng, label)
+      stream, ops = script(rng, mode)
+      steps, stall_at = Plans.build(mode, rng, stream, ops, index)
+      sock, peer = Wire.socket_pair(transport)
+      release = Thread::Queue.new
+      writer = Wire.start_writer(peer, steps, release)
+      faults = Faults.new(Random.new(rng.rand(2**32)))
+      faults.attach(sock)
+      timeout = TIMEOUTS.fetch(mode)
+      check(Case.new(label, Reader.new(sock, timeout), faults, Model.new(stream), ops, stall_at, timeout), mode, rng)
+    ensure
+      Wire.close_quietly(sock)
+      release&.push(true)
+      assert writer.join(10), "#{label}: writer still running" if writer
+    end
+
+    def script(rng, mode)
+      return Scripts.protocol_script(rng, with_large_value: mode == :stall) unless mode == :exact
+      return Scripts.random_script(rng) if rng.rand(3).zero?
+
+      stream, ops = Scripts.protocol_script(rng)
+      # EOF partway through a reply.
+      stream = stream.byteslice(0, rng.rand(0..stream.bytesize)) if rng.rand(4).zero?
+      [stream, ops]
+    end
+
+    def check(test_case, mode, rng)
+      started = TimeoutRecoveryReaderFuzz.now
+      test_case.faults.mark
+      STATS[:cases] += 1
+      unless run_ops(test_case) == :stalled
+        assert_same_bytes(test_case.model.rest, drain(test_case.reader, rng), "#{test_case.label} drain")
+        assert_paused_through(started, test_case.label) if mode == :pauses
+      end
+
+      assert_bounded_waits(test_case)
+      assert_bounded_scratch(test_case)
+    end
+
+    def run_ops(test_case)
+      test_case.ops.each_with_index do |(kind, arg), index|
+        context = "#{test_case.label} op #{index} #{kind}(#{arg.inspect})"
+        if test_case.stalls_before?(kind, arg)
+          assert_fails_within_timeout(test_case, kind, arg, context)
+          return :stalled
+        end
+
+        large_read = kind == :read && arg - test_case.buffered >= CHUNK
+        actual = test_case.reader.public_send(kind, arg)
+        test_case.returned!(actual)
+
+        assert_same_bytes(test_case.model.apply(kind, arg), actual, context)
+        assert_buffer_consistent(test_case, large_read, context)
+        STATS[:ops] += 1
+      end
+    end
+
+    def drain(reader, rng)
+      rest = String.new(encoding: Encoding::BINARY)
+      while (chunk = reader.read([1, CHUNK, LARGE + 1, 3 * LARGE].sample(random: rng)))
+        rest << chunk
+      end
+      rest
+    end
+  end
+end
+
+if ENV['FUZZ_STATS']
+  Minitest.after_run do
+    puts "TimeoutRecoveryReader fuzz FUZZ_SEED=#{TimeoutRecoveryReaderFuzz::SEED}: #{TimeoutRecoveryReaderFuzz::STATS}"
+  end
+end
+
+describe 'TimeoutRecoveryReader fuzz' do
+  include TimeoutRecoveryReaderFuzz::Cases
+
+  TimeoutRecoveryReaderFuzz::TRANSPORTS.each do |transport|
+    TimeoutRecoveryReaderFuzz::SCHEDULERS.each do |scheduler|
+      describe "over #{transport} #{scheduler ? "under Async with #{scheduler}" : 'without a fiber scheduler'}" do
+        it 'matches IO#gets and IO#read on randomized streams and faults' do
+          fuzz(transport, scheduler, :exact, TimeoutRecoveryReaderFuzz::ITERATIONS)
+        end
+
+        it 'fails within one timeout of the last byte when the stream stops' do
+          fuzz(transport, scheduler, :stall, [TimeoutRecoveryReaderFuzz::ITERATIONS / 2, 2].max)
+        end
+
+        it 'keeps reading through pauses that are each shorter than the timeout' do
+          fuzz(transport, scheduler, :pauses, [TimeoutRecoveryReaderFuzz::ITERATIONS / 8, 1].max)
+        end
+      end
+    end
+  end
+end

From f3b596d69b3ba0c7bba5ed909275b1d916f5229a Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 15:57:02 -0700
Subject: [PATCH 4/8] Re-enable GC for the async_get benchmark

Under BENCH_TARGET=all, the GC suite of the preceding get target leaves GC disabled, so async_get ran without collection on top of the garbage left by get, exhausting memory on the CI runner. It now enables GC before running.
---
 bin/benchmark | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/bin/benchmark b/bin/benchmark
index a9556071..cf7e3555 100755
--- a/bin/benchmark
+++ b/bin/benchmark
@@ -187,10 +187,12 @@ if %w[all get].include?(bench_target)
 end
 
 # Large-value gets under a fiber scheduler, through the default read path and the
-# exp_async_timeout_recovery reader. GC stays enabled: with the GC suite, the
-# discarded values exhaust memory within a few seconds at the default payload size.
+# exp_async_timeout_recovery reader. GC must run here: the GC suite of earlier targets
+# leaves it disabled, and the discarded values exhaust memory within a few seconds at the
+# default payload size.
 if %w[all async_get].include?(bench_target)
   require 'async'
+  GC.enable
   Sync do
     Benchmark.ips do |x|
       x.config(warmup: bench_warmup, time: bench_time)

From 7bdfde9c576253ab136dcb16d7662098d5cf5160 Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 16:31:14 -0700
Subject: [PATCH 5/8] Keep fuzz faults realistic for TLS and report socket
 history on failure

A TLS socket holding decrypted bytes returns them from read_nonblock, so it never reports EAGAIN; injecting a spurious EAGAIN there left the reader waiting on an empty TCP socket for bytes OpenSSL already had, which a stalled stream turned into a false timeout. Failure messages for unexpected and out-of-bounds timeouts now include the case's recent reads and waits.
---
 .../test_timeout_recovery_reader_fuzz.rb      | 76 +++++++++++++++----
 1 file changed, 60 insertions(+), 16 deletions(-)

diff --git a/test/protocol/test_timeout_recovery_reader_fuzz.rb b/test/protocol/test_timeout_recovery_reader_fuzz.rb
index 433e665e..afe8da15 100644
--- a/test/protocol/test_timeout_recovery_reader_fuzz.rb
+++ b/test/protocol/test_timeout_recovery_reader_fuzz.rb
@@ -104,6 +104,9 @@ def take(length)
   # Fault injection and bookkeeping for one reader socket.
   ##
   class Faults
+    # Recent reads and waits, for failure messages.
+    HISTORY = 24
+
     attr_reader :waits, :delivered, :last_data_at
 
     def initialize(rng)
@@ -113,6 +116,7 @@ def initialize(rng)
       @short = [0.0, 0.2, 0.5].sample(random: rng)
       @starve = [0.0, 0.3, 1.0].sample(random: rng)
       @waits = []
+      @history = []
       @delivered = 0
       mark
     end
@@ -127,16 +131,26 @@ def mark
     end
 
     # The length to request, or the Symbol a spurious EAGAIN would return.
-    def read_length(maxlen)
+    def read_length(sock, maxlen)
       STATS[:large_reads] += 1 if maxlen > CHUNK
       roll = @rng.rand
-      return :wait_readable if roll < @spurious
-      return :wait_writable if roll < @spurious + @writable
+      spurious = spurious_result(sock, roll)
+      return spurious.tap { record(:read, maxlen, :spurious, spurious) } if spurious
 
       roll > 1 - @short ? @rng.rand(1..maxlen) : maxlen
     end
 
-    def received(result)
+    # A TLS socket holding decrypted bytes always returns them, so it never gets a spurious EAGAIN.
+    def spurious_result(sock, roll)
+      return if sock.respond_to?(:pending) && sock.pending.positive?
+      return :wait_readable if roll < @spurious
+
+      :wait_writable if roll < @spurious + @writable
+    end
+
+    def read(maxlen, length)
+      result = yield
+      record(:read, maxlen, length, result.is_a?(String) ? result.bytesize : result.inspect)
       return result unless result.is_a?(String)
 
       @delivered += result.bytesize
@@ -144,15 +158,36 @@ def received(result)
       result
     end
 
-    # Records the wait and decides whether to report a timeout although the socket is ready.
-    def starve?(sock, event, timeout)
+    # Reports a timeout without waiting when the socket is ready and the roll says the scheduler
+    # resumed the reader late; otherwise waits.
+    def wait(sock, event, timeout)
+      started = TimeoutRecoveryReaderFuzz.now
       @waits << [timeout, Fiber.blocking?, !Fiber.scheduler.nil?]
+      starved = starve?(sock, event, timeout)
+      result = starved ? nil : yield
+      record(event, timeout, starved ? :starved : :waited, result ? :ready : result.inspect,
+             (TimeoutRecoveryReaderFuzz.now - started).round(4))
+      result
+    end
+
+    def history
+      @history.map { |event| event.join(' ') }.join("\n")
+    end
+
+    private
+
+    def starve?(sock, event, timeout)
       return false if timeout.nil? || timeout.zero? || @rng.rand >= @starve
       return false unless Fiber.blocking { IO.instance_method(event).bind_call(sock.to_io, 0) }
 
       STATS[:starved_waits] += 1
       true
     end
+
+    def record(*event)
+      @history << [(TimeoutRecoveryReaderFuzz.now - @last_data_at).round(4), *event]
+      @history.shift if @history.size > HISTORY
+    end
   end
 
   ##
@@ -160,18 +195,18 @@ def starve?(sock, event, timeout)
   ##
   module Hooks
     def read_nonblock(maxlen, buf = nil, exception: true)
-      length = @fuzz_faults.read_length(maxlen)
+      length = @fuzz_faults.read_length(self, maxlen)
       return length if length.is_a?(Symbol)
 
-      @fuzz_faults.received(super(length, buf, exception: exception))
+      @fuzz_faults.read(maxlen, length) { super(length, buf, exception: exception) }
     end
 
     def wait_readable(timeout = nil)
-      @fuzz_faults.starve?(self, :wait_readable, timeout) ? nil : super
+      @fuzz_faults.wait(self, :wait_readable, timeout) { super }
     end
 
     def wait_writable(timeout = nil)
-      @fuzz_faults.starve?(self, :wait_writable, timeout) ? nil : super
+      @fuzz_faults.wait(self, :wait_writable, timeout) { super }
     end
   end
 
@@ -394,10 +429,11 @@ def assert_fails_within_timeout(test_case, kind, arg, context)
       timeout = test_case.timeout
       STATS[:timeouts] += 1
       STATS[:max_hang_past_timeout] = [STATS[:max_hang_past_timeout], hang - timeout].max
+      events = "; recent socket events (seconds since data, event, ...):\n#{test_case.faults.history}"
 
-      assert_operator hang, :>=, timeout * 0.9, "#{context}: failed #{hang.round(3)}s after the last byte"
+      assert_operator hang, :>=, timeout * 0.9, "#{context}: failed #{hang.round(3)}s after the last byte#{events}"
       assert_operator hang, :<=, timeout + TIMEOUT_SLACK,
-                      "#{context}: failed #{hang.round(3)}s after the last byte with a #{timeout}s timeout"
+                      "#{context}: failed #{hang.round(3)}s after the last byte with a #{timeout}s timeout#{events}"
     end
 
     # Every pause is shorter than the timeout, but together they outlast it.
@@ -491,7 +527,7 @@ def check(test_case, mode, rng)
       test_case.faults.mark
       STATS[:cases] += 1
       unless run_ops(test_case) == :stalled
-        assert_same_bytes(test_case.model.rest, drain(test_case.reader, rng), "#{test_case.label} drain")
+        assert_same_bytes(test_case.model.rest, drain(test_case, rng), "#{test_case.label} drain")
         assert_paused_through(started, test_case.label) if mode == :pauses
       end
 
@@ -508,7 +544,7 @@ def run_ops(test_case)
         end
 
         large_read = kind == :read && arg - test_case.buffered >= CHUNK
-        actual = test_case.reader.public_send(kind, arg)
+        actual = read_reporting_history(test_case, kind, arg, context)
         test_case.returned!(actual)
 
         assert_same_bytes(test_case.model.apply(kind, arg), actual, context)
@@ -517,13 +553,21 @@ def run_ops(test_case)
       end
     end
 
-    def drain(reader, rng)
+    def drain(test_case, rng)
       rest = String.new(encoding: Encoding::BINARY)
-      while (chunk = reader.read([1, CHUNK, LARGE + 1, 3 * LARGE].sample(random: rng)))
+      size = -> { [1, CHUNK, LARGE + 1, 3 * LARGE].sample(random: rng) }
+      while (chunk = read_reporting_history(test_case, :read, size.call, test_case.label))
         rest << chunk
       end
       rest
     end
+
+    def read_reporting_history(test_case, kind, arg, context)
+      test_case.reader.public_send(kind, arg)
+    rescue IO::TimeoutError => e
+      flunk "#{context}: unexpected #{e.message}; recent socket events " \
+            "(seconds since data, event, ...):\n#{test_case.faults.history}"
+    end
   end
 end
 

From 6b647631ebc7f7d8d7b25bd877370bc4b918e9bd Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 17:00:30 -0700
Subject: [PATCH 6/8] Read once more before failing when the post-timeout poll
 finds nothing

A zero-timeout poll can report nothing without checking the socket while the thread has a pending interrupt (from Thread#wakeup, a queue push, and the like), so after a wait that timed out, the reader could fail a read whose data had already arrived. When the poll finds nothing, the reader now makes one more non-blocking read and fails only if that read also has nothing. Waits and their bounds are unchanged.

The fuzz no longer injects a spurious EAGAIN into that final read, which no socket returns while it holds data, and the large-value integration test counts a recovery found by that read.
---
 CHANGELOG.md                                  |  3 +++
 lib/dalli/protocol/timeout_recovery_reader.rb | 26 +++++++++----------
 .../test_async_timeout_recovery.rb            | 13 +++++++---
 test/protocol/test_timeout_recovery_reader.rb | 13 ++++++++++
 .../test_timeout_recovery_reader_fuzz.rb      |  9 +++++--
 5 files changed, 45 insertions(+), 19 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d80bdc4a..151be551 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,9 @@ Unreleased
 - Fix `exp_async_timeout_recovery` replaying the previous reply instead of reporting EOF after a TLS server
   closes the connection cleanly. (nherson)
 
+- Fix `exp_async_timeout_recovery` failing a read whose data is ready when the poll after a timed-out wait
+  misses it, which a zero-timeout poll can do while the thread has a pending interrupt. (nherson)
+
 - Add per-request opaque-correlation controls and a client `opaque_correlation_request_only` mode
   for opt-in rollouts, with eager PRNG initialization and unchanged unflagged reads. (mrattle)
 
diff --git a/lib/dalli/protocol/timeout_recovery_reader.rb b/lib/dalli/protocol/timeout_recovery_reader.rb
index 88d597b6..c2c00c31 100644
--- a/lib/dalli/protocol/timeout_recovery_reader.rb
+++ b/lib/dalli/protocol/timeout_recovery_reader.rb
@@ -87,10 +87,13 @@ def scratch
         @scratch ||= String.new(encoding: Encoding::BINARY)
       end
 
-      # Waits like read_chunk, but reads at most maxlen bytes into outbuf.
+      # Reads at most maxlen bytes into outbuf, waiting while the socket has none; nil at EOF.
       def read_partial(maxlen, outbuf)
+        last_try = false
         while (result = @sock.read_nonblock(maxlen, outbuf, exception: false)).is_a?(Symbol)
-          wait_for(result)
+          raise IO::TimeoutError, "IO timeout after #{@timeout}s" if last_try
+
+          last_try = !wait_for(result)
         end
         result
       end
@@ -100,26 +103,21 @@ def read_partial(maxlen, outbuf)
       # buffer but OpenSSL leaves it untouched, so it is cleared here to avoid replaying old bytes.
       def fill
         partial = @buffer.byteslice(@offset, @buffer.bytesize - @offset) if pending?
-        filled = read_chunk
+        filled = !read_partial(CHUNK_SIZE, @buffer).nil?
         @buffer.clear unless filled
         @offset = 0
         @buffer.prepend(partial) if partial
         filled
       end
 
-      # Waits on :wait_readable or :wait_writable, which name the socket method to call.
-      def read_chunk
-        while (result = @sock.read_nonblock(CHUNK_SIZE, @buffer, exception: false)).is_a?(Symbol)
-          wait_for(result)
-        end
-        !result.nil?
-      end
-
+      # Waits on :wait_readable or :wait_writable, which name the socket method to call. After a
+      # wait that times out, polls once outside the fiber scheduler. Returns false when that poll
+      # finds nothing too, and the caller then reads once more before failing: a zero-timeout poll
+      # can report nothing without checking the socket while the thread has a pending interrupt.
       def wait_for(event)
-        return if @sock.public_send(event, @timeout)
-        return if Fiber.blocking { @sock.public_send(event, 0) }
+        return true if @sock.public_send(event, @timeout)
 
-        raise IO::TimeoutError, "IO timeout after #{@timeout}s"
+        Fiber.blocking { @sock.public_send(event, 0) } ? true : false
       end
     end
   end
diff --git a/test/integration/test_async_timeout_recovery.rb b/test/integration/test_async_timeout_recovery.rb
index 7618b73b..4088b446 100644
--- a/test/integration/test_async_timeout_recovery.rb
+++ b/test/integration/test_async_timeout_recovery.rb
@@ -203,14 +203,21 @@ def last_data_at
       @events.reverse_each.find { |kind, bytes| kind == :read && bytes.positive? }&.last
     end
 
-    # Bytes received before each wait that reported a timeout and was followed by a poll that found data.
+    # Bytes received before each wait that reported a timeout after which the reader still found
+    # data: through the poll that follows or, when that poll misses it, the read after the poll.
     def recoveries(timeout)
       received = 0
-      @events.each_cons(2).filter_map do |event, following|
+      @events.each_cons(3).filter_map do |event, poll, after|
         received += event[1] if event.first == :read
-        received if event == [:wait, timeout, false] && following == [:wait, 0, true]
+        received if event == [:wait, timeout, false] && found_data?(poll, after)
       end
     end
+
+    private
+
+    def found_data?(poll, after)
+      poll == [:wait, 0, true] || (poll == [:wait, 0, false] && after.first == :read && after[1].positive?)
+    end
   end
 
   def in_scheduler(selector, &)
diff --git a/test/protocol/test_timeout_recovery_reader.rb b/test/protocol/test_timeout_recovery_reader.rb
index 8d48c0e9..bef81355 100644
--- a/test/protocol/test_timeout_recovery_reader.rb
+++ b/test/protocol/test_timeout_recovery_reader.rb
@@ -94,6 +94,19 @@ def write_in_background(peer, *chunks)
     end
   end
 
+  it 'reads data that is ready when the poll after a timed-out wait misses it' do
+    with_reader do |reader, peer, sock|
+      # A zero-timeout poll can report nothing without checking the socket while the thread has a
+      # pending interrupt, such as one from Thread#wakeup or a queue push.
+      sock.define_singleton_method(:wait_readable) do |timeout|
+        peer.write("HD\r\n") if timeout.positive?
+        nil
+      end
+
+      assert_equal "HD\r\n", reader.gets("\r\n")
+    end
+  end
+
   it 'keeps its buffer at the chunk size after a large read' do
     value = SecureRandom.random_bytes(1_000_000)
     with_reader do |reader, peer|
diff --git a/test/protocol/test_timeout_recovery_reader_fuzz.rb b/test/protocol/test_timeout_recovery_reader_fuzz.rb
index afe8da15..529382cf 100644
--- a/test/protocol/test_timeout_recovery_reader_fuzz.rb
+++ b/test/protocol/test_timeout_recovery_reader_fuzz.rb
@@ -130,11 +130,15 @@ def mark
       @last_data_at = TimeoutRecoveryReaderFuzz.now
     end
 
-    # The length to request, or the Symbol a spurious EAGAIN would return.
+    # The length to request, or the Symbol a spurious EAGAIN would return. The read after a poll
+    # that found nothing is the reader's last check before failing, so it gets no spurious EAGAIN:
+    # that would fail a read whose data is there, which no socket does.
     def read_length(sock, maxlen)
       STATS[:large_reads] += 1 if maxlen > CHUNK
+      last_check = @last_check
+      @last_check = false
       roll = @rng.rand
-      spurious = spurious_result(sock, roll)
+      spurious = spurious_result(sock, roll) unless last_check
       return spurious.tap { record(:read, maxlen, :spurious, spurious) } if spurious
 
       roll > 1 - @short ? @rng.rand(1..maxlen) : maxlen
@@ -165,6 +169,7 @@ def wait(sock, event, timeout)
       @waits << [timeout, Fiber.blocking?, !Fiber.scheduler.nil?]
       starved = starve?(sock, event, timeout)
       result = starved ? nil : yield
+      @last_check = timeout&.zero? && !result
       record(event, timeout, starved ? :starved : :waited, result ? :ready : result.inspect,
              (TimeoutRecoveryReaderFuzz.now - started).round(4))
       result

From 19d7107e9c2c6c7cd5a4eae01abca55c9e1ad8f4 Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 17:16:10 -0700
Subject: [PATCH 7/8] Run the async_get benchmark after the existing targets

With BENCH_TARGET=all, async_get ran between get and get_multi and the async gem was required at startup, so every later target ran in a different process state than before it existed; the Benchmarks job was then cancelled partway through delete_multi. async_get now runs last, and async is required only by that target.
---
 bin/benchmark | 36 ++++++++++++++++++------------------
 1 file changed, 18 insertions(+), 18 deletions(-)

diff --git a/bin/benchmark b/bin/benchmark
index cf7e3555..5d5196ed 100755
--- a/bin/benchmark
+++ b/bin/benchmark
@@ -13,7 +13,7 @@ require 'json'
 
 gemfile do
   source 'https://rubygems.org'
-  gem 'async'
+  gem 'async', require: false
   gem 'benchmark-ips'
   gem 'logger'
   gem 'opentelemetry-sdk'
@@ -186,23 +186,6 @@ if %w[all get].include?(bench_target)
   end
 end
 
-# Large-value gets under a fiber scheduler, through the default read path and the
-# exp_async_timeout_recovery reader. GC must run here: the GC suite of earlier targets
-# leaves it disabled, and the discarded values exhaust memory within a few seconds at the
-# default payload size.
-if %w[all async_get].include?(bench_target)
-  require 'async'
-  GC.enable
-  Sync do
-    Benchmark.ips do |x|
-      x.config(warmup: bench_warmup, time: bench_time)
-      x.report('get dalli async') { client.get('key') }
-      x.report('get dalli async timeout recovery') { client.get('key', exp_async_timeout_recovery: true) }
-      x.compare!
-    end
-  end
-end
-
 if %w[all get_multi].include?(bench_target)
   Benchmark.ips do |x|
     x.config(warmup: bench_warmup, time: bench_time, suite: suite)
@@ -253,3 +236,20 @@ if %w[all delete_multi].include?(bench_target)
     x.compare!
   end
 end
+
+# Large-value gets under a fiber scheduler, through the default read path and the
+# exp_async_timeout_recovery reader. Runs last so the targets above run as they did before
+# it existed. GC must run here: the GC suite of earlier targets leaves it disabled, and the
+# discarded values exhaust memory within a few seconds at the default payload size.
+if %w[all async_get].include?(bench_target)
+  require 'async'
+  GC.enable
+  Sync do
+    Benchmark.ips do |x|
+      x.config(warmup: bench_warmup, time: bench_time)
+      x.report('get dalli async') { client.get('key') }
+      x.report('get dalli async timeout recovery') { client.get('key', exp_async_timeout_recovery: true) }
+      x.compare!
+    end
+  end
+end

From 649eaf0ae81dc2154635e96e26d478fa661c9d1e Mon Sep 17 00:00:00 2001
From: Nicholas Herson <nicholas.herson@shopify.com>
Date: Thu, 15 Oct 2026 17:48:13 -0700
Subject: [PATCH 8/8] Make the async_get benchmark opt-in

Main's Benchmarks job passed 18 of 18 runs; with async_get in BENCH_TARGET=all it was cancelled partway through delete_multi in 3 of 10, even after async_get moved to the end and stopped loading async at startup. The only remaining difference before delete_multi was the extra inline gems (async and its dependencies activate newer json, tempfile, and others). async_get and its gem now apply only with BENCH_TARGET=async_get, so BENCH_TARGET=all runs exactly as on main.
---
 bin/benchmark | 13 +++++--------
 1 file changed, 5 insertions(+), 8 deletions(-)

diff --git a/bin/benchmark b/bin/benchmark
index 5d5196ed..bb3150f7 100755
--- a/bin/benchmark
+++ b/bin/benchmark
@@ -7,13 +7,14 @@
 # run with:
 # bundle exec bin/benchmark
 # RUBY_YJIT_ENABLE=1 BENCH_TARGET=get bundle exec bin/benchmark
-# BENCH_TARGET=async_get bundle exec bin/benchmark
+# BENCH_TARGET=async_get bundle exec bin/benchmark (not part of BENCH_TARGET=all)
 require 'bundler/inline'
 require 'json'
 
 gemfile do
   source 'https://rubygems.org'
-  gem 'async', require: false
+  # Only async_get needs it; other runs keep the same gems as before that target existed.
+  gem 'async' if ENV['BENCH_TARGET'] == 'async_get'
   gem 'benchmark-ips'
   gem 'logger'
   gem 'opentelemetry-sdk'
@@ -238,12 +239,8 @@ if %w[all delete_multi].include?(bench_target)
 end
 
 # Large-value gets under a fiber scheduler, through the default read path and the
-# exp_async_timeout_recovery reader. Runs last so the targets above run as they did before
-# it existed. GC must run here: the GC suite of earlier targets leaves it disabled, and the
-# discarded values exhaust memory within a few seconds at the default payload size.
-if %w[all async_get].include?(bench_target)
-  require 'async'
-  GC.enable
+# exp_async_timeout_recovery reader. Runs only when selected, so BENCH_TARGET=all is unchanged.
+if bench_target == 'async_get'
   Sync do
     Benchmark.ips do |x|
       x.config(warmup: bench_warmup, time: bench_time)