#!/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()