S3 Directory Support

gems/aws-sdk-s3/CHANGELOG.md

@@ -1,6 +1,8 @@
 Unreleased Changes
 ------------------
 
+* Feature - Added `#upload_directory` and `#download_directory` to `Aws::S3::TransferManager` for bulk directory transfers.
+
 1.211.0 (2026-01-08)
 ------------------
 

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

@@ -7,15 +7,23 @@ module S3
     autoload :Encryption, 'aws-sdk-s3/encryption'
     autoload :EncryptionV2, 'aws-sdk-s3/encryption_v2'
     autoload :EncryptionV3, 'aws-sdk-s3/encryption_v3'
+    autoload :LegacySigner, 'aws-sdk-s3/legacy_signer'
+
+    # transfer manager + multipart upload/download utilities
+    autoload :DefaultExecutor, 'aws-sdk-s3/default_executor'
     autoload :FilePart, 'aws-sdk-s3/file_part'
     autoload :DefaultExecutor, 'aws-sdk-s3/default_executor'
     autoload :FileUploader, 'aws-sdk-s3/file_uploader'
     autoload :FileDownloader, 'aws-sdk-s3/file_downloader'
-    autoload :LegacySigner, 'aws-sdk-s3/legacy_signer'
     autoload :MultipartDownloadError, 'aws-sdk-s3/multipart_download_error'
     autoload :MultipartFileUploader, 'aws-sdk-s3/multipart_file_uploader'
     autoload :MultipartStreamUploader, 'aws-sdk-s3/multipart_stream_uploader'
     autoload :MultipartUploadError, 'aws-sdk-s3/multipart_upload_error'
+    autoload :DirectoryProgress, 'aws-sdk-s3/directory_progress'
+    autoload :DirectoryDownloadError, 'aws-sdk-s3/directory_download_error'
+    autoload :DirectoryDownloader, 'aws-sdk-s3/directory_downloader'
+    autoload :DirectoryUploadError, 'aws-sdk-s3/directory_upload_error'
+    autoload :DirectoryUploader, 'aws-sdk-s3/directory_uploader'
     autoload :ObjectCopier, 'aws-sdk-s3/object_copier'
     autoload :ObjectMultipartCopier, 'aws-sdk-s3/object_multipart_copier'
     autoload :PresignedPost, 'aws-sdk-s3/presigned_post'

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

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+module Aws
+  module S3
+    # Raised when DirectoryDownloader fails to download objects from S3 bucket
+    class DirectoryDownloadError < StandardError
+      def initialize(message, errors = [])
+        @errors = errors
+        super(message)
+      end
+
+      # @return [Array<StandardError>] The list of errors encountered when downloading objects
+      attr_reader :errors
+    end
+  end
+end

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

