Make iter_content a chunked stream with a configurable chunk size

iter_content read at a size the caller could not change, and the sizes
were hardcoded and inconsistent: 4 KiB on AWS, 64 KiB on OpenStack
Swift, whatever the service chose on Azure. The AWS value was never a
tuning decision - it arrived with the 2017 boto2-to-boto3 migration,
where a BucketObjIterator shim replaced boto2's Key (which read in 8 KiB
BufferSize chunks) - and it is small enough to dominate a large read.
Reading an HTTP body at 1 MiB measures ~12x cheaper per byte than at
4 KiB, and the curve is flat from 1 MiB up, so consumers streaming large
objects paid for it in per-chunk overhead with no way out. save_content
was never affected: it copied via shutil.copyfileobj, which reads in
64 KiB blocks regardless of what iter_content yields.

Two providers could not have honoured a chunk size at all, because they
were not chunking by size to begin with. Azure returned an io.RawIOBase
wrapper, and iterating a raw stream calls readline(), so chunks broke at
b"\n" wherever the content happened to put one and a blob with no
newline in it buffered whole however large it was. GCP returned an
io.BytesIO around a full get_media().execute(), materialising the entire
object in memory before yielding anything (and, being a BytesIO, also
iterating by line).

Give iter_content and save_content a chunk_size, defaulting to 1 MiB and
settable globally through the iter_chunk_size provider config value or
the CB_ITER_CHUNK_SIZE environment variable, alongside the existing
CB_MULTIPART_* knobs. Azure now reads chunk_size slices off the
downloader over a single connection; GCP fetches successive ranged reads,
keeping memory flat at one chunk. Both yield chunks sized by chunk_size
rather than by the content, so binary data is never split on newlines.

save_content is now written in terms of iter_content rather than
shutil.copyfileobj, which required a .read() that the interface never
promised - it only ever promised Iterable[bytes]. A provider returning a
plain generator now works.

chunk_size is optional everywhere, so existing calls are unaffected, and
the AWS return value still exposes read/close. The Azure and GCP return
values are plain generators now, so code calling .read() on those must
iterate instead.

The object store service suite only exercises the one provider it is
configured against - the AWS-backed mock, in CI - so the per-provider
chunking loops are pinned in a new tests/test_iter_content.py against
fake SDK objects, following test_download_driver.py.

CHANGELOG.rst

@@ -1,22 +1,28 @@
 4.4.0 - unreleased
 ------------------
 
-## Fixes
-* **Paginated AWS calls no longer use the caller's result limit as the
-  transport page size.** ``BotoEC2Service._get_paginated_results`` set
-  ``PaginationConfig={'MaxItems': limit, 'PageSize': limit}``, conflating how
-  many results the caller wants with how many the service returns per
-  request. Against a scan that matches sparsely that walks the collection in
-  tiny increments: the same filtered ``describe_images`` took 977.6s at a
-  page size of 5 and 10.0s at 1000, for one result either way. ``MaxItems``
-  still bounds what the caller receives; ``PageSize`` is now a full page,
-  clamped to whatever bounds the service model declares for the operation
-  (``DescribeRouteTables`` permits 100 where most permit more, several require
-  at least 5, and falling outside them is a hard ``InvalidParameterValue``).
-  Since ``DEFAULT_RESULT_LIMIT`` is 50, every paginated AWS call was affected,
-  not just filtered searches.
+## Release highlights
+``BucketObject.iter_content`` becomes a real chunked stream on every provider:
+it takes a ``chunk_size``, yields chunks sized by that rather than by the
+content, and no longer materialises a whole object in memory anywhere.
+Separately, two pathological patterns in AWS listing are removed - an image
+search that scanned the region's entire public catalogue, and a paginated
+call that used the caller's result limit as its transport page size.
 
 ## Enhancements
+* **``iter_content`` and ``save_content`` accept a ``chunk_size``.** It
+  defaults to 1 MiB and is settable globally through the ``iter_chunk_size``
+  provider config value or the ``CB_ITER_CHUNK_SIZE`` environment variable,
+  alongside the existing ``CB_MULTIPART_*`` knobs. Previously the read size
+  was hardcoded and inconsistent - 4 KiB on AWS, 64 KiB on OpenStack Swift,
+  the service's own chunking on Azure - and callers had no way to change it.
+  The AWS value dated from the 2017 boto2-to-boto3 migration, where a
+  ``BucketObjIterator`` shim replaced boto2's ``Key`` (which read in 8 KiB
+  ``BufferSize`` chunks); it was never a tuning decision. Reading an HTTP body
+  at 1 MiB measures ~12x cheaper per byte than at 4 KiB, and the curve is flat
+  from 1 MiB up, so larger defaults would only cost memory. Note that
+  ``save_content`` was never affected: it copied via ``shutil.copyfileobj``,
+  which reads in 64 KiB blocks regardless.
 * **New ``aws_page_size`` configuration value.** How many records to request
   from AWS per call while satisfying a list method, defaulting to 500. It is
   a transport setting, distinct from ``default_result_limit``, which bounds
