feat(corpus): publish verified pairs concurrently

Author: vex0209-bt

docs/CORPUS_V2_PILOT_RUNBOOK.md

@@ -191,15 +191,18 @@ the successful publish gate permanently excludes them.
 ```bash
 venv/bin/leoma corpus v2 publish \
   --workdir "$WORK" \
-  --prefix corpus-v2/leoma-ti2v-pilot-10k-v1
+  --prefix corpus-v2/leoma-ti2v-pilot-10k-v1 \
+  --workers 64
 ```
 
 Every MP4 and PNG is uploaded under its own content digest, then checked by size and
 SHA-256 object metadata. A ledger row advances to `published` only after both objects
 verify. The default removes both local files only after that checkpoint. An
 interrupted command resumes the already-approved batch without requiring the same
 200 audited rows to remain unpublished. Use `--keep-local` only when NVMe capacity
-has been planned.
+has been planned. Concurrent publishing keeps a bounded set of pairs in flight;
+successful uploads remain content-addressed and resumable even if another worker
+fails before they are checkpointed.
 
 For a non-production smoke test with fewer than 200 total clips, the QA threshold can
 be explicitly reduced. Never use this override for the real corpus:

leoma/delivery/commands.py

@@ -568,14 +568,27 @@ def corpus_v2_qa_retry_rejected(workdir):
 @click.option("--workdir", default="corpus-v2-pilot", show_default=True)
 @click.option("--prefix", required=True, help="Versioned bucket prefix, e.g. corpus-v2/pilot-10k")
 @click.option("--limit", type=click.IntRange(min=1), default=None)
+@click.option(
+    "--workers",
+    type=click.IntRange(min=1, max=128),
+    default=1,
+    show_default=True,
+    help="Concurrent verified sample-pair uploads",
+)
 @click.option("--qa-min-reviews", type=click.IntRange(min=0), default=200, show_default=True)
 @click.option(
     "--qa-min-pass-rate", type=click.FloatRange(min=0.0, max=1.0),
     default=0.95, show_default=True,
 )
 @click.option("--delete-local-after-upload/--keep-local", default=True, show_default=True)
 def corpus_v2_publish(
-    workdir, prefix, limit, qa_min_reviews, qa_min_pass_rate, delete_local_after_upload,
+    workdir,
+    prefix,
+    limit,
+    workers,
+    qa_min_reviews,
+    qa_min_pass_rate,
+    delete_local_after_upload,
 ):
     """Upload captioned clip/first-frame pairs and checkpoint each verified sample."""
     from leoma.bootstrap import SOURCE_BUCKET, emit_header, emit_log
