Batch log uploads to avoid HF 256-commits/hour rate limit

Replaces per-inference commit (prune + upload = 2 commits) with a
LogUploader class that queues parquet files locally and flushes them
in a single batched commit every LOG_BATCH_INTERVAL seconds (default 60).
At full load this reduces commit rate from O(requests) to ~1/minute.

Also splits _build_commit_ops into _build_add_ops and _build_delete_ops
so each helper has a single responsibility, and updates tests accordingly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app.py

@@ -14,7 +14,14 @@ import json
 import html as html_lib
 from io import BytesIO
 from PIL import Image
-from logging_utils import log_inference
+from logging_utils import LogUploader
+
+_log_uploader = LogUploader(
+    token=os.environ.get("HF_TOKEN"),
+    repo_id=os.environ.get("LOG_DATASET_REPO"),
+    max_files=int(os.environ.get("LOG_MAX_FILES", "5000")),
+    batch_interval=int(os.environ.get("LOG_BATCH_INTERVAL", "60")),
+)
 
 MAX_SEED = np.iinfo(np.int32).max
 LANCZOS = getattr(Image, "Resampling", Image).LANCZOS
@@ -295,7 +302,7 @@ def _resolve_seed(seed: int, randomize_seed: bool) -> int:
 def _spawn_log(pil_images, result_image, prompt, seed, steps, guidance_scale,
                width, height, duration, success, error=""):
     threading.Thread(
-        target=log_inference,
+        target=_log_uploader.log_inference,
         args=(pil_images, result_image, prompt, seed, steps, guidance_scale,
               width, height, duration, success, error),
         daemon=True,

logging_utils.py

@@ -1,12 +1,10 @@
 import os
 import uuid
+import threading
+import time as _time
 from io import BytesIO
-from datetime import datetime, timezone, timedelta
-from huggingface_hub import hf_hub_download, CommitOperationDelete
-
-HF_TOKEN      = os.environ.get("HF_TOKEN")
-DATASET_REPO  = os.environ.get("LOG_DATASET_REPO")
-MAX_LOG_FILES = int(os.environ.get("LOG_MAX_FILES", "5000"))
+from datetime import datetime, timezone
+from huggingface_hub import hf_hub_download, CommitOperationAdd, CommitOperationDelete
 
 
 def _img_to_jpeg(img, quality=85):
@@ -74,40 +72,51 @@ def _build_table(pil_inputs, output_pil, prompt, seed, steps, guidance_scale,
     }, schema=schema)
 
 
-def _upload_parquet(api, repo_id, table, path_in_repo):
+def _write_parquet(table):
     import tempfile
     import pyarrow.parquet as pq
-
-    tmp_path = None
-    try:
-        with tempfile.NamedTemporaryFile(suffix=".parquet", delete=False) as tmp:
-            tmp_path = tmp.name
-        pq.write_table(table, tmp_path)
-        print(f"[log] uploading {path_in_repo} ({os.path.getsize(tmp_path)//1024}KB)")
-        api.upload_file(
-            path_or_fileobj=tmp_path, path_in_repo=path_in_repo,
-            repo_id=repo_id, repo_type="dataset",
-        )
-        print(f"[log] upload done — {repo_id}/{path_in_repo}")
-    finally:
-        if tmp_path:
-            try:
-                os.unlink(tmp_path)
-            except Exception as e:
-                print(f"[log] failed to delete temp file {tmp_path}: {e}")
+    with tempfile.NamedTemporaryFile(suffix=".parquet", delete=False) as tmp:
+        path = tmp.name
+    pq.write_table(table, path)
+    return path
 
 
 def _make_path(now, uid):
     return f"data/{now.strftime('%Y-%m-%d-%H%M%S')}-{uid}.parquet"
 
 
-def _file_date(path):
-    return os.path.basename(path)[:10]
+def _list_existing_files(api, repo_id):
+    return sorted(
+        f.path
+        for f in api.list_repo_tree(repo_id, repo_type="dataset", path_in_repo="data")
+        if f.path.endswith(".parquet")
+    )
+
+
+def _build_add_ops(batch):
+    return [CommitOperationAdd(path_in_repo=p, path_or_fileobj=local)
+            for p, local in batch]
+
 
+def _build_delete_ops(existing_files, n_new, max_files):
+    total_after = len(existing_files) + n_new
+    if max_files <= 0 or total_after <= max_files:
+        return []
+    n_delete = total_after - max_files
+    return [CommitOperationDelete(path_in_repo=p) for p in existing_files[:n_delete]]
 
-def _maybe_squash_history(api, repo_id, now):
+
+def _delete_temp_files(batch):
+    for _, local in batch:
+        try:
+            os.unlink(local)
+        except Exception:
+            pass
+
+
+def _squash_if_needed(api, repo_id):
     marker = "metadata/last_squash.txt"
-    today = now.strftime("%Y-%m-%d")
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
     try:
         try:
             local = hf_hub_download(repo_id=repo_id, filename=marker,
@@ -116,78 +125,85 @@ def _maybe_squash_history(api, repo_id, now):
                 return
         except Exception as e:
             print(f"[log] squash marker not found ({e}), proceeding with squash")
-
         api.super_squash_history(repo_id=repo_id, repo_type="dataset")
+        api.upload_file(path_or_fileobj=today.encode(), path_in_repo=marker,
+                        repo_id=repo_id, repo_type="dataset")
         print(f"[log] squashed history for {repo_id}")
-
-        api.upload_file(
-            path_or_fileobj=today.encode(), path_in_repo=marker,
-            repo_id=repo_id, repo_type="dataset",
-        )
-        print(f"[log] updated squash marker: {today}")
     except Exception as e:
         print(f"[log] squash warning: {e}")
 
 
-def _prune_old_files(api, repo_id, keep_count):
-    if keep_count <= 0:
-        return
-    try:
-        all_files = sorted(
-            f.path
-            for f in api.list_repo_tree(repo_id, repo_type="dataset", path_in_repo="data")
-            if f.path.endswith(".parquet")
-        )
-        to_delete = all_files[:-keep_count] if len(all_files) > keep_count else []
-        if to_delete:
-            ops = [CommitOperationDelete(path_in_repo=p) for p in to_delete]
-            api.create_commit(
-                repo_id=repo_id, repo_type="dataset", operations=ops,
-                commit_message=f"[log] prune {len(to_delete)} old file(s)",
-            )
-            print(f"[log] pruned {len(to_delete)} old file(s)")
-    except Exception as e:
-        print(f"[log] prune warning: {e}")
-
+class LogUploader:
+    def __init__(self, token, repo_id, max_files=5000, batch_interval=60):
+        self._token = token
+        self._repo_id = repo_id
+        self._max_files = max_files
+        self._batch_interval = batch_interval
+        self._pending = []
+        self._lock = threading.Lock()
+        if token and repo_id:
+            threading.Thread(target=self._loop, daemon=True, name="log-uploader").start()
+
+    def log_inference(self, pil_inputs, output_pil, prompt, seed, steps, guidance_scale,
+                      input_width, input_height, duration_seconds, success, error_message=""):
+        if not self._token or not self._repo_id:
+            print(f"[log] skipped — token={'set' if self._token else 'missing'}, repo={'set' if self._repo_id else 'missing'}")
+            return
+        t0 = _time.perf_counter()
+        try:
+            now = datetime.now(timezone.utc)
+            table = _build_table(pil_inputs, output_pil, prompt, seed, steps, guidance_scale,
+                                 input_width, input_height, duration_seconds, success, error_message, now)
+            local_path = _write_parquet(table)
+            path_in_repo = _make_path(now, uuid.uuid4().hex[:8])
+            self._enqueue(path_in_repo, local_path)
+            print(f"[log] queued {path_in_repo} (pending={len(self._pending)})")
+        except Exception as e:
+            import traceback as _tb
+            print(f"[log] WARNING: {e}\n{_tb.format_exc()}")
+        print(f"[log] log_inference total: {_time.perf_counter() - t0:.3f}s")
+
+    def _enqueue(self, path_in_repo, local_path):
+        with self._lock:
+            self._pending.append((path_in_repo, local_path))
+
+    def _drain(self):
+        with self._lock:
+            batch = self._pending[:]
+            self._pending.clear()
+        return batch
+
+    def _requeue(self, batch):
+        with self._lock:
+            self._pending[:0] = batch
+
+    def _loop(self):
+        while True:
+            _time.sleep(self._batch_interval)
+            self._flush()
+
+    def _flush(self):
+        batch = self._drain()
+        if not batch:
+            return
+        try:
+            self._commit_batch(batch)
+            _delete_temp_files(batch)
+        except Exception as e:
+            print(f"[log] batch upload warning: {e}")
+            self._requeue(batch)
 
-def log_inference(pil_inputs, output_pil, prompt, seed, steps, guidance_scale,
-                  input_width, input_height, duration_seconds, success, error_message=""):
-    import time as _time
-    _t0 = _time.perf_counter()
-    if not HF_TOKEN or not DATASET_REPO:
-        print(f"[log] skipped — HF_TOKEN={'set' if HF_TOKEN else 'missing'}, DATASET_REPO={'set' if DATASET_REPO else 'missing'}")
-        return
-    try:
+    def _commit_batch(self, batch):
         from huggingface_hub import HfApi
-
-        now = datetime.now(timezone.utc)
-
-        _t1 = _time.perf_counter()
-        table = _build_table(pil_inputs, output_pil, prompt, seed, steps, guidance_scale,
-                             input_width, input_height, duration_seconds, success, error_message, now)
-        print(f"[log]   build_table:       {_time.perf_counter() - _t1:.3f}s")
-
-        uid = uuid.uuid4().hex[:8]
-        path_in_repo = _make_path(now, uid)
-
-        _t2 = _time.perf_counter()
-        api = HfApi(token=HF_TOKEN)
-        api.create_repo(repo_id=DATASET_REPO, repo_type="dataset", private=True, exist_ok=True)
-        print(f"[log]   create_repo:       {_time.perf_counter() - _t2:.3f}s")
-
-        _t3 = _time.perf_counter()
-        _prune_old_files(api, DATASET_REPO, MAX_LOG_FILES)
-        print(f"[log]   prune_old_files:   {_time.perf_counter() - _t3:.3f}s")
-
-        _t4 = _time.perf_counter()
-        _upload_parquet(api, DATASET_REPO, table, path_in_repo)
-        print(f"[log]   upload_parquet:    {_time.perf_counter() - _t4:.3f}s")
-
-        _t5 = _time.perf_counter()
-        _maybe_squash_history(api, DATASET_REPO, now)
-        print(f"[log]   squash_history:    {_time.perf_counter() - _t5:.3f}s")
-    except Exception as log_err:
-        import traceback as _tb
-        print(f"[log] WARNING: {log_err}\n{_tb.format_exc()}")
-    finally:
-        print(f"[log] log_inference total: {_time.perf_counter() - _t0:.3f}s")
+        api = HfApi(token=self._token)
+        api.create_repo(repo_id=self._repo_id, repo_type="dataset", private=True, exist_ok=True)
+        existing = _list_existing_files(api, self._repo_id)
+        add_ops = _build_add_ops(batch)
+        del_ops = _build_delete_ops(existing, len(batch), self._max_files)
+        api.create_commit(
+            repo_id=self._repo_id, repo_type="dataset",
+            operations=add_ops + del_ops,
+            commit_message=f"[log] batch {len(batch)}" + (f", prune {len(del_ops)}" if del_ops else ""),
+        )
+        print(f"[log] committed {len(batch)} file(s), pruned {len(del_ops)}")
+        _squash_if_needed(api, self._repo_id)

tests/test_logging_utils.py

@@ -2,104 +2,93 @@ from datetime import datetime, timezone, timedelta
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
-from huggingface_hub import CommitOperationDelete
-from logging_utils import _prune_old_files, _file_date, _make_path, _maybe_squash_history
+from huggingface_hub import CommitOperationAdd, CommitOperationDelete
+from logging_utils import (
+    _build_add_ops, _build_delete_ops, _make_path, _squash_if_needed,
+)
 
 NOW = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
 KEEP_COUNT = 10
 
 
-def _make_api(paths):
-    api = MagicMock()
-    api.list_repo_tree.return_value = [SimpleNamespace(path=p) for p in paths]
-    return api
-
-
 def _file(days_ago):
     return _make_path(NOW - timedelta(days=days_ago), "abcd1234")
 
 
-def _deleted_paths(api):
-    ops = api.create_commit.call_args.kwargs["operations"]
-    return {op.path_in_repo for op in ops}
+# ── _build_add_ops ────────────────────────────────────────────────────────────
 
+def test_build_add_ops_returns_one_op_per_entry(tmp_path):
+    files = [tmp_path / "a.parquet", tmp_path / "b.parquet"]
+    for f in files:
+        f.write_bytes(b"")
+    batch = [("data/a.parquet", str(files[0])), ("data/b.parquet", str(files[1]))]
+    ops = _build_add_ops(batch)
+    assert len(ops) == 2
+    assert all(isinstance(op, CommitOperationAdd) for op in ops)
+    assert [op.path_in_repo for op in ops] == ["data/a.parquet", "data/b.parquet"]
 
-def test_file_date_matches_make_path():
-    """Contract: _file_date(_make_path(now, uid)) must return now's date.
 
-    _make_path is the single source of truth for file naming.
-    _file_date must correctly parse whatever _make_path produces.
-    If either changes incompatibly, this test breaks immediately.
-    """
-    import uuid
-    path = _make_path(NOW, uuid.uuid4().hex[:8])
-    assert _file_date(path) == NOW.strftime("%Y-%m-%d")
+def test_build_add_ops_empty_batch():
+    assert _build_add_ops([]) == []
 
 
+# ── _build_delete_ops ─────────────────────────────────────────────────────────
+
 def test_oldest_files_deleted_when_over_limit():
-    files = [_file(i) for i in range(12)]  # 12 files, oldest = largest days_ago
-    api = _make_api(files)
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    assert len(_deleted_paths(api)) == 2
-    # oldest two files (days_ago=11, days_ago=10) should be deleted
-    assert _deleted_paths(api) == {_file(11), _file(10)}
+    existing = sorted([_file(i) for i in range(12)])  # 12 files
+    ops = _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT)
+    deleted = {op.path_in_repo for op in ops}
+    assert len(deleted) == 2
+    assert deleted == {_file(11), _file(10)}
 
 
 def test_files_kept_when_under_limit():
-    api = _make_api([_file(i) for i in range(5)])
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    api.create_commit.assert_not_called()
+    existing = [_file(i) for i in range(5)]
+    assert _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT) == []
 
 
 def test_exactly_at_limit_nothing_deleted():
-    api = _make_api([_file(i) for i in range(KEEP_COUNT)])
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    api.create_commit.assert_not_called()
+    existing = [_file(i) for i in range(KEEP_COUNT)]
+    assert _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT) == []
 
 
 def test_one_over_limit_oldest_deleted():
-    files = [_file(i) for i in range(KEEP_COUNT + 1)]
-    api = _make_api(files)
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    assert _deleted_paths(api) == {_file(KEEP_COUNT)}
+    existing = sorted([_file(i) for i in range(KEEP_COUNT + 1)])
+    ops = _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT)
+    assert {op.path_in_repo for op in ops} == {_file(KEEP_COUNT)}
 
 
-def test_empty_directory_does_nothing():
-    api = _make_api([])
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    api.create_commit.assert_not_called()
+def test_empty_existing_does_nothing():
+    assert _build_delete_ops([], n_new=0, max_files=KEEP_COUNT) == []
 
 
-def test_keep_count_zero_skips_pruning():
-    api = _make_api([_file(0)])
-    _prune_old_files(api, "org/repo", keep_count=0)
-    api.list_repo_tree.assert_not_called()
-    api.create_commit.assert_not_called()
+def test_max_files_zero_skips_pruning():
+    existing = [_file(0)]
+    assert _build_delete_ops(existing, n_new=0, max_files=0) == []
 
 
-def test_non_parquet_files_are_ignored():
-    # 10 parquet files + 1 non-parquet: total parquet == KEEP_COUNT, nothing deleted
-    api = _make_api([_file(i) for i in range(KEEP_COUNT)] + ["data/2020-01-01-README.md"])
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    api.create_commit.assert_not_called()
+def test_n_new_counted_toward_total():
+    # 8 existing + 4 new = 12 total, need to delete 2
+    existing = sorted([_file(i) for i in range(8)])
+    ops = _build_delete_ops(existing, n_new=4, max_files=KEEP_COUNT)
+    assert len(ops) == 2
 
 
-def test_api_error_does_not_raise():
-    api = MagicMock()
-    api.list_repo_tree.side_effect = RuntimeError("network error")
-    _prune_old_files(api, "org/repo", KEEP_COUNT)  # must not raise
+def test_oldest_files_deleted_regardless_of_input_order():
+    files = [_file(3), _file(11), _file(0), _file(10), _file(1),
+             _file(2), _file(4), _file(5), _file(6), _file(7), _file(8)]
+    existing = sorted(files)  # caller sorts before passing
+    ops = _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT)
+    assert {op.path_in_repo for op in ops} == {_file(11)}
 
 
-def test_oldest_files_deleted_correct_order():
-    # Provide files in non-sorted order; pruning should still delete the oldest
-    files = [_file(3), _file(11), _file(0), _file(10), _file(1),
-             _file(2), _file(4), _file(5), _file(6), _file(7), _file(8)]  # 11 files
-    api = _make_api(files)
-    _prune_old_files(api, "org/repo", KEEP_COUNT)
-    assert _deleted_paths(api) == {_file(11)}
+def test_all_ops_are_delete_type():
+    existing = sorted([_file(i) for i in range(12)])
+    ops = _build_delete_ops(existing, n_new=0, max_files=KEEP_COUNT)
+    assert all(isinstance(op, CommitOperationDelete) for op in ops)
 
 
-# ── _maybe_squash_history ────────────────────────────────────────────────────
+# ── _squash_if_needed ─────────────────────────────────────────────────────────
 
 def _squash_api():
     api = MagicMock()
@@ -113,26 +102,28 @@ def test_squash_runs_when_no_marker(monkeypatch):
         MagicMock(side_effect=FileNotFoundError("no marker")),
     )
     api = _squash_api()
-    _maybe_squash_history(api, "org/repo", NOW)
+    _squash_if_needed(api, "org/repo")
     api.super_squash_history.assert_called_once()
     api.upload_file.assert_called_once()
 
 
 def test_squash_skipped_when_marker_is_today(monkeypatch, tmp_path):
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
     marker = tmp_path / "last_squash.txt"
-    marker.write_text(NOW.strftime("%Y-%m-%d"))
+    marker.write_text(today)
     monkeypatch.setattr("logging_utils.hf_hub_download", MagicMock(return_value=str(marker)))
     api = _squash_api()
-    _maybe_squash_history(api, "org/repo", NOW)
+    _squash_if_needed(api, "org/repo")
     api.super_squash_history.assert_not_called()
 
 
 def test_squash_runs_when_marker_is_yesterday(monkeypatch, tmp_path):
+    yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d")
     marker = tmp_path / "last_squash.txt"
-    marker.write_text((NOW - timedelta(days=1)).strftime("%Y-%m-%d"))
+    marker.write_text(yesterday)
     monkeypatch.setattr("logging_utils.hf_hub_download", MagicMock(return_value=str(marker)))
     api = _squash_api()
-    _maybe_squash_history(api, "org/repo", NOW)
+    _squash_if_needed(api, "org/repo")
     api.super_squash_history.assert_called_once()
 
 
@@ -143,4 +134,4 @@ def test_squash_error_does_not_raise(monkeypatch):
     )
     api = _squash_api()
     api.super_squash_history.side_effect = RuntimeError("squash failed")
-    _maybe_squash_history(api, "org/repo", NOW)  # must not raise
+    _squash_if_needed(api, "org/repo")  # must not raise