ComfyUI API 비동기 큐·재시도 파이프라인 구축

# app.py
import json
import uuid
from fastapi import FastAPI, Header, HTTPException
import redis

app = FastAPI()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# 실제로는 DB에 저장해야 함
JOB_KEY_PREFIX = "job:"  # job:{job_id}
IDEMPOTENCY_PREFIX = "idem:"  # idem:{key} -> job_id
QUEUE_NAME = "comfyui:jobs"

@app.post("/v1/generate")
def generate(payload: dict, x_idempotency_key: str | None = Header(default=None)):
    if not x_idempotency_key:
        x_idempotency_key = str(uuid.uuid4())

    # 멱등성 키가 이미 있으면 기존 job_id 반환
    existing = r.get(f"{IDEMPOTENCY_PREFIX}{x_idempotency_key}")
    if existing:
        return {"job_id": existing, "idempotency_key": x_idempotency_key}

    job_id = str(uuid.uuid4())
    r.set(f"{IDEMPOTENCY_PREFIX}{x_idempotency_key}", job_id)

    job = {
        "job_id": job_id,
        "status": "QUEUED",
        "attempt": 0,
        "payload": payload,
        "prompt_id": None,
    }

    r.set(f"{JOB_KEY_PREFIX}{job_id}", json.dumps(job))
    r.lpush(QUEUE_NAME, job_id)

    return {"job_id": job_id, "idempotency_key": x_idempotency_key}

@app.get("/v1/jobs/{job_id}")
def get_job(job_id: str):
    raw = r.get(f"{JOB_KEY_PREFIX}{job_id}")
    if not raw:
        raise HTTPException(status_code=404, detail="job not found")
    return json.loads(raw)

# worker.py
import json
import random
import time
import requests
import redis

COMFY_BASE = "http://127.0.0.1:8188"
QUEUE_NAME = "comfyui:jobs"
JOB_KEY_PREFIX = "job:"

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

MAX_ATTEMPTS = 5
BASE_BACKOFF_SEC = 2.0
MAX_BACKOFF_SEC = 60.0


def backoff(attempt: int) -> float:
    # 지수 백오프 + 지터
    exp = min(MAX_BACKOFF_SEC, BASE_BACKOFF_SEC * (2 ** max(0, attempt - 1)))
    jitter = random.uniform(0, exp * 0.2)
    return exp + jitter


def update_job(job_id: str, patch: dict):
    raw = r.get(f"{JOB_KEY_PREFIX}{job_id}")
    job = json.loads(raw)
    job.update(patch)
    r.set(f"{JOB_KEY_PREFIX}{job_id}", json.dumps(job))
    return job


def comfy_submit(workflow: dict) -> str:
    # ComfyUI는 {"prompt": workflow} 형태를 주로 사용
    resp = requests.post(f"{COMFY_BASE}/prompt", json={"prompt": workflow}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    return data["prompt_id"]


def comfy_history(prompt_id: str) -> dict:
    resp = requests.get(f"{COMFY_BASE}/history/{prompt_id}", timeout=10)
    resp.raise_for_status()
    return resp.json()


def is_done(history: dict, prompt_id: str) -> bool:
    # history 응답 구조는 워크플로/버전에 따라 다를 수 있음
    return prompt_id in history


def extract_images(history: dict, prompt_id: str) -> list[dict]:
    # 예시: outputs에서 filename, subfolder 등을 뽑아낸다고 가정
    item = history[prompt_id]
    outputs = item.get("outputs", {})
    images = []
    for node_id, out in outputs.items():
        for img in out.get("images", []):
            images.append({
                "filename": img.get("filename"),
                "subfolder": img.get("subfolder"),
                "type": img.get("type"),
            })
    return images


def should_retry_http(status_code: int) -> bool:
    if status_code in [502, 503, 504]:
        return True
    return False


def run_once(job_id: str):
    raw = r.get(f"{JOB_KEY_PREFIX}{job_id}")
    job = json.loads(raw)

    attempt = int(job.get("attempt", 0))
    if attempt >= MAX_ATTEMPTS:
        update_job(job_id, {"status": "FAILED", "error": "max attempts exceeded"})
        return

    try:
        update_job(job_id, {"status": "DISPATCHING"})

        # 멱등성: prompt_id가 이미 있으면 제출 생략
        prompt_id = job.get("prompt_id")
        if not prompt_id:
            workflow = job["payload"]["workflow"]
            prompt_id = comfy_submit(workflow)
            update_job(job_id, {"prompt_id": prompt_id})

        update_job(job_id, {"status": "RUNNING"})

        # 폴링(운영에서는 WebSocket 이벤트로 대체 가능)
        deadline = time.time() + 300
        while time.time() < deadline:
            hist = comfy_history(prompt_id)
            if is_done(hist, prompt_id):
                images = extract_images(hist, prompt_id)
                update_job(job_id, {"status": "SUCCEEDED", "images": images})
                return
            time.sleep(1.0)

        raise TimeoutError("history polling timeout")

    except requests.HTTPError as e:
        code = e.response.status_code if e.response is not None else 0
        if code and should_retry_http(code):
            attempt += 1
            update_job(job_id, {"status": "RETRYING", "attempt": attempt, "error": f"http {code}"})
            time.sleep(backoff(attempt))
            r.lpush(QUEUE_NAME, job_id)
            return

        # 400 계열은 보통 워크플로/입력 오류로 즉시 실패
        update_job(job_id, {"status": "FAILED", "error": f"http error {code}"})

    except Exception as e:
        attempt += 1
        if attempt >= MAX_ATTEMPTS:
            update_job(job_id, {"status": "FAILED", "attempt": attempt, "error": str(e)})
            return
        update_job(job_id, {"status": "RETRYING", "attempt": attempt, "error": str(e)})
        time.sleep(backoff(attempt))
        r.lpush(QUEUE_NAME, job_id)


def main():
    while True:
        item = r.brpop(QUEUE_NAME, timeout=5)
        if not item:
            continue
        _, job_id = item
        run_once(job_id)


if __name__ == "__main__":
    main()