Robert Wojciechowski
6b94242110
* feat: add Runpod serverless image collector pools * fix: prevent worker settings clipping and remove preview copy * fix: initialize worker settings before mounting the form * fix: validate complete serverless requests against 10 MB limit * fix: simplify serverless cards and keep status dots circular * fix: match serverless endpoint box to master card * fix: preserve serverless capacity and recovery across job failures Retain durable reservations until provider termination or explicit recovery, protect unresolved pools and credentials, and add scoped reconciliation. Consolidate lifecycle and secret policies and split prompt/media stages with regression coverage. * fix(serverless): disable Clear saved API key while the worker is enabled The checkbox read "Clear saved API key (disable pool first)", but the panel never defines "pool" and saving with it ticked on an enabled worker was rejected by the backend. - Disable the checkbox while the worker is enabled, with a tooltip saying to turn the worker off first. Also disable it when no key is saved. - Use "worker" instead of "pool" in user-facing serverless errors, the serverless guide and the README entry. * feat(serverless): replace Concurrent jobs with a single Images per run setting - Remove the separate Concurrent jobs field; rename Jobs per run to Images per run (1-16) with a tooltip explaining that Runpod's max workers decides how many GPUs run them. - Backend uses a fixed per-worker limit of 16 outstanding jobs, so all images of a run are submitted at once. Unresolved jobs still count against it. The limit is constant, so it never has to change while jobs are unresolved. - Legacy 'concurrency' in saved configs and older browser tabs is accepted and ignored/dropped; it no longer triggers the busy guard. - Revert the enable toggle and show the reason when the backend rejects it. - Trim a pasted api.runpod.ai/v2/<id> URL to the endpoint ID; plainer endpoint-ID error otherwise. - Confirm before a type change deletes a saved Runpod API key. - Update docs and tests.
421 lines
16 KiB
Python
421 lines
16 KiB
Python
"""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
|