From 80808215f75713292c908bbfc052aa53fd25ce80 Mon Sep 17 00:00:00 2001
From: Richard Wang <wangrch@amazon.com>
Date: Tue, 7 Jul 2026 13:37:53 -0700
Subject: [PATCH 1/5] Add fix for upload stream bug

---
 .../lib/aws-sdk-s3/default_executor.rb        |  4 +-
 .../aws-sdk-s3/multipart_stream_uploader.rb   | 19 +++--
 .../spec/multipart_stream_uploader_spec.rb    | 77 +++++++++++++++++++
 3 files changed, 90 insertions(+), 10 deletions(-)

diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
index 13a719f4397..f57c583a379 100644
--- a/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
+++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
@@ -12,7 +12,7 @@ class DefaultExecutor
       def initialize(options = {})
         @max_threads = options[:max_threads] || DEFAULT_MAX_THREADS
         @state = RUNNING
-        @queue = Queue.new
+        @queue = SizedQueue.new(@max_threads)
         @pool = []
         @mutex = Mutex.new
       end
@@ -25,9 +25,9 @@ def post(*args, &block)
         @mutex.synchronize do
           raise 'Executor has been shutdown and is no longer accepting tasks' unless @state == RUNNING
 
-          @queue << [args, block]
           ensure_worker_available
         end
+        @queue << [args, block]
         true
       end
 
diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
index ae6f75a47a1..dd025c01fc7 100644
--- a/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
+++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
@@ -113,18 +113,21 @@ def complete_opts(options)
       def read_to_part_body(read_pipe)
         return if read_pipe.closed?
 
-        temp_io = @tempfile ? Tempfile.new('aws-sdk-s3-upload_stream') : StringIO.new(String.new)
-        temp_io.binmode
-        bytes_copied = IO.copy_stream(read_pipe, temp_io, @part_size)
-        temp_io.rewind
-        if bytes_copied.zero?
-          if temp_io.is_a?(Tempfile)
+        if @tempfile
+          temp_io = Tempfile.new('aws-sdk-s3-upload_stream')
+          temp_io.binmode
+          bytes_copied = IO.copy_stream(read_pipe, temp_io, @part_size)
+          temp_io.rewind
+          if bytes_copied.zero?
             temp_io.close
             temp_io.unlink
+            nil
+          else
+            temp_io
           end
-          nil
         else
-          temp_io
+          data = read_pipe.read(@part_size)
+          data.nil? || data.empty? ? nil : StringIO.new(data)
         end
       end
 
diff --git a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
index 6262a42b560..a1e778a124a 100644
--- a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
+++ b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
@@ -22,6 +22,83 @@ module S3
         end
       end
 
+      describe '#upload_stream memory bounds', :jruby_flaky do
+        it 'bounds queued parts to the thread count when source outpaces upload' do
+          num_threads = 4
+          part_size = 1024 * 1024 # 1 MB parts for faster test
+          total_parts = 20
+          total_data = part_size * total_parts
+
+          client.stub_responses(:create_multipart_upload, upload_id: 'id')
+          client.stub_responses(:complete_multipart_upload)
+
+          mutex = Mutex.new
+          queue_depth_samples = []
+
+          executor = DefaultExecutor.new(max_threads: num_threads)
+
+          # Simulate slow uploads so parts queue up faster than they drain
+          allow(client).to receive(:upload_part) do |_part|
+            mutex.synchronize do
+              queue_depth_samples << executor.instance_variable_get(:@queue).size
+            end
+            sleep(0.05)
+            double(:upload_part, etag: 'etag')
+          end
+
+          uploader = MultipartStreamUploader.new(
+            client: client,
+            executor: executor,
+            part_size: part_size
+          )
+
+          uploader.upload(params) do |write_stream|
+            write_stream << ('a' * total_data)
+          end
+
+          peak_queue_depth = queue_depth_samples.max || 0
+
+          # With backpressure (SizedQueue), the reader blocks when the
+          # queue is full, so depth never exceeds the thread count.
+          expect(peak_queue_depth).to be <= num_threads,
+            "Expected peak queue depth (#{peak_queue_depth}) to be at most " \
+            "num_threads (#{num_threads}), but the queue grew unbounded."
+        end
+
+        it 'completes all parts under backpressure' do
+          num_threads = 2
+          part_size = 1024 * 1024 # 1 MB
+          total_parts = 10
+          total_data = part_size * total_parts
+
+          client.stub_responses(:create_multipart_upload, upload_id: 'id')
+          client.stub_responses(:complete_multipart_upload)
+
+          mutex = Mutex.new
+          uploaded_parts = []
+
+          executor = DefaultExecutor.new(max_threads: num_threads)
+
+          allow(client).to receive(:upload_part) do |part|
+            sleep(0.05) # slow upload
+            mutex.synchronize { uploaded_parts << part[:part_number] }
+            double(:upload_part, etag: 'etag')
+          end
+
+          uploader = MultipartStreamUploader.new(
+            client: client,
+            executor: executor,
+            part_size: part_size
+          )
+
+          uploader.upload(params) do |write_stream|
+            write_stream << ('a' * total_data)
+          end
+
+          expect(uploaded_parts.sort).to eq((1..total_parts).to_a)
+        end
+      end
+
       describe '#upload_stream', :jruby_flaky do
         it 'can upload empty stream' do
           client.stub_responses(:create_multipart_upload, upload_id: 'id')

From 842f7e7d2e42312c62ed3fb4fa16b0fb800ae2ef Mon Sep 17 00:00:00 2001
From: Richard Wang <wangrch@amazon.com>
Date: Tue, 7 Jul 2026 14:50:41 -0700
Subject: [PATCH 2/5] Add repro script

---
 repro_upload_stream.rb | 123 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 123 insertions(+)
 create mode 100755 repro_upload_stream.rb

diff --git a/repro_upload_stream.rb b/repro_upload_stream.rb
new file mode 100755
index 00000000000..0872fee8d53
--- /dev/null
+++ b/repro_upload_stream.rb
@@ -0,0 +1,123 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+# Reproduction for https://github.com/aws/aws-sdk-ruby/issues/3393
+# Object#upload_stream retains unbounded memory when the source outpaces the upload.
+#
+# This script demonstrates the bug and verifies the fix by:
+# 1. Simulating a fast writer (200 MB) feeding a slow S3 upload (0.1s per 5 MB part)
+# 2. Measuring peak memory (RSS) during the upload
+#
+# BUG behavior (unbounded Queue):  ~200 MB memory growth (all parts buffered)
+# FIX behavior (SizedQueue):       ~30-50 MB memory growth (bounded to thread count)
+#
+# Usage: ruby repro_upload_stream.rb
+#   Run from the aws-sdk-ruby repo root on the fix/upload_stream branch.
+
+$LOAD_PATH.unshift(File.expand_path('gems/aws-sdk-s3/lib', __dir__))
+$LOAD_PATH.unshift(File.expand_path('gems/aws-sdk-core/lib', __dir__))
+$LOAD_PATH.unshift(File.expand_path('gems/aws-sigv4/lib', __dir__))
+$LOAD_PATH.unshift(File.expand_path('gems/aws-partitions/lib', __dir__))
+
+require 'aws-sdk-s3'
+
+PART_SIZE = 5 * 1024 * 1024  # 5 MB (S3 default)
+NUM_PARTS = 40               # 200 MB total
+TOTAL_DATA = PART_SIZE * NUM_PARTS
+MAX_THREADS = 4
+SIMULATED_UPLOAD_DELAY = 0.1 # seconds per part (simulates slow network)
+
+def rss_mb
+  `ps -o rss= -p #{Process.pid}`.strip.to_i / 1024.0
+end
+
+puts "=== Reproduction: upload_stream unbounded memory (issue #3393) ==="
+puts
+puts "Configuration:"
+puts "  Part size:       #{PART_SIZE / (1024*1024)} MB"
+puts "  Total parts:     #{NUM_PARTS}"
+puts "  Total data:      #{TOTAL_DATA / (1024*1024)} MB"
+puts "  Max threads:     #{MAX_THREADS}"
+puts "  Upload delay:    #{SIMULATED_UPLOAD_DELAY}s per part"
+puts
+
+# Verify source and queue type
+source_file = Aws::S3::DefaultExecutor.instance_method(:initialize).source_location[0]
+puts "DefaultExecutor source: #{source_file}"
+test_executor = Aws::S3::DefaultExecutor.new(max_threads: MAX_THREADS)
+queue = test_executor.instance_variable_get(:@queue)
+puts "Queue type: #{queue.class} (max: #{queue.respond_to?(:max) ? queue.max : 'unbounded'})"
+test_executor.shutdown
+puts
+
+# Set up stubbed client (no real AWS calls)
+client = Aws::S3::Client.new(region: 'us-east-1', stub_responses: true)
+client.stub_responses(:create_multipart_upload, { upload_id: 'test-upload-id' })
+client.stub_responses(:upload_part, ->(context) {
+  sleep(SIMULATED_UPLOAD_DELAY)
+  { etag: "\"etag-#{context.params[:part_number]}\"" }
+})
+client.stub_responses(:complete_multipart_upload, {
+  location: 'https://bucket.s3.amazonaws.com/key',
+  bucket: 'test-bucket',
+  key: 'test-key',
+  etag: '"final-etag"'
+})
+
+tm = Aws::S3::TransferManager.new(client: client)
+
+baseline_rss = rss_mb
+peak_rss = baseline_rss
+
+sampler = Thread.new do
+  loop do
+    current = rss_mb
+    peak_rss = current if current > peak_rss
+    sleep(0.01)
+  end
+end
+
+puts "Baseline RSS: #{baseline_rss.round(1)} MB"
+puts "Uploading #{TOTAL_DATA / (1024*1024)} MB with #{MAX_THREADS} threads..."
+start_time = Time.now
+
+tm.upload_stream(
+  bucket: 'test-bucket',
+  key: 'test-key',
+  part_size: PART_SIZE,
+  thread_count: MAX_THREADS
+) do |write_stream|
+  chunk = 'x' * (1024 * 1024) # write in 1 MB chunks (fast local source)
+  bytes_written = 0
+  while bytes_written < TOTAL_DATA
+    write_stream.write(chunk)
+    bytes_written += chunk.size
+  end
+end
+
+elapsed = Time.now - start_time
+sampler.kill
+
+memory_growth = peak_rss - baseline_rss
+
+puts "Done in #{elapsed.round(2)}s"
+puts
+puts "=== Memory Results ==="
+puts "  Peak RSS:        #{peak_rss.round(1)} MB"
+puts "  Memory growth:   #{memory_growth.round(1)} MB"
+puts
+
+# With backpressure (SizedQueue), expect bounded memory:
+#   ~max_threads * part_size + pipe buffer + overhead ≈ 30-50 MB
+# Without backpressure (Queue), all 40 parts buffer in memory:
+#   ~40 * 5 MB = 200 MB growth
+bounded_limit_mb = (MAX_THREADS + 2) * (PART_SIZE / (1024.0 * 1024)) + 50
+
+if memory_growth < bounded_limit_mb
+  puts "PASS: Memory is bounded (#{memory_growth.round(1)} MB < #{bounded_limit_mb.round(0)} MB limit)"
+  puts "      The SizedQueue backpressure is working correctly."
+else
+  puts "FAIL: Memory grew unbounded (#{memory_growth.round(1)} MB > #{bounded_limit_mb.round(0)} MB limit)"
+  puts "      All #{NUM_PARTS} parts (#{NUM_PARTS * PART_SIZE / (1024*1024)} MB) were buffered in the queue."
+  puts "      This confirms bug #3393."
+end

From 906c099edd839ed23acbb093842a38434c3671f9 Mon Sep 17 00:00:00 2001
From: Richard Wang <wangrch@amazon.com>
Date: Wed, 8 Jul 2026 13:44:18 -0700
Subject: [PATCH 3/5] Remove repro script

---
 repro_upload_stream.rb | 123 -----------------------------------------
 1 file changed, 123 deletions(-)
 delete mode 100755 repro_upload_stream.rb

diff --git a/repro_upload_stream.rb b/repro_upload_stream.rb
deleted file mode 100755
index 0872fee8d53..00000000000
--- a/repro_upload_stream.rb
+++ /dev/null
@@ -1,123 +0,0 @@
-#!/usr/bin/env ruby
-# frozen_string_literal: true
-
-# Reproduction for https://github.com/aws/aws-sdk-ruby/issues/3393
-# Object#upload_stream retains unbounded memory when the source outpaces the upload.
-#
-# This script demonstrates the bug and verifies the fix by:
-# 1. Simulating a fast writer (200 MB) feeding a slow S3 upload (0.1s per 5 MB part)
-# 2. Measuring peak memory (RSS) during the upload
-#
-# BUG behavior (unbounded Queue):  ~200 MB memory growth (all parts buffered)
-# FIX behavior (SizedQueue):       ~30-50 MB memory growth (bounded to thread count)
-#
-# Usage: ruby repro_upload_stream.rb
-#   Run from the aws-sdk-ruby repo root on the fix/upload_stream branch.
-
-$LOAD_PATH.unshift(File.expand_path('gems/aws-sdk-s3/lib', __dir__))
-$LOAD_PATH.unshift(File.expand_path('gems/aws-sdk-core/lib', __dir__))
-$LOAD_PATH.unshift(File.expand_path('gems/aws-sigv4/lib', __dir__))
-$LOAD_PATH.unshift(File.expand_path('gems/aws-partitions/lib', __dir__))
-
-require 'aws-sdk-s3'
-
-PART_SIZE = 5 * 1024 * 1024  # 5 MB (S3 default)
-NUM_PARTS = 40               # 200 MB total
-TOTAL_DATA = PART_SIZE * NUM_PARTS
-MAX_THREADS = 4
-SIMULATED_UPLOAD_DELAY = 0.1 # seconds per part (simulates slow network)
-
-def rss_mb
-  `ps -o rss= -p #{Process.pid}`.strip.to_i / 1024.0
-end
-
-puts "=== Reproduction: upload_stream unbounded memory (issue #3393) ==="
-puts
-puts "Configuration:"
-puts "  Part size:       #{PART_SIZE / (1024*1024)} MB"
-puts "  Total parts:     #{NUM_PARTS}"
-puts "  Total data:      #{TOTAL_DATA / (1024*1024)} MB"
-puts "  Max threads:     #{MAX_THREADS}"
-puts "  Upload delay:    #{SIMULATED_UPLOAD_DELAY}s per part"
-puts
-
-# Verify source and queue type
-source_file = Aws::S3::DefaultExecutor.instance_method(:initialize).source_location[0]
-puts "DefaultExecutor source: #{source_file}"
-test_executor = Aws::S3::DefaultExecutor.new(max_threads: MAX_THREADS)
-queue = test_executor.instance_variable_get(:@queue)
-puts "Queue type: #{queue.class} (max: #{queue.respond_to?(:max) ? queue.max : 'unbounded'})"
-test_executor.shutdown
-puts
-
-# Set up stubbed client (no real AWS calls)
-client = Aws::S3::Client.new(region: 'us-east-1', stub_responses: true)
-client.stub_responses(:create_multipart_upload, { upload_id: 'test-upload-id' })
-client.stub_responses(:upload_part, ->(context) {
-  sleep(SIMULATED_UPLOAD_DELAY)
-  { etag: "\"etag-#{context.params[:part_number]}\"" }
-})
-client.stub_responses(:complete_multipart_upload, {
-  location: 'https://bucket.s3.amazonaws.com/key',
-  bucket: 'test-bucket',
-  key: 'test-key',
-  etag: '"final-etag"'
-})
-
-tm = Aws::S3::TransferManager.new(client: client)
-
-baseline_rss = rss_mb
-peak_rss = baseline_rss
-
-sampler = Thread.new do
-  loop do
-    current = rss_mb
-    peak_rss = current if current > peak_rss
-    sleep(0.01)
-  end
-end
-
-puts "Baseline RSS: #{baseline_rss.round(1)} MB"
-puts "Uploading #{TOTAL_DATA / (1024*1024)} MB with #{MAX_THREADS} threads..."
-start_time = Time.now
-
-tm.upload_stream(
-  bucket: 'test-bucket',
-  key: 'test-key',
-  part_size: PART_SIZE,
-  thread_count: MAX_THREADS
-) do |write_stream|
-  chunk = 'x' * (1024 * 1024) # write in 1 MB chunks (fast local source)
-  bytes_written = 0
-  while bytes_written < TOTAL_DATA
-    write_stream.write(chunk)
-    bytes_written += chunk.size
-  end
-end
-
-elapsed = Time.now - start_time
-sampler.kill
-
-memory_growth = peak_rss - baseline_rss
-
-puts "Done in #{elapsed.round(2)}s"
-puts
-puts "=== Memory Results ==="
-puts "  Peak RSS:        #{peak_rss.round(1)} MB"
-puts "  Memory growth:   #{memory_growth.round(1)} MB"
-puts
-
-# With backpressure (SizedQueue), expect bounded memory:
-#   ~max_threads * part_size + pipe buffer + overhead ≈ 30-50 MB
-# Without backpressure (Queue), all 40 parts buffer in memory:
-#   ~40 * 5 MB = 200 MB growth
-bounded_limit_mb = (MAX_THREADS + 2) * (PART_SIZE / (1024.0 * 1024)) + 50
-
-if memory_growth < bounded_limit_mb
-  puts "PASS: Memory is bounded (#{memory_growth.round(1)} MB < #{bounded_limit_mb.round(0)} MB limit)"
-  puts "      The SizedQueue backpressure is working correctly."
-else
-  puts "FAIL: Memory grew unbounded (#{memory_growth.round(1)} MB > #{bounded_limit_mb.round(0)} MB limit)"
-  puts "      All #{NUM_PARTS} parts (#{NUM_PARTS * PART_SIZE / (1024*1024)} MB) were buffered in the queue."
-  puts "      This confirms bug #3393."
-end

From c796bc5261232af13debeb6b93eb3086caa55199 Mon Sep 17 00:00:00 2001
From: Richard Wang <wangrch@amazon.com>
Date: Thu, 9 Jul 2026 10:48:53 -0700
Subject: [PATCH 4/5] Update tests

---
 .../spec/multipart_stream_uploader_spec.rb    | 122 +++++++-----------
 1 file changed, 45 insertions(+), 77 deletions(-)

diff --git a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
index a1e778a124a..af2ff1798e0 100644
--- a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
+++ b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
@@ -22,83 +22,6 @@ module S3
         end
       end
 
-      describe '#upload_stream memory bounds', :jruby_flaky do
-        it 'bounds queued parts to the thread count when source outpaces upload' do
-          num_threads = 4
-          part_size = 1024 * 1024 # 1 MB parts for faster test
-          total_parts = 20
-          total_data = part_size * total_parts
-
-          client.stub_responses(:create_multipart_upload, upload_id: 'id')
-          client.stub_responses(:complete_multipart_upload)
-
-          mutex = Mutex.new
-          queue_depth_samples = []
-
-          executor = DefaultExecutor.new(max_threads: num_threads)
-
-          # Simulate slow uploads so parts queue up faster than they drain
-          allow(client).to receive(:upload_part) do |_part|
-            mutex.synchronize do
-              queue_depth_samples << executor.instance_variable_get(:@queue).size
-            end
-            sleep(0.05)
-            double(:upload_part, etag: 'etag')
-          end
-
-          uploader = MultipartStreamUploader.new(
-            client: client,
-            executor: executor,
-            part_size: part_size
-          )
-
-          uploader.upload(params) do |write_stream|
-            write_stream << ('a' * total_data)
-          end
-
-          peak_queue_depth = queue_depth_samples.max || 0
-
-          # With backpressure (SizedQueue), the reader blocks when the
-          # queue is full, so depth never exceeds the thread count.
-          expect(peak_queue_depth).to be <= num_threads,
-            "Expected peak queue depth (#{peak_queue_depth}) to be at most " \
-            "num_threads (#{num_threads}), but the queue grew unbounded."
-        end
-
-        it 'completes all parts under backpressure' do
-          num_threads = 2
-          part_size = 1024 * 1024 # 1 MB
-          total_parts = 10
-          total_data = part_size * total_parts
-
-          client.stub_responses(:create_multipart_upload, upload_id: 'id')
-          client.stub_responses(:complete_multipart_upload)
-
-          mutex = Mutex.new
-          uploaded_parts = []
-
-          executor = DefaultExecutor.new(max_threads: num_threads)
-
-          allow(client).to receive(:upload_part) do |part|
-            sleep(0.05) # slow upload
-            mutex.synchronize { uploaded_parts << part[:part_number] }
-            double(:upload_part, etag: 'etag')
-          end
-
-          uploader = MultipartStreamUploader.new(
-            client: client,
-            executor: executor,
-            part_size: part_size
-          )
-
-          uploader.upload(params) do |write_stream|
-            write_stream << ('a' * total_data)
-          end
-
-          expect(uploaded_parts.sort).to eq((1..total_parts).to_a)
-        end
-      end
-
       describe '#upload_stream', :jruby_flaky do
         it 'can upload empty stream' do
           client.stub_responses(:create_multipart_upload, upload_id: 'id')
@@ -230,6 +153,51 @@ module S3
           end.to raise_error(S3::MultipartUploadError, /failed to abort multipart upload: network-error/)
         end
 
+        context 'when source outpaces upload' do
+          let(:num_threads) { 4 }
+          let(:executor) { DefaultExecutor.new(max_threads: num_threads) }
+          let(:subject) { MultipartStreamUploader.new(client: client, executor: executor, part_size: 1024 * 1024) }
+
+          it 'bounds concurrent in-flight parts to the thread count' do
+            client.stub_responses(:create_multipart_upload, upload_id: 'id')
+            client.stub_responses(:complete_multipart_upload)
+            mutex = Mutex.new
+            in_flight = 0
+            peak_in_flight = 0
+            allow(client).to receive(:upload_part) do |_part|
+              mutex.synchronize do
+                in_flight += 1
+                peak_in_flight = in_flight if in_flight > peak_in_flight
+              end
+              sleep(0.05)
+              mutex.synchronize { in_flight -= 1 }
+            end.and_return(double(:upload_part, etag: 'etag'))
+
+            subject.upload(params) do |write_stream|
+              20.times { write_stream << one_mb }
+            end
+
+            expect(peak_in_flight).to be <= num_threads
+          end
+
+          it 'completes all parts under backpressure' do
+            client.stub_responses(:create_multipart_upload, upload_id: 'id')
+            client.stub_responses(:complete_multipart_upload)
+            mutex = Mutex.new
+            uploaded_parts = []
+            allow(client).to receive(:upload_part) do |part|
+              sleep(0.05)
+              mutex.synchronize { uploaded_parts << part[:part_number] }
+            end.and_return(double(:upload_part, etag: 'etag'))
+
+            subject.upload(params) do |write_stream|
+              10.times { write_stream << one_mb }
+            end
+
+            expect(uploaded_parts.sort).to eq((1..10).to_a)
+          end
+        end
+
         context 'when tempfile is true' do
           let(:subject) { MultipartStreamUploader.new(client: client, tempfile: true, executor: DefaultExecutor.new) }
 

