Merge branch 'main' of git.stroblme.de:Fluksio/app

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01C5H4uLCCpsbipL1R7WKCee

backend/fluksio/flow/connector.py

@@ -86,7 +86,14 @@ class ConnectorNode(Node):
             description="Seconds between polls; 0 polls never.",
         )
 
-    __slots__ = ("config", "_poll_task", "_stop_event", "_last_published", "_artifacts")
+    __slots__ = (
+        "config",
+        "_poll_task",
+        "_stop_event",
+        "_last_published",
+        "_artifacts",
+        "_down",
+    )
 
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(f=self._dispatch, **kwargs)
@@ -95,6 +102,7 @@ class ConnectorNode(Node):
         self._stop_event: asyncio.Event | None = None
         self._last_published: dict[str, Any] = {}
         self._artifacts: ArtifactStore | None = None
+        self._down = False
 
     def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None:
         """The scheduler's entry point. Settings are already on ``self.config``."""
@@ -161,11 +169,7 @@ class ConnectorNode(Node):
             return
         self._stop_event.set()
         if self._poll_task is not None:
-            self._poll_task.cancel()
-            try:
-                await self._poll_task
-            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
-                pass
+            await self._cancel_task(self._poll_task)
         self._poll_task = None
         self._stop_event = None
         self._last_published = {}
@@ -175,24 +179,29 @@ class ConnectorNode(Node):
 
         Only changed ports are published: a device polled every few seconds is
         usually saying the same thing, and every publication wakes everything
-        downstream of it.
+        downstream of it. What is remembered is what was *published*, not what
+        the poll returned — a publication that raised is retried next tick
+        rather than counting as said.
         """
         while not (self._stop_event and self._stop_event.is_set()):
             try:
                 values = await self.poll()
-                self.report_health("ok")
                 changed = {
                     port: value
                     for port, value in (values or {}).items()
                     if self._last_published.get(port, object()) != value
                 }
                 if changed:
-                    self._last_published.update(changed)
                     # inject runs the graph, which is blocking work.
                     await asyncio.to_thread(self.inject, changed)
+                    self._last_published.update(changed)
+                self.report_health("ok")
+                self._down = False
             except asyncio.CancelledError:
                 break
             except Exception as exc:
-                logger.warning("Connector '%s' failed to poll: %s", self.id, exc)
+                if not self._down:
+                    logger.warning("Connector '%s' failed to poll: %s", self.id, exc)
+                self._down = True
                 self.report_health("down", f"{type(exc).__name__}: {exc}")
             await asyncio.sleep(self.config.poll_interval)

backend/fluksio/flow/controller.py

@@ -916,12 +916,12 @@ class FlowController:
         placer, pool = self.placer, self.workers
 
         def call(**kwargs: Any) -> Any:
-            with placer.claim(
+            with placer.claim(  # type: ignore[union-attr]
                 wanted, device=device, policy=policy, node=node_id, run=run_id
             ) as (target, allocation):
                 env = derive_env(wanted, allocation)
                 if target.worker is None:
-                    return pool.for_env(env).run(
+                    return pool.for_env(env).run(  # type: ignore[union-attr]
                         owner,
                         local,
                         code,
@@ -1224,8 +1224,31 @@ class FlowController:
             }
         )
 
+    def _health_issues(self, flow: str | None = None) -> list[ValidationIssue]:
+        """Nodes that are running but not working, as issues on their flow.
+
+        Not part of `self.issues`: that list is what a build found, and this is
+        what is happening now. Derived on read, so a node reporting itself well
+        again clears it with nothing to remember.
+        """
+        return [
+            ValidationIssue(
+                code="node_unhealthy",
+                message=(
+                    f"Node '{entry.id.rpartition('.')[2]}' is down: "
+                    f"{entry.health_detail or 'no detail given'}"
+                ),
+                flow=entry.flow,
+                node=entry.id,
+            )
+            for entry in self.loaded.values()
+            if entry.health == "down" and (flow is None or entry.flow == flow)
+        ]
+
     def flow_issues(self, flow: str) -> list[ValidationIssue]:
-        return [issue for issue in self.issues if not issue.flow or issue.flow == flow]
+        return [
+            issue for issue in self.issues if not issue.flow or issue.flow == flow
+        ] + self._health_issues(flow)
 
     def preview(self, name: str) -> Preview:
         """Build a flow's unpublished draft without deploying it.

