diff --git a/README.md b/README.md index 8397368..1b8320f 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ ComfyUI Distributed supports four types of workers: - **Local Workers** - Additional GPUs on the same machine (auto-configured on first launch) - **Remote Workers** - GPUs on other computers in your network - **Cloud Workers** - GPUs hosted on a cloud service like Runpod, accessible via secure tunnels -- **Serverless (Runpod)** - Queue-based endpoint pools for image Collector jobs; no Distributed install or inbound tunnel on workers. [Setup, endpoint contract, and preview limitations](/docs/serverless-workers.md). +- **Serverless (Runpod)** - Queue-based Runpod endpoints for image Collector jobs; no Distributed install or inbound tunnel on workers. [Setup, endpoint contract, and preview limitations](/docs/serverless-workers.md). > For detailed setup instructions, see the [setup guide](/docs/worker-setup-guides.md) diff --git a/api/config_routes.py b/api/config_routes.py index 882d380..db59743 100644 --- a/api/config_routes.py +++ b/api/config_routes.py @@ -299,7 +299,7 @@ async def update_config_endpoint(request): if isinstance(worker, dict) and worker.get("type") == "serverless" and worker.get("enabled", False): store = store or _serverless_secret_store() if worker["id"] in changed_ids or not store.has(worker["id"]): - return _error_response("Save an API key before enabling this pool", 400) + return _error_response("Save an API key before enabling this worker", 400) if changed_ids: store = store or _serverless_secret_store() for worker_id in sorted(changed_ids): @@ -316,7 +316,7 @@ async def update_config_endpoint(request): secret_changes.commit() return response except _ActivePoolCheckError: - return _error_response("Finish or reconcile unresolved serverless jobs before changing this pool", 409) + return _error_response("Finish or reconcile unresolved serverless jobs before changing this worker", 409) except Exception: return _error_response("Unable to update config", 500) @@ -434,7 +434,7 @@ async def _update_worker_endpoint(request, secret_changes): _ensure_pool_idle(worker_id) if validated.get("enabled", False) and next_secret is None: - return _error_response("Save an API key before enabling this pool; disable it before clearing the key", 400) + return _error_response("Save an API key before enabling this worker; turn the worker off before clearing its key", 400) if next_secret != prior_secret: if next_secret is None: @@ -489,7 +489,7 @@ async def _update_worker_endpoint(request, secret_changes): return web.json_response(result) except _ActivePoolCheckError: - return _error_response("Finish or reconcile unresolved serverless jobs before changing this pool", 409) + return _error_response("Finish or reconcile unresolved serverless jobs before changing this worker", 409) except ValueError: return _error_response("Invalid worker configuration", 400) except Exception: @@ -537,7 +537,7 @@ async def _delete_worker_endpoint(request, secret_changes): return web.json_response(result) except _ActivePoolCheckError: - return _error_response("Finish or reconcile unresolved serverless jobs before changing this pool", 409) + return _error_response("Finish or reconcile unresolved serverless jobs before changing this worker", 409) except Exception: return _error_response("Unable to delete worker", 500) diff --git a/api/orchestration/serverless_runtime.py b/api/orchestration/serverless_runtime.py index 128ef4a..c7968f9 100644 --- a/api/orchestration/serverless_runtime.py +++ b/api/orchestration/serverless_runtime.py @@ -130,7 +130,7 @@ class ServerlessManager: self.pool_limits[pool_id] = current elif current[0] != limit: if self.pool_is_active(pool_id): - raise ValueError("Wait for this serverless pool's current jobs before changing concurrency") + raise ValueError("Wait for this serverless worker's current jobs before changing concurrency") current = (limit, asyncio.Semaphore(limit)) self.pool_limits[pool_id] = current return current[1] @@ -320,7 +320,7 @@ class ServerlessManager: async with self.recovery_lock: record = next((r for r in self._detached_records(worker_id) if r["record_id"] == record_id), None) if record is None: - raise ValueError("Unresolved inactive job not found for this pool") + raise ValueError("Unresolved inactive job not found for this worker") self.ledger.update(record_id, status="RESOLVED", updated_at=time.time(), error="Recovery explicitly acknowledged after operator checked Runpod; not provider-confirmed") await self._notify_capacity() diff --git a/api/serverless_routes.py b/api/serverless_routes.py index 27592df..4c904cd 100644 --- a/api/serverless_routes.py +++ b/api/serverless_routes.py @@ -83,7 +83,7 @@ async def _recover(request, *, acknowledge=False): result = {"resolved": await manager.reconcile(worker_id=worker_id)} return web.json_response(dict(result, ok=True, jobs=manager.public_jobs(worker_id))) except ValueError: - return web.json_response({"ok": False, "message": "Unresolved inactive job not found for this pool"}, status=409) + return web.json_response({"ok": False, "message": "Unresolved inactive job not found for this worker"}, status=409) except Exception: return web.json_response({"ok": False, "message": "Recovery could not be confirmed; retained jobs remain protected"}, status=502) diff --git a/api/worker_routes.py b/api/worker_routes.py index 709fb77..a92b541 100644 --- a/api/worker_routes.py +++ b/api/worker_routes.py @@ -461,7 +461,7 @@ async def launch_worker_endpoint(request): return await handle_api_error(request, f"Worker {worker_id} not found", 404) if worker.get("type") == "serverless": - return await handle_api_error(request, "Serverless pools are managed by Runpod, not local processes", 400) + return await handle_api_error(request, "Serverless workers are managed by Runpod, not local processes", 400) # Ensure consistent string ID worker_id_str = worker_id diff --git a/docs/serverless-workers.md b/docs/serverless-workers.md index b3abd2f..ca79d9c 100644 --- a/docs/serverless-workers.md +++ b/docs/serverless-workers.md @@ -1,6 +1,6 @@ -# Runpod serverless pools (preview) +# Runpod serverless workers (preview) -A **Serverless (Runpod)** worker is a queue-based endpoint/pool, not a permanent +A **Serverless (Runpod)** worker is a queue-based endpoint, not a permanent ComfyUI machine. Runpod chooses the actual GPU workers. Distributed submits one normal ComfyUI API workflow per logical slot and collects the returned images. The endpoint workers **do not need ComfyUI-Distributed installed**. @@ -12,13 +12,14 @@ The endpoint workers **do not need ComfyUI-Distributed installed**. 3. Start with **Jobs per run = 1**, **Concurrent jobs = 1**, and **Timeout = 1800**. 4. Save, then choose **Test Saved Endpoint**. This checks authentication and endpoint health only; it does not start GPU inference or prove model/node compatibility. -5. Enable the pool when you want queued Distributed workflows to use it. +5. Enable the worker when you want queued Distributed workflows to use it. 6. Use an image **Distributed Collector** and queue through Distributed as usual. Enable **Delegate Only** if the master should collect without generating locally. Keys are write-only. Leave the key field blank to retain a saved key, or select -**Clear saved API key** to remove it (disable the pool first). A saved key is -required before enabling a pool. Use HTTPS or a trusted local connection when +**Clear saved API key** to remove it. The checkbox is unavailable while the +worker is enabled, so turn the worker off first. A saved key is required before +enabling a worker. Use HTTPS or a trusted local connection when entering keys. Like ComfyUI itself, the panel must not be exposed to untrusted users: anyone able to queue arbitrary workflows can run code and spend endpoint credits. @@ -73,11 +74,11 @@ incomplete serverless batch. - **Jobs per run** adds logical collector participants. Each receives its own Distributed Seed/Value overrides. It is not a physical GPU count. -- **Concurrent jobs** bounds this master's outstanding submissions per pool, +- **Concurrent jobs** bounds this master's outstanding submissions per worker, including retained jobs after a restart. Ambiguous submissions and unconfirmed cancellations keep their capacity reservation until reconciled. Other applications using the same endpoint are unaffected. -- **Load balance** selects one participant, so it does not fan out every pool slot. +- **Load balance** selects one participant, so it does not fan out every job slot. - **Timeout** includes waiting for a local slot, Runpod queue/cold-start time, execution, and output retrieval. Provider policy limits are submitted as well. - A zero-worker endpoint can be healthy: Runpod scales from zero on submission. @@ -98,13 +99,13 @@ available on Windows. `distributed/serverless/jobs.json` records provider job IDs and sanitized states, not image data, workflow payloads, or API keys. A restarted master cannot resume an interrupted collector. It does **not** automatically resubmit jobs. Retained -IDs are available through `/distributed/serverless/status/`; the pool's +IDs are available through `/distributed/serverless/status/`; the worker's cancel action can request cancellation of known pre-restart jobs. If no job ID was received, check Runpod manually. Cancellation is best-effort and may race with completion; `CANCEL_REQUESTED` is not a provider-confirmed cancellation. -Unresolved records protect the pool's endpoint, concurrency, API key, worker type, -and deletion even when no local task remains. Disabling a pool is still allowed; +Unresolved records protect the worker's endpoint, concurrency, API key, worker type, +and deletion even when no local task remains. Disabling the worker is still allowed; it does not cancel existing jobs. On restart, never-submitted `WAITING` slots are closed locally, while interrupted `SUBMITTING` records without IDs become `UNKNOWN` and remain protected. @@ -122,7 +123,7 @@ Recovery uses the same trusted-master API as the panel; it never resubmits a job Network errors, expired IDs, or ambiguous responses do not. ```json - {"worker_id": "YOUR_POOL_ID"} + {"worker_id": "YOUR_WORKER_ID"} ``` 3. If no ID was returned, or Runpod no longer retains its status, inspect Runpod @@ -130,7 +131,7 @@ Recovery uses the same trusted-master API as the panel; it never resubmits a job queued**, send `POST /distributed/serverless/resolve` for that one record: ```json - {"worker_id": "YOUR_POOL_ID", "record_id": "LOCAL_RECORD_ID", "provider_checked": true} + {"worker_id": "YOUR_WORKER_ID", "record_id": "LOCAL_RECORD_ID", "provider_checked": true} ``` This records `RESOLVED`: an explicit operator acknowledgment, **not** a @@ -139,7 +140,7 @@ Recovery uses the same trusted-master API as the panel; it never resubmits a job already-waiting jobs proceed. Never automate this acknowledgment as a retry mechanism; an unverified job can still be billable. -Pool changes become available once all its local tasks and unresolved records +Worker setting changes become available once all its local tasks and unresolved records are finished. No credentials or ledger files need to be edited by hand. ## First-version limits diff --git a/utils/network.py b/utils/network.py index d24cc92..4760cd8 100644 --- a/utils/network.py +++ b/utils/network.py @@ -88,7 +88,7 @@ def _split_host_and_port(host): def build_worker_url(worker, endpoint=""): """Construct the worker base URL with optional endpoint.""" if worker.get("type") == "serverless": - raise ValueError("Serverless pools have no ComfyUI worker URL") + raise ValueError("Serverless workers have no ComfyUI worker URL") host = (worker.get("host") or "").strip() port = int(worker.get("port", worker.get("listen_port", 8188)) or 8188) diff --git a/web/tests/serverlessWorkers.test.js b/web/tests/serverlessWorkers.test.js index e1a40b9..d975fbf 100644 --- a/web/tests/serverlessWorkers.test.js +++ b/web/tests/serverlessWorkers.test.js @@ -192,6 +192,23 @@ describe("Runpod serverless worker panel", () => { expect(form.children.length).toBeGreaterThan(0); }); + it("disables Clear saved API key while the worker is enabled and explains why", () => { + vi.spyOn(document, "getElementById").mockReturnValue(null); + const base = { name: "Runpod", type: "serverless", endpoint_id: "ep_123" }; + createWorkerSettingsForm(createFormUi(), {}, { ...base, id: "on", enabled: true, api_key_configured: true }); + createWorkerSettingsForm(createFormUi(), {}, { ...base, id: "off", enabled: false, api_key_configured: true }); + createWorkerSettingsForm(createFormUi(), {}, { ...base, id: "nokey", enabled: false, api_key_configured: false }); + + expect(elements.get("clear-api-key-on").disabled).toBe(true); + expect(elements.get("clear-api-key-group-on").title).toMatch(/turn this worker off/i); + expect(elements.get("clear-api-key-off").disabled).toBe(false); + expect(elements.get("clear-api-key-nokey").disabled).toBe(true); + expect(elements.get("clear-api-key-group-nokey").title).toMatch(/no api key is saved/i); + for (const id of ["on", "off", "nokey"]) { + expect(elements.get(`clear-api-key-group-${id}`).title).not.toMatch(/pool/i); + } + }); + it("saves fixed Runpod settings without enabling or retaining the API key", async () => { const worker = { id: "w2", name: "Old", type: "remote", enabled: false, api_key_configured: false }; const extension = makeSaveExtension(worker); diff --git a/web/ui/settingsForm.js b/web/ui/settingsForm.js index ec0149e..13e879e 100644 --- a/web/ui/settingsForm.js +++ b/web/ui/settingsForm.js @@ -150,7 +150,17 @@ export function createWorkerSettingsForm(ui, extension, worker) { clearApiKeyInput.id = `clear-api-key-${worker.id}`; const clearApiKeyLabel = document.createElement("label"); clearApiKeyLabel.htmlFor = clearApiKeyInput.id; - clearApiKeyLabel.textContent = "Clear saved API key (disable pool first)"; + clearApiKeyLabel.textContent = "Clear saved API key"; + // The backend rejects clearing the key of an enabled worker; say so up front. + // The form is rebuilt when the enable toggle changes, so this stays current. + const clearApiKeyBlockedReason = !worker.api_key_configured + ? "No API key is saved for this worker." + : worker.enabled + ? "Turn this worker off before clearing its API key. An enabled serverless worker needs a saved key." + : ""; + clearApiKeyInput.disabled = Boolean(clearApiKeyBlockedReason); + clearApiKeyGroup.title = clearApiKeyBlockedReason || "Remove the saved Runpod API key when you save."; + clearApiKeyLabel.style.cssText = clearApiKeyBlockedReason ? "opacity: 0.5; cursor: not-allowed;" : "cursor: pointer;"; clearApiKeyGroup.appendChild(clearApiKeyInput); clearApiKeyGroup.appendChild(clearApiKeyLabel); clearApiKeyInput.onchange = () => {