From 1b2f69f0829f645701100b54a27c3cce64115353 Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Tue, 20 Oct 2026 14:43:13 +0200 Subject: [PATCH 1/2] fix: retry requests on dropped connections The load balancer front end in front of the API occasionally drops connections. A pooled keep-alive connection closed without a TLS close_notify makes the next request fail with "SSL_read: unexpected eof while reading", and faraday-net_http_persistent disables Net::HTTP's own retry (max_retries = 0). Add StreamChat::RetryMiddleware to the client's Faraday connection. It retries up to 2 times with jittered backoff (~100ms, ~200ms): - any method when the failure happened in Net::HTTP's connect or begin_transport phase, so the request was never sent; - GET, HEAD and OPTIONS also when the connection dropped mid-flight. HTTP error responses and timeouts are never retried, and IO bodies are retried only when they can be rewound. Co-Authored-By: Claude Opus 5.5 --- lib/stream-chat/client.rb | 3 + lib/stream-chat/retry_middleware.rb | 93 ++++++++++++++++ spec/retry_middleware_spec.rb | 164 ++++++++++++++++++++++++++++ spec/support/tls_test_server.rb | 138 +++++++++++++++++++++++ 4 files changed, 398 insertions(+) create mode 100644 lib/stream-chat/retry_middleware.rb create mode 100644 spec/retry_middleware_spec.rb create mode 100644 spec/support/tls_test_server.rb diff --git a/lib/stream-chat/client.rb b/lib/stream-chat/client.rb index 78bf025..fec3d52 100644 --- a/lib/stream-chat/client.rb +++ b/lib/stream-chat/client.rb @@ -23,6 +23,7 @@ require 'stream-chat/moderation' require 'stream-chat/channel_batch_updater' require 'stream-chat/webhook' +require 'stream-chat/retry_middleware' module StreamChat DEFAULT_BLOCKLIST = 'profanity' @@ -70,6 +71,8 @@ def initialize(api_key, api_secret, timeout = nil, **options) faraday.options[:open_timeout] = @timeout faraday.options[:timeout] = @timeout faraday.request :multipart + # Retries requests whose connection was dropped before a response; see RetryMiddleware. + faraday.use StreamChat::RetryMiddleware faraday.adapter :net_http_persistent, pool_size: 5 do |http| # AWS load balancer idle timeout is 60 secs, so let's make it 59 http.idle_timeout = 59 diff --git a/lib/stream-chat/retry_middleware.rb b/lib/stream-chat/retry_middleware.rb new file mode 100644 index 0000000..9d43783 --- /dev/null +++ b/lib/stream-chat/retry_middleware.rb @@ -0,0 +1,93 @@ +# typed: strict +# frozen_string_literal: true + +require 'faraday' +require 'sorbet-runtime' + +module StreamChat + # Retries a request whose connection was dropped at the transport level. + # + # Why: the load balancer front end in front of the API occasionally drops + # connections (e.g. scheduled proxy restarts). Most often an idle pooled + # keep-alive connection is closed without a TLS close_notify, and the next + # request on it fails with "SSL_read: unexpected eof while reading" before + # anything is written. faraday-net_http_persistent sets + # Net::HTTP#max_retries = 0, so nothing below this middleware retries. + # + # A request is retried only when repeating it is safe: + # - any method, when the failure happened while Net::HTTP was opening or + # checking the connection (`connect` / `begin_transport`), so the request + # provably never left the client; + # - GET, HEAD and OPTIONS also when the connection dropped later, since the + # server may have received them but repeating them is harmless. + # HTTP error responses and timeouts are never retried, and a body that + # cannot be rewound is never sent twice. + class RetryMiddleware < Faraday::Middleware + extend T::Sig + + MAX_RETRIES = 2 + BASE_DELAY = 0.1 + SAFE_METHODS = T.let(%i[get head options].freeze, T::Array[Symbol]) + # A Net::HTTP frame that runs before the request is written: opening the + # socket/TLS session, or the stale-connection check of a pooled socket. + # Matches both backtrace label styles (`connect' and 'Net::HTTP#connect'). + PRE_SEND_FRAME = T.let(%r{/net/http\.rb:\d+:in [`'](?:Net::HTTP#)?(?:connect|begin_transport)'}, Regexp) + + sig { params(env: Faraday::Env).returns(Faraday::Response) } + def call(env) + request_body = env.request_body + retries = 0 + begin + app.call(env) + rescue Faraday::ConnectionFailed, Faraday::SSLError => e + raise unless retries < MAX_RETRIES && retryable?(e, env.method) && rewind(request_body) + + retries += 1 + sleep(backoff(retries)) + env.request_body = request_body + retry + end + end + + private + + sig { params(error: Exception, method: Symbol).returns(T::Boolean) } + def retryable?(error, method) + SAFE_METHODS.include?(method) || not_sent?(error) + end + + # True when the failure was raised from Net::HTTP before the request was + # written. The adapter wraps the original error, so walk the cause chain. + sig { params(error: Exception).returns(T::Boolean) } + def not_sent?(error) + current = T.let(error, T.nilable(Exception)) + while current + return true if (current.backtrace || []).any? { |frame| PRE_SEND_FRAME.match?(frame) } + + current = current.cause + end + false + end + + # Strings (and no body) can be sent again as is; IO bodies, such as + # multipart uploads, only when they can be rewound to the start. + sig { params(body: T.untyped).returns(T::Boolean) } + def rewind(body) + return true unless body.respond_to?(:read) + return false unless body.respond_to?(:rewind) + + body.rewind + true + rescue IOError, SystemCallError + false + end + + # ~100ms then ~200ms, with jitter so clients that lost connections at the + # same moment do not reconnect in lockstep. + sig { params(retry_number: Integer).returns(Float) } + def backoff(retry_number) + delay = BASE_DELAY * (1 << (retry_number - 1)) + delay + (rand * delay / 2) + end + end +end diff --git a/spec/retry_middleware_spec.rb b/spec/retry_middleware_spec.rb new file mode 100644 index 0000000..cf97bc8 --- /dev/null +++ b/spec/retry_middleware_spec.rb @@ -0,0 +1,164 @@ +# frozen_string_literal: true + +require 'socket' +require 'tempfile' +require 'stream-chat' + +# Exercises StreamChat::RetryMiddleware through the real client stack +# (Faraday + net_http_persistent) against a local TLS server. No network +# access or API credentials needed. +describe StreamChat::RetryMiddleware do + let(:server) { TLSTestServer.new } + let(:client) { tls_client(server.url) } + + after { server.close } + + def tls_client(url, timeout: 3.0) + client = StreamChat::Client.new('key', 'secret', timeout, base_url: url) + # The test server uses a self-signed certificate. + client.conn.ssl.verify = false + client + end + + # Opens a pooled keep-alive connection, then has the server close it without + # a TLS close_notify, the way the load balancer front end drops connections. + def make_pooled_connection_stale + server.script(:respond_close) + client.get('warmup') + sleep 0.3 # let the server's close reach the client socket + server.requests + end + + def open_pooled_connection + client.get('warmup') + server.requests + end + + context 'when the pooled connection went stale (request never sent)' do + it 'retries a POST, and the server receives it once' do + make_pooled_connection_stale + + response = client.post('messages', data: { message: { text: 'hi' } }) + + expect(response['duration']).to eq('1ms') + requests = server.requests + expect(requests.map(&:http_method)).to eq(['POST']) + expect(JSON.parse(requests.first.body)).to eq('message' => { 'text' => 'hi' }) + end + + it 'retries a GET, and the server receives it once' do + make_pooled_connection_stale + + client.get('channels') + + expect(server.requests.map(&:http_method)).to eq(['GET']) + end + + it 'retries a multipart upload with the full body' do + make_pooled_connection_stale + content = "line of the uploaded file\n" * 4096 + file = Tempfile.new('upload') + file.write(content) + file.close + + client.send_file('uploads', file.path, { 'id' => 'u1' }, 'text/plain') + + requests = server.requests + expect(requests.map(&:http_method)).to eq(['POST']) + body = requests.first.body + expect(body.bytesize).to eq(requests.first.headers['content-length'].to_i) + expect(body).to include(content) + expect(body).to include('{"id":"u1"}') + boundary = requests.first.headers['content-type'][/boundary=(\S+)/, 1] + expect(body).to end_with("--#{boundary}--\r\n") + ensure + file&.unlink + end + + it 'does not retry an IO body that cannot be rewound' do + make_pooled_connection_stale + body = Class.new do + def initialize(data) + @io = StringIO.new(data) + end + + def read(*args) + @io.read(*args) + end + end.new('{"a":1}') + + expect do + client.conn.post('messages', body, 'Content-Length' => '7', 'Content-Type' => 'application/json') + end.to raise_error(Faraday::SSLError) + expect(server.requests).to be_empty + end + end + + context 'when the connection drops after the request was sent' do + it 'retries a GET, and the server receives it twice' do + open_pooled_connection + server.script(:drop) + + client.get('channels') + + expect(server.requests.map(&:http_method)).to eq(%w[GET GET]) + end + + it 'does not retry a POST and raises the transport error' do + open_pooled_connection + server.script(:drop) + + expect { client.post('messages', data: { message: { text: 'hi' } }) } + .to raise_error(Faraday::SSLError, /unexpected eof/) + expect(server.requests.map(&:http_method)).to eq(['POST']) + end + + it 'stops after 3 attempts' do + open_pooled_connection + server.script(:drop, :drop, :drop, :drop) + + expect { client.get('channels') }.to raise_error(Faraday::SSLError) + expect(server.requests.map(&:http_method)).to eq(%w[GET GET GET]) + end + end + + context 'when the connection is refused' do + it 'retries a POST once the server is listening' do + port = TCPServer.new('127.0.0.1', 0).then do |s| + s.addr[1].tap { s.close } + end + late_client = tls_client("https://127.0.0.1:#{port}") + late_server = nil + starter = Thread.new do + sleep 0.05 # first attempt is refused; the first retry waits >= 100ms + late_server = TLSTestServer.new(port: port) + end + + late_client.post('messages', data: { message: { text: 'hi' } }) + + starter.join + expect(late_server.requests.map(&:http_method)).to eq(['POST']) + ensure + starter&.join + late_server&.close + end + end + + context 'when the server answers or times out' do + it 'does not retry an HTTP 500' do + open_pooled_connection + server.script(:server_error) + + expect { client.get('channels') }.to raise_error(StreamChat::StreamAPIException) + expect(server.requests.map(&:http_method)).to eq(['GET']) + end + + it 'does not retry a read timeout' do + slow_client = tls_client(server.url, timeout: 0.5) + server.script(:stall) + + expect { slow_client.get('channels') }.to raise_error(Faraday::TimeoutError) + expect(server.requests.map(&:http_method)).to eq(['GET']) + end + end +end diff --git a/spec/support/tls_test_server.rb b/spec/support/tls_test_server.rb new file mode 100644 index 0000000..b029745 --- /dev/null +++ b/spec/support/tls_test_server.rb @@ -0,0 +1,138 @@ +# frozen_string_literal: true + +require 'openssl' +require 'socket' + +# A local HTTPS server for transport-level tests. Each request it reads is +# handled by the next scripted action (default :respond): +# :respond - answer 200 and keep the connection open +# :respond_close - answer 200, then close the TCP socket without a TLS +# close_notify, leaving the client's pooled connection stale +# :drop - read the request, close the TCP socket without answering +# :server_error - answer 500 with a Stream error body +# :stall - read the request and never answer (client read timeout) +# Every request read is recorded in #requests, including dropped ones. +class TLSTestServer + Request = Struct.new(:http_method, :path, :headers, :body, keyword_init: true) + + attr_reader :port + + def initialize(port: 0) + @tcp = TCPServer.new('127.0.0.1', port) + @port = @tcp.addr[1] + @ssl = OpenSSL::SSL::SSLServer.new(@tcp, self.class.ssl_context) + @ssl.start_immediately = true + @actions = Queue.new + @requests = Queue.new + @threads = [] + @thread = Thread.new { accept_loop } + end + + def self.ssl_context + @ssl_context ||= begin + key = OpenSSL::PKey::RSA.new(2048) + cert = OpenSSL::X509::Certificate.new + cert.version = 2 + cert.serial = 1 + cert.subject = cert.issuer = OpenSSL::X509::Name.parse('/CN=localhost') + cert.public_key = key.public_key + cert.not_before = Time.now - 60 + cert.not_after = Time.now + 3600 + cert.sign(key, OpenSSL::Digest.new('SHA256')) + ctx = OpenSSL::SSL::SSLContext.new + ctx.cert = cert + ctx.key = key + ctx + end + end + + def url + "https://127.0.0.1:#{@port}" + end + + def script(*actions) + actions.each { |a| @actions << a } + end + + # Returns and clears the requests recorded so far. + def requests + out = [] + out << @requests.pop until @requests.empty? + out + end + + def close + @thread.kill + @threads.each(&:kill) + @tcp.close + end + + private + + def accept_loop + loop do + conn = begin + @ssl.accept + rescue OpenSSL::SSL::SSLError, SystemCallError, IOError + next + end + @threads << Thread.new(conn) { |c| serve(c) } + end + end + + def serve(conn) + loop do + request = read_request(conn) + break unless request + + @requests << request + break unless handle(conn, @actions.empty? ? :respond : @actions.pop) + end + rescue StandardError + nil + end + + # Returns whether the connection stays open for another request. + def handle(conn, action) + case action + when :drop + conn.to_io.close + false + when :stall + sleep 5 + false + when :server_error + write_response(conn, '500 Internal Server Error', '{"code":-1,"message":"boom","StatusCode":500}') + true + else + write_response(conn, '200 OK', '{"duration":"1ms"}') + return true unless action == :respond_close + + sleep 0.1 + conn.to_io.close + false + end + end + + def read_request(conn) + head = +'' + head << conn.readpartial(65_536) until head.include?("\r\n\r\n") + head, body = head.split("\r\n\r\n", 2) + lines = head.split("\r\n") + method, path = lines.shift.split + headers = lines.to_h do |l| + k, v = l.split(':', 2) + [k.downcase, v.strip] + end + length = headers['content-length'].to_i + body = body.to_s + body << conn.readpartial(65_536) while body.bytesize < length + Request.new(http_method: method, path: path, headers: headers, body: body) + rescue IOError, SystemCallError, OpenSSL::SSL::SSLError + nil + end + + def write_response(conn, status, body) + conn.write("HTTP/1.1 #{status}\r\nContent-Type: application/json\r\nContent-Length: #{body.bytesize}\r\n\r\n#{body}") + end +end From c5073687fbeb548139e775f829a147d2858ab24c Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Thu, 22 Oct 2026 16:29:21 +0200 Subject: [PATCH 2/2] test: make retry specs independent of OpenSSL EOF handling On Linux CI a dropped connection surfaces as Faraday::ConnectionFailed (EOFError) rather than Faraday::SSLError, and Net::HTTP may reconnect a stale pooled connection by itself. Accept either transport error, assert only the outcome for stale connections, and check the unrewindable-body rule with a refused connection, which fails the same way everywhere. Co-Authored-By: Claude Opus 5.5 --- spec/retry_middleware_spec.rb | 72 +++++++++++++++++++++++++++-------- 1 file changed, 57 insertions(+), 15 deletions(-) diff --git a/spec/retry_middleware_spec.rb b/spec/retry_middleware_spec.rb index cf97bc8..a984206 100644 --- a/spec/retry_middleware_spec.rb +++ b/spec/retry_middleware_spec.rb @@ -22,6 +22,10 @@ def tls_client(url, timeout: 3.0) # Opens a pooled keep-alive connection, then has the server close it without # a TLS close_notify, the way the load balancer front end drops connections. + # Depending on the OpenSSL build, the next request either fails in + # Net::HTTP#begin_transport (and the middleware retries it) or Net::HTTP sees + # a clean EOF and reconnects by itself. Either way the request must succeed + # and reach the server exactly once, which is what these specs assert. def make_pooled_connection_stale server.script(:respond_close) client.get('warmup') @@ -34,8 +38,21 @@ def open_pooled_connection server.requests end - context 'when the pooled connection went stale (request never sent)' do - it 'retries a POST, and the server receives it once' do + # How a dropped connection surfaces depends on the Ruby/OpenSSL build: + # Faraday::SSLError ("unexpected eof") on some, Faraday::ConnectionFailed + # (EOFError) on others. Callers get whichever the adapter raises. + def expect_transport_error(error) + expect(error).to be_a(Faraday::SSLError).or be_a(Faraday::ConnectionFailed) + end + + def closed_port + TCPServer.new('127.0.0.1', 0).then do |s| + s.addr[1].tap { s.close } + end + end + + context 'when the pooled connection went stale' do + it 'succeeds with a POST, and the server receives it once' do make_pooled_connection_stale response = client.post('messages', data: { message: { text: 'hi' } }) @@ -46,7 +63,7 @@ def open_pooled_connection expect(JSON.parse(requests.first.body)).to eq('message' => { 'text' => 'hi' }) end - it 'retries a GET, and the server receives it once' do + it 'succeeds with a GET, and the server receives it once' do make_pooled_connection_stale client.get('channels') @@ -54,7 +71,7 @@ def open_pooled_connection expect(server.requests.map(&:http_method)).to eq(['GET']) end - it 'retries a multipart upload with the full body' do + it 'succeeds with a multipart upload, and the server receives the full body once' do make_pooled_connection_stale content = "line of the uploaded file\n" * 4096 file = Tempfile.new('upload') @@ -74,23 +91,50 @@ def open_pooled_connection ensure file&.unlink end + end - it 'does not retry an IO body that cannot be rewound' do - make_pooled_connection_stale + context 'when the request body is an IO' do + # Connection refused fails before anything is sent, the same way on every + # platform, so it isolates the body check. A counting middleware placed + # after the retry middleware sees each attempt. + def refused_attempts(body) + refused = tls_client("https://127.0.0.1:#{closed_port}") + attempts = 0 + counter = Class.new(Faraday::Middleware) do + define_method(:call) do |env| + attempts += 1 + app.call(env) + end + end + refused.conn.builder.insert_after(StreamChat::RetryMiddleware, counter) + + expect do + refused.conn.post('messages', body, 'Content-Length' => '7', 'Content-Type' => 'application/json') + end.to raise_error(Faraday::ConnectionFailed, /connection refused/) + attempts + end + + it 'does not retry a body that cannot be rewound' do body = Class.new do + attr_reader :reads + def initialize(data) @io = StringIO.new(data) + @reads = 0 end def read(*args) + @reads += 1 @io.read(*args) end end.new('{"a":1}') - expect do - client.conn.post('messages', body, 'Content-Length' => '7', 'Content-Type' => 'application/json') - end.to raise_error(Faraday::SSLError) - expect(server.requests).to be_empty + expect(refused_attempts(body)).to eq(1) + expect(body.reads).to eq(0) + end + + it 'retries a body that can be rewound' do + expect(refused_attempts(StringIO.new('{"a":1}'))).to eq(3) end end @@ -109,7 +153,7 @@ def read(*args) server.script(:drop) expect { client.post('messages', data: { message: { text: 'hi' } }) } - .to raise_error(Faraday::SSLError, /unexpected eof/) + .to(raise_error { |e| expect_transport_error(e) }) expect(server.requests.map(&:http_method)).to eq(['POST']) end @@ -117,16 +161,14 @@ def read(*args) open_pooled_connection server.script(:drop, :drop, :drop, :drop) - expect { client.get('channels') }.to raise_error(Faraday::SSLError) + expect { client.get('channels') }.to(raise_error { |e| expect_transport_error(e) }) expect(server.requests.map(&:http_method)).to eq(%w[GET GET GET]) end end context 'when the connection is refused' do it 'retries a POST once the server is listening' do - port = TCPServer.new('127.0.0.1', 0).then do |s| - s.addr[1].tap { s.close } - end + port = closed_port late_client = tls_client("https://127.0.0.1:#{port}") late_server = nil starter = Thread.new do