perf(photos): stream album enumeration in fixed-size chunks (bounds peak RSS on large libraries)

src/album_sync_orchestrator.py

@@ -9,7 +9,7 @@
 import os
 from typing import Any
 
-from src import get_logger
+from src import config_parser, get_logger
 from src.hardlink_registry import HardlinkRegistry
 from src.photo_download_manager import (
     DownloadTaskInfo,
@@ -21,6 +21,13 @@
 
 LOGGER = get_logger()
 
+# Default chunk size for streaming photo enumeration. Picked to keep peak
+# RSS bounded at ~10–20 MB of DownloadTaskInfo objects per chunk on
+# typical iCloud libraries while still giving execute_parallel_downloads
+# enough work to amortize HTTP connection setup. Users can override via
+# ``photos.enumeration_chunk_size`` in config.yaml.
+DEFAULT_ENUMERATION_CHUNK_SIZE = 1000
+
 
 def sync_album_photos(
     album,
@@ -61,22 +68,25 @@ def sync_album_photos(
     os.makedirs(normalized_destination, exist_ok=True)
     LOGGER.info(f"Syncing {album.title}")
 
-    # Collect download tasks for photos
-    download_tasks = _collect_album_download_tasks(
+    # Stream the album in fixed-size chunks: collect → download →
+    # release → next chunk. Memory is bounded by chunk_size × per-task
+    # size (~10 MB at chunk=1000) instead of len(album) × per-task size
+    # (which OOM-kills containers on ~100K+ libraries: empirically a
+    # 111K-photo library peaks at ~4 GB RSS without chunking, kernel-
+    # confirmed via cgroup OOM at the 4 GB cap).
+    chunk_size = config_parser.get_photos_enumeration_chunk_size(config=config)
+    total_successful, total_failed = _collect_and_execute_album_in_chunks(
         album,
         normalized_destination,
         file_sizes,
         extensions,
         files,
         folder_format,
         hardlink_registry,
+        config,
+        chunk_size=chunk_size,
     )
 
-    # Execute downloads in parallel if there are tasks
-    total_successful, total_failed = 0, 0
-    if download_tasks:
-        total_successful, total_failed = execute_parallel_downloads(download_tasks, config)
-
     # Recursively sync subalbums and aggregate counts
     sub_successful, sub_failed = _sync_subalbums(
         album,
@@ -143,7 +153,9 @@ def _collect_photo_download_tasks(
             photo_id = photo.id
         except Exception:
             photo_id = "<unknown>"
-        LOGGER.warning(f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}")
+        LOGGER.warning(
+            f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}",
+        )
         return []
 
 
@@ -158,6 +170,12 @@ def _collect_album_download_tasks(
 ) -> list:
     """Collect download tasks for all photos in an album.
 
+    .. deprecated::
+        Materializes the full task list — unbounded memory on large
+        libraries. Kept as a thin wrapper for test backward-compat
+        only. Production code path goes through
+        ``_collect_and_execute_album_in_chunks`` which streams instead.
+
     Args:
         album: Album object from iCloudPy
         destination_path: Path where photos should be saved
@@ -188,6 +206,88 @@ def _collect_album_download_tasks(
     return download_tasks
 
 
+def _collect_and_execute_album_in_chunks(
+    album,
+    destination_path: str,
+    file_sizes: list[str],
+    extensions: list[str] | None,
+    files: set[str] | None,
+    folder_format: str | None,
+    hardlink_registry: HardlinkRegistry | None,
+    config,
+    chunk_size: int = DEFAULT_ENUMERATION_CHUNK_SIZE,
+) -> tuple[int, int]:
+    """Stream album → fixed-size chunks → download → release.
+
+    Buffers up to ``chunk_size`` download tasks, then drains them via
+    ``execute_parallel_downloads`` and clears the buffer before
+    collecting the next chunk. Memory is bounded by chunk_size, not by
+    len(album). Semantically equivalent to building the full task list
+    and downloading once — same total counts, same per-photo
+    side-effects — but resident-set stays flat instead of growing
+    monotonically through enumeration.
+
+    Args:
+        album: Album object from iCloudPy
+        destination_path: Path where photos should be saved
+        file_sizes: List of file size variants to download
+        extensions: List of allowed file extensions
+        files: Set to track downloaded files
+        folder_format: strftime format string for folder organization
+        hardlink_registry: Registry for tracking downloaded files
+        config: Configuration dictionary (passed through to
+            ``execute_parallel_downloads`` for per-album thread count)
+        chunk_size: Tasks to buffer before draining. Smaller = lower
+            peak memory but more per-chunk HTTP setup overhead.
+
+    Returns:
+        Tuple of (total_successful, total_failed) summed across chunks.
+    """
+    if chunk_size <= 0:
+        # Degenerate config; fall back to default rather than refusing
+        # to sync. Logging is at INFO so operators see the fallback.
+        LOGGER.info(
+            f"Invalid photos.enumeration_chunk_size={chunk_size!r}; "
+            f"using default {DEFAULT_ENUMERATION_CHUNK_SIZE}.",
+        )
+        chunk_size = DEFAULT_ENUMERATION_CHUNK_SIZE
+
+    buffer: list[DownloadTaskInfo] = []
+    total_successful = 0
+    total_failed = 0
+
+    def _drain():
+        nonlocal total_successful, total_failed, buffer
+        if not buffer:
+            return
+        succ, fail = execute_parallel_downloads(buffer, config)
+        total_successful += succ
+        total_failed += fail
+        # Explicit reassign-to-empty so the chunk's task objects (and
+        # the photo references they hold) become collectable as soon
+        # as the parallel-download call returns.
+        buffer = []
+
+    for photo in album:
+        buffer.extend(
+            _collect_photo_download_tasks(
+                photo,
+                destination_path,
+                file_sizes,
+                extensions,
+                files,
+                folder_format,
+                hardlink_registry,
+            ),
+        )
+        if len(buffer) >= chunk_size:
+            _drain()
+
+    _drain()  # final partial chunk
+
+    return total_successful, total_failed
+
+
 def _sync_subalbums(
     album,
     destination_path: str,