stroblmeandClaude Opus 5
da528340a9
Measured with `make bench-engine` against a real Redis: 103.6 -> 164.4 messages a second on a five-node chain (p50 latency 2125 -> 1171 ms) and 34.8 -> 63.2 on a fan-out of twenty. Against the memory backend, which is what a pip install runs on, 262 -> 626. The two that bought most of it: - `StateBackend.record` puts a published value, its timestamp, its series and its version counter in one round trip. They were four calls building four pipelines, and a value crossing an edge pays them twice. A released rate-limit hold rides along instead of a DEL per port. - the readiness check reads a node's inputs and hands them to the node, rather than reading the triggering ones to count them and having the node read the same keys again a moment later. `apply_outputs` was a second copy of `_record_outputs` and is now the same code plus the event that distinguishes it. The rest, each small: - `_derive` builds a node-by-id map and a `consumes` index, so dispatching an item and publishing a value stop scanning every node in the installation. - `read_all` is memoised against the store revision — it sits on the publish path, so a dashboard slider was reading and validating every flow file per value. Same mechanism `_wiring` already uses. - the `message_value` source block is built once per node instead of per emission. - both timer threads ask the queue to promote only when something is actually due, which takes an idle engine from ~4 Redis round trips a second to one. - the shared httpx client is bounded (32 connections, one retry); its default pool is 100 with no per-host cap, so one slow endpoint could take it and every other sender node with it. - the MQTT and delay nodes no longer log a line per message at INFO. Robustness, in the same pass: - `MemoryWorkQueue._done` was a set nothing ever removed from — one entry per non-idempotent node per item, for the life of the process, in the default configuration. Capped, the way the Redis side expires its markers. - a saturated engine can claim from the due lane past the cascade limit. The capacity gate sits in front of the claim, so the due lane's priority — decided inside it — did not apply while every slot was held: a motor's stop was not behind the long nodes, it was unread. Only after a slot has genuinely failed to free for half a second, and briefly, so the backlog is not starved in turn. - `reclaim_stale` dispatches through that same gate. It could return sixty entries and push in-flight far past the limit the gate exists to hold. - a flow's nodes are stopped together rather than one after another. Each gets `NODE_STOP_TIMEOUT`, so a flow whose broker was unreachable took five seconds per node — long enough to outlast `REBUILD_WAIT` and 503 the deploy. - the worker pool and the HTTP client are closed on a thread, not on the event loop, and a run closes the state backend it built (on Redis, a client and a connection pool per run). - the five background tasks say something when they die. Each catches exceptions inside its loop, so one raised anywhere else left the engine serving with no metrics, no alerts or no artifact sweep, silently. `tests/flow/test_round_trips.py` counts the state operations one message costs — four, where it was about eleven — because none of the above would fail a behavioural test if it were undone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T
487 lines
17 KiB
Python
487 lines
17 KiB
Python
"""HTTP nodes: an inbound webhook trigger and an outbound request sender."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hmac
|
|
import logging
|
|
import threading
|
|
from collections.abc import Iterable
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Any, Literal
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from fluksio.flow.messages import MessageSpec
|
|
from fluksio.flow.nodes.base import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# One pooled client for every sender node: connections are the expensive part
|
|
# of an HTTP request, and a node that fires every second should keep its own.
|
|
# The ceiling is process-wide rather than per node, which is why it is a
|
|
# constant here rather than a `Params` field on the node.
|
|
HTTP_POOL_LIMIT = 32
|
|
_client: httpx.Client | None = None
|
|
_client_lock = threading.Lock()
|
|
|
|
|
|
def shared_client() -> httpx.Client:
|
|
"""The process-wide HTTP client, built on first use.
|
|
|
|
Bounded on purpose: httpx's default pool is 100 connections with no
|
|
per-host cap, so one endpoint that stops answering could take the whole
|
|
pool and every other sender node with it. Read outside the lock once it
|
|
exists — this is on the per-request path, and rebinding the global is
|
|
what the lock is for.
|
|
"""
|
|
global _client
|
|
if _client is not None:
|
|
return _client
|
|
with _client_lock:
|
|
if _client is None:
|
|
_client = httpx.Client(
|
|
limits=httpx.Limits(
|
|
max_connections=HTTP_POOL_LIMIT,
|
|
max_keepalive_connections=HTTP_POOL_LIMIT // 2,
|
|
keepalive_expiry=30.0,
|
|
),
|
|
transport=httpx.HTTPTransport(retries=1),
|
|
)
|
|
return _client
|
|
|
|
|
|
def close_shared_client() -> None:
|
|
"""Release pooled connections at shutdown. Safe to call more than once."""
|
|
global _client
|
|
with _client_lock:
|
|
if _client is not None:
|
|
_client.close()
|
|
_client = None
|
|
|
|
|
|
def _first_catch_all(app: FastAPI) -> int:
|
|
"""Where a webhook has to go in to be reachable.
|
|
|
|
A mount at the root answers for every path, so anything registered after
|
|
it is dead. Returns that mount's index, or the end of the table.
|
|
"""
|
|
from starlette.routing import Mount
|
|
|
|
for index, route in enumerate(app.routes):
|
|
if isinstance(route, Mount) and route.path in ("", "/"):
|
|
return index
|
|
return len(app.routes)
|
|
|
|
|
|
class HttpNode(Node):
|
|
"""
|
|
HTTP node that can act as a trigger (receiver) or sender based on configuration.
|
|
|
|
This node integrates with FastAPI to either:
|
|
|
|
- **Trigger mode**: Receive incoming HTTP requests (GET/POST) and inject data
|
|
into the pipeline. Used when ``provides`` is specified but ``requires`` is empty.
|
|
- **Sender mode**: Make outgoing HTTP requests with pipeline data. Used when
|
|
``requires`` is specified.
|
|
|
|
:param url: The URL endpoint. For trigger mode, this is the route path
|
|
(e.g., "/sensors/temperature"). For sender mode, this is the full URL
|
|
to send requests to.
|
|
:type url: str
|
|
:param method: HTTP method - "GET" or "POST".
|
|
:type method: Literal["GET", "POST"]
|
|
:param requires: Messages required by this node (makes it a sender node).
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages provided by this node (makes it a trigger node).
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Additional parameters for the node.
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
:param timeout: Request timeout in seconds (for sender mode).
|
|
:type timeout: float
|
|
:param headers: Additional HTTP headers.
|
|
:type headers: dict[str, str] | None
|
|
:param secret: Shared secret callers append to the webhook URL (trigger mode).
|
|
:type secret: str
|
|
|
|
:raises ValueError: If both ``requires`` and ``provides`` are empty, or if
|
|
the configuration is invalid.
|
|
|
|
:example:
|
|
Trigger node (receives POST requests):
|
|
|
|
>>> trigger = HttpNode(
|
|
... url="/api/sensors/temperature",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
|
|
Sender node (makes POST requests):
|
|
|
|
>>> sender = HttpNode(
|
|
... url="https://api.example.com/data",
|
|
... method="POST",
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
"""
|
|
|
|
class Mode(Enum):
|
|
"""Operating mode of the HTTP node."""
|
|
|
|
TRIGGER = "trigger" # Receives HTTP requests
|
|
SENDER = "sender" # Sends HTTP requests
|
|
|
|
# A repeated request is a repeated request, whatever the endpoint does
|
|
# with it.
|
|
idempotent = False
|
|
|
|
__slots__ = (
|
|
"url",
|
|
"method",
|
|
"mode",
|
|
"timeout",
|
|
"headers",
|
|
"query",
|
|
"send_inputs",
|
|
"secret",
|
|
"_route_registered",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
url: str
|
|
method: Literal["GET", "POST"] = "POST"
|
|
timeout: float = 30.0
|
|
headers: dict[str, str] = {}
|
|
query: dict[str, Any] = Field(
|
|
default_factory=dict,
|
|
description=(
|
|
"Fixed query parameters, sender mode. A value may be a secret "
|
|
"reference, which is how an API key stays out of the flow file."
|
|
),
|
|
)
|
|
send_inputs: bool = Field(
|
|
default=True,
|
|
description=(
|
|
"Send the node's inputs as the request body or query. Off when "
|
|
"the inputs are only a trigger and the request is fully "
|
|
"described by 'url' and 'query'."
|
|
),
|
|
)
|
|
secret: str = Field(
|
|
default="",
|
|
description=(
|
|
"Shared secret for a webhook, appended to its URL: "
|
|
"/hooks/<flow>/<url>/<secret>. Empty leaves the webhook open "
|
|
"to anyone."
|
|
),
|
|
json_schema_extra={"x-secret": True},
|
|
)
|
|
|
|
@classmethod
|
|
def instance_key(cls, params: dict[str, Any]) -> str | None:
|
|
# ponytail: a webhook's stored url is the path before the flow name is
|
|
# prefixed onto it, so two flows both receiving on "/tick" merge into
|
|
# one neuron. Key on mode as well if that ever misleads.
|
|
url = params.get("url")
|
|
if not url:
|
|
return None
|
|
# A URL may legally carry credentials, and this key becomes a group id
|
|
# in the brain graph — rendered into the response and into the DOM.
|
|
parts = urlsplit(str(url))
|
|
if parts.username or parts.password:
|
|
host = parts.netloc.rsplit("@", 1)[-1]
|
|
return urlunsplit(parts._replace(netloc=host))
|
|
return str(url)
|
|
|
|
def __init__(
|
|
self,
|
|
url: str | None = None,
|
|
method: Literal["GET", "POST"] | None = None,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
params = dict(params) if params else {}
|
|
if url is not None:
|
|
params["url"] = url
|
|
if method is not None:
|
|
params["method"] = method
|
|
cfg = self.Params.model_validate(params)
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An HTTP node needs either inputs (to send) or outputs (to receive)"
|
|
)
|
|
|
|
# Inputs mean this node sends data out; outputs mean it receives.
|
|
self.mode = HttpNode.Mode.SENDER if requires else HttpNode.Mode.TRIGGER
|
|
|
|
self.url = cfg.url
|
|
self.method = cfg.method.upper()
|
|
self.timeout = cfg.timeout
|
|
self.headers = cfg.headers
|
|
self.query = cfg.query
|
|
self.send_inputs = cfg.send_inputs
|
|
self.secret = cfg.secret
|
|
self._route_registered = False
|
|
|
|
# Set default name based on mode and URL
|
|
if name is None:
|
|
safe_url = self.url.replace("/", "_").replace(":", "").strip("_")
|
|
name = f"http_{self.mode.value}_{safe_url}"
|
|
|
|
# Initialize parent with appropriate function
|
|
# For trigger mode, f is a no-op since data is injected via inject()
|
|
# For sender mode, f handles the outgoing HTTP request
|
|
super().__init__(
|
|
f=(
|
|
self._noop_trigger
|
|
if self.mode == HttpNode.Mode.TRIGGER
|
|
else self._sender_handler
|
|
),
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
@staticmethod
|
|
def _noop_trigger(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""
|
|
No-op function for trigger mode nodes.
|
|
|
|
Trigger mode nodes inject data via :meth:`inject`, not :meth:`__call__`.
|
|
This function exists only to satisfy the Node interface.
|
|
"""
|
|
return None
|
|
|
|
def _sender_handler(
|
|
self, params: dict[str, Any], **kwargs: Any
|
|
) -> dict[str, Any] | None:
|
|
"""
|
|
Send HTTP request with pipeline data (sender mode).
|
|
|
|
This method is called when upstream dependencies are satisfied.
|
|
It sends the required data via HTTP request.
|
|
|
|
Node functions already run on a worker thread, so the request is sent
|
|
synchronously over the shared pooled client: a node publishing every
|
|
second should reuse its connection, not open one per message.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Pipeline data to send (from required messages).
|
|
:type kwargs: Any
|
|
:returns: Response data if the endpoint returns JSON, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
client = shared_client()
|
|
payload = dict(kwargs) if self.send_inputs else {}
|
|
try:
|
|
if self.method == "GET":
|
|
response = client.get(
|
|
self.url,
|
|
params={**self.query, **payload},
|
|
headers=self.headers,
|
|
timeout=self.timeout,
|
|
)
|
|
else: # POST
|
|
response = client.post(
|
|
self.url,
|
|
params=self.query or None,
|
|
json=payload,
|
|
headers=self.headers,
|
|
timeout=self.timeout,
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
body = response.json()
|
|
except Exception:
|
|
return None
|
|
# Outputs are keyed by port, so only an object can be one. A bare
|
|
# scalar or list is a valid reply, just not something to publish.
|
|
return body if isinstance(body, dict) else None
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(
|
|
"HTTP error in node '%s': status=%s, url=%s",
|
|
self.name,
|
|
e.response.status_code,
|
|
self.url,
|
|
)
|
|
raise
|
|
except httpx.RequestError as e:
|
|
logger.error("Request error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""A webhook needs a route; a sender reaches out on its own."""
|
|
if self.mode is HttpNode.Mode.TRIGGER and app is not None:
|
|
self.register_route(app)
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self.mode is HttpNode.Mode.TRIGGER and app is not None:
|
|
self.unregister_route(app)
|
|
|
|
def register_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Register this node's HTTP endpoint with a FastAPI application.
|
|
|
|
This method should only be called for trigger mode nodes. It creates
|
|
a route that, when called, triggers the node in the pipeline.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
:raises RuntimeError: If called on a sender mode node.
|
|
|
|
:example:
|
|
>>> from fastapi import FastAPI
|
|
>>> app = FastAPI()
|
|
>>> trigger_node = HttpNode(
|
|
... url="/sensors/data",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
>>> trigger_node.register_route(app)
|
|
"""
|
|
if self.mode != HttpNode.Mode.TRIGGER:
|
|
raise RuntimeError("Can only register routes for trigger mode nodes")
|
|
|
|
if self._route_registered:
|
|
return
|
|
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Route
|
|
|
|
async def handle_request(request: Request) -> JSONResponse:
|
|
"""
|
|
Handle incoming HTTP request and trigger the pipeline.
|
|
|
|
:param request: The incoming Starlette request.
|
|
:type request: Request
|
|
:returns: JSON response with trigger result.
|
|
:rtype: JSONResponse
|
|
"""
|
|
if self.secret and not hmac.compare_digest(
|
|
str(request.path_params.get("secret", "")).encode(),
|
|
self.secret.encode(),
|
|
):
|
|
# A wrong secret must look like no such hook at all.
|
|
return JSONResponse(content={"detail": "Not Found"}, status_code=404)
|
|
|
|
try:
|
|
# Parse request data
|
|
data: dict[str, Any]
|
|
if self.method == "GET":
|
|
data = dict(request.query_params)
|
|
else: # POST
|
|
content_type = request.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
data = await request.json()
|
|
elif "application/x-www-form-urlencoded" in content_type:
|
|
form = await request.form()
|
|
data = dict(form)
|
|
else:
|
|
data = await request.json() # Default to JSON
|
|
|
|
# Payload keys are port names; values arrive as text.
|
|
typed_data = {}
|
|
for spec in self.output_ports:
|
|
if spec.port in data:
|
|
typed_data[spec.port] = spec.coerce(data[spec.port])
|
|
|
|
# Inject data into the pipeline (trigger mode nodes inject via
|
|
# provides). Off the loop: journalling is a blocking Redis
|
|
# round trip, and every webhook was making it on the thread
|
|
# the whole API answers from.
|
|
result = await asyncio.to_thread(self.inject, typed_data)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "triggered",
|
|
"node": self.name,
|
|
"data": typed_data,
|
|
"result": result if isinstance(result, dict) else None,
|
|
}
|
|
)
|
|
|
|
except ValueError as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=400,
|
|
)
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=500,
|
|
)
|
|
|
|
# The secret is a path parameter, not part of the registered path, so
|
|
# neither the route table nor the logs below carry its value.
|
|
path = f"{self.url}/{{secret}}" if self.secret else self.url
|
|
|
|
# Create a Starlette Route and add it directly to the app's routes
|
|
route = Route(
|
|
path,
|
|
handle_request,
|
|
methods=[self.method],
|
|
name=self.id,
|
|
)
|
|
# Ahead of any catch-all mount. Starlette takes the first route that
|
|
# matches, and the MCP app is mounted at "/", so appending would put
|
|
# every webhook behind something that answers for every path.
|
|
app.routes.insert(_first_catch_all(app), route)
|
|
|
|
self._route_registered = True
|
|
logger.info(
|
|
"Registered %s route for node '%s': %s",
|
|
self.method,
|
|
self.name,
|
|
path,
|
|
)
|
|
|
|
def unregister_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Unregister this node's HTTP endpoint from a FastAPI application.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
|
|
.. note::
|
|
FastAPI doesn't natively support route removal. This method
|
|
removes the route from the internal routes list, but the change
|
|
may not take effect until the application is restarted or
|
|
the OpenAPI schema is regenerated.
|
|
"""
|
|
if not self._route_registered:
|
|
return
|
|
|
|
# FastAPI doesn't have a clean way to remove routes
|
|
# We need to filter them out from the routes list
|
|
app.routes[:] = [
|
|
route
|
|
for route in app.routes
|
|
if not (hasattr(route, "name") and route.name == self.id)
|
|
]
|
|
|
|
self._route_registered = False
|
|
logger.info("Unregistered route for node '%s': %s", self.name, self.url)
|