From f1aff6049d203fb3b89ee3b3864d6379b8364aab Mon Sep 17 00:00:00 2001 From: Julian Kimmig Date: Thu, 7 May 2026 09:40:38 +0200 Subject: [PATCH 1/2] feat: add single-worker Docker mode and corresponding CLI support for host and port configuration --- DOCKERFILE | 18 ++++++ docs/content/examples/docker-compose.yaml | 6 ++ docs/content/getting-started/docker.md | 56 +++++++++++++++++- pyproject.toml | 2 +- scripts/docker-entrypoint.sh | 72 +++++++++++++++++++++++ src/funcnodes/cli/parser.py | 11 ++++ src/funcnodes/cli/tasks.py | 2 + src/funcnodes/runner/_simple_server.py | 15 ++++- tests/test_main_cli.py | 67 +++++++++++++++++++++ tests/test_release_docker.py | 36 ++++++++++++ tests/test_simple_server.py | 25 +++++++- uv.lock | 2 +- 12 files changed, 307 insertions(+), 5 deletions(-) diff --git a/DOCKERFILE b/DOCKERFILE index 4c97888..c0a2096 100644 --- a/DOCKERFILE +++ b/DOCKERFILE @@ -42,6 +42,24 @@ ENV FUNCNODES_WORKER_MANAGER_PORT=9380 ENV FUNCNODES_HOST=0.0.0.0 ENV FUNCNODES_WS_WORKER_STARTPORT=9382 +# Docker startup mode. Empty/default starts the normal UI + Workermanager setup. +# Set FUNCNODES_DOCKER_MODE=single-worker to start one persistent worker and a +# frontend connected directly to it, without a Workermanager. +ENV FUNCNODES_DOCKER_MODE= + +# Defaults for the Docker-only single-worker mode. The fixed UUID keeps the +# same worker config across container restarts when FUNCNODES_CONFIG_DIR is +# mounted as a volume. +ENV FUNCNODES_SINGLE_WORKER_UUID=00000000000000000000000000000001 +ENV FUNCNODES_SINGLE_WORKER_NAME=docker-single-worker +ENV FUNCNODES_SINGLE_WORKER_HOST=0.0.0.0 +ENV FUNCNODES_SINGLE_WORKER_PORT=9382 + +# Optional browser-facing worker host for reverse proxy deployments. Leave empty +# for normal local Docker use; the frontend server can infer the request host +# for local/container bind hosts such as 0.0.0.0. +ENV FUNCNODES_SINGLE_WORKER_PUBLIC_HOST= + # Optional space-separated package list, for example: # FUNCNODES_UPDATE_PACKAGES="funcnodes-react-flow funcnodes-worker" # Leave empty for reproducible startup without network-dependent upgrades. diff --git a/docs/content/examples/docker-compose.yaml b/docs/content/examples/docker-compose.yaml index d0e4f84..29cc8ee 100644 --- a/docs/content/examples/docker-compose.yaml +++ b/docs/content/examples/docker-compose.yaml @@ -13,5 +13,11 @@ services: FUNCNODES_HOST: "0.0.0.0" FUNCNODES_WS_WORKER_STARTPORT: "9382" FUNCNODES_UPDATE_PACKAGES: "" + # Uncomment these values to run one worker and the frontend without a + # Workermanager. In that mode, port 9380 is not needed and 9382 is the + # single worker websocket port. + # FUNCNODES_DOCKER_MODE: "single-worker" + # FUNCNODES_SINGLE_WORKER_HOST: "0.0.0.0" + # FUNCNODES_SINGLE_WORKER_PORT: "9382" volumes: - ./funcnodes_config:/usr/local/app/.funcnodes diff --git a/docs/content/getting-started/docker.md b/docs/content/getting-started/docker.md index 4b8437c..5b33b21 100644 --- a/docs/content/getting-started/docker.md +++ b/docs/content/getting-started/docker.md @@ -6,7 +6,10 @@ FuncNodes publishes Docker images to GitHub Container Registry: ghcr.io/linkdlab/funcnodes ``` -Use Docker when you want to run the FuncNodes web UI and worker manager without installing Python packages on the host. +Use Docker when you want to run FuncNodes without installing Python packages on the host. The image supports two runtime modes: + +- **Manager mode**: starts the web UI and uses a Workermanager. This is the default. +- **Single-worker mode**: starts one persistent worker and a web UI connected directly to it, without a Workermanager. ## Pull the Image @@ -56,6 +59,37 @@ http://localhost:8000 The worker manager listens on port `9380`. Worker websocket ports use the exposed range `9382-9482`. +## Single-Worker Mode + +Use single-worker mode when you want one empty worker and the web frontend without a Workermanager: + +```bash +docker run --rm \ + -p 8000:8000 \ + -p 9382:9382 \ + -e FUNCNODES_DOCKER_MODE=single-worker \ + -e FUNCNODES_RUNSERVER_HOST=0.0.0.0 \ + -e FUNCNODES_RUNSERVER_PORT=8000 \ + -e FUNCNODES_SINGLE_WORKER_HOST=0.0.0.0 \ + -e FUNCNODES_SINGLE_WORKER_PORT=9382 \ + -v ./funcnodes_config:/usr/local/app/.funcnodes \ + ghcr.io/linkdlab/funcnodes:latest +``` + +The container creates the worker once and reuses it on later starts through the mounted `funcnodes_config` volume. + +Optional single-worker settings: + +| Environment variable | Default | Description | +| -------------------- | ------- | ----------- | +| `FUNCNODES_SINGLE_WORKER_UUID` | `00000000000000000000000000000001` | Stable worker id used for the persisted worker config. | +| `FUNCNODES_SINGLE_WORKER_NAME` | `docker-single-worker` | Display name for the worker. | +| `FUNCNODES_SINGLE_WORKER_HOST` | `0.0.0.0` | Bind host inside the container. | +| `FUNCNODES_SINGLE_WORKER_PORT` | `9382` | Worker websocket port. | +| `FUNCNODES_SINGLE_WORKER_PUBLIC_HOST` | empty | Browser-facing worker host for reverse proxies or remote deployments. | + +For local Docker use, leave `FUNCNODES_SINGLE_WORKER_PUBLIC_HOST` empty. Set it only when the browser must connect to a public hostname that differs from the HTTP request host. + ## Run with Docker Compose Use this `docker-compose.yaml`: @@ -80,6 +114,26 @@ services: - ./funcnodes_config:/usr/local/app/.funcnodes ``` +For single-worker mode, remove the worker-manager port and expose one worker port: + +```yaml +services: + funcnodes: + image: ghcr.io/linkdlab/funcnodes:latest + ports: + - "8000:8000" + - "9382:9382" + environment: + FUNCNODES_DOCKER_MODE: "single-worker" + FUNCNODES_RUNSERVER_HOST: "0.0.0.0" + FUNCNODES_RUNSERVER_PORT: "8000" + FUNCNODES_SINGLE_WORKER_HOST: "0.0.0.0" + FUNCNODES_SINGLE_WORKER_PORT: "9382" + FUNCNODES_UPDATE_PACKAGES: "" + volumes: + - ./funcnodes_config:/usr/local/app/.funcnodes +``` + Start the service: ```bash diff --git a/pyproject.toml b/pyproject.toml index 9daafd4..33f0cbe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "funcnodes" -version = "1.7.1" +version = "1.7.2a0" description = "funcnodes" authors = [{name = "Julian Kimmig", email = "julian.kimmig@linkdlab.de"}] diff --git a/scripts/docker-entrypoint.sh b/scripts/docker-entrypoint.sh index 25e7a00..9794f3a 100755 --- a/scripts/docker-entrypoint.sh +++ b/scripts/docker-entrypoint.sh @@ -23,6 +23,74 @@ if [ -n "${FUNCNODES_UPDATE_PACKAGES:-}" ]; then set +f fi +wait_for_worker() { + worker_host="$1" + worker_port="$2" + attempts=60 + + # The worker process is started in the background. Poll its websocket port + # before launching the frontend so the browser receives a reachable worker + # endpoint immediately. + while [ "$attempts" -gt 0 ]; do + if python -c 'import socket, sys; socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=1).close()' "$worker_host" "$worker_port"; then + return 0 + fi + + attempts=$((attempts - 1)) + sleep 2 + done + + echo "Worker did not become reachable at ${worker_host}:${worker_port}" >&2 + return 1 +} + +run_single_worker() { + worker_config="${FUNCNODES_CONFIG_DIR}/workers/worker_${FUNCNODES_SINGLE_WORKER_UUID}.json" + worker_public_host="${FUNCNODES_SINGLE_WORKER_PUBLIC_HOST:-${FUNCNODES_SINGLE_WORKER_HOST}}" + worker_connect_host="${FUNCNODES_SINGLE_WORKER_HOST}" + + # Bind hosts such as 0.0.0.0 are valid for the worker process but cannot be + # used as a client target from inside the container. Use loopback for the + # local readiness check while keeping the public host for the frontend. + if [ "$worker_connect_host" = "0.0.0.0" ] || [ "$worker_connect_host" = "::" ] || [ -z "$worker_connect_host" ]; then + worker_connect_host="127.0.0.1" + fi + + # Create the fixed worker once. `--not-in-venv` keeps the worker in the + # container environment, which is already isolated by Docker and can be + # updated at startup via FUNCNODES_UPDATE_PACKAGES. + if [ ! -f "$worker_config" ]; then + funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}" \ + --name "${FUNCNODES_SINGLE_WORKER_NAME}" \ + new --create-only --not-in-venv \ + --host "${FUNCNODES_SINGLE_WORKER_HOST}" \ + --port "${FUNCNODES_SINGLE_WORKER_PORT}" + fi + + # Start the worker as a background process. The frontend becomes PID 1 after + # exec below, and Docker will stop the whole container when it exits. + funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}" start & + worker_pid="$!" + + if ! wait_for_worker "$worker_connect_host" "${FUNCNODES_SINGLE_WORKER_PORT}"; then + kill "$worker_pid" 2>/dev/null || true + wait "$worker_pid" 2>/dev/null || true + exit 1 + fi + + # Serve the React Flow frontend without Workermanager discovery. The + # frontend reads /worker from this server and connects directly to the one + # worker started above. + exec funcnodes runserver \ + --host "${FUNCNODES_RUNSERVER_HOST}" \ + --port "${FUNCNODES_RUNSERVER_PORT}" \ + --no-browser \ + --no-manager \ + --worker_host "$worker_public_host" \ + --worker_port "${FUNCNODES_SINGLE_WORKER_PORT}" \ + "$@" +} + # Default command path. # # `docker run image` and `docker run image runserver` both start the FuncNodes @@ -33,6 +101,10 @@ if [ "$#" -eq 0 ] || [ "$1" = "runserver" ]; then shift fi + if [ "${FUNCNODES_DOCKER_MODE:-manager}" = "single-worker" ]; then + run_single_worker "$@" + fi + exec funcnodes runserver \ --host "${FUNCNODES_RUNSERVER_HOST}" \ --port "${FUNCNODES_RUNSERVER_PORT}" \ diff --git a/src/funcnodes/cli/parser.py b/src/funcnodes/cli/parser.py index 2c5c0bb..335122f 100644 --- a/src/funcnodes/cli/parser.py +++ b/src/funcnodes/cli/parser.py @@ -220,6 +220,17 @@ def add_worker_parser(subparsers): new_worker_parser.add_argument( "--not-in-venv", action="store_false", dest="in_venv", help="Do not use a venv" ) + new_worker_parser.add_argument( + "--host", + default=None, + help="The host to bind the new worker to", + ) + new_worker_parser.add_argument( + "--port", + default=None, + type=int, + help="The port to bind the new worker to", + ) autostart_group = new_worker_parser.add_mutually_exclusive_group() autostart_group.add_argument( "--autostart", diff --git a/src/funcnodes/cli/tasks.py b/src/funcnodes/cli/tasks.py index 591c0f1..a78fd94 100644 --- a/src/funcnodes/cli/tasks.py +++ b/src/funcnodes/cli/tasks.py @@ -156,6 +156,8 @@ def task_worker(args: argparse.Namespace): create_only=args.create_only, profile=args.profile, autostart_policy=args.autostart_policy, + host=getattr(args, "host", None), + port=getattr(args, "port", None), ) elif workertask == "list": return list_workers(args) diff --git a/src/funcnodes/runner/_simple_server.py b/src/funcnodes/runner/_simple_server.py index 3faf631..80e9986 100644 --- a/src/funcnodes/runner/_simple_server.py +++ b/src/funcnodes/runner/_simple_server.py @@ -59,6 +59,15 @@ def _public_worker_manager_host( return _strip_port_from_host(request_host) +def _public_worker_host( + configured_host: Optional[str], request_host: Optional[str] +) -> str: + return _public_worker_manager_host( + configured_host=configured_host, + request_host=request_host, + ) + + class Methods(Enum): GET = "GET" DELETE = "DELETE" @@ -281,9 +290,13 @@ def run(self, loop=None, **kwargs): loop.run_until_complete(self.shutdown()) async def get_worker(self, request): + public_host = _public_worker_host( + configured_host=self.worker_host, + request_host=request.host, + ) return web.json_response( data={ - "host": self.worker_host, + "host": public_host, "port": self.worker_port, "ssl": self.worker_ssl, }, diff --git a/tests/test_main_cli.py b/tests/test_main_cli.py index ed01258..d7e7650 100644 --- a/tests/test_main_cli.py +++ b/tests/test_main_cli.py @@ -281,6 +281,73 @@ def test_add_worker_parser_parses_new_autostart_policy(): assert args.autostart_policy == "always" +@pytest_funcnodes.funcnodes_test +def test_add_worker_parser_parses_new_worker_host_and_port(): + from funcnodes.__main__ import add_worker_parser + + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="task", required=True) + add_worker_parser(subparsers) + + args = parser.parse_args( + [ + "worker", + "--uuid", + "worker-docker", + "--name", + "docker", + "new", + "--host", + "0.0.0.0", + "--port", + "9382", + ] + ) + + assert args.task == "worker" + assert args.workertask == "new" + assert args.uuid == "worker-docker" + assert args.name == "docker" + assert args.host == "0.0.0.0" + assert args.port == 9382 + + +@pytest_funcnodes.funcnodes_test +def test_task_worker_new_passes_host_and_port(monkeypatch): + from funcnodes.cli import tasks as tasks_mod + from funcnodes.__main__ import task_worker + + captured: dict = {} + + def fake_start_new_worker(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr(tasks_mod, "start_new_worker", fake_start_new_worker) + + args = SimpleNamespace( + workertask="new", + uuid="worker-docker", + name="docker", + workertype="WSWorker", + debug=False, + in_venv=False, + create_only=True, + profile=False, + autostart_policy="never", + host="0.0.0.0", + port=9382, + ) + + task_worker(args) + + assert captured["uuid"] == "worker-docker" + assert captured["name"] == "docker" + assert captured["host"] == "0.0.0.0" + assert captured["port"] == 9382 + assert captured["in_venv"] is False + assert captured["create_only"] is True + + @pytest_funcnodes.funcnodes_test def test_add_worker_parser_rejects_autostart_flag_and_policy_together(): from funcnodes.__main__ import add_worker_parser diff --git a/tests/test_release_docker.py b/tests/test_release_docker.py index ee703fc..5aa9666 100644 --- a/tests/test_release_docker.py +++ b/tests/test_release_docker.py @@ -48,6 +48,42 @@ def test_docker_entrypoint_updates_configured_packages_before_startup(): assert 'exec "$@"' in entrypoint +def test_dockerfile_declares_single_worker_mode_defaults(): + dockerfile = (REPO_ROOT / "DOCKERFILE").read_text(encoding="utf-8") + + assert "ENV FUNCNODES_DOCKER_MODE=" in dockerfile + assert ( + "ENV FUNCNODES_SINGLE_WORKER_UUID=00000000000000000000000000000001" + in dockerfile + ) + assert "ENV FUNCNODES_SINGLE_WORKER_NAME=docker-single-worker" in dockerfile + assert "ENV FUNCNODES_SINGLE_WORKER_HOST=0.0.0.0" in dockerfile + assert "ENV FUNCNODES_SINGLE_WORKER_PORT=9382" in dockerfile + assert "ENV FUNCNODES_SINGLE_WORKER_PUBLIC_HOST=" in dockerfile + + +def test_docker_entrypoint_supports_single_worker_mode(): + entrypoint = (REPO_ROOT / "scripts" / "docker-entrypoint.sh").read_text( + encoding="utf-8" + ) + + assert "FUNCNODES_DOCKER_MODE:-manager" in entrypoint + assert "single-worker" in entrypoint + assert 'funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}"' in entrypoint + assert "new --create-only --not-in-venv" in entrypoint + assert '--host "${FUNCNODES_SINGLE_WORKER_HOST}"' in entrypoint + assert '--port "${FUNCNODES_SINGLE_WORKER_PORT}"' in entrypoint + assert ( + 'funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}" start &' + in entrypoint + ) + assert 'worker_connect_host="127.0.0.1"' in entrypoint + assert 'wait_for_worker "$worker_connect_host"' in entrypoint + assert "--no-manager" in entrypoint + assert '--worker_host "$worker_public_host"' in entrypoint + assert '--worker_port "${FUNCNODES_SINGLE_WORKER_PORT}"' in entrypoint + + def test_release_workflow_pushes_versioned_and_latest_docker_image(): workflow = ( REPO_ROOT / ".github" / "workflows" / "version_publish_main.yml" diff --git a/tests/test_simple_server.py b/tests/test_simple_server.py index 5c6352c..307a4f3 100644 --- a/tests/test_simple_server.py +++ b/tests/test_simple_server.py @@ -1,4 +1,7 @@ -from funcnodes.runner._simple_server import _public_worker_manager_host +from funcnodes.runner._simple_server import ( + _public_worker_host, + _public_worker_manager_host, +) def test_public_worker_manager_host_uses_request_host_for_localhost(): @@ -39,3 +42,23 @@ def test_public_worker_manager_host_handles_ipv6_request_host(): ) == "[2001:db8::1]" ) + + +def test_public_worker_host_uses_request_host_for_container_bind_host(): + assert ( + _public_worker_host( + configured_host="0.0.0.0", + request_host="funcnodes.demoserver.xyz:8001", + ) + == "funcnodes.demoserver.xyz" + ) + + +def test_public_worker_host_keeps_explicit_public_host(): + assert ( + _public_worker_host( + configured_host="workers.example.org", + request_host="funcnodes.demoserver.xyz:8001", + ) + == "workers.example.org" + ) diff --git a/uv.lock b/uv.lock index e4ecae7..0689260 100644 --- a/uv.lock +++ b/uv.lock @@ -792,7 +792,7 @@ wheels = [ [[package]] name = "funcnodes" -version = "1.7.1" +version = "1.7.2a0" source = { editable = "." } dependencies = [ { name = "funcnodes-basic" }, From 644b15250faf17f9ec7180f69fe646c790f8290d Mon Sep 17 00:00:00 2001 From: Julian Kimmig Date: Thu, 7 May 2026 11:43:33 +0200 Subject: [PATCH 2/2] refactor: integrate worker lifecycle management directly into runserver via --worker-uuid to simplify docker-compose startup. --- docs/content/examples/docker-compose.yaml | 4 +- docs/content/getting-started/docker.md | 2 +- scripts/docker-entrypoint.sh | 47 +--- src/funcnodes/cli/parser.py | 8 + src/funcnodes/cli/tasks.py | 131 +++++++++-- src/funcnodes/worker/worker_manager.py | 13 ++ tests/test_main_cli.py | 261 ++++++++++++++++++++++ tests/test_release_docker.py | 5 +- 8 files changed, 409 insertions(+), 62 deletions(-) diff --git a/docs/content/examples/docker-compose.yaml b/docs/content/examples/docker-compose.yaml index 29cc8ee..e6a2534 100644 --- a/docs/content/examples/docker-compose.yaml +++ b/docs/content/examples/docker-compose.yaml @@ -14,8 +14,8 @@ services: FUNCNODES_WS_WORKER_STARTPORT: "9382" FUNCNODES_UPDATE_PACKAGES: "" # Uncomment these values to run one worker and the frontend without a - # Workermanager. In that mode, port 9380 is not needed and 9382 is the - # single worker websocket port. + # Workermanager. In that mode, runserver attaches to the worker by UUID, + # port 9380 is not needed, and 9382 is the single worker websocket port. # FUNCNODES_DOCKER_MODE: "single-worker" # FUNCNODES_SINGLE_WORKER_HOST: "0.0.0.0" # FUNCNODES_SINGLE_WORKER_PORT: "9382" diff --git a/docs/content/getting-started/docker.md b/docs/content/getting-started/docker.md index 5b33b21..4b31d63 100644 --- a/docs/content/getting-started/docker.md +++ b/docs/content/getting-started/docker.md @@ -76,7 +76,7 @@ docker run --rm \ ghcr.io/linkdlab/funcnodes:latest ``` -The container creates the worker once and reuses it on later starts through the mounted `funcnodes_config` volume. +The container creates the worker once and reuses it on later starts through the mounted `funcnodes_config` volume. Startup uses `funcnodes runserver --no-manager --worker-uuid ...`: if the worker is already running, the frontend attaches to its existing port; otherwise the server starts it on `FUNCNODES_SINGLE_WORKER_PORT` and stops it again when the server exits. Optional single-worker settings: diff --git a/scripts/docker-entrypoint.sh b/scripts/docker-entrypoint.sh index 9794f3a..10ee2e6 100755 --- a/scripts/docker-entrypoint.sh +++ b/scripts/docker-entrypoint.sh @@ -23,38 +23,9 @@ if [ -n "${FUNCNODES_UPDATE_PACKAGES:-}" ]; then set +f fi -wait_for_worker() { - worker_host="$1" - worker_port="$2" - attempts=60 - - # The worker process is started in the background. Poll its websocket port - # before launching the frontend so the browser receives a reachable worker - # endpoint immediately. - while [ "$attempts" -gt 0 ]; do - if python -c 'import socket, sys; socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=1).close()' "$worker_host" "$worker_port"; then - return 0 - fi - - attempts=$((attempts - 1)) - sleep 2 - done - - echo "Worker did not become reachable at ${worker_host}:${worker_port}" >&2 - return 1 -} - run_single_worker() { worker_config="${FUNCNODES_CONFIG_DIR}/workers/worker_${FUNCNODES_SINGLE_WORKER_UUID}.json" worker_public_host="${FUNCNODES_SINGLE_WORKER_PUBLIC_HOST:-${FUNCNODES_SINGLE_WORKER_HOST}}" - worker_connect_host="${FUNCNODES_SINGLE_WORKER_HOST}" - - # Bind hosts such as 0.0.0.0 are valid for the worker process but cannot be - # used as a client target from inside the container. Use loopback for the - # local readiness check while keeping the public host for the frontend. - if [ "$worker_connect_host" = "0.0.0.0" ] || [ "$worker_connect_host" = "::" ] || [ -z "$worker_connect_host" ]; then - worker_connect_host="127.0.0.1" - fi # Create the fixed worker once. `--not-in-venv` keeps the worker in the # container environment, which is already isolated by Docker and can be @@ -67,25 +38,15 @@ run_single_worker() { --port "${FUNCNODES_SINGLE_WORKER_PORT}" fi - # Start the worker as a background process. The frontend becomes PID 1 after - # exec below, and Docker will stop the whole container when it exits. - funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}" start & - worker_pid="$!" - - if ! wait_for_worker "$worker_connect_host" "${FUNCNODES_SINGLE_WORKER_PORT}"; then - kill "$worker_pid" 2>/dev/null || true - wait "$worker_pid" 2>/dev/null || true - exit 1 - fi - - # Serve the React Flow frontend without Workermanager discovery. The - # frontend reads /worker from this server and connects directly to the one - # worker started above. + # Serve the React Flow frontend without Workermanager discovery. `runserver` + # attaches to the configured worker if it is already running; otherwise it + # starts it with the configured port and stops it during server shutdown. exec funcnodes runserver \ --host "${FUNCNODES_RUNSERVER_HOST}" \ --port "${FUNCNODES_RUNSERVER_PORT}" \ --no-browser \ --no-manager \ + --worker-uuid "${FUNCNODES_SINGLE_WORKER_UUID}" \ --worker_host "$worker_public_host" \ --worker_port "${FUNCNODES_SINGLE_WORKER_PORT}" \ "$@" diff --git a/src/funcnodes/cli/parser.py b/src/funcnodes/cli/parser.py index 335122f..99b5eae 100644 --- a/src/funcnodes/cli/parser.py +++ b/src/funcnodes/cli/parser.py @@ -81,6 +81,14 @@ def add_runserver_parser(subparsers): type=int, help="The port to run the worker on", ) + parser.add_argument( + "--worker-uuid", + default=None, + help=( + "Existing worker UUID to use in --no-manager mode. If the worker is " + "not running, runserver starts it and stops it on shutdown." + ), + ) parser.add_argument( "--worker_ssl", action="store_true", diff --git a/src/funcnodes/cli/tasks.py b/src/funcnodes/cli/tasks.py index a78fd94..681dd89 100644 --- a/src/funcnodes/cli/tasks.py +++ b/src/funcnodes/cli/tasks.py @@ -4,11 +4,13 @@ import asyncio import textwrap import threading +import time from typing import Optional from pathlib import Path import funcnodes as fn +from funcnodes_core.utils.files import write_json_secure from .utils import parse_command_kwargs from .worker import ( @@ -29,6 +31,83 @@ # ============================================================================= +def _get_runserver_worker_config(worker_uuid: str, debug: bool): + """Return the manager and worker config for a direct runserver worker.""" + manager = fn.worker.worker_manager.WorkerManager(debug=debug) + for worker_config in manager.get_all_workercfg(): + if worker_config["uuid"] == worker_uuid: + return manager, worker_config + + raise ValueError(f"No worker found with uuid {worker_uuid!r}") + + +def _write_runserver_worker_config(manager, worker_config): + """Persist direct worker host/port changes before starting the worker.""" + worker_config_file = ( + Path(manager.worker_dir) / f"worker_{worker_config['uuid']}.json" + ) + write_json_secure(worker_config, worker_config_file, indent=2) + + +def _check_runserver_worker(worker_config) -> bool: + """Check whether the configured worker websocket is reachable.""" + _, is_running = asyncio.run(fn.worker.worker_manager.check_worker(worker_config)) + return is_running + + +def _wait_for_runserver_worker(worker_uuid: str, debug: bool, timeout: float = 30.0): + """Wait for a worker started by runserver to publish a reachable endpoint.""" + deadline = time.time() + timeout + + while time.time() < deadline: + _, worker_config = _get_runserver_worker_config(worker_uuid, debug=debug) + if _check_runserver_worker(worker_config): + return worker_config + time.sleep(0.5) + + raise TimeoutError(f"Worker {worker_uuid!r} did not become reachable") + + +def _stop_runserver_worker(worker_uuid: str, debug: bool): + """Stop a worker that was started for a direct runserver session.""" + manager = fn.worker.worker_manager.WorkerManager(debug=debug) + asyncio.run(manager.stop_worker(worker_uuid)) + + +def _prepare_direct_worker_for_runserver(args: argparse.Namespace): + """Attach to or start the worker requested by runserver --worker-uuid.""" + worker_uuid = getattr(args, "worker_uuid", None) + if not worker_uuid: + return None, False + + if getattr(args, "no_manager", True): + raise ValueError("--worker-uuid requires --no-manager") + + debug = getattr(args, "debug", False) + manager, worker_config = _get_runserver_worker_config(worker_uuid, debug=debug) + + if _check_runserver_worker(worker_config): + return worker_config, False + + worker_config["host"] = ( + worker_config.get("host") or getattr(args, "worker_host", None) or "localhost" + ) + worker_config["port"] = getattr(args, "worker_port", None) or worker_config.get( + "port", 9380 + ) + worker_config["ssl"] = getattr(args, "worker_ssl", False) + worker_config.pop("pid", None) + _write_runserver_worker_config(manager, worker_config) + + fn.worker.worker_manager.start_worker(worker_config, debug=debug) + try: + worker_config = _wait_for_runserver_worker(worker_uuid, debug=debug) + except Exception: + _stop_runserver_worker(worker_uuid, debug=debug) + raise + return worker_config, True + + def task_run_server(args: argparse.Namespace): """Run the FuncNodes server with the specified frontend.""" frontend = args.frontend @@ -37,21 +116,47 @@ def task_run_server(args: argparse.Namespace): else: raise Exception(f"Unknown frontend: {frontend}") - run_server( - port=args.port, - host=args.host, - open_browser=args.no_browser, - worker_manager_host=args.worker_manager_host, - worker_manager_port=args.worker_manager_port, - worker_manager_ssl=args.worker_manager_ssl, - start_worker_manager=args.no_manager, - has_worker_manager=args.no_manager, - worker_host=args.worker_host, - worker_port=args.worker_port, - worker_ssl=args.worker_ssl, - debug=args.debug, + direct_worker_config, started_direct_worker = _prepare_direct_worker_for_runserver( + args ) + worker_host = args.worker_host + worker_port = args.worker_port + worker_ssl = args.worker_ssl + shutdown_handler_callback = None + + if direct_worker_config is not None: + worker_host = direct_worker_config.get("host") + worker_port = direct_worker_config.get("port") + worker_ssl = direct_worker_config.get("ssl", False) + if args.worker_host and args.worker_host != "localhost": + worker_host = args.worker_host + + def _direct_worker_shutdown_handler(handler): + return handler + + shutdown_handler_callback = _direct_worker_shutdown_handler + + try: + run_server( + port=args.port, + host=args.host, + open_browser=args.no_browser, + worker_manager_host=args.worker_manager_host, + worker_manager_port=args.worker_manager_port, + worker_manager_ssl=args.worker_manager_ssl, + start_worker_manager=args.no_manager, + has_worker_manager=args.no_manager, + worker_host=worker_host, + worker_port=worker_port, + worker_ssl=worker_ssl, + debug=args.debug, + register_shutdown_handler=shutdown_handler_callback, + ) + finally: + if started_direct_worker and direct_worker_config is not None: + _stop_runserver_worker(direct_worker_config["uuid"], debug=args.debug) + def task_standalone(args: argparse.Namespace): """Run a standalone .fnw file with its own worker.""" diff --git a/src/funcnodes/worker/worker_manager.py b/src/funcnodes/worker/worker_manager.py index 7a4214e..ed3e664 100644 --- a/src/funcnodes/worker/worker_manager.py +++ b/src/funcnodes/worker/worker_manager.py @@ -1253,6 +1253,10 @@ async def stop_worker(self, workerid, websocket: web.WebSocketResponse = None): """ Stops a worker. + The target is resolved from the manager's active/inactive caches first. + If this manager instance has not loaded workers yet, the persisted worker + JSON is used as a fallback so CLI commands can stop a known worker UUID. + Args: workerid (str): The id of the worker to stop. websocket (WebSocketResponse): The websocket connection to send status updates to. @@ -1276,6 +1280,15 @@ async def stop_worker(self, workerid, websocket: web.WebSocketResponse = None): target_worker = worker break + if target_worker is None: + jsonfilepath = os.path.join(self.worker_dir, f"worker_{workerid}.json") + if os.path.exists(jsonfilepath): + try: + with open(jsonfilepath, "r", encoding="utf-8") as file: + target_worker = WorkerJson(**json.load(file)) + except Exception: + target_worker = None + if target_worker is None: return diff --git a/tests/test_main_cli.py b/tests/test_main_cli.py index d7e7650..508ee86 100644 --- a/tests/test_main_cli.py +++ b/tests/test_main_cli.py @@ -76,6 +76,264 @@ def run_server(**kwargs): assert captured["debug"] is True +@pytest_funcnodes.funcnodes_test +def test_task_run_server_attaches_to_running_worker(monkeypatch, worker_config): + from funcnodes.__main__ import task_run_server + from funcnodes.worker import worker_manager as worker_manager_mod + + worker_config["host"] = "existing.local" + worker_config["port"] = 9399 + worker_dir = Path(worker_manager_mod.WorkerManager(debug=False).worker_dir) + worker_file = worker_dir / f"worker_{worker_config['uuid']}.json" + worker_file.write_text(json.dumps(worker_config), encoding="utf-8") + + captured: dict = {} + module = ModuleType("funcnodes_react_flow") + + def run_server(**kwargs): + captured.update(kwargs) + + async def fake_check_worker(config): + return config["uuid"], True + + def fail_start_worker(config, debug=False): # pragma: no cover - assertion guard + raise AssertionError("running worker must not be started") + + async def fail_stop_worker(self, workerid, websocket=None): + raise AssertionError("attached worker must not be stopped") + + module.run_server = run_server + monkeypatch.setitem(sys.modules, "funcnodes_react_flow", module) + monkeypatch.setattr(worker_manager_mod, "check_worker", fake_check_worker) + monkeypatch.setattr(worker_manager_mod, "start_worker", fail_start_worker) + monkeypatch.setattr( + worker_manager_mod.WorkerManager, + "stop_worker", + fail_stop_worker, + ) + + args = SimpleNamespace( + frontend="react_flow", + port=9001, + host="127.0.0.1", + no_browser=False, + worker_manager_host=None, + worker_manager_port=None, + worker_manager_ssl=False, + no_manager=False, + worker_host="localhost", + worker_port=9200, + worker_ssl=False, + worker_uuid=worker_config["uuid"], + debug=True, + ) + + task_run_server(args) + + assert captured["has_worker_manager"] is False + assert captured["start_worker_manager"] is False + assert captured["worker_host"] == "existing.local" + assert captured["worker_port"] == 9399 + + +@pytest_funcnodes.funcnodes_test +def test_task_run_server_starts_and_stops_stopped_worker(monkeypatch, worker_config): + from funcnodes.__main__ import task_run_server + from funcnodes.worker import worker_manager as worker_manager_mod + + captured: dict = {} + started: list[dict] = [] + stopped: list[str] = [] + module = ModuleType("funcnodes_react_flow") + + def run_server(**kwargs): + captured.update(kwargs) + + async def fake_check_worker(config): + return config["uuid"], bool(started) + + def fake_start_worker(config, debug=False): + started.append(dict(config)) + + async def fake_stop_worker(self, workerid, websocket=None): + stopped.append(workerid) + + module.run_server = run_server + monkeypatch.setitem(sys.modules, "funcnodes_react_flow", module) + monkeypatch.setattr(worker_manager_mod, "check_worker", fake_check_worker) + monkeypatch.setattr(worker_manager_mod, "start_worker", fake_start_worker) + monkeypatch.setattr( + worker_manager_mod.WorkerManager, + "stop_worker", + fake_stop_worker, + ) + + args = SimpleNamespace( + frontend="react_flow", + port=9001, + host="127.0.0.1", + no_browser=False, + worker_manager_host=None, + worker_manager_port=None, + worker_manager_ssl=False, + no_manager=False, + worker_host="requested.local", + worker_port=9200, + worker_ssl=True, + worker_uuid=worker_config["uuid"], + debug=True, + ) + + task_run_server(args) + + assert started[0]["host"] == "requested.local" + assert started[0]["port"] == 9200 + assert "register_shutdown_handler" in captured + assert captured["worker_host"] == "requested.local" + assert captured["worker_port"] == 9200 + assert captured["worker_ssl"] is True + assert stopped == [worker_config["uuid"]] + + +@pytest_funcnodes.funcnodes_test +def test_task_run_server_preserves_worker_bind_host_for_public_host_override( + monkeypatch, worker_config +): + from funcnodes.__main__ import task_run_server + from funcnodes.worker import worker_manager as worker_manager_mod + + worker_config["host"] = "0.0.0.0" + worker_dir = Path(worker_manager_mod.WorkerManager(debug=False).worker_dir) + worker_file = worker_dir / f"worker_{worker_config['uuid']}.json" + worker_file.write_text(json.dumps(worker_config), encoding="utf-8") + + captured: dict = {} + started: list[dict] = [] + stopped: list[str] = [] + module = ModuleType("funcnodes_react_flow") + + def run_server(**kwargs): + captured.update(kwargs) + + async def fake_check_worker(config): + return config["uuid"], bool(started) + + def fake_start_worker(config, debug=False): + started.append(dict(config)) + + async def fake_stop_worker(self, workerid, websocket=None): + stopped.append(workerid) + + module.run_server = run_server + monkeypatch.setitem(sys.modules, "funcnodes_react_flow", module) + monkeypatch.setattr(worker_manager_mod, "check_worker", fake_check_worker) + monkeypatch.setattr(worker_manager_mod, "start_worker", fake_start_worker) + monkeypatch.setattr( + worker_manager_mod.WorkerManager, + "stop_worker", + fake_stop_worker, + ) + + args = SimpleNamespace( + frontend="react_flow", + port=9001, + host="127.0.0.1", + no_browser=False, + worker_manager_host=None, + worker_manager_port=None, + worker_manager_ssl=False, + no_manager=False, + worker_host="worker.example.com", + worker_port=9200, + worker_ssl=False, + worker_uuid=worker_config["uuid"], + debug=True, + ) + + task_run_server(args) + + assert started[0]["host"] == "0.0.0.0" + assert captured["worker_host"] == "worker.example.com" + assert stopped == [worker_config["uuid"]] + + +@pytest_funcnodes.funcnodes_test +def test_task_run_server_stops_worker_when_startup_check_fails( + monkeypatch, worker_config +): + from funcnodes.cli import tasks as tasks_mod + from funcnodes.worker import worker_manager as worker_manager_mod + + started: list[dict] = [] + stopped: list[str] = [] + + async def fake_check_worker(config): + return config["uuid"], False + + def fake_start_worker(config, debug=False): + started.append(dict(config)) + + async def fake_stop_worker(self, workerid, websocket=None): + stopped.append(workerid) + + ticks = iter([0, 31]) + + monkeypatch.setattr(worker_manager_mod, "check_worker", fake_check_worker) + monkeypatch.setattr(worker_manager_mod, "start_worker", fake_start_worker) + monkeypatch.setattr( + worker_manager_mod.WorkerManager, + "stop_worker", + fake_stop_worker, + ) + monkeypatch.setattr(tasks_mod.time, "time", lambda: next(ticks)) + monkeypatch.setattr(tasks_mod.time, "sleep", lambda delay: None) + + args = SimpleNamespace( + frontend="react_flow", + no_manager=False, + worker_host="localhost", + worker_port=9200, + worker_ssl=False, + worker_uuid=worker_config["uuid"], + debug=True, + ) + + with pytest.raises(TimeoutError, match="did not become reachable"): + tasks_mod.task_run_server(args) + + assert started + assert stopped == [worker_config["uuid"]] + + +@pytest_funcnodes.funcnodes_test +def test_task_run_server_worker_uuid_requires_no_manager(worker_config): + from funcnodes.__main__ import task_run_server + + args = SimpleNamespace( + frontend="react_flow", + no_manager=True, + worker_uuid=worker_config["uuid"], + ) + + with pytest.raises(ValueError, match="--worker-uuid requires --no-manager"): + task_run_server(args) + + +@pytest_funcnodes.funcnodes_test +def test_task_run_server_unknown_worker_uuid_raises(): + from funcnodes.__main__ import task_run_server + + args = SimpleNamespace( + frontend="react_flow", + no_manager=False, + worker_uuid="missing-worker", + debug=False, + ) + + with pytest.raises(ValueError, match="No worker found"): + task_run_server(args) + + @pytest_funcnodes.funcnodes_test def test_task_run_server_rejects_unknown_frontend(): from funcnodes.__main__ import task_run_server @@ -111,6 +369,8 @@ def test_add_runserver_parser_parses_flags(): "worker", "--worker_port", "9201", + "--worker-uuid", + "worker-123", "--worker_ssl", ] ) @@ -125,6 +385,7 @@ def test_add_runserver_parser_parses_flags(): assert args.worker_manager_ssl is True assert args.worker_host == "worker" assert args.worker_port == 9201 + assert args.worker_uuid == "worker-123" assert args.worker_ssl is True diff --git a/tests/test_release_docker.py b/tests/test_release_docker.py index 5aa9666..f7aec07 100644 --- a/tests/test_release_docker.py +++ b/tests/test_release_docker.py @@ -75,11 +75,10 @@ def test_docker_entrypoint_supports_single_worker_mode(): assert '--port "${FUNCNODES_SINGLE_WORKER_PORT}"' in entrypoint assert ( 'funcnodes worker --uuid "${FUNCNODES_SINGLE_WORKER_UUID}" start &' - in entrypoint + not in entrypoint ) - assert 'worker_connect_host="127.0.0.1"' in entrypoint - assert 'wait_for_worker "$worker_connect_host"' in entrypoint assert "--no-manager" in entrypoint + assert '--worker-uuid "${FUNCNODES_SINGLE_WORKER_UUID}"' in entrypoint assert '--worker_host "$worker_public_host"' in entrypoint assert '--worker_port "${FUNCNODES_SINGLE_WORKER_PORT}"' in entrypoint