@@ -26,6 +32,26 @@
   because it only means anything where the provider walks pages itself:
   GCP, Azure and OpenStack each return a single page plus a continuation
   token and let the caller drive.
+
+## Fixes
+* **Azure no longer splits object content on newlines.** ``iter_content``
+  returned an ``io.RawIOBase`` wrapper, and iterating a raw stream calls
+  ``readline()`` - so chunks broke at ``b"\n"`` at whatever sizes the content
+  happened to dictate, and a blob with no newline in it was buffered whole
+  however large it was. It now yields ``chunk_size`` chunks read over a single
+  connection.
+* **GCP no longer loads the entire object into memory.** ``iter_content``
+  returned ``io.BytesIO`` wrapped around a full ``get_media().execute()``, so
+  streaming a large object cost its full size in RAM (and, being a
+  ``BytesIO``, also iterated by line). Content is now fetched as successive
+  ranged reads of ``chunk_size`` bytes, keeping memory flat at one chunk.
+  ``download_to_file`` remains the faster path for downloading to disk, as it
+  fetches ranges in parallel.
+* **``save_content`` no longer requires ``iter_content`` to return a
+  file-like object.** It copied with ``shutil.copyfileobj``, which needs a
+  ``.read()`` that the interface never promised - only ``Iterable[bytes]``. It
+  now writes the iterated chunks directly, so a provider returning a plain
+  generator works.
 * **``AWSImageService.find`` no longer scans every public image to run its
   tag search.** ``find(label=...)`` issues two ``describe_images`` calls, one
   filtered on ``name`` and one on ``tag:Name``, and neither was scoped by
@@ -38,6 +64,25 @@
   The ``name`` half is unchanged and still searches public images, which is
   what most callers want; an explicit ``owners`` argument still overrides
   both.
+* **Paginated AWS calls no longer use the caller's result limit as the
+  transport page size.** ``BotoEC2Service._get_paginated_results`` set
+  ``PaginationConfig={'MaxItems': limit, 'PageSize': limit}``, conflating how
+  many results the caller wants with how many the service returns per
+  request. Against a scan that matches sparsely that walks the collection in
+  tiny increments: the same filtered ``describe_images`` took 977.6s at a
+  page size of 5 and 10.0s at 1000, for one result either way. ``MaxItems``
+  still bounds what the caller receives; ``PageSize`` is now a full page,
+  clamped to whatever bounds the service model declares for the operation
+  (``DescribeRouteTables`` permits 100 where most permit more, several require
+  at least 5, and falling outside them is a hard ``InvalidParameterValue``).
+  Since ``DEFAULT_RESULT_LIMIT`` is 50, every paginated AWS call was affected,
+  not just filtered searches.
+
+## Backward compatibility
+``chunk_size`` is optional everywhere, so existing calls keep working. The
+AWS return value still exposes ``read``/``close`` as before. The Azure and GCP
+return values are now plain generators: code that called ``.read()`` on them
+must iterate instead, or use ``save_content``/``download_to_file``.
 
 4.3.1 - August 2, 2026 (sha 8fabc1e2d3916e2c100bdb18075f2caa3bd38b38)
 ---------------------------------------------------------------------

cloudbridge/base/resources.py

@@ -8,7 +8,6 @@ import logging
 import os
 import queue
 import re
-import shutil
 import threading
 import time
 import uuid
@@ -71,8 +70,6 @@ from cloudbridge.interfaces.resources import VolumeState
 from . import helpers as cb_helpers
 
 if TYPE_CHECKING:
-    from _typeshed import SupportsRead
-
     from cloudbridge.base.provider import BaseCloudProvider
     from cloudbridge.base.services import BaseStorageService
     from cloudbridge.interfaces.services import BucketObjectService
@@ -840,6 +837,15 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
     # Number of parts uploaded in parallel by the transparent multipart path.
     CB_MULTIPART_MAX_CONCURRENCY = int(os.environ.get(
         'CB_MULTIPART_MAX_CONCURRENCY', 5))
