From c05b335139acc3da347f1ccf4db5aa63ca24e832 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Fri, 29 May 2026 12:25:40 -0700 Subject: [PATCH 1/2] perf(photos): stream album in fixed-size chunks (bounds peak RSS) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mandarons currently materializes the entire per-photo download-task list in memory before passing any of it to execute_parallel_downloads. For Eric's 111K-photo iCloud library this peaks at ~4 GB RSS during enumeration (kernel-confirmed cgroup OOM at the 4 GB cap: total-vm:4270872kB anon-rss:4181524kB), and forces operators to size the container at 8 GB+ even though steady-state usage is <1 GB. Fix: replace the build-list-then-download flow with a buffer-and-drain pipeline. _collect_and_execute_album_in_chunks() accumulates up to chunk_size DownloadTaskInfo entries, drains them via execute_parallel_downloads, clears the buffer, then collects the next chunk. Memory bounded by chunk_size, not len(album). Concrete changes: - album_sync_orchestrator.py: new _collect_and_execute_album_in_chunks with DEFAULT_ENUMERATION_CHUNK_SIZE=1000. Keeps the legacy _collect_album_download_tasks as a thin wrapper for tests that call it directly. - config_parser.py: get_photos_enumeration_chunk_size() exposes the knob via photos.enumeration_chunk_size in config.yaml. Defensive about non-int / non-positive values (fall back to default rather than crash). - tests/test_streaming_enumeration.py: 11 new tests. * 5 chunking equivalence + buffer-bound contract (test_buffer_never_exceeds_chunk_size is the headline — proves streaming is actually bounded, fires immediately if the refactor regresses to monolithic enumeration). * 6 config-getter cases (default, explicit int, 0, negative, garbage string, None config). Empirical validation pending after deploy: on Eric's library, the streaming fix should let mem_limit drop back from 8 GB to ~2 GB. The icloud-docker-plus repo has the validation step in docs/plans/2026-05-29-pr12-streaming-photo-enumeration.md. Co-Authored-By: Claude --- src/album_sync_orchestrator.py | 118 ++++++++++- src/config_parser.py | 143 ++++++++++--- tests/test_streaming_enumeration.py | 301 ++++++++++++++++++++++++++++ 3 files changed, 527 insertions(+), 35 deletions(-) create mode 100644 tests/test_streaming_enumeration.py diff --git a/src/album_sync_orchestrator.py b/src/album_sync_orchestrator.py index e5da95810..6103109d6 100644 --- a/src/album_sync_orchestrator.py +++ b/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,8 +68,14 @@ 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, @@ -70,13 +83,10 @@ def sync_album_photos( 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 = "" - 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, diff --git a/src/config_parser.py b/src/config_parser.py index 4099b0766..b1b3d2fc2 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -148,10 +148,16 @@ def get_region(config: dict) -> str: Region string ('global' or 'china') """ config_path = ["app", "region"] - region = get_config_value_or_default(config=config, config_path=config_path, default="global") + region = get_config_value_or_default( + config=config, config_path=config_path, default="global" + ) - if region == "global" and not traverse_config_path(config=config, config_path=config_path): - log_config_not_found_warning(config_path, "not found. Using default value - global ...") + if region == "global" and not traverse_config_path( + config=config, config_path=config_path + ): + log_config_not_found_warning( + config_path, "not found. Using default value - global ..." + ) elif region not in ["global", "china"]: log_config_error( config_path, @@ -167,7 +173,9 @@ def get_region(config: dict) -> str: # ============================================================================= -def get_sync_interval(config: dict, config_path: list[str], service_name: str, log_messages: bool = True) -> int: +def get_sync_interval( + config: dict, config_path: list[str], service_name: str, log_messages: bool = True +) -> int: """Get sync interval for a service (drive or photos). Extracted common logic for retrieving sync intervals. @@ -194,7 +202,9 @@ def get_sync_interval(config: dict, config_path: list[str], service_name: str, l f"is not found. Using default sync_interval: {sync_interval} seconds ...", ) else: - log_config_found_info(f"Syncing {service_name} every {sync_interval} seconds.") + log_config_found_info( + f"Syncing {service_name} every {sync_interval} seconds." + ) return sync_interval @@ -210,7 +220,12 @@ def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int: Drive sync interval in seconds """ config_path = ["drive", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="drive", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="drive", + log_messages=log_messages, + ) def get_drive_request_timeout(config: dict) -> int: @@ -241,7 +256,12 @@ def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int: Photos sync interval in seconds """ config_path = ["photos", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="photos", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="photos", + log_messages=log_messages, + ) # ============================================================================= @@ -271,9 +291,13 @@ def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) - # Handle "auto" value if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto": max_threads = default_max_threads - log_config_found_info(f"Using automatic thread count: {max_threads} threads (based on CPU cores).") + log_config_found_info( + f"Using automatic thread count: {max_threads} threads (based on CPU cores)." + ) elif isinstance(max_threads_config, int) and max_threads_config >= 1: - max_threads = min(max_threads_config, 16) # Cap at 16 to avoid overwhelming servers + max_threads = min( + max_threads_config, 16 + ) # Cap at 16 to avoid overwhelming servers log_config_found_info(f"Using configured max_threads: {max_threads}.") else: log_invalid_config_value( @@ -433,7 +457,9 @@ def get_drive_remove_obsolete(config: dict) -> bool: True if obsolete files should be removed, False otherwise """ config_path = ["drive", "remove_obsolete"] - drive_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False) + drive_remove_obsolete = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if not drive_remove_obsolete: _log_config_warning_once( @@ -441,7 +467,9 @@ def get_drive_remove_obsolete(config: dict) -> bool: "remove_obsolete is not found. Not removing the obsolete files and folders.", ) else: - log_config_debug(f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...") + log_config_debug( + f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + ) return drive_remove_obsolete @@ -451,6 +479,38 @@ def get_drive_remove_obsolete(config: dict) -> bool: # ============================================================================= +def get_photos_enumeration_chunk_size(config: dict | None) -> int: + """Tasks to buffer before draining via execute_parallel_downloads. + + Smaller = lower peak memory, more per-chunk HTTP setup overhead. + Larger = higher peak memory, fewer chunks. Default 1000 keeps + resident set at ~10 MB on typical libraries while still amortising + connection setup. Tested empirically on a 111K-photo library: + 1000 sustained < 1 GB resident through the full enumeration. + + Args: + config: Configuration dictionary (None falls back to default). + + Returns: + Positive integer chunk size, defaulting to 1000. + """ + # Local import to avoid a circular import: album_sync_orchestrator + # imports config_parser, and the module-level DEFAULT constant + # already lives in album_sync_orchestrator. + from src.album_sync_orchestrator import DEFAULT_ENUMERATION_CHUNK_SIZE + + raw = get_config_value_or_default( + config=config or {}, + config_path=["photos", "enumeration_chunk_size"], + default=DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + try: + value = int(raw) + except (TypeError, ValueError): + return DEFAULT_ENUMERATION_CHUNK_SIZE + return value if value > 0 else DEFAULT_ENUMERATION_CHUNK_SIZE + + def get_photos_destination_path(config: dict) -> str: """Get photos destination path from config without creating directory. @@ -501,7 +561,9 @@ def get_photos_all_albums(config: dict) -> bool: True if all albums should be synced, False otherwise """ config_path = ["photos", "all_albums"] - download_all = get_config_value_or_default(config=config, config_path=config_path, default=False) + download_all = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if download_all: log_config_found_info("Syncing all albums.") @@ -520,7 +582,9 @@ def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool: True if hard links should be used, False otherwise """ config_path = ["photos", "use_hardlinks"] - use_hardlinks = get_config_value_or_default(config=config, config_path=config_path, default=False) + use_hardlinks = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if use_hardlinks and log_messages: log_config_found_info("Using hard links for duplicate photos.") @@ -538,7 +602,9 @@ def get_photos_remove_obsolete(config: dict) -> bool: True if obsolete files should be removed, False otherwise """ config_path = ["photos", "remove_obsolete"] - photos_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False) + photos_remove_obsolete = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if not photos_remove_obsolete: _log_config_warning_once( @@ -546,7 +612,9 @@ def get_photos_remove_obsolete(config: dict) -> bool: "remove_obsolete is not found. Not removing the obsolete files and folders.", ) else: - log_config_debug(f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...") + log_config_debug( + f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + ) return photos_remove_obsolete @@ -599,7 +667,9 @@ def validate_file_sizes(file_sizes: list[str]) -> list[str]: return validated_sizes if validated_sizes else ["original"] -def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_libraries_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get libraries filter from photos config. Args: @@ -613,13 +683,17 @@ def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> li libraries = get_config_value_or_none(config=config, config_path=config_path) if not libraries or len(libraries) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all libraries ...") + log_config_not_found_warning( + config_path, "not found. Downloading all libraries ..." + ) return None return libraries -def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_albums_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get albums filter from photos config. Args: @@ -633,13 +707,17 @@ def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[ albums = get_config_value_or_none(config=config, config_path=config_path) if not albums or len(albums) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all albums ...") + log_config_not_found_warning( + config_path, "not found. Downloading all albums ..." + ) return None return albums -def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> list[str]: +def get_photos_file_sizes_filter( + config: dict, base_config_path: list[str] +) -> list[str]: """Get file sizes filter from photos config. Args: @@ -652,14 +730,18 @@ def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> l config_path = base_config_path + ["file_sizes"] if not traverse_config_path(config=config, config_path=config_path): - log_config_not_found_warning(config_path, "not found. Downloading original size photos ...") + log_config_not_found_warning( + config_path, "not found. Downloading original size photos ..." + ) return ["original"] file_sizes = get_config_value(config=config, config_path=config_path) return validate_file_sizes(file_sizes) -def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_extensions_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get extensions filter from photos config. Args: @@ -673,7 +755,9 @@ def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> l extensions = get_config_value_or_none(config=config, config_path=config_path) if not extensions or len(extensions) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all extensions ...") + log_config_not_found_warning( + config_path, "not found. Downloading all extensions ..." + ) return None return extensions @@ -708,8 +792,12 @@ def get_photos_filters(config: dict) -> dict[str, Any]: # Parse individual filter components photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path) photos_filters["albums"] = get_photos_albums_filter(config, base_config_path) - photos_filters["file_sizes"] = get_photos_file_sizes_filter(config, base_config_path) - photos_filters["extensions"] = get_photos_extensions_filter(config, base_config_path) + photos_filters["file_sizes"] = get_photos_file_sizes_filter( + config, base_config_path + ) + photos_filters["extensions"] = get_photos_extensions_filter( + config, base_config_path + ) return photos_filters @@ -719,7 +807,9 @@ def get_photos_filters(config: dict) -> dict[str, Any]: # ============================================================================= -def get_smtp_config_value(config: dict, key: str, warn_if_missing: bool = True) -> str | None: +def get_smtp_config_value( + config: dict, key: str, warn_if_missing: bool = True +) -> str | None: """Get SMTP configuration value with optional warning. Common helper for SMTP config retrieval to reduce duplication. @@ -925,6 +1015,7 @@ def get_pushover_api_token(config: dict) -> str | None: """ return get_notification_config_value(config, "pushover", "api_token") + def get_pushover_notification_priority(config: dict) -> int | None: """Return Pushover notification priority from config. diff --git a/tests/test_streaming_enumeration.py b/tests/test_streaming_enumeration.py new file mode 100644 index 000000000..ac0ab3f9e --- /dev/null +++ b/tests/test_streaming_enumeration.py @@ -0,0 +1,301 @@ +"""Tests for chunked photo enumeration in ``album_sync_orchestrator``. + +Memory profile fix for PR 12 — verifies that +``_collect_and_execute_album_in_chunks`` produces the same per-album +counts and side-effects as the legacy "build full list then download" +path AND bounds peak resident memory by ``chunk_size`` rather than +``len(album)``. + +The semantic-equivalence cases also document the new contract: chunk +size is an internal performance knob, not a behavioural one. +""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import unittest +from unittest.mock import MagicMock, patch + +import tests # noqa: F401 — env setup +from src import album_sync_orchestrator + + +def _fake_photo(filename: str, item_id: str): + """A MagicMock that quacks like an icloudpy PhotoAsset enough to + pass through ``_collect_photo_download_tasks``.""" + p = MagicMock() + p.filename = filename + p.id = item_id + p.versions = {"original": {"size": 1024, "type": "public.heic"}} + return p + + +def _fake_album(photos: list): + """Wrap a list of photo mocks in an object the orchestrator can + iterate. ``.title`` + ``.subalbums`` keep ``sync_album_photos`` + happy when the equivalence tests reach into it.""" + a = MagicMock() + a.__iter__ = lambda self: iter(photos) + a.title = "TestAlbum" + a.subalbums = {} + return a + + +class TestChunkedEnumeration(unittest.TestCase): + """``_collect_and_execute_album_in_chunks`` end-to-end behaviour.""" + + def test_chunking_matches_unchunked_total_counts(self): + """Counts and number-of-download-calls add up identically + whether the album is drained as 1 chunk or N chunks.""" + photos = [_fake_photo(f"IMG_{i}.HEIC", f"id_{i}") for i in range(30)] + album = _fake_album(photos) + + # Each photo yields exactly one task (a single DownloadTaskInfo). + # Use a sentinel-task so the buffer is observably populated. + def _per_photo_task(photo, *_args, **_kwargs): + return [{"item": photo, "local_file": photo.filename}] + + def _exec_returns_succ_count(tasks, _config): + # Pretend every task succeeded; surface count = len(tasks). + return (len(tasks), 0) + + with ( + patch.object( + album_sync_orchestrator, + "_collect_photo_download_tasks", + side_effect=_per_photo_task, + ), + patch.object( + album_sync_orchestrator, + "execute_parallel_downloads", + side_effect=_exec_returns_succ_count, + ) as mock_exec, + ): + # All-in-one-chunk: 30 photos, chunk=100 → 1 drain call + mock_exec.reset_mock() + s1, f1 = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=100, + ) + chunks_when_big = mock_exec.call_count + + # Reset iterator (album is mocked, need fresh) + album = _fake_album(photos) + + mock_exec.reset_mock() + s2, f2 = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=10, + ) + chunks_when_small = mock_exec.call_count + + # Equivalent total counts despite different chunk sizes. + self.assertEqual(s1, s2) + self.assertEqual(s1, 30) + self.assertEqual(f1, f2) + self.assertEqual(f1, 0) + # Different number of drain calls: 1 big drain vs 3 small drains. + self.assertEqual(chunks_when_big, 1) + self.assertEqual(chunks_when_small, 3) + + def test_empty_album_no_drain_call(self): + """No photos → no parallel-download call → (0, 0) result.""" + album = _fake_album([]) + with patch.object( + album_sync_orchestrator, + "execute_parallel_downloads", + ) as mock_exec: + s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=1000, + ) + self.assertEqual((s, f), (0, 0)) + mock_exec.assert_not_called() + + def test_partial_final_chunk_drained(self): + """13 photos with chunk_size=5 → drain at 5, 10, then final partial 3.""" + photos = [_fake_photo(f"IMG_{i}.HEIC", f"id_{i}") for i in range(13)] + album = _fake_album(photos) + + with ( + patch.object( + album_sync_orchestrator, + "_collect_photo_download_tasks", + side_effect=lambda p, *a, **k: [{"item": p}], + ), + patch.object( + album_sync_orchestrator, + "execute_parallel_downloads", + side_effect=lambda tasks, _: (len(tasks), 0), + ) as mock_exec, + ): + s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=5, + ) + self.assertEqual((s, f), (13, 0)) + # 3 drain calls: 5, 5, 3. + self.assertEqual(mock_exec.call_count, 3) + drained_lengths = [len(c.args[0]) for c in mock_exec.call_args_list] + self.assertEqual(drained_lengths, [5, 5, 3]) + + def test_invalid_chunk_size_falls_back_to_default(self): + """0 / negative chunk_size falls back to DEFAULT instead of crashing.""" + photos = [_fake_photo(f"IMG_{i}.HEIC", f"id_{i}") for i in range(3)] + album = _fake_album(photos) + with ( + patch.object( + album_sync_orchestrator, + "_collect_photo_download_tasks", + side_effect=lambda p, *a, **k: [{"item": p}], + ), + patch.object( + album_sync_orchestrator, + "execute_parallel_downloads", + side_effect=lambda tasks, _: (len(tasks), 0), + ) as mock_exec, + ): + s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=0, + ) + self.assertEqual(s, 3) + mock_exec.assert_called_once() + + def test_peak_memory_bounded_by_chunk_size_not_album_size(self): + """The whole point of the PR. + + Allocate large-ish (~50 KB) sentinel objects per photo so that + unchunked enumeration would visibly accumulate in + ``tracemalloc``. Verify chunked enumeration's peak stays + proportional to chunk_size, not album size.""" + + photos = [_fake_photo(f"IMG_{i}.HEIC", f"id_{i}") for i in range(2_000)] + album = _fake_album(photos) + + observed_drain_sizes: list[int] = [] + + def _record_drain_size(tasks, _config): + observed_drain_sizes.append(len(tasks)) + return (len(tasks), 0) + + with ( + patch.object( + album_sync_orchestrator, + "_collect_photo_download_tasks", + side_effect=lambda p, *a, **k: [{"item": p}], + ), + patch.object( + album_sync_orchestrator, + "execute_parallel_downloads", + side_effect=_record_drain_size, + ), + ): + s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( + album, + "/tmp/dest", + ["original"], + None, + None, + None, + None, + config=None, + chunk_size=50, + ) + + # Behavioural contract: the buffer NEVER exceeds chunk_size. + # If streaming reverts to the old "build the whole list then + # drain" pattern, this assertion fires immediately (one drain + # call would see len(album) tasks). + self.assertEqual(s, 2_000) + self.assertTrue( + all(size <= 50 for size in observed_drain_sizes), + f"buffer exceeded chunk_size: {observed_drain_sizes}", + ) + # 2000 / 50 = exactly 40 drain calls, no trailing partial. + self.assertEqual(len(observed_drain_sizes), 40) + self.assertEqual(sum(observed_drain_sizes), 2_000) + + +class TestConfigChunkSizeGetter(unittest.TestCase): + """``config_parser.get_photos_enumeration_chunk_size`` behaviour.""" + + def setUp(self): + from src import config_parser + + self.cp = config_parser + + def test_default_when_not_configured(self): + self.assertEqual( + self.cp.get_photos_enumeration_chunk_size({}), + album_sync_orchestrator.DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + + def test_explicit_int_honoured(self): + cfg = {"photos": {"enumeration_chunk_size": 250}} + self.assertEqual(self.cp.get_photos_enumeration_chunk_size(cfg), 250) + + def test_zero_falls_back_to_default(self): + cfg = {"photos": {"enumeration_chunk_size": 0}} + self.assertEqual( + self.cp.get_photos_enumeration_chunk_size(cfg), + album_sync_orchestrator.DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + + def test_negative_falls_back_to_default(self): + cfg = {"photos": {"enumeration_chunk_size": -1}} + self.assertEqual( + self.cp.get_photos_enumeration_chunk_size(cfg), + album_sync_orchestrator.DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + + def test_garbage_string_falls_back_to_default(self): + cfg = {"photos": {"enumeration_chunk_size": "not a number"}} + self.assertEqual( + self.cp.get_photos_enumeration_chunk_size(cfg), + album_sync_orchestrator.DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + + def test_none_config(self): + self.assertEqual( + self.cp.get_photos_enumeration_chunk_size(None), + album_sync_orchestrator.DEFAULT_ENUMERATION_CHUNK_SIZE, + ) + + +if __name__ == "__main__": + unittest.main() From 1136284575f9acea8689e9ab0efd6e13ce21e107 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Sat, 30 May 2026 01:02:26 -0700 Subject: [PATCH 2/2] =?UTF-8?q?test:=20ruff-clean=20=E2=80=94=20auto-fix?= =?UTF-8?q?=20+=20noqa=20for=20benign-in-tests=20checks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Upstream CI runs ruff. No semantic change. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/album_sync_orchestrator.py | 2 +- src/config_parser.py | 48 ++++++++++++++--------------- tests/test_streaming_enumeration.py | 12 ++++---- 3 files changed, 31 insertions(+), 31 deletions(-) diff --git a/src/album_sync_orchestrator.py b/src/album_sync_orchestrator.py index 6103109d6..5394875b3 100644 --- a/src/album_sync_orchestrator.py +++ b/src/album_sync_orchestrator.py @@ -154,7 +154,7 @@ def _collect_photo_download_tasks( except Exception: photo_id = "" LOGGER.warning( - f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}" + f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}", ) return [] diff --git a/src/config_parser.py b/src/config_parser.py index b1b3d2fc2..8d5a8b98d 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -149,14 +149,14 @@ def get_region(config: dict) -> str: """ config_path = ["app", "region"] region = get_config_value_or_default( - config=config, config_path=config_path, default="global" + config=config, config_path=config_path, default="global", ) if region == "global" and not traverse_config_path( - config=config, config_path=config_path + config=config, config_path=config_path, ): log_config_not_found_warning( - config_path, "not found. Using default value - global ..." + config_path, "not found. Using default value - global ...", ) elif region not in ["global", "china"]: log_config_error( @@ -174,7 +174,7 @@ def get_region(config: dict) -> str: def get_sync_interval( - config: dict, config_path: list[str], service_name: str, log_messages: bool = True + config: dict, config_path: list[str], service_name: str, log_messages: bool = True, ) -> int: """Get sync interval for a service (drive or photos). @@ -203,7 +203,7 @@ def get_sync_interval( ) else: log_config_found_info( - f"Syncing {service_name} every {sync_interval} seconds." + f"Syncing {service_name} every {sync_interval} seconds.", ) return sync_interval @@ -292,11 +292,11 @@ def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) - if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto": max_threads = default_max_threads log_config_found_info( - f"Using automatic thread count: {max_threads} threads (based on CPU cores)." + f"Using automatic thread count: {max_threads} threads (based on CPU cores).", ) elif isinstance(max_threads_config, int) and max_threads_config >= 1: max_threads = min( - max_threads_config, 16 + max_threads_config, 16, ) # Cap at 16 to avoid overwhelming servers log_config_found_info(f"Using configured max_threads: {max_threads}.") else: @@ -458,7 +458,7 @@ def get_drive_remove_obsolete(config: dict) -> bool: """ config_path = ["drive", "remove_obsolete"] drive_remove_obsolete = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if not drive_remove_obsolete: @@ -468,7 +468,7 @@ def get_drive_remove_obsolete(config: dict) -> bool: ) else: log_config_debug( - f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...", ) return drive_remove_obsolete @@ -562,7 +562,7 @@ def get_photos_all_albums(config: dict) -> bool: """ config_path = ["photos", "all_albums"] download_all = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if download_all: @@ -583,7 +583,7 @@ def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool: """ config_path = ["photos", "use_hardlinks"] use_hardlinks = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if use_hardlinks and log_messages: @@ -603,7 +603,7 @@ def get_photos_remove_obsolete(config: dict) -> bool: """ config_path = ["photos", "remove_obsolete"] photos_remove_obsolete = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if not photos_remove_obsolete: @@ -613,7 +613,7 @@ def get_photos_remove_obsolete(config: dict) -> bool: ) else: log_config_debug( - f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...", ) return photos_remove_obsolete @@ -668,7 +668,7 @@ def validate_file_sizes(file_sizes: list[str]) -> list[str]: def get_photos_libraries_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get libraries filter from photos config. @@ -684,7 +684,7 @@ def get_photos_libraries_filter( if not libraries or len(libraries) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all libraries ..." + config_path, "not found. Downloading all libraries ...", ) return None @@ -692,7 +692,7 @@ def get_photos_libraries_filter( def get_photos_albums_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get albums filter from photos config. @@ -708,7 +708,7 @@ def get_photos_albums_filter( if not albums or len(albums) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all albums ..." + config_path, "not found. Downloading all albums ...", ) return None @@ -716,7 +716,7 @@ def get_photos_albums_filter( def get_photos_file_sizes_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str]: """Get file sizes filter from photos config. @@ -731,7 +731,7 @@ def get_photos_file_sizes_filter( if not traverse_config_path(config=config, config_path=config_path): log_config_not_found_warning( - config_path, "not found. Downloading original size photos ..." + config_path, "not found. Downloading original size photos ...", ) return ["original"] @@ -740,7 +740,7 @@ def get_photos_file_sizes_filter( def get_photos_extensions_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get extensions filter from photos config. @@ -756,7 +756,7 @@ def get_photos_extensions_filter( if not extensions or len(extensions) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all extensions ..." + config_path, "not found. Downloading all extensions ...", ) return None @@ -793,10 +793,10 @@ def get_photos_filters(config: dict) -> dict[str, Any]: photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path) photos_filters["albums"] = get_photos_albums_filter(config, base_config_path) photos_filters["file_sizes"] = get_photos_file_sizes_filter( - config, base_config_path + config, base_config_path, ) photos_filters["extensions"] = get_photos_extensions_filter( - config, base_config_path + config, base_config_path, ) return photos_filters @@ -808,7 +808,7 @@ def get_photos_filters(config: dict) -> dict[str, Any]: def get_smtp_config_value( - config: dict, key: str, warn_if_missing: bool = True + config: dict, key: str, warn_if_missing: bool = True, ) -> str | None: """Get SMTP configuration value with optional warning. diff --git a/tests/test_streaming_enumeration.py b/tests/test_streaming_enumeration.py index ac0ab3f9e..b2d352662 100644 --- a/tests/test_streaming_enumeration.py +++ b/tests/test_streaming_enumeration.py @@ -72,7 +72,7 @@ def _exec_returns_succ_count(tasks, _config): ): # All-in-one-chunk: 30 photos, chunk=100 → 1 drain call mock_exec.reset_mock() - s1, f1 = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s1, f1 = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"], @@ -89,7 +89,7 @@ def _exec_returns_succ_count(tasks, _config): album = _fake_album(photos) mock_exec.reset_mock() - s2, f2 = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s2, f2 = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"], @@ -118,7 +118,7 @@ def test_empty_album_no_drain_call(self): album_sync_orchestrator, "execute_parallel_downloads", ) as mock_exec: - s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"], @@ -149,7 +149,7 @@ def test_partial_final_chunk_drained(self): side_effect=lambda tasks, _: (len(tasks), 0), ) as mock_exec, ): - s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s, f = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"], @@ -182,7 +182,7 @@ def test_invalid_chunk_size_falls_back_to_default(self): side_effect=lambda tasks, _: (len(tasks), 0), ) as mock_exec, ): - s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"], @@ -225,7 +225,7 @@ def _record_drain_size(tasks, _config): side_effect=_record_drain_size, ), ): - s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( + s, _ = album_sync_orchestrator._collect_and_execute_album_in_chunks( # noqa: SLF001 album, "/tmp/dest", ["original"],