FlowNotFound says whether a published flow or any flow is missing; read_all stamps its cache with the revision from before its scan; dashboard discard and delete commits name the key; a RedisState ttl expires counters and the names set too; bench_startup publishes the version after its saves, and only when there is a draft

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.

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()
 

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:

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]]:

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)