diff --git a/api/config_routes.py b/api/config_routes.py index 3c8e1d7..dd39fb6 100644 --- a/api/config_routes.py +++ b/api/config_routes.py @@ -226,6 +226,8 @@ async def update_setting_endpoint(request): if not key or value is None: return await handle_api_error(request, "Missing 'key' or 'value' in request", 400) + if key not in _SETTINGS_FIELDS: + return await handle_api_error(request, f"Unknown setting: {key}", 400) config = load_config() if 'settings' not in config: @@ -254,7 +256,10 @@ async def update_master_endpoint(request): if "name" in data: config['master']['name'] = data['name'] if "host" in data: - config['master']['host'] = data['host'] + if data["host"] is None: + config['master'].pop('host', None) + else: + config['master']['host'] = normalize_host(data["host"]) if "port" in data: config['master']['port'] = data['port'] if "cuda_device" in data: diff --git a/nodes/collector.py b/nodes/collector.py index e790298..3e0c205 100644 --- a/nodes/collector.py +++ b/nodes/collector.py @@ -52,7 +52,7 @@ class DistributedCollectorNode: if not multi_job_id or pass_through: if pass_through: - print(f"[Distributed Collector] Pass-through mode enabled, returning images unchanged") + debug_log("Collector: pass-through mode enabled, returning images unchanged") return (images, audio if audio is not None else empty_audio) # Use async helper to run in server loop diff --git a/nodes/queue_node.py b/nodes/queue_node.py index 539178c..d7259cc 100644 --- a/nodes/queue_node.py +++ b/nodes/queue_node.py @@ -94,8 +94,8 @@ class DistributedQueueNode: async def _fetch_worker_statuses(self, workers): session = await get_client_session() - statuses = [] - for worker in workers: + + async def _probe(worker): url = build_worker_url(worker, "/prompt") try: async with session.get( @@ -103,17 +103,18 @@ class DistributedQueueNode: timeout=aiohttp.ClientTimeout(total=3), ) as resp: if resp.status != 200: - continue + return None data = await resp.json() - queue_remaining = int(data.get("exec_info", {}).get("queue_remaining", 0)) - statuses.append({ + return { "worker": worker, - "queue_remaining": queue_remaining, - }) + "queue_remaining": int(data.get("exec_info", {}).get("queue_remaining", 0)), + } except Exception as e: debug_log(f"DistributedQueue: Worker status check failed ({url}): {e}") + return None - return statuses + results = await asyncio.gather(*[_probe(worker) for worker in workers]) + return [result for result in results if result is not None] def _select_worker(self, statuses): idle_workers = [status for status in statuses if status["queue_remaining"] == 0]