+    # Size of each chunk yielded by the single-stream read path
+    # (``iter_content``/``save_content``). Sits at the knee of the
+    # throughput curve: reading an HTTP body at 1 MiB is ~12x cheaper per
+    # byte than at 4 KiB, while 4 MiB and above measure the same as 1 MiB
+    # and cost proportionally more memory per concurrent stream. Unrelated
+    # to CB_MULTIPART_PART_SIZE, which sizes the parts of a *parallel*
+    # transfer rather than the chunks of a sequential read.
+    CB_ITER_CHUNK_SIZE = int(os.environ.get(
+        'CB_ITER_CHUNK_SIZE', 1024 * 1024))              # 1 MiB
 
     def __init__(self, provider: CloudProvider) -> None:
         super(BaseBucketObject, self).__init__(provider)
@@ -878,12 +884,14 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
                 "in: http://docs.aws.amazon.com/AmazonS3/latest/dev/UsingMeta"
                 "data.html#object-key-guidelines" % name)
 
-    def save_content(self, target_stream: IO[bytes]) -> None:
-        # iter_content() is declared Iterable[bytes] on the interface, but the
-        # concrete objects returned by providers also support .read(); cast so
-        # copyfileobj accepts it without changing behavior.
-        shutil.copyfileobj(
-            cast("SupportsRead[bytes]", self.iter_content()), target_stream)
+    def save_content(self, target_stream: IO[bytes],
+                     chunk_size: int | None = None) -> None:
+        # Written in terms of iter_content so that the interface's promise -
+        # an Iterable[bytes] - is all a provider has to deliver. Copying via
+        # shutil.copyfileobj would additionally require a .read(), which not
+        # every provider's return value has.
+        for chunk in self.iter_content(chunk_size=chunk_size):
+            target_stream.write(chunk)
 
     def download_to_file(self, path: str,
                          config: TransferConfig | None = None) -> None:
@@ -1008,6 +1016,22 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
         return int(self._provider._get_config_value(
             'multipart_max_concurrency', self.CB_MULTIPART_MAX_CONCURRENCY))
 
+    def _iter_chunk_size(self, chunk_size: int | None = None) -> int:
+        """
+        Resolve the chunk size for a single-stream read: an explicit
+        ``chunk_size``, else the provider/global config, else the class
+        default. Providers call this at the top of ``iter_content`` so the
+        value is validated before any request is issued.
+        """
+        if chunk_size is None:
+            chunk_size = int(self._provider._get_config_value(
+                'iter_chunk_size', self.CB_ITER_CHUNK_SIZE))
+        else:
+            chunk_size = int(chunk_size)
+        if chunk_size <= 0:
+            raise InvalidValueException('iter_chunk_size', chunk_size)
+        return chunk_size
+
     @staticmethod
     def _data_size(data: str | bytes | IO[bytes]) -> int | None:
         """

cloudbridge/interfaces/resources.py

@@ -2357,20 +2357,52 @@ class BucketObject(CloudResource):
         pass
 
     @abstractmethod
-    def iter_content(self) -> Iterable[bytes]:
+    def iter_content(self, chunk_size: int | None = None) -> Iterable[bytes]:
         """
-        Returns this object's content as an iterable.
+        Returns this object's content as an iterable of byte chunks.
+
+        The object is streamed rather than held in memory, so this is safe
+        for objects of any size. Chunks are sized by ``chunk_size``, never by
+        the content itself - binary data is never split on newlines - and
+        only the final chunk may be shorter.
+
+        ``chunk_size`` trades per-chunk overhead against memory and latency.
+        Each chunk costs a read from the provider plus whatever the caller
+        does per chunk, so small values are expensive over a large object;
+        conversely a chunk is buffered in full before it is yielded, so large
+        values cost memory per concurrent stream and delay the first chunk.
+        The default suits most callers.
+
+        :type chunk_size: ``int``
+        :param chunk_size: Maximum size in bytes of each chunk yielded. If
+            ``None``, falls back to the provider/global configuration
+            (``iter_chunk_size`` / the ``CB_ITER_CHUNK_SIZE`` setting,
+            1 MiB by default). Must be positive.
 
         :rtype: Iterable
         :return: An iterable of the file contents
 
+        :raise: ``InvalidValueException`` if ``chunk_size`` is not positive.
         """
         pass
 
     @abstractmethod
-    def save_content(self, target_stream: IO[bytes]) -> None:
+    def save_content(self, target_stream: IO[bytes],
+                     chunk_size: int | None = None) -> None:
         """
         Save this object and write its contents to the ``target_stream``.