From b20cd27335e3b176f0a8529550fee5667ebe9479 Mon Sep 17 00:00:00 2001
From: Richard Wang <wangrch@amazon.com>
Date: Tue, 28 Jul 2026 11:30:57 -0700
Subject: [PATCH 5/5] PR comments

---
 gems/aws-sdk-s3/CHANGELOG.md                  |  2 +
 .../lib/aws-sdk-s3/customizations/object.rb   |  9 ++-
 .../lib/aws-sdk-s3/default_executor.rb        | 18 ++++--
 .../aws-sdk-s3/multipart_stream_uploader.rb   |  5 +-
 .../lib/aws-sdk-s3/transfer_manager.rb        |  8 ++-
 gems/aws-sdk-s3/spec/default_executor_spec.rb | 56 +++++++++++++++++++
 .../spec/multipart_stream_uploader_spec.rb    | 31 ++++++----
 .../spec/object/upload_stream_spec.rb         |  5 +-
 8 files changed, 113 insertions(+), 21 deletions(-)

diff --git a/gems/aws-sdk-s3/CHANGELOG.md b/gems/aws-sdk-s3/CHANGELOG.md
index e9c3f1a878f..40ccd6f2e9b 100644
--- a/gems/aws-sdk-s3/CHANGELOG.md
+++ b/gems/aws-sdk-s3/CHANGELOG.md
@@ -1,6 +1,8 @@
 Unreleased Changes
 ------------------
 
