fix(usage-keeper): 收紧HF快照轮转上传

.env.example

@@ -50,6 +50,19 @@ REQUEST_TIMEOUT=30s
 # Application work directory. SQLite database, logs, and backups are stored here by default. In Docker the runtime working directory is /, so ./data maps to /data; for binaries loaded with --env or package-local .env, ./data maps to a data directory next to the env file. Required: no. Default: ./data.
 WORK_DIR=./data
 
+# Hugging Face Space SQLite 快照上传间隔。Space 默认每 60 秒上传一次；配合 rotate 只保留最近 KEEPER_HF_ROTATE_KEEP 个历史文件。
+# Hugging Face Space SQLite snapshot upload interval. The Space default uploads every 60 seconds; rotation keeps only the latest KEEPER_HF_ROTATE_KEEP history files.
+KEEPER_HF_UPLOAD_INTERVAL=60
+
+# Hugging Face SQLite 快照轮转粒度和保留数量。默认 60 秒一个 history 文件、保留 48 个；如需更长恢复窗口可调大 KEEP。
+# Hugging Face SQLite snapshot rotation bucket and retention count. Defaults to one history file per 60 seconds and keeps 48 files; increase KEEP for a longer restore window.
+KEEPER_HF_ROTATE_INTERVAL=60
+KEEPER_HF_ROTATE_KEEP=48
+
+# 是否额外维护 KEEPER_HF_PATH 指向的最新 app.db。默认关闭，避免每分钟覆盖同一路径导致远端历史/LFS 膨胀；恢复会优先读 rotate manifest。
+# Whether to also maintain the latest app.db at KEEPER_HF_PATH. Disabled by default to avoid growing remote history/LFS by overwriting the same path every minute; restore reads the rotate manifest first.
+KEEPER_HF_WRITE_LATEST=0
+
 # 应用日志级别。必填：否。默认值：info。
 # Application log level. Required: no. Default: info.
 LOG_LEVEL=info

entrypoint.space.sh

@@ -13,9 +13,10 @@ export AUTH_ENABLED="${AUTH_ENABLED:-true}"
 export KEEPER_HF_REPO_ID="${KEEPER_HF_REPO_ID:-pjpjq/daili-usage-state}"
 export KEEPER_HF_REPO_TYPE="${KEEPER_HF_REPO_TYPE:-dataset}"
 export KEEPER_HF_PATH="${KEEPER_HF_PATH:-usage-keeper/app.db}"
-export KEEPER_HF_ROTATE_INTERVAL="${KEEPER_HF_ROTATE_INTERVAL:-3600}"
+export KEEPER_HF_ROTATE_INTERVAL="${KEEPER_HF_ROTATE_INTERVAL:-60}"
 export KEEPER_HF_ROTATE_KEEP="${KEEPER_HF_ROTATE_KEEP:-48}"
-export KEEPER_HF_UPLOAD_INTERVAL="${KEEPER_HF_UPLOAD_INTERVAL:-300}"
+export KEEPER_HF_WRITE_LATEST="${KEEPER_HF_WRITE_LATEST:-0}"
+export KEEPER_HF_UPLOAD_INTERVAL="${KEEPER_HF_UPLOAD_INTERVAL:-60}"
 
 missing=""
 [ -n "${CPA_MANAGEMENT_KEY:-}" ] || missing="$missing CPA_MANAGEMENT_KEY"

hf_state_snapshot.py

@@ -18,7 +18,7 @@ from huggingface_hub.errors import EntryNotFoundError, RepositoryNotFoundError
 
 
 ROTATION_MANIFEST_TYPE = "daili_usage_keeper_sqlite_rotation_v1"
-DEFAULT_ROTATE_INTERVAL_SECONDS = 3600
+DEFAULT_ROTATE_INTERVAL_SECONDS = 60
 DEFAULT_ROTATE_KEEP = 48
 
 
@@ -43,6 +43,13 @@ def env_int(name: str, default: int) -> int:
         return default
 
 
+def env_bool(name: str, default: bool = False) -> bool:
+    raw = env(name)
+    if not raw:
+        return default
+    return raw.lower() in {"1", "true", "yes", "on"}
+
+
 def isoformat_utc(value: datetime) -> str:
     return value.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
 
@@ -126,10 +133,66 @@ def latest_snapshot_from_manifest(token: str, repo_id: str, path_in_repo: str) -
     return download_repo_file(token, repo_id, latest_path)
 
 
+def is_history_path(path_in_repo: str, candidate: str) -> bool:
+    prefix = history_dir_for(path_in_repo).rstrip("/") + "/"
+    return candidate.startswith(prefix) and candidate != prefix
+
+
+def list_repo_files_safe(api: HfApi, repo_id: str) -> set[str] | None:
+    try:
+        return set(api.list_repo_files(repo_id=repo_id, repo_type=repo_type()))
+    except Exception:
+        return None
+
+
+def existing_history_paths(path_in_repo: str, repo_files: set[str] | None) -> list[str]:
+    if repo_files is None:
+        return []
+    return sorted(
+        (item for item in repo_files if is_history_path(path_in_repo, item)),
+        reverse=True,
+    )
+
+
+def retained_history_paths(
+    path_in_repo: str,
+    current_path: str,
+    previous_paths: list[str],
+    existing_paths: list[str],
+    rotate_keep: int,
+) -> list[str]:
+    retained: list[str] = []
+    seen: set[str] = set()
+    for item in [current_path, *existing_paths, *previous_paths]:
+        if not item or item in seen or not is_history_path(path_in_repo, item):
+            continue
+        seen.add(item)
+        retained.append(item)
+        if len(retained) >= rotate_keep:
+            break
+    return retained
+
+
+def deleted_history_paths(
+    path_in_repo: str,
+    previous_paths: list[str],
+    existing_paths: list[str],
+    retained_paths: list[str],
+    repo_listing_available: bool,
+) -> list[str]:
+    retained = set(retained_paths)
+    candidates = existing_paths if repo_listing_available else previous_paths
+    return [
+        item
+        for item in candidates
+        if is_history_path(path_in_repo, item) and item not in retained
+    ]
+
+
 def download_latest(token: str, repo_id: str, path_in_repo: str, local_path: Path) -> int:
-    downloaded = download_repo_file(token, repo_id, path_in_repo)
+    downloaded = latest_snapshot_from_manifest(token, repo_id, path_in_repo)
     if downloaded is None:
-        downloaded = latest_snapshot_from_manifest(token, repo_id, path_in_repo)
+        downloaded = download_repo_file(token, repo_id, path_in_repo)
     if downloaded is None:
         return 0
     local_path.parent.mkdir(parents=True, exist_ok=True)
@@ -174,46 +237,93 @@ def upload_snapshot(token: str, repo_id: str, path_in_repo: str, local_path: Pat
 
     rotate_interval = env_int("KEEPER_HF_ROTATE_INTERVAL", DEFAULT_ROTATE_INTERVAL_SECONDS)
     rotate_keep = env_int("KEEPER_HF_ROTATE_KEEP", DEFAULT_ROTATE_KEEP)
+    rotate_enabled = rotate_interval > 0 and rotate_keep > 0
+    write_latest = env_bool("KEEPER_HF_WRITE_LATEST", False)
+    if not rotate_enabled:
+        write_latest = True
+
     api = HfApi(token=token)
     temp_copy = backup_sqlite(local_path)
     try:
         checksum = sha256_file(temp_copy)
         metadata_path = manifest_path_for(path_in_repo) + ".state.json"
         current_state = load_json(download_repo_file(token, repo_id, metadata_path) or Path("/nonexistent"))
-        if current_state and current_state.get("sha256") == checksum:
-            return 0
+        same_checksum = bool(current_state and current_state.get("sha256") == checksum)
+        repo_files = list_repo_files_safe(api, repo_id) if rotate_enabled else None
+
+        operations = []
+        if write_latest:
+            if not same_checksum:
+                operations.append(CommitOperationAdd(path_in_repo=path_in_repo, path_or_fileobj=str(temp_copy)))
+        elif repo_files is not None and path_in_repo in repo_files:
+            operations.append(CommitOperationDelete(path_in_repo=path_in_repo))
+
+        if not same_checksum:
+            operations.append(
+                CommitOperationAdd(
+                    path_in_repo=metadata_path,
+                    path_or_fileobj=json.dumps(
+                        {
+                            "updated_at": isoformat_utc(datetime.now(UTC)),
+                            "sha256": checksum,
+                            "size_bytes": temp_copy.stat().st_size,
+                        },
+                        ensure_ascii=False,
+                        indent=2,
+                        sort_keys=True,
+                    ).encode("utf-8"),
+                )
+            )
 
-        operations = [
-            CommitOperationAdd(path_in_repo=path_in_repo, path_or_fileobj=str(temp_copy)),
-            CommitOperationAdd(
-                path_in_repo=metadata_path,
-                path_or_fileobj=json.dumps(
-                    {
-                        "updated_at": isoformat_utc(datetime.now(UTC)),
-                        "sha256": checksum,
-                        "size_bytes": temp_copy.stat().st_size,
-                    },
-                    ensure_ascii=False,
-                    indent=2,
-                    sort_keys=True,
-                ).encode("utf-8"),
-            ),
-        ]
-
-        if rotate_interval > 0 and rotate_keep > 0:
+        if rotate_enabled:
             bucket_time = floor_time(snapshot_time_for(temp_copy), rotate_interval)
             rotated_path = history_path_for(path_in_repo, bucket_time)
             manifest_path = manifest_path_for(path_in_repo)
             manifest = load_manifest(token, repo_id, manifest_path) or {}
+            latest_path = manifest.get("latest_path")
+            latest_path_known = (
+                isinstance(latest_path, str)
+                and bool(latest_path)
+                and is_history_path(path_in_repo, latest_path)
+            )
+            latest_path_missing = bool(
+                repo_files is not None
+                and latest_path_known
+                and latest_path not in repo_files
+            )
+            needs_rotated_upload = (not same_checksum) or (not latest_path_known) or latest_path_missing
+            current_history_path = rotated_path if needs_rotated_upload else latest_path
             previous_paths = [
                 item
                 for item in manifest.get("retained_paths", [])
                 if isinstance(item, str) and item
             ]
-            retained_paths = [rotated_path, *[item for item in previous_paths if item != rotated_path]][:rotate_keep]
-            delete_paths = [item for item in previous_paths if item not in retained_paths]
-            if manifest.get("latest_path") != rotated_path:
+            repo_listing_available = repo_files is not None
+            existing_paths = existing_history_paths(path_in_repo, repo_files)
+            retained_paths = retained_history_paths(
+                path_in_repo,
+                current_history_path,
+                previous_paths,
+                existing_paths,
+                rotate_keep,
+            )
+            delete_paths = deleted_history_paths(
+                path_in_repo,
+                previous_paths,
+                existing_paths,
+                retained_paths,
+                repo_listing_available,
+            )
+            needs_manifest_update = (
+                needs_rotated_upload
+                or manifest.get("retained_paths") != retained_paths
+                or manifest.get("rotation_interval_seconds") != rotate_interval
+                or manifest.get("retention_count") != rotate_keep
+                or bool(delete_paths)
+            )
+            if needs_rotated_upload:
                 operations.append(CommitOperationAdd(path_in_repo=rotated_path, path_or_fileobj=str(temp_copy)))
+            if needs_manifest_update:
                 operations.append(
                     CommitOperationAdd(
                         path_in_repo=manifest_path,
@@ -221,7 +331,7 @@ def upload_snapshot(token: str, repo_id: str, path_in_repo: str, local_path: Pat
                             {
                                 "type": ROTATION_MANIFEST_TYPE,
                                 "version": 1,
-                                "latest_path": rotated_path,
+                                "latest_path": current_history_path,
                                 "updated_at": isoformat_utc(datetime.now(UTC)),
                                 "source_path": path_in_repo,
                                 "repo_type": repo_type(),
@@ -237,7 +347,10 @@ def upload_snapshot(token: str, repo_id: str, path_in_repo: str, local_path: Pat
                         ).encode("utf-8"),
                     )
                 )
-                operations.extend(CommitOperationDelete(path_in_repo=item) for item in delete_paths)
+            operations.extend(CommitOperationDelete(path_in_repo=item) for item in delete_paths)
+
+        if not operations:
+            return 0
 
         api.create_commit(
             repo_id=repo_id,

hf_state_snapshot_test.py

@@ -0,0 +1,135 @@
+import hashlib
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import hf_state_snapshot as snapshot
+
+
+class RotationRetentionTest(unittest.TestCase):
+    def test_default_rotation_bucket_matches_upload_interval(self) -> None:
+        self.assertEqual(snapshot.DEFAULT_ROTATE_INTERVAL_SECONDS, 60)
+
+    def test_retention_keeps_current_and_newest_existing_paths(self) -> None:
+        path = "usage-keeper/app.db"
+        retained = snapshot.retained_history_paths(
+            path,
+            "usage-keeper/app.history/20260522T030000Z.db",
+            [
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+            ],
+            [
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+                "usage-keeper/app.history/20260522T000000Z.db",
+            ],
+            3,
+        )
+
+        self.assertEqual(
+            retained,
+            [
+                "usage-keeper/app.history/20260522T030000Z.db",
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+            ],
+        )
+
+    def test_cleanup_removes_orphaned_history_files_from_repo_listing(self) -> None:
+        path = "usage-keeper/app.db"
+        delete_paths = snapshot.deleted_history_paths(
+            path,
+            previous_paths=[
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+            ],
+            existing_paths=[
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+                "usage-keeper/app.history/20260522T000000Z.db",
+                "usage-keeper/other.history/20260522T000000Z.db",
+            ],
+            retained_paths=[
+                "usage-keeper/app.history/20260522T020000Z.db",
+                "usage-keeper/app.history/20260522T010000Z.db",
+            ],
+            repo_listing_available=True,
+        )
+
+        self.assertEqual(delete_paths, ["usage-keeper/app.history/20260522T000000Z.db"])
+
+    def test_upload_cleanup_does_not_overwrite_latest_path_when_rotating(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            temp_copy = Path(tmpdir) / "app.db"
+            temp_copy.write_bytes(b"sqlite snapshot")
+            checksum = hashlib.sha256(temp_copy.read_bytes()).hexdigest()
+            state_file = Path(tmpdir) / "state.json"
+            state_file.write_text(json.dumps({"sha256": checksum}), encoding="utf-8")
+            local_db = Path(tmpdir) / "local.db"
+            local_db.write_bytes(b"source")
+
+            class FakeApi:
+                def __init__(self) -> None:
+                    self.operations = []
+
+                def create_commit(self, **kwargs) -> None:
+                    self.operations = kwargs["operations"]
+
+            fake_api = FakeApi()
+
+            def fake_download(_token: str, _repo_id: str, path_in_repo: str):
+                if path_in_repo.endswith(".state.json"):
+                    return state_file
+                return None
+
+            with (
+                patch.object(snapshot, "HfApi", return_value=fake_api),
+                patch.object(snapshot, "backup_sqlite", return_value=temp_copy),
+                patch.object(snapshot, "download_repo_file", side_effect=fake_download),
+                patch.object(
+                    snapshot,
+                    "list_repo_files_safe",
+                    return_value={
+                        "usage-keeper/app.db",
+                        "usage-keeper/app.history/20260522T030000Z.db",
+                        "usage-keeper/app.history/20260522T020000Z.db",
+                    },
+                ),
+                patch.object(
+                    snapshot,
+                    "load_manifest",
+                    return_value={
+                        "type": snapshot.ROTATION_MANIFEST_TYPE,
+                        "latest_path": "usage-keeper/app.history/20260522T030000Z.db",
+                        "rotation_interval_seconds": 3600,
+                        "retention_count": 48,
+                        "retained_paths": [
+                            "usage-keeper/app.history/20260522T030000Z.db",
+                            "usage-keeper/app.history/20260522T020000Z.db",
+                        ],
+                    },
+                ),
+            ):
+                self.assertEqual(snapshot.upload_snapshot("token", "repo", "usage-keeper/app.db", local_db), 0)
+
+            added_paths = {
+                operation.path_in_repo
+                for operation in fake_api.operations
+                if isinstance(operation, snapshot.CommitOperationAdd)
+            }
+            deleted_paths = {
+                operation.path_in_repo
+                for operation in fake_api.operations
+                if isinstance(operation, snapshot.CommitOperationDelete)
+            }
+
+            self.assertNotIn("usage-keeper/app.db", added_paths)
+            self.assertIn("usage-keeper/app.db", deleted_paths)
+            self.assertIn("usage-keeper/app.manifest.json", added_paths)
+
+
+if __name__ == "__main__":
+    unittest.main()