+
+        The object is streamed through ``iter_content``, so ``chunk_size``
+        means what it does there.
+
+        :type target_stream: ``IO[bytes]``
+        :param target_stream: A writable binary stream to write to.
+
+        :type chunk_size: ``int``
+        :param chunk_size: Maximum size in bytes of each chunk read from the
+            provider. See :meth:`.iter_content`.
+
+        :raise: ``InvalidValueException`` if ``chunk_size`` is not positive.
         """
         pass
 

cloudbridge/providers/aws/resources.py

@@ -861,14 +861,21 @@ class AWSVMFirewallRule(BaseVMFirewallRule):
 
 class AWSBucketObject(BaseBucketObject):
     class BucketObjIterator():
-        CHUNK_SIZE = 4096
+        """
+        Chunked reader over a boto3 ``StreamingBody``.
+
+        Also exposes ``read``/``close`` so the value handed back by
+        ``iter_content`` stays usable as a file-like object, as it has been
+        since this replaced boto2's ``Key``.
+        """
 
-        def __init__(self, body: Any) -> None:
+        def __init__(self, body: Any, chunk_size: int) -> None:
             self.body = body
+            self.chunk_size = chunk_size
 
         def __iter__(self) -> Iterator[bytes]:
             while True:
-                data = self.read(self.CHUNK_SIZE)
+                data = self.read(self.chunk_size)
                 if data:
                     yield data
                 else:
@@ -910,8 +917,12 @@ class AWSBucketObject(BaseBucketObject):
             cast("AWSCloudProvider", self._provider).s3_conn
             .Bucket(self._obj.bucket_name))
 
-    def iter_content(self) -> Iterable[bytes]:
-        return self.BucketObjIterator(self._obj.get().get('Body'))
+    def iter_content(self, chunk_size: int | None = None) -> Iterable[bytes]:
+        # Resolve (and validate) the chunk size before the GET, so a bad
+        # value does not leave an unread response body behind.
+        chunk_size = self._iter_chunk_size(chunk_size)
+        return self.BucketObjIterator(
+            self._obj.get().get('Body'), chunk_size)
 
     def _upload_single_shot(self,
                             data: str | bytes | IO[bytes]) -> BucketObject:

cloudbridge/providers/azure/resources.py

@@ -4,7 +4,6 @@ DataTypes used by this provider
 from __future__ import annotations
 
 import collections
-import io
 import logging
 from datetime import datetime
 from typing import Any
@@ -245,37 +244,27 @@ class AzureBucketObject(BaseBucketObject):
         """
         return self._blob_properties.last_modified.strftime("%Y-%m-%dT%H:%M:%S.%f")
 
-    def iter_content(self) -> Iterable[bytes]:
+    def iter_content(self, chunk_size: int | None = None) -> Iterable[bytes]:
         """
-        Returns this object's content as an
-        iterable stream.
+        Returns this object's content as an iterable of byte chunks.
         """
-
-        def iterable_to_stream(iterable: Iterator[bytes]) -> io.RawIOBase:
-            class IterStream(io.RawIOBase):
-                def __init__(self) -> None:
-                    self.leftover: bytes | None = None
-
-                def readable(self) -> bool:
-                    return True
-
-                def readinto(self, b: Any) -> int:
-                    try:
-                        buffer_length = len(b)  # We're supposed to return at most this much
-                        chunk = self.leftover or next(iterable)
-                        output, self.leftover = chunk[:buffer_length], chunk[buffer_length:]
-                        b[:len(output)] = output
-                        return len(output)
-                    except StopIteration:
-                        return 0  # indicate EOF
-
-            return IterStream()
+        chunk_size = self._iter_chunk_size(chunk_size)
+        # The downloader buffers the service's own chunks (max_chunk_get_size,
+        # 4 MiB by default) and read() serves chunk_size slices out of them
+        # over a single connection, so the caller's chunk size is independent
+        # of the service's. Previously this returned a RawIOBase wrapper,
+        # which iterates by *line* rather than by size - splitting binary
+        # content on b"\n", and buffering a newline-free blob whole.
+        downloader = self._blob_client.download_blob()
 
         def blob_iterator() -> Iterator[bytes]:
-            for chunk in self._blob_client.download_blob().chunks():
-                yield chunk
+            while True:
+                data = downloader.read(chunk_size)
+                if not data:
+                    break
+                yield data
 
-        return iterable_to_stream(blob_iterator())
+        return blob_iterator()
 
     @property
     def bucket(self) -> AzureBucket:

cloudbridge/providers/gcp/resources.py

@@ -2132,14 +2132,31 @@ class GCPBucketObject(BaseBucketObject):
     def last_modified(self) -> str:
         return self._obj['updated']
 
