Give connectors a store that survives restarts, and a way to clear it

`ConnectorNode.store` keeps short strings per connector package in the
engine's database (`connector_state`, migration f8c3e5a91d27), so the nodes of
one package share them and `fluksio serve` needs nothing new: get, set and
delete, at most 64 kB a value and 256 keys a package. A superuser lists the
keys (never the values) and clears them through `/connectors/{node_type}/state`,
or under Stored state in a connector node's panel. Documented in the connector
contract.

backend/fluksio/alembic/versions/f8c3e5a91d27_connector_state.py

@@ -0,0 +1,38 @@
+"""connector_state
+
+A place for a connector to keep what has to outlive a restart, such as the
+certificate a Hue bridge presented on first contact. Keyed by the connector's
+package, so the nodes of one package share it.
+
+Revision ID: f8c3e5a91d27
+Revises: d5a3c8e1f246
+Create Date: 2026-10-06
+
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "f8c3e5a91d27"
+down_revision = "d5a3c8e1f246"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "connector_state",
+        sa.Column(
+            "connector", sqlmodel.sql.sqltypes.AutoString(length=100), nullable=False
+        ),
+        sa.Column("key", sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False),
+        sa.Column("value", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("connector", "key"),
+    )
+
+
+def downgrade():
+    op.drop_table("connector_state")

backend/fluksio/api/main.py

@@ -5,6 +5,7 @@ from fluksio.api.routes import (
     alerts,
     artifacts,
     cloud,
+    connectors,
     dashboards,
     flavors,
     flows,
@@ -31,6 +32,7 @@ api_router.include_router(flavors.router)
 api_router.include_router(flows.router)
 api_router.include_router(flows.ws_router)
 api_router.include_router(secrets.router)
+api_router.include_router(connectors.router)
 api_router.include_router(alerts.router)
 api_router.include_router(dashboards.router)
 api_router.include_router(panels.router)

backend/fluksio/api/routes/connectors.py

@@ -0,0 +1,79 @@
+"""What a connector keeps across restarts, for a superuser to see and reset.
+
+A connector writes these itself (``ConnectorNode.store``). This is the way back
+when one of them is wrong: a Hue bridge replaced, so the certificate pinned for
+it no longer matches. Values are never handed out, since a connector may keep a
+token here; the keys say enough to decide what to clear.
+
+A connector is named by any node type it installs, as the canvas knows it, and
+its keys are shared by every node type of the same package.
+"""
+
+from datetime import datetime
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from fluksio.api.deps import get_current_active_superuser
+from fluksio.flow.connector import ConnectorNode
+from fluksio.flow.connector_store import KEY, ConnectorStore, namespace_of
+from fluksio.flow.controller import NODE_TYPES
+from fluksio.models import Message
+
+router = APIRouter(
+    prefix="/connectors",
+    tags=["connectors"],
+    dependencies=[Depends(get_current_active_superuser)],
+)
+
+
+class StoredKey(BaseModel):
+    key: str
+    #: Bytes, so a superuser can tell a fingerprint from a blob.
+    size: int
+    updated_at: datetime
+
+
+class ConnectorStatePublic(BaseModel):
+    #: The package the keys belong to.
+    connector: str
+    data: list[StoredKey]
+    count: int
+
+
+def _store(node_type: str) -> ConnectorStore:
+    """The store of the connector that installed *node_type*, or a 404."""
+    known = NODE_TYPES.get(node_type)
+    cls = known.cls if known is not None else None
+    if not (isinstance(cls, type) and issubclass(cls, ConnectorNode)):
+        raise HTTPException(status_code=404, detail=f"No connector named '{node_type}'")
+    return ConnectorStore(namespace_of(cls))
+
+
+@router.get("/{node_type}/state", response_model=ConnectorStatePublic)
+def read_connector_state(node_type: str) -> Any:
+    """The keys this connector keeps, without their values."""
+    store = _store(node_type)
+    keys = [
+        StoredKey(key=row.key, size=len(row.value.encode()), updated_at=row.updated_at)
+        for row in store.entries()
+    ]
+    return ConnectorStatePublic(connector=store.connector, data=keys, count=len(keys))
+
+
+@router.delete("/{node_type}/state", response_model=Message)
+def clear_connector_state(node_type: str) -> Any:
+    """Forget everything this connector keeps. It starts over on next use."""
+    store = _store(node_type)
+    cleared = store.clear()
+    return Message(message=f"Cleared {cleared} key(s) of {store.connector}")
+
+
+@router.delete("/{node_type}/state/{key}", response_model=Message)
+def delete_connector_state(node_type: str, key: str) -> Any:
+    """Forget one key, such as the certificate pinned for one bridge."""
+    store = _store(node_type)
+    if not KEY.fullmatch(key) or not store.delete(key):
+        raise HTTPException(status_code=404, detail=f"No key '{key}' stored")
+    return Message(message=f"Deleted '{key}' of {store.connector}")

backend/fluksio/flow/connector.py

@@ -20,7 +20,10 @@ What a connector gets from the base class:
   reading it;
 * :meth:`ConnectorNode.save_artifact`, for a device whose readings are bytes —
   a camera frame, a recorded clip — which travel as a reference on a media
-  port rather than as the message itself.
+  port rather than as the message itself;
+* :attr:`ConnectorNode.store`, for the few short strings that have to outlive
+  a restart — a certificate pinned on first contact, a pairing token — shared
+  by every node of the connector's package.
 
 The message schemas and the parameter model are the rest of the contract, and
 they are the same ones the built-in nodes use. See ``docs/connectors/`` for the
@@ -35,6 +38,7 @@ from typing import TYPE_CHECKING, Any, ClassVar
 
 from pydantic import BaseModel, Field
 
+from fluksio.flow.connector_store import ConnectorStore, namespace_of
 from fluksio.flow.nodes import Node
 
 if TYPE_CHECKING:
@@ -92,6 +96,7 @@ class ConnectorNode(Node):
         "_stop_event",
         "_last_published",
         "_artifacts",
+        "_store",
         "_down",
     )
 
@@ -102,6 +107,7 @@ class ConnectorNode(Node):
         self._stop_event: asyncio.Event | None = None
         self._last_published: dict[str, Any] = {}
         self._artifacts: ArtifactStore | None = None
+        self._store: ConnectorStore | None = None
         self._down = False
 
     def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None:
@@ -166,6 +172,19 @@ class ConnectorNode(Node):
             [data], name=name, media_type=media_type, volatile=volatile
         )
 
+    @property
+    def store(self) -> ConnectorStore:
+        """Short strings this connector keeps across restarts.
+
+        ``get(key)``, ``set(key, value)`` and ``delete(key)``, namespaced by the
+        connector's top-level package, so every node of the package sees the
+        same keys. Blocking: call it through ``asyncio.to_thread`` from the
+        event loop. Available before :meth:`start`, unlike the artifact store.
+        """
+        if self._store is None:
+            self._store = ConnectorStore(namespace_of(type(self)))
+        return self._store
+
     async def start(self, app: FastAPI | None = None) -> None:
         self._artifacts = getattr(app.state, "artifact_store", None) if app else None
         if self.config.poll_interval > 0 and self._stop_event is None:

backend/fluksio/flow/connector_store.py

@@ -0,0 +1,139 @@
+"""What a connector keeps across restarts.
+
+A connector holds connections and handles, and those are meant to go when the
+process does. A few things are not: the certificate a device presented on first
+contact, a token it paired with. :attr:`ConnectorNode.store` is where those go —
+a handful of short strings per connector package, in the engine's own database,
+so they survive a restart and need nothing ``fluksio serve`` does not have.
+
+Keyed by the connector's top-level package rather than by node: the nodes of
+one Hue bridge share the certificate pinned for it, and every node type a
+package installs shares the same keys. A superuser lists and clears them
+through ``/connectors/{node_type}/state``.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import UTC, datetime
+
+from sqlalchemy import delete, func
+from sqlmodel import Session, col, select
+
+from fluksio.models import ConnectorState
+
+#: A value is a short string — a fingerprint, a PEM, a token. Bytes worth
+#: keeping belong in the artifact store.
+MAX_VALUE_BYTES = 64 * 1024
+#: Keys per connector package. Past this a connector is keeping something per
+#: reading rather than per device.
+MAX_KEYS = 256
+#: What a top-level package can be called.
+CONNECTOR_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,99}")
+#: One URL path segment, so the reset route can name every key.
+KEY = re.compile(r"[^/\s]{1,200}")
+
+
+def namespace_of(cls: type) -> str:
+    """The package a connector class keeps its state under."""
+    return cls.__module__.partition(".")[0]
+
+
+def _checked(key: str) -> str:
+    if not KEY.fullmatch(key):
+        raise ValueError(
+            f"'{key}' is not a valid key: 1 to 200 characters, no slash or whitespace"
+        )
+    return key
+
+
+class ConnectorStore:
+    """Short strings one connector package keeps across restarts.
+
+    Every call is a database round trip, so from the event loop it goes
+    through ``asyncio.to_thread`` like any other blocking I/O.
+    """
+
+    def __init__(self, connector: str) -> None:
+        if not CONNECTOR_NAME.fullmatch(connector):
+            raise ValueError(f"'{connector}' is not a connector package name")
+        self.connector = connector
+
+    def get(self, key: str) -> str | None:
+        """The value stored under *key*, or ``None``."""
+        # Imported here: a connector imports this module, and its tests have
+        # no business opening the engine's database by importing it.
+        from fluksio.core.db import engine
+
+        with Session(engine) as session:
+            row = session.get(ConnectorState, (self.connector, _checked(key)))
+        return None if row is None else row.value
+
+    def set(self, key: str, value: str) -> None:
+        """Store *value* under *key*, replacing what was there."""
+        from fluksio.core.db import engine, writing
+
+        _checked(key)
+        if len(value.encode()) > MAX_VALUE_BYTES:
+            raise ValueError(
+                f"'{key}' is too large to store: {MAX_VALUE_BYTES} bytes at most"
+            )
+        with Session(engine) as session, writing(session):
+            row = session.get(ConnectorState, (self.connector, key))
+            if row is None:
+                count = session.exec(
+                    select(func.count())
+                    .select_from(ConnectorState)
+                    .where(ConnectorState.connector == self.connector)
+                ).one()
+                if count >= MAX_KEYS:
+                    raise ValueError(
+                        f"{self.connector} already stores {MAX_KEYS} keys, the most "
+                        "a connector may keep"
+                    )
+                row = ConnectorState(connector=self.connector, key=key, value=value)
+            else:
+                row.value = value
+                row.updated_at = datetime.now(UTC)
+            session.add(row)
+            session.commit()
+
+    def delete(self, key: str) -> bool:
+        """Forget *key*. Whether there was anything to forget."""
+        from fluksio.core.db import engine
+
+        with Session(engine) as session:
+            result = session.exec(
+                delete(ConnectorState).where(
+                    col(ConnectorState.connector) == self.connector,
+                    col(ConnectorState.key) == _checked(key),
+                )
+            )
+            session.commit()
+        return bool(result.rowcount)
+
+    def entries(self) -> list[ConnectorState]:
+        """Every key this package keeps, in key order."""
+        from fluksio.core.db import engine
+
+        with Session(engine) as session:
+            return list(
+                session.exec(
+                    select(ConnectorState)
+                    .where(ConnectorState.connector == self.connector)
+                    .order_by(col(ConnectorState.key))
+                ).all()
+            )
+
+    def clear(self) -> int:
+        """Forget every key. How many there were."""
+        from fluksio.core.db import engine
+
+        with Session(engine) as session:
+            result = session.exec(
+                delete(ConnectorState).where(
+                    col(ConnectorState.connector) == self.connector
+                )
+            )
+            session.commit()
+        return int(result.rowcount)

