fix: Address unbounded memory growth in upload_stream when source outpaces upload - #3407

Fixes #3393.

Description

upload_stream held memory proportional to the whole object when the source wrote faster than parts uploaded. Since #3302, the reader offloads each part into DefaultExecutor, whose unbounded queue let it read ahead of the upload threads without limit (#3393).

• Streaming a fast local source to a slower sink buffered almost the entire object. In a repro, peak RSS was 1045 MB for a 1 GB source and 2078 MB for 2 GB.
• IO.copy_stream into a StringIO grows the buffer by doubling, so each 5 MB part held an 8 MB string and fragmented the heap.
• A part task that raised a non-StandardError (NoMemoryError, SystemStackError, LoadError) killed its worker and was left out of the completed parts. On real S3, upload_stream returned normally and committed a 15 MB object missing part 2 of a 20 MB source. upload_file was protected, because S3 rejects its incomplete completion through mpu_object_size.
• If every worker died that way, queued tasks never ran and upload_stream and ranged downloads hung.
• A task rejected by a shared executor shut down mid-upload hung upload_stream and left upload_file's multipart upload in progress with no abort.

Fix

DefaultExecutor accepts an optional max_queue, and upload_stream sets it to thread_count, so post blocks while the queue is full.

# before
executor = DefaultExecutor.new(max_threads: thread_count)
IO.copy_stream(read_pipe, StringIO.new(String.new), @part_size)

# after
executor = DefaultExecutor.new(max_threads: thread_count, max_queue: thread_count)
StringIO.new(read_pipe.read(@part_size))

• Buffered data is capped at about 2 * thread_count + 1 parts. Peak RSS is about 110 MB regardless of source size, with no change in elapsed time. The tempfile path keeps copy_stream.
• shutdown and kill close the queue instead of pushing sentinels, and post pushes outside the mutex, so a producer waiting on a full queue can't delay them. Without this, shutdown(1) and kill each took 10 s with a bounded queue.
• Rejected posts raise DefaultExecutor::RejectedExecutionError, a RuntimeError subclass, and both uploaders abort the multipart upload when a post is rejected.
• Uploader part tasks treat any exception as a failed part, so the upload aborts with MultipartUploadError instead of completing without the part.
• A worker killed by its task starts a replacement before exiting. shutdown keeps joining until the pool is empty, ignores errors re-raised by joining a dead worker, then re-raises the first fatal error once all tasks finish.
• Caller-provided executors are still unbounded, which is now documented on :thread_count. The same FileDownloader issues (a zero-filled hole at the destination on a fatal part error, and rejected-post handling) predate this PR and will be fixed in a follow-up.

Testing

aws-sdk-s3 1681 examples, 0 failures. That includes default_executor 12, multipart_stream_uploader 19, multipart_file_uploader 11 and object/upload_stream 5. Each new spec fails with its fix reverted. Repro scripts were run against version-3 and this branch for memory, rejected posts, worker death and non-StandardError part failures. The missing-part case was also checked on real S3.

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!