@@ -0,0 +1,212 @@
+# frozen_string_literal: true
+
+module Aws
+  module S3
+    # @api private
+    class DirectoryDownloader
+      def initialize(options = {})
+        @client = options[:client]
+        @executor = options[:executor]
+        @abort_requested = false
+        @mutex = Mutex.new
+      end
+
+      attr_reader :client, :executor
+
+      def abort_requested
+        @mutex.synchronize { @abort_requested }
+      end
+
+      def request_abort
+        @mutex.synchronize { @abort_requested = true }
+      end
+
+      def download(destination, bucket:, **options)
+        if File.exist?(destination)
+          raise ArgumentError, 'invalid destination, expected a directory' unless File.directory?(destination)
+        else
+          FileUtils.mkdir_p(destination)
+        end
+
+        download_opts, producer_opts = build_opts(destination, bucket, options)
+        downloader = FileDownloader.new(client: @client, executor: @executor)
+        producer = ObjectProducer.new(producer_opts)
+        downloads, errors = process_download_queue(producer, downloader, download_opts)
+        build_result(downloads, errors)
+      ensure
+        @abort_requested = false
+      end
+
+      private
+
+      def build_opts(destination, bucket, opts)
+        download_opts = {
+          progress_callback: opts[:progress_callback],
+          destination: destination,
+          ignore_failure: opts[:ignore_failure] || false
+        }
+        producer_opts = {
+          client: @client,
+          directory_downloader: self,
+          destination: destination,
+          bucket: bucket,
+          s3_prefix: opts[:s3_prefix],
+          filter_callback: opts[:filter_callback],
+          request_callback: opts[:request_callback]
+        }
+        [download_opts, producer_opts]
+      end
+
+      def build_result(download_count, errors)
+        if abort_requested
+          msg = "directory download failed: #{errors.map(&:message).join('; ')}"
+          raise DirectoryDownloadError.new(msg, errors)
+        else
+          {
+            completed_downloads: [download_count - errors.count, 0].max,
+            failed_downloads: errors.count,
+            errors: errors.any? ? errors : nil
+          }.compact
+        end
+      end
+
+      def download_object(entry, downloader, opts, progress, errors)
+        raise entry.error if entry.error
+
+        FileUtils.mkdir_p(File.dirname(entry.path)) unless Dir.exist?(File.dirname(entry.path))
+        downloader.download(entry.path, entry.params)
+        progress&.call(File.size(entry.path))
+      rescue StandardError => e
+        @mutex.synchronize { errors << e }
+        request_abort unless opts[:ignore_failure]
+      end
+
+      def process_download_queue(producer, downloader, opts)
+        progress = DirectoryProgress.new(opts[:progress_callback]) if opts[:progress_callback]
+        queue_executor = DefaultExecutor.new
+        completion_queue = Queue.new
+        download_attempts = 0
+        errors = []
+        begin
+          producer.each do |object|
+            break if abort_requested
+
+            download_attempts += 1
+            queue_executor.post(object) do |o|
+              download_object(o, downloader, opts, progress, errors)
+            ensure
+              completion_queue << :done
+            end
+          end
+        rescue StandardError => e
+          @mutex.synchronize { errors << e }
+          request_abort
+        end
+        download_attempts.times { completion_queue.pop }
+        [download_attempts, errors]
+      ensure
+        queue_executor&.shutdown
+      end
+
+      # @api private
+      class ObjectProducer
+        include Enumerable
+
+        DEFAULT_QUEUE_SIZE = 100
+        DONE_MARKER = :done
+
+        def initialize(opts = {})
+          @directory_downloader = opts[:directory_downloader]
+          @destination_dir = opts[:destination]
+          @bucket = opts[:bucket]
+          @client = opts[:client]
+          @s3_prefix = opts[:s3_prefix]
+          @filter_callback = opts[:filter_callback]
+          @request_callback = opts[:request_callback]
+          @object_queue = SizedQueue.new(DEFAULT_QUEUE_SIZE)
+        end
+
+        def each
+          producer_thread = Thread.new do
+            stream_objects
+          ensure
+            @object_queue << DONE_MARKER
+          end
+
+          # Yield objects from internal queue
+          while (object = @object_queue.shift) != DONE_MARKER
+            break if @directory_downloader.abort_requested
+
+            yield object
+          end
+        ensure
+          producer_thread.join
+        end
+
+        private
+
+        def apply_request_callback(key, params)
+          @request_callback&.call(key, params.dup)
+        end
+
+        def build_object_entry(key)
+          params = { bucket: @bucket, key: key }
+          params = apply_request_callback(key, params) if @request_callback
+          normalized_key = normalize_key(key)
+          full_path = File.join(@destination_dir, normalized_key)
+          error = validate_path(full_path, key)
+          DownloadEntry.new(path: full_path, params: params, error: error)
+        end
+
+        def include_object?(key)
+          return true unless @filter_callback
+
+          @filter_callback.call(key)
+        end
+
+        def directory_marker?(obj)
+          obj.key.end_with?('/') && obj.size.zero?
+        end
+
+        def normalize_key(key)
+          if @s3_prefix
+            prefix = @s3_prefix.end_with?('/') ? @s3_prefix : "#{@s3_prefix}/"
+            key = key.delete_prefix(prefix)
+          end
+          File::SEPARATOR == '/' ? key : key.tr('/', File::SEPARATOR)
+        end
+
+        def stream_objects(continuation_token: nil)
+          resp = @client.list_objects_v2(bucket: @bucket, prefix: @s3_prefix, continuation_token: continuation_token)
+          resp.contents&.each do |o|
+            break if @directory_downloader.abort_requested
+
+            next if directory_marker?(o)
+            next unless include_object?(o.key)
+
+            @object_queue << build_object_entry(o.key)
+          end
+          stream_objects(continuation_token: resp.next_continuation_token) if resp.next_continuation_token
+        end
+
+        def validate_path(path, key)
+          segments = path.split('/')
+          return unless segments.any? { |s| %w[. ..].include?(s) }
+
+          DirectoryDownloadError.new("invalid key '#{key}': contains '.' or '..' path segments")
+        end
+
+        # @api private
+        class DownloadEntry
+          def initialize(opts = {})
+            @path = opts[:path]
+            @params = opts[:params]
+            @error = opts[:error]
+          end
+
+          attr_reader :path, :params, :error
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module Aws
+  module S3
+    # @api private
+    class DirectoryProgress
+      def initialize(progress_callback)
+        @transferred_bytes = 0
+        @transferred_files = 0
+        @progress_callback = progress_callback
+        @mutex = Mutex.new
+      end
+
+      def call(bytes_transferred)
+        @mutex.synchronize do
+          @transferred_bytes += bytes_transferred
+          @transferred_files += 1
+
+          @progress_callback.call(@transferred_bytes, @transferred_files)
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+module Aws
+  module S3
+    # Raised when DirectoryUploader fails to upload files to S3 bucket
+    class DirectoryUploadError < StandardError
+      def initialize(message, errors = [])
+        @errors = errors
+        super(message)
+      end
+
+      # @return [Array<StandardError>] The list of errors encountered when uploading files
+      attr_reader :errors
+    end
+  end
+end