-    def iter_content(self) -> io.BytesIO:
-        provider = cast("GCPCloudProvider", self._provider)
-        return io.BytesIO(provider
-                          .gcp_storage
-                          .objects()
-                          .get_media(bucket=self._obj['bucket'],
-                                     object=self.name)
-                          .execute())
+    def iter_content(self, chunk_size: int | None = None) -> Iterable[bytes]:
+        """
+        Returns this object's content as an iterable of byte chunks.
+
+        google-api-python-client buffers whole responses, so there is no
+        single-connection streaming read to be had here; content is fetched
+        as successive ranged GETs of ``chunk_size`` bytes instead. That keeps
+        memory flat at one chunk - this previously materialised the entire
+        object in a ``BytesIO`` - at the cost of one request per chunk, so
+        prefer a large ``chunk_size`` when streaming a large object, or
+        :meth:`.download_to_file`, which fetches ranges in parallel.
+        """
+        chunk_size = self._iter_chunk_size(chunk_size)
+        bucket_objects = self._bucket_objects
+        size = self.size
+
+        def range_iterator() -> Iterator[bytes]:
+            offset = 0
+            while offset < size:
+                length = min(chunk_size, size - offset)
+                yield bucket_objects.download_range(
+                    self.bucket, self.name, offset, length)
+                offset += length
+
+        return range_iterator()
 
     @property
     def bucket(self) -> Bucket:

cloudbridge/providers/openstack/resources.py

@@ -1389,10 +1389,14 @@ class OpenStackBucketObject(BaseBucketObject):
     def last_modified(self) -> str:
         return self._obj.get("last_modified")
 
-    def iter_content(self) -> Iterable[bytes]:
-        """Returns this object's content as an iterable."""
+    def iter_content(self, chunk_size: int | None = None) -> Iterable[bytes]:
+        """Returns this object's content as an iterable of byte chunks."""
+        # resp_chunk_size makes swiftclient hand back a streaming _ObjectBody
+        # (iterable, and file-like) instead of the whole object as bytes.
+        chunk_size = self._iter_chunk_size(chunk_size)
         _, content = cast("OpenStackCloudProvider", self._provider).swift \
-            .get_object(self.cbcontainer.name, self.name, resp_chunk_size=65536)
+            .get_object(self.cbcontainer.name, self.name,
+                        resp_chunk_size=chunk_size)
         return content
 
     @property

docs/topics/object_storage.rst

@@ -39,7 +39,40 @@ To locate and download this uploaded file again, you can do the following:
     print("Size: {0}, Modified: {1}".format(obj.size, obj.last_modified))
     with open('/tmp/myfile.txt', 'wb') as f:
         obj.save_content(f)
- 
+
+To download to a local path, prefer download_to_file(), which fetches large
+objects as parallel ranged reads. save_content() and iter_content() are the
+single-stream alternatives, for writing to an arbitrary stream or for handing
+the content on somewhere else - proxying it to an HTTP response, say.
+
+.. code-block:: python
+
+    for chunk in obj.iter_content():
+        process(chunk)
+
+Content is streamed, never held in memory whole, and chunks are sized by
+chunk_size rather than by the content - binary data is never split on
+newlines. Only the last chunk may be short.
+
+.. code-block:: python
+
+    for chunk in obj.iter_content(chunk_size=4 * 1024 * 1024):
+        process(chunk)
+
+chunk_size defaults to 1 MiB, which suits most callers. It trades per-chunk
+overhead against memory and latency: each chunk costs a read from the provider
+plus whatever your loop does per chunk, so small values get expensive over a
+large object, while a large value is buffered in full before it is yielded and
+so costs memory per concurrent stream. To change the default globally, set the
+iter_chunk_size provider config value or the CB_ITER_CHUNK_SIZE environment
+variable. save_content() takes the same argument.
+
+.. note::
+    On GCP the content is fetched as successive ranged requests, because the
+    underlying client library has no single-connection streaming read, so
+    chunk_size is also the request size there. Prefer a large chunk_size when
+    streaming a large object from GCP.
+
 
 Using tokens for authentication
 -------------------------------

tests/test_download_driver.py

@@ -107,7 +107,7 @@ class _DriverObject(BaseBucketObject):
     def bucket(self):
         return "BUCKET"
 
-    def save_content(self, target_stream):
+    def save_content(self, target_stream, chunk_size=None):
         self._provider._recorder.single_shot = True
         target_stream.write(self._provider._recorder.content)
 

tests/test_iter_content.py

