Harden parallel fal queue submissions

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.
 

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: