diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 26c7efb..6fd370d 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/backend/fluksio/flow/dashboards.py @@ -961,21 +961,23 @@ class DashboardStore: self._dump(shown.model_copy(update={"version": top + 1})) ) draft.unlink() - self.flows._commit(f"Discard draft of dashboard '{name}'") - return self.read(name) + shown = self.read(name) + self.flows._commit(f"Discard draft of dashboard {named(shown.name, name)}") + return shown def delete(self, name: str) -> None: if not self.exists(name): raise DashboardNotFound(name) path = self._file(name) with self.flows._write_lock: + key = next((k for k, i in self.key_index().items() if i == name), name) path.unlink(missing_ok=True) self._draft_file(name).unlink(missing_ok=True) try: path.parent.rmdir() except OSError: pass - self.flows._commit(f"Delete dashboard '{name}'") + self.flows._commit(f"Delete dashboard {named(key, name)}") def duplicate(self, name: str) -> DashboardDef: """Copy the working copy into a new dashboard, as a draft only. diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py index 74c8d8f..fcd0133 100644 --- a/backend/fluksio/flow/state.py +++ b/backend/fluksio/flow/state.py @@ -639,6 +639,14 @@ class RedisState(StateBackend): names = [k for k in keys if not k.startswith("__")] if names: pipe.sadd(self._names_key(), *names) + if self._ttl: + pipe.expire(self._names_key(), self._ttl) + + def _incr(self, pipe: Any, key: str) -> None: + """INCR a counter in the caller's pipeline, expiring with the values.""" + pipe.incr(self._key(key)) + if self._ttl: + pipe.expire(self._key(key), self._ttl) def set(self, key: str, value: Any) -> None: data = self._serialize(value) @@ -747,7 +755,9 @@ class RedisState(StateBackend): def increment(self, key: str) -> int: """Atomically increment a counter using Redis INCR.""" - return cast(int, self._client.incr(self._key(key))) + pipe = self._client.pipeline() + self._incr(pipe, key) + return cast(int, pipe.execute()[0]) def increment_multi(self, keys: list[str]) -> None: """Bump every counter in one round trip rather than one INCR each.""" @@ -755,7 +765,7 @@ class RedisState(StateBackend): return pipe = self._client.pipeline() for key in keys: - pipe.incr(self._key(key)) + self._incr(pipe, key) pipe.execute() def get_multi(self, keys: list[str]) -> dict[str, Any]: @@ -916,7 +926,7 @@ class RedisState(StateBackend): if self._ttl: pipe.expire(history_key, self._ttl) for key in counters: - pipe.incr(self._key(key)) + self._incr(pipe, key) self._note(pipe, values) pipe.execute() diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 9f9b58c..f3f117b 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -202,12 +202,14 @@ def new_id(kind: Literal["f", "d"], taken: Callable[[str], bool]) -> str: class FlowNotFound(KeyError): - def __init__(self, name: str) -> None: + def __init__(self, name: str, published: bool = False) -> None: super().__init__(name) self.name = name + self.published = published def __str__(self) -> str: - return f"No flow named '{self.name}'" + what = "published flow" if self.published else "flow" + return f"No {what} with the id '{self.name}'" class FlowExists(ValueError): @@ -720,7 +722,7 @@ class FlowStore: return self._load(path) path = self._flow_file(name) if not path.exists(): - raise FlowNotFound(name) + raise FlowNotFound(name, published=not draft) return self._load(path) def key_index(self) -> dict[str, str]: @@ -782,7 +784,9 @@ class FlowStore: and returned as a copy so a caller sorting it cannot disturb the next. """ with self._read_all_lock: - if self._read_all_at != self.revision: + # Taken before scanning, so a commit landing meanwhile scans again. + at = self.revision + if self._read_all_at != at: flows = [] for path in sorted(self.root.glob("*/flow.json")): name = path.parent.name @@ -791,7 +795,7 @@ class FlowStore: except Exception: logger.exception("Skipping unreadable flow '%s'", name) self._read_all = flows - self._read_all_at = self.revision + self._read_all_at = at return list(self._read_all) def write_flow(self, flow: FlowDef) -> bool: diff --git a/backend/scripts/bench_startup.py b/backend/scripts/bench_startup.py index 3610ee8..6bb3a46 100644 --- a/backend/scripts/bench_startup.py +++ b/backend/scripts/bench_startup.py @@ -96,11 +96,15 @@ def ensure_flow(api: Api) -> None: api.call( "PUT", f"/api/v1/flows/{BENCH_FLOW}/nodes/noop/source", {"code": NODE_SOURCE} ) - api.call( - "POST", - f"/api/v1/flows/{BENCH_FLOW}/publish", - {"version": definition["version"]}, - ) + # Read again, since the saves move the version. A flow an interrupted run + # left published, and unchanged since, has nothing to publish. + current = api.call("GET", f"/api/v1/flows/{BENCH_FLOW}") + if current["has_draft"]: + api.call( + "POST", + f"/api/v1/flows/{BENCH_FLOW}/publish", + {"version": current["definition"]["version"]}, + ) def time_runs(api: Api, count: int) -> tuple[list[float], list[float]]: diff --git a/backend/tests/flow/test_redis_state.py b/backend/tests/flow/test_redis_state.py new file mode 100644 index 0000000..525c090 --- /dev/null +++ b/backend/tests/flow/test_redis_state.py @@ -0,0 +1,34 @@ +"""RedisState against a real server, enabled by FLUKSIO_TEST_REDIS_HOST (and +optional PORT).""" + +import os +import uuid + +import pytest + +from fluksio.flow.state import RedisState + + +@pytest.fixture +def state(): + host = os.environ.get("FLUKSIO_TEST_REDIS_HOST") + if not host: + pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks") + port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379")) + state = RedisState(host, port, namespace=f"test-state-{uuid.uuid4().hex}", ttl=600) + try: + yield state + finally: + state.clear() + state.close() + + +def test_a_ttl_expires_counters_and_the_names_set(state): + state.set("heat.temp", 21) + state.increment("__version__:heat.a") + state.increment_multi(["__version__:heat.b"]) + state.record({"heat.v": 1}, {}, ["__version__:heat.c"], 0.0) + + keys = state.keys() + assert "__names__" in keys and "__version__:heat.c" in keys + assert all(state._client.ttl(state._key(k)) > 0 for k in keys)