fix: Fix FileDownloader and directory transfer failure paths (#3434)

Follow-up PR to #3407. 

Fixes `download_file` leaving truncated file at the destination when a
part fails with a non-StandardError, `download_file` leaving a stray
temp file behind when a caller-provided executor rejects a part
mid-download, and `upload_directory`/`download_directory` hanging
forever when listing files or objects fails with a non-StandardError
(such as from a `:filter_callback`).

---
By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license.

1. To make sure we include your contribution in the release notes,
please make sure to add description entry for your changes in the
"unreleased changes" section of the `CHANGELOG.md` file (at
corresponding gem). For the description entry, please make sure it lives
in one line and starts with `Feature` or `Issue` in the correct format.

2. For generated code changes, please checkout below instructions first:
  https://github.com/aws/aws-sdk-ruby/blob/version-3/CONTRIBUTING.md

Thank you for your contribution!

Author: richardwang1124

gems/aws-sdk-s3/CHANGELOG.md

@@ -1,6 +1,10 @@
 Unreleased Changes
 ------------------
 
+* Issue - Fix `download_file` on `TransferManager` and `Aws::S3::Object` leaving a corrupt file at the destination when a part raises a non-`StandardError`, and leaving a temp file behind when a caller-provided executor rejects a part.
+
+* Issue - Prevent `upload_directory` and `download_directory` on `TransferManager` from hanging when listing files or objects raises a non-`StandardError`, such as from a `:filter_callback`.
+
 1.233.1 (2026-10-01)
 ------------------
 

gems/aws-sdk-s3/lib/aws-sdk-s3/directory_downloader.rb

@@ -148,8 +148,9 @@ def each
             stream_objects
             @object_queue << DONE_MARKER
           rescue ClosedQueueError
-            # abort requested
-          rescue StandardError => e
+            nil # abort requested
+          # Any error, or the consumer waits on the queue forever
+          rescue Exception => e # rubocop:disable Lint/RescueException
             close
             raise e
           end

gems/aws-sdk-s3/lib/aws-sdk-s3/directory_uploader.rb

@@ -156,6 +156,10 @@ def each
             # encountered a traversal error, we must abort immediately
             close
             raise DirectoryUploadError, "Directory traversal failed for '#{@source_dir}': #{e.message}"
+          # Any other error, or the consumer waits on the queue forever
+          rescue Exception => e # rubocop:disable Lint/RescueException
+            close
+            raise e
           end
 
           while (file = @file_queue.shift) && file != DONE_MARKER

gems/aws-sdk-s3/lib/aws-sdk-s3/file_downloader.rb

@@ -60,7 +60,7 @@ def cleanup_temp_file(opts)
       end
 
       def download_with_executor(part_list, total_size, opts)
-        download_attempts = 0
+        queued_parts = 0
         completion_queue = Queue.new
         abort_download = false
         error = nil
@@ -69,24 +69,31 @@ def download_with_executor(part_list, total_size, opts)
         while (part = part_list.shift)
           break if abort_download
 
-          download_attempts += 1
-          @executor.post(part) do |p|
-            update_progress(progress, p)
-            resp = @client.get_object(p.params)
-            range = extract_range(resp.content_range)
-            validate_range(range, p.params[:range]) if p.params[:range]
-            write(resp.body, range, opts)
-
-            execute_checksum_callback(resp, opts)
+          begin
+            @executor.post(part) do |p|
+              update_progress(progress, p)
+              resp = @client.get_object(p.params)
+              range = extract_range(resp.content_range)
+              validate_range(range, p.params[:range]) if p.params[:range]
+              write(resp.body, range, opts)
+
+              execute_checksum_callback(resp, opts)
+            # Any error, or a corrupt file replaces the destination
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              abort_download = true
+              error = e
+            ensure
+              completion_queue << :done
+            end
+            queued_parts += 1
           rescue StandardError => e
-            abort_download = true
+            # Rejected by the executor, wait for queued parts so none write after cleanup
             error = e
-          ensure
-            completion_queue << :done
+            break
           end
         end
 
-        download_attempts.times { completion_queue.pop }
+        queued_parts.times { completion_queue.pop }
         raise error unless error.nil?
       end
 

gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_file_uploader.rb

@@ -169,7 +169,6 @@ def upload_with_executor(pending, completed, options)
             queued_parts += 1
           rescue StandardError => e
             # Rejected by the executor; abort rather than orphan the upload.
-            abort_upload = true
             errors << e
             break
           end

gems/aws-sdk-s3/spec/directory_downloader_spec.rb

@@ -131,6 +131,14 @@ module S3
 
             expect(result[:completed_downloads]).to eq(2)
           end
+
+          it 'raises instead of hanging when the callback raises a non-StandardError',
+             thread_report_on_exception: false do
+            filter = ->(obj) { obj.key == 'file2.json' ? raise(SystemStackError, 'filter failed') : true }
+            download = Thread.new { downloader.download(temp_dir, bucket: 'test-bucket', filter_callback: filter) }
+
+            expect { download.join(5) }.to raise_error(SystemStackError, 'filter failed')
+          end
         end
 
         context 'request callbacks' do

gems/aws-sdk-s3/spec/directory_uploader_spec.rb

@@ -128,6 +128,14 @@ module S3
             expect(uploaded_keys).not_to include('huge.bin')
             expect(result[:completed_uploads]).to eq(4)
           end
+
+          it 'raises instead of hanging when the callback raises a non-StandardError',
+             thread_report_on_exception: false do
+            filter_callback = ->(_path, file) { file == 'medium.log' ? raise(SystemStackError, 'filter failed') : true }
+            upload = Thread.new { uploader.upload(temp_dir, 'test-bucket', filter_callback: filter_callback) }
+
+            expect { upload.join(5) }.to raise_error(SystemStackError, 'filter failed')
+          end
         end
 
         context 'request callbacks', :jruby_flaky do

gems/aws-sdk-s3/spec/file_downloader_spec.rb

@@ -270,6 +270,53 @@ module S3
             expect(File.exist?(path)).to be(true)
             expect(File.read(path)).to eq('existing content')
           end
+
+          it 'does not overwrite existing file when a part raises a non-StandardError' do
+            # Held locally: `path` alone is a bare string, and the Tempfile backing it
+            # has no other reference, so it can be GC'd and unlinked before the assertion.
+            destination = Tempfile.new('destination')
+            path = destination.path
+            File.write(path, 'existing content')
+            client.stub_responses(:get_object, lambda { |context|
+              first, last = context.params[:range].scan(/\d+/).map(&:to_i)
+              raise NoMemoryError, 'part 2 failed' if first == 5 * one_meg
+
+              { body: 'x' * (last - first + 1), content_range: "bytes #{first}-#{last}/#{15 * one_meg}" }
+            })
+
+            expect { subject.download(path, range_params.merge(chunk_size: 5 * one_meg, mode: 'get_range')) }
+              .to raise_error(NoMemoryError, 'part 2 failed')
+            expect(File.read(path)).to eq('existing content')
+            expect(Dir.glob("#{path}.s3tmp.*")).to be_empty
+          end
+
+          it 'does not leave a temp file behind when the executor rejects a task mid-download' do
+            executor = DefaultExecutor.new
+            calls = 0
+            allow(executor).to receive(:post).and_wrap_original do |original, *args, &blk|
+              calls += 1
+              raise DefaultExecutor::RejectedExecutionError if calls == 2
+
+              original.call(*args, &blk)
+            end
+            downloader = FileDownloader.new(client: client, executor: executor)
+            written = Queue.new
+            allow(downloader).to receive(:write).and_wrap_original do |original, *args|
+              sleep(0.1)
+              original.call(*args)
+              written << :part
+            end
+            client.stub_responses(:get_object, lambda { |context|
+              first, last = context.params[:range].scan(/\d+/).map(&:to_i)
+              { body: 'x' * (last - first + 1), content_range: "bytes #{first}-#{last}/#{15 * one_meg}" }
+            })
+
+            expect { downloader.download(path, range_params.merge(chunk_size: 5 * one_meg, mode: 'get_range')) }
+              .to raise_error(DefaultExecutor::RejectedExecutionError)
+            expect(written.size).to eq(1)
+            expect(Dir.glob("#{path}.s3tmp.*")).to be_empty
+            executor.shutdown
+          end
         end
       end
     end