Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Engine holds for a rename: the pipeline refuses publish and inline cascades with EngineHeld once those running have finished, the execution service stops claiming and its timers once claimed cascades have finished; both wait without a limit, report what is busy and can be aborted. REST answers 503 with Retry-After, a device frame busy. Redis adds no longer wait for each other, only a queue hold waits for them

Browse Source
This commit is contained in:
stroblme
2026-11-02 20:21:17 +01:00
parent 46b57de4ec
commit badcf94491
10 changed files with 687 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/api/routes/flows.py
+6 -1
View File
@@ -65,7 +65,7 @@ from fluksio.flow.devices import (
from fluksio.flow.events import event_bus
from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify
from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta
from fluksio.flow.pipeline import ValidationIssue
from fluksio.flow.pipeline import EngineHeld, ValidationIssue
from fluksio.flow.runs import RunRejected
from fluksio.flow.schemas import (
NAME_PATTERN,
@@ -1558,6 +1558,7 @@ FRAME_REFUSALS = {
"not_scalar": "the port it names is not wired as a bool, int, float or str",
"undeclared": "no flow declares its message",
"refused": "its message refused the value",
"busy": "a rename holds the engine",
}
@@ -1604,6 +1605,8 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
return "undeclared"
except TypeError:
return "refused"
except EngineHeld:
return "busy"
return ""
@@ -1657,6 +1660,8 @@ def ingest_publish(app: Any, device: DeviceDef, payload: dict[str, Any]) -> str:
return "undeclared"
except TypeError:
return "refused"
except EngineHeld:
return "busy"
return ""
backend/fluksio/flow/controller.py
+3
View File
@@ -1858,6 +1858,9 @@ class FlowController:
Refuses a name no flow declares: the flows own the namespace, and a
message nothing reads or writes would just be a key nobody sees.
:raises EngineHeld: While a rename holds the engine, for the caller to
answer: a 503, a device's ``busy``, a log line.
"""
if self.pipeline is None:
raise KeyError(name)
backend/fluksio/flow/executor.py
+175 -83
View File
@@ -17,7 +17,7 @@ from __future__ import annotations
import logging
import threading
import time
from collections import deque
from collections import Counter, deque
from collections.abc import Callable
from concurrent.futures import Executor, Future, ThreadPoolExecutor
from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
@@ -26,6 +26,7 @@ from fluksio.flow.queue import MAX_DELIVERIES, WorkItem, WorkQueue
if TYPE_CHECKING:
from fluksio.flow.events import EventBus
from fluksio.flow.namespace import Rename
from fluksio.flow.pipeline import Pipeline
logger = logging.getLogger(__name__)
@@ -69,10 +70,41 @@ BACKLOG_INTERVAL_S = 5.0
BACKLOG_DEGRADED = 50
BACKLOG_STRIKES = 3
#: How often an engine hold looks at its abort while it waits.
HOLD_POLL_S = 0.5
_P = ParamSpec("_P")
_T = TypeVar("_T")
def wait_out(
lock: threading.Condition,
busy: Callable[[], list[str]],
abort: threading.Event,
report: Callable[[list[str]], None] | None = None,
) -> bool:
"""Wait until ``busy``, read under ``lock``, names nothing.
Without a limit, because an engine hold waits for as long as the work
takes. ``report`` hears each new set of names, so whoever asked can show
what it waits for, and setting ``abort`` gives up with False.
"""
said: list[str] = []
while True:
with lock:
names = busy()
while names and names == said and not abort.is_set():
lock.wait(HOLD_POLL_S)
names = busy()
if abort.is_set():
return False
if not names:
return True
said = names
if report is not None:
report(names)
class _PoolLane(Executor):
def __init__(self, pool: _ReservedPool, due: bool) -> None:
self._pool = pool
@@ -158,7 +190,8 @@ class ExecutionService:
queue.on_delayed = self._timer_wake.set
self._intake = threading.Event()
self._intake.set()
self._inflight = 0
# Cascades claimed and still running, by the node each came from.
self._running: Counter[str] = Counter()
self._inflight_lock = threading.Condition()
# Entry ids claimed and still running, under _inflight_lock.
self._active: set[str] = set()
@@ -174,6 +207,12 @@ class ExecutionService:
)
self._consumer: threading.Thread | None = None
self._timers: threading.Thread | None = None
# Held by the consumer for one pass of capacity check, claim and
# dispatch, and by the timer thread for one pass of promotion, touch
# and reclaim. `hold` takes both. Plain locks, because `release` may
# come from another thread than the one that held.
self._consumer_lock = threading.Lock()
self._timer_lock = threading.Lock()
# Consecutive backlog readings over the threshold, and whether the last
# of them said so out loud.
self._backlog_strikes = 0
@@ -225,11 +264,11 @@ class ExecutionService:
self._intake.clear()
deadline = time.monotonic() + DRAIN_TIMEOUT_S
with self._inflight_lock:
while self._inflight:
while self._running:
remaining = deadline - time.monotonic()
if remaining <= 0:
logger.warning(
"Rebuild did not wait out %d cascades", self._inflight
"Rebuild did not wait out %d cascades", self._running.total()
)
return
self._inflight_lock.wait(remaining)
@@ -237,13 +276,59 @@ class ExecutionService:
def resume_intake(self) -> None:
self._intake.set()
def hold(
self,
abort: threading.Event,
report: Callable[[list[str]], None] | None = None,
) -> bool:
"""Stop claiming, promoting, touching and reclaiming, once every
claimed cascade has finished. For a rename, which rewrites the queue.
Unlike `pause_intake` this also waits out a claim already blocking
when intake clears, and a timer pass. It waits without a limit and
tells ``report`` the nodes whose cascades are still running each time
they change. Setting ``abort`` gives up: claiming resumes and it
returns False. True means held until `release`.
"""
self._intake.clear()
# So a consumer waiting for capacity sees the intake closed.
self._capacity_changed()
self._consumer_lock.acquire()
# The timer lock only after the drain: while a cascade runs, the
# timer thread keeps touching it, so it is not reclaimed as dropped.
if not wait_out(self._inflight_lock, self._busy, abort, report):
self._consumer_lock.release()
self._intake.set()
return False
self._timer_lock.acquire()
return True
def release(self) -> None:
"""End a `hold`."""
self._timer_lock.release()
self._consumer_lock.release()
self._intake.set()
def rename_reclaimed(self, rename: Rename) -> None:
"""Move the names in reclaimed work still waiting to be run, which
the queue's own `rename` cannot reach. Under a `hold`."""
with self._inflight_lock:
self._reclaimed = deque(
rename.item(item) or item for item in self._reclaimed
)
def _busy(self) -> list[str]:
"""The nodes whose cascades are running. Under _inflight_lock."""
return sorted(self._running)
def alive(self) -> bool:
return self._consumer is not None and self._consumer.is_alive()
@property
def inflight(self) -> int:
"""Cascades claimed and still running."""
return self._inflight
with self._inflight_lock:
return self._running.total()
# -------------------------------------------------------------------------
# Threads
@@ -255,40 +340,45 @@ class ExecutionService:
if not self._intake.is_set():
self._intake.wait(timeout=0.5)
continue
free, due_only = self._await_capacity()
if not free:
continue
with self._inflight_lock:
recovered = [
item
for item in self._reclaimed
if not due_only or item.not_before > 0
][:free]
for item in recovered:
self._reclaimed.remove(item)
if recovered:
for item in recovered:
self._dispatch(item)
continue
try:
items = self.queue.claim(
free,
# Briefly, in the due-only case: this is the saturated
# engine, and a slot freeing has to be noticed promptly.
DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS,
due_only,
)
failures = 0
except Exception as exc:
failures += 1
logger.error("Could not claim work: %s", exc)
self._publish_unavailable(exc)
# Backing off hard: a queue that is down stays down for a while.
self._stop.wait(min(30.0, 2.0**failures))
continue
backoff = 0.0
with self._consumer_lock:
free, due_only = self._await_capacity()
if not free:
continue
with self._inflight_lock:
recovered = [
item
for item in self._reclaimed
if not due_only or item.not_before > 0
][:free]
for item in recovered:
self._reclaimed.remove(item)
if recovered:
for item in recovered:
self._dispatch(item)
continue
try:
items = self.queue.claim(
free,
# Briefly, in the due-only case: this is the saturated
# engine, and a slot freeing has to be noticed promptly.
DUE_CLAIM_BLOCK_MS if due_only else CLAIM_BLOCK_MS,
due_only,
)
failures = 0
except Exception as exc:
failures += 1
logger.error("Could not claim work: %s", exc)
self._publish_unavailable(exc)
# Backing off hard: a queue that is down stays down for a
# while. Outside the lock, so a hold does not wait it out.
backoff = min(30.0, 2.0**failures)
items = []
for item in items:
self._dispatch(item)
for item in items:
self._dispatch(item)
if backoff:
self._stop.wait(backoff)
def _sleep_until_due(self) -> bool:
"""Wait for the soonest deadline, the housekeeping cap, or a new one.
@@ -321,54 +411,56 @@ class ExecutionService:
promote = self._sleep_until_due()
if self._stop.is_set():
break
try:
if promote:
self.queue.move_due(time.time())
except Exception as exc:
logger.error("Could not promote delayed work: %s", exc)
# The item is still due, so the wait above would be zero and
# this would spin on a queue that is down. Back off to what a
# fixed poll used to cost.
self._stop.wait(DELAYED_INTERVAL_S)
now = time.monotonic()
if now - last_backlog >= BACKLOG_INTERVAL_S:
last_backlog = now
# One pass, not the sleep above: a hold waits for it to end.
with self._timer_lock:
try:
self._check_backlog()
if promote:
self.queue.move_due(time.time())
except Exception as exc:
logger.error("Could not read the queue backlog: %s", exc)
logger.error("Could not promote delayed work: %s", exc)
# The item is still due, so the wait above would be zero and
# this would spin on a queue that is down. Back off to what a
# fixed poll used to cost.
self._stop.wait(DELAYED_INTERVAL_S)
if now - last_touch >= TOUCH_INTERVAL_S:
last_touch = now
with self._inflight_lock:
running = list(self._active)
try:
self.queue.touch(running)
except Exception as exc:
logger.error("Could not touch claimed work: %s", exc)
now = time.monotonic()
if now - last_backlog >= BACKLOG_INTERVAL_S:
last_backlog = now
try:
self._check_backlog()
except Exception as exc:
logger.error("Could not read the queue backlog: %s", exc)
if now - last_reclaim < RECLAIM_INTERVAL_S:
continue
with self._inflight_lock:
if self._reclaimed or not self._intake.is_set():
if now - last_touch >= TOUCH_INTERVAL_S:
last_touch = now
with self._inflight_lock:
running = list(self._active)
try:
self.queue.touch(running)
except Exception as exc:
logger.error("Could not touch claimed work: %s", exc)
if now - last_reclaim < RECLAIM_INTERVAL_S:
continue
last_reclaim = now
try:
recovered = self.queue.reclaim_stale(RECLAIM_IDLE_MS)
for item in recovered:
logger.info(
"Reclaimed work for '%s' (delivery %d)",
item.node,
item.deliveries,
)
with self._inflight_lock:
self._reclaimed.extend(recovered)
self._active.update(
item.entry_id for item in recovered if item.entry_id
)
except Exception as exc:
logger.error("Could not reclaim stale work: %s", exc)
if self._reclaimed or not self._intake.is_set():
continue
last_reclaim = now
try:
recovered = self.queue.reclaim_stale(RECLAIM_IDLE_MS)
for item in recovered:
logger.info(
"Reclaimed work for '%s' (delivery %d)",
item.node,
item.deliveries,
)
with self._inflight_lock:
self._reclaimed.extend(recovered)
self._active.update(
item.entry_id for item in recovered if item.entry_id
)
except Exception as exc:
logger.error("Could not reclaim stale work: %s", exc)
def _check_backlog(self) -> None:
"""Say so when work has been waiting in the stream for a while.
@@ -439,7 +531,7 @@ class ExecutionService:
def _dispatch(self, item: WorkItem) -> None:
with self._inflight_lock:
self._inflight += 1
self._running[item.node] += 1
if item.entry_id:
self._active.add(item.entry_id)
try:
@@ -455,7 +547,7 @@ class ExecutionService:
def _done(self, item: WorkItem) -> None:
with self._inflight_lock:
self._inflight -= 1
self._running -= Counter((item.node,))
self._active.discard(item.entry_id)
self._inflight_lock.notify_all()
@@ -621,7 +713,7 @@ class ExecutionService:
except Exception as exc:
return {"error": str(exc), "consumer_alive": self.alive()}
stats["consumer_alive"] = self.alive()
stats["cascades_busy"] = self._inflight
stats["cascades_busy"] = self.inflight
stats["behind"] = self.behind
return stats
backend/fluksio/flow/nodes/device.py
+5 -1
View File
@@ -23,7 +23,7 @@ from pydantic import BaseModel, Field
from fluksio.flow.devices import PORT_PATTERN, presence
from fluksio.flow.messages import MessageSpec
from fluksio.flow.nodes.base import Node
from fluksio.flow.pipeline import ValueSource
from fluksio.flow.pipeline import EngineHeld, ValueSource
if TYPE_CHECKING:
from fastapi import FastAPI
@@ -135,5 +135,9 @@ class DeviceNode(Node):
)
try:
pipeline.publish(flags, source)
except EngineHeld:
logger.warning(
"Kept the flags of '%s': a rename holds the engine", self.cfg.device
)
except Exception:
logger.exception("Could not reset the flags of '%s'", self.cfg.device)
backend/fluksio/flow/pipeline.py
+136 -65
View File
@@ -19,7 +19,7 @@ import logging
import threading
import time
import uuid
from collections import deque
from collections import Counter, deque
from collections.abc import Callable, Iterator
from concurrent.futures import Executor, Future, ThreadPoolExecutor
from contextlib import contextmanager
@@ -33,6 +33,7 @@ from pydantic import BaseModel, model_validator
from fluksio.flow import logs
from fluksio.flow.artifacts import is_reference
from fluksio.flow.events import EventBus
from fluksio.flow.executor import wait_out
from fluksio.flow.messages import flow_of, requalify
from fluksio.flow.namespace import (
DELIVERED,
@@ -265,6 +266,14 @@ def _derive(
return produces, dependencies, consumes, by_id
class EngineHeld(RuntimeError):
"""A value arrived while a rename holds the engine.
Answered as a 503 with ``Retry-After``: nothing is wrong with the value,
and the same publish works once the rename is done.
"""
class Pipeline:
"""Directed graph of nodes with automatic dependency resolution."""
@@ -286,6 +295,9 @@ class Pipeline:
"_gate_lock",
"_gate_revision",
"_graph_lock",
"_holding",
"_inline",
"_inline_lock",
"_queue",
"_node_pool",
"history_limits",
@@ -324,6 +336,12 @@ class Pipeline:
# `_gate_lock` or `state.lock()` in either order, which is what keeps
# three locks from needing an ordering rule.
self._graph_lock = threading.RLock()
# Set while a rename holds the engine (`hold`). What runs outside the
# queue, by node id or message name, is counted so a hold can wait
# it out.
self._holding = False
self._inline: Counter[str] = Counter()
self._inline_lock = threading.Condition()
# An empty state backend is falsy, so this cannot be ``state or ...``:
# that would quietly hand the pipeline a second, private state and
# leave everyone reading the shared one seeing nothing.
@@ -1325,6 +1343,49 @@ class Pipeline:
with self._gate_lock:
return flow in self._stepping
def hold(
self,
abort: threading.Event,
report: Callable[[list[str]], None] | None = None,
) -> bool:
"""Refuse `publish` and every cascade outside the queue, once those
already running have finished. For a rename, which nothing may write
past.
Waits without a limit and tells ``report`` the nodes and messages
still running inline each time they change. Setting ``abort`` gives
up and lets them through again: False. True means held until
`release`.
"""
with self._inline_lock:
self._holding = True
if wait_out(self._inline_lock, lambda: sorted(self._inline), abort, report):
return True
self.release()
return False
def release(self) -> None:
"""End a `hold`."""
with self._inline_lock:
self._holding = False
@contextmanager
def _inline_run(self, names: list[str]) -> Iterator[None]:
"""Count a run outside the queue, which a hold waits out.
:raises EngineHeld: While held.
"""
with self._inline_lock:
if self._holding:
raise EngineHeld("A rename holds the engine; try again shortly")
self._inline.update(names)
try:
yield
finally:
with self._inline_lock:
self._inline -= Counter(names)
self._inline_lock.notify_all()
def _execute_parallel(
self,
nodes_subset: set[Node] | None,
@@ -1625,44 +1686,45 @@ class Pipeline:
already in state, so they are handed to their readers rather than
written again.
"""
run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}"
self._publish(
{
"type": "cascade_started",
"run": run_id,
"flow": node.flow,
"node": node.id,
"cause": cause,
"deliveries": 1,
"ts": time.time(),
}
)
try:
published = (
set(outputs or {})
if delivered
else self.apply_outputs(node, outputs, run_id)
)
state = self.run_downstream(
node,
entry_id=run_id,
# No payload means the value is already in state and this is a
# wake-up, which has nothing to name as changed.
changed=published if outputs else None,
overrides=outputs if delivered else None,
)
finally:
# Paired, or a cascade that raised leaves the run open until the
# abandoned sweep ten minutes later.
with self._inline_run([node.id]):
run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}"
self._publish(
{
"type": "cascade_finished",
"type": "cascade_started",
"run": run_id,
"flow": node.flow,
"node": node.id,
"cause": cause,
"deliveries": 1,
"ts": time.time(),
}
)
return state
try:
published = (
set(outputs or {})
if delivered
else self.apply_outputs(node, outputs, run_id)
)
state = self.run_downstream(
node,
entry_id=run_id,
# No payload means the value is already in state and this is a
# wake-up, which has nothing to name as changed.
changed=published if outputs else None,
overrides=outputs if delivered else None,
)
finally:
# Paired, or a cascade that raised leaves the run open until the
# abandoned sweep ten minutes later.
self._publish(
{
"type": "cascade_finished",
"run": run_id,
"flow": node.flow,
"ts": time.time(),
}
)
return state
def publish(
self, values: dict[str, Any], source: ValueSource | None = None
@@ -1676,45 +1738,48 @@ class Pipeline:
``source`` says what did, so the canvas can show the value arriving
from outside instead of blaming whichever node happens to be drawn as
a producer of that message.
:raises EngineHeld: While a rename holds the engine.
"""
if not values:
return
origin = source or ValueSource(kind="api", label="API")
ts = time.time()
self._state.record(
values,
{self._timestamp_key(name): ts for name in values},
[self._version_key(name) for name in values],
ts,
self.history_limits,
)
source_dump = origin.model_dump()
for name, value in values.items():
self._publish(
{
"type": "message_value",
"flow": flow_of(name),
"name": name,
"value": value,
"ts": ts,
"source": source_dump,
}
with self._inline_run(list(values)):
ts = time.time()
self._state.record(
values,
{self._timestamp_key(name): ts for name in values},
[self._version_key(name) for name in values],
ts,
self.history_limits,
)
source_dump = origin.model_dump()
for name, value in values.items():
self._publish(
{
"type": "message_value",
"flow": flow_of(name),
"name": name,
"value": value,
"ts": ts,
"source": source_dump,
}
)
targets: set[Node] = set()
for name in values:
for consumer in self.consumes.get(name, ()):
targets.add(consumer)
targets.update(self._get_downstream(consumer))
if targets:
# Only what these values reach, as in a cascade: a node further
# down whose producer published nothing would otherwise run again
# on what it already read, once per microphone chunk.
self._execute_parallel(
targets, self._state, check_ready=True, changed=set(values)
)
targets: set[Node] = set()
for name in values:
for consumer in self.consumes.get(name, ()):
targets.add(consumer)
targets.update(self._get_downstream(consumer))
if targets:
# Only what these values reach, as in a cascade: a node further
# down whose producer published nothing would otherwise run again
# on what it already read, once per microphone chunk.
self._execute_parallel(
targets, self._state, check_ready=True, changed=set(values)
)
def defer(
self,
@@ -1785,8 +1850,14 @@ class Pipeline:
"ts": time.time(),
}
)
# Losing the value outright would be worse than running it here.
self._run_here(node, outputs, cause="external", delivered=kind == "emission")
# Losing the value outright would be worse than running it here,
# unless a rename holds the engine, which nothing runs past.
try:
self._run_here(
node, outputs, cause="external", delivered=kind == "emission"
)
except EngineHeld:
logger.warning("Dropped work for '%s' while the engine is held", node.id)
def value(self, name: str) -> dict[str, Any]:
"""One message and its timestamp, read together without listing state."""
backend/fluksio/flow/queue.py
+16 -5
View File
@@ -756,12 +756,15 @@ class RedisWorkQueue(WorkQueue):
self._held_key = f"{namespace}:__held__"
self._move_due = self._redis.register_script(_MOVE_DUE)
self._release = self._redis.register_script(_RELEASE)
# `add` reads the flag and writes under this, and `release` empties the
# held list and clears the flag under it. So nothing added after `hold`
# lands in a stream a rename is rebuilding, and nothing added while
# held overtakes held work.
self._hold_lock = threading.Lock()
# `add` reads the flag under this, and counts an `XADD` in `_adding`
# until it is done, so adds from parallel producers do not wait for
# each other. `hold` sets the flag and waits for `_adding` to reach 0,
# and `release` empties the held list and clears the flag under it. So
# nothing added after `hold` lands in a stream a rename is rebuilding,
# and nothing added while held overtakes held work.
self._hold_lock = threading.Condition()
self._holding = False
self._adding = 0
self._ensure_group()
self._drop_stale_consumers()
@@ -811,12 +814,18 @@ class RedisWorkQueue(WorkQueue):
if self._holding:
self._redis.rpush(self._held_key, orjson.dumps(item.to_fields()))
return
self._adding += 1
try:
self._redis.xadd(
self._stream,
cast(Any, item.to_fields()),
maxlen=STREAM_MAXLEN,
approximate=True,
)
finally:
with self._hold_lock:
self._adding -= 1
self._hold_lock.notify_all()
def add_delayed(self, item: WorkItem, not_before: float) -> None:
item.not_before = not_before
@@ -1154,6 +1163,8 @@ class RedisWorkQueue(WorkQueue):
def hold(self) -> None:
with self._hold_lock:
self._holding = True
while self._adding:
self._hold_lock.wait()
def release(self) -> None:
with self._hold_lock:
backend/fluksio/main.py
+9 -1
View File
@@ -43,7 +43,7 @@ from fluksio.flow.executor import ExecutionService
from fluksio.flow.identity import assign_ids
from fluksio.flow.metrics import MetricsCollector
from fluksio.flow.nodes.http import close_shared_client
from fluksio.flow.pipeline import ValueSource
from fluksio.flow.pipeline import EngineHeld, ValueSource
from fluksio.flow.placement import Placer
from fluksio.flow.plugins import load_plugins
from fluksio.flow.provision import load_provisioners
@@ -609,6 +609,14 @@ async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse: # n
return JSONResponse(status_code=503, content={"detail": str(exc)})
@app.exception_handler(EngineHeld)
async def engine_held(request: Request, exc: EngineHeld) -> JSONResponse: # noqa: ARG001
"""A value published while a rename holds the engine, which ends shortly."""
return JSONResponse(
status_code=503, content={"detail": str(exc)}, headers={"Retry-After": "5"}
)
# Tagged because the operation-id builder reads the first tag; the route
# itself stays out of the schema.
@app.get(
backend/tests/api/routes/test_flows.py
+34
View File
@@ -9,7 +9,9 @@ from sqlmodel import Session, select
from fluksio.api.routes import flows as flows_route
from fluksio.core.config import settings
from fluksio.flow.devices import Binding
from fluksio.flow.events import event_bus
from fluksio.flow.pipeline import EngineHeld
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
from tests.utils.flows import put_flow
@@ -600,3 +602,35 @@ def test_a_shared_source_write_is_audited(
headers=superuser_token_headers,
)
client.delete(f"{PREFIX}/audited", headers=superuser_token_headers)
def test_a_publish_while_a_rename_holds_the_engine_is_refused(
client: TestClient,
superuser_token_headers: dict[str, str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Over REST as a 503 to retry, from a device's socket as ``busy``."""
controller = client.app.state.flow_controller # type: ignore[attr-defined]
def held(*_: object) -> None:
raise EngineHeld("A rename holds the engine; try again shortly")
monkeypatch.setattr(controller, "publish_message", held)
response = client.post(
f"{settings.API_V1_STR}/messages/porch.doorbell",
headers=superuser_token_headers,
json={"value": True},
)
assert response.status_code == 503
assert response.headers["Retry-After"] == "5"
bound = Binding("panel", "porch", "panel", "gpio.4", "porch.doorbell", "in", "bool")
app = SimpleNamespace(
state=SimpleNamespace(
device_store=SimpleNamespace(bound=lambda _: [bound]),
flow_controller=controller,
)
)
device = SimpleNamespace(id="panel", title="Panel")
value = {"type": "publish", "name": "porch.doorbell", "value": True}
assert flows_route.ingest_publish(app, device, value) == "busy" # type: ignore[arg-type]
backend/tests/flow/test_engine_hold.py
+272
View File
@@ -0,0 +1,272 @@
"""Engine holds for a rename (section 2.2 of docs/private/id_refac_rename.md).
The pipeline refuses what it would run outside the queue, and the execution
service stops claiming and its timers once claimed work has finished. Both
wait without a limit and can be aborted.
"""
import threading
import time
import pytest
from fluksio.flow import executor
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.namespace import Rename
from fluksio.flow.nodes import Node
from fluksio.flow.pipeline import EngineHeld, Pipeline
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.state import MemoryState
def _engine(consume, queue=None) -> tuple[Pipeline, ExecutionService, Node]:
"""A source whose readings ``consume`` gets, and a service draining the
queue one cascade at a time, so order shows."""
def consumer_body(reading, params):
consume(reading)
source = Node(
f=lambda params: None,
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
name="source",
)
consumer = Node(
f=consumer_body,
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
name="consumer",
)
source.assign_flow("f", "source")
consumer.assign_flow("f", "consumer")
queue = queue or MemoryWorkQueue()
service = ExecutionService(queue, max_cascades=1)
pipeline = Pipeline(
nodes=[source, consumer],
state=MemoryState(),
work_queue=queue,
node_pool=service.node_pool,
)
service.bind(pipeline)
return pipeline, service, source
def _in_thread(fn, *args) -> tuple[threading.Thread, list]:
result: list = []
thread = threading.Thread(target=lambda: result.append(fn(*args)), daemon=True)
thread.start()
return thread, result
def _until(check, timeout: float = 5.0) -> None:
deadline = time.monotonic() + timeout
while not check():
assert time.monotonic() < deadline, "timed out"
time.sleep(0.02)
def _blocking(seen: list) -> tuple[threading.Event, threading.Event, object]:
"""A consumer that records and then waits until told to finish."""
running, finish = threading.Event(), threading.Event()
def consume(reading):
seen.append(reading)
running.set()
finish.wait(5)
return running, finish, consume
def test_a_hold_waits_out_a_running_cascade_without_a_limit(monkeypatch):
# What a rebuild's drain gives up after, which a hold does not.
monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
seen: list = []
running, finish, consume = _blocking(seen)
_, service, source = _engine(consume)
service.start()
try:
source.inject({"reading": 1.0})
assert running.wait(5)
busy: list = []
thread, held = _in_thread(service.hold, threading.Event(), busy.append)
thread.join(0.3)
assert thread.is_alive()
assert busy == [["f.source"]]
source.inject({"reading": 2.0})
finish.set()
thread.join(5)
assert held == [True]
time.sleep(0.2)
assert seen == [1.0]
service.release()
_until(lambda: seen == [1.0, 2.0])
finally:
finish.set()
service.stop()
class SlowClaim(MemoryWorkQueue):
"""A queue whose claim is still under way when a hold begins."""
def __init__(self) -> None:
super().__init__()
self.claiming = threading.Event()
self.go = threading.Event()
def claim(self, count, block_ms, due_only=False):
items = super().claim(count, block_ms, due_only)
if items:
self.claiming.set()
self.go.wait(5)
return items
def test_a_hold_waits_out_a_claim_already_under_way():
"""`pause_intake` returns here and the item runs after it."""
seen: list = []
queue = SlowClaim()
_, service, source = _engine(seen.append, queue)
service.start()
try:
source.inject({"reading": 1.0})
assert queue.claiming.wait(5)
thread, held = _in_thread(service.hold, threading.Event())
thread.join(0.2)
assert thread.is_alive()
queue.go.set()
thread.join(5)
assert held == [True]
assert seen == [1.0]
assert service.inflight == 0
service.release()
finally:
queue.go.set()
service.stop()
def test_an_aborted_hold_leaves_the_engine_as_it_was():
seen: list = []
running, finish, consume = _blocking(seen)
pipeline, service, source = _engine(consume)
service.start()
try:
source.inject({"reading": 1.0})
assert running.wait(5)
abort = threading.Event()
thread, held = _in_thread(service.hold, abort)
thread.join(0.2)
assert thread.is_alive()
abort.set()
thread.join(5)
assert held == [False]
finish.set()
# Claiming and promotion carry on.
assert pipeline.defer(source, {"f.reading": 2.0}, 0.05)
source.inject({"reading": 3.0})
_until(lambda: sorted(seen) == [1.0, 2.0, 3.0])
assert not service.queue.held
finally:
finish.set()
service.stop()
def test_an_aborted_pipeline_hold_lets_publishes_through_again():
seen: list = []
running, finish, consume = _blocking(seen)
pipeline, _, _ = _engine(consume)
publishing, _ = _in_thread(pipeline.publish, {"f.reading": 1.0})
try:
assert running.wait(5)
busy: list = []
abort = threading.Event()
thread, held = _in_thread(pipeline.hold, abort, busy.append)
thread.join(0.2)
assert thread.is_alive()
assert busy == [["f.reading"]]
abort.set()
thread.join(5)
assert held == [False]
finish.set()
publishing.join(5)
pipeline.publish({"f.reading": 2.0})
assert seen == [1.0, 2.0]
finally:
finish.set()
def test_publish_is_refused_while_held():
seen: list = []
pipeline, _, _ = _engine(seen.append)
assert pipeline.hold(threading.Event())
with pytest.raises(EngineHeld):
pipeline.publish({"f.reading": 1.0})
assert "f.reading" not in pipeline.state
pipeline.release()
pipeline.publish({"f.reading": 2.0})
assert seen == [2.0]
def test_work_arriving_during_a_hold_runs_after_it_in_order():
seen: list = []
pipeline, service, source = _engine(seen.append)
service.start()
try:
abort = threading.Event()
assert pipeline.hold(abort)
assert service.hold(abort)
source.inject({"reading": 1.0})
service.queue.hold()
source.inject({"reading": 2.0})
source.inject({"reading": 3.0})
time.sleep(0.2)
assert seen == []
service.queue.release()
service.release()
pipeline.release()
_until(lambda: seen == [1.0, 2.0, 3.0])
finally:
service.stop()
def test_a_queue_failure_while_held_drops_the_value_rather_than_run_it(
monkeypatch,
):
seen: list = []
pipeline, service, source = _engine(seen.append)
def down(item):
raise ConnectionError("queue down")
monkeypatch.setattr(service.queue, "add", down)
assert pipeline.hold(threading.Event())
source.inject({"reading": 1.0})
assert seen == []
pipeline.release()
source.inject({"reading": 2.0})
assert seen == [2.0]
def test_reclaimed_work_waiting_to_run_follows_a_rename():
service = ExecutionService(MemoryWorkQueue())
service._reclaimed.extend(
[
WorkItem(kind="cascade", node="heat.valve", flow="heat", entry_id="1-0"),
WorkItem(kind="cascade", node="heating.pump", flow="heating"),
]
)
service.rename_reclaimed(Rename("heat", "warm"))
assert [(i.node, i.flow) for i in service._reclaimed] == [
("warm.valve", "warm"),
("heating.pump", "heating"),
]
assert service._reclaimed[0].entry_id == "1-0"
backend/tests/flow/test_rename_runtime.py
+31
View File
@@ -3,6 +3,7 @@ and 3 of docs/private/id_refac_rename.md). Memory always, Redis with
FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""
import os
import threading
import uuid
import pytest
@@ -218,6 +219,36 @@ def test_held_work_is_released_in_order_behind_older_work(queue):
]
def test_a_hold_waits_for_an_add_already_writing_to_the_stream(
redis_queue, monkeypatch
):
"""Adds do not wait for each other, but a hold waits for them, so no
entry lands in a stream after the rebuild has read it."""
writing, finish = threading.Event(), threading.Event()
xadd = redis_queue._redis.xadd
def slow(*args, **kwargs):
writing.set()
finish.wait(5)
return xadd(*args, **kwargs)
monkeypatch.setattr(redis_queue._redis, "xadd", slow)
adding = threading.Thread(target=redis_queue.add, args=(item("heat.a"),))
adding.start()
assert writing.wait(5)
holding = threading.Thread(target=redis_queue.hold)
holding.start()
holding.join(0.2)
assert holding.is_alive()
finish.set()
holding.join(5)
adding.join(5)
assert redis_queue.count(HEAT)["queue_items"] == 1
assert redis_queue.held
redis_queue.release()
def test_a_guarded_timer_still_finds_its_guard(queue):
state = MemoryState()
state["__node__:heat.limit:generation"] = 3