@@ -596,6 +609,7 @@ def corpus_v2_publish(
             ledger=ledger,
             prefix=prefix,
             limit=limit,
+            workers=workers,
             qa_min_reviews=qa_min_reviews,
             qa_min_pass_rate=qa_min_pass_rate,
             delete_local_after_upload=delete_local_after_upload,

leoma/infra/corpus_v2.py

@@ -1290,6 +1290,7 @@ def publish_captioned_samples(
     ledger: PilotLedger,
     prefix: str,
     limit: Optional[int] = None,
+    workers: int = 1,
     qa_min_reviews: int = 0,
     qa_min_pass_rate: float = 0.0,
     delete_local_after_upload: bool = False,
@@ -1299,6 +1300,8 @@ def publish_captioned_samples(
     clean_prefix = prefix.strip("/")
     if not clean_prefix:
         raise ValueError("a versioned publish prefix is required")
+    if workers < 1:
+        raise ValueError("publish workers must be at least one")
     ledger.approve_publish_batch(
         min_reviews=qa_min_reviews,
         min_pass_rate=qa_min_pass_rate,
@@ -1311,7 +1314,8 @@ def publish_captioned_samples(
         params = (int(limit),)
     with ledger.connect() as db:
         approved_rows = list(db.execute(query, params))
-    for row in approved_rows:
+
+    def upload_pair(row: sqlite3.Row) -> tuple[str, str, str, str, str]:
         clip_path = str(row["clip_path"])
         first_frame_path = str(row["first_frame_path"])
         if not os.path.isfile(clip_path):
@@ -1385,14 +1389,49 @@ def publish_captioned_samples(
                     f"s3://{bucket}/{first_frame_key}"
                 )
 
-        ledger.mark_published(row["sample_id"], clip_key, first_frame_key)
+        return (
+            str(row["sample_id"]),
+            clip_key,
+            first_frame_key,
+            clip_path,
+            first_frame_path,
+        )
+
+    def checkpoint(result: tuple[str, str, str, str, str]) -> None:
+        nonlocal published
+        sample_id, clip_key, first_frame_key, clip_path, first_frame_path = result
+        ledger.mark_published(sample_id, clip_key, first_frame_key)
         if delete_local_after_upload:
             Path(clip_path).unlink(missing_ok=True)
             Path(first_frame_path).unlink(missing_ok=True)
         published += 1
-        log(
-            f"published {row['sample_id']} -> {clip_key} + {first_frame_key}"
-        )
+        log(f"published {sample_id} -> {clip_key} + {first_frame_key}")
+
+    if workers == 1:
+        for row in approved_rows:
+            checkpoint(upload_pair(row))
+    else:
+        from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+
+        # Keep the in-flight set bounded. If a worker fails, at most twice the
+        # configured concurrency can have uploaded but not yet checkpointed.
+        # Resumption safely verifies and reuses those content-addressed objects.
+        pending = set()
+        with ThreadPoolExecutor(
+            max_workers=workers,
+            thread_name_prefix="corpus-v2-publish",
+        ) as pool:
+            for row in approved_rows:
+                pending.add(pool.submit(upload_pair, row))
+                if len(pending) < workers * 2:
+                    continue
+                completed, pending = wait(pending, return_when=FIRST_COMPLETED)
+                for future in completed:
+                    checkpoint(future.result())
+            while pending:
+                completed, pending = wait(pending, return_when=FIRST_COMPLETED)
+                for future in completed:
+                    checkpoint(future.result())
     return {**ledger.stats(), "published_this_run": published}
 
 

tests/unit/test_corpus_v2.py

@@ -574,6 +574,75 @@ def fput_object(self, bucket, key, path, *, content_type, metadata):
         self.objects[(bucket, key)] = (Path(path).read_bytes(), stored)
 
 
+class _ConcurrentMemoryObjectStore(_MemoryObjectStore):
+    def __init__(self):
+        super().__init__()
+        self._active = 0
+        self._lock = threading.Lock()
+        self.max_active = 0
+
+    def fput_object(self, bucket, key, path, *, content_type, metadata):
+        with self._lock:
+            self._active += 1
+            self.max_active = max(self.max_active, self._active)
+        try:
+            time.sleep(0.01)
+            super().fput_object(
+                bucket,
+                key,
+                path,
+                content_type=content_type,
+                metadata=metadata,
+            )
+        finally:
+            with self._lock:
+                self._active -= 1
+
+
+def test_publish_workers_upload_concurrently_and_checkpoint_before_delete(tmp_path):
+    ledger = PilotLedger(tmp_path / "pilot.sqlite3", _spec())
+    _bind_caption(ledger)
+    local_paths = []
+    for index in range(6):
+        original = _sample(
+            tmp_path,
+            sample_id=f"{index + 500:064x}",
+            truth="sha256:" + f"{index + 600:064x}",
+        )
+        sample = PreparedSample(**{
+            **original.__dict__,
+            "source_key": f"raw/concurrent-{index}.mp4",
+            "source_sha256": "sha256:" + f"{index + 700:064x}",
+            "source_start_ms": index * 10_000,
+        })
+        local_paths.extend((Path(sample.clip_path), Path(sample.first_frame_path)))
+        ledger.add_sample(sample)
+        task = ledger.claim_caption("gpu-0")
+        ledger.finish_caption(
+            task.sample_id,
+            caption="A person walks through a room while the camera follows behind.",
+            model="org/caption-model",
+            revision="f" * 40,
+            worker="gpu-0",
+            frame_count=16,
+        )
+
+    store = _ConcurrentMemoryObjectStore()
+    result = publish_captioned_samples(
+        store,
+        "bucket",
+        ledger=ledger,
+        prefix="corpus-v2/concurrent",
+        workers=3,
+        delete_local_after_upload=True,
+    )
+
+    assert result["published_this_run"] == 6
+    assert result["samples"] == {"published": 6}
+    assert store.max_active >= 2
+    assert not any(path.exists() for path in local_paths)
+
+
 def test_approved_publish_batch_resumes_without_requiring_the_audit_twice(tmp_path):
     ledger = PilotLedger(tmp_path / "pilot.sqlite3", _spec())
     _bind_caption(ledger)