feat: add --upload-blobs for processing deferred blob manifests

Reads a TSV manifest (from --deferred-blobs) and uploads blobs
to S3 with parallel workers and retry logic.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jensens

src/zodb_convert/cli.py

@@ -115,7 +115,13 @@ def parse_args(argv):
     args = parser.parse_args(argv)
 
     # Require at least one source/destination specification
-    if not args.config_file and not args.source_zope_conf and not args.dest_zope_conf:
+    # (--upload-blobs only needs a destination, validated later)
+    if (
+        not args.config_file
+        and not args.source_zope_conf
+        and not args.dest_zope_conf
+        and not args.upload_blobs
+    ):
         parser.error(
             "At least one of config_file, --source-zope-conf, or --dest-zope-conf is required."
         )
@@ -145,11 +151,80 @@ def _setup_logging(verbose):
     logging.getLogger("zodb-convert").setLevel(level)
 
 
+def _open_destination(args):
+    """Open only the destination storage from CLI args.
+
+    Returns (destination_storage, closables).
+    """
+    from zodb_convert.config import open_storage_from_zope_conf
+    from zodb_convert.config import open_storages_from_config
+
+    destination = None
+    closables = []
+
+    if args.config_file:
+        _cfg_source, cfg_dest = open_storages_from_config(args.config_file)
+        if cfg_dest is not None:
+            destination = cfg_dest
+        # Close unused source if opened
+        if _cfg_source is not None:
+            closables.append(_cfg_source)
+
+    if args.dest_zope_conf:
+        if destination is not None:
+            raise ValueError(
+                "Destination specified in both config file and --dest-zope-conf"
+            )
+        db = open_storage_from_zope_conf(args.dest_zope_conf, args.dest_db)
+        destination = db.storage
+        closables.append(db)
+
+    if destination is None:
+        raise ValueError(
+            "No destination storage configured. Use a config file or --dest-zope-conf."
+        )
+
+    return destination, closables
+
+
 def main(argv=None):
     """Main entry point for zodb-convert."""
     args = parse_args(argv if argv is not None else sys.argv[1:])
     _setup_logging(args.verbose)
 
+    if args.upload_blobs:
+        from zodb_convert.manifest import upload_from_manifest
+
+        closables = []
+        try:
+            destination, closables = _open_destination(args)
+            s3_client = getattr(destination, "_s3_client", None)
+            if s3_client is None:
+                log.error("Destination storage has no S3 client configured")
+                return 1
+            stats = upload_from_manifest(
+                args.upload_blobs,
+                s3_client,
+                workers=args.workers or 8,
+            )
+            if stats["failed"]:
+                log.error("%d blob upload(s) failed", stats["failed"])
+                return 1
+            return 0
+        except KeyboardInterrupt:
+            log.warning("Interrupted by user, aborting...")
+            return 130
+        except (ValueError, FileNotFoundError) as e:
+            log.error("%s", e)
+            sys.exit(1)
+        except Exception as e:
+            log.error("Upload failed: %s", e, exc_info=True)
+            sys.exit(2)
+        finally:
+            for obj in closables:
+                with contextlib.suppress(Exception):
+                    obj.close()
+
     closables = []
     try:
         source, destination, closables = open_storages(args)

src/zodb_convert/manifest.py

@@ -0,0 +1,105 @@
+"""Upload deferred blobs from a manifest file."""
+
+from concurrent.futures import as_completed
+from concurrent.futures import ThreadPoolExecutor
+
+import contextlib
+import logging
+import os
+import time
+
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_MAX_RETRIES = 3
+_DEFAULT_RETRY_BASE_DELAY = 2.0
+
+
+def upload_from_manifest(
+    manifest_path,
+    s3_client,
+    workers=8,
+    max_retries=_DEFAULT_MAX_RETRIES,
+    retry_base_delay=_DEFAULT_RETRY_BASE_DELAY,
+    cleanup=False,
+):
+    """Read manifest TSV and upload blobs to S3.
+
+    Returns dict with counts: uploaded, failed, skipped.
+    """
+    entries = []
+    with open(manifest_path) as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            parts = line.split("\t")
+            if len(parts) != 4:
+                logger.warning("Skipping malformed manifest line: %s", line)
+                continue
+            blob_path, s3_key, zoid_str, size_str = parts
+            entries.append((blob_path, s3_key, int(zoid_str), int(size_str)))
+
+    logger.info("Manifest: %d blob(s) to upload", len(entries))
+
+    uploaded = 0
+    failed = 0
+    skipped = 0
+
+    def _upload_one(blob_path, s3_key, zoid, size):
+        if not os.path.exists(blob_path):
+            logger.warning(
+                "Blob file missing, skipping: %s (oid=0x%016x)", blob_path, zoid
+            )
+            return "skipped"
+        last_exc = None
+        for attempt in range(max_retries):
+            try:
+                s3_client.upload_file(blob_path, s3_key)
+                if cleanup:
+                    with contextlib.suppress(OSError):
+                        os.unlink(blob_path)
+                return "uploaded"
+            except Exception as exc:
+                last_exc = exc
+                if attempt < max_retries - 1:
+                    delay = retry_base_delay * (2**attempt)
+                    logger.warning(
+                        "Upload oid=0x%016x attempt %d/%d failed (%s), "
+                        "retrying in %.0fs ...",
+                        zoid,
+                        attempt + 1,
+                        max_retries,
+                        exc,
+                        delay,
+                    )
+                    time.sleep(delay)
+        logger.error(
+            "Upload oid=0x%016x FAILED after %d attempts: %s",
+            zoid,
+            max_retries,
+            last_exc,
+        )
+        return "failed"
+
+    t0 = time.time()
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        futures = {pool.submit(_upload_one, *entry): entry[1] for entry in entries}
+        for fut in as_completed(futures):
+            result = fut.result()
+            if result == "uploaded":
+                uploaded += 1
+            elif result == "failed":
+                failed += 1
+            elif result == "skipped":
+                skipped += 1
+
+    elapsed = time.time() - t0
+    logger.info(
+        "Manifest upload complete: %d uploaded, %d failed, %d skipped (%.1fs)",
+        uploaded,
+        failed,
+        skipped,
+        elapsed,
+    )
+    return {"uploaded": uploaded, "failed": failed, "skipped": skipped}

tests/test_manifest.py

@@ -0,0 +1,61 @@
+from unittest.mock import MagicMock
+from zodb_convert.manifest import upload_from_manifest
+
+
+class TestUploadFromManifest:
+    def test_uploads_all_entries(self, tmp_path):
+        s3 = MagicMock()
+        manifest = tmp_path / "manifest.tsv"
+
+        entries = []
+        for i in range(3):
+            blob = tmp_path / f"blob{i}"
+            blob.write_bytes(f"data{i}".encode())
+            entries.append(f"{blob}\tblobs/{i}.blob\t{i}\t5")
+        manifest.write_text("\n".join(entries) + "\n")
+
+        stats = upload_from_manifest(str(manifest), s3, workers=2)
+
+        assert s3.upload_file.call_count == 3
+        assert stats["uploaded"] == 3
+        assert stats["failed"] == 0
+
+    def test_cleans_temp_files_after_upload(self, tmp_path):
+        s3 = MagicMock()
+        manifest = tmp_path / "manifest.tsv"
+        blob = tmp_path / "blob0"
+        blob.write_bytes(b"data")
+        manifest.write_text(f"{blob}\tblobs/0.blob\t0\t4\n")
+
+        upload_from_manifest(str(manifest), s3, workers=1, cleanup=True)
+
+        assert not blob.exists()
+
+    def test_retries_and_reports_failures(self, tmp_path):
+        s3 = MagicMock()
+        s3.upload_file.side_effect = Exception("permanent")
+        manifest = tmp_path / "manifest.tsv"
+        blob = tmp_path / "blob0"
+        blob.write_bytes(b"data")
+        manifest.write_text(f"{blob}\tblobs/0.blob\t0\t4\n")
+
+        stats = upload_from_manifest(
+            str(manifest),
+            s3,
+            workers=1,
+            max_retries=1,
+            retry_base_delay=0,
+        )
+
+        assert stats["failed"] == 1
+        assert stats["uploaded"] == 0
+
+    def test_skips_missing_files(self, tmp_path):
+        s3 = MagicMock()
+        manifest = tmp_path / "manifest.tsv"
+        manifest.write_text("/nonexistent/blob\tblobs/0.blob\t0\t4\n")
+
+        stats = upload_from_manifest(str(manifest), s3, workers=1)
+
+        assert stats["skipped"] == 1
+        assert s3.upload_file.call_count == 0