diff --git a/.env.example b/.env.example index 9a0b2af..f183ba5 100644 --- a/.env.example +++ b/.env.example @@ -15,7 +15,9 @@ PROJECT_NAME="Fluksio" STACK_NAME=fluksio-app # Backend -BACKEND_CORS_ORIGINS="http://localhost,http://localhost:5173,http://app.localhost,https://localhost,https://localhost:5173" +# `http://frontend` is the compose-network origin the Playwright job loads the +# SPA from (docker/compose.ci.yml); without it every spec fails CORS at login. +BACKEND_CORS_ORIGINS="http://localhost,http://localhost:5173,http://app.localhost,https://localhost,https://localhost:5173,http://frontend" SECRET_KEY=changethis FIRST_SUPERUSER=admin@fluksio.com FIRST_SUPERUSER_PASSWORD=changethis @@ -36,6 +38,11 @@ POSTGRES_DB=app POSTGRES_USER=postgres POSTGRES_PASSWORD=changethis +# Redis — flow engine state. Empty keeps the state in memory; the compose stack +# points the backend at its own `redis` service. +REDIS_HOST= +REDIS_PORT=6379 + SENTRY_DSN= # Docker registry images diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index 0cdfcc0..2970073 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -32,6 +32,7 @@ class ValidationIssue(BaseModel): "unconnected_input", "missing_initial_value", "node_error", + "unauthenticated_hook", ] message: str flow: str = "" @@ -243,7 +244,7 @@ class Pipeline: def _timestamp_key(self, msg_name: str) -> str: return f"__ts__:{msg_name}" - def _increment_message_versions(self, outputs: dict) -> None: + def _increment_message_versions(self, outputs: dict[str, Any]) -> None: for msg_name in outputs: self._state.increment(self._version_key(msg_name)) @@ -305,7 +306,7 @@ class Pipeline: if self._events is not None: self._events.publish(event) - def _execute_node(self, node: Node, state: StateBackend) -> dict | None: + def _execute_node(self, node: Node, state: StateBackend) -> dict[str, Any] | None: """Run one node and record its outputs. Never raises.""" started = time.perf_counter() try: @@ -378,7 +379,7 @@ class Pipeline: submitted: set[Node] = set() skipped: set[Node] = set() - node_futures: dict[Node, Future] = {} + node_futures: dict[Node, Future[dict[str, Any] | None]] = {} def is_ready(n: Node) -> bool: if in_degree[n] != 0: @@ -416,7 +417,7 @@ class Pipeline: return state def run( - self, inputs: dict | None = None, nodes: set[Node] | None = None + self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None ) -> StateBackend: """Execute the graph (or one flow's nodes) against the shared state.""" if inputs: @@ -426,7 +427,7 @@ class Pipeline: return self._execute_parallel(nodes, self._state, check_ready=False) - def trigger(self, node: Node, outputs: dict | None) -> StateBackend: + def trigger(self, node: Node, outputs: dict[str, Any] | None) -> StateBackend: """Publish a node's outputs and run everything downstream of it.""" state = self._state diff --git a/backend/app/flow/state.py b/backend/app/flow/state.py index 42d1ca1..f0589df 100644 --- a/backend/app/flow/state.py +++ b/backend/app/flow/state.py @@ -12,7 +12,7 @@ from abc import ABC, abstractmethod from collections.abc import Iterator from contextlib import contextmanager from threading import RLock -from typing import Any +from typing import Any, cast import redis @@ -90,7 +90,7 @@ class StateBackend(ABC): @abstractmethod @contextmanager - def lock(self, timeout: float = 10.0): + def lock(self, timeout: float = 10.0) -> Iterator[None]: """ Acquire a lock for thread-safe operations. @@ -211,7 +211,7 @@ class MemoryState(StateBackend): self._data.update(mapping) @contextmanager - def lock(self, timeout: float = 10.0): + def lock(self, timeout: float = 10.0) -> Iterator[None]: """Acquire the internal lock.""" acquired = self._lock.acquire(timeout=timeout) if not acquired: @@ -224,7 +224,7 @@ class MemoryState(StateBackend): def increment(self, key: str) -> int: """Atomically increment a counter and return the new value.""" with self._lock: - value = self._data.get(key, 0) + value: int = self._data.get(key, 0) new_value = value + 1 self._data[key] = new_value return new_value @@ -313,8 +313,10 @@ class RedisState(StateBackend): """Deserialize value from storage.""" return json.loads(data) if data else None + # redis-py types every command as a sync/async union; this is the + # synchronous client, so the results are narrowed where they are consumed. def get(self, key: str, default: Any = None) -> Any: - data = self._client.get(self._key(key)) + data = cast(bytes | None, self._client.get(self._key(key))) return self._deserialize(data) if data is not None else default def set(self, key: str, value: Any) -> None: @@ -332,7 +334,10 @@ class RedisState(StateBackend): pattern = f"{self._namespace}:*" cursor = 0 while True: - cursor, keys = self._client.scan(cursor, match=pattern, count=100) + cursor, keys = cast( + tuple[int, list[bytes]], + self._client.scan(cursor, match=pattern, count=100), + ) if keys: self._client.delete(*keys) if cursor == 0: @@ -345,7 +350,10 @@ class RedisState(StateBackend): result = [] cursor = 0 while True: - cursor, keys = self._client.scan(cursor, match=pattern, count=100) + cursor, keys = cast( + tuple[int, list[bytes]], + self._client.scan(cursor, match=pattern, count=100), + ) for k in keys: key_str = k.decode("utf-8") # Skip the lock key @@ -369,7 +377,7 @@ class RedisState(StateBackend): pipe.execute() @contextmanager - def lock(self, timeout: float = 10.0): + def lock(self, timeout: float = 10.0) -> Iterator[None]: """ Acquire a distributed lock using Redis. @@ -402,13 +410,13 @@ class RedisState(StateBackend): :rtype: bool """ try: - return self._client.ping() + return bool(self._client.ping()) except redis.ConnectionError: return False def increment(self, key: str) -> int: """Atomically increment a counter using Redis INCR.""" - return self._client.incr(self._key(key)) + return cast(int, self._client.incr(self._key(key))) def get_multi(self, keys: list[str]) -> dict[str, Any]: """Get multiple values atomically using Redis MGET.""" @@ -416,7 +424,7 @@ class RedisState(StateBackend): return {} full_keys = [self._key(k) for k in keys] - values = self._client.mget(full_keys) + values = cast(list[bytes | None], self._client.mget(full_keys)) result = {} for key, value in zip(keys, values, strict=True): diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 4f25839..994a1ba 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -44,11 +44,11 @@ build-backend = "hatchling.build" strict = true exclude = ["venv", ".venv", "alembic"] -# app/flow/nodes.py still carries the prototype's node classes, and state.py's -# redis calls type as sync/async unions under the current stubs. Both are being -# revisited; every module around them is checked strictly. +# app/flow/nodes.py still carries the prototype's node classes: the integration +# nodes need annotations of their own, plus stubs for croniter and influxdb_client. +# It is being revisited per integration; every module around it is checked strictly. [[tool.mypy.overrides]] -module = ["app.flow.nodes", "app.flow.pipeline", "app.flow.state"] +module = ["app.flow.nodes"] ignore_errors = true [tool.ruff] diff --git a/docker/compose.dev.yml b/docker/compose.dev.yml index 9025456..834287e 100644 --- a/docker/compose.dev.yml +++ b/docker/compose.dev.yml @@ -45,6 +45,9 @@ services: ports: - "127.0.0.1:8080:8080" + redis: + restart: "no" + backend: restart: "no" ports: diff --git a/docker/compose.yml b/docker/compose.yml index ee8e11f..cd7c261 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -29,6 +29,27 @@ services: - POSTGRES_USER=${POSTGRES_USER?Variable not set} - POSTGRES_DB=${POSTGRES_DB?Variable not set} + redis: + image: redis:8.2-alpine + container_name: fluksio-redis + restart: always + security_opt: + - no-new-privileges:true + # Flow state, not a cache: append-only persistence keeps the last value of + # every message across a restart. + command: redis-server --appendonly yes + # Deliberately not on `proxy` and not published: only the backend talks to it. + networks: + - default + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + retries: 5 + start_period: 10s + timeout: 5s + volumes: + - app-redis-data:/data + adminer: image: adminer container_name: fluksio-adminer @@ -94,6 +115,9 @@ services: db: condition: service_healthy restart: true + redis: + condition: service_healthy + restart: true prestart: condition: service_completed_successfully env_file: @@ -116,6 +140,8 @@ services: - POSTGRES_USER=${POSTGRES_USER?Variable not set} - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set} - SENTRY_DSN=${SENTRY_DSN} + # Flow state survives a restart in Redis; without a host it stays in memory. + - REDIS_HOST=redis # Flows are files in a git repository; secrets sit encrypted beside it. - FLOWS_DIR=/data/flows - SECRETS_FILE=/data/secrets.enc @@ -183,6 +209,7 @@ services: volumes: app-db-data: + app-redis-data: app-flow-data: networks: