diff --git a/gems/aws-sdk-s3/CHANGELOG.md b/gems/aws-sdk-s3/CHANGELOG.md index 129db9dbd12..b9ab65508bb 100644 --- a/gems/aws-sdk-s3/CHANGELOG.md +++ b/gems/aws-sdk-s3/CHANGELOG.md @@ -1,6 +1,8 @@ Unreleased Changes ------------------ +* Issue - Ensure the internally-created executor is shutdown on error in `TransferManager` and `Aws::S3::Object` resource methods, preventing leaked worker threads on multipart transfer failures (#3419). + 1.232.1 (2026-09-16) ------------------ diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/customizations/object.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/customizations/object.rb index dd2cb036edf..5febedae568 100644 --- a/gems/aws-sdk-s3/lib/aws-sdk-s3/customizations/object.rb +++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/customizations/object.rb @@ -406,17 +406,20 @@ def public_url(options = {}) def upload_stream(options = {}, &block) upload_opts = options.merge(bucket: bucket_name, key: key) executor = DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count)) - uploader = MultipartStreamUploader.new( - client: client, - executor: executor, - tempfile: upload_opts.delete(:tempfile), - part_size: upload_opts.delete(:part_size) - ) - Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do - uploader.upload(upload_opts, &block) + begin + uploader = MultipartStreamUploader.new( + client: client, + executor: executor, + tempfile: upload_opts.delete(:tempfile), + part_size: upload_opts.delete(:part_size) + ) + Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do + uploader.upload(upload_opts, &block) + end + true + ensure + executor.shutdown end - executor.shutdown - true end deprecated(:upload_stream, use: 'Aws::S3::TransferManager#upload_stream', version: 'next major version') @@ -480,17 +483,20 @@ def upload_stream(options = {}, &block) def upload_file(source, options = {}) upload_opts = options.merge(bucket: bucket_name, key: key) executor = DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count)) - uploader = FileUploader.new( - client: client, - executor: executor, - multipart_threshold: upload_opts.delete(:multipart_threshold) - ) - response = Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do - uploader.upload(source, upload_opts) + begin + uploader = FileUploader.new( + client: client, + executor: executor, + multipart_threshold: upload_opts.delete(:multipart_threshold) + ) + response = Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do + uploader.upload(source, upload_opts) + end + yield response if block_given? + true + ensure + executor.shutdown end - yield response if block_given? - executor.shutdown - true end deprecated(:upload_file, use: 'Aws::S3::TransferManager#upload_file', version: 'next major version') @@ -562,12 +568,15 @@ def upload_file(source, options = {}) def download_file(destination, options = {}) download_opts = options.merge(bucket: bucket_name, key: key) executor = DefaultExecutor.new(max_threads: download_opts.delete([:thread_count])) - downloader = FileDownloader.new(client: client, executor: executor) - Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do - downloader.download(destination, download_opts) + begin + downloader = FileDownloader.new(client: client, executor: executor) + Aws::Plugins::UserAgent.metric('RESOURCE_MODEL') do + downloader.download(destination, download_opts) + end + true + ensure + executor.shutdown end - executor.shutdown - true end deprecated(:download_file, use: 'Aws::S3::TransferManager#download_file', version: 'next major version') diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/transfer_manager.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/transfer_manager.rb index 864dd543c14..7668bfdde0e 100644 --- a/gems/aws-sdk-s3/lib/aws-sdk-s3/transfer_manager.rb +++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/transfer_manager.rb @@ -161,10 +161,12 @@ def initialize(options = {}) def download_directory(destination, bucket:, **options) Aws::Plugins::UserAgent.metric('S3_TRANSFER', 'S3_TRANSFER_DOWNLOAD_DIRECTORY') do executor = @executor || DefaultExecutor.new(max_threads: options.delete(:thread_count)) - downloader = DirectoryDownloader.new(client: @client, executor: executor, logger: @logger) - result = downloader.download(destination, bucket: bucket, **options) - executor.shutdown unless @executor - result + begin + downloader = DirectoryDownloader.new(client: @client, executor: executor, logger: @logger) + downloader.download(destination, bucket: bucket, **options) + ensure + executor.shutdown unless @executor + end end end @@ -246,10 +248,13 @@ def download_directory(destination, bucket:, **options) def download_file(destination, bucket:, key:, **options) download_opts = options.merge(bucket: bucket, key: key) executor = @executor || DefaultExecutor.new(max_threads: download_opts.delete(:thread_count)) - downloader = FileDownloader.new(client: @client, executor: executor) - downloader.download(destination, download_opts) - executor.shutdown unless @executor - true + begin + downloader = FileDownloader.new(client: @client, executor: executor) + downloader.download(destination, download_opts) + true + ensure + executor.shutdown unless @executor + end end # Uploads all files under the given directory to the provided S3 bucket. @@ -361,10 +366,12 @@ def download_file(destination, bucket:, key:, **options) def upload_directory(source, bucket:, **options) Aws::Plugins::UserAgent.metric('S3_TRANSFER', 'S3_TRANSFER_UPLOAD_DIRECTORY') do executor = @executor || DefaultExecutor.new(max_threads: options.delete(:thread_count)) - uploader = DirectoryUploader.new(client: @client, executor: executor, logger: @logger) - result = uploader.upload(source, bucket, **options.merge(http_chunk_size: resolve_http_chunk_size(options))) - executor.shutdown unless @executor - result + begin + uploader = DirectoryUploader.new(client: @client, executor: executor, logger: @logger) + uploader.upload(source, bucket, **options.merge(http_chunk_size: resolve_http_chunk_size(options))) + ensure + executor.shutdown unless @executor + end end end @@ -445,16 +452,19 @@ def upload_file(source, bucket:, key:, **options) http_chunk_size = resolve_http_chunk_size(upload_opts) executor = @executor || DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count)) - uploader = FileUploader.new( - multipart_threshold: upload_opts.delete(:multipart_threshold), - http_chunk_size: http_chunk_size, - client: @client, - executor: executor - ) - response = uploader.upload(source, upload_opts) - yield response if block_given? - executor.shutdown unless @executor - true + begin + uploader = FileUploader.new( + multipart_threshold: upload_opts.delete(:multipart_threshold), + http_chunk_size: http_chunk_size, + client: @client, + executor: executor + ) + response = uploader.upload(source, upload_opts) + yield response if block_given? + true + ensure + executor.shutdown unless @executor + end end # Uploads a stream in a streaming fashion to S3. @@ -512,15 +522,18 @@ def upload_file(source, bucket:, key:, **options) def upload_stream(bucket:, key:, **options, &block) upload_opts = options.merge(bucket: bucket, key: key) executor = @executor || DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count)) - uploader = MultipartStreamUploader.new( - client: @client, - executor: executor, - tempfile: upload_opts.delete(:tempfile), - part_size: upload_opts.delete(:part_size) - ) - uploader.upload(upload_opts, &block) - executor.shutdown unless @executor - true + begin + uploader = MultipartStreamUploader.new( + client: @client, + executor: executor, + tempfile: upload_opts.delete(:tempfile), + part_size: upload_opts.delete(:part_size) + ) + uploader.upload(upload_opts, &block) + true + ensure + executor.shutdown unless @executor + end end private diff --git a/gems/aws-sdk-s3/spec/object/download_file_spec.rb b/gems/aws-sdk-s3/spec/object/download_file_spec.rb index 455a51a5113..dbac0baa67d 100644 --- a/gems/aws-sdk-s3/spec/object/download_file_spec.rb +++ b/gems/aws-sdk-s3/spec/object/download_file_spec.rb @@ -28,6 +28,18 @@ module S3 expect { subject.download_file(path) }.to raise_error(Aws::S3::Errors::NoSuchKey) end + it 'shuts down the internally-created executor when download raises', thread_report_on_exception: false do + client.stub_responses(:head_object, 'NoSuchKey') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect { subject.download_file(path) }.to raise_error(Aws::S3::Errors::NoSuchKey) + expect(executor).to have_received(:shutdown) + end + it 'calls progress callback when given' do n_calls = 0 callback = proc { |_b, _p, _t| n_calls += 1 } diff --git a/gems/aws-sdk-s3/spec/object/upload_file_spec.rb b/gems/aws-sdk-s3/spec/object/upload_file_spec.rb index 81fad4efd23..4e12cf61d5b 100644 --- a/gems/aws-sdk-s3/spec/object/upload_file_spec.rb +++ b/gems/aws-sdk-s3/spec/object/upload_file_spec.rb @@ -36,6 +36,18 @@ module S3 expect { subject.upload_file(file) }.to raise_error(Aws::S3::Errors::AccessDenied) end + it 'shuts down the internally-created executor when upload raises' do + client.stub_responses(:put_object, 'AccessDenied') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect { subject.upload_file(file) }.to raise_error(Aws::S3::Errors::AccessDenied) + expect(executor).to have_received(:shutdown) + end + it 'yields the response to the given block' do subject.upload_file(file) do |response| expect(response).to be_kind_of(Seahorse::Client::Response) diff --git a/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb b/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb index 8f2a5739813..33b1d2635c2 100644 --- a/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb +++ b/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb @@ -25,6 +25,20 @@ module S3 end.to raise_error(Aws::S3::MultipartUploadError, /part failed/) end + it 'shuts down the internally-created executor when upload raises' do + client.stub_responses(:upload_part, RuntimeError.new('part failed')) + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect do + subject.upload_stream { |write_stream| write_stream << seventeen_mb } + end.to raise_error(Aws::S3::MultipartUploadError, /part failed/) + expect(executor).to have_received(:shutdown) + end + it 'respects the thread_count option' do custom_thread_count = 20 client.stub_responses(:create_multipart_upload, upload_id: 'id') diff --git a/gems/aws-sdk-s3/spec/transfer_manager_spec.rb b/gems/aws-sdk-s3/spec/transfer_manager_spec.rb index d938428e2c2..036fb3725e9 100644 --- a/gems/aws-sdk-s3/spec/transfer_manager_spec.rb +++ b/gems/aws-sdk-s3/spec/transfer_manager_spec.rb @@ -51,6 +51,23 @@ module S3 subject.download_directory(temp_dir, bucket: 'bucket', ignore_failure: false) end.to raise_error(DirectoryDownloadError) end + + it 'shuts down the internally-created executor when download raises' do + client.stub_responses(:get_object, 'AccessDenied') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + instance = orig.call(*args, **kwargs) + if executor.nil? + executor = instance + allow(instance).to receive(:shutdown).and_call_original + end + instance + end + expect do + subject.download_directory(temp_dir, bucket: 'bucket', ignore_failure: false) + end.to raise_error(DirectoryDownloadError) + expect(executor).to have_received(:shutdown) + end end describe '#download_file' do @@ -72,6 +89,19 @@ module S3 .to raise_error(Aws::S3::Errors::NoSuchKey) end + it 'shuts down the internally-created executor when download raises' do + client.stub_responses(:head_object, 'NoSuchKey') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect { subject.download_file(path, bucket: 'bucket', key: 'missing-key') } + .to raise_error(Aws::S3::Errors::NoSuchKey) + expect(executor).to have_received(:shutdown) + end + it 'calls progress callback when given' do n_calls = 0 callback = proc { |_b, _p, _t| n_calls += 1 } @@ -108,6 +138,23 @@ module S3 subject.upload_directory(temp_dir, bucket: 'bucket', ignore_failure: false) end.to raise_error(DirectoryUploadError) end + + it 'shuts down the internally-created executor when upload raises' do + client.stub_responses(:put_object, 'AccessDenied') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + instance = orig.call(*args, **kwargs) + if executor.nil? + executor = instance + allow(instance).to receive(:shutdown).and_call_original + end + instance + end + expect do + subject.upload_directory(temp_dir, bucket: 'bucket', ignore_failure: false) + end.to raise_error(DirectoryUploadError) + expect(executor).to have_received(:shutdown) + end end describe '#upload_file' do @@ -135,6 +182,19 @@ module S3 .to raise_error(Aws::S3::Errors::AccessDenied) end + it 'shuts down the internally-created executor when upload raises' do + client.stub_responses(:put_object, 'AccessDenied') + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect { subject.upload_file(file, bucket: 'forbidden-bucket', key: 'key') } + .to raise_error(Aws::S3::Errors::AccessDenied) + expect(executor).to have_received(:shutdown) + end + it 'yields response when block given' do subject.upload_file(file, bucket: 'bucket', key: 'key') do |response| expect(response).to be_kind_of(Seahorse::Client::Response) @@ -243,6 +303,20 @@ module S3 subject.upload_stream(bucket: 'bucket', key: 'key') { |write_stream| write_stream << seventeen_mb } end.to raise_error(Aws::S3::MultipartUploadError, /part failed/) end + + it 'shuts down the internally-created executor when upload raises' do + client.stub_responses(:upload_part, RuntimeError.new('part failed')) + executor = nil + allow(DefaultExecutor).to receive(:new).and_wrap_original do |orig, *args, **kwargs| + executor = orig.call(*args, **kwargs) + allow(executor).to receive(:shutdown).and_call_original + executor + end + expect do + subject.upload_stream(bucket: 'bucket', key: 'key') { |write_stream| write_stream << seventeen_mb } + end.to raise_error(Aws::S3::MultipartUploadError, /part failed/) + expect(executor).to have_received(:shutdown) + end end end end