+* Issue - Bound memory usage in `upload_stream` when the source produces data faster than parts can be uploaded.
+
 1.226.0 (2026-06-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..99b66013c4a 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
@@ -380,7 +380,9 @@ def public_url(options = {})
       #   and {Client#upload_part} can be provided.
       #
       # @option options [Integer] :thread_count (10) The number of parallel multipart uploads.
-      #   An additional thread is used internally for task coordination.
+      #   An additional thread is used internally for task coordination. This also bounds
+      #   how many parts are buffered ahead of the upload, limiting memory usage to roughly
+      #   `2 * :thread_count * :part_size`.
       #
       # @option options [Boolean] :tempfile (false) Normally read data is stored
       #   in memory when building the parts in order to complete the underlying
@@ -405,7 +407,10 @@ def public_url(options = {})
       # @see Client#upload_part
       def upload_stream(options = {}, &block)
         upload_opts = options.merge(bucket: bucket_name, key: key)
-        executor = DefaultExecutor.new(max_threads: upload_opts.delete(:thread_count))
+        thread_count = upload_opts.delete(:thread_count) || DefaultExecutor::DEFAULT_MAX_THREADS
+        # A bounded queue prevents the source from reading ahead without limit when it
+        # produces data faster than parts can be uploaded.
+        executor = DefaultExecutor.new(max_threads: thread_count, max_queue: thread_count)
         uploader = MultipartStreamUploader.new(
           client: client,
           executor: executor,
diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
index f57c583a379..8c16f393ed3 100644
--- a/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
+++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/default_executor.rb
@@ -11,8 +11,10 @@ class DefaultExecutor
 
       def initialize(options = {})
         @max_threads = options[:max_threads] || DEFAULT_MAX_THREADS
+        @max_queue = options[:max_queue] || 0
         @state = RUNNING
-        @queue = SizedQueue.new(@max_threads)
+        # A bounded queue applies backpressure to producers when full. 0 means unbounded.
+        @queue = @max_queue.zero? ? Queue.new : SizedQueue.new(@max_queue)
         @pool = []
         @mutex = Mutex.new
       end
@@ -27,8 +29,13 @@ def post(*args, &block)
 
           ensure_worker_available
         end
-        @queue << [args, block]
+        # Pushed outside the mutex because a bounded queue blocks the caller when
+        # full and holding the lock while parked would deadlock #shutdown and #kill.
+        @queue.push([args, block])
         true
+      rescue ClosedQueueError
+        # shutdown or kill happened while parked on a full queue
+        raise 'Executor has been shutdown and is no longer accepting tasks'
       end
 
       # Immediately terminates all worker threads and clears pending tasks.
@@ -38,6 +45,7 @@ def post(*args, &block)
       def kill
         @mutex.synchronize do
           @state = SHUTDOWN
+          @queue.close # wakes any producer parked on a full queue
           @pool.each(&:kill)
           @pool.clear
           @queue.clear
@@ -56,7 +64,9 @@ def shutdown(timeout = nil)
           return true if @state == SHUTDOWN
 
           @state = SHUTTING_DOWN
-          @pool.size.times { @queue << :shutdown }
+          # Closing wakes parked producers and lets workers drain remaining tasks
+          # before exiting without pushing sentinels onto a queue that may be full.
+          @queue.close
         end
 
         if timeout
@@ -91,8 +101,6 @@ def ensure_worker_available
       def spawn_worker
         Thread.new do
           while (job = @queue.shift)
-            break if job == :shutdown
-
             args, block = job
             block.call(*args)
           end
diff --git a/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb b/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
index dd025c01fc7..812dbacbac7 100644
--- a/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
+++ b/gems/aws-sdk-s3/lib/aws-sdk-s3/multipart_stream_uploader.rb
@@ -126,8 +126,11 @@ def read_to_part_body(read_pipe)
             temp_io
           end
         else
+          # Read into a single right-sized buffer. IO.copy_stream into a StringIO grows
+          # the backing string geometrically (an 8MB buffer for a 5MB part) and discards
+          # the intermediates, fragmenting the heap across concurrent parts.
           data = read_pipe.read(@part_size)
-          data.nil? || data.empty? ? nil : StringIO.new(data)
+          data.nil? ? nil : StringIO.new(data)
         end
       end
 
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..a3a49597e25 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
@@ -491,6 +491,9 @@ def upload_file(source, bucket:, key:, **options)
       # @option options [Integer] :thread_count (10)
       #   The number of parallel multipart uploads. Only used when no custom executor is provided (creates
       #   {DefaultExecutor} with the given thread count). An additional thread is used internally for task coordination.
+      #   This also bounds how many parts are buffered ahead of the upload, limiting memory usage to roughly
+      #   `2 * :thread_count * :part_size`. When a custom `:executor` is provided, it is responsible for applying
+      #   its own backpressure.
       #
       # @option options [Boolean] :tempfile (false)
       #   Normally read data is stored in memory when building the parts in order to complete the underlying
@@ -511,7 +514,10 @@ def upload_file(source, bucket:, key:, **options)
       # @see Client#upload_part
       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))