@@ -0,0 +1,293 @@
+"""
+Provider-agnostic unit tests for chunked streaming reads
+(``BucketObject.iter_content`` / ``save_content``).
+
+Every provider streams object content through ``iter_content``, but only the
+configured test provider gets exercised by the object store service suite. The
+chunk-size contract those implementations have to honour is pinned here
+against in-memory fakes so it has coverage in CI without cloud credentials.
+"""
+import unittest
+from io import BytesIO
+
+from cloudbridge.base.resources import BaseBucketObject
+from cloudbridge.interfaces.exceptions import InvalidValueException
+
+
+class _FakeProvider:
+    def __init__(self, config=None, bucket_objects=None):
+        self._config = config or {}
+        self.storage = _FakeStorage(bucket_objects)
+
+    def _get_config_value(self, key, default_value=None):
+        return self._config.get(key, default_value)
+
+
+class _FakeStorage:
+    def __init__(self, bucket_objects=None):
+        self._bucket_objects = bucket_objects
+
+
+class _NoRangeService:
+    def download_range(self, bucket, name, offset, length):
+        raise AssertionError("no range should have been requested")
+
+
+class _FakeAzureContainer:
+    """Stands in for AzureBucket: AzureBucketObject reaches its blob client
+    through ``container._bucket.get_blob_client(name)``."""
+
+    def __init__(self, blob_client):
+        self._bucket = self
+        self._blob_client = blob_client
+
+    def get_blob_client(self, name):
+        return self._blob_client
+
+
+class _FakeBlobProperties:
+    def __init__(self, name):
+        self.name = name
+
+
+class _FakeSwiftContainer:
+    name = "bucket"
+
+
+class _StreamingObject(BaseBucketObject):
+    """A BaseBucketObject that streams from an in-memory buffer."""
+
+    def __init__(self, provider, content):
+        super(_StreamingObject, self).__init__(provider)
+        self._content = content
+        self.chunk_sizes_seen = []
+
+    @property
+    def id(self):
+        return "obj"
+
+    @property
+    def name(self):
+        return "obj"
+
+    @property
+    def size(self):
+        return len(self._content)
+
+    @property
+    def bucket(self):
+        return "BUCKET"
+
+    def iter_content(self, chunk_size=None):
+        size = self._iter_chunk_size(chunk_size)
+        self.chunk_sizes_seen.append(size)
+        return (self._content[i:i + size]
+                for i in range(0, len(self._content), size))
+
+
+class IterChunkSizeTestCase(unittest.TestCase):
+    """The resolver that turns an optional chunk_size into a concrete one."""
+
+    def _obj(self, config=None, content=b""):
+        return _StreamingObject(_FakeProvider(config), content)
+
+    def test_defaults_to_class_constant_when_unset(self):
+        obj = self._obj()
+        self.assertEqual(obj._iter_chunk_size(),
+                         BaseBucketObject.CB_ITER_CHUNK_SIZE)
+
+    def test_default_is_one_mebibyte(self):
+        # Large enough that per-chunk overhead disappears against network
+        # throughput, small enough to stay cheap per concurrent stream.
+        self.assertEqual(BaseBucketObject.CB_ITER_CHUNK_SIZE, 1024 * 1024)
+
+    def test_provider_config_overrides_class_constant(self):
+        obj = self._obj({'iter_chunk_size': 8192})
+        self.assertEqual(obj._iter_chunk_size(), 8192)
+
+    def test_explicit_argument_overrides_provider_config(self):
+        obj = self._obj({'iter_chunk_size': 8192})
+        self.assertEqual(obj._iter_chunk_size(4096), 4096)
+
+    def test_rejects_zero_chunk_size(self):
+        obj = self._obj()
+        with self.assertRaises(InvalidValueException):
+            obj._iter_chunk_size(0)
+
+    def test_rejects_negative_chunk_size(self):
+        obj = self._obj()
+        with self.assertRaises(InvalidValueException):
+            obj._iter_chunk_size(-1)
+
+
+class SaveContentTestCase(unittest.TestCase):
+    """save_content is defined in terms of iter_content."""
+
+    def _obj(self, content, config=None):
+        return _StreamingObject(_FakeProvider(config), content)
+
+    def test_writes_whole_content_to_target_stream(self):
+        content = bytes(range(256)) * 40
+        obj = self._obj(content)
+        target = BytesIO()
+        obj.save_content(target)
+        self.assertEqual(target.getvalue(), content)
+
+    def test_passes_chunk_size_through_to_iter_content(self):
+        obj = self._obj(b"x" * 100)
+        obj.save_content(BytesIO(), chunk_size=16)
+        self.assertEqual(obj.chunk_sizes_seen, [16])
+
+    def test_uses_default_chunk_size_when_unset(self):
+        obj = self._obj(b"x" * 10)
+        obj.save_content(BytesIO())
+        self.assertEqual(obj.chunk_sizes_seen,
+                         [BaseBucketObject.CB_ITER_CHUNK_SIZE])
+
+    def test_handles_empty_object(self):
+        obj = self._obj(b"")
+        target = BytesIO()
+        obj.save_content(target)
+        self.assertEqual(target.getvalue(), b"")
+
+    def test_does_not_require_a_readable_iter_content(self):
+        # iter_content promises Iterable[bytes] and nothing more; providers
+        # that return a bare generator must still work with save_content.
+        obj = self._obj(b"abc")
+        self.assertFalse(hasattr(obj.iter_content(), 'read'))
+        target = BytesIO()
+        obj.save_content(target)
+        self.assertEqual(target.getvalue(), b"abc")
+
+
+class ProviderIterContentTestCase(unittest.TestCase):
+    """
+    The per-provider chunking loops.
+
+    The object store service suite only ever exercises the one provider it is
+    configured against - in CI, the AWS-backed mock - so the loops the other
+    providers use to turn an SDK handle into sized chunks are pinned here
+    against fake SDK objects instead.
+    """
+
+    def test_azure_reads_chunk_size_slices_from_one_download(self):
+        from cloudbridge.providers.azure.resources import AzureBucketObject
+
+        content = bytes(range(256)) * 40   # 10 KiB, newline-free by design
+        reads = []
+        downloads = []
+
+        class _Downloader:
+            def __init__(self):
+                self.offset = 0
+
+            def read(self, size):
+                reads.append(size)
+                data = content[self.offset:self.offset + size]
+                self.offset += len(data)
+                return data
+
+        class _BlobClient:
+            def download_blob(self):
+                downloads.append(1)
+                return _Downloader()
+
+        obj = AzureBucketObject(
+            _FakeProvider(), _FakeAzureContainer(_BlobClient()),
+            _FakeBlobProperties("obj"))
+
+        chunks = list(obj.iter_content(chunk_size=1024))
+
+        self.assertEqual(b"".join(chunks), content)
+        self.assertEqual([len(c) for c in chunks], [1024] * 10)
+        self.assertEqual(set(reads), {1024},
+                         "Chunk size must be passed straight to read().")
+        self.assertEqual(len(downloads), 1,
+                         "The whole object should stream from a single "
+                         "download, not one request per chunk.")
+
+    def test_azure_uses_resolved_default_chunk_size(self):
+        from cloudbridge.providers.azure.resources import AzureBucketObject
+
+        reads = []
+
+        class _Downloader:
+            def read(self, size):
+                reads.append(size)
+                return b""
+
+        class _BlobClient:
+            def download_blob(self):
+                return _Downloader()
+
+        obj = AzureBucketObject(
+            _FakeProvider({'iter_chunk_size': 4096}),
+            _FakeAzureContainer(_BlobClient()), _FakeBlobProperties("obj"))
+
+        self.assertEqual(list(obj.iter_content()), [])
+        self.assertEqual(reads, [4096])
+
+    def test_gcp_fetches_successive_ranges_of_chunk_size(self):
+        from cloudbridge.providers.gcp.resources import GCPBucketObject
+
+        content = bytes(range(256)) * 40   # 10 KiB
+        ranges = []
+
+        class _BucketObjects:
+            def download_range(self, bucket, name, offset, length):
+                ranges.append((offset, length))
+                return content[offset:offset + length]
+
+        obj = GCPBucketObject(
+            _FakeProvider(bucket_objects=_BucketObjects()), "BUCKET",
+            {'name': 'obj', 'size': str(len(content))})
+
+        chunks = list(obj.iter_content(chunk_size=4096))
+
+        self.assertEqual(b"".join(chunks), content)
+        self.assertEqual(
+            ranges, [(0, 4096), (4096, 4096), (8192, 2048)],
+            "Ranges must tile the object exactly and the last must be "
+            "clamped to the object size, not overrun it.")
+
+    def test_gcp_reads_nothing_for_an_empty_object(self):
+        from cloudbridge.providers.gcp.resources import GCPBucketObject
+
+        obj = GCPBucketObject(
+            _FakeProvider(bucket_objects=_NoRangeService()), "BUCKET",
+            {'name': 'obj', 'size': '0'})
+        self.assertEqual(list(obj.iter_content()), [])
+
+    def test_gcp_rejects_bad_chunk_size_before_any_request(self):
+        from cloudbridge.providers.gcp.resources import GCPBucketObject
+
+        obj = GCPBucketObject(
+            _FakeProvider(bucket_objects=_NoRangeService()), "BUCKET",
+            {'name': 'obj', 'size': '100'})
+        with self.assertRaises(InvalidValueException):
+            obj.iter_content(chunk_size=0)
+
+    def test_openstack_passes_chunk_size_as_resp_chunk_size(self):
+        from cloudbridge.providers.openstack.resources import (
+            OpenStackBucketObject)
+
+        calls = []
+
+        class _Swift:
+            def get_object(self, container, name, resp_chunk_size=None):
+                calls.append(resp_chunk_size)
+                return {}, iter([b"data"])
+
+        provider = _FakeProvider()
+        provider.swift = _Swift()
+        obj = OpenStackBucketObject(
+            provider, _FakeSwiftContainer(), {'name': 'obj'})
+
+        self.assertEqual(list(obj.iter_content(chunk_size=8192)), [b"data"])
+        self.assertEqual(calls, [8192],
+                         "resp_chunk_size is what makes swiftclient stream "
+                         "rather than return the whole object.")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_object_store_service.py

