diff --git a/backend/fluksio/flow/nodes/influx.py b/backend/fluksio/flow/nodes/influx.py index 3b192a0..c0018f8 100644 --- a/backend/fluksio/flow/nodes/influx.py +++ b/backend/fluksio/flow/nodes/influx.py @@ -58,6 +58,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 +155,7 @@ class InfluxDbNode(Node): "bucket", "write_precision", "query_range", + "timeout", "writes", "queries", "_write_client", @@ -169,6 +171,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 +211,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 @@ -285,7 +296,12 @@ class InfluxDbNode(Node): 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 = [ @@ -326,7 +342,12 @@ class InfluxDbNode(Node): from influxdb_client.client.write_api import 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 = { @@ -422,7 +443,12 @@ class InfluxDbNode(Node): 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: diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index 499ac2b..f6b8e35 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -19,10 +19,6 @@ 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 topic_matches(filter_: str, topic: str) -> bool: """Does an MQTT topic filter cover this topic? @@ -85,6 +81,7 @@ class MqttNode(Node): - ``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 @@ -154,6 +151,7 @@ class MqttNode(Node): "retain", "keepalive", "timeout", + "publish_queue_size", "json_keys", "_topic_to_ports", "_wildcards", @@ -187,6 +185,17 @@ class MqttNode(Node): 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. @@ -260,6 +269,7 @@ class MqttNode(Node): 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 @@ -467,7 +477,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) diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index 7cc1dc1..056593c 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -197,6 +197,47 @@ def test_a_flux_request_is_run_rather_than_written(monkeypatch): assert out == {"answer": {"rows": [], "range_s": 3600}} +def test_the_configured_timeout_reaches_the_influx_client(monkeypatch): + """The client counts in milliseconds; the param is seconds like its peers.""" + import influxdb_client + + from fluksio.flow.nodes import InfluxDbNode + + seen: dict = {} + + class FakeClient: + def __init__(self, **kwargs): + seen.update(kwargs) + + def __enter__(self): + return self + + def __exit__(self, *_): + return False + + def query_api(self): + return self + + def query(self, *_, **__): + return [] + + monkeypatch.setattr(influxdb_client, "InfluxDBClient", FakeClient) + + node = InfluxDbNode( + provides=[MessageSpec(name="answer", dtype=DType.JSON)], + params={ + "url": "http://influx", + "token": "t", + "org": "o", + "bucket": "b", + "timeout": 2.5, + }, + ) + node._run_flux({"flux": 'from(bucket: "b")'}) + + assert seen["timeout"] == 2500 + + def test_a_falsy_return_is_a_mistake_not_silence(): """Only None means "nothing to publish".""" from fluksio.flow.nodes import Node diff --git a/backend/tests/flow/test_senders.py b/backend/tests/flow/test_senders.py index 953d246..655066a 100644 --- a/backend/tests/flow/test_senders.py +++ b/backend/tests/flow/test_senders.py @@ -119,3 +119,35 @@ def test_the_configured_timeout_reaches_the_broker_client(monkeypatch): asyncio.run(node._publish_once({"setpoint": 21.0})) assert seen["timeout"] == 2.5 + + +def test_the_configured_backlog_reaches_the_publish_queue(monkeypatch): + """The depth is read when the queue is built, so it has to be per node.""" + import aiomqtt + + class FakeClient: + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + monkeypatch.setattr(aiomqtt, "Client", FakeClient) + + node = MqttNode( + requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], + params={"topic": {"setpoint": "heating/setpoint"}, "publish_queue_size": 8}, + ) + node.assign_flow("heating", "out") + + async def scenario() -> int: + await node.start_publisher() + assert node._publish_queue is not None + size = node._publish_queue.maxsize + await node.stop_publisher() + return size + + assert asyncio.run(scenario()) == 8