Z User
fix(entrypoint): start the celery worker only when CELERY_BULK_ENABLED=true — a misconfigured rediss:// broker caused a pointless 5s crash-restart loop while bulk actually runs sync
4c14cbd | #!/usr/bin/env python3 | |
| """ | |
| Container entrypoint for BRIDGE Modul-X on Hugging Face Spaces. | |
| Responsibilities | |
| ---------------- | |
| 1. Ensure Redis for cache + Celery broker/results, in priority order: | |
| a. Space secrets win ONLY if they answer PING: | |
| REDIS_URL / CELERY_BROKER_URL / CELERY_RESULT_BACKEND. | |
| (A stale secret — e.g. an expired Upstash DB whose domain no longer | |
| resolves — must never shadow the sidecar; it used to crash-loop the | |
| whole app.) | |
| b. Otherwise run a local redis-server sidecar on 127.0.0.1:6379 | |
| (installed in the image; volatile memory, no persistence). This is the | |
| default production path — no external dependency, no expiry. | |
| c. If the sidecar binary is unavailable, fall back to Upstash free-tier | |
| provisioning (deterministic idempotency keys; DBs expire in ~3 days | |
| unless claimed at the console URL printed here). | |
| 2. Supervise child processes: | |
| - redis-server (sidecar mode only) | |
| - Celery worker (bulk validation queue) | |
| - uvicorn (FastAPI web app on $PORT / 7860) | |
| Sidecar/worker are restarted if they die; if the web process dies the | |
| container exits and the platform restarts it. | |
| """ | |
| import os | |
| import signal | |
| import subprocess | |
| import sys | |
| import time | |
| import uuid | |
| # Fixed namespace so the per-store Idempotency-Keys are deterministic. | |
| _NS = uuid.UUID("7c9e6679-7425-40de-944b-e07fc1f90ae7") | |
| SIDECAR_PORT = 6379 | |
| def _stable_db_id(store: str) -> str: | |
| """Deterministic UUIDv4-FORM id (Upstash rejects non-v4 idempotency keys): | |
| derive bytes from uuid5, then patch version/variant bits to v4/RFC-4122.""" | |
| b = bytearray(uuid.uuid5(_NS, f"bridge-modul-x/{store}").bytes) | |
| b[6] = (b[6] & 0x0F) | 0x40 | |
| b[8] = (b[8] & 0x3F) | 0x80 | |
| return str(uuid.UUID(bytes=bytes(b))) | |
| # --------------------------------------------------------------------------- # | |
| # Upstash free-tier fallback (only used when no local redis-server exists) | |
| # --------------------------------------------------------------------------- # | |
| def _provision_upstash(store: str): | |
| from app.utils.upstash_free import create_database | |
| for attempt, db_id in enumerate([_stable_db_id(store), ""]): | |
| try: | |
| db = create_database(db_id=db_id) | |
| print( | |
| f"[ENTRYPOINT] upstash {store}: db {db.db_id} ok " | |
| f"(expires {db.expires}) - claim to keep it: {db.console_url}", | |
| flush=True, | |
| ) | |
| return db.connection_string(0) | |
| except Exception as exc: | |
| print(f"[ENTRYPOINT] upstash {store}: failed (try {attempt}): {exc}", flush=True) | |
| return None | |
| # --------------------------------------------------------------------------- # | |
| # Redis resolution | |
| # --------------------------------------------------------------------------- # | |
| def _secret_url(env_name: str): | |
| """An externally-provided (Space secret) URL wins over any fallback.""" | |
| url = os.environ.get(env_name, "").strip() | |
| if url and "localhost" not in url and "127.0.0.1" not in url: | |
| return url | |
| return None | |
| def _wait_for_redis(url: str, seconds: int = 30) -> bool: | |
| import redis as _redis | |
| deadline = time.monotonic() + seconds | |
| while time.monotonic() < deadline: | |
| try: | |
| client = _redis.Redis.from_url(url, socket_connect_timeout=2, socket_timeout=2) | |
| if client.ping(): | |
| client.close() | |
| return True | |
| except Exception: | |
| time.sleep(0.5) | |
| return False | |
| def _resolve_redis(children: dict): | |
| """Return (cache, broker, result) URLs and populate `children` with a | |
| redis-server sidecar process when one is started.""" | |
| cache = _secret_url("REDIS_URL") | |
| broker = _secret_url("CELERY_BROKER_URL") | |
| result = _secret_url("CELERY_RESULT_BACKEND") | |
| if cache or broker or result: | |
| # Trust-but-verify: every provided secret URL must answer PING, | |
| # otherwise ignore all of them and use the sidecar instead. | |
| dead = [ | |
| u for u in (cache, broker, result) if u and not _wait_for_redis(u, seconds=6) | |
| ] | |
| if not dead: | |
| print("[ENTRYPOINT] using Redis from Space secrets (PING ok)", flush=True) | |
| cache = cache or broker or result | |
| broker = broker or cache | |
| result = result or broker | |
| return cache, broker, result | |
| print( | |
| "[ENTRYPOINT] WARNING: Space secret Redis unreachable " | |
| f"({len(dead)} of the URLs failed PING) - ignoring secrets", | |
| flush=True, | |
| ) | |
| cache = broker = result = None | |
| # Local sidecar: same image, same container, always reachable. | |
| try: | |
| proc = subprocess.Popen( | |
| [ | |
| "redis-server", | |
| "--port", str(SIDECAR_PORT), | |
| "--bind", "127.0.0.1", | |
| "--save", "", | |
| "--appendonly", "no", | |
| "--maxmemory", "120mb", | |
| "--maxmemory-policy", "allkeys-lru", | |
| ], | |
| env=os.environ, | |
| ) | |
| children["redis"] = proc | |
| print(f"[ENTRYPOINT] redis-server sidecar started (pid {proc.pid})", flush=True) | |
| except FileNotFoundError: | |
| print("[ENTRYPOINT] redis-server binary missing - using Upstash fallback", flush=True) | |
| except Exception as exc: | |
| print(f"[ENTRYPOINT] sidecar spawn failed ({exc}) - using Upstash fallback", flush=True) | |
| if "redis" in children: | |
| base = f"redis://127.0.0.1:{SIDECAR_PORT}" | |
| if _wait_for_redis(f"{base}/0"): | |
| print("[ENTRYPOINT] sidecar PONG ok", flush=True) | |
| # Separate logical DBs for cache / broker / results. | |
| return f"{base}/0", f"{base}/1", f"{base}/2" | |
| print("[ENTRYPOINT] sidecar never answered PING - falling back", flush=True) | |
| try: | |
| children.pop("redis").terminate() | |
| except Exception: | |
| pass | |
| cache = _provision_upstash("cache") | |
| broker = _provision_upstash("celery-broker") or cache | |
| result = _provision_upstash("celery-result") or broker | |
| return cache, broker, result | |
| def main(): | |
| children = {} | |
| state = {"stopping": False} | |
| cache_url, broker_url, result_url = _resolve_redis(children) | |
| if cache_url: | |
| os.environ["REDIS_URL"] = cache_url | |
| if broker_url: | |
| os.environ["CELERY_BROKER_URL"] = broker_url | |
| if result_url: | |
| os.environ["CELERY_RESULT_BACKEND"] = result_url | |
| if not broker_url: | |
| print("[ENTRYPOINT] WARNING: no Redis reachable - bulk jobs use the synchronous fallback", flush=True) | |
| # Hugging Face Spaces always proxies to container port 7860; a PORT | |
| # env/secret with another value (e.g. 8000) must not steer uvicorn, | |
| # or the platform never sees the app start. | |
| port = "7860" | |
| def spawn(kind): | |
| if kind == "worker": | |
| cmd = [ | |
| sys.executable, "-m", "celery", | |
| "-A", "app.tasks.celery_tasks:celery_app", "worker", | |
| "--loglevel=info", "--concurrency=4", | |
| ] | |
| elif kind == "redis": | |
| cmd = [ | |
| "redis-server", "--port", str(SIDECAR_PORT), "--bind", "127.0.0.1", | |
| "--save", "", "--appendonly", "no", | |
| "--maxmemory", "120mb", "--maxmemory-policy", "allkeys-lru", | |
| ] | |
| else: | |
| cmd = [ | |
| sys.executable, "-m", "uvicorn", "app.main:app", | |
| "--host", "0.0.0.0", "--port", port, | |
| ] | |
| proc = subprocess.Popen(cmd, env=os.environ) | |
| children[kind] = proc | |
| print(f"[ENTRYPOINT] started {kind} (pid {proc.pid})", flush=True) | |
| return proc | |
| def on_signal(signum, frame): | |
| state["stopping"] = True | |
| for proc in list(children.values()): | |
| try: | |
| proc.terminate() | |
| except Exception: | |
| pass | |
| signal.signal(signal.SIGTERM, on_signal) | |
| signal.signal(signal.SIGINT, on_signal) | |
| # Celery worker (bulk validation queue) starts ONLY when bulk is routed | |
| # through Celery on this host. With CELERY_BULK_ENABLED=false (the Space | |
| # default) bulk runs synchronously in-process, and spawning a worker | |
| # against a misconfigured broker (e.g. a rediss:// URL without | |
| # ssl_cert_reqs) produced a pointless 5s crash-restart loop. | |
| run_celery_worker = ( | |
| os.environ.get("CELERY_BULK_ENABLED", "").strip().lower() in ("1", "true", "yes") | |
| ) | |
| if broker_url and run_celery_worker: | |
| spawn("worker") | |
| elif broker_url: | |
| print( | |
| "[ENTRYPOINT] CELERY_BULK_ENABLED!=true — bulk runs sync; " | |
| "celery worker NOT started", | |
| flush=True, | |
| ) | |
| web = spawn("web") | |
| while not state["stopping"]: | |
| if web.poll() is not None: | |
| print(f"[ENTRYPOINT] web exited ({web.returncode}) - shutting down", flush=True) | |
| break | |
| if "redis" in children and children["redis"].poll() is not None: | |
| print("[ENTRYPOINT] redis sidecar exited - restarting in 2s", flush=True) | |
| time.sleep(2) | |
| if not state["stopping"]: | |
| spawn("redis") | |
| if run_celery_worker and broker_url and children.get("worker") and children["worker"].poll() is not None: | |
| print(f"[ENTRYPOINT] worker exited ({children['worker'].returncode}) - restarting in 5s", flush=True) | |
| time.sleep(5) | |
| if not state["stopping"]: | |
| spawn("worker") | |
| time.sleep(2) | |
| for proc in list(children.values()): | |
| try: | |
| proc.terminate() | |
| proc.wait(timeout=10) | |
| except Exception: | |
| try: | |
| proc.kill() | |
| except Exception: | |
| pass | |
| sys.exit(web.returncode or 0) | |
| if __name__ == "__main__": | |
| main() | |