Shutdown internally-created executor on error in TransferManager

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)
 ------------------
 

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')
 

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

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 }

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)

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')

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