feat: parallelize export downloads with asyncio.gather (#545)

previously tracks were downloaded sequentially, making exports slow
for users with many tracks or large files. now downloads happen
concurrently (up to 4 at a time) while zip creation remains sequential
(zipfile isn't thread-safe).

- refactor process_export to use asyncio.gather + semaphore
- add regression test verifying concurrent download behavior

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude <[e-mail]>

backend/src/backend/_internal/background_tasks.py

···
 requires DOCKET_URL to be set (Redis is always available).
 """
 
+import asyncio
 import logging
 import os
 import tempfile
···
 async def process_export(export_id: str, artist_did: str) -> None:
     """process a media export in the background.
 
-    downloads all tracks for the given artist, zips them, and uploads
-    to R2. progress is tracked via job_service.
+    downloads all tracks for the given artist concurrently, zips them,
+    and uploads to R2. progress is tracked via job_service.
 
     args:
         export_id: job ID for tracking progress
···
             zip_path = temp_path / f"{export_id}.zip"
             async_session = aioboto3.Session()
 
-            async with async_session.client(
-                "s3",
-                endpoint_url=settings.storage.r2_endpoint_url,
-                aws_access_key_id=settings.storage.aws_access_key_id,
-                aws_secret_access_key=settings.storage.aws_secret_access_key,
-            ) as s3_client:
-                with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zip_file:
-                    # track counter for duplicate titles
-                    title_counts: dict[str, int] = {}
-                    processed = 0
-                    total = len(tracks)
+            # prepare track metadata before downloading
+            title_counts: dict[str, int] = {}
+            track_info: list[dict] = []
 
-                    for track in tracks:
-                        if not track.file_id or not track.file_type:
-                            logfire.warn(
-                                "skipping track: missing file_id or file_type",
-                                track_id=track.id,
-                            )
-                            continue
+            for track in tracks:
+                if not track.file_id or not track.file_type:
+                    logfire.warn(
+                        "skipping track: missing file_id or file_type",
+                        track_id=track.id,
+                    )
+                    continue
 
-                        # construct R2 key
-                        key = f"audio/{track.file_id}.{track.file_type}"
+                # create safe filename with duplicate handling
+                base_filename = f"{track.title}.{track.file_type}"
+                if base_filename in title_counts:
+                    title_counts[base_filename] += 1
+                    filename = f"{track.title} ({title_counts[base_filename]}).{track.file_type}"
+                else:
+                    title_counts[base_filename] = 0
+                    filename = base_filename
 
-                        try:
-                            # update progress (current_track is 1-indexed for display)
-                            current_track = processed + 1
-                            pct = (processed / total) * 100
+                # sanitize filename (remove invalid chars)
+                filename = "".join(
+                    c
+                    for c in filename
+                    if c.isalnum() or c in (" ", ".", "-", "_", "(", ")")
+                )
+
+                track_info.append(
+                    {
+                        "track": track,
+                        "key": f"audio/{track.file_id}.{track.file_type}",
+                        "filename": filename,
+                        "temp_path": temp_path / filename,
+                    }
+                )
+
+            total = len(track_info)
+            if total == 0:
+                await job_service.update_progress(
+                    export_id,
+                    JobStatus.FAILED,
+                    "export failed",
+                    error="no valid tracks found to export",
+                )
+                return
+
+            # download counter for progress updates
+            downloaded = 0
+            download_lock = asyncio.Lock()
+
+            async def download_track(
+                info: dict,
+                s3_client,
+                semaphore: asyncio.Semaphore,
+            ) -> dict | None:
+                """download a single track, returning info on success or None on failure."""
+                nonlocal downloaded
+
+                async with semaphore:
+                    track = info["track"]
+                    try:
+                        response = await s3_client.get_object(
+                            Bucket=settings.storage.r2_bucket,
+                            Key=info["key"],
+                        )
+
+                        # stream to disk in chunks
+                        async with aiofiles.open(info["temp_path"], "wb") as f:
+                            async for chunk in response["Body"].iter_chunks():
+                                await f.write(chunk)
+
+                        # update progress
+                        async with download_lock:
+                            downloaded += 1
+                            pct = (downloaded / total) * 100
                             await job_service.update_progress(
                                 export_id,
                                 JobStatus.PROCESSING,
-                                f"downloading track {current_track} of {total}...",
+                                f"downloading tracks ({downloaded}/{total})...",
                                 progress_pct=pct,
                                 result={
-                                    "processed_count": processed,
+                                    "processed_count": downloaded,
                                     "total_count": total,
-                                    "current_track": current_track,
-                                    "current_title": track.title,
                                 },
                             )
 
-                            # create safe filename
-                            # handle duplicate titles by appending counter
-                            base_filename = f"{track.title}.{track.file_type}"
-                            if base_filename in title_counts:
-                                title_counts[base_filename] += 1
-                                filename = f"{track.title} ({title_counts[base_filename]}).{track.file_type}"
-                            else:
-                                title_counts[base_filename] = 0
-                                filename = base_filename
+                        logfire.info(
+                            "downloaded track: {track_title}",
+                            track_id=track.id,
+                            track_title=track.title,
+                        )
+                        return info
 
-                            # sanitize filename (remove invalid chars)
-                            filename = "".join(
-                                c
-                                for c in filename
-                                if c.isalnum() or c in (" ", ".", "-", "_", "(", ")")
-                            )
+                    except Exception as e:
+                        logfire.error(
+                            "failed to download track: {track_title}",
+                            track_id=track.id,
+                            track_title=track.title,
+                            error=str(e),
+                            _exc_info=True,
+                        )
+                        return None
 
-                            # download file to temp location, streaming to disk
-                            temp_file_path = temp_path / filename
-                            response = await s3_client.get_object(
-                                Bucket=settings.storage.r2_bucket,
-                                Key=key,
-                            )
+            # download all tracks concurrently (limit to 4 concurrent downloads)
+            await job_service.update_progress(
+                export_id,
+                JobStatus.PROCESSING,
+                f"downloading {total} tracks...",
+                progress_pct=0,
+                result={"processed_count": 0, "total_count": total},
+            )
 
-                            # stream to disk in chunks to avoid loading into memory
-                            async with aiofiles.open(temp_file_path, "wb") as f:
-                                async for chunk in response["Body"].iter_chunks():
-                                    await f.write(chunk)
+            semaphore = asyncio.Semaphore(4)
+            async with async_session.client(
+                "s3",
+                endpoint_url=settings.storage.r2_endpoint_url,
+                aws_access_key_id=settings.storage.aws_access_key_id,
+                aws_secret_access_key=settings.storage.aws_secret_access_key,
+            ) as s3_client:
+                results = await asyncio.gather(
+                    *[download_track(info, s3_client, semaphore) for info in track_info]
+                )
 
-                            # add to zip from disk (streams from file, doesn't load into memory)
-                            zip_file.write(temp_file_path, arcname=filename)
+            # filter out failed downloads
+            successful_downloads = [r for r in results if r is not None]
 
-                            # remove temp file immediately to free disk space
-                            os.unlink(temp_file_path)
+            # create zip file from downloaded tracks (sequential - zipfile not thread-safe)
+            await job_service.update_progress(
+                export_id,
+                JobStatus.PROCESSING,
+                "creating zip archive...",
+                progress_pct=100,
+                result={
+                    "processed_count": len(successful_downloads),
+                    "total_count": total,
+                },
+            )
 
-                            processed += 1
-                            logfire.info(
-                                "added track to export: {track_title}",
-                                track_id=track.id,
-                                track_title=track.title,
-                                filename=filename,
-                            )
+            with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zip_file:
+                for info in successful_downloads:
+                    zip_file.write(info["temp_path"], arcname=info["filename"])
+                    logfire.info(
+                        "added track to export: {track_title}",
+                        track_id=info["track"].id,
+                        track_title=info["track"].title,
+                        filename=info["filename"],
+                    )
+                    # remove temp file to free disk space
+                    os.unlink(info["temp_path"])
 
-                        except Exception as e:
-                            logfire.error(
-                                "failed to add track to export: {track_title}",
-                                track_id=track.id,
-                                track_title=track.title,
-                                error=str(e),
-                                _exc_info=True,
-                            )
-                            # continue with other tracks instead of failing entire export
+            processed = len(successful_downloads)
 
             # upload the zip file to R2 (still inside temp directory context)
             r2_key = f"exports/{export_id}.zip"

backend/tests/test_background_tasks.py

···
 """tests for background task scheduling."""
 
-from unittest.mock import MagicMock, patch
+import asyncio
+import contextlib
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import backend._internal.background_tasks as bg_tasks
 
···
         assert calls == [
             ("session-xyz", 42, "Test Track", "Test Artist", 180, "Test Album")
         ]
+
+
+async def test_process_export_downloads_concurrently() -> None:
+    """process_export should download tracks concurrently, not sequentially.
+
+    regression test: previously tracks were downloaded one at a time,
+    making exports slow for users with many tracks or large files.
+    """
+    download_times: list[float] = []
+    download_start_event = asyncio.Event()
+
+    async def mock_get_object(Bucket: str, Key: str) -> dict:
+        """track when downloads start and simulate network delay."""
+        download_times.append(asyncio.get_event_loop().time())
+        # signal that at least one download has started
+        download_start_event.set()
+        # simulate network delay
+        await asyncio.sleep(0.1)
+        # return mock response with async body
+        body = AsyncMock()
+        body.iter_chunks = lambda: async_chunk_gen()
+        return {"Body": body}
+
+    async def async_chunk_gen():
+        yield b"mock audio data"
+
+    # create mock tracks
+    mock_tracks = []
+    for i in range(4):
+        track = MagicMock()
+        track.id = i
+        track.title = f"Track {i}"
+        track.file_id = f"file_{i}"
+        track.file_type = "mp3"
+        mock_tracks.append(track)
+
+    # mock database query
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.all.return_value = mock_tracks
+
+    mock_db = AsyncMock()
+    mock_db.execute.return_value = mock_result
+
+    # mock S3 client
+    mock_s3 = AsyncMock()
+    mock_s3.get_object = mock_get_object
+
+    # mock session that returns mock s3 client
+    mock_session = MagicMock()
+    mock_session.client.return_value.__aenter__.return_value = mock_s3
+
+    # mock job service
+    mock_job_service = AsyncMock()
+
+    # mock aiofiles.open to be a no-op
+    mock_file = AsyncMock()
+    mock_file.__aenter__.return_value = mock_file
+    mock_file.__aexit__.return_value = None
+    mock_file.write = AsyncMock()
+
+    with (
+        patch(
+            "backend._internal.background_tasks.aioboto3.Session",
+            return_value=mock_session,
+        ),
+        patch(
+            "backend._internal.background_tasks.aiofiles.open", return_value=mock_file
+        ),
+        patch("backend._internal.background_tasks.zipfile.ZipFile"),
+        patch("backend._internal.background_tasks.os.unlink"),
+        patch("backend.utilities.database.db_session") as mock_db_session,
+        patch("backend._internal.jobs.job_service", mock_job_service),
+    ):
+        mock_db_session.return_value.__aenter__.return_value = mock_db
+
+        # run process_export but cancel before upload phase
+        # (we only care about testing download concurrency)
+        with contextlib.suppress(TimeoutError):
+            await asyncio.wait_for(
+                bg_tasks.process_export("export-123", "did:plc:testuser"),
+                timeout=2.0,
+            )
+
+    # verify downloads started concurrently:
+    # if sequential, each download would start ~0.1s after the previous
+    # if concurrent, all 4 downloads should start within ~0.05s of each other
+    assert len(download_times) == 4, f"expected 4 downloads, got {len(download_times)}"
+
+    # check that all downloads started within a small time window (concurrent)
+    # not spread out over 0.4s (sequential)
+    time_spread = max(download_times) - min(download_times)
+    assert time_spread < 0.1, (
+        f"downloads should start concurrently (within 0.1s), "
+        f"but time spread was {time_spread:.3f}s - likely still sequential"
+    )