backend/fluksio/models.py

@@ -544,3 +544,25 @@ class FlavorPublic(FlavorBase):
 class FlavorsPublic(SQLModel):
     data: list[FlavorPublic]
     count: int
+
+
+# -----------------------------------------------------------------------------
+# Connector state
+#
+# What a connector has to remember across restarts and has no other home for:
+# a certificate it pinned on first contact, a pairing token a device handed it.
+# Small values by design — `fluksio.flow.connector_store` caps them.
+# -----------------------------------------------------------------------------
+
+
+class ConnectorState(SQLModel, table=True):
+    """One value a connector package stored under a key of its own."""
+
+    __tablename__ = "connector_state"
+
+    #: The connector's top-level package, so every node type it installs and
+    #: every node of those shares one set of keys.
+    connector: str = Field(primary_key=True, max_length=100)
+    key: str = Field(primary_key=True, max_length=200)
+    value: str
+    updated_at: datetime = Field(default_factory=get_datetime_utc, sa_type=UTCDateTime)

backend/tests/api/routes/test_connectors.py

@@ -0,0 +1,74 @@
+"""Resetting what a connector keeps: a superuser's, named by node type."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.core.config import settings
+from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode
+from fluksio.flow.controller import NODE_TYPES, NodeType
+
+URL = f"{settings.API_V1_STR}/connectors"
+
+
+class Bridge(ConnectorNode):
+    contract = CONTRACT_VERSION
+
+
+@pytest.fixture
+def bridge(monkeypatch):
+    monkeypatch.setitem(NODE_TYPES, "test_bridge", NodeType("Bridge", "", Bridge))
+    store = Bridge(name="b").store
+    store.set("cert:10.0.0.2", "PEM")
+    store.set("cert:10.0.0.3", "PEM")
+    yield store
+    store.clear()
+
+
+def test_a_superuser_lists_and_clears_the_keys(
+    client: TestClient, superuser_token_headers: dict[str, str], bridge
+) -> None:
+    listed = client.get(f"{URL}/test_bridge/state", headers=superuser_token_headers)
+    assert listed.status_code == 200
+    body = listed.json()
+    assert body["connector"] == "tests"
+    assert [row["key"] for row in body["data"]] == ["cert:10.0.0.2", "cert:10.0.0.3"]
+    assert "value" not in body["data"][0]
+
+    one = client.delete(
+        f"{URL}/test_bridge/state/cert:10.0.0.2", headers=superuser_token_headers
+    )
+    assert one.status_code == 200
+    assert bridge.get("cert:10.0.0.2") is None
+
+    rest = client.delete(f"{URL}/test_bridge/state", headers=superuser_token_headers)
+    assert rest.status_code == 200
+    assert bridge.entries() == []
+
+
+def test_anyone_else_is_refused(
+    client: TestClient, normal_user_token_headers: dict[str, str], bridge
+) -> None:
+    for method in ("GET", "DELETE"):
+        response = client.request(
+            method, f"{URL}/test_bridge/state", headers=normal_user_token_headers
+        )
+        assert response.status_code == 403
+    assert bridge.get("cert:10.0.0.2") == "PEM"
+
+
+@pytest.mark.parametrize("path", ["python/state", "nothing/state", "..%2F../state"])
+def test_only_a_connector_has_state(
+    client: TestClient, superuser_token_headers: dict[str, str], path: str
+) -> None:
+    response = client.get(f"{URL}/{path}", headers=superuser_token_headers)
+    assert response.status_code == 404
+
+
+@pytest.mark.usefixtures("bridge")
+def test_a_key_that_is_not_there_is_a_404(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.delete(
+        f"{URL}/test_bridge/state/cert:nowhere", headers=superuser_token_headers
+    )
+    assert response.status_code == 404

backend/tests/test_connector_store.py

@@ -0,0 +1,65 @@
+"""What a connector keeps across restarts: per package, small, and durable."""
+
+import pytest
+
+from fluksio.flow import connector_store
+from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode
+from fluksio.flow.connector_store import MAX_VALUE_BYTES, ConnectorStore
+
+
+class Bridge(ConnectorNode):
+    contract = CONTRACT_VERSION
+
+
+@pytest.fixture
+def store():
+    store = ConnectorStore("store_test")
+    yield store
+    store.clear()
+
+
+def test_the_nodes_of_one_package_share_its_keys():
+    first, second = Bridge(name="a"), Bridge(name="b")
+    try:
+        first.store.set("cert:10.0.0.2", "PEM")
+        assert first.store.connector == "tests"
+        assert second.store.get("cert:10.0.0.2") == "PEM"
+        assert ConnectorStore("other").get("cert:10.0.0.2") is None
+    finally:
+        first.store.clear()
+
+
+def test_a_value_survives_a_fresh_store(store):
+    store.set("token", "one")
+    store.set("token", "two")
+
+    again = ConnectorStore("store_test")
+    assert again.get("token") == "two"
+    assert again.delete("token") is True
+    assert again.delete("token") is False
+    assert store.get("token") is None
+
+
+def test_a_value_or_a_key_count_past_the_limit_is_refused(store, monkeypatch):
+    with pytest.raises(ValueError, match="too large"):
+        store.set("big", "x" * (MAX_VALUE_BYTES + 1))
+
+    monkeypatch.setattr(connector_store, "MAX_KEYS", 2)
+    store.set("one", "1")
+    store.set("two", "2")
+    with pytest.raises(ValueError, match="2 keys"):
+        store.set("three", "3")
+    # Replacing a key that is already there is not a new one.
+    store.set("two", "22")
+    assert [row.key for row in store.entries()] == ["one", "two"]
+
+
+@pytest.mark.parametrize("key", ["", "a/b", "a b", "x" * 201])
+def test_a_key_the_reset_route_could_not_name_is_refused(store, key):
+    with pytest.raises(ValueError, match="not a valid key"):
+        store.set(key, "v")
+
+
+def test_a_package_name_is_a_name():
+    with pytest.raises(ValueError):
+        ConnectorStore("../flows")