fix: integrate alpha preservation with current collector transports

Merge main while retaining exact transfer, serverless cleanup and image assembly error propagation. Update RGB fixtures and verify mixed RGB/RGBA exact HTTP transport without value loss.
This commit is contained in:
Robert Wojciechowski
2026-10-14 10:46:16 +00:00
161 changed files with 32349 additions and 612 deletions
+2
View File
@@ -1,5 +1,7 @@
from . import config_routes # noqa: F401
from . import serverless_routes # noqa: F401
from . import tunnel_routes # noqa: F401
from . import worker_routes # noqa: F401
from . import job_routes # noqa: F401
from . import usdu_routes # noqa: F401
from . import image_list_routes # noqa: F401
+382 -69
View File
@@ -1,5 +1,6 @@
import json
from contextlib import asynccontextmanager
from typing import Any
from aiohttp import web
import server
@@ -24,6 +25,12 @@ except ImportError:
save_config(config)
from ..utils.logging import debug_log, log
from ..utils.network import handle_api_error, normalize_host
from ..utils.secret_fields import is_secret_field_name
from ..utils.serverless_config import (
ServerlessSecretChanges,
public_config,
validate_serverless_worker,
)
def _positive_int(value):
@@ -86,22 +93,169 @@ def _apply_field_patch(target: dict, data: dict, field_rules: list) -> None:
target[key] = normalizer(value) if (normalizer and value is not None) else value
_SECRET_FIELDS = {"api_key", "clear_api_key"}
_SERVERLESS_LOCAL_FIELDS = {"host", "port", "listen_port", "cuda_device", "extra_args"}
_SERVERLESS_ONLY_FIELDS = {
"provider", "endpoint_id", "jobs_per_run", "concurrency", "timeout_seconds", "api_key_configured",
"media_profile",
}
_SERVERLESS_CONFIG_FIELDS = {
"worker_id", "type", "provider", "endpoint_id", "jobs_per_run", "concurrency",
"timeout_seconds", "enabled", "name", "api_key", "clear_api_key", "media_profile",
}
class _ActivePoolCheckError(RuntimeError):
pass
def _error_response(message, status):
return web.json_response({"status": "error", "message": message}, status=status)
def _contains_secret_fields(value, *, allow_root_serverless_secrets=False, depth=0):
if isinstance(value, dict):
for key, item in value.items():
field_name = key.casefold() if isinstance(key, str) else ""
if is_secret_field_name(field_name):
if not (allow_root_serverless_secrets and depth == 0 and key in _SECRET_FIELDS):
return True
if _contains_secret_fields(
item, allow_root_serverless_secrets=allow_root_serverless_secrets, depth=depth + 1
):
return True
elif isinstance(value, (list, tuple)):
return any(
_contains_secret_fields(
item, allow_root_serverless_secrets=allow_root_serverless_secrets, depth=depth + 1
)
for item in value
)
return False
def _serverless_secret_store():
from .orchestration.serverless_runtime import get_secret_store
return get_secret_store()
def _serverless_pool_is_active(worker_id):
from .orchestration.serverless_runtime import get_manager
return get_manager().pool_is_active(worker_id)
def _ensure_pool_idle(worker_id):
try:
active = _serverless_pool_is_active(worker_id)
except Exception:
raise _ActivePoolCheckError from None
if active:
raise _ActivePoolCheckError
def _worker_by_id(workers, worker_id):
for worker in workers:
if isinstance(worker, dict) and worker.get("id") == worker_id:
return worker
return None
def _worker_type(worker):
return worker.get("type", "local") if isinstance(worker, dict) else None
def _strip_sensitive_fields(value: Any) -> Any:
if isinstance(value, dict):
return {
key: _strip_sensitive_fields(item)
for key, item in value.items()
if not is_secret_field_name(key)
}
if isinstance(value, list):
return [_strip_sensitive_fields(item) for item in value]
return value
def _clean_serverless_worker(worker):
clean = _strip_sensitive_fields(dict(worker))
for field in _SERVERLESS_LOCAL_FIELDS | {"api_key_configured"}:
clean.pop(field, None)
return clean
def _validate_bulk_workers(workers):
validated = []
for worker in workers:
if isinstance(worker, dict) and worker.get("type") == "serverless":
clean = _clean_serverless_worker(worker)
validated.append(validate_serverless_worker(clean))
else:
validated.append(worker)
return validated
def _clean_bulk_type_conversions(old_workers, new_workers):
old_by_id = {str(w.get("id")): w for w in old_workers if isinstance(w, dict) and w.get("id") is not None}
for index, worker in enumerate(new_workers):
if not isinstance(worker, dict) or worker.get("id") is None:
continue
old = old_by_id.get(str(worker["id"]))
if old is None or old.get("type") != "serverless":
continue
if worker.get("type") == "serverless":
# Older clients do not know this field. Only an explicit selection
# may change the endpoint contract, including on bulk saves.
if "media_profile" not in worker and "media_profile" in old:
worker["media_profile"] = old["media_profile"]
continue
converted = dict(worker)
for field in _SERVERLESS_ONLY_FIELDS:
converted.pop(field, None)
new_workers[index] = converted
return new_workers
def _bulk_serverless_changes(old_workers, new_workers):
old_by_id = {str(w.get("id")): w for w in old_workers if isinstance(w, dict) and w.get("id") is not None}
new_by_id = {str(w.get("id")): w for w in new_workers if isinstance(w, dict) and w.get("id") is not None}
secret_ids = set()
for worker_id, old in old_by_id.items():
new = new_by_id.get(worker_id)
if old.get("type") == "serverless":
if new is None or new.get("type") != "serverless":
_ensure_pool_idle(worker_id)
secret_ids.add(worker_id)
elif (
old.get("endpoint_id") != new.get("endpoint_id")
or old.get("media_profile", "legacy-image-v1") != new.get("media_profile", "legacy-image-v1")
):
_ensure_pool_idle(worker_id)
elif new is not None and new.get("type") == "serverless":
secret_ids.add(worker_id)
return secret_ids
@server.PromptServer.instance.routes.get("/distributed/config")
async def get_config_endpoint(request):
config = load_config()
return web.json_response(config)
try:
config = load_config()
return web.json_response(public_config(config, _serverless_secret_store()))
except Exception:
return _error_response("Unable to read config", 500)
@server.PromptServer.instance.routes.post("/distributed/config")
async def update_config_endpoint(request):
"""Bulk config update with schema validation."""
"""Bulk config update with schema validation and credential redaction."""
try:
data = await request.json()
except Exception as e:
return await handle_api_error(request, f"Invalid JSON payload: {e}", 400)
except Exception:
return _error_response("Invalid JSON payload", 400)
if not isinstance(data, dict):
return await handle_api_error(request, "Config payload must be an object", 400)
return _error_response("Config payload must be an object", 400)
if _contains_secret_fields(data):
return _error_response("API keys cannot be set through bulk config", 400)
validated_settings = {}
validated_root = {}
@@ -118,9 +272,15 @@ async def update_config_endpoint(request):
continue
if validator and not validator(value):
errors.append(f"{key}: value {value!r} failed validation")
errors.append(f"{key}: failed validation")
continue
if key == "workers":
try:
value = _validate_bulk_workers(value)
except (TypeError, ValueError):
return _error_response("Invalid serverless worker configuration", 400)
if key in _SETTINGS_FIELDS:
validated_settings[key] = value
else:
@@ -133,15 +293,41 @@ async def update_config_endpoint(request):
"message": "; ".join(errors),
}, status=400)
store = None
response_config = None
try:
async with config_transaction() as config:
settings = config.setdefault("settings", {})
settings.update(validated_settings)
for key, value in validated_root.items():
config[key] = value
return web.json_response({"status": "success", "config": config})
except Exception as e:
return await handle_api_error(request, e)
with ServerlessSecretChanges() as secret_changes:
async with config_transaction() as config:
if "workers" in validated_root:
old_workers = config.get("workers", [])
validated_root["workers"] = _clean_bulk_type_conversions(
old_workers, validated_root["workers"]
)
changed_ids = _bulk_serverless_changes(old_workers, validated_root["workers"])
for worker in validated_root["workers"]:
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 worker", 400)
if changed_ids:
store = store or _serverless_secret_store()
for worker_id in sorted(changed_ids):
prior_secret = secret_changes.capture(store, worker_id)
if prior_secret is not None:
secret_changes.delete(store, worker_id)
settings = config.setdefault("settings", {})
settings.update(validated_settings)
for key, value in validated_root.items():
config[key] = value
response_config = public_config(config, store or _serverless_secret_store())
response = web.json_response({"status": "success", "config": response_config})
secret_changes.commit()
return response
except _ActivePoolCheckError:
return _error_response("Finish or reconcile unresolved serverless jobs before changing this worker", 409)
except Exception:
return _error_response("Unable to update config", 500)
@server.PromptServer.instance.routes.get("/distributed/queue_status/{job_id}")
@@ -164,79 +350,206 @@ async def queue_status_endpoint(request):
@server.PromptServer.instance.routes.post("/distributed/config/update_worker")
async def update_worker_endpoint(request):
return await _run_with_serverless_secret_changes(_update_worker_endpoint, request)
async def _run_with_serverless_secret_changes(endpoint, request):
with ServerlessSecretChanges() as secret_changes:
response = await endpoint(request, secret_changes)
if response.status < 400:
secret_changes.commit()
return response
async def _update_worker_endpoint(request, secret_changes):
store = None
worker_id = None
try:
data = await request.json()
try:
data = await request.json()
except Exception:
return _error_response("Invalid JSON payload", 400)
if not isinstance(data, dict):
return _error_response("Worker payload must be an object", 400)
worker_id = data.get("worker_id")
if worker_id is None:
return await handle_api_error(request, "Missing worker_id", 400)
return _error_response("Missing worker_id", 400)
if _contains_secret_fields(data, allow_root_serverless_secrets=True):
return _error_response("Invalid worker credentials", 400)
result = None
async with config_transaction() as config:
worker_found = False
workers = config.setdefault("workers", [])
existing = _worker_by_id(workers, worker_id)
current_type = _worker_type(existing)
target_type = data.get("type", current_type or "local")
for worker in workers:
if worker["id"] == worker_id:
_apply_field_patch(worker, data, _WORKER_FIELDS)
worker_found = True
break
if target_type == "serverless":
unknown = set(data) - _SERVERLESS_CONFIG_FIELDS - _SERVERLESS_LOCAL_FIELDS
if unknown:
return _error_response("Unknown serverless worker field", 400)
if not isinstance(worker_id, str) or not worker_id.strip():
return _error_response("worker_id must be a non-empty string", 400)
if "api_key" in data and not isinstance(data["api_key"], str):
return _error_response("Invalid worker credentials", 400)
if "clear_api_key" in data and type(data["clear_api_key"]) is not bool:
return _error_response("Invalid worker credentials", 400)
supplied_key = data.get("api_key", "")
clear_key = data.get("clear_api_key", False)
if supplied_key and clear_key:
return _error_response("Choose either api_key or clear_api_key", 400)
if "enabled" in data and type(data["enabled"]) is not bool:
return _error_response("enabled must be a boolean", 400)
if "name" in data and not isinstance(data["name"], str):
return _error_response("name must be a string", 400)
if not worker_found:
# If worker not found and all required fields are provided, create new worker
if all(key in data for key in ["name", "port", "cuda_device"]):
new_worker = {
"id": worker_id,
"name": data["name"],
"host": normalize_host(data.get("host", "localhost")),
"port": data["port"],
"cuda_device": data["cuda_device"],
"enabled": data.get("enabled", False),
"extra_args": data.get("extra_args", ""),
"type": data.get("type", "local")
}
workers.append(new_worker)
candidate = dict(existing or {})
candidate["id"] = worker_id
candidate["type"] = "serverless"
candidate.setdefault("provider", "runpod")
candidate.setdefault("enabled", False)
# Legacy "concurrency" is accepted (see _SERVERLESS_CONFIG_FIELDS) but ignored.
for field in ("provider", "endpoint_id", "jobs_per_run", "timeout_seconds", "enabled", "name", "media_profile"):
if field in data:
candidate[field] = data[field]
candidate = _clean_serverless_worker(candidate)
validated = validate_serverless_worker(candidate)
if "enabled" in validated and type(validated["enabled"]) is not bool:
return _error_response("enabled must be a boolean", 400)
if "name" in validated and not isinstance(validated["name"], str):
return _error_response("name must be a string", 400)
store = _serverless_secret_store()
prior_secret = secret_changes.capture(store, worker_id)
converting_to_serverless = current_type != "serverless"
next_secret = None if converting_to_serverless else prior_secret
if supplied_key:
next_secret = supplied_key
elif clear_key:
next_secret = None
endpoint_changed = (
current_type == "serverless"
and existing.get("endpoint_id") != validated.get("endpoint_id")
)
media_profile_changed = (
current_type == "serverless"
and existing.get("media_profile", "legacy-image-v1") != validated.get("media_profile", "legacy-image-v1")
)
if current_type == "serverless" and (
endpoint_changed or media_profile_changed or next_secret != prior_secret
):
_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 worker; turn the worker off before clearing its key", 400)
if next_secret != prior_secret:
if next_secret is None:
secret_changes.delete(store, worker_id)
else:
secret_changes.set(store, worker_id, next_secret)
if existing is None:
workers.append(validated)
else:
return await handle_api_error(
request,
f"Worker {worker_id} not found and missing required fields for creation",
404,
)
existing.clear()
existing.update(validated)
result = {"status": "success", "api_key_configured": next_secret is not None}
elif existing is not None:
cleaned_existing = _strip_sensitive_fields(existing)
existing.clear()
existing.update(cleaned_existing)
if current_type == "serverless":
if _contains_secret_fields(data):
return _error_response("Invalid worker credentials", 400)
_ensure_pool_idle(worker_id)
store = _serverless_secret_store()
prior_secret = secret_changes.capture(store, worker_id)
if prior_secret is not None:
secret_changes.delete(store, worker_id)
for field in _SERVERLESS_ONLY_FIELDS:
existing.pop(field, None)
elif _contains_secret_fields(data):
return _error_response("API keys are only accepted for serverless workers", 400)
_apply_field_patch(existing, data, _WORKER_FIELDS)
result = {"status": "success"}
elif all(key in data for key in ["name", "port", "cuda_device"]):
if _contains_secret_fields(data):
return _error_response("API keys are only accepted for serverless workers", 400)
new_worker = {
"id": worker_id,
"name": data["name"],
"host": normalize_host(data.get("host", "localhost")),
"port": data["port"],
"cuda_device": data["cuda_device"],
"enabled": data.get("enabled", False),
"extra_args": data.get("extra_args", ""),
"type": data.get("type", "local")
}
workers.append(new_worker)
result = {"status": "success"}
else:
return _error_response(
"Worker not found and missing required fields for creation", 404
)
return web.json_response(result)
except _ActivePoolCheckError:
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:
return _error_response("Unable to update worker configuration", 500)
return web.json_response({"status": "success"})
except Exception as e:
return await handle_api_error(request, e, 400)
@server.PromptServer.instance.routes.post("/distributed/config/delete_worker")
async def delete_worker_endpoint(request):
return await _run_with_serverless_secret_changes(_delete_worker_endpoint, request)
async def _delete_worker_endpoint(request, secret_changes):
store = None
worker_id = None
try:
data = await request.json()
try:
data = await request.json()
except Exception:
return _error_response("Invalid JSON payload", 400)
if not isinstance(data, dict):
return _error_response("Worker payload must be an object", 400)
worker_id = data.get("worker_id")
if worker_id is None:
return await handle_api_error(request, "Missing worker_id", 400)
return _error_response("Missing worker_id", 400)
result = None
async with config_transaction() as config:
workers = config.get("workers", [])
# Find and remove the worker
worker_index = -1
for i, worker in enumerate(workers):
if worker["id"] == worker_id:
worker_index = i
break
if worker_index == -1:
return await handle_api_error(request, f"Worker {worker_id} not found", 404)
# Remove the worker
removed_worker = workers.pop(worker_index)
return web.json_response({
removed = _worker_by_id(workers, worker_id)
if removed is None:
return _error_response("Worker not found", 404)
if _worker_type(removed) == "serverless":
_ensure_pool_idle(worker_id)
store = _serverless_secret_store()
secret_worker_id = str(worker_id)
prior_secret = secret_changes.capture(store, secret_worker_id)
if prior_secret is not None:
secret_changes.delete(store, secret_worker_id)
workers.remove(removed)
result = {
"status": "success",
"message": f"Worker {removed_worker.get('name', worker_id)} deleted"
})
except Exception as e:
return await handle_api_error(request, e, 400)
"message": "Worker deleted" if _worker_type(removed) == "serverless"
else f"Worker {removed.get('name', worker_id)} deleted",
}
return web.json_response(result)
except _ActivePoolCheckError:
return _error_response("Finish or reconcile unresolved serverless jobs before changing this worker", 409)
except Exception:
return _error_response("Unable to delete worker", 500)
@server.PromptServer.instance.routes.post("/distributed/config/update_setting")
async def update_setting_endpoint(request):
+94
View File
@@ -0,0 +1,94 @@
"""Bounded, typed HTTP transport for dynamic image-list worker tasks."""
import json
from aiohttp import web
import server
from ..image_list.runtime import (
IMAGE_TRANSFER_FORMATS,
MAX_REQUEST_BYTES,
PROTOCOL_VERSION,
decode_result,
delete_worker_item,
set_master_result,
store_worker_item,
)
prompt_server = server.PromptServer.instance
async def _json_body(request):
length = request.content_length
if length is not None and length > MAX_REQUEST_BYTES:
raise ValueError(f"Request exceeds the {MAX_REQUEST_BYTES}-byte image-list limit")
raw = bytearray()
async for chunk in request.content.iter_chunked(64 * 1024):
raw.extend(chunk)
if len(raw) > MAX_REQUEST_BYTES:
raise ValueError(f"Request exceeds the {MAX_REQUEST_BYTES}-byte image-list limit")
try:
return json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"Invalid image-list JSON body: {exc}") from exc
@prompt_server.routes.get("/distributed/image-list/capabilities")
async def image_list_capabilities_endpoint(_request):
return web.json_response({
"supported": True,
"execution_targets": True,
"value_results": True,
"protocol_version": PROTOCOL_VERSION,
"node_ids": ["DistributedImageListStart", "DistributedImageListEnd", "DistributedImageListGetValue"],
"transport": "typed-float-tensor-json-v1",
"image_transfer_formats": list(IMAGE_TRANSFER_FORMATS),
})
@prompt_server.routes.post("/distributed/image-list/item")
async def image_list_item_endpoint(request):
try:
data = await _json_body(request)
if not isinstance(data, dict):
raise ValueError("Expected an image-list item object")
context_id = data.get("context_id")
if not isinstance(context_id, str) or not context_id:
raise ValueError("context_id must be a non-empty string")
item = data.get("item")
identity = store_worker_item(data.get("token"), item)
return web.json_response({"status": "stored", "item_index": item["index"],
"item_identity": identity})
except (ValueError, TypeError) as exc:
return web.json_response({"error": str(exc)}, status=400)
except Exception as exc:
return web.json_response({"error": str(exc)}, status=500)
@prompt_server.routes.delete("/distributed/image-list/item/{token}")
async def image_list_delete_item_endpoint(request):
token = request.match_info.get("token", "")
return web.json_response({"status": "deleted", "removed": delete_worker_item(token)})
@prompt_server.routes.post("/distributed/image-list/result")
async def image_list_result_endpoint(request):
try:
data = await _json_body(request)
if not isinstance(data, dict):
raise ValueError("Expected an image-list result object")
context_id = data.get("context_id")
token = data.get("token")
item_index = data.get("item_index")
if not isinstance(context_id, str) or not context_id:
raise ValueError("context_id must be a non-empty string")
if not isinstance(token, str) or not token:
raise ValueError("token must be a non-empty string")
if isinstance(item_index, bool) or not isinstance(item_index, int) or item_index < 0:
raise ValueError("item_index must be a non-negative integer")
result = decode_result(data.get("result"))
set_master_result(context_id, item_index, token, result)
return web.json_response({"status": "accepted", "item_index": item_index})
except (ValueError, TypeError) as exc:
return web.json_response({"error": str(exc)}, status=400)
except Exception as exc:
return web.json_response({"error": str(exc)}, status=500)
+188 -24
View File
@@ -15,6 +15,7 @@ from ..utils.logging import debug_log
from ..utils.image import pil_to_tensor, ensure_contiguous
from ..utils.network import handle_api_error
from ..utils.constants import JOB_INIT_GRACE_PERIOD, MEMORY_CLEAR_DELAY
from ..utils import transfer_codec
try:
from .queue_orchestration import ensure_distributed_state, orchestrate_distributed_execution
except ImportError:
@@ -26,8 +27,18 @@ from .queue_request import parse_queue_request_payload
prompt_server = server.PromptServer.instance
# Canonical worker result envelope accepted by POST /distributed/job_complete:
# { "job_id": str, "worker_id": str, "batch_idx": int, "image": <base64 PNG>, "is_last": bool }
# Worker result envelopes accepted by POST /distributed/job_complete:
#
# application/json (legacy, one frame per request):
# { "job_id": str, "worker_id": str, "batch_idx": int, "image": <base64 PNG>, "is_last": bool,
# "audio"?: {...} }
#
# multipart/form-data (compressed, many frames per request):
# envelope: JSON { "job_id", "worker_id", "format", "start_index", "count", "width", "height",
# "is_last", "audio"?: {...} }
# media: one part per frame for still formats, or one video file for video formats
# format "exact" adds envelope "descriptors" (one tensor-le-v1 descriptor per
# frame) and sends one raw tensor part per frame; values are not quantized.
def _decode_image_sync(image_path):
@@ -272,8 +283,168 @@ async def check_file_endpoint(request):
return await handle_api_error(request, e, 500)
async def _enqueue_worker_items(multi_job_id, items):
"""Put decoded items on the job queue, waiting briefly for the master to create it.
Returns the queue size after enqueueing, or None if the job never appeared.
"""
deadline = time.monotonic() + float(JOB_INIT_GRACE_PERIOD)
while True:
async with prompt_server.distributed_jobs_lock:
pending = prompt_server.distributed_pending_jobs.get(multi_job_id)
if pending is not None:
for item in items:
await pending.put(item)
return pending.qsize()
if time.monotonic() > deadline:
return None
await asyncio.sleep(0.05)
def _decode_compressed_chunk_sync(envelope, parts):
"""Decode a compressed chunk into per-frame IMAGE tensors (threadpool)."""
if envelope.get("format") == transfer_codec.EXACT_FORMAT:
return _decode_exact_chunk_sync(envelope, parts)
frames = transfer_codec.decode_chunk(envelope, parts)
return [
ensure_contiguous(torch.from_numpy(transfer_codec.uint8_frame_to_tensor_array(frame)))
for frame in frames
]
def _decode_exact_chunk_sync(envelope, parts):
"""Decode raw exact IMAGE frames, checking each against the envelope geometry."""
from ..serverless_bridge import codec as bridge_codec
count, width, height = envelope.get("count"), envelope.get("width"), envelope.get("height")
for name, value in (("count", count), ("width", width), ("height", height)):
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise transfer_codec.TransferCodecError(f"envelope.{name}: expected positive integer")
descriptors = envelope.get("descriptors")
if not isinstance(descriptors, list) or len(descriptors) != count or len(parts) != count:
raise transfer_codec.TransferCodecError(
f"exact chunk expected {count} descriptors and frames, got "
f"{len(descriptors) if isinstance(descriptors, list) else 'no'} descriptors and {len(parts)} frames"
)
tensors = []
for descriptor, payload in zip(descriptors, parts):
if not isinstance(descriptor, dict) or descriptor.get("encoding") != "tensor-le-v1":
raise transfer_codec.TransferCodecError("exact frames require tensor-le-v1 encoding")
shape = descriptor.get("shape") if isinstance(descriptor, dict) else None
if not isinstance(shape, list) or len(shape) != 4 or shape[:3] != [1, height, width]:
raise transfer_codec.TransferCodecError(
f"exact frame shape {shape} does not match the envelope's 1x{height}x{width} frames"
)
try:
tensors.append(bridge_codec.decode_payload("IMAGE", descriptor, payload))
except ValueError as exc:
raise transfer_codec.TransferCodecError(f"exact frame: {exc}") from None
return tensors
def _validate_compressed_envelope(envelope):
errors = []
if not isinstance(envelope, dict):
return ["envelope: expected JSON object"]
for key in ("job_id", "worker_id"):
value = envelope.get(key)
if not isinstance(value, str) or not value.strip():
errors.append(f"{key}: expected non-empty string")
start_index = envelope.get("start_index")
if not isinstance(start_index, int) or isinstance(start_index, bool) or start_index < 0:
errors.append("start_index: expected non-negative integer")
if not isinstance(envelope.get("is_last"), bool):
errors.append("is_last: expected boolean")
try:
if transfer_codec.normalize_format(envelope.get("format")) == transfer_codec.LEGACY_FORMAT:
errors.append("format: legacy_png must be sent as JSON")
except transfer_codec.TransferCodecError as exc:
errors.append(f"format: {exc}")
audio_payload = envelope.get("audio")
if audio_payload is not None and not isinstance(audio_payload, dict):
errors.append("audio: expected object when provided")
return errors
async def _job_complete_compressed(request):
try:
form = await request.post()
except Exception as exc:
too_large = getattr(web, "HTTPRequestEntityTooLarge", None)
if too_large is not None and isinstance(exc, too_large):
return await handle_api_error(
request, f"Request body exceeds the master's upload limit (--max-upload-size): {exc}", 413)
return await handle_api_error(request, f"Invalid multipart payload: {exc}", 400)
raw_envelope = form.get("envelope")
if isinstance(raw_envelope, (bytes, bytearray)):
raw_envelope = raw_envelope.decode("utf-8")
elif hasattr(raw_envelope, "file"):
raw_envelope = raw_envelope.file.read().decode("utf-8")
try:
envelope = json.loads(raw_envelope) if isinstance(raw_envelope, str) else None
except ValueError as exc:
return await handle_api_error(request, f"envelope: invalid JSON ({exc})", 400)
if envelope is None:
return await handle_api_error(request, "envelope: missing field", 400)
errors = _validate_compressed_envelope(envelope)
if errors:
return await handle_api_error(request, errors, 400)
parts = []
for field in form.getall("media", []):
if hasattr(field, "file"):
parts.append(field.file.read())
elif isinstance(field, (bytes, bytearray)):
parts.append(bytes(field))
else:
return await handle_api_error(request, "media: expected binary file parts", 400)
loop = asyncio.get_running_loop()
try:
tensors = await loop.run_in_executor(None, _decode_compressed_chunk_sync, envelope, parts)
decoded_audio = _decode_audio_payload(envelope.get("audio")) if envelope.get("audio") is not None else None
except ValueError as exc:
return await handle_api_error(request, exc, 400)
multi_job_id = envelope["job_id"].strip()
worker_id = envelope["worker_id"].strip()
start_index = envelope["start_index"]
is_last = envelope["is_last"]
last_pos = len(tensors) - 1
items = [
{
"tensor": tensor,
"worker_id": worker_id,
"image_index": start_index + pos,
"is_last": bool(is_last and pos == last_pos),
"audio": decoded_audio if pos == last_pos else None,
"format": envelope["format"],
}
for pos, tensor in enumerate(tensors)
]
queue_size = await _enqueue_worker_items(multi_job_id, items)
if queue_size is None:
return await handle_api_error(request, "job not initialized", 404)
debug_log(
f"job_complete received {envelope['format']} chunk - job_id: {multi_job_id}, worker: {worker_id}, "
f"frames: {start_index}..{start_index + len(items) - 1}, "
f"bytes: {sum(len(p) for p in parts)}, is_last: {is_last}, queue_size: {queue_size}"
)
return web.json_response({"status": "success", "frames": len(items)})
@server.PromptServer.instance.routes.post("/distributed/job_complete")
async def job_complete_endpoint(request):
content_type = str(getattr(request, "content_type", "") or "").lower()
if content_type.startswith("multipart/"):
try:
return await _job_complete_compressed(request)
except Exception as e:
return await handle_api_error(request, e)
try:
data = await request.json()
except Exception as exc:
@@ -318,28 +489,21 @@ async def job_complete_endpoint(request):
multi_job_id = job_id.strip()
worker_id = worker_id.strip()
pending = None
queue_size = 0
deadline = time.monotonic() + float(JOB_INIT_GRACE_PERIOD)
while pending is None:
async with prompt_server.distributed_jobs_lock:
pending = prompt_server.distributed_pending_jobs.get(multi_job_id)
if pending is not None:
await pending.put(
{
"tensor": tensor,
"worker_id": worker_id,
"image_index": int(batch_idx),
"is_last": is_last,
"audio": decoded_audio,
}
)
queue_size = pending.qsize()
break
if time.monotonic() > deadline:
return await handle_api_error(request, "job not initialized", 404)
await asyncio.sleep(0.05)
queue_size = await _enqueue_worker_items(
multi_job_id,
[
{
"tensor": tensor,
"worker_id": worker_id,
"image_index": int(batch_idx),
"is_last": is_last,
"audio": decoded_audio,
"format": transfer_codec.LEGACY_FORMAT,
}
],
)
if queue_size is None:
return await handle_api_error(request, "job not initialized", 404)
debug_log(
f"job_complete received canonical envelope - job_id: {multi_job_id}, "
+184 -22
View File
@@ -6,6 +6,9 @@ import aiohttp
from ...utils.logging import debug_log, log
from ...utils.network import build_worker_url, get_client_session, probe_worker
# Must match image_list.runtime.PROTOCOL_VERSION (kept local so this module
# stays importable without the image-list runtime; a test pins them equal).
IMAGE_LIST_PROTOCOL_VERSION = 2
try:
from ...utils.trace_logger import trace_debug, trace_info
except ImportError:
@@ -55,6 +58,11 @@ async def worker_ws_is_active(worker):
async def _probe_worker_active(worker, use_websocket, semaphore):
async with semaphore:
if worker.get("type") == "serverless":
from .serverless_runtime import get_manager
# Authentication/configuration failures must surface, not silently skip paid work.
status = await get_manager().health(worker)
return worker, status["ok"]
is_active = await (worker_ws_is_active(worker) if use_websocket else worker_is_active(worker))
return worker, is_active
@@ -69,30 +77,49 @@ async def _dispatch_via_websocket(worker_url, payload, client_id, timeout=60.0):
"workflow": payload.get("workflow"),
"client_id": client_id,
}
if "partial_execution_targets" in payload:
ws_payload["partial_execution_targets"] = payload["partial_execution_targets"]
ws_url = worker_url.replace("http://", "ws://").replace("https://", "wss://")
ws_url = f"{ws_url}/distributed/worker_ws"
session = await get_client_session()
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
ws = None
async with session.ws_connect(ws_url, heartbeat=20, timeout=timeout) as ws:
await ws.send_json(ws_payload)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data or "{}")
if data.get("type") == "dispatch_ack" and data.get("request_id") == request_id:
if data.get("ok"):
return
error_text = data.get("error") or "Worker rejected websocket dispatch."
validation_error = data.get("validation_error")
node_errors = data.get("node_errors")
if validation_error:
error_text = f"{error_text} | validation_error={validation_error}"
if node_errors:
error_text = f"{error_text} | node_errors={node_errors}"
raise RuntimeError(error_text)
elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED):
raise RuntimeError(f"Worker websocket closed unexpectedly: {msg.type}")
raise RuntimeError("Worker websocket closed before dispatch_ack was received.")
try:
async with asyncio.timeout_at(deadline):
ws = await session.ws_connect(ws_url, heartbeat=20)
await ws.send_json(ws_payload)
while True:
msg = await ws.receive()
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data or "{}")
if data.get("type") == "dispatch_ack" and data.get("request_id") == request_id:
if data.get("ok"):
return data.get("prompt_id")
error_text = data.get("error") or "Worker rejected websocket dispatch."
validation_error = data.get("validation_error")
node_errors = data.get("node_errors")
if validation_error:
error_text = f"{error_text} | validation_error={validation_error}"
if node_errors:
error_text = f"{error_text} | node_errors={node_errors}"
raise RuntimeError(error_text)
elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED):
raise RuntimeError(f"Worker websocket closed before acknowledgement: {msg.type}; submission status is ambiguous")
except asyncio.TimeoutError as exc:
raise RuntimeError(
"Worker websocket dispatch timed out before dispatch_ack; submission status is ambiguous."
) from exc
finally:
if ws is not None:
# Closing must not discard an already acknowledged prompt ID or
# replace a definite rejection with an ambiguous timeout.
try:
async with asyncio.timeout_at(min(deadline, loop.time() + 1.0)):
await ws.close()
except (asyncio.TimeoutError, aiohttp.ClientError):
pass
async def dispatch_worker_prompt(
@@ -102,11 +129,18 @@ async def dispatch_worker_prompt(
client_id=None,
use_websocket=False,
trace_execution_id=None,
return_prompt_id=False,
*,
partial_execution_targets=None,
):
"""Send the prepared prompt to a worker ComfyUI instance."""
worker_url = build_worker_url(worker)
url = build_worker_url(worker, "/prompt")
payload = {"prompt": prompt_obj}
target_payload = {}
if partial_execution_targets is not None:
target_payload["partial_execution_targets"] = list(partial_execution_targets)
payload.update(target_payload)
extra_data = {}
if workflow_meta:
extra_data.setdefault("extra_pnginfo", {})["workflow"] = workflow_meta
@@ -115,15 +149,18 @@ async def dispatch_worker_prompt(
if use_websocket:
try:
await _dispatch_via_websocket(
prompt_id = await _dispatch_via_websocket(
worker_url,
{
"prompt": prompt_obj,
"workflow": workflow_meta,
**target_payload,
},
client_id,
)
return
if return_prompt_id and (not isinstance(prompt_id, str) or not prompt_id):
raise RuntimeError("Worker websocket acknowledgement omitted prompt_id")
return prompt_id if return_prompt_id else None
except Exception as exc:
worker_id = worker.get("id")
if trace_execution_id:
@@ -139,6 +176,127 @@ async def dispatch_worker_prompt(
timeout=aiohttp.ClientTimeout(total=60),
) as resp:
resp.raise_for_status()
if not return_prompt_id:
return None
try:
response_data = await resp.json()
except Exception as exc:
raise RuntimeError("Worker /prompt response did not contain JSON acknowledgement") from exc
prompt_id = response_data.get("prompt_id") if isinstance(response_data, dict) else None
if not isinstance(prompt_id, str) or not prompt_id:
raise RuntimeError("Worker /prompt response omitted prompt_id")
return prompt_id if return_prompt_id else None
async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0, *,
required_transfer_formats=(), require_value_results=False):
"""Require protocol/node support and report which direct workers are idle.
This performs read-only GETs only; it never submits a prompt or contacts a
serverless provider. Unsupported worker releases fail before master inputs run.
"""
if not workers:
raise RuntimeError("Distributed Image List requires at least one selected direct worker")
if (not isinstance(required_transfer_formats, (list, tuple, set, frozenset))
or any(type(fmt) is not str for fmt in required_transfer_formats)):
raise ValueError("required_transfer_formats must contain image transfer format names")
required_formats = set(required_transfer_formats) - {"exact"}
if required_formats:
from ...serverless_bridge.image_transfer import validate_options
for fmt in required_formats:
validate_options(fmt)
session = await get_client_session()
required_nodes = {"DistributedImageListStart", "DistributedImageListEnd"}
semaphore = asyncio.Semaphore(8)
async def inspect(worker):
worker_id = str(worker.get("id") or "unknown")
check = "capability"
try:
async with semaphore:
cap_url = build_worker_url(worker, "/distributed/image-list/capabilities")
async with session.get(cap_url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
if response.status != 200:
raise RuntimeError(
f"Image-list worker {worker_id} does not expose the capability endpoint "
f"(HTTP {response.status}); upgrade that worker before running this feature"
)
capability = await response.json()
if not isinstance(capability, dict):
raise RuntimeError(f"Image-list worker {worker_id} returned malformed capability data")
supported = capability.get("supported")
protocol_version = capability.get("protocol_version")
nodes = capability.get("node_ids")
if (
not isinstance(supported, bool)
or type(protocol_version) is not int
or not isinstance(nodes, list)
or any(not isinstance(node_id, str) for node_id in nodes)
):
raise RuntimeError(f"Image-list worker {worker_id} returned malformed capability data")
if (
supported is not True
or protocol_version != IMAGE_LIST_PROTOCOL_VERSION
or not required_nodes.issubset(nodes)
):
raise RuntimeError(
f"Image-list worker {worker_id} is incompatible: requires protocol "
f"v{IMAGE_LIST_PROTOCOL_VERSION} and nodes "
f"{sorted(required_nodes)}; no work was submitted"
)
if capability.get("execution_targets") is not True:
raise RuntimeError(
f"Image-list worker {worker_id} lacks explicit execution_targets support; "
"upgrade that worker before running this feature; no work was submitted"
)
if required_formats:
advertised = capability.get("image_transfer_formats")
if (not isinstance(advertised, list)
or any(type(fmt) is not str for fmt in advertised)
or not required_formats.issubset(advertised)):
raise RuntimeError(
f"Image-list worker {worker_id} lacks required image_transfer_formats "
f"{sorted(required_formats)}; upgrade that worker before compressed IMAGE returns; "
"no work was submitted"
)
if require_value_results and capability.get("value_results") is not True:
raise RuntimeError(
f"Image-list worker {worker_id} lacks explicit value_results support; "
"upgrade that worker before using End overrides; no work was submitted"
)
check = "queue"
queue_url = build_worker_url(worker, "/queue")
async with session.get(queue_url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
if response.status != 200:
raise RuntimeError(f"Image-list worker {worker_id} queue preflight returned HTTP {response.status}")
queue = await response.json()
if (
not isinstance(queue, dict)
or not isinstance(queue.get("queue_running"), list)
or not isinstance(queue.get("queue_pending"), list)
):
raise RuntimeError(f"Image-list worker {worker_id} returned malformed /queue state")
idle = not queue["queue_running"] and not queue["queue_pending"]
return worker_id, idle
except (asyncio.TimeoutError, aiohttp.ClientConnectionError) as exc:
log(
f"[Distributed] Image-list worker {worker_id} unavailable during {check} check; "
f"skipping: {exc}"
)
return worker_id, None
statuses = await asyncio.gather(*(inspect(worker) for worker in workers))
idle_workers = {
worker_id: True
for worker_id, idle in statuses
if idle is True
}
if require_idle and not idle_workers:
raise RuntimeError("No selected feature-compatible direct worker is idle; unrelated worker prompts were left untouched")
return idle_workers
async def select_active_workers(
@@ -203,6 +361,10 @@ def _extract_queue_remaining(payload):
async def _probe_worker_queue(worker, semaphore, probe_timeout):
async with semaphore:
if worker.get("type") == "serverless":
from .serverless_runtime import get_manager
status = await get_manager().health(worker)
return {"worker": worker, "queue_remaining": status["queueRemaining"]}
worker_url = build_worker_url(worker)
payload = await probe_worker(worker_url, timeout=probe_timeout)
if payload is None:
+35
View File
@@ -29,6 +29,11 @@ def _normalize_media_reference(value):
return None
cleaned = re.sub(r"\s*\[\w+\]$", "", value).strip().replace("\\", "/")
if MEDIA_FILE_RE.search(cleaned):
# Output/temp annotations select a different master-side media root.
# Preserve them through discovery/loading and as distinct rewrite keys.
annotation = re.search(r"\s*\[(output|temp)\]$", value)
if annotation:
return f"{cleaned} [{annotation.group(1)}]"
return cleaned
return None
@@ -143,10 +148,40 @@ async def fetch_worker_path_separator(worker, trace_execution_id=None):
return None
async def fetch_worker_has_node(worker, node_class, trace_execution_id=None):
"""Best-effort check that a worker has ``node_class`` registered.
Uses ComfyUI's ``GET /object_info/{node_class}``, which returns ``{}`` for
unknown classes. Any failure counts as "not available" so callers fall
back to nodes every ComfyUI install has.
"""
url = build_worker_url(worker, f"/object_info/{node_class}")
session = await get_client_session()
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
if resp.status != 200:
return False
payload = await resp.json()
return isinstance(payload, dict) and node_class in payload
except Exception as exc:
message = f"Failed to probe node '{node_class}' on worker {worker.get('id')}: {exc}"
if trace_execution_id:
trace_debug(trace_execution_id, message)
else:
debug_log(f"[Distributed] {message}")
return False
async def _upload_media_to_worker(worker, filename, file_bytes, file_hash, mime_type, trace_execution_id=None):
"""Upload one media file to worker iff missing or hash-mismatched."""
session = await get_client_session()
normalized = filename.replace("\\", "/")
annotation = re.search(r"\s*\[(output|temp)\]$", normalized)
if annotation:
# Worker uploads live under input/. Do not collide across master roots,
# or overwrite a different reference with the same basename.
basename = normalized[:annotation.start()].rstrip().rsplit("/", 1)[-1]
normalized = f"distributed-media/{annotation.group(1)}/{file_hash}/{basename}"
check_url = build_worker_url(worker, "/distributed/check_file")
try:
+93 -5
View File
@@ -328,8 +328,19 @@ def _find_delegate_master_retained_upstream_nodes(prompt_obj, start_ids):
return connected
def prune_prompt_for_worker(prompt_obj):
"""Prune worker prompt to distributed nodes and their upstream dependencies."""
WORKER_SINK_CLASS = "DistributedWorkerSink"
def prune_prompt_for_worker(prompt_obj, use_worker_sink=False):
"""Prune worker prompt to distributed nodes and their upstream dependencies.
Removing downstream nodes can leave the worker prompt without an output
node. One is re-added after each such distributed node: the hidden
``DistributedWorkerSink`` when ``use_worker_sink`` is set (the worker must
have it registered), otherwise PreviewImage/PreviewAudio for workers
running older versions. The preview nodes write every frame to the
worker's temp directory, which the sink avoids.
"""
collector_ids = find_nodes_by_class(prompt_obj, "DistributedCollector")
upscale_ids = find_nodes_by_class(prompt_obj, "UltimateSDUpscaleDistributed")
distributed_ids = collector_ids + upscale_ids
@@ -364,7 +375,13 @@ def prune_prompt_for_worker(prompt_obj):
and len(inputs["audio"]) == 2
)
if image_connected:
if use_worker_sink and (image_connected or audio_connected):
pruned_prompt[next_id()] = {
"inputs": {"value": [dist_id, 0 if image_connected else 1]},
"class_type": WORKER_SINK_CLASS,
"_meta": {"title": "Distributed Worker Sink (auto-added)"},
}
elif image_connected:
preview_id = next_id()
pruned_prompt[preview_id] = {
"inputs": {"images": [dist_id, 0]},
@@ -441,11 +458,55 @@ def prepare_delegate_master_prompt(prompt_obj, collector_ids):
return pruned_prompt
def _resolve_direct_workers_for_image_list(config, requested_ids):
"""Resolve enabled workers by default, or exact explicitly selected direct workers."""
requested = None if requested_ids is None else {str(worker_id) for worker_id in requested_ids}
configured_workers = (config or {}).get("workers", [])
configured_ids = {
str(worker.get("id") or "").strip()
for worker in configured_workers
if str(worker.get("id") or "").strip()
}
if requested is not None:
unknown_ids = requested - configured_ids
if unknown_ids:
raise RuntimeError(
"Unknown explicitly selected image-list worker ID(s): "
+ ", ".join(sorted(unknown_ids))
)
direct = []
for worker in configured_workers:
worker_id = str(worker.get("id") or "").strip()
if not worker_id:
continue
if requested is None:
if not worker.get("enabled", False):
continue
elif worker_id not in requested:
continue
if str(worker.get("type") or "local").strip().lower() == "serverless":
continue
try:
port = int(worker.get("port", worker.get("listen_port", 8188)) or 8188)
except (TypeError, ValueError) as exc:
raise RuntimeError(f"Invalid port for image-list worker {worker_id}") from exc
direct.append({
"id": worker_id,
"name": str(worker.get("name") or worker_id),
"host": worker.get("host"),
"port": port,
"type": worker.get("type", "local"),
})
return direct
def generate_job_id_map(prompt_index, prefix):
"""Create stable per-node job IDs for distributed nodes."""
job_map = {}
distributed_nodes = prompt_index.nodes_for_class("DistributedCollector") + prompt_index.nodes_for_class(
"UltimateSDUpscaleDistributed"
distributed_nodes = (
prompt_index.nodes_for_class("DistributedCollector")
+ prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")
+ prompt_index.nodes_for_class("DistributedImageListStart")
)
for node_id in distributed_nodes:
job_map[node_id] = f"{prefix}_{node_id}"
@@ -540,6 +601,22 @@ def _override_value_nodes(prompt_copy, prompt_index, is_master, participant_id,
inputs["worker_id"] = f"worker_{worker_index_map.get(participant_id, 0)}"
def _override_image_list_nodes(prompt_copy, prompt_index, participant_id, worker_ids,
job_id_map, master_url, delegate_master, workflow_meta=None):
"""Attach protocol context to Start without changing its visible V3 sockets."""
for node_id in prompt_index.nodes_for_class("DistributedImageListStart"):
node = prompt_copy.get(node_id)
if not isinstance(node, dict):
continue
inputs = node.setdefault("inputs", {})
inputs["image_list_enabled_worker_ids"] = json.dumps(list(worker_ids))
inputs["image_list_master_participates"] = participant_id == "master" and not delegate_master
inputs["image_list_master_callback"] = str(master_url or "")
inputs["image_list_execution_id"] = job_id_map.get(node_id, node_id)
if workflow_meta is not None:
inputs["image_list_workflow_meta"] = workflow_meta
def apply_participant_overrides(
prompt_copy,
participant_id,
@@ -548,6 +625,7 @@ def apply_participant_overrides(
master_url,
delegate_master,
prompt_index,
workflow_meta=None,
):
"""Return a prompt copy with hidden inputs configured for master/worker."""
is_master = participant_id == "master"
@@ -575,5 +653,15 @@ def apply_participant_overrides(
master_url,
enabled_json,
)
_override_image_list_nodes(
prompt_copy,
prompt_index,
participant_id,
enabled_worker_ids,
job_id_map,
master_url,
delegate_master,
workflow_meta=workflow_meta,
)
return prompt_copy
+127
View File
@@ -0,0 +1,127 @@
"""Small Runpod queue API client. Never retries a potentially billable POST."""
import asyncio
import json
import re
import aiohttp
API_BASE = "https://api.runpod.ai/v2"
MAX_REQUEST_BYTES = 10_000_000
_ID = re.compile(r"[A-Za-z0-9_-]{1,200}\Z")
class RunpodError(RuntimeError):
"""Sanitized provider failure safe to display in the panel."""
class AmbiguousSubmission(RunpodError):
"""Submission may have been accepted: never retry it automatically."""
def serialize_request(workflow, images, timeout_seconds, *, media=None):
"""Serialize and size-check the exact JSON body used by Runpod ``/run``."""
payload = {
"input": {"workflow": workflow, "images": images},
"policy": {
"executionTimeout": timeout_seconds * 1000,
"ttl": timeout_seconds * 1000,
},
}
if media is not None:
from ...serverless_bridge.handler import _validate_request
payload["input"]["media"] = media
_validate_request(payload["input"])
body = json.dumps(
payload,
allow_nan=False,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
size = len(body)
if size > MAX_REQUEST_BYTES:
raise ValueError(
f"Serialized serverless request body is {size:,} bytes; Runpod allows at most "
f"{MAX_REQUEST_BYTES:,} bytes. Base64-encoded images add about 33% over raw "
"image bytes; reduce image or workflow size."
)
return body
def checked_id(value):
if not isinstance(value, str) or not _ID.fullmatch(value):
raise ValueError("Invalid Runpod endpoint or job ID")
return value
class RunpodClient:
def __init__(self, session, endpoint_id, api_key):
self.session = session
self.endpoint_id = checked_id(endpoint_id)
if not isinstance(api_key, str) or not api_key or any(c in api_key for c in "\r\n"):
raise ValueError("A valid Runpod API key is required")
self._api_key = api_key
async def _request(self, method, path, payload=None, request_body=None):
# Base origin is code-owned, never a user-supplied URL. Do not follow redirects
# with credentials, or include response bodies/transport errors in exceptions.
url = f"{API_BASE}/{self.endpoint_id}/{path}"
attempts = 3 if method == "GET" else 1
for attempt in range(attempts):
try:
request_kwargs = {
"headers": {"Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json"},
"timeout": aiohttp.ClientTimeout(total=30),
"allow_redirects": False,
}
if request_body is None:
request_kwargs["json"] = payload
else:
request_kwargs["data"] = request_body
async with self.session.request(
method, url,
**request_kwargs,
) as response:
status = response.status
if status != 200:
if method == "GET" and (status == 429 or status >= 500) and attempt + 1 < attempts:
await asyncio.sleep(0.5 * (2 ** attempt))
continue
if method == "POST" and path == "run" and status >= 500:
raise AmbiguousSubmission("Runpod submission returned a server error; acceptance is unknown. Check the provider before retrying.")
raise RunpodError(f"Runpod {path.split('/')[0]} request failed (HTTP {status})")
try:
result = await response.json()
except (ValueError, aiohttp.ContentTypeError):
if method == "POST" and path == "run":
raise AmbiguousSubmission("Runpod submission returned an unreadable response; check the provider before retrying.") from None
raise RunpodError("Runpod returned an invalid JSON response") from None
if not isinstance(result, dict):
if method == "POST" and path == "run":
raise AmbiguousSubmission("Runpod submission did not return a job record; check the provider before retrying.")
raise RunpodError("Runpod returned an invalid response")
return result
except (asyncio.TimeoutError, aiohttp.ClientError, OSError):
if method == "POST" and path == "run":
raise AmbiguousSubmission("Runpod submission connection was interrupted; acceptance is unknown. Check the provider before retrying.") from None
if method == "GET" and attempt + 1 < attempts:
await asyncio.sleep(0.5 * (2 ** attempt))
continue
raise RunpodError(f"Runpod {path.split('/')[0]} connection failed") from None
raise RunpodError("Runpod request failed")
async def health(self):
return await self._request("GET", "health")
async def submit(self, workflow, images, timeout_seconds, *, media=None):
body = serialize_request(workflow, images, timeout_seconds, media=media)
result = await self._request("POST", "run", request_body=body)
try:
return checked_id(result.get("id"))
except ValueError:
raise AmbiguousSubmission("Runpod submission returned no valid job ID; check the provider before retrying.") from None
async def status(self, job_id):
return await self._request("GET", f"status/{checked_id(job_id)}")
async def cancel(self, job_id):
return await self._request("POST", f"cancel/{checked_id(job_id)}")
+57
View File
@@ -0,0 +1,57 @@
"""Private, atomic job-ID journal. Contains identifiers/status only, never payloads."""
import json
import os
from pathlib import Path
import tempfile
import threading
from .serverless_state import TERMINAL
FIELDS = {"record_id", "execution_id", "worker_id", "participant_id", "endpoint_id", "job_id", "status", "created_at", "updated_at", "error", "prompt_id"}
class JobLedger:
def __init__(self, path):
self.path = Path(path)
self._lock = threading.RLock()
def _read(self):
if self.path.is_symlink():
raise RuntimeError("Unsafe serverless job journal path")
if not self.path.exists():
return {}
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
if not isinstance(data, dict) or any(not isinstance(v, dict) or set(v) - FIELDS for v in data.values()):
raise ValueError()
return data
except (ValueError, OSError):
raise RuntimeError("Serverless job journal is unreadable; preserve it for recovery before submitting more work") from None
def all(self):
with self._lock:
return list(self._read().values())
def update(self, record_id, **fields):
if set(fields) - FIELDS:
raise ValueError("Unexpected serverless journal field")
with self._lock:
data = self._read()
data.setdefault(record_id, {"record_id": record_id}).update(fields)
# Keep every unresolved job, plus a bounded terminal history.
completed = sorted((k for k, v in data.items() if v.get("status") in TERMINAL), key=lambda k: data[k].get("updated_at", 0))
for key in completed[:-200]:
data.pop(key)
if self.path.parent.is_symlink():
raise RuntimeError("Unsafe serverless journal directory")
self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=".jobs-", dir=self.path.parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as stream:
json.dump(data, stream, allow_nan=False)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, self.path)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
return dict(data[record_id])
+601
View File
@@ -0,0 +1,601 @@
"""Bounded image transport for generic Runpod-backed ComfyUI jobs."""
import asyncio
import base64
import binascii
import hashlib
import io
import ipaddress
import os
import re
import socket
import stat
import warnings
from pathlib import Path, PurePosixPath
from urllib.parse import urljoin, urlsplit
import aiohttp
import numpy as np
from PIL import Image, UnidentifiedImageError
MAX_INPUT_IMAGE_BYTES = 25 * 1024 * 1024
MAX_INPUT_TOTAL_BYTES = 10_000_000
MAX_OUTPUT_FILES = 1024
MAX_OUTPUT_IMAGE_BYTES = 128 * 1024 * 1024
MAX_OUTPUT_TOTAL_BYTES = 256 * 1024 * 1024
MAX_IMAGE_PIXELS = 40_000_000
MAX_OUTPUT_ARRAY_BYTES = 256 * 1024 * 1024
MAX_REDIRECTS = 3
_DOWNLOAD_TIMEOUT_SECONDS = 60
_CHUNK_SIZE = 64 * 1024
_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"}
_UNSUPPORTED_IMAGE_EXTENSIONS = {".gif", ".tif", ".tiff", ".avif", ".heic", ".ico", ".svg"}
_AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".m4a", ".aac", ".ogg", ".opus", ".aiff", ".aif", ".wma"}
_VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm"}
_MEDIA_EXTENSIONS = _IMAGE_EXTENSIONS | _UNSUPPORTED_IMAGE_EXTENSIONS | _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS
_MEDIA_ANNOTATION = re.compile(r"\s+\[([A-Za-z0-9_-]{1,64})\]$")
_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,199}\Z", re.ASCII)
_SAFE_FILENAME_RE = re.compile(r"[A-Za-z0-9_.-]{1,255}\Z", re.ASCII)
class ServerlessMediaError(ValueError):
"""Sanitized input, artifact, or download validation failure."""
def _folder_roots(folder_paths):
roots = {}
for kind, getter_name in (
("input", "get_input_directory"),
("output", "get_output_directory"),
("temp", "get_temp_directory"),
):
getter = getattr(folder_paths, getter_name, None)
if not callable(getter):
raise ServerlessMediaError("ComfyUI media directories are unavailable")
try:
roots[kind] = Path(getter()).resolve(strict=True)
except Exception:
raise ServerlessMediaError("ComfyUI media directories are unavailable") from None
return roots
def _inside_roots(path, roots):
for root in roots.values():
try:
path.relative_to(root)
return True
except ValueError:
continue
return False
def _normalize_media_reference(value):
if not isinstance(value, str) or not value or value != value.strip() or "\x00" in value:
raise ServerlessMediaError("Image input reference is invalid")
value = value.replace("\\", "/")
explicit_root = None
annotation = _MEDIA_ANNOTATION.search(value)
if annotation:
marker = annotation.group(1).lower()
if marker not in {"input", "output", "temp"}:
raise ServerlessMediaError("Image input annotation is unsupported")
explicit_root = marker
value = value[: annotation.start()]
if "://" in value:
raise ServerlessMediaError("Remote media references are not supported as image inputs")
if re.match(r"^[A-Za-z]:", value) or value.startswith("//"):
raise ServerlessMediaError("Image input path is outside the allowed ComfyUI media directories")
path = PurePosixPath(value)
if any(part == ".." for part in path.parts):
raise ServerlessMediaError("Image input path traversal is not allowed")
if not path.parts or value in {".", ""}:
raise ServerlessMediaError("Image input reference is invalid")
if explicit_root:
value = f"{value} [{explicit_root}]"
suffix = Path(path.name).suffix.lower()
return value, suffix
def _resolve_input_path(value, suffix, folder_paths, roots):
if suffix not in _IMAGE_EXTENSIONS:
if suffix in _MEDIA_EXTENSIONS:
raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
raise ServerlessMediaError("Unsupported serverless image file type")
try:
source = Path(value)
if source.is_absolute():
annotation = _MEDIA_ANNOTATION.search(value)
unannotated = value[:annotation.start()] if annotation else value
resolved = Path(unannotated).resolve(strict=True)
else:
resolver = getattr(folder_paths, "get_annotated_filepath", None)
if not callable(resolver):
raise ServerlessMediaError("ComfyUI media path resolver is unavailable")
resolved = Path(resolver(value)).resolve(strict=True)
except ServerlessMediaError:
raise
except Exception:
raise ServerlessMediaError("Image input file is missing or invalid") from None
if not _inside_roots(resolved, roots):
raise ServerlessMediaError("Image input path is outside the allowed ComfyUI media directories")
try:
mode = resolved.stat().st_mode
except OSError:
raise ServerlessMediaError("Image input file is missing or invalid") from None
if not stat.S_ISREG(mode):
raise ServerlessMediaError("Image input is not a regular file")
return resolved
def _iter_image_references(workflow):
if not isinstance(workflow, dict):
raise ServerlessMediaError("Serverless workflow must be an API graph")
references = []
for node_id, node in workflow.items():
if not isinstance(node, dict):
continue
class_type = node.get("class_type")
inputs = node.get("inputs", {})
if class_type == "LoadImage" and not isinstance(inputs, dict):
raise ServerlessMediaError("LoadImage requires a literal image filename; linked image inputs are unsupported")
if not isinstance(inputs, dict):
continue
if class_type == "LoadImage":
value = inputs.get("image")
if not isinstance(value, str):
raise ServerlessMediaError("LoadImage requires a literal image filename; linked image inputs are unsupported")
references.append((node_id, inputs, "image", value, True))
for key in ("image", "file"):
if key == "image" and class_type == "LoadImage":
continue
value = inputs.get(key)
if not isinstance(value, str):
continue
normalized, suffix = _normalize_media_reference(value)
if suffix in _IMAGE_EXTENSIONS:
references.append((node_id, inputs, key, normalized, False))
elif suffix in _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS:
raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
elif suffix in _UNSUPPORTED_IMAGE_EXTENSIONS or (key == "image" and suffix):
raise ServerlessMediaError("Unsupported serverless image file type")
for key in ("audio", "video"):
value = inputs.get(key)
if not isinstance(value, str):
continue
_normalized, suffix = _normalize_media_reference(value)
if suffix in _MEDIA_EXTENSIONS:
raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
return references
def _read_limited(path, expected_size):
try:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
with os.fdopen(descriptor, "rb") as handle:
info = os.fstat(handle.fileno())
if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_INPUT_IMAGE_BYTES:
raise ServerlessMediaError("Image input exceeds the permitted size")
if info.st_size > MAX_INPUT_TOTAL_BYTES:
raise ServerlessMediaError(
f"Total raw serverless image input size exceeds {MAX_INPUT_TOTAL_BYTES:,} bytes; "
"base64 encoding adds about 33%, and the complete request body is also limited to 10,000,000 bytes"
)
data = handle.read(MAX_INPUT_TOTAL_BYTES + 1)
except ServerlessMediaError:
raise
except OSError:
raise ServerlessMediaError("Image input file could not be read safely") from None
if len(data) > MAX_INPUT_TOTAL_BYTES:
raise ServerlessMediaError(
f"Total raw serverless image input size exceeds {MAX_INPUT_TOTAL_BYTES:,} bytes; "
"base64 encoding adds about 33%, and the complete request body is also limited to 10,000,000 bytes"
)
if len(data) != expected_size:
raise ServerlessMediaError("Image input changed while being read")
return data
async def prepare_images(workflow: dict) -> list[dict]:
"""Pack referenced local images and rewrite only their graph input references.
``workflow`` is expected to be a private API-graph copy owned by the caller.
Returned ``name`` values are flat generated names suitable for ``input.images``.
"""
references = _iter_image_references(workflow)
if not references:
return []
# Import only in ComfyUI, keeping this module importable in the serverless client.
try:
import folder_paths
except Exception:
raise ServerlessMediaError("ComfyUI media directories are unavailable") from None
roots = _folder_roots(folder_paths)
resolved_refs = []
unique_paths = {}
total_source_bytes = 0
for node_id, inputs, key, value, _is_load_image in references:
normalized, suffix = _normalize_media_reference(value)
if key == "image" and suffix not in _IMAGE_EXTENSIONS:
if suffix in _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS:
raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
raise ServerlessMediaError("Unsupported serverless image file type")
path = _resolve_input_path(normalized, suffix, folder_paths, roots)
if path not in unique_paths:
try:
size = path.stat().st_size
except OSError:
raise ServerlessMediaError("Image input file is missing or invalid") from None
if size < 0 or size > MAX_INPUT_IMAGE_BYTES:
raise ServerlessMediaError("Image input exceeds the permitted size")
total_source_bytes += size
if total_source_bytes > MAX_INPUT_TOTAL_BYTES:
raise ServerlessMediaError(
f"Total raw serverless image input size is {total_source_bytes:,} bytes; "
f"the input read guard is {MAX_INPUT_TOTAL_BYTES:,} bytes. Base64 encoding "
"adds about 33%, and the complete request body is also limited to 10,000,000 bytes"
)
unique_paths[path] = {"size": size, "suffix": suffix}
resolved_refs.append((inputs, key, path))
packed_by_name = {}
path_to_name = {}
for path, metadata in unique_paths.items():
data = _read_limited(path, metadata["size"])
digest = hashlib.sha256(data).hexdigest()
name = f"serverless_{digest}{metadata['suffix']}"
path_to_name[path] = name
packed_by_name.setdefault(name, {"name": name, "image": base64.b64encode(data).decode("ascii")})
for inputs, key, path in resolved_refs:
inputs[key] = path_to_name[path]
return [packed_by_name[name] for name in sorted(packed_by_name)]
def _validate_artifact_url(url):
if not isinstance(url, str) or not url or any(ord(char) < 0x20 or char.isspace() for char in url):
raise ServerlessMediaError("Serverless image artifact URL is invalid")
try:
parsed = urlsplit(url)
port = parsed.port
except ValueError:
raise ServerlessMediaError("Serverless image artifact URL is invalid") from None
if (
parsed.scheme.lower() != "https"
or not parsed.hostname
or parsed.username is not None
or parsed.password is not None
or port not in (None, 443)
or parsed.fragment
):
raise ServerlessMediaError("Serverless image artifact URL must be a public HTTPS URL")
host = parsed.hostname.rstrip(".").lower()
if host in {"localhost", "metadata", "metadata.google.internal"} or host.endswith((".localhost", ".local", ".internal")):
raise ServerlessMediaError("Serverless image artifact URL must be a public HTTPS URL")
try:
address = ipaddress.ip_address(host)
except ValueError:
return url
if not address.is_global:
raise ServerlessMediaError("Serverless image artifact URL must resolve to a public address")
return url
class _PublicResolver(aiohttp.abc.AbstractResolver):
"""Resolve once, reject non-public answers, and pin the checked address."""
async def _getaddrinfo(self, host, port, family):
return await asyncio.get_running_loop().getaddrinfo(
host, port, family=family, type=socket.SOCK_STREAM
)
async def resolve(self, host, port=0, family=socket.AF_INET):
try:
results = await self._getaddrinfo(host, port, family)
except (OSError, asyncio.TimeoutError):
raise OSError("Public artifact host could not be resolved") from None
if not results:
raise OSError("Public artifact host could not be resolved")
resolved = []
seen = set()
for result_family, _socktype, proto, _canonname, sockaddr in results:
try:
address = ipaddress.ip_address(sockaddr[0])
except (ValueError, TypeError, IndexError):
raise OSError("Public artifact host resolved to an invalid address") from None
if not address.is_global:
raise OSError("Public artifact host resolved to a non-public address")
key = (str(address), result_family)
if key in seen:
continue
seen.add(key)
resolved.append({
"hostname": host,
"host": str(address),
"port": port,
"family": result_family,
"proto": proto,
"flags": 0,
})
if not resolved:
raise OSError("Public artifact host could not be resolved")
return resolved
async def close(self):
return None
async def download_public_artifact(url: str, max_bytes: int = MAX_OUTPUT_IMAGE_BYTES) -> bytes:
"""Download a signed public HTTPS artifact without provider credentials."""
if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or not 0 < max_bytes <= MAX_OUTPUT_IMAGE_BYTES:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
current_url = _validate_artifact_url(url)
resolver = _PublicResolver()
connector = aiohttp.TCPConnector(resolver=resolver, use_dns_cache=False, limit=4)
timeout = aiohttp.ClientTimeout(
total=_DOWNLOAD_TIMEOUT_SECONDS,
connect=10,
sock_read=15,
)
try:
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout,
trust_env=False,
cookie_jar=aiohttp.DummyCookieJar(),
) as session:
for redirect_count in range(MAX_REDIRECTS + 1):
_validate_artifact_url(current_url)
async with session.get(current_url, allow_redirects=False) as response:
if response.status in {301, 302, 303, 307, 308}:
location = response.headers.get("Location")
if not location or redirect_count >= MAX_REDIRECTS:
raise ServerlessMediaError("Serverless image artifact redirect is invalid")
current_url = _validate_artifact_url(urljoin(current_url, location))
continue
if response.status != 200:
raise ServerlessMediaError("Serverless image artifact download failed")
payload = bytearray()
async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
if not isinstance(chunk, bytes):
raise ServerlessMediaError("Serverless image artifact response is invalid")
if len(payload) + len(chunk) > max_bytes:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
payload.extend(chunk)
if not payload:
raise ServerlessMediaError("Serverless image artifact is empty")
return bytes(payload)
except ServerlessMediaError:
raise
except Exception:
raise ServerlessMediaError("Serverless image artifact download failed") from None
raise ServerlessMediaError("Serverless image artifact download failed")
def _validated_manifest(manifest):
if not isinstance(manifest, list) or len(manifest) > MAX_OUTPUT_FILES:
raise ServerlessMediaError("Serverless image manifest is invalid")
normalized = []
seen_prefixes = set()
prefixes = []
for item in manifest:
if not isinstance(item, dict):
raise ServerlessMediaError("Serverless image manifest is invalid")
if item.get("kind") != "image":
raise ServerlessMediaError("Only image collectors are supported by serverless media transport")
for key in ("collector_id", "job_id", "node_id"):
value = item.get(key)
if not isinstance(value, (str, int)) or isinstance(value, bool) or not str(value):
raise ServerlessMediaError("Serverless image manifest is invalid")
prefix = item.get("prefix")
if not isinstance(prefix, str) or not _PREFIX_RE.fullmatch(prefix) or prefix in {".", ".."}:
raise ServerlessMediaError("Serverless image manifest contains an unsafe filename prefix")
folded = prefix.casefold()
if folded in seen_prefixes:
raise ServerlessMediaError("Serverless image manifest prefixes must be unique")
seen_prefixes.add(folded)
prefixes.append(prefix)
normalized.append({
"collector_id": str(item["collector_id"]),
"job_id": str(item["job_id"]),
"node_id": str(item["node_id"]),
"prefix": prefix,
})
for first in prefixes:
for second in prefixes:
if first != second and second.casefold().startswith(first.casefold() + "_"):
raise ServerlessMediaError("Serverless image manifest prefixes are ambiguous")
return normalized
def _artifact_filename_matches(filename, prefixes):
if not isinstance(filename, str):
return None
basename = filename.replace("\\", "/").rsplit("/", 1)[-1]
matches = [item for item in prefixes if basename.startswith(item["prefix"] + "_")]
if len(matches) > 1:
raise ServerlessMediaError("Serverless image artifact matches multiple collectors")
return matches[0] if matches else None
def _canonical_artifact_identity(entry, filename):
artifact_path = entry.get("artifact_path")
artifact_type = entry.get("artifact_type", "output")
if artifact_path is not None:
if not isinstance(artifact_path, str) or not artifact_path or "\x00" in artifact_path:
raise ServerlessMediaError("Serverless image artifact identity is invalid")
normalized_path = os.path.normcase(os.path.normpath(artifact_path.replace("\\", "/")))
else:
subfolder = entry.get("subfolder", "")
if not isinstance(subfolder, str) or "\x00" in subfolder:
raise ServerlessMediaError("Serverless image artifact identity is invalid")
normalized_path = os.path.normcase(os.path.normpath(f"{subfolder}/{filename}"))
return str(artifact_type), normalized_path
def _decode_base64_data(value, max_bytes=MAX_OUTPUT_IMAGE_BYTES):
if not isinstance(value, str) or not value:
raise ServerlessMediaError("Serverless image artifact data is invalid")
max_bytes = min(max_bytes, MAX_OUTPUT_IMAGE_BYTES)
max_encoded = ((max_bytes + 2) // 3) * 4
if len(value) > max_encoded:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
padding = len(value) - len(value.rstrip("="))
estimated_size = (len(value) // 4) * 3 - padding
if estimated_size > max_bytes:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
try:
payload = base64.b64decode(value, validate=True)
except (binascii.Error, ValueError):
raise ServerlessMediaError("Serverless image artifact data is invalid") from None
if not payload:
raise ServerlessMediaError("Serverless image artifact is empty")
if len(payload) > max_bytes:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
return payload
def _decode_image_array(payload, total_array_bytes):
try:
with warnings.catch_warnings():
warnings.simplefilter("error", Image.DecompressionBombWarning)
image = Image.open(io.BytesIO(payload))
width, height = image.size
if width <= 0 or height <= 0 or width * height > MAX_IMAGE_PIXELS:
raise ServerlessMediaError("Serverless image dimensions exceed the permitted limit")
has_alpha = image.mode in {"RGBA", "LA"} or "transparency" in image.info
mode = "RGBA" if has_alpha else "RGB"
output_bytes = width * height * (4 if has_alpha else 3) * 4
if total_array_bytes + output_bytes > MAX_OUTPUT_ARRAY_BYTES:
raise ServerlessMediaError("Decoded serverless images exceed the permitted memory limit")
converted = image.convert(mode)
converted.load()
array = np.array(converted, dtype=np.float32, copy=True)
np.multiply(array, 1.0 / 255.0, out=array)
return array, output_bytes
except ServerlessMediaError:
raise
except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError, Image.DecompressionBombWarning):
raise ServerlessMediaError("Serverless image artifact could not be decoded") from None
except Exception:
raise ServerlessMediaError("Serverless image artifact could not be decoded") from None
def _image_output(output):
"""Unwrap handler output and reject explicit execution errors."""
if not isinstance(output, dict):
raise ServerlessMediaError("Serverless job output is invalid")
nested = output.get("output")
data = nested if isinstance(nested, dict) else output
for candidate in (output, data):
if candidate.get("error") or candidate.get("errors"):
raise ServerlessMediaError("Serverless job reported an execution error")
return data
async def _select_image_payloads(data, expected, download):
"""Match artifact identities and fetch bounded bytes, without image decoding."""
artifacts = data.get("images", [])
if not isinstance(artifacts, list) or len(artifacts) > MAX_OUTPUT_FILES:
raise ServerlessMediaError("Serverless job returned too many image files")
selected = {item["prefix"]: [] for item in expected}
identities = set()
total_payload_bytes = 0
for entry in artifacts:
if not isinstance(entry, dict):
continue
filename = entry.get("filename")
manifest_entry = _artifact_filename_matches(filename, expected)
if manifest_entry is None:
continue
normalized_filename = filename.replace("\\", "/")
if "/" in normalized_filename or not _SAFE_FILENAME_RE.fullmatch(filename):
raise ServerlessMediaError("Serverless image artifact filename is invalid")
if Path(filename).suffix.lower() not in _IMAGE_EXTENSIONS:
raise ServerlessMediaError("Serverless image artifact has an unsupported file type")
if "artifact_type" in entry and entry["artifact_type"] != "output":
raise ServerlessMediaError("Expected serverless image artifact was not an output image")
identity = _canonical_artifact_identity(entry, filename)
if identity in identities:
raise ServerlessMediaError("Duplicate serverless image artifact identity")
identities.add(identity)
match = re.match(rf"^{re.escape(manifest_entry['prefix'])}_(\d+)(?:_|\.)", filename)
if match is None:
raise ServerlessMediaError("Expected serverless image filename counter is invalid")
artifact_type = entry.get("type")
remaining_bytes = MAX_OUTPUT_TOTAL_BYTES - total_payload_bytes
if remaining_bytes <= 0:
raise ServerlessMediaError("Serverless image artifacts exceed the permitted total size")
if artifact_type == "base64":
payload = _decode_base64_data(entry.get("data"), remaining_bytes)
elif artifact_type in {"gcs_url", "s3_url", "url"}:
url = entry.get("data") or entry.get("url")
url = _validate_artifact_url(url)
try:
if download is None:
payload = await download_public_artifact(
url, max_bytes=min(MAX_OUTPUT_IMAGE_BYTES, remaining_bytes)
)
else:
payload = await download(url)
except ServerlessMediaError:
raise
except Exception:
raise ServerlessMediaError("Serverless image artifact download failed") from None
if not isinstance(payload, bytes) or not payload:
raise ServerlessMediaError("Serverless image artifact response is invalid")
if len(payload) > MAX_OUTPUT_IMAGE_BYTES:
raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
else:
raise ServerlessMediaError("Serverless image artifact transport type is unsupported")
total_payload_bytes += len(payload)
if total_payload_bytes > MAX_OUTPUT_TOTAL_BYTES:
raise ServerlessMediaError("Serverless image artifacts exceed the permitted total size")
selected[manifest_entry["prefix"]].append((int(match.group(1)), filename, identity, payload))
if any(not selected[item["prefix"]] for item in expected):
raise ServerlessMediaError("Serverless job is missing an expected collector image")
return selected
def _decode_image_groups(selected, expected):
"""Decode in stable counter order, bounding array memory across collectors."""
result = {}
total_array_bytes = 0
for manifest_entry in expected:
prefix = manifest_entry["prefix"]
rows = sorted(selected[prefix], key=lambda row: (row[0], row[1], row[2]))
arrays = []
shape = None
for _counter, _filename, _identity, payload in rows:
array, added_bytes = _decode_image_array(payload, total_array_bytes)
total_array_bytes += added_bytes
if shape is None:
shape = array.shape
elif array.shape != shape:
raise ServerlessMediaError("Images from one serverless collector have inconsistent dimensions")
arrays.append(array)
result.setdefault(manifest_entry["job_id"], []).extend(arrays)
return result
async def collect_images(output: dict, manifest: list, download=None) -> dict[str, list[np.ndarray]]:
"""Validate and decode expected image artifacts, grouped by serverless job ID."""
data = _image_output(output)
expected = _validated_manifest(manifest)
if not expected:
return {}
selected = await _select_image_payloads(data, expected, download)
return _decode_image_groups(selected, expected)
@@ -0,0 +1,239 @@
"""Pure compilation of prepared Distributed Collectors for comfy-media-v1 pools.
The legacy image compiler remains separate. This compiler emits one MediaExportV1
execution target per connected Collector socket (``images`` and/or ``audio``),
retaining only its upstream computation. It never imports ComfyUI or
inspects/executes runtime node classes.
"""
from __future__ import annotations
import copy
import hashlib
import json
from graphlib import CycleError, TopologicalSorter
from ...serverless_bridge.contract import SOCKET_KINDS, VALUE_SOCKETS, identifier, validate_bindings
from ...serverless_bridge.image_transfer import DEFAULT_FORMAT, validate_options
from .serverless_prompt import (
_PRIMITIVE_NODES,
_REMOVED_SCALAR_NODES,
_is_link,
_is_unsupported_distributed,
_make_scalar_resolver,
_prune_graph,
)
_COLLECTOR = "DistributedCollector"
# Collector inputs, in output-slot order, that a typed pool returns.
COLLECTOR_SOCKETS = ("images", "audio")
# Collector widget defaults (the saved-workflow contract is legacy_png / 90).
COLLECTOR_DEFAULT_FORMAT = "legacy_png"
COLLECTOR_DEFAULT_QUALITY = 90
def collector_image_policy(inputs):
"""Return the endpoint IMAGE (format, quality) for one Collector.
legacy_png is the Collector's saved default and names a JSON/base64 framing
that only applies to direct workers; serverless returns the same 8-bit
pixels as compressed png. exact stays opt-in. No other format is altered.
"""
fmt, quality = validate_options(inputs.get("transfer_format", COLLECTOR_DEFAULT_FORMAT),
inputs.get("transfer_quality", COLLECTOR_DEFAULT_QUALITY))
return ("png" if fmt == "legacy_png" else fmt), quality
def _validate_identifier(value, context):
# Validate the original token before any lookup. In particular, neither
# integer node IDs nor whitespace in prepared identities is normalized.
try:
identifier(value)
except ValueError:
raise ValueError(f"{context} must be a valid nonempty media identifier") from None
def _validate_prompt(prepared_prompt, *, require_collectors=True):
"""Validate the entire source graph, including branches later discarded."""
if not isinstance(prepared_prompt, dict) or not prepared_prompt:
raise ValueError("prepared prompt must be a nonempty API graph mapping")
collectors = []
for node_id, node in prepared_prompt.items():
_validate_identifier(node_id, "prompt node ID")
if not isinstance(node, dict):
raise ValueError(f"prompt node {node_id!r} must be an object")
class_type = node.get("class_type")
if not isinstance(class_type, str) or not class_type.strip():
raise ValueError(f"prompt node {node_id!r} must have a nonempty class_type")
if class_type == _COLLECTOR:
collectors.append(node_id)
elif _is_unsupported_distributed(class_type):
raise ValueError(f"unsupported distributed or batch node: {class_type}")
inputs = node.get("inputs", {})
if not isinstance(inputs, dict):
raise ValueError(f"prompt node {node_id!r} inputs must be an object")
if any(not isinstance(name, str) or not name for name in inputs):
raise ValueError(f"prompt node {node_id!r} input names must be nonempty strings")
if require_collectors and not collectors:
raise ValueError("prepared prompt contains no DistributedCollector nodes")
dependencies = {}
for node_id, node in prepared_prompt.items():
parents = set()
for input_name, value in node.get("inputs", {}).items():
# Ordinary inputs may contain literal lists. As in the legacy
# compiler, exactly two elements denotes an API link, not media.
if not _is_link(value):
continue
source_id, slot = value
context = f"link at node {node_id!r} input {input_name!r}"
_validate_identifier(source_id, context + " source ID")
if source_id not in prepared_prompt:
raise ValueError(f"invalid {context}: source {source_id!r} does not exist")
if type(slot) is not int or slot < 0:
raise ValueError(f"invalid {context}: output slot must be a nonnegative integer")
source_class = prepared_prompt[source_id]["class_type"]
if source_class in _REMOVED_SCALAR_NODES or source_class in _PRIMITIVE_NODES:
if slot != 0:
raise ValueError(f"invalid output slot {slot} for scalar node {source_id!r}")
elif source_class == _COLLECTOR and slot >= len(COLLECTOR_SOCKETS):
raise ValueError(f"invalid output slot {slot} for collector {source_id!r}")
parents.add(source_id)
dependencies[node_id] = parents
for collector_id in collectors:
inputs = prepared_prompt[collector_id].get("inputs", {})
collector_image_policy(inputs)
for field in ("multi_job_id", "worker_id"):
_validate_identifier(inputs.get(field), f"collector {collector_id!r} {field}")
connected = [socket for socket in COLLECTOR_SOCKETS if socket in inputs]
if not connected:
raise ValueError(f"collector {collector_id!r} requires at least one connected media socket")
for socket in connected:
value = inputs[socket]
if not _is_link(value):
raise ValueError(f"collector {collector_id!r} {socket} must be linked; use PrimitiveString for literal text")
if prepared_prompt[value[0]]["class_type"] in _REMOVED_SCALAR_NODES:
# Export sockets must remain actual links after scalar folding.
# A PrimitiveString wrapper supports distributed text values.
raise ValueError(f"collector {collector_id!r} {socket} cannot link directly to a distributed scalar; use a typed node or PrimitiveString")
# Iterative topological sorting checks ordinary and unused cycles too, not
# only the primitive chains that the shared scalar resolver traverses.
try:
ordered = list(TopologicalSorter(dependencies).static_order())
except CycleError:
raise ValueError("cycle in prepared media prompt") from None
collector_set = set(collectors)
has_collector_ancestor = {}
for node_id in ordered:
has_collector_ancestor[node_id] = any(
parent in collector_set or has_collector_ancestor[parent]
for parent in dependencies[node_id]
)
if node_id in collector_set and has_collector_ancestor[node_id]:
raise ValueError("collector-to-collector dependence is unsupported")
return sorted(collectors)
def _build_explicit_exports(source_nodes, specs, request_id):
bindings, exports = [], {}
reserved = set(source_nodes)
for spec in specs:
if not isinstance(spec, dict) or set(spec) != {
"collector_id", "job_id", "participant_id", "socket", "source",
"transfer_format", "transfer_quality"}:
raise ValueError("Invalid explicit media export specification")
collector_id, socket = spec["collector_id"], spec["socket"]
kind = "VALUE" if socket in VALUE_SOCKETS else SOCKET_KINDS.get(socket)
if kind is None:
raise ValueError("Unsupported explicit media export socket")
link = spec["source"]
if (not _is_link(link) or link[0] not in source_nodes
or type(link[1]) is not int or link[1] < 0):
raise ValueError("Explicit media export requires a valid source link")
fmt, quality = validate_options(spec["transfer_format"], spec["transfer_quality"])
if kind != "IMAGE" and fmt != DEFAULT_FORMAT:
raise ValueError("Image transfer controls apply only to IMAGE exports")
# Same stable identities as Collector compilation, including collisions.
identity = json.dumps([collector_id, socket], separators=(",", ":"))
digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
base_id = f"serverless:media:{digest}"
node_id, suffix = base_id, 0
while node_id in reserved:
suffix += 1
node_id = f"{base_id}:{suffix}"
reserved.add(node_id)
binding = {"binding_id": f"b:{digest}", "collector_id": collector_id,
"job_id": spec["job_id"], "participant_id": spec["participant_id"],
"node_id": node_id, "socket": socket, "kind": kind}
bindings.append(binding)
exports[node_id] = {"class_type": "MediaExportV1", "inputs": {
"request_id": request_id, "binding_json": json.dumps(binding),
socket: copy.deepcopy(link)}}
if fmt != DEFAULT_FORMAT:
exports[node_id]["inputs"].update(transfer_format=fmt, transfer_quality=quality)
validate_bindings(bindings)
return bindings, exports
def _build_exports(source_nodes, collector_ids, request_id):
specs = []
for collector_id in collector_ids:
inputs = source_nodes[collector_id].get("inputs", {})
for socket in COLLECTOR_SOCKETS:
if socket not in inputs:
continue
fmt, quality = collector_image_policy(inputs) if socket == "images" else (DEFAULT_FORMAT, 90)
specs.append({"collector_id": collector_id, "job_id": inputs["multi_job_id"],
"participant_id": inputs["worker_id"], "socket": socket,
"source": inputs[socket], "transfer_format": fmt, "transfer_quality": quality})
return _build_explicit_exports(source_nodes, specs, request_id)
def compile_export_prompt(prepared_prompt, request_id, specs):
"""Compile explicit internal exports, without fabricating a Collector node.
Callers supply linked source sockets and literal binding identities; only
exporters and required ancestors survive. This does not add Collector
sockets: its public compiler remains images/audio only.
"""
_validate_identifier(request_id, "request_id")
if _validate_prompt(prepared_prompt, require_collectors=False):
raise ValueError("Explicit exports cannot contain Collector nodes")
bindings, exports = _build_explicit_exports(prepared_prompt, specs, request_id)
compiled = _prune_graph(prepared_prompt, bindings, exports, _make_scalar_resolver(prepared_prompt))
return {node_id: compiled[node_id] for node_id in sorted(compiled)}, bindings
def compile_media_prompt(prepared_prompt: dict, request_id: str) -> tuple[dict, list[dict]]:
"""Return a worker graph and canonical typed-media export bindings.
Each DistributedCollector must have prepared literal ``multi_job_id`` and
``worker_id`` identities and at least one linked ``images``/``audio`` input.
Present media inputs must be API links. Collector transport controls become
the IMAGE export policy (``legacy_png`` is sent as ``png``); other controls
are not exported.
Only MediaExportV1 targets and their ancestors survive. Required ordinary
output/utility nodes are preserved regardless of their runtime OUTPUT_NODE
flag; callers must dispatch exactly the returned binding ``node_id`` values
as partial_execution_targets. Distributed scalar dependencies use the same
resolver as the legacy compiler. Direct scalar-to-media links need a typed
wrapper so that each export input remains a link.
Validation and compilation never modify the caller's source graph. This
structural compiler does not infer socket types or unknown nodes' output
counts; ComfyUI performs runtime type validation at the endpoint.
"""
_validate_identifier(request_id, "request_id")
collector_ids = _validate_prompt(prepared_prompt)
bindings, exports = _build_exports(prepared_prompt, collector_ids, request_id)
compiled = _prune_graph(
prepared_prompt, bindings, exports, _make_scalar_resolver(prepared_prompt)
)
return {node_id: compiled[node_id] for node_id in sorted(compiled)}, bindings
+420
View File
@@ -0,0 +1,420 @@
"""Compile prepared distributed prompts into worker-side ComfyUI image jobs.
This module is deliberately pure: it only copies and rewrites an API prompt
mapping and never imports or executes ComfyUI nodes.
"""
from __future__ import annotations
import copy
import hashlib
import json
import re
_SUPPORTED_DISTRIBUTED = {
"DistributedCollector",
"DistributedSeed",
"DistributedValue",
"DistributedModelName",
}
_REMOVED_SCALAR_NODES = {
"DistributedSeed",
"DistributedValue",
"DistributedModelName",
}
_PRIMITIVE_NODES = {
"PrimitiveBoolean": False,
"PrimitiveFloat": 0.0,
"PrimitiveInt": 0,
"PrimitiveString": "",
}
_REMOVED_NODE_OUTPUTS = {
"DistributedCollector": {0, 1},
"DistributedSeed": {0},
"DistributedValue": {0},
"DistributedModelName": {0},
}
_SEED_DEFAULT = 1125899906842
def _is_link(value):
return isinstance(value, list) and len(value) == 2
def _is_unsupported_distributed(class_type):
if not isinstance(class_type, str) or class_type in _SUPPORTED_DISTRIBUTED:
return False
lowered = class_type.lower()
return "distributed" in lowered or "batchdivider" in lowered
def _safe_component(value):
cleaned = re.sub(r"[^A-Za-z0-9_-]+", "_", str(value)).strip("_-")
return cleaned[:24] or "empty"
def _filename_prefix(job_id, worker_id, collector_id, kind):
identity = json.dumps(
[job_id, worker_id, collector_id, kind],
ensure_ascii=False,
separators=(",", ":"),
)
digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
parts = (
"serverless",
_safe_component(job_id),
_safe_component(worker_id),
_safe_component(collector_id),
kind,
digest,
)
return "_".join(parts)
def _worker_index(worker_id):
"""Mirror nodes.utilities worker-index parsing; None means fallback."""
try:
if worker_id.startswith("worker_"):
return int(worker_id.split("_")[1])
return int(worker_id)
except (ValueError, IndexError):
return None
def _coerce(value, value_type):
if value_type == "INT":
return int(float(value))
if value_type == "FLOAT":
return float(value)
return value
def _coerce_safe(value, value_type):
try:
return _coerce(value, value_type)
except (TypeError, ValueError):
return value
def _stringify_model_name(value):
if isinstance(value, str):
return value
if isinstance(value, (int, float, bool)):
return str(value)
try:
return json.dumps(value, indent=4)
except (TypeError, ValueError):
return str(value)
def _model_name_result(text):
if isinstance(text, list):
values = [_stringify_model_name(value) for value in text]
return values[0] if len(values) == 1 else values
return _stringify_model_name(text)
def _validate_prepared_prompt(prepared_prompt):
"""Validate graph shape, links, collectors, and supported topology."""
if not isinstance(prepared_prompt, dict):
raise ValueError("prepared prompt must be an API graph mapping")
source_nodes = {}
for raw_id, node in prepared_prompt.items():
node_id = str(raw_id)
if node_id in source_nodes:
raise ValueError("prompt has duplicate node IDs after string normalization")
if not isinstance(node, dict):
raise ValueError(f"prompt node {node_id!r} must be an object")
class_type = node.get("class_type")
if not isinstance(class_type, str):
raise ValueError(f"prompt node {node_id!r} is missing a class_type")
if _is_unsupported_distributed(class_type):
raise ValueError(f"unsupported distributed or batch node: {class_type}")
inputs = node.get("inputs", {})
if not isinstance(inputs, dict):
raise ValueError(f"prompt node {node_id!r} inputs must be an object")
source_nodes[node_id] = node
if not source_nodes:
raise ValueError("prepared prompt contains no nodes")
# Validate every API link before any graph rewriting. Slots on the nodes
# removed by this compiler have a known schema and are checked exactly.
for node_id, node in source_nodes.items():
for input_name, value in node.get("inputs", {}).items():
if not _is_link(value):
continue
upstream_id, output_slot = value
upstream_id = str(upstream_id)
if upstream_id not in source_nodes:
raise ValueError(
f"invalid link at node {node_id!r} input {input_name!r}: "
f"source {upstream_id!r} does not exist"
)
if isinstance(output_slot, bool) or not isinstance(output_slot, int) or output_slot < 0:
raise ValueError(
f"invalid link at node {node_id!r} input {input_name!r}: "
"output slot must be a nonnegative integer"
)
upstream_class = source_nodes[upstream_id]["class_type"]
valid_slots = _REMOVED_NODE_OUTPUTS.get(upstream_class)
if valid_slots is not None and output_slot not in valid_slots:
raise ValueError(
f"invalid output slot {output_slot} for removed node "
f"{upstream_class} {upstream_id!r}"
)
collector_ids = sorted(
(node_id for node_id, node in source_nodes.items() if node["class_type"] == "DistributedCollector"),
key=str,
)
if not collector_ids:
raise ValueError("prepared prompt contains no DistributedCollector nodes")
# Only image collectors are supported by the current generic handler.
for collector_id in collector_ids:
inputs = source_nodes[collector_id].get("inputs", {})
if "audio" in inputs:
raise ValueError("collector audio input is unsupported by this image-only compiler")
if not _is_link(inputs.get("images")):
raise ValueError(f"collector {collector_id!r} is missing a linked image input")
source_id = str(inputs["images"][0])
source_class = source_nodes[source_id]["class_type"]
if source_class in _REMOVED_SCALAR_NODES:
raise ValueError(f"collector {collector_id!r} image input must link to an image node")
# A collector cannot be used, directly or through processing, as another
# collector's input: this first compiler handles disconnected jobs only.
for collector_id in collector_ids:
pending = [str(source_nodes[collector_id]["inputs"]["images"][0])]
seen = set()
while pending:
node_id = pending.pop()
if node_id in seen:
continue
seen.add(node_id)
if source_nodes[node_id]["class_type"] == "DistributedCollector":
raise ValueError("collector-to-collector dependence is unsupported")
for value in source_nodes[node_id].get("inputs", {}).values():
if _is_link(value):
pending.append(str(value[0]))
return source_nodes, collector_ids
def _make_scalar_resolver(source_nodes):
"""Build pure resolvers for distributed scalar nodes and stock primitives."""
# Resolve only stock primitive nodes. A visited set makes malformed
# primitive-to-primitive cycles fail clearly instead of recursing forever.
def resolve_primitive(node_id, visiting=None):
node_id = str(node_id)
node = source_nodes[node_id]
class_type = node["class_type"]
if class_type not in _PRIMITIVE_NODES:
raise ValueError("linked value must be literal or supported PrimitiveInt/Float/String/Boolean")
if visiting is None:
visiting = set()
if node_id in visiting:
raise ValueError(f"cycle while resolving primitive value at node {node_id!r}")
visiting.add(node_id)
value = node.get("inputs", {}).get("value", _PRIMITIVE_NODES[class_type])
if _is_link(value):
parent_id, output_slot = str(value[0]), value[1]
if output_slot != 0:
raise ValueError(f"invalid output slot {output_slot} for primitive node {parent_id!r}")
if source_nodes[parent_id]["class_type"] not in _PRIMITIVE_NODES:
raise ValueError("linked primitive value must come from a stock Primitive node")
result = resolve_primitive(parent_id, visiting)
else:
result = copy.deepcopy(value)
visiting.remove(node_id)
return result
def resolve_literal_or_primitive(value, context):
if _is_link(value):
source_id, output_slot = str(value[0]), value[1]
if output_slot != 0 or source_nodes[source_id]["class_type"] not in _PRIMITIVE_NODES:
raise ValueError(f"{context} must be a literal or supported Primitive link")
try:
return resolve_primitive(source_id)
except ValueError as exc:
if "cycle" in str(exc):
raise
raise ValueError(f"{context} must be a literal or supported Primitive link") from None
return copy.deepcopy(value)
def resolve_hidden(node_id, name, default):
value = source_nodes[node_id].get("inputs", {}).get(name, default)
if _is_link(value):
raise ValueError(f"{name} on distributed node {node_id!r} must be a literal")
return copy.deepcopy(value)
resolved_distributed = {}
def resolve_distributed(node_id, output_slot):
node_id = str(node_id)
class_type = source_nodes[node_id]["class_type"]
if output_slot != 0:
raise ValueError(f"invalid output slot {output_slot} for {class_type} {node_id!r}")
cache_key = (node_id, output_slot)
if cache_key in resolved_distributed:
return copy.deepcopy(resolved_distributed[cache_key])
inputs = source_nodes[node_id].get("inputs", {})
if class_type == "DistributedSeed":
seed = resolve_literal_or_primitive(inputs.get("seed", _SEED_DEFAULT), f"seed on {node_id!r}")
is_worker = resolve_hidden(node_id, "is_worker", False)
worker_id = resolve_hidden(node_id, "worker_id", "")
if is_worker:
worker_index = _worker_index(worker_id)
if worker_index is not None:
seed = seed + worker_index + 1
result = seed
elif class_type == "DistributedValue":
default_value = resolve_literal_or_primitive(
inputs.get("default_value", ""), f"default_value on {node_id!r}"
)
worker_values = resolve_literal_or_primitive(
inputs.get("worker_values", "{}"), f"worker_values on {node_id!r}"
)
is_worker = resolve_hidden(node_id, "is_worker", False)
worker_id = resolve_hidden(node_id, "worker_id", "")
try:
values = json.loads(worker_values) if isinstance(worker_values, str) else worker_values
except json.JSONDecodeError:
values = {}
if not isinstance(values, dict):
values = {}
value_type = values.get("_type", "STRING")
fallback = _coerce_safe(default_value, value_type)
result = fallback
if is_worker:
worker_index = _worker_index(worker_id)
if worker_index is not None:
raw_value = values.get(str(worker_index + 1), "")
if raw_value:
try:
result = _coerce(raw_value, value_type)
except (ValueError, IndexError):
result = fallback
elif class_type == "DistributedModelName":
text = resolve_literal_or_primitive(
inputs.get("text", ""), f"DistributedModelName text on {node_id!r}"
)
result = _model_name_result(text)
else:
raise ValueError(f"unsupported distributed node: {class_type}")
resolved_distributed[cache_key] = copy.deepcopy(result)
return result
return resolve_distributed
def _build_savers(source_nodes, collector_ids):
"""Construct stable SaveImage nodes and their collector metadata."""
# The prepared collector values define the artifact identity. The fallback
# for absent hidden values is stable and retains the collector's job scope.
collector_specs = []
max_numeric_id = 0
for node_id in source_nodes:
try:
max_numeric_id = max(max_numeric_id, int(node_id))
except ValueError:
pass
next_numeric_id = max_numeric_id + 1
for collector_id in collector_ids:
inputs = source_nodes[collector_id].get("inputs", {})
raw_job_id = inputs.get("multi_job_id", collector_id)
if _is_link(raw_job_id):
raise ValueError(f"collector {collector_id!r} multi_job_id must be a literal")
job_id = str(collector_id if raw_job_id in (None, "") else raw_job_id)
raw_worker_id = inputs.get("worker_id", "")
if _is_link(raw_worker_id):
raise ValueError(f"collector {collector_id!r} worker_id must be a literal")
worker_id = "" if raw_worker_id is None else str(raw_worker_id)
saver_id = str(next_numeric_id)
next_numeric_id += 1
prefix = _filename_prefix(job_id, worker_id, collector_id, "image")
collector_specs.append(
{
"collector_id": collector_id,
"job_id": job_id,
"node_id": saver_id,
"kind": "image",
"prefix": prefix,
"source_link": copy.deepcopy(inputs["images"]),
}
)
saver_nodes = {}
for spec in collector_specs:
saver_nodes[spec["node_id"]] = {
"class_type": "SaveImage",
"inputs": {
"images": spec["source_link"],
"filename_prefix": spec["prefix"],
},
"_meta": {"title": f"Serverless output: collector {spec['collector_id']} (image)"},
}
return collector_specs, saver_nodes
def _prune_graph(source_nodes, collector_specs, saver_nodes, resolve_distributed):
"""Keep saver ancestors, replacing distributed scalar links with literals."""
all_nodes = {node_id: copy.deepcopy(node) for node_id, node in source_nodes.items()}
all_nodes.update(copy.deepcopy(saver_nodes))
needed = set()
pending = [spec["node_id"] for spec in collector_specs]
while pending:
node_id = pending.pop()
if node_id in needed:
continue
needed.add(node_id)
node = all_nodes[node_id]
if node["class_type"] == "DistributedCollector":
raise ValueError("collector-to-collector dependence is unsupported")
for input_name, value in list(node.get("inputs", {}).items()):
if not _is_link(value):
continue
upstream_id, output_slot = str(value[0]), value[1]
upstream_class = source_nodes[upstream_id]["class_type"]
if upstream_class in _REMOVED_SCALAR_NODES:
node["inputs"][input_name] = resolve_distributed(upstream_id, output_slot)
else:
pending.append(upstream_id)
compiled = {
node_id: copy.deepcopy(node)
for node_id, node in all_nodes.items()
if node_id in needed
}
if any("distributed" in node["class_type"].lower() for node in compiled.values()):
raise ValueError("compiled prompt still contains a distributed dependency")
return compiled
def compile_serverless_prompt(prepared_prompt: dict) -> tuple[dict, list[dict]]:
"""Compile a prepared worker prompt to an image-only serverless graph.
Every DistributedCollector's original ``images`` input is connected to a
new SaveImage node. Only those savers and their upstream dependencies are
retained. DistributedSeed/Value/ModelName links are resolved as scalar
literals; unsupported node execution is never attempted.
"""
source_nodes, collector_ids = _validate_prepared_prompt(prepared_prompt)
resolve_distributed = _make_scalar_resolver(source_nodes)
collector_specs, saver_nodes = _build_savers(source_nodes, collector_ids)
compiled = _prune_graph(
source_nodes,
collector_specs,
saver_nodes,
resolve_distributed,
)
manifest = [
{key: spec[key] for key in ("collector_id", "job_id", "node_id", "kind", "prefix")}
for spec in collector_specs
]
return compiled, manifest
+413
View File
@@ -0,0 +1,413 @@
"""Master-owned asynchronous Runpod jobs, independent of worker callbacks."""
import asyncio
import time
import uuid
from pathlib import Path
from .runpod_client import RunpodClient, RunpodError, AmbiguousSubmission
from .serverless_ledger import JobLedger
from .serverless_state import (
PROVIDER_STATES, PROVIDER_TERMINAL, belongs_to_pool, is_unresolved,
occupies_capacity, terminal_provider_status,
)
# Fixed per-worker cap on this master's outstanding Runpod jobs; equals the
# Images per run maximum (MAX_JOBS_PER_RUN in utils/serverless_config.py), so
# every image of a run is submitted at once. Deliberately not configurable or
# derived from jobs_per_run: unresolved jobs from earlier runs (ambiguous
# submissions, unconfirmed cancels) still count against it, and a constant
# never has to change while such jobs exist.
PER_WORKER_JOB_LIMIT = 16
async def decode_results(output, manifest):
from .serverless_media import collect_images
import torch
images = await collect_images(output, manifest)
return {job_id: [torch.from_numpy(image).unsqueeze(0) for image in batch] for job_id, batch in images.items()}
def state_directory():
import folder_paths
return Path(folder_paths.get_user_directory()) / "distributed" / "serverless"
def get_secret_store():
from ...utils.serverless_config import SecretStore
return SecretStore(state_directory() / "credentials.json")
def get_manager():
import server
from ...utils.network import get_client_session
ps = server.PromptServer.instance
manager = getattr(ps, "distributed_serverless", None)
if manager is None:
manager = ServerlessManager(ps, get_secret_store(), state_directory() / "jobs.json", get_client_session)
ps.distributed_serverless = manager
return manager
class ServerlessManager:
def __init__(self, prompt_server, secrets, ledger_path, session_factory):
self.server = prompt_server
self.secrets = secrets
self.ledger = JobLedger(ledger_path)
self.session_factory = session_factory
self.tasks = {}
self.runs = {}
self.pool_limits = {}
self.health_cache = {}
self.health_locks = {}
self.poll_seconds = 2.0
self.capacity_changed = asyncio.Condition()
self.recovery_lock = asyncio.Lock()
self._recover_local_records()
def _recover_local_records(self):
# WAITING is committed before admission; it cannot have reached Runpod.
# SUBMITTING is an intent record, so a lost ID must remain reserved.
for record in self.ledger.all():
if record.get("status") == "WAITING" and not record.get("job_id"):
self.ledger.update(record["record_id"], status="CANCELLED", updated_at=time.time())
elif record.get("status") == "SUBMITTING" and not record.get("job_id"):
self.ledger.update(record["record_id"], status="UNKNOWN", updated_at=time.time())
def _task_is_active(self, record_id):
task = self.tasks.get(record_id)
return task is not None and not task.done()
def pool_is_active(self, worker_id):
"""Protect both live local work and unresolved pre-restart provider work."""
for rid, run in self.runs.items():
pool_id = run["worker"].get("pool_id", run["worker"]["id"])
if str(pool_id) == str(worker_id) and self._task_is_active(rid):
return True
return any(belongs_to_pool(record, worker_id) and is_unresolved(record)
for record in self.ledger.all())
async def _reserve_provider_capacity(self, run):
pool_id = run["worker"].get("pool_id", run["worker"]["id"])
async with self.capacity_changed:
while True:
self._check_master(run)
outstanding = sum(belongs_to_pool(record, pool_id) and occupies_capacity(record)
for record in self.ledger.all())
if outstanding < PER_WORKER_JOB_LIMIT:
# Claim atomically before any await that could submit a billed job.
self.ledger.update(run["record_id"], status="SUBMITTING", updated_at=time.time())
return
await self.capacity_changed.wait()
async def _notify_capacity(self):
async with self.capacity_changed:
self.capacity_changed.notify_all()
async def _client(self, worker):
pool_id = worker.get("pool_id", worker["id"])
key = self.secrets.get(pool_id)
return RunpodClient(await self.session_factory(), worker["endpoint_id"], key)
async def health(self, worker, *, fresh=False):
pool_id = worker.get("pool_id", worker["id"])
cache_key = (pool_id, worker["endpoint_id"])
lock = self.health_locks.setdefault(cache_key, asyncio.Lock())
async with lock:
cached = self.health_cache.get(cache_key)
if cached and not fresh and time.monotonic() - cached[0] < 10:
return cached[1]
client = await self._client(worker)
data = await client.health()
jobs = data.get("jobs", {})
if not isinstance(jobs, dict) or not isinstance(data.get("workers"), dict):
raise RunpodError("Runpod returned an invalid endpoint health record")
try:
pending = max(0, int(jobs.get("inQueue", 0))) + max(0, int(jobs.get("inProgress", 0)))
except (ValueError, TypeError):
raise RunpodError("Runpod returned invalid queue counts") from None
result = {"ok": True, "queueRemaining": pending, "message": "Queued / processing" if pending else "Ready (scales from zero)"}
self.health_cache[cache_key] = (time.monotonic(), result)
return result
def _semaphore(self, worker):
# The limit is a constant, so it never has to be swapped while this
# worker still has unresolved jobs.
pool_id = worker.get("pool_id", worker["id"])
semaphore = self.pool_limits.get(pool_id)
if semaphore is None:
semaphore = asyncio.Semaphore(PER_WORKER_JOB_LIMIT)
self.pool_limits[pool_id] = semaphore
return semaphore
def ensure_capacity(self, count):
if sum(not task.done() for task in self.tasks.values()) + count > 64:
raise ValueError("Distributed already has too many pending serverless jobs (limit 64)")
async def start(self, execution_id, prompt_id, prepared_jobs):
self.ensure_capacity(len(prepared_jobs))
from .serverless_typed_results import validate_prepared_job
# Validate the entire batch before creating even the first local slot.
for prepared in prepared_jobs:
validate_prepared_job(prepared)
new_runs = []
# Journal every local slot before starting any billed submission.
for prepared in prepared_jobs:
worker = prepared["worker"]
semaphore = self._semaphore(worker)
record_id = uuid.uuid4().hex
now = time.time()
self.ledger.update(record_id, execution_id=execution_id, prompt_id=prompt_id,
worker_id=worker.get("pool_id", worker["id"]), participant_id=worker["id"],
endpoint_id=worker["endpoint_id"], job_id=None, status="WAITING", created_at=now, updated_at=now)
run = dict(prepared, execution_id=execution_id, prompt_id=prompt_id, record_id=record_id,
provider_job_id=None, client=None, deadline=time.monotonic() + worker["timeout_seconds"])
self.runs[record_id] = run
new_runs.append((record_id, run, semaphore))
for record_id, run, semaphore in new_runs:
self.tasks[record_id] = asyncio.create_task(self._run(run, semaphore), name=f"distributed-serverless-{record_id}")
# Bound in-memory completed tracking as well as the on-disk journal.
done = [rid for rid, task in self.tasks.items() if task.done()]
for rid in done[:-200]:
self.tasks.pop(rid, None)
self.runs.pop(rid, None)
# Additive ownership handle for item schedulers; legacy callers ignore it.
return [record_id for record_id, _run, _semaphore in new_runs]
def is_pending(self, collector_job_id, participant_id):
return any(not task.done() and run["worker"]["id"] == participant_id and
any(m["job_id"] == collector_job_id for m in run["manifest"])
for rid, task in self.tasks.items() for run in [self.runs[rid]])
def owns(self, collector_job_id, participant_id):
return any(run["worker"]["id"] == participant_id and
any(m["job_id"] == collector_job_id for m in run["manifest"]) for run in self.runs.values())
async def _deliver(self, run, images=None, error=None, *, media=None):
if media is not None and not error:
# Validate all destinations before exposing any partial success.
for binding in run["manifest"]:
if binding["participant_id"] != run["worker"]["id"]:
raise ValueError("Serverless media participant identity mismatch")
values = media[binding["job_id"]][binding["socket"]]
if not isinstance(values, list) or not values:
raise ValueError("Missing serverless media values")
async with self.server.distributed_jobs_lock:
delivered = set()
for binding in run["manifest"]:
job_id = binding["job_id"]
if error or media is not None:
if job_id in delivered:
continue
delivered.add(job_id)
queue = self.server.distributed_pending_jobs.get(binding["job_id"])
if queue is None:
continue # A cancelled/finished collector must never be resurrected.
if error:
queue.put_nowait({"worker_id": run["worker"]["id"], "serverless_error": error, "is_last": True})
elif media is not None:
queue.put_nowait({"worker_id": run["worker"]["id"], "media": media[job_id], "is_last": True})
else:
batch = images[binding["job_id"]]
for index, tensor in enumerate(batch):
queue.put_nowait({"worker_id": run["worker"]["id"], "image_index": index,
"tensor": tensor, "is_last": index == len(batch) - 1})
async def _cancel_provider(self, run):
if run["provider_job_id"] and run["client"]:
try:
result = await run["client"].cancel(run["provider_job_id"])
return terminal_provider_status(result)
except Exception:
return None
return None
def _check_master(self, run):
queue = getattr(self.server, "prompt_queue", None)
if queue is None:
return
history = queue.get_history(run["prompt_id"])
record = history.get(run["prompt_id"], {})
if record.get("status", {}).get("status_str") == "error":
raise RunpodError("Master workflow failed; serverless work stopped")
if not record and hasattr(queue, "get_current_queue"):
running, pending = queue.get_current_queue()
if not any(str(item[1]) == run["prompt_id"] for item in [*running, *pending]):
raise RunpodError("Master workflow was removed from the queue; serverless work stopped")
async def _run(self, run, semaphore):
rid = run["record_id"]
submitted = False
provider_terminal = False
try:
from .serverless_typed_results import validate_prepared_job, collect_typed_results
typed = validate_prepared_job(run)
async with asyncio.timeout(max(0, run["deadline"] - time.monotonic())):
async with semaphore:
await self._reserve_provider_capacity(run)
client = await self._client(run["worker"])
run["client"] = client
submitted = True
if typed:
job_id = await client.submit(run["workflow"], run["images"], run["worker"]["timeout_seconds"], media=run["media"])
else:
job_id = await client.submit(run["workflow"], run["images"], run["worker"]["timeout_seconds"])
run["provider_job_id"] = job_id
# Persist immediately: all subsequent actions use this exact ID.
self.ledger.update(rid, job_id=job_id, status="IN_QUEUE", updated_at=time.time())
while True:
self._check_master(run)
response = await client.status(job_id)
status = response.get("status")
if not isinstance(status, str) or status not in PROVIDER_STATES:
raise RunpodError("Runpod returned an unknown job state")
provider_terminal = status in PROVIDER_TERMINAL
if status == "COMPLETED":
output = response.get("output")
if not isinstance(output, dict) or output.get("error") or output.get("errors"):
raise RunpodError("Serverless workflow reported an output error; inspect the provider job")
self.ledger.update(rid, status="DOWNLOADING", updated_at=time.time())
if typed:
media = await collect_typed_results(output, run["media"], run["manifest"])
await self._deliver(run, media=media)
else:
images = await decode_results(output, run["manifest"])
await self._deliver(run, images=images)
self.ledger.update(rid, status="COMPLETED", updated_at=time.time())
return
self.ledger.update(rid, status=status, updated_at=time.time())
if provider_terminal:
raise RunpodError(f"Serverless job ended with {status}")
await asyncio.sleep(self.poll_seconds)
except AmbiguousSubmission as exc:
self.ledger.update(rid, status="UNKNOWN", error=str(exc), updated_at=time.time())
await self._deliver(run, error=str(exc))
except (asyncio.TimeoutError, asyncio.CancelledError) as exc:
terminal = "FAILED" if provider_terminal else await self._cancel_provider(run)
if submitted and not run["provider_job_id"]:
status = "UNKNOWN"
error = "Submission interrupted before its job ID was received; check Runpod before retrying"
elif isinstance(exc, asyncio.TimeoutError):
status = terminal or ("CANCEL_REQUESTED" if run["provider_job_id"] else "TIMED_OUT")
error = "Serverless job exceeded its queue/execution deadline"
else:
status = terminal or ("CANCEL_REQUESTED" if run["provider_job_id"] else "CANCELLED")
error = "Serverless work stopped" if terminal or not submitted else "Serverless cancellation requested; provider confirmation is pending"
self.ledger.update(rid, status=status, error=error, updated_at=time.time())
await self._deliver(run, error=error)
except Exception as exc:
# If polling/decoding failed, stop only our own provider job, never the pool.
terminal = "FAILED" if provider_terminal else await self._cancel_provider(run)
error = str(exc) if isinstance(exc, RunpodError) else "Serverless processing failed; check the workflow and endpoint contract"
status = terminal or ("CANCEL_REQUESTED" if run["provider_job_id"] else "FAILED")
self.ledger.update(rid, status=status, error=error, updated_at=time.time())
await self._deliver(run, error=error)
finally:
# Completed tracking retains IDs/bindings only, not potentially large
# base64 inputs, workflow metadata, or a client holding credentials.
run.pop("workflow", None)
run.pop("images", None)
run.pop("media", None)
run["client"] = None
# A released local semaphore alone cannot release a durable reservation.
await self._notify_capacity()
async def wait_for_execution(self, execution_id):
tasks = [task for rid, task in self.tasks.items() if self.runs[rid]["execution_id"] == execution_id]
await asyncio.gather(*tasks, return_exceptions=False)
def _detached_records(self, worker_id=None, execution_id=None):
return [record for record in self.ledger.all()
if is_unresolved(record) and not self._task_is_active(record["record_id"])
and (worker_id is None or belongs_to_pool(record, worker_id))
and (execution_id is None or record.get("execution_id") == execution_id)]
async def _recover_provider_record(self, record, *, cancel=False):
"""Reconcile an exact saved ID, never submit or resume a workflow."""
if not record.get("job_id"):
return False
status = None
try:
client = await self._client({"id": record["worker_id"], "endpoint_id": record["endpoint_id"]})
response = await (client.cancel(record["job_id"]) if cancel else client.status(record["job_id"]))
status = terminal_provider_status(response)
except Exception:
pass # Unreachable/expired is not proof of provider termination.
if status is not None or cancel:
self.ledger.update(record["record_id"], status=status or "CANCEL_REQUESTED", updated_at=time.time())
return status is not None
async def reconcile(self, worker_id=None):
"""Read terminal states for detached jobs and release only proven capacity."""
resolved = 0
async with self.recovery_lock:
for record in self._detached_records(worker_id):
resolved += await self._recover_provider_record(record)
await self._notify_capacity()
return resolved
async def acknowledge_recovery(self, worker_id, record_id, *, provider_checked=False):
"""Explicit operator escape hatch for lost IDs/expired provider records.
This records a human's verification, not a cancellation or provider
status. It is never called automatically or allowed for live tasks.
"""
if provider_checked is not True:
raise ValueError("Check Runpod and confirm that this job is no longer running before resolving it")
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 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()
async def cancel(self, worker_id=None, execution_id=None, collector_job_id=None):
targets = {}
for rid, task in self.tasks.items():
run = self.runs[rid]
if worker_id and run["worker"].get("pool_id", run["worker"]["id"]) != worker_id:
continue
if execution_id and run["execution_id"] != execution_id:
continue
if collector_job_id and not any(m["job_id"] == collector_job_id for m in run["manifest"]):
continue
if not task.done():
# Repeated requests must not interrupt the first cancellation's cleanup.
if not task.cancelling():
task.cancel()
targets[rid] = task
if targets:
await asyncio.gather(*targets.values(), return_exceptions=True)
recovered = 0
async with self.recovery_lock:
for record in self._detached_records(worker_id, execution_id):
rid = record["record_id"]
# A task cancelled before its first step never enters _run's handlers.
# A failed start() may also leave a journalled but unstarted local slot.
if record.get("status") == "WAITING" and not record.get("job_id"):
run = self.runs.get(rid)
if run is not None and (collector_job_id is None or any(m["job_id"] == collector_job_id for m in run["manifest"])):
self.ledger.update(rid, status="CANCELLED", updated_at=time.time())
run.pop("workflow", None)
run.pop("images", None)
run.pop("media", None)
await self._deliver(run, error="Serverless job cancelled before submission")
continue
# Do not send two cancellation requests for the same live job in
# one action. Its unresolved reservation survives for later recovery.
if rid in targets or collector_job_id or not record.get("job_id"):
continue
await self._recover_provider_record(record, cancel=True)
recovered += 1
await self._notify_capacity()
return len(targets) + recovered
def public_jobs(self, worker_id):
records = [record for record in reversed(self.ledger.all()) if belongs_to_pool(record, worker_id)]
# Never hide unresolved IDs behind the bounded completed history.
visible = [record for record in records if is_unresolved(record)]
visible.extend(record for record in records if not is_unresolved(record))
unresolved_count = sum(is_unresolved(record) for record in records)
return [{key: record.get(key) for key in ("record_id", "job_id", "status", "worker_id", "execution_id", "error")}
for record in visible[:max(20, unresolved_count)]]
+29
View File
@@ -0,0 +1,29 @@
"""Shared rules for provider capacity and durable recovery protection.
A local task ending is not evidence that provider work ended. RESOLVED is an
explicit operator acknowledgment, never a provider-reported state.
"""
PROVIDER_TERMINAL = frozenset({"COMPLETED", "FAILED", "CANCELLED", "TIMED_OUT"})
TERMINAL = PROVIDER_TERMINAL | {"RESOLVED"}
PROVIDER_STATES = PROVIDER_TERMINAL | {"IN_QUEUE", "IN_PROGRESS"}
def terminal_provider_status(response):
status = response.get("status") if isinstance(response, dict) else None
return status if isinstance(status, str) and status in PROVIDER_TERMINAL else None
def is_unresolved(record):
return record.get("status") not in TERMINAL
def occupies_capacity(record):
# WAITING is journalled before admission and cannot have been submitted.
# Every other unresolved state is conservatively reserved, including a
# submission whose provider ID was lost and an unconfirmed cancellation.
return is_unresolved(record) and (record.get("status") != "WAITING" or bool(record.get("job_id")))
def belongs_to_pool(record, worker_id):
return str(record.get("worker_id")) == str(worker_id)
@@ -0,0 +1,225 @@
"""Bounded local native inputs for comfy-media-v1 requests.
Only stock LoadAudio/LoadVideo schemas are adapted. Original nodes and links
remain in the private request graph for endpoint validation; only the existing
image preparer rewrites filenames. No upload, transcode, resample or provider
operation belongs here. The caller still enforces the complete request-body cap.
"""
import hashlib
import io
import json
from pathlib import Path
from ...serverless_bridge import codec, contract, video_codec
from . import serverless_media
_LOADERS = {"LoadAudio": ("audio", "AUDIO"), "LoadVideo": ("file", "VIDEO")}
Error = serverless_media.ServerlessMediaError
def _loader_references(workflow):
if not isinstance(workflow, dict):
raise Error("Serverless workflow must be an API graph")
loaders = {}
for node_id, node in workflow.items():
if not isinstance(node, dict):
continue
class_type = node.get("class_type")
if not isinstance(class_type, str) or class_type not in _LOADERS:
continue
contract.identifier(node_id)
key, kind = _LOADERS[class_type]
inputs = node.get("inputs")
if not isinstance(inputs, dict) or set(inputs) != {key}:
raise Error(f"{class_type} inputs must contain only its native filename; extra controls are unsupported")
if not isinstance(inputs[key], str):
raise Error(f"{class_type} requires a literal filename; linked inputs are unsupported")
normalized, _suffix = serverless_media._normalize_media_reference(inputs[key])
loaders[node_id] = (kind, normalized)
references = []
used = set()
for node_id, node in workflow.items():
if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict):
continue
for key, value in node["inputs"].items():
if not isinstance(value, (list, tuple)) or not value or not isinstance(value[0], str):
continue
if value[0] not in loaders:
continue
if type(value) is not list or len(value) != 2 or type(value[1]) is not int or value[1] != 0:
raise Error("Native media loader links require output slot 0")
contract.identifier(node_id)
contract.identifier(key)
references.append((node_id, key, value[0]))
used.add(value[0])
if used != set(loaders):
raise Error("Native media loader has no consumer")
return loaders, references
def _source_path(value, folder_paths):
"""Resolve native annotations; authorization/opening is owned by _read_path."""
annotation = serverless_media._MEDIA_ANNOTATION.search(value)
unannotated = value[:annotation.start()] if annotation else value
if Path(unannotated).is_absolute():
return unannotated
try:
return folder_paths.get_annotated_filepath(value)
except Exception:
raise Error("Native media input path is missing or invalid") from None
def _decode_audio(data, limit):
"""Match core nodes_audio.load, bounding each frame *before* accumulation.
Float32 output bytes are charged even for narrow integer PCM. Native float64
remains float64, as in f32_pcm (there is no silent dtype conversion). Raw
retained arrays and each live AV plane allocation have a separate <= limit
guard, so wider/padded raw frames cannot bypass the output budget. PyAV owns
the one frame being decoded; it is checked before to_ndarray or torch.cat.
Concatenation/conversion use bounded temporary copies, not an RSS guarantee.
"""
if limit <= 0:
raise Error("AUDIO decoded input exceeds byte limit")
try:
import av
import torch
from comfy_extras.nodes_audio import f32_pcm
except ImportError:
raise Error("Native AUDIO decoding is unavailable") from None
try:
with av.open(io.BytesIO(data)) as container:
if not container.streams.audio:
raise Error("Native AUDIO input has no audio stream")
stream = container.streams.audio[0]
rate = stream.codec_context.sample_rate
channels = stream.channels
if type(rate) is not int or rate <= 0 or type(channels) is not int or channels <= 0:
raise Error("Native AUDIO stream shape or sample rate is invalid")
frames = []
output_bytes = raw_bytes = 0
dtype = None
for frame in container.decode(streams=stream.index):
samples = frame.samples
if (type(samples) is not int or samples <= 0 or frame.sample_rate != rate
or len(frame.layout.channels) != channels):
raise Error("Native AUDIO frame shape or sample rate changed")
if frame.format.name not in {"s16", "s16p", "s32", "s32p", "flt", "fltp", "dbl", "dblp"}:
raise Error("Unsupported native AUDIO PCM format")
itemsize = frame.format.bytes
count = samples * channels
# Native f32_pcm promotes integer PCM to float32, not float64.
decoded_bytes = count * max(4, itemsize)
frame_bytes = count * itemsize
plane_bytes = sum(plane.buffer_size for plane in frame.planes)
if (decoded_bytes > limit - output_bytes or frame_bytes > limit - raw_bytes
or plane_bytes > limit - raw_bytes):
raise Error("AUDIO decoded input or raw frame exceeds byte limit")
output_bytes += decoded_bytes
raw_bytes += frame_bytes
array = frame.to_ndarray()
if array.ndim != 2 or array.size != count or array.nbytes != frame_bytes:
raise Error("Native AUDIO frame shape is invalid")
buf = torch.from_numpy(array)
if buf.dtype not in (torch.int16, torch.int32, torch.float32, torch.float64):
raise Error("Unsupported native AUDIO PCM dtype")
if dtype is not None and buf.dtype != dtype:
raise Error("Native AUDIO frame dtype changed")
dtype = buf.dtype
if buf.shape[0] != channels:
buf = buf.view(-1, channels).t()
if tuple(buf.shape) != (channels, samples):
raise Error("Native AUDIO frame shape is invalid")
frames.append(buf)
if not frames:
raise Error("Native AUDIO input has no decoded frames")
waveform = f32_pcm(torch.cat(frames, dim=1)).unsqueeze(0)
return {"waveform": waveform, "sample_rate": rate}
except Error:
raise
except Exception:
raise Error("Native AUDIO input could not be decoded safely") from None
def _artifact_id(kind, descriptor, payload):
digest = hashlib.sha256()
digest.update(kind.encode("ascii") + b"\0")
# Include interpretation (not just waveform bytes), notably sample_rate.
digest.update(json.dumps(descriptor, sort_keys=True, separators=(",", ":"), allow_nan=False).encode())
digest.update(b"\0")
digest.update(payload)
return "input-" + kind.lower() + "-" + digest.hexdigest()
async def prepare_media_inputs(workflow: dict) -> tuple[list, list, list]:
"""Return legacy images, inline typed artifacts, and every consumer reference.
Paths are read once into bounded bytes through the video adapter's no-follow
directory-fd reader. Decoders receive BytesIO, never a second path open.
Source files and unique decoded artifacts have independent aggregate caps.
Repeated content of the same kind is decoded once and charged once.
"""
loaders, references = _loader_references(workflow)
image_view = {node_id: node for node_id, node in workflow.items() if node_id not in loaders}
# Retain the strict scan for every other loader before expensive decoding.
serverless_media._iter_image_references(image_view)
if not loaders:
return await serverless_media.prepare_images(image_view), [], []
try:
import folder_paths
except ImportError:
raise Error("ComfyUI media directories are unavailable") from None
roots = list(serverless_media._folder_roots(folder_paths).values())
artifacts = {}
source_artifacts = {}
source_digests = set()
filenames = {}
loader_artifacts = {}
file_total = decoded_total = 0
for node_id, (kind, filename) in loaders.items():
# Avoid retaining file buffers merely to deduplicate repeat references.
if (kind, filename) in filenames:
loader_artifacts[node_id] = filenames[kind, filename]
continue
path = _source_path(filename, folder_paths)
data = video_codec._read_path(path, contract.MAX_ARTIFACT_BYTES, roots)
source_digest = hashlib.sha256(data).hexdigest()
if source_digest not in source_digests:
if len(data) > contract.MAX_TOTAL_BYTES - file_total:
raise Error("Native media source files exceed total byte limit")
file_total += len(data)
source_digests.add(source_digest)
source_key = (kind, source_digest)
if source_key not in source_artifacts:
if len(artifacts) >= contract.MAX_ARTIFACTS:
raise Error("Native media input artifact count exceeds limit")
limit = min(contract.MAX_ARTIFACT_BYTES, contract.MAX_TOTAL_BYTES - decoded_total)
if kind == "AUDIO":
value = _decode_audio(data, limit)
descriptor, payload = codec.encode_payload(kind, value, max_bytes=limit)
del value
else:
from comfy_api.latest import InputImpl
value = InputImpl.VideoFromFile(io.BytesIO(data))
descriptor, payload = video_codec.encode_video(value, max_bytes=limit, allowed_roots=roots)
del value
aid = _artifact_id(kind, descriptor, payload)
if aid not in artifacts:
if len(payload) > limit:
raise Error("Native media decoded inputs exceed total byte limit")
artifacts[aid] = contract.artifact_from_payload(
aid, kind, descriptor, payload, max_bytes=contract.MAX_ARTIFACT_BYTES)
decoded_total += len(payload)
source_artifacts[source_key] = aid
del payload
aid = source_artifacts[source_key]
loader_artifacts[node_id] = filenames[kind, filename] = aid
del data
refs = [{"node_id": node_id, "input_name": key, "artifact_id": loader_artifacts[source]}
for node_id, key, source in references]
# Shared node dictionaries deliberately propagate only legacy image rewrites.
images = await serverless_media.prepare_images(image_view)
return images, list(artifacts.values()), refs
@@ -0,0 +1,79 @@
"""Explicit media-v1 preflight and atomic, lossless result materialization."""
from urllib.parse import urljoin
import aiohttp
from ...serverless_bridge import contract
from ...serverless_bridge.contract import MAX_ARTIFACT_BYTES, PROFILE, validate_bindings
from .serverless_media import (
MAX_REDIRECTS, _CHUNK_SIZE, _DOWNLOAD_TIMEOUT_SECONDS,
_PublicResolver, _validate_artifact_url,
)
def validate_prepared_job(prepared):
"""Fail before journalling/submission; legacy jobs keep their original path."""
profile = prepared["worker"].get("media_profile", "legacy-image-v1")
media = prepared.get("media")
if profile == "legacy-image-v1" and media is None:
return False
if profile != PROFILE or media is None:
raise ValueError("Serverless media profile/request mismatch")
from .runpod_client import serialize_request
# The shared endpoint validator also checks exporter controls and input refs.
serialize_request(prepared["workflow"], prepared["images"],
prepared["worker"]["timeout_seconds"], media=media)
expected = validate_bindings(prepared["manifest"])
if expected != validate_bindings(media["outputs"]):
raise ValueError("Serverless media manifest identity mismatch")
if any(binding["participant_id"] != prepared["worker"]["id"] for binding in expected.values()):
raise ValueError("Serverless media participant identity mismatch")
return True
async def download_public_artifact(url, *, max_bytes=MAX_ARTIFACT_BYTES):
"""Contract-sized public HTTPS download, never using a provider session/key."""
if type(max_bytes) is not int or not 0 <= max_bytes <= MAX_ARTIFACT_BYTES:
raise ValueError("Media artifact exceeds the permitted size")
current_url = _validate_artifact_url(url)
connector = aiohttp.TCPConnector(resolver=_PublicResolver(), use_dns_cache=False, limit=4)
timeout = aiohttp.ClientTimeout(total=_DOWNLOAD_TIMEOUT_SECONDS, connect=10, sock_read=15)
try:
async with aiohttp.ClientSession(connector=connector, timeout=timeout, trust_env=False,
cookie_jar=aiohttp.DummyCookieJar()) as session:
for redirect_count in range(MAX_REDIRECTS + 1):
async with session.get(current_url, allow_redirects=False) as response:
if response.status in {301, 302, 303, 307, 308}:
location = response.headers.get("Location")
if not location or redirect_count >= MAX_REDIRECTS:
raise ValueError("Media artifact redirect is invalid")
current_url = _validate_artifact_url(urljoin(current_url, location))
continue
if response.status != 200:
raise ValueError("Media artifact download failed")
payload = bytearray()
async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
if not isinstance(chunk, bytes) or len(payload) + len(chunk) > max_bytes:
raise ValueError("Media artifact exceeds the permitted size")
payload.extend(chunk)
return bytes(payload)
except Exception:
# Signed URLs and transport exception details must not reach the ledger.
raise ValueError("Media artifact download failed or exceeded its byte limit") from None
raise ValueError("Media artifact download failed")
async def collect_typed_results(output, media, manifest, *, download=None):
"""Decode every binding before returning any collector's native value lists."""
if not isinstance(output, dict) or output.get("error") or output.get("errors"):
raise ValueError("Invalid serverless media output")
decoded = await contract.decode_response(output.get("media"), request_id=media["request_id"],
bindings=manifest,
download=download if download is not None else download_public_artifact)
grouped = {}
for binding in manifest:
values = decoded[binding["binding_id"]]
if not isinstance(values, list) or not values:
raise ValueError("Missing serverless media values")
grouped.setdefault(binding["job_id"], {}).setdefault(binding["socket"], []).extend(values)
return grouped
+233 -29
View File
@@ -1,4 +1,5 @@
import asyncio
import copy
import time
import uuid
@@ -21,7 +22,12 @@ from .orchestration.dispatch import (
select_active_workers,
select_least_busy_worker,
)
from .orchestration.media_sync import convert_paths_for_platform, fetch_worker_path_separator, sync_worker_media
from .orchestration.media_sync import (
convert_paths_for_platform,
fetch_worker_has_node,
fetch_worker_path_separator,
sync_worker_media,
)
from .orchestration.prompt_transform import (
PromptIndex,
apply_participant_overrides,
@@ -29,7 +35,11 @@ from .orchestration.prompt_transform import (
generate_job_id_map,
prepare_delegate_master_prompt,
prune_prompt_for_worker,
WORKER_SINK_CLASS,
_resolve_direct_workers_for_image_list,
)
from ..image_list.runtime import validate_image_list_regions, _has_selected_serverless_workers
from ..image_list.execution_plan import prepare_image_list_execution
prompt_server = server.PromptServer.instance
@@ -74,6 +84,18 @@ def _resolve_enabled_workers(config, requested_ids=None):
elif not worker.get("enabled", False):
continue
if worker.get("type") == "serverless":
from ..utils.serverless_config import validate_serverless_worker
pool = validate_serverless_worker(worker)
if pool.get("media_profile") == "comfy-media-v1":
from ..serverless_bridge.contract import identifier
# Do not repair malformed media identities by stripping them.
identifier(pool["id"])
for slot in range(pool["jobs_per_run"]):
workers.append(dict(pool, pool_id=worker_id, id=f"serverless:{worker_id}:{slot + 1}",
name=f"{pool.get('name', worker_id)} / job {slot + 1}"))
continue
raw_port = worker.get("port", worker.get("listen_port", 8188))
try:
port = int(raw_port or 8188)
@@ -138,6 +160,32 @@ def _prompt_requests_load_balance(prompt_index):
return False
def _validate_media_workers(workers, prompt_index=None):
"""Reject unknown serverless media profiles before probing or dispatching.
A Distributed Collector accepts direct workers, legacy-image-v1 pools and
comfy-media-v1 pools in the same run; each pool's profile selects its
endpoint contract. Unknown profiles fail rather than guessing a contract.
legacy-image-v1 endpoints return 8-bit files, so a Collector set to exact
cannot use them.
"""
exact = prompt_index is not None and any(
str(prompt_index.inputs_by_node.get(node_id, {}).get("transfer_format", "")).strip().lower() == "exact"
for node_id in prompt_index.nodes_for_class("DistributedCollector")
)
for worker in workers:
if worker.get("type") != "serverless":
continue
profile = worker.get("media_profile", "legacy-image-v1")
if profile not in ("legacy-image-v1", "comfy-media-v1"):
raise ValueError("Unsupported serverless media_profile")
if exact and profile == "legacy-image-v1":
raise ValueError(
"transfer_format exact requires comfy-media-v1 serverless pools; "
"legacy-image-v1 endpoints return 8-bit images"
)
async def _prepare_worker_payload(
worker,
prompt_index,
@@ -153,7 +201,44 @@ async def _prepare_worker_payload(
):
"""Prepare one worker prompt payload with bounded concurrency and media-sync timeout."""
async with worker_prep_semaphore:
_validate_media_workers([worker], prompt_index)
worker_prompt = prompt_index.copy_prompt()
if worker.get("type") == "serverless":
from .orchestration.runpod_client import serialize_request
if worker.get("media_profile", "legacy-image-v1") == "comfy-media-v1":
from ..serverless_bridge.contract import PROFILE, VERSION
from .orchestration.serverless_media_prompt import compile_media_prompt
from .orchestration.serverless_typed_inputs import prepare_media_inputs
# The typed compiler validates the complete private graph and
# prunes it itself, retaining every connected Collector branch.
worker_prompt = apply_participant_overrides(
worker_prompt, worker["id"], enabled_ids, job_id_map,
"", delegate_master, prompt_index,
)
request_id = str(uuid.uuid4())
workflow, manifest = compile_media_prompt(worker_prompt, request_id)
images, artifacts, refs = await prepare_media_inputs(workflow)
media = {
"version": VERSION, "profile": PROFILE, "request_id": request_id,
"artifacts": artifacts, "inputs": refs, "outputs": copy.deepcopy(manifest),
}
serialize_request(workflow, images, worker["timeout_seconds"], media=media)
return worker, {"worker": worker, "workflow": workflow, "manifest": manifest,
"images": images, "media": media}
from .orchestration.serverless_prompt import compile_serverless_prompt
from .orchestration.serverless_media import prepare_images
worker_prompt = apply_participant_overrides(
prune_prompt_for_worker(worker_prompt), worker["id"], enabled_ids,
job_id_map, "", delegate_master, prompt_index,
)
workflow, manifest = compile_serverless_prompt(worker_prompt)
images = await prepare_images(workflow)
# Fail during preparation, before regular workers or the master are
# dispatched. submit() uses this same serializer for the sent bytes.
serialize_request(workflow, images, worker["timeout_seconds"])
return worker, {"worker": worker, "workflow": workflow, "manifest": manifest, "images": images}
worker_master_url = build_master_callback_url(
worker,
config=config,
@@ -162,12 +247,21 @@ async def _prepare_worker_payload(
worker_type = str(worker.get("type") or "local").strip().lower()
is_remote_like = bool(worker.get("host")) and worker_type != "local"
if is_remote_like:
path_separator = await fetch_worker_path_separator(worker, trace_execution_id=trace_execution_id)
if path_separator:
worker_prompt = convert_paths_for_platform(worker_prompt, path_separator)
# Older workers lack the sink node; they keep the PreviewImage terminator.
has_sink_task = asyncio.ensure_future(
fetch_worker_has_node(worker, WORKER_SINK_CLASS, trace_execution_id=trace_execution_id)
)
try:
if is_remote_like:
path_separator = await fetch_worker_path_separator(worker, trace_execution_id=trace_execution_id)
if path_separator:
worker_prompt = convert_paths_for_platform(worker_prompt, path_separator)
use_worker_sink = await has_sink_task
finally:
if not has_sink_task.done():
has_sink_task.cancel()
worker_prompt = prune_prompt_for_worker(worker_prompt)
worker_prompt = prune_prompt_for_worker(worker_prompt, use_worker_sink=use_worker_sink)
worker_prompt = apply_participant_overrides(
worker_prompt,
worker["id"],
@@ -214,6 +308,8 @@ async def orchestrate_distributed_execution(
execution_trace_id = trace_execution_id or _generate_execution_trace_id()
config = load_config()
if delegate_master is None:
delegate_master = bool(config.get("settings", {}).get("master_delegate_only", False))
use_websocket = bool(config.get("settings", {}).get("websocket_orchestration", False))
master_url = build_master_url(config=config, prompt_server_instance=prompt_server)
(
@@ -223,8 +319,53 @@ async def orchestrate_distributed_execution(
media_sync_timeout_seconds,
) = _resolve_orchestration_limits(config)
requested_ids = enabled_worker_ids if enabled_worker_ids is not None else None
workers = _resolve_enabled_workers(config, requested_ids)
image_list_mode = any(isinstance(node, dict) and node.get("class_type") == "DistributedImageListStart"
for node in prompt_obj.values())
target_kwargs = {}
if image_list_mode:
# Check the source graph before execution-root pruning can remove an
# incompatible mode (or mask it with an unrelated boundary error).
if any(isinstance(node, dict) and node.get("class_type") in (
"DistributedCollector", "UltimateSDUpscaleDistributed") for node in prompt_obj.values()):
raise ValueError(
"Distributed Image List v1 cannot be mixed with Distributed Collector or "
"Ultimate SD Upscale nodes in one prompt"
)
prompt_obj, outer_targets = prepare_image_list_execution(prompt_obj)
target_kwargs["partial_execution_targets"] = outer_targets
prompt_index = PromptIndex(prompt_obj)
if image_list_mode:
# Validate before probing or queueing anything so malformed boundaries and
# old worker releases fail before any upstream image is materialized.
validate_image_list_regions(prompt_obj, explicit_targets=True)
from ..serverless_bridge.image_transfer import DEFAULT_FORMAT, DEFAULT_QUALITY, validate_options
required_transfer_formats = set()
for end_id in prompt_index.nodes_for_class("DistributedImageListEnd"):
end_inputs = prompt_obj[end_id].get("inputs", {})
fmt, _ = validate_options(end_inputs.get("transfer_format", DEFAULT_FORMAT),
end_inputs.get("transfer_quality", DEFAULT_QUALITY))
if fmt != DEFAULT_FORMAT:
required_transfer_formats.add(fmt)
workers = _resolve_direct_workers_for_image_list(config, requested_ids)
if _has_selected_serverless_workers(config, requested_ids):
from ..image_list.serverless import resolve_serverless_workers
# Pool identity is retained once, never expanded by Images per run.
# Validate profiles before reachability can hide an invalid selection.
workers += resolve_serverless_workers(config, requested_ids)
if not workers and delegate_master:
raise RuntimeError(
"Distributed Image List in orchestrator-only mode requires a selected direct worker "
"or comfy-media-v1 serverless pool"
)
else:
workers = _resolve_enabled_workers(config, requested_ids)
_validate_media_workers(workers, prompt_index)
if any(worker.get("media_profile") == "comfy-media-v1" for worker in workers):
from ..serverless_bridge.contract import identifier
# Validate the source, not PromptIndex's JSON copy (which coerces
# integer keys to strings), before probing or dispatching workers.
for node_id in prompt_obj:
identifier(node_id)
load_balance_requested = _prompt_requests_load_balance(prompt_index)
trace_debug(
execution_trace_id,
@@ -240,26 +381,44 @@ async def orchestrate_distributed_execution(
),
)
# Respect master delegate-only configuration
if delegate_master is None:
delegate_master = bool(config.get("settings", {}).get("master_delegate_only", False))
if not workers and delegate_master:
# Keep the legacy Collector fallback, but never override image-list
# orchestrator-only mode merely because its direct workers are unavailable.
if not image_list_mode and not workers and delegate_master:
trace_debug(
execution_trace_id,
"Delegate-only requested but no workers are enabled. Falling back to master execution.",
)
delegate_master = False
active_workers, delegate_master = await select_active_workers(
active_workers, selected_delegate_master = await select_active_workers(
workers,
use_websocket,
delegate_master,
False if image_list_mode else delegate_master,
trace_execution_id=execution_trace_id,
probe_concurrency=worker_probe_concurrency,
)
if image_list_mode:
from .orchestration.dispatch import preflight_image_list_workers
if not active_workers and delegate_master:
raise RuntimeError(
"No selected worker is reachable for Distributed Image List in orchestrator-only mode"
)
from ..image_list.runtime import END_CLASS, end_override_names
requires_values = any(end_override_names(node.get("inputs", {}))
for node in prompt_obj.values() if node.get("class_type") == END_CLASS)
direct_workers = [worker for worker in active_workers if worker.get("type") != "serverless"]
preflight_kwargs = ({"required_transfer_formats": sorted(required_transfer_formats)}
if required_transfer_formats else {})
if requires_values:
preflight_kwargs["require_value_results"] = True
# Preserve busy direct candidates for queued runs; runtime rechecks idle
# availability before assignment. Never direct-probe serverless pools.
if direct_workers:
await preflight_image_list_workers(direct_workers, require_idle=False, **preflight_kwargs)
else:
delegate_master = selected_delegate_master
if load_balance_requested:
if load_balance_requested and not image_list_mode:
candidate_workers = list(active_workers)
if not delegate_master:
# Include master in load balancing only when master participation is enabled.
@@ -322,6 +481,7 @@ async def orchestrate_distributed_execution(
workflow_meta,
client_id,
include_queue_metadata=True,
**target_kwargs,
)
return (
queue_result["prompt_id"],
@@ -330,9 +490,6 @@ async def orchestrate_distributed_execution(
queue_result.get("node_errors", {}),
)
for job_id in job_id_map.values():
await _ensure_distributed_queue(job_id)
master_prompt = prompt_index.copy_prompt()
master_prompt = apply_participant_overrides(
master_prompt,
@@ -342,9 +499,10 @@ async def orchestrate_distributed_execution(
master_url,
delegate_master,
prompt_index,
workflow_meta=workflow_meta,
)
if delegate_master:
if delegate_master and not image_list_mode:
collector_ids = find_nodes_by_class(master_prompt, "DistributedCollector")
upscale_nodes = find_nodes_by_class(master_prompt, "UltimateSDUpscaleDistributed")
if upscale_nodes:
@@ -365,7 +523,7 @@ async def orchestrate_distributed_execution(
+ ", ".join(f"{worker['name']} ({worker['id']})" for worker in active_workers),
)
worker_payloads = []
if active_workers:
if active_workers and not image_list_mode:
worker_prep_semaphore = asyncio.Semaphore(worker_prep_concurrency)
media_sync_semaphore = asyncio.Semaphore(media_sync_concurrency)
worker_payloads = await asyncio.gather(
@@ -387,7 +545,29 @@ async def orchestrate_distributed_execution(
]
)
if worker_payloads:
serverless_jobs = [payload for worker, payload in worker_payloads if worker.get("type") == "serverless"]
regular_payloads = [(worker, payload) for worker, payload in worker_payloads if worker.get("type") != "serverless"]
manager = None
if serverless_jobs:
from .orchestration.serverless_runtime import get_manager
# Repeat the manager's whole-batch identity/profile preflight *before*
# any direct dispatch, master queueing or collector queue allocation.
# Pure legacy jobs are a no-op in that validator; keep their path free
# of typed imports (and compatible with existing legacy media adapters).
for prepared in serverless_jobs:
if (prepared["worker"].get("media_profile", "legacy-image-v1") != "legacy-image-v1"
or prepared.get("media") is not None):
from .orchestration.serverless_typed_results import validate_prepared_job
validate_prepared_job(prepared)
manager = get_manager()
manager.ensure_capacity(len(serverless_jobs))
# Do not leave collector queues behind when offline compilation/media validation fails.
if not image_list_mode:
for job_id in job_id_map.values():
await _ensure_distributed_queue(job_id)
if regular_payloads:
await asyncio.gather(
*[
dispatch_worker_prompt(
@@ -398,21 +578,45 @@ async def orchestrate_distributed_execution(
use_websocket=use_websocket,
trace_execution_id=execution_trace_id,
)
for worker, wprompt in worker_payloads
for worker, wprompt in regular_payloads
]
)
queue_result = await queue_prompt_payload(
master_prompt,
workflow_meta,
client_id,
include_queue_metadata=True,
)
try:
queue_result = await queue_prompt_payload(
master_prompt,
workflow_meta,
client_id,
include_queue_metadata=True,
**target_kwargs,
)
except Exception:
if serverless_jobs:
async with prompt_server.distributed_jobs_lock:
for job_id in job_id_map.values():
prompt_server.distributed_pending_jobs.pop(job_id, None)
raise
prompt_id = queue_result["prompt_id"]
if manager is not None:
try:
await manager.start(execution_trace_id, prompt_id, serverless_jobs)
except Exception:
# Wake queued collectors with an explicit error if journal/capacity setup failed.
await manager.cancel(execution_id=execution_trace_id)
async with prompt_server.distributed_jobs_lock:
for prepared in serverless_jobs:
# Typed manifests have several sockets per collector, but
# each participant gets only one terminal collector event.
for job_id in {binding["job_id"] for binding in prepared["manifest"]}:
q = prompt_server.distributed_pending_jobs.get(job_id)
if q is not None:
q.put_nowait({"worker_id": prepared["worker"]["id"], "is_last": True,
"serverless_error": "Serverless job setup failed; no automatic retry was attempted"})
raise
prompt_number = queue_result["number"]
node_errors = queue_result.get("node_errors", {})
trace_debug(
execution_trace_id,
f"Orchestration complete: prompt_id={prompt_id}, dispatched_workers={len(worker_payloads)}, delegate_master={delegate_master}",
)
return prompt_id, prompt_number, len(worker_payloads), node_errors
return prompt_id, prompt_number, (len(active_workers) if image_list_mode else len(worker_payloads)), node_errors
+98
View File
@@ -0,0 +1,98 @@
"""Master-only Runpod pool monitoring. Health checks never submit inference."""
from aiohttp import web
import server
from ..utils.config import load_config
from .orchestration.serverless_runtime import get_manager
from .orchestration.runpod_client import RunpodError
def _pool(worker_id):
return next((w for w in load_config().get("workers", [])
if w.get("id") == worker_id and w.get("type") == "serverless"), None)
async def _status(worker_id, fresh=False):
worker = _pool(worker_id)
if worker is None:
return web.json_response({"ok": False, "message": "Serverless worker not found"}, status=404)
jobs = []
try:
manager = get_manager()
jobs = manager.public_jobs(worker_id)
result = await manager.health(worker, fresh=fresh)
return web.json_response(dict(result, jobs=jobs))
except Exception as exc:
message = str(exc) if isinstance(exc, RunpodError) else "Check the saved endpoint ID and API key"
return web.json_response({"ok": False, "queueRemaining": 0, "message": message, "jobs": jobs})
@server.PromptServer.instance.routes.post("/distributed/serverless/test")
async def test_serverless_endpoint(request):
try:
data = await request.json()
worker_id = data.get("worker_id")
except Exception:
return web.json_response({"ok": False, "message": "Invalid request"}, status=400)
return await _status(worker_id, fresh=True)
@server.PromptServer.instance.routes.get("/distributed/serverless/status/{worker_id}")
async def serverless_status_endpoint(request):
return await _status(request.match_info["worker_id"])
@server.PromptServer.instance.routes.post("/distributed/serverless/cancel")
async def cancel_serverless_endpoint(request):
try:
data = await request.json()
worker_id = data.get("worker_id")
except Exception:
return web.json_response({"ok": False, "message": "Invalid request"}, status=400)
if _pool(worker_id) is None:
return web.json_response({"ok": False, "message": "Serverless worker not found"}, status=404)
try:
manager = get_manager()
count = await manager.cancel(worker_id=worker_id)
return web.json_response({"ok": True, "cancelled": count, "jobs": manager.public_jobs(worker_id),
"message": "Cancellation requested for this master's jobs only"})
except Exception:
return web.json_response({"ok": False, "message": "Cancellation could not be confirmed; check provider jobs"}, status=502)
async def _recover(request, *, acknowledge=False):
try:
data = await request.json()
worker_id = data.get("worker_id")
if not isinstance(worker_id, str) or not worker_id.strip():
raise ValueError()
record_id = data.get("record_id")
if acknowledge:
if not isinstance(record_id, str) or not record_id.strip() or data.get("provider_checked") is not True:
raise ValueError()
except Exception:
return web.json_response({"ok": False, "message": "Provide a worker ID; manual resolution also requires a record ID and provider_checked: true"}, status=400)
if _pool(worker_id) is None:
return web.json_response({"ok": False, "message": "Serverless worker not found"}, status=404)
try:
manager = get_manager()
if acknowledge:
await manager.acknowledge_recovery(worker_id, record_id, provider_checked=True)
result = {"message": "Recovery acknowledged by operator; not provider-confirmed cancellation"}
else:
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 worker"}, status=409)
except Exception:
return web.json_response({"ok": False, "message": "Recovery could not be confirmed; retained jobs remain protected"}, status=502)
@server.PromptServer.instance.routes.post("/distributed/serverless/reconcile")
async def reconcile_serverless_endpoint(request):
return await _recover(request)
@server.PromptServer.instance.routes.post("/distributed/serverless/resolve")
async def resolve_serverless_endpoint(request):
return await _recover(request, acknowledge=True)
+33 -1
View File
@@ -16,12 +16,14 @@ from ..utils.logging import debug_log, log
from ..utils.network import (
build_worker_url,
get_client_session,
get_server_port,
handle_api_error,
normalize_host,
probe_worker,
)
from ..utils.constants import CHUNK_SIZE
from ..workers import get_worker_manager
from ..workers.ports import allocate_worker_ports
from .schemas import require_fields, validate_worker_id
from ..workers.detection import (
get_machine_id,
@@ -59,6 +61,13 @@ async def worker_ws_endpoint(request):
})
continue
if not isinstance(data, dict):
await ws.send_json({
"type": "dispatch_ack", "request_id": None, "ok": False,
"error": "Expected a websocket message object.",
})
continue
if data.get("type") != "dispatch_prompt":
await ws.send_json({
"type": "dispatch_ack",
@@ -79,10 +88,16 @@ async def worker_ws_endpoint(request):
continue
try:
target_kwargs = {}
if "partial_execution_targets" in data:
from ..utils.async_helpers import validate_execution_targets
target_kwargs["partial_execution_targets"] = validate_execution_targets(
prompt, data["partial_execution_targets"])
prompt_id = await queue_prompt_payload(
prompt,
workflow_meta=data.get("workflow"),
client_id=data.get("client_id"),
**target_kwargs,
)
await ws.send_json({
"type": "dispatch_ack",
@@ -277,6 +292,13 @@ def _get_cuda_info():
def _collect_network_info_sync():
"""Collect network/cuda info in a worker thread to avoid blocking route handlers."""
cuda_device, cuda_device_count, physical_device_count = _get_cuda_info()
device_count = physical_device_count if physical_device_count > 0 else cuda_device_count
master_port = get_server_port()
config = load_config()
worker_ports = []
if not config.get("settings", {}).get("has_auto_populated_workers") and not config.get("workers"):
worker_count = device_count - (1 if cuda_device is not None and 0 <= cuda_device < device_count else 0)
worker_ports = allocate_worker_ports(master_port, config.get("workers", []), worker_count)
hostname = socket.gethostname()
all_ips = get_network_ips()
recommended_ip = get_recommended_ip(all_ips)
@@ -285,7 +307,9 @@ def _collect_network_info_sync():
"all_ips": all_ips,
"recommended_ip": recommended_ip,
"cuda_device": cuda_device,
"cuda_device_count": physical_device_count if physical_device_count > 0 else cuda_device_count,
"cuda_device_count": device_count,
"master_port": master_port,
"local_worker_ports": worker_ports,
}
@@ -449,6 +473,9 @@ async def launch_worker_endpoint(request):
if not worker:
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 workers are managed by Runpod, not local processes", 400)
# Ensure consistent string ID
worker_id_str = worker_id
@@ -483,6 +510,8 @@ async def launch_worker_endpoint(request):
"message": f"Worker {worker['name']} launched",
"log_file": log_file
})
except ValueError as e:
return await handle_api_error(request, f"Failed to launch worker: {str(e)}", 400)
except Exception as e:
return await handle_api_error(request, f"Failed to launch worker: {str(e)}", 500)
@@ -541,6 +570,9 @@ async def get_local_worker_status_endpoint(request):
worker_statuses = {}
for worker in config.get("workers", []):
# Serverless pools have no host, port, or local process.
if worker.get("type") == "serverless":
continue
# Only check local workers
host = normalize_host(worker.get("host")) or ""
if not host or host in ["localhost", "127.0.0.1"]: