Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions gems/aws-sdk-core/lib/seahorse/client/net_http/connection_pool.rb
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,17 @@ def session_for(endpoint, &block)
session.continue_timeout = http_continue_timeout if
session.respond_to?(:continue_timeout=)
yield(session)
rescue Net::OpenTimeout, Timeout::Error => error
session.finish if session
# For timeout errors, clear the entire pool for this endpoint
# to force fresh connections on subsequent requests
@pool_mutex.synchronize do
if @pool.key?(endpoint)
@pool[endpoint].each(&:finish)
@pool[endpoint].clear
end
end
raise
rescue
session.finish if session
raise
Expand Down Expand Up @@ -147,6 +158,22 @@ def empty!
nil
end

# Closes and removes all sessions for a specific endpoint from the pool.
# This is useful for clearing potentially stale connections after
# timeout errors.
# @param [URI::HTTP, URI::HTTPS] endpoint The endpoint to clear
# @return [nil]
def clear_endpoint!(endpoint)
endpoint = remove_path_and_query(endpoint)
@pool_mutex.synchronize do
if @pool.key?(endpoint)
@pool[endpoint].each(&:finish)
@pool[endpoint].clear
end
end
nil
end

private

def remove_path_and_query(endpoint)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,107 @@ module NetHttp
expect(first_pool).to eq second_pool
end
end

describe "#session_for" do
let(:pool) { described_class.new }
let(:endpoint) { URI.parse('http://example.com') }

describe "timeout error handling" do
it "clears endpoint pool on Net::OpenTimeout" do
mock_session = double('session')
allow(pool).to receive(:start_session).and_return(mock_session)
allow(mock_session).to receive(:read_timeout=)
allow(mock_session).to receive(:continue_timeout=)
allow(mock_session).to receive(:respond_to?).with(:continue_timeout=).and_return(false)
allow(mock_session).to receive(:finish)

# First call succeeds and session gets added to pool
pool.session_for(endpoint) { |s| }
expect(pool.size).to eq(1)

# Second call raises Net::OpenTimeout
expect(mock_session).to receive(:finish)
expect do
pool.session_for(endpoint) { |s| raise Net::OpenTimeout.new }
end.to raise_error(Net::OpenTimeout)

# Pool should be cleared for this endpoint
expect(pool.size).to eq(0)
end

it "clears endpoint pool on Timeout::Error" do
mock_session = double('session')
allow(pool).to receive(:start_session).and_return(mock_session)
allow(mock_session).to receive(:read_timeout=)
allow(mock_session).to receive(:continue_timeout=)
allow(mock_session).to receive(:respond_to?).with(:continue_timeout=).and_return(false)
allow(mock_session).to receive(:finish)

# First call succeeds and session gets added to pool
pool.session_for(endpoint) { |s| }
expect(pool.size).to eq(1)

# Second call raises Timeout::Error
expect(mock_session).to receive(:finish)
expect do
pool.session_for(endpoint) { |s| raise Timeout::Error.new }
end.to raise_error(Timeout::Error)

# Pool should be cleared for this endpoint
expect(pool.size).to eq(0)
end

it "does not clear pool for other errors" do
mock_session = double('session')
allow(pool).to receive(:start_session).and_return(mock_session)
allow(mock_session).to receive(:read_timeout=)
allow(mock_session).to receive(:continue_timeout=)
allow(mock_session).to receive(:respond_to?).with(:continue_timeout=).and_return(false)
allow(mock_session).to receive(:finish)

# First call succeeds and session gets added to pool
pool.session_for(endpoint) { |s| }
expect(pool.size).to eq(1)

# Second call raises different error
expect(mock_session).to receive(:finish)
expect do
pool.session_for(endpoint) { |s| raise SocketError.new }
end.to raise_error(SocketError)

# Pool should still have the session since it wasn't a timeout error
expect(pool.size).to eq(1)
end
end
end

describe "#clear_endpoint!" do
let(:pool) { described_class.new }
let(:endpoint) { URI.parse('http://example.com') }

it "clears sessions for a specific endpoint" do
mock_session = double('session')
allow(pool).to receive(:start_session).and_return(mock_session)
allow(mock_session).to receive(:read_timeout=)
allow(mock_session).to receive(:continue_timeout=)
allow(mock_session).to receive(:respond_to?).with(:continue_timeout=).and_return(false)
allow(mock_session).to receive(:finish)

# Add session to pool
pool.session_for(endpoint) { |s| }
expect(pool.size).to eq(1)

# Clear the endpoint
expect(mock_session).to receive(:finish)
pool.clear_endpoint!(endpoint)
expect(pool.size).to eq(0)
end

it "handles non-existent endpoints gracefully" do
endpoint = URI.parse('http://nonexistent.com')
expect { pool.clear_endpoint!(endpoint) }.not_to raise_error
end
end
end
end
end
Expand Down