+        thread_count = upload_opts.delete(:thread_count) || DefaultExecutor::DEFAULT_MAX_THREADS
+        # A bounded queue prevents the source from reading ahead without limit when it
+        # produces data faster than parts can be uploaded.
+        executor = @executor || DefaultExecutor.new(max_threads: thread_count, max_queue: thread_count)
         uploader = MultipartStreamUploader.new(
           client: @client,
           executor: executor,
diff --git a/gems/aws-sdk-s3/spec/default_executor_spec.rb b/gems/aws-sdk-s3/spec/default_executor_spec.rb
index ffb6974d096..27e2aa889e6 100644
--- a/gems/aws-sdk-s3/spec/default_executor_spec.rb
+++ b/gems/aws-sdk-s3/spec/default_executor_spec.rb
@@ -24,6 +24,62 @@ module S3
         end
       end
 
+      context 'when the queue is full' do
+        let(:executor) { DefaultExecutor.new(max_threads: 1, max_queue: 1) }
+        let(:release) { Queue.new }
+        let(:errors) { [] }
+
+        # occupies the only worker, then fills the single queue slot
+        def fill_queue
+          started = Queue.new
+          executor.post do
+            started << :running
+            release.pop
+          end
+          started.pop
+          executor.post {}
+        end
+
+        def park_producer
+          fill_queue
+          producer = Thread.new do
+            executor.post {}
+          rescue RuntimeError => e
+            errors << e
+          end
+          sleep(0.1)
+          raise 'producer did not park' unless producer.status == 'sleep'
+
+          producer
+        end
+
+        it 'blocks the caller until a worker frees a slot' do
+          fill_queue
+          parked = Thread.new { executor.post {} }
+          sleep(0.1)
+          expect(parked.status).to eq('sleep')
+
+          release << :go
+          expect(parked.value).to be(true)
+          executor.shutdown
+        end
+
+        it 'kill unblocks the producer instead of silently dropping the task' do
+          producer = park_producer
+          expect(executor.kill).to be(true)
+          expect(producer.join(2)).to_not be_nil
+          expect(errors.first).to be_a(RuntimeError)
+        end
+
+        it 'shutdown does not deadlock while holding the lock' do
+          producer = park_producer
+          shutdown = Thread.new { executor.shutdown(0.5) }
+          expect(shutdown.join(2)).to_not be_nil
+          expect(producer.join(2)).to_not be_nil
+          expect(errors.first).to be_a(RuntimeError)
+        end
+      end
+
       describe '#shutdown' do
         it 'waits for running tasks to be complete' do
           result = nil
diff --git a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
index af2ff1798e0..28d75fbbb3e 100644
--- a/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
+++ b/gems/aws-sdk-s3/spec/multipart_stream_uploader_spec.rb
@@ -154,30 +154,39 @@ module S3
         end
 
         context 'when source outpaces upload' do
-          let(:num_threads) { 4 }
-          let(:executor) { DefaultExecutor.new(max_threads: num_threads) }
+          let(:num_threads) { 2 }
+          let(:executor) { DefaultExecutor.new(max_threads: num_threads, max_queue: num_threads) }
           let(:subject) { MultipartStreamUploader.new(client: client, executor: executor, part_size: 1024 * 1024) }
 
-          it 'bounds concurrent in-flight parts to the thread count' do
+          it 'bounds the number of parts buffered ahead of the upload' do
             client.stub_responses(:create_multipart_upload, upload_id: 'id')
             client.stub_responses(:complete_multipart_upload)
             mutex = Mutex.new
-            in_flight = 0
-            peak_in_flight = 0
-            allow(client).to receive(:upload_part) do |_part|
+            buffered = 0
+            peak_buffered = 0
+            # count parts read off the pipe but not yet uploaded
+            allow(subject).to receive(:read_to_part_body).and_wrap_original do |original, *args|
+              body = original.call(*args)
               mutex.synchronize do
-                in_flight += 1
-                peak_in_flight = in_flight if in_flight > peak_in_flight
+                if body
+                  buffered += 1
+                  peak_buffered = buffered if buffered > peak_buffered
+                end
               end
+              body
+            end
+            allow(client).to receive(:upload_part) do |_part|
               sleep(0.05)
-              mutex.synchronize { in_flight -= 1 }
+              mutex.synchronize { buffered -= 1 }
             end.and_return(double(:upload_part, etag: 'etag'))
 
             subject.upload(params) do |write_stream|
-              20.times { write_stream << one_mb }
+              30.times { write_stream << one_mb }
             end
 
-            expect(peak_in_flight).to be <= num_threads
+            # at most max_queue queued + max_threads in flight + 1 being read.
+            # without a bounded queue all 30 parts are read into memory up front.
+            expect(peak_buffered).to be <= (num_threads * 2) + 1
           end
 
           it 'completes all parts under backpressure' do
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..f8683144930 100644
--- a/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb
+++ b/gems/aws-sdk-s3/spec/object/upload_stream_spec.rb
@@ -29,7 +29,10 @@ module S3
           custom_thread_count = 20
           client.stub_responses(:create_multipart_upload, upload_id: 'id')
           client.stub_responses(:complete_multipart_upload)
-          expect(DefaultExecutor).to receive(:new).with(max_threads: custom_thread_count).and_call_original
+          expect(DefaultExecutor)
+            .to receive(:new)
+            .with(max_threads: custom_thread_count, max_queue: custom_thread_count)
+            .and_call_original
           subject.upload_stream(thread_count: custom_thread_count) { |_write_stream| }
         end