@@ -178,6 +178,114 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
                     target_stream2.write(data)
                 self.assertEqual(target_stream2.getvalue(), content)
 
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_iter_content_honours_chunk_size(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("chunked_binary.bin")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                # Binary content with no newline in it at all: a reader that
+                # chunks by line rather than by size hands this back as one
+                # buffer, however large the object is.
+                content = bytes(
+                    b for b in range(256) if b != 0x0A) * 256   # ~64 KiB
+                obj.upload(content)
+
+                chunk_size = 4096
+                chunks = list(obj.iter_content(chunk_size=chunk_size))
+
+                self.assertEqual(
+                    b"".join(chunks), content,
+                    "Chunked read must reassemble to the original content.")
+                self.assertTrue(
+                    all(len(c) <= chunk_size for c in chunks),
+                    "No chunk may exceed the requested chunk_size of {0}, but "
+                    "sizes were {1}.".format(
+                        chunk_size, sorted({len(c) for c in chunks})))
+                self.assertGreaterEqual(
+                    len(chunks), len(content) // chunk_size,
+                    "A {0} byte object read at chunk_size={1} should yield at "
+                    "least {2} chunks, but yielded {3}.".format(
+                        len(content), chunk_size,
+                        len(content) // chunk_size, len(chunks)))
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_iter_content_chunks_by_size_not_by_line(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("chunked_lines.txt")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                # Newline every 10 bytes. Line-oriented chunking yields 10
+                # byte chunks regardless of chunk_size; size-oriented
+                # chunking fills each chunk.
+                content = b"123456789\n" * 1000
+                obj.upload(content)
+
+                chunk_size = 1024
+                chunks = list(obj.iter_content(chunk_size=chunk_size))
+
+                self.assertEqual(b"".join(chunks), content)
+                self.assertTrue(all(len(c) <= chunk_size for c in chunks))
+                # Sizing by chunk_size needs ~10 chunks here; splitting on
+                # newlines needs 1000. Allow slack for short reads, but not
+                # two orders of magnitude of it.
+                self.assertLessEqual(
+                    len(chunks), 2 * -(-len(content) // chunk_size),
+                    "A {0} byte object read at chunk_size={1} yielded {2} "
+                    "chunks, far more than size-based chunking needs - "
+                    "content is being split on newlines.".format(
+                        len(content), chunk_size, len(chunks)))
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_iter_content_default_chunk_size(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("default_chunked.bin")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                # Spans two default chunks, so the default actually reaches
+                # the wire rather than being hidden by a one-chunk object.
+                default = BaseBucketObject.CB_ITER_CHUNK_SIZE
+                content = b"\x00\xff" * default   # 2 x the default chunk
+                obj.upload(content)
+
+                chunks = list(obj.iter_content())
+
+                self.assertEqual(b"".join(chunks), content)
+                self.assertTrue(
+                    all(len(c) <= default for c in chunks),
+                    "Chunks must not exceed the default chunk size.")
+                self.assertLessEqual(
+                    len(chunks), 2 * (len(content) // default),
+                    "A {0} byte object should come back in about {1} chunks "
+                    "at the {2} byte default, but came back in {3}.".format(
+                        len(content), len(content) // default, default,
+                        len(chunks)))
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_save_content_honours_chunk_size(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("saved_chunked.bin")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                content = bytes(range(256)) * 100   # 25 KiB
+                obj.upload(content)
+
+                target_stream = BytesIO()
+                obj.save_content(target_stream, chunk_size=1024)
+                self.assertEqual(target_stream.getvalue(), content)
+
     @helpers.skipIfNoService(['storage.buckets'])
     def test_generate_url(self):
         name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())