From 4f1496c1e708d654dc5c3aa81c8eeb4e599a3783 Mon Sep 17 00:00:00 2001 From: Alexander Tujicov Date: Wed, 9 Sep 2026 19:42:34 +0300 Subject: [PATCH] Harden parallel fal queue submissions --- README.md | 1 + nano_seed.py | 94 ++++++++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 84 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index b4c3a4a..e19b16e 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,7 @@ Different models have different requirements implemented in the node: * **Queue failures**: The node reports the failed run number and fal request ID. Check account balance, model access, input validation, and safety filters. * **Queue timeout**: Increase `queue_timeout` when fal.ai is busy or when using slower high-resolution models. +* **Many parallel nodes**: NanoSeed limits simultaneous payload uploads to four and retries transient upload failures once. Queue generations remain asynchronous and can continue in parallel after submission. * **Nodes still run in order**: Restart ComfyUI after updating the node. NanoSeed nodes connected by data dependencies must remain sequential; independent branches run concurrently on ComfyUI versions with async-node scheduling. * **Resolution Errors**: If using Seedream, ensure your resolution is set high enough (min 1920x1920). If using Flux, ensure it is within 512-2048. diff --git a/nano_seed.py b/nano_seed.py index 40717e5..2aa78f9 100644 --- a/nano_seed.py +++ b/nano_seed.py @@ -14,8 +14,16 @@ from PIL import Image FAL_RUN_URL_PREFIX = "https://fal.run/" FAL_QUEUE_URL_PREFIX = "https://queue.fal.run/" FAL_HTTP_TIMEOUT_SECONDS = 60 +FAL_QUEUE_SUBMIT_TIMEOUT_SECONDS = 300 FAL_QUEUE_POLL_INTERVAL_SECONDS = 1.0 FAL_MAX_CONCURRENT_RUNS = 8 +FAL_MAX_CONCURRENT_SUBMISSIONS = 4 +FAL_QUEUE_SUBMIT_MAX_ATTEMPTS = 2 +FAL_QUEUE_SUBMIT_RETRY_BASE_SECONDS = 2.0 +FAL_RETRYABLE_STATUS_CODES = {408, 425, 429, 500, 502, 503, 504} + +_fal_submission_semaphore = None +_fal_submission_semaphore_loop = None GPT_IMAGE_25_EDIT_URLS = { "gpt_image_2_5_flare_edit": "https://fal.run/openai/gpt-image-2.5/flare/edit", @@ -310,6 +318,74 @@ def _queue_operation_url(value, fallback): return f"{FAL_QUEUE_URL_PREFIX.rstrip('/')}/{value.lstrip('/')}" +def _get_fal_submission_semaphore(): + global _fal_submission_semaphore + global _fal_submission_semaphore_loop + + loop = asyncio.get_running_loop() + if _fal_submission_semaphore is None or _fal_submission_semaphore_loop is not loop: + _fal_submission_semaphore = asyncio.Semaphore(FAL_MAX_CONCURRENT_SUBMISSIONS) + _fal_submission_semaphore_loop = loop + return _fal_submission_semaphore + + +def _fal_retry_delay(response, attempt): + if response is not None: + retry_after = response.headers.get("Retry-After") + if retry_after: + try: + return min(max(float(retry_after), 0.0), 30.0) + except ValueError: + pass + return FAL_QUEUE_SUBMIT_RETRY_BASE_SECONDS * (2 ** (attempt - 1)) + + +async def _post_fal_queue_request(queue_url, payload, headers, run_label): + for attempt in range(1, FAL_QUEUE_SUBMIT_MAX_ATTEMPTS + 1): + submit_response = None + try: + async with _get_fal_submission_semaphore(): + submit_response = await asyncio.to_thread( + requests.post, + queue_url, + json=payload, + headers=headers, + timeout=FAL_QUEUE_SUBMIT_TIMEOUT_SECONDS, + ) + except (requests.ConnectionError, requests.Timeout) as error: + if attempt >= FAL_QUEUE_SUBMIT_MAX_ATTEMPTS: + raise ValueError( + f"fal.ai queue submission failed after {attempt} attempts: {error}" + ) from error + delay = _fal_retry_delay(None, attempt) + print( + f"NanoSeed: queue upload {run_label} failed; retrying in {delay:g}s " + f"(attempt {attempt + 1}/{FAL_QUEUE_SUBMIT_MAX_ATTEMPTS})." + ) + await asyncio.sleep(delay) + continue + except requests.RequestException as error: + raise ValueError(f"fal.ai queue submission failed: {error}") from error + + if ( + submit_response.status_code in FAL_RETRYABLE_STATUS_CODES + and attempt < FAL_QUEUE_SUBMIT_MAX_ATTEMPTS + ): + delay = _fal_retry_delay(submit_response, attempt) + submit_response.close() + print( + f"NanoSeed: fal.ai returned HTTP {submit_response.status_code} for " + f"queue upload {run_label}; retrying in {delay:g}s " + f"(attempt {attempt + 1}/{FAL_QUEUE_SUBMIT_MAX_ATTEMPTS})." + ) + await asyncio.sleep(delay) + continue + + return submit_response + + raise ValueError("fal.ai queue submission exhausted its retry attempts.") + + async def _cancel_queue_request(cancel_url, headers): try: await asyncio.to_thread( @@ -324,16 +400,13 @@ async def _cancel_queue_request(cancel_url, headers): async def _submit_fal_queue_request(run_url, payload, headers, queue_timeout, run_index, run_count): queue_url = _queue_endpoint(run_url) - try: - submit_response = await asyncio.to_thread( - requests.post, - queue_url, - json=payload, - headers=headers, - timeout=FAL_HTTP_TIMEOUT_SECONDS, - ) - except requests.RequestException as error: - raise ValueError(f"fal.ai queue submission failed: {error}") from error + run_label = f"{run_index + 1}/{run_count}" + submit_response = await _post_fal_queue_request( + queue_url, + payload, + headers, + run_label, + ) submit_data = _response_json(submit_response, "fal.ai queue submission") request_id = submit_data.get("request_id") @@ -354,7 +427,6 @@ async def _submit_fal_queue_request(run_url, payload, headers, queue_timeout, ru f"{request_base_url}/cancel", ) deadline = time.monotonic() + queue_timeout - run_label = f"{run_index + 1}/{run_count}" print(f"NanoSeed: queued async run {run_label} ({request_id}).") try: