| """Persist the whole Space to a Hugging Face **Storage Bucket**. |
| |
| The bucket is the single source of truth: |
| |
| <bucket>/workspace/... -> /home/node/workspace |
| <bucket>/opencode/... -> /home/node/.local/share/opencode |
| <bucket>/global/... -> /home/node/.config/opencode |
| |
| Flow |
| ---- |
| * boot : bucket -> container (``python3 sync.py --pull``, run before the web server) |
| * every N s : container -> bucket (incremental, deletions propagated) |
| * SIGTERM : one last push so a restart/rebuild keeps everything |
| |
| Env |
| --- |
| BUCKET_REPO_ID namespace/bucket-name (required) |
| HF_TOKEN token with write access |
| SYNC_INTERVAL_SECONDS default 60 |
| CREATE_BUCKET "0" to disable auto-creation (default: created private) |
| WORKSPACE_DIR default /home/node/workspace |
| OPENCODE_DATA_DIR default /home/node/.local/share/opencode |
| OPENCODE_CONFIG_DIR default /home/node/.config/opencode |
| """ |
|
|
| import argparse |
| import os |
| import signal |
| import sys |
| import threading |
| import time |
| from pathlib import Path |
|
|
| from huggingface_hub import HfApi |
|
|
| HOME = Path(os.getenv("HOME", "/home/node")) |
| WORKSPACE = Path(os.getenv("WORKSPACE_DIR", HOME / "workspace")) |
| OPENCODE_DATA = Path(os.getenv("OPENCODE_DATA_DIR", HOME / ".local/share/opencode")) |
| GLOBAL_CONFIG = Path(os.getenv("OPENCODE_CONFIG_DIR", HOME / ".config/opencode")) |
|
|
| BUCKET_ID = os.getenv("BUCKET_REPO_ID", "").strip() |
| TOKEN = os.getenv("HF_TOKEN") or None |
| INTERVAL = int(os.getenv("SYNC_INTERVAL_SECONDS", "60")) |
| CREATE_BUCKET = os.getenv("CREATE_BUCKET", "1") not in {"0", "false", "False", ""} |
|
|
| COMMON_IGNORE = ["**/.git/**", ".git/**", "**/__pycache__/**", "**/*.pyc"] |
|
|
| TREES = [ |
| ("workspace", WORKSPACE, COMMON_IGNORE + ["**/node_modules/**", "node_modules/**"]), |
| ("opencode", OPENCODE_DATA, ["**/log/**", "log/**", "**/*.log", "**/__pycache__/**"]), |
| ("global", GLOBAL_CONFIG, COMMON_IGNORE + ["**/node_modules/**", "node_modules/**"]), |
| ] |
|
|
| api = HfApi(token=TOKEN) |
| _lock = threading.Lock() |
| _stop = threading.Event() |
|
|
|
|
| def log(message): |
| print(f"[sync {time.strftime('%H:%M:%S')}] {message}", flush=True) |
|
|
|
|
| def bucket_url(prefix): |
| return f"hf://buckets/{BUCKET_ID}/{prefix}" |
|
|
|
|
| def ensure_bucket(): |
| if not CREATE_BUCKET: |
| return |
| try: |
| api.create_bucket(BUCKET_ID, private=True, exist_ok=True, token=TOKEN) |
| log(f"bucket ready: {BUCKET_ID} (private)") |
| except Exception as exc: |
| log(f"could not create bucket (assuming it exists): {exc}") |
|
|
|
|
| def pull(): |
| if not BUCKET_ID: |
| log("BUCKET_REPO_ID not set - skipping pull") |
| return |
| for prefix, local, _ignore in TREES: |
| local.mkdir(parents=True, exist_ok=True) |
| try: |
| api.sync_bucket( |
| source=bucket_url(prefix), |
| dest=str(local), |
| token=TOKEN, |
| quiet=True, |
| ) |
| log(f"pulled {prefix} -> {local}") |
| except Exception as exc: |
| log(f"pull {prefix} failed: {exc}") |
|
|
|
|
| def push(): |
| if not BUCKET_ID: |
| return True |
| ok = True |
| for prefix, local, ignore in TREES: |
| if not local.exists(): |
| continue |
| try: |
| api.sync_bucket( |
| source=str(local), |
| dest=bucket_url(prefix), |
| delete=True, |
| exclude=ignore, |
| token=TOKEN, |
| quiet=True, |
| ) |
| log(f"pushed {prefix}") |
| except Exception as exc: |
| log(f"push {prefix} failed: {exc}") |
| ok = False |
| return ok |
|
|
|
|
| def full_sync(): |
| if not _lock.acquire(blocking=False): |
| log("previous sync still running - skipping this tick") |
| return True |
| try: |
| return push() |
| finally: |
| _lock.release() |
|
|
|
|
| def handle_signal(signum, _frame): |
| log(f"signal {signum} - flushing to bucket before exit") |
| _stop.set() |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--pull", action="store_true", help="one-shot bucket -> disk") |
| parser.add_argument("--push", action="store_true", help="one-shot disk -> bucket") |
| args = parser.parse_args() |
|
|
| if not BUCKET_ID: |
| log("BUCKET_REPO_ID is empty - sync daemon disabled") |
| return 0 |
|
|
| ensure_bucket() |
|
|
| if args.pull: |
| pull() |
| return 0 |
| if args.push: |
| return 0 if push() else 1 |
|
|
| signal.signal(signal.SIGTERM, handle_signal) |
| signal.signal(signal.SIGINT, handle_signal) |
|
|
| log(f"daemon started: interval {INTERVAL}s, bucket {BUCKET_ID}") |
| failures = 0 |
| while not _stop.is_set(): |
| wait = INTERVAL if failures == 0 else min(INTERVAL * 2 ** failures, 600) |
| if _stop.wait(wait): |
| break |
| failures = 0 if full_sync() else failures + 1 |
| if failures: |
| log(f"{failures} consecutive failure(s)") |
|
|
| full_sync() |
| log("flushed - bye") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