backend/fluksio/flow/executor.py

@@ -46,6 +46,11 @@ DELAYED_INTERVAL_S = 1.0
 #: whose nodes wait on a network rather than a CPU may want more of them —
 #: `FLOW_MAX_CASCADES` is where that is said.
 MAX_CASCADES = 4
+#: Node threads, unless the service is given a number. Both this and the one
+#: above are taken as written: only ``None`` means "nobody said", so a number
+#: that reached here is one somebody chose, and an unusable one is the pool's
+#: ``ValueError`` rather than a silent 4.
+MAX_WORKERS = 4
 # How long a reload waits for claimed work to finish before rebuilding anyway.
 DRAIN_TIMEOUT_S = 10.0
 # Work waiting in the stream, undelivered. A burst is normal — the pool claims
@@ -68,7 +73,7 @@ class ExecutionService:
     ) -> None:
         self.queue = queue
         self._events = events
-        self.max_cascades = max_cascades or MAX_CASCADES
+        self.max_cascades = MAX_CASCADES if max_cascades is None else max_cascades
         self._pipeline: Pipeline | None = None
         self._stop = threading.Event()
         # Set when a deadline moves closer, so the timer thread stops waiting
@@ -82,7 +87,8 @@ class ExecutionService:
         # Entry ids claimed and still running, under _inflight_lock.
         self._active: set[str] = set()
         self.node_pool = ThreadPoolExecutor(
-            max_workers=max_workers or 4, thread_name_prefix="node"
+            max_workers=MAX_WORKERS if max_workers is None else max_workers,
+            thread_name_prefix="node",
         )
         self._cascade_pool = ThreadPoolExecutor(
             max_workers=self.max_cascades, thread_name_prefix="cascade"

backend/fluksio/flow/nodes/base.py

@@ -212,6 +212,22 @@ class Node:
             return None
         return asyncio.create_task(factory())
 
+    @staticmethod
+    async def _cancel_task(task: asyncio.Task[None]) -> None:
+        """Stop an unsupervised loop and wait for it to be gone.
+
+        `wait` keeps whatever the task raises on its way out to itself, and
+        lets a cancellation aimed at *this* coroutine through — the
+        `except CancelledError` around `await task` it replaces swallowed that,
+        which left whoever asked for the teardown unkillable. The same trap
+        `Supervisor._cancel` documents.
+        """
+        task.cancel()
+        await asyncio.wait([task])
+        if not task.cancelled():
+            # Retrieved so a crash on the way out is not reported at exit.
+            task.exception()
+
     def report_health(self, status: str, detail: str | None = None) -> None:
         """Say how this node's connection is doing: ok, degraded or down."""
         if self._on_health is not None:

backend/fluksio/flow/nodes/delay.py

@@ -211,11 +211,7 @@ class DelayNode(Node):
         self._stop_cron.set()
 
         if self._cron_task is not None:
-            self._cron_task.cancel()
-            try:
-                await self._cron_task
-            except asyncio.CancelledError:
-                pass
+            await self._cancel_task(self._cron_task)
 
         self._cron_task = None
         self._stop_cron = None

backend/fluksio/flow/nodes/influx.py

@@ -14,6 +14,22 @@ from fluksio.flow.nodes.base import Node, NodeResult
 logger = logging.getLogger(__name__)
 
 
+def _influxdb() -> Any:
+    """The client library, which is a `fluksio[server]` extra.
+
+    Imported per use rather than at module level, because the node type is
+    registered at boot and an installation with no InfluxDB behind it should
+    not have to carry the library to start.
+    """
+    try:
+        import influxdb_client
+    except ImportError:
+        raise RuntimeError(
+            "the influxdb node needs the server extra: pip install 'fluksio[server]'"
+        ) from None
+    return influxdb_client
+
+
 class InfluxDbNode(Node):
     """
     InfluxDB node for writing to and reading from InfluxDB.
@@ -58,6 +74,7 @@ class InfluxDbNode(Node):
         - ``bucket`` (str): Bucket name (required)
         - ``write_precision`` (str): Write precision ("ns", "us", "ms", "s"), default "ms"
         - ``query_range`` (str): Default time range for queries, e.g., "-1h", "-24h"
+        - ``timeout`` (float): Deadline for a request in seconds (default: 10.0)
         - ``writes`` (dict): Write configurations keyed by message name, each with:
             - ``measurement`` (str): Measurement name to write to
             - ``field`` (str): Field name to write (default: "value")
@@ -154,6 +171,7 @@ class InfluxDbNode(Node):
         "bucket",
         "write_precision",
         "query_range",
+        "timeout",
         "writes",
         "queries",
         "_write_client",
@@ -169,6 +187,14 @@ class InfluxDbNode(Node):
         bucket: str
         write_precision: str = "ms"
         query_range: str = "-1h"
+        # Bounds every request to the server. Without one the client falls
+        # back to its own default, which no flow can see or change. Held in
+        # seconds like every other node; the client counts in milliseconds.
+        timeout: float = Field(
+            default=10.0,
+            gt=0,
+            description="Give up on a query or write after this many seconds.",
+        )
         # Per-port write and query configuration.
         writes: dict[str, dict[str, Any]] = {}
         queries: dict[str, dict[str, Any]] = {}
@@ -201,6 +227,7 @@ class InfluxDbNode(Node):
         self.bucket = cfg.bucket
         self.write_precision = cfg.write_precision
         self.query_range = cfg.query_range
+        self.timeout = cfg.timeout
         self.writes = cfg.writes
         self.queries = cfg.queries
 
@@ -279,13 +306,18 @@ class InfluxDbNode(Node):
         :returns: ``{"rows": [...], **echo}``.
         :rtype: dict
         """
-        from influxdb_client import InfluxDBClient
+        InfluxDBClient = _influxdb().InfluxDBClient
 
         flux = str(request["flux"])
         echo = {key: value for key, value in request.items() if key != "flux"}
 
         logger.info("Running Flux for node '%s': %s", self.name, flux)
-        with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
+        with InfluxDBClient(
+            url=self.url,
+            token=self.token,
+            org=self.org,
+            timeout=int(self.timeout * 1000),
+        ) as client:
             tables = client.query_api().query(flux, org=self.org)
 
         rows = [
@@ -322,11 +354,18 @@ class InfluxDbNode(Node):
             - Dict with "value" and "tags" keys: Value written with merged tags
         :type data: dict
         """
-        from influxdb_client import InfluxDBClient, Point, WritePrecision
-        from influxdb_client.client.write_api import SYNCHRONOUS
+        influxdb_client = _influxdb()
+        InfluxDBClient = influxdb_client.InfluxDBClient
+        Point, WritePrecision = influxdb_client.Point, influxdb_client.WritePrecision
+        SYNCHRONOUS = influxdb_client.client.write_api.SYNCHRONOUS
 
         try:
-            with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
+            with InfluxDBClient(
+                url=self.url,
+                token=self.token,
+                org=self.org,
+                timeout=int(self.timeout * 1000),
+            ) as client:
                 write_api = client.write_api(write_options=SYNCHRONOUS)
 
                 precision_map = {
@@ -417,12 +456,17 @@ class InfluxDbNode(Node):
         :returns: Dict of port name to queried value.
         :rtype: dict
         """
-        from influxdb_client import InfluxDBClient
+        InfluxDBClient = _influxdb().InfluxDBClient
 
         results = {}
 
         try:
-            with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
+            with InfluxDBClient(
+                url=self.url,
+                token=self.token,
+                org=self.org,
+                timeout=int(self.timeout * 1000),
+            ) as client:
                 query_api = client.query_api()
 
                 for spec in self.output_ports:

backend/fluksio/flow/nodes/mqtt.py

@@ -19,9 +19,21 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# Deep enough to ride out a broker hiccup, shallow enough that a publisher
-# which cannot keep up drops old values instead of growing without bound.
-PUBLISH_QUEUE_SIZE = 256
+
+def _aiomqtt() -> Any:
+    """The client library, which is a `fluksio[server]` extra.
+
+    Imported per use rather than at module level, because the node type is
+    registered at boot and an installation that talks to no broker should not
+    have to carry the library to start.
+    """
+    try:
+        import aiomqtt
+    except ImportError:
+        raise RuntimeError(
+            "the mqtt node needs the server extra: pip install 'fluksio[server]'"
+        ) from None
+    return aiomqtt
 
 
 def topic_matches(filter_: str, topic: str) -> bool:
@@ -83,6 +95,9 @@ class MqttNode(Node):
         - ``qos`` (int): Quality of Service level 0, 1, or 2 (default: 0)
         - ``retain`` (bool): Retain flag for published messages (default: False)
         - ``keepalive`` (int): Keepalive interval in seconds (default: 60)
+        - ``timeout`` (float): Deadline for a broker operation in seconds
+          (default: 10.0)
+        - ``publish_queue_size`` (int): Publisher backlog depth (default: 256)
     :type params: dict
     :param name: Optional name for the node.
     :type name: str | None
@@ -151,6 +166,8 @@ class MqttNode(Node):
         "qos",
         "retain",
         "keepalive",
+        "timeout",
+        "publish_queue_size",
         "json_keys",
         "_topic_to_ports",
         "_wildcards",
@@ -175,6 +192,26 @@ class MqttNode(Node):
         qos: int = 0
         retain: bool = False
         keepalive: int = 60
+        # Bounds every broker operation: subscribe, publish, and the
+        # disconnect acknowledgement on the way out. Without one a client
+        # whose socket died waits for that ack forever, and the task never
+        # finishes unwinding. Brokers differ, so it is per node.
+        timeout: float = Field(
+            default=10.0,
+            gt=0,
+            description="Give up on a broker operation after this many seconds.",
+        )
+        # Deep enough to ride out a broker hiccup, shallow enough that a
+        # publisher which cannot keep up drops old values instead of growing
+        # without bound. A node that bursts wants more than one that trickles.
+        publish_queue_size: int = Field(
+            default=256,
+            gt=0,
+            description=(
+                "How many payloads may wait for the broker. Past this the oldest "
+                "is dropped and the node reports degraded."
+            ),
+        )
         # Which key to lift out of a JSON object payload. A device that wraps
         # its reading — Victron's ``{"value": 5}`` — is otherwise a Python node
         # per port. One key for every port, or a per-port mapping.
@@ -247,6 +284,8 @@ class MqttNode(Node):
         self.qos = cfg.qos
         self.retain = cfg.retain
         self.keepalive = cfg.keepalive
+        self.timeout = cfg.timeout
+        self.publish_queue_size = cfg.publish_queue_size
 
         # Runtime state
         self._subscription_task: asyncio.Task[None] | None = None
@@ -359,7 +398,7 @@ class MqttNode(Node):
         A dropped connection raises, and the supervisor decides when to
         reconnect — the same arrangement the subscriber uses.
         """
-        import aiomqtt
+        aiomqtt = _aiomqtt()
 
         queue = self._publish_queue
         if queue is None:
@@ -372,6 +411,7 @@ class MqttNode(Node):
             password=self.password,
             identifier=self.client_id,
             keepalive=self.keepalive,
+            timeout=self.timeout,
         ) as client:
             self.report_health("ok")
             while True:
@@ -384,7 +424,7 @@ class MqttNode(Node):
 
     async def _publish_once(self, data: dict[str, Any]) -> None:
         """Connect, publish, disconnect — the unstarted node's path."""
-        import aiomqtt
+        aiomqtt = _aiomqtt()
 
         async with aiomqtt.Client(
             hostname=self.broker_host,
@@ -393,6 +433,7 @@ class MqttNode(Node):
             password=self.password,
             identifier=self.client_id,
             keepalive=self.keepalive,
+            timeout=self.timeout,
         ) as client:
             await self._publish_with(client, data)
 
@@ -452,7 +493,7 @@ class MqttNode(Node):
         """Run the task that owns this node's connection to the broker."""
         if self._publish_queue is not None:
             return
-        self._publish_queue = asyncio.Queue(maxsize=PUBLISH_QUEUE_SIZE)
+        self._publish_queue = asyncio.Queue(maxsize=self.publish_queue_size)
         self._loop = asyncio.get_running_loop()
         self._publisher_task = self._run_supervised("mqtt-out", self._publisher_loop)
 
@@ -461,11 +502,7 @@ class MqttNode(Node):
         if self._publish_queue is None:
             return
         if self._publisher_task is not None:
-            self._publisher_task.cancel()
-            try:
-                await self._publisher_task
-            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
-                pass
+            await self._cancel_task(self._publisher_task)
         self._publisher_task = None
         self._publish_queue = None
         self._loop = None
@@ -515,11 +552,7 @@ class MqttNode(Node):
         self._stop_event.set()
 
         if self._subscription_task is not None:
-            self._subscription_task.cancel()
-            try:
-                await self._subscription_task
-            except asyncio.CancelledError:
-                pass
+            await self._cancel_task(self._subscription_task)
 
         self._subscription_task = None
         self._stop_event = None
@@ -553,7 +586,7 @@ class MqttNode(Node):
         """
         import json
 
-        import aiomqtt
+        aiomqtt = _aiomqtt()
 
         if not (self._stop_event and self._stop_event.is_set()):
             try:
@@ -564,6 +597,7 @@ class MqttNode(Node):
                     password=self.password,
                     identifier=self.client_id,
                     keepalive=self.keepalive,
+                    timeout=self.timeout,
                 ) as client:
                     # Subscribe to every unique topic
                     for topic in self._topic_to_ports:

backend/fluksio/flow/pipeline.py

@@ -59,6 +59,7 @@ class ValidationIssue(BaseModel):
         "unauthenticated_hook",
         "self_loop_needs_initial",
         "missing_source",
+        "node_unhealthy",
     ]
     message: str
     flow: str = ""

backend/fluksio/flow/placement.py

@@ -215,9 +215,18 @@ class Placer:
         said = [shape[2] for shape in capable if shape[2]]
         ram = min(ram, max(said)) if said and ram else ram
         if (cpus, gpus, ram) != (wanted.cpus, wanted.gpus, wanted.ram or 0):
+            # Cards are not detected, so a machine that has one still reports
+            # none until it is told — which reads as "no GPU here" to a node
+            # that then runs unserialised beside every other one.
+            hint = (
+                "; no machine here declares a GPU — `fluksio serve --gpus N` "
+                "(or FLOW_GPUS) says how many this one has"
+                if wanted.gpus and not gpus
+                else ""
+            )
             logger.warning(
                 "%s asked for %d cpu(s), %d gpu(s) and %s MB; "
-                "the largest machine here can give %d, %d and %s",
+                "the largest machine here can give %d, %d and %s%s",
                 node or "a node",
                 wanted.cpus,
                 wanted.gpus,
@@ -225,6 +234,7 @@ class Placer:
                 cpus,
                 gpus,
                 ram or "no stated",
+                hint,
             )
         return cpus, gpus, ram
 

backend/fluksio/flow/runs.py

@@ -769,7 +769,9 @@ class RunService:
         # the isolation it wants, minus surviving the process.
         self._state_factory = state_factory or (lambda _ns: MemoryState())
         self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64]
-        self.parallel = max(1, parallel)
+        # Taken as written: clamping a 0 up to 1 would hide a limit somebody
+        # set, and the pool below rejects an unusable one loudly anyway.
+        self.parallel = parallel
         self._pool = ThreadPoolExecutor(
             max_workers=self.parallel, thread_name_prefix="run"
         )
@@ -844,6 +846,26 @@ class RunService:
         # reference a python caller would have passed and every later reader —
         # the digest, the cache, the run detail — sees one spelling.
         params = resolve_references(flow, params, self._artifacts)
+        # What the run actually starts from, not only what was passed: an input
+        # left out takes its declared value, and a row that records `{}` cannot
+        # say which. Folded literally — an initial is a value from the
+        # definition, never a reference to resolve.
+        declared = {
+            one.spec.name: one.initial for one in flow.inputs if one.initial is not None
+        }
+        # The run's own seed fills an input of that name, outranking what the
+        # flow declares and outranked by one passed as a parameter — the order
+        # `seed_values` applies, moved to where the record is written.
+        if seed is not None and any(one.spec.name == "seed" for one in flow.inputs):
+            declared["seed"] = seed
+        params = {**declared, **params}
+        # And back the other way, so the run-level column holds the seed the
+        # run actually used however it arrived. Otherwise `--seed 1` fills one
+        # column and a declared seed the other, and that is the single field an
+        # export still has to coalesce.
+        resolved_seed = params.get("seed")
+        if isinstance(resolved_seed, int) and not isinstance(resolved_seed, bool):
+            seed = resolved_seed
         # Checked here rather than in the driver: a caller who mistyped a
         # parameter should be told now, not by a run that fails in a minute.
         seed_values(flow, params, seed)
@@ -1174,6 +1196,7 @@ class RunService:
             self._finish(run_id, status, reason, result, duration)
             run.status = status
             self._publish(run, "run_finished")
+            self._release_cards(run)
             # Its values were only ever this run's; nothing reads them once it
             # has a result. On Redis the namespace would expire anyway.
             if state is not None and status != "error":
@@ -1182,6 +1205,29 @@ class RunService:
                 except Exception:
                     logger.warning("Could not clear state of run %s", run_id)
 
+    def _release_cards(self, run: Run) -> None:
+        """Hand a GPU run's device memory back when the run is over.
+
+        The accountant frees the card the moment the node returns, but the
+        worker that ran on it is kept warm and a library that preallocated
+        most of the VRAM never gives it up — so the next process to want the
+        card found it taken by one sitting idle.
+        """
+        # ponytail: retires every CUDA pool rather than the ones this run used,
+        # which needs no bookkeeping — a concurrent GPU run's busy worker only
+        # dies when it returns, which is when its own memory should go back
+        # anyway. What it costs is the warm worker of a *live* flow's GPU node.
+        # Track the pools per run if that ever matters.
+        if not (run.needs or {}).get("gpus"):
+            return
+        pool = getattr(self.controller, "workers", None)
+        if pool is None:
+            return
+        try:
+            pool.retire_gpu_children()
+        except Exception:
+            logger.warning("Could not retire the GPU workers of run %s", run.id)
+
     def _record_node(self, run_id: str, outcome: NodeOutcome) -> None:
         outputs = _cacheable(outcome)
         row = RunNode(

backend/fluksio/flow/workers.py

@@ -321,6 +321,24 @@ class PythonWorkerPool:
         for child in children:
             child.respawn_all()
 
+    def retire_gpu_children(self) -> None:
+        """Retire the pools holding a card, so the VRAM goes back.
+
+        A library like JAX takes most of the device when it imports and never
+        releases it, so a warm worker that has run one such node is a held
+        card — and warm is the point of a pool, so nothing retires it. At the
+        end of a run there is something to key on: the environments carrying a
+        GPU assignment are exactly the pools that ran on one.
+        """
+        with self._lock:
+            children = [
+                child
+                for key, child in self._children.items()
+                if any(name == "CUDA_VISIBLE_DEVICES" for name, _ in key)
+            ]
+        for child in children:
+            child.respawn_all()
+
     def _drain(self) -> list[_Worker | None]:
         slots = []
         while True: