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

Bound what the API accepts, and close the holes the audit found

Browse Source 
**SQLite is the database, and now says so.** `metric_minute` and every run
table are written with `sqlalchemy.dialects.sqlite.insert(...)
.on_conflict_do_update` and with `max(a, b)`, neither of which another
dialect has — so pointing `DATABASE_URL` at Postgres migrated cleanly,
served, logged in, and then lost every observability flush into the
collector's hold buffer and failed every run. It refuses at startup
instead. (The Postgres in the compose stack is Umami's; the engine's own
database has been a file beside the flows since 2026-08-21.)

**Every integer query parameter is bounded.** The caps were written as
`min(limit, 500)`, which a negative walks straight through — `?limit=-1`
compiles to `LIMIT -1` and SQLite returns the whole table. Ten signatures,
now `Query(ge=…, le=…)`. `hours=0` still means an hour, which
`_window_hours` was already deliberate about.

**Exports are capped at 10 000 runs** and say so with `X-Truncated`. The
filters bounded a sensible request and nothing bounded an unfiltered one,
which read every row into memory before a byte was streamed. `_series`
resolves cached curves in two queries rather than a `Run` lookup and a
`RunMetric` query per restored node — a comparison of twenty runs was
calling that twenty times over.

**`PUT /artifacts` has a size limit** (`MAX_ARTIFACT_BYTES`, 2 GiB, 0 to
disable), checked against `Content-Length` and again against the stream for
a chunked body, and its writes moved off the event loop.

**`/observability/timeseries` takes `since`/`until`**, the same window
`/runs` and `/events` take, capped at 2000 points — `hours=720&bucket_s=60`
was 43 200 of them in one array. It is also what a dragged chart needs to
re-fetch at its own resolution rather than magnifying buckets it has.

**Composite indexes** for the three list screens: `run(flow, created_at)`
and `(status, created_at)`, `flow_run(flow, started_at)`,
`engine_event(type, ts)`. Every index was single-column, so SQLite picked
one and sorted the rest by hand. Verified against a copy of a live database
(250k `flow_run` rows): the planner takes all four.

**Redis clients have socket timeouts.** A Redis that stops answering
without closing the connection hung the caller until the kernel gave up —
including `/utils/health/`, whose job is to notice.

**The panels file is written under one lock.** `save_panels` and
`unpair_panel` are both read-modify-write, and a save that read before an
unpair wrote put the old nonce back — silently un-revoking a screen that
had just been unpaired. The nonce carry-forward was written to make that
impossible; the gap between its read and its write is where it happened.

**Startup releases what it acquired.** Everything past `event_bus.bind`
registers how to close itself and the `finally` walks that list backwards;
a failure part-way through used to reach none of the shutdown steps and
leave the worker pool's subprocesses and every background task behind —
under `--reload`, once per bad edit. `modules.reconcile` moved into the
background: `uv` gets five minutes twice over, the healthcheck allows
eighty seconds, and the autoheal restarted the container before it could
finish installing.

`delete_run` takes SQLite's write lock up front (`core.db.writing`) rather
than upgrading a deferred transaction and losing to whichever flush
committed in between. `modules.sync` is serialised — two applies mutated
one venv at once. The proxied-call and stream dicts are bounded, and a
reused id cancels its predecessor instead of dropping the reference.

Security, found in passing and small enough to fix here:

- **`/secrets/` required only a signed-in user.** The names alone say what
  this installation talks to, and `PUT /{name}` takes any name, so any
  account could overwrite the credential a flow authenticates with.
  Superuser now — which `/search` already assumed and said so.
- **`POST /login/access-token` had no rate limit.** Argon2 is deliberately
  expensive and the route is unauthenticated and runs in the shared
  threadpool. Ten *failed* attempts per address per five minutes; a
  successful sign-in spends nothing.
- **a password reset link worked repeatedly for 48 hours.** The token now
  carries a digest of the password hash it was minted against, so it stops
  verifying once it has set one. No table of spent tokens needed.
- **enrolment accepted `http://`**, sending the claim code and then this
  installation's credential in clear. https, or a local address.
- the rate limiter read `request.client.host`, which behind Traefik is the
  proxy — so every per-address limit was one global bucket and one caller
  could lock out everyone. It reads the forwarded address, and its
  bucket table is capped rather than growing one key per address forever.
- SMTP has a timeout and sends after the response, so an unreachable mail
  host cannot pin a threadpool worker, and a reply's timing no longer says
  whether the address exists.

Test suite: engine-written rows are cleared between modules. A `FlowRun`
left `running` by one module turned up in another's query. Per-test
rollback is not available here — the module-scoped `client` runs the real
lifespan and its collector and run service write through sessions of their
own — so this bounds it where the writes come from. Three consecutive
green runs, orders randomised.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T
This commit is contained in:
stroblme
2026-08-29 20:40:05 +02:00
co-authored by Claude Opus 5
parent 1069247085
commit 57eace2226
24 changed files with 821 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/alembic/env.py
+2 -1
View File
@@ -20,7 +20,7 @@ if config.config_file_name is not None:
# target_metadata = None
from fluksio.models import SQLModel # noqa
from fluksio.core.config import settings # noqa
from fluksio.core.config import settings # noqa
from fluksio.core.types import UTCDateTime # noqa
target_metadata = SQLModel.metadata
@@ -38,6 +38,7 @@ def render_item(type_, obj, autogen_context):
return "sa.DateTime(timezone=True)"
return False
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
backend/fluksio/alembic/versions/e7d3b1a9c624_list_indexes.py
+37
View File
@@ -0,0 +1,37 @@
"""Composite indexes for the three list screens.
Each of these filters on one column and orders by another, and every index on
the table was single-column — so SQLite used one of them and sorted the rest by
hand. `engine_event` is the worst of the three: its filter is `type != 'audit'`,
which no index on `type` can serve at all.
Revision ID: e7d3b1a9c624
Revises: d4a71e9c2b58
Create Date: 2026-08-29
"""
from alembic import op
# revision identifiers, used by Alembic.
revision = "e7d3b1a9c624"
down_revision = "d4a71e9c2b58"
branch_labels = None
depends_on = None
def upgrade() -> None:
# `GET /runs` filters flow and/or status, newest first.
op.create_index("ix_run_flow_created", "run", ["flow", "created_at"])
op.create_index("ix_run_status_created", "run", ["status", "created_at"])
# `GET /observability/runs` does the same over the execution history.
op.create_index("ix_flow_run_flow_started", "flow_run", ["flow", "started_at"])
# `GET /observability/events` excludes audit rows, newest first.
op.create_index("ix_engine_event_type_ts", "engine_event", ["type", "ts"])
def downgrade() -> None:
op.drop_index("ix_engine_event_type_ts", table_name="engine_event")
op.drop_index("ix_flow_run_flow_started", table_name="flow_run")
op.drop_index("ix_run_status_created", table_name="run")
op.drop_index("ix_run_flow_created", table_name="run")
backend/fluksio/api/routes/artifacts.py
+21 -2
View File
@@ -19,6 +19,7 @@ from starlette.concurrency import run_in_threadpool
from fluksio.api.deps import user_from_token
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow.artifacts import ArtifactStore
@@ -84,14 +85,32 @@ async def put_artifact(
Spooled to disk as it arrives rather than buffered: a video segment is as
legitimate a body here as a checkpoint, and neither should have to fit in
memory twice.
memory twice. Capped, because nothing else here was: any account, and any
worker credential, could otherwise fill the data volume.
"""
store = _store(request)
cap = settings.MAX_ARTIFACT_BYTES
declared = request.headers.get("content-length")
if cap and declared and declared.isdigit() and int(declared) > cap:
raise HTTPException(
status_code=413, detail=f"An artifact may be at most {cap} bytes"
)
handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False)
written = 0
try:
with handle:
async for chunk in request.stream():
handle.write(chunk)
written += len(chunk)
# A chunked body declares no length, so the stream is what
# actually holds the limit.
if cap and written > cap:
raise HTTPException(
status_code=413,
detail=f"An artifact may be at most {cap} bytes",
)
# Off the event loop: this is a write syscall per chunk, for
# as long as the upload lasts.
await run_in_threadpool(handle.write, chunk)
return await run_in_threadpool(
store.put_file, Path(handle.name), media_type, name
)
backend/fluksio/api/routes/login.py
+46 -8
View File
@@ -1,18 +1,21 @@
from datetime import timedelta
from secrets import compare_digest
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.security import OAuth2PasswordRequestForm
from fluksio import crud
from fluksio.api.deps import CurrentUser, SessionDep, get_current_active_superuser
from fluksio.api.routes.oauth import too_many
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.models import Message, NewPassword, Token, UserPublic, UserUpdate
from fluksio.utils import (
generate_password_reset_token,
generate_reset_password_email,
password_fingerprint,
send_email,
verify_password_reset_token,
)
@@ -20,17 +23,35 @@ from fluksio.utils import (
router = APIRouter(tags=["login"])
#: Attempts per address per window, and the window. Argon2 is deliberately
#: expensive — tens of megabytes and several passes per verification — and
#: this route is unauthenticated and runs in the shared threadpool, so a
#: burst of guesses is a memory and a concurrency problem before it is a
#: credential one. A person who mistypes twice is nowhere near it.
LOGIN_ATTEMPTS = 10
LOGIN_WINDOW_S = 300.0
@router.post("/login/access-token")
def login_access_token(
session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
request: Request,
session: SessionDep,
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
"""
OAuth2 compatible token login, get an access token for future requests
"""
if too_many(request, "login", LOGIN_ATTEMPTS, LOGIN_WINDOW_S, record=False):
raise HTTPException(
status_code=429, detail="Too many sign-in attempts. Try again shortly."
)
user = crud.authenticate(
session=session, email=form_data.username, password=form_data.password
)
if not user:
# Only the failures are counted, so somebody signing in all day is
# never near the limit and somebody guessing is.
too_many(request, "login", LOGIN_ATTEMPTS, LOGIN_WINDOW_S)
raise HTTPException(status_code=400, detail="Incorrect email or password")
elif not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
@@ -51,7 +72,9 @@ def test_token(current_user: CurrentUser) -> Any:
@router.post("/password-recovery/{email}")
def recover_password(email: str, session: SessionDep) -> Message:
def recover_password(
email: str, session: SessionDep, background: BackgroundTasks
) -> Message:
"""
Password Recovery
"""
@@ -60,11 +83,18 @@ def recover_password(email: str, session: SessionDep) -> Message:
# Always return the same response to prevent email enumeration attacks
# Only send email if user actually exists
if user:
password_reset_token = generate_password_reset_token(email=email)
password_reset_token = generate_password_reset_token(
email=email, hashed_password=user.hashed_password
)
email_data = generate_reset_password_email(
email_to=user.email, email=email, token=password_reset_token
)
send_email(
# After the response, not before it: sending is a synchronous SMTP
# conversation, and doing it inline made the reply slow exactly when
# the address exists — which is the enumeration the identical wording
# below is there to prevent.
background.add_task(
send_email,
email_to=user.email,
subject=email_data.subject,
html_content=email_data.html_content,
@@ -79,15 +109,21 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
"""
Reset password
"""
email = verify_password_reset_token(token=body.token)
if not email:
claims = verify_password_reset_token(token=body.token)
if not claims:
raise HTTPException(status_code=400, detail="Invalid token")
email, fingerprint = claims
user = crud.get_user_by_email(session=session, email=email)
if not user:
# Don't reveal that the user doesn't exist - use same error as invalid token
raise HTTPException(status_code=400, detail="Invalid token")
elif not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
if not compare_digest(fingerprint, password_fingerprint(user.hashed_password)):
# The link already set a password, or the password moved since it was
# sent. Either way this one is spent — a reset link used to work over
# and over for the whole of its 48 hours.
raise HTTPException(status_code=400, detail="Invalid token")
user_in_update = UserUpdate(password=body.new_password)
crud.update_user(
session=session,
@@ -113,7 +149,9 @@ def recover_password_html_content(email: str, session: SessionDep) -> Any:
status_code=404,
detail="The user with this username does not exist in the system.",
)
password_reset_token = generate_password_reset_token(email=email)
password_reset_token = generate_password_reset_token(
email=email, hashed_password=user.hashed_password
)
email_data = generate_reset_password_email(
email_to=user.email, email=email, token=password_reset_token
)
backend/fluksio/api/routes/oauth.py
+37 -5
View File
@@ -26,7 +26,7 @@ import re
import secrets
import time
import uuid
from collections import defaultdict, deque
from collections import OrderedDict, defaultdict, deque
from datetime import UTC, datetime, timedelta
from typing import Any
from urllib.parse import urlencode, urlparse
@@ -81,24 +81,56 @@ def _error(code: str, description: str, status: int = 400) -> JSONResponse:
# enough: the engine is one process, and this is a speed bump, not a boundary.
# -----------------------------------------------------------------------------
_hits: dict[str, deque[float]] = defaultdict(deque)
#: How many distinct buckets are remembered at once. Entries were pruned
#: *within* a deque and never removed, so one key per source address survived
#: for the life of the process.
_MAX_BUCKETS = 4096
_hits: OrderedDict[str, deque[float]] = OrderedDict()
def _too_many(bucket: str, limit: int, window: float) -> bool:
def _too_many(bucket: str, limit: int, window: float, record: bool = True) -> bool:
now = time.monotonic()
seen = _hits[bucket]
seen = _hits.get(bucket)
if seen is None:
seen = _hits[bucket] = deque()
while len(_hits) > _MAX_BUCKETS:
_hits.popitem(last=False)
else:
_hits.move_to_end(bucket)
while seen and now - seen[0] > window:
seen.popleft()
if len(seen) >= limit:
return True
seen.append(now)
if record:
seen.append(now)
return False
def _client_ip(request: Request) -> str:
"""The caller's address, or the proxy's if it did not forward one.
Behind Traefik `request.client.host` is the proxy for every request, which
made every per-address limit here one global bucket — so one caller could
lock out everyone. The leftmost forwarded entry is the client, and is what
the panel pairing screen already reads.
"""
forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
if forwarded:
return forwarded
return request.client.host if request.client else "unknown"
def too_many(
request: Request, action: str, limit: int, window: float, record: bool = True
) -> bool:
"""Per-address rate limit, for anything unauthenticated on any router.
``record=False`` asks the question without spending an attempt, for a
caller that only wants to count the ones that failed.
"""
return _too_many(f"{action}:{_client_ip(request)}", limit, window, record)
# -----------------------------------------------------------------------------
# Helpers
# -----------------------------------------------------------------------------
backend/fluksio/api/routes/observability.py
+34 -11
View File
@@ -8,9 +8,9 @@ cannot render while the engine is degraded is the wrong way round.
"""
from datetime import UTC, datetime, timedelta
from typing import Any, Literal
from typing import Annotated, Any, Literal
from fastapi import APIRouter, Depends, Request
from fastapi import APIRouter, Depends, Query, Request
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel, model_validator
from sqlalchemy import ColumnElement, Integer, cast, func
@@ -141,6 +141,11 @@ def _closed_before(stride: int) -> datetime:
return datetime.fromtimestamp(now // stride * stride, UTC)
#: The most slices one timeseries answers with. `hours=720&bucket_s=60` is
#: 43 200 points into a single JSON array; a chart draws a few hundred.
MAX_POINTS = 2000
def _window_hours(hours: int) -> int:
"""A window the rollups can answer for: an hour at least, retention at most.
@@ -245,14 +250,28 @@ def read_timeseries(
session: SessionDep,
flow: str | None = None,
node: str | None = None,
hours: int = 24,
bucket_s: int = 60,
hours: Annotated[int, Query(ge=0, le=8760)] = 24,
bucket_s: Annotated[int, Query(ge=1, le=86400)] = 60,
since: datetime | None = None,
until: datetime | None = None,
) -> Any:
"""Executions, errors and timings over time, summed across nodes."""
"""Executions, errors and timings over time, summed across nodes.
``hours`` measures back from now, which is what a range picker asks for.
``since`` and ``until`` name a window instead — inclusive and exclusive,
the same pair ``/runs`` and ``/events`` take — so a chart dragged to a
span can be re-fetched at that span's own resolution rather than
magnifying the buckets it already has.
"""
hours = _window_hours(hours)
# The database does the fold: a week of minute rows per node used to cross
# the wire on every poll, and only the slices need to.
stride = max(60, bucket_s)
if since is not None and until is not None:
# Enough slices to draw with, and never so many that the response is
# the problem the fold was there to solve.
span = (_aware(until) - _aware(since)).total_seconds()
stride = max(stride, int(span // MAX_POINTS) + 1)
slot = (_epoch(col(MetricBucket.bucket)) // stride * stride).label("slot")
statement = sa_select(
slot,
@@ -264,8 +283,8 @@ def read_timeseries(
func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
func.sum(col(MetricBucket.items)).label("items"),
).where(
col(MetricBucket.bucket) >= _since(hours),
col(MetricBucket.bucket) < _closed_before(stride),
col(MetricBucket.bucket) >= (_aware(since) if since else _since(hours)),
col(MetricBucket.bucket) < (_aware(until) if until else _closed_before(stride)),
)
if flow:
statement = statement.where(col(MetricBucket.flow) == flow)
@@ -287,7 +306,9 @@ def read_timeseries(
@router.get("/flows", response_model=list[FlowRollup])
def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
def read_flow_rollups(
session: SessionDep, hours: Annotated[int, Query(ge=0, le=8760)] = 24
) -> Any:
"""One row per flow, with a coarse trend of how much it ran."""
hours = _window_hours(hours)
window = hours * 3600
@@ -388,7 +409,7 @@ def read_runs(
status: str | None = None,
since: datetime | None = None,
until: datetime | None = None,
limit: int = 50,
limit: Annotated[int, Query(ge=1, le=200)] = 50,
) -> Any:
"""Recent cascades, newest first, and how many there were in total.
@@ -432,7 +453,7 @@ def read_events(
run: str | None = None,
since: datetime | None = None,
until: datetime | None = None,
limit: int = 100,
limit: Annotated[int, Query(ge=1, le=500)] = 100,
) -> Any:
"""What went wrong, or who changed what. Newest first.
@@ -459,7 +480,9 @@ def read_events(
@router.get("/dead-letter", response_model=list[DeadLetter])
async def read_dead_letters(controller: FlowControllerDep, limit: int = 50) -> Any:
async def read_dead_letters(
controller: FlowControllerDep, limit: Annotated[int, Query(ge=1, le=200)] = 50
) -> Any:
"""Work the engine gave up on, which nothing else surfaces."""
if controller.execution is None:
return []
backend/fluksio/api/routes/panels.py
+38 -12
View File
@@ -35,7 +35,15 @@ from fluksio.cloud import config as cloud_config
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.flow.events import event_bus
from fluksio.flow.panels import PanelDef, PanelsConfig, find, read_config, write_config
from fluksio.flow.panels import (
PanelDef,
PanelsConfig,
edit_lock,
find,
read_config,
write_config,
)
from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
from fluksio.models import Message
#: Gated per route rather than on the router: the two pairing endpoints are the
@@ -104,6 +112,10 @@ class _PendingStore:
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
decode_responses=True,
# A silent Redis must fail this request, not hold it open.
socket_timeout=SOCKET_TIMEOUT_S,
socket_connect_timeout=CONNECT_TIMEOUT_S,
health_check_interval=30,
)
if settings.REDIS_HOST
else None
@@ -288,7 +300,6 @@ async def save_panels(body: PanelsConfig) -> Any:
how a device is unpaired along with its assignment. Re-pairing one screen
and keeping the assignment is ``/{panel_id}/unpair`` below.
"""
stored = {p.id: p.nonce for p in (await run_in_threadpool(read_config)).panels}
seen = set()
for panel in body.panels:
if panel.id in seen:
@@ -296,12 +307,20 @@ async def save_panels(body: PanelsConfig) -> Any:
status_code=422, detail=f"Two panels named {panel.id!r}"
)
seen.add(panel.id)
# The nonce belongs to this installation, not to whoever is writing the
# panels back: a client holding an older copy must not be able to undo
# a revocation by saving an arrangement.
panel.nonce = stored.get(panel.id, 0)
await run_in_threadpool(write_config, body)
def _save() -> None:
# Read and write under one lock. The nonce belongs to this
# installation, not to whoever is writing the panels back: a client
# holding an older copy must not be able to undo a revocation by
# saving an arrangement — and reading it in a separate step from
# writing it is exactly how an unpair in between was undone.
with edit_lock:
stored = {p.id: p.nonce for p in read_config().panels}
for panel in body.panels:
panel.nonce = stored.get(panel.id, 0)
write_config(body)
await run_in_threadpool(_save)
# Which dashboards hang on which panel just changed. An empty name says
# that much and no more: every screen listening rescopes and refetches
# what it shows, rather than waiting for whenever it next reads.
@@ -436,12 +455,19 @@ async def unpair_panel(panel_id: str) -> Any:
A credential the portal minted for a remote screen carries no nonce, so
this does not reach it; that one is revoked at the hub.
"""
config = await run_in_threadpool(read_config)
panel = next((p for p in config.panels if p.id == panel_id), None)
if panel is None:
def _unpair() -> bool:
with edit_lock:
config = read_config()
panel = next((p for p in config.panels if p.id == panel_id), None)
if panel is None:
return False
panel.nonce += 1
write_config(config)
return True
if not await run_in_threadpool(_unpair):
raise HTTPException(status_code=404, detail=f"No panel named {panel_id!r}")
panel.nonce += 1
await run_in_threadpool(write_config, config)
# The screen is still holding a socket. One event and it refetches, which
# is where it meets the 401 that sends it back to the pairing code.
event_bus.publish({"type": "dashboard_changed", "dashboard": "", "ts": time.time()})
backend/fluksio/api/routes/runs.py
+104 -52
View File
@@ -12,9 +12,9 @@ import time
from collections.abc import Iterator
from datetime import UTC, datetime
from itertools import groupby
from typing import Any, Literal
from typing import Annotated, Any, Literal
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field, model_validator
@@ -23,6 +23,7 @@ from sqlalchemy import select as sa_select
from sqlmodel import Session, col, select
from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
from fluksio.core.db import writing
from fluksio.flow.events import event_bus
from fluksio.flow.messages import requalify
from fluksio.flow.runs import RunRejected, RunService, new_run_id
@@ -271,8 +272,8 @@ def read_runs(
digest: str | None = None,
since: datetime | None = None,
before: datetime | None = None,
limit: int = 50,
offset: int = 0,
limit: Annotated[int, Query(ge=1, le=500)] = 50,
offset: Annotated[int, Query(ge=0)] = 0,
) -> Any:
"""Runs, newest first. The queryable table an experiment log needs.
@@ -306,6 +307,11 @@ def read_overview(session: SessionDep) -> Any:
The list caps at 500 newest runs, so counting flows on the client goes
wrong the moment a history outgrows one page. The database counts instead.
"""
# Deliberately unwindowed. It is a group-by over the whole `run` table,
# which is never pruned — but a run kept for years is the point of the
# table, and a window here would drop a finished experiment off the rail
# rather than making it cheaper to find. A rollup is the answer if the
# scan ever measures.
statement = sa_select(
col(Run.flow),
col(Run.status),
@@ -346,6 +352,12 @@ METRIC_COLUMNS = ("run", "name", "step", "ts", "value")
#: A run's own columns in the wide table. Its inputs and its final numbers
#: follow, prefixed, so an input named "status" cannot collide with the run's.
#: The most runs one export covers. High enough that a real sweep fits and low
#: enough that an unfiltered request cannot read the whole table into memory —
#: which it did, before a byte was streamed, while holding one of the fifteen
#: pooled database connections for the length of the download.
EXPORT_CAP = 10_000
RUN_COLUMNS = (
"id",
"flow",
@@ -370,7 +382,12 @@ def _selected(
since: datetime | None,
until: datetime | None,
) -> list[Run]:
"""The runs an export covers, newest first — the filters the list takes."""
"""The runs an export covers, newest first — the filters the list takes.
Capped: the filters bound a *sensible* request and nothing bounded an
unfiltered one, so asking for everything read every row of the table into
memory before a byte was sent. A truncated export says so in a header.
"""
statement = select(Run).order_by(col(Run.created_at).desc())
if flow:
statement = statement.where(col(Run.flow) == flow)
@@ -385,7 +402,7 @@ def _selected(
statement = statement.where(col(Run.created_at) >= _aware(since))
if until:
statement = statement.where(col(Run.created_at) < _aware(until))
return list(session.exec(statement))
return list(session.exec(statement.limit(EXPORT_CAP + 1)))
def _cell(value: Any) -> Any:
@@ -458,7 +475,11 @@ def _csv(columns: list[str], chunks: Iterator[list[dict[str, Any]]]) -> Iterator
def _stream(
fmt: str, name: str, columns: list[str], chunks: Iterator[list[dict[str, Any]]]
fmt: str,
name: str,
columns: list[str],
chunks: Iterator[list[dict[str, Any]]],
truncated: bool = False,
) -> StreamingResponse:
"""The rows out, a chunk at a time rather than a list.
@@ -473,11 +494,13 @@ def _stream(
else:
body = _csv(columns, chunks)
media = "text/csv; charset=utf-8"
return StreamingResponse(
body,
media_type=media,
headers={"Content-Disposition": f'attachment; filename="{name}.{fmt}"'},
)
headers = {"Content-Disposition": f'attachment; filename="{name}.{fmt}"'}
if truncated:
# The body is otherwise indistinguishable from a complete one, and an
# export quietly missing its oldest runs is the wrong thing to hand
# somebody who is about to plot it.
headers["X-Truncated"] = str(EXPORT_CAP)
return StreamingResponse(body, media_type=media, headers=headers)
@router.get("/export/metrics")
@@ -490,7 +513,7 @@ def export_metrics(
since: datetime | None = None,
until: datetime | None = None,
name: str = "",
stride: int = 1,
stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
format: ExportFormat = "csv",
) -> Any:
"""Every selected run's series as one long table: run, name, step, ts, value.
@@ -500,6 +523,8 @@ def export_metrics(
for every tenth point of two metrics gives every tenth point of both.
"""
runs = _selected(session, flow, status, group, ids, since, until)
truncated = len(runs) > EXPORT_CAP
runs = runs[:EXPORT_CAP]
wanted = [part for part in name.split(",") if part]
step = max(1, stride)
@@ -523,7 +548,7 @@ def export_metrics(
)
yield rows
return _stream(format, "metrics", list(METRIC_COLUMNS), chunks())
return _stream(format, "metrics", list(METRIC_COLUMNS), chunks(), truncated)
@router.get("/export/runs")
@@ -552,6 +577,8 @@ def export_runs(
depth.
"""
runs = _selected(session, flow, status, group, ids, since, until)
truncated = len(runs) > EXPORT_CAP
runs = runs[:EXPORT_CAP]
inputs = [part for part in params.split(",") if part] or sorted(
{path for run in runs for path, _ in _leaves(run.params)}
)
@@ -569,7 +596,7 @@ def export_runs(
row.update((f"metric.{k}", _cell(_dig(run.result, k))) for k in scores)
yield [row]
return _stream(format, "runs", columns, chunks())
return _stream(format, "runs", columns, chunks(), truncated)
@router.get("/{run_id}", response_model=RunDetail)
@@ -623,23 +650,27 @@ def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
keeps whatever a ``run_artifact`` row or a live message still names, so
dropping the rows is enough and the hourly sweep reclaims the blobs.
"""
run = session.get(Run, run_id)
if run is None:
raise HTTPException(status_code=404, detail="No such run")
if run.status in ("running", "queued"):
raise HTTPException(
status_code=409,
detail=(
f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
"finish, before deleting it."
),
)
flow = run.flow
session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
session.execute(delete(Run).where(col(Run.id) == run_id))
session.commit()
# Reads the run, then decides whether to delete it — so it takes the write
# lock up front rather than upgrading and losing to whichever flush
# committed in between.
with writing(session):
run = session.get(Run, run_id)
if run is None:
raise HTTPException(status_code=404, detail="No such run")
if run.status in ("running", "queued"):
raise HTTPException(
status_code=409,
detail=(
f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
"finish, before deleting it."
),
)
flow = run.flow
session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
session.execute(delete(Run).where(col(Run.id) == run_id))
session.commit()
event_bus.publish(
{
"type": "audit",
@@ -674,32 +705,50 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
if restored:
run = session.get(Run, run_id)
flow = run.flow if run is not None else ""
# Two queries for the lot, rather than a `Run` lookup and a `RunMetric`
# query per restored node. A comparison of twenty runs calls this
# twenty times, so the per-node round trips multiplied.
sources = {
row.id: row.flow
for row in session.exec(
select(Run).where(col(Run.id).in_({n.cached_from for n in restored}))
)
}
wanted: dict[tuple[str, str], list[RunNode]] = {}
for node_row in restored:
source = session.get(Run, node_row.cached_from)
if source is None:
source_flow = sources.get(node_row.cached_from)
if source_flow is None:
# The run it came from is gone — deleted with its flow. The
# outputs are still on this run; the curve is not recoverable.
continue
source_node = requalify(node_row.node, flow, source.flow)
for row in session.exec(
key = (
node_row.cached_from,
requalify(node_row.node, flow, source_flow),
)
wanted.setdefault(key, []).append(node_row)
if wanted:
cached = session.exec(
select(RunMetric).where(
col(RunMetric.run_id) == node_row.cached_from,
col(RunMetric.node) == source_node,
col(RunMetric.run_id).in_({run_id for run_id, _ in wanted}),
col(RunMetric.node).in_({node for _, node in wanted}),
)
):
renamed = requalify(row.name, source.flow, flow)
if name and renamed != name:
continue
rows.append(
RunMetric(
run_id=run_id,
name=renamed,
step=row.step,
node=node_row.node,
ts=row.ts,
value=row.value,
)
for row in cached:
for node_row in wanted.get((row.run_id, row.node), ()):
source_flow = sources[node_row.cached_from]
renamed = requalify(row.name, source_flow, flow)
if name and renamed != name:
continue
rows.append(
RunMetric(
run_id=run_id,
name=renamed,
step=row.step,
node=node_row.node,
ts=row.ts,
value=row.value,
)
)
)
rows.sort(key=lambda row: (row.name, row.step))
return rows
@@ -707,7 +756,10 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
@router.get("/{run_id}/metrics", response_model=list[MetricPoint])
def read_metrics(
run_id: str, session: SessionDep, name: str = "", stride: int = 1
run_id: str,
session: SessionDep,
name: str = "",
stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
) -> Any:
"""One metric's series, in step order — or every one of them, unnamed.
backend/fluksio/api/routes/secrets.py
+8 -2
View File
@@ -9,12 +9,18 @@ from fastapi import APIRouter, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from fluksio.api.deps import get_current_user
from fluksio.api.deps import get_current_active_superuser
from fluksio.flow.secrets import SecretNotFound, get_secrets
from fluksio.models import Message
# Superuser, not merely signed in. The names alone say what this installation
# talks to, and `PUT /{name}` takes any name — so an ordinary account could
# overwrite the credential a flow authenticates with. `/search` already gates
# secret names this way and said so; this router was the half that did not.
router = APIRouter(
prefix="/secrets", tags=["secrets"], dependencies=[Depends(get_current_user)]
prefix="/secrets",
tags=["secrets"],
dependencies=[Depends(get_current_active_superuser)],
)
backend/fluksio/api/routes/users.py
+7 -3
View File
@@ -1,7 +1,7 @@
import uuid
from typing import Any
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import col, func, select
from fluksio import crud
@@ -34,7 +34,11 @@ router = APIRouter(prefix="/users", tags=["users"])
dependencies=[Depends(get_current_active_superuser)],
response_model=UsersPublic,
)
def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
def read_users(
session: SessionDep,
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=500)] = 100,
) -> Any:
"""
Retrieve users.
"""
backend/fluksio/cloud/connector.py
+22
View File
@@ -50,6 +50,10 @@ MAX_WS_MESSAGE = 1024 * 1024
#: Only the versioned API is served over the tunnel. The MCP mount and the
#: OAuth endpoints live outside it and stay local-only.
ALLOWED_PREFIX = "/api/v1/"
#: Proxied calls, and bridged streams, this installation will run at once.
#: Neither dict was bounded, and an id the hub reused silently dropped the
#: reference to a task that was still running.
MAX_IN_FLIGHT = 256
#: How often to look for a config that appeared while the engine was up.
#: Enrolment is something a person does and then waits on, so this is the
#: delay they sit through — short enough not to be worth a restart.
@@ -380,6 +384,16 @@ class CloudConnector:
call_id = str(frame.get("id") or "")
if not call_id:
return
if len(self._calls) >= MAX_IN_FLIGHT:
logger.warning(
"Refusing proxied call %s: %d already in flight",
call_id,
len(self._calls),
)
return
# An id the hub reuses would otherwise drop the reference to a task
# still running, leaving it with nothing able to cancel it.
self._cancel(call_id)
task = asyncio.create_task(self._serve(socket, frame))
self._calls[call_id] = task
task.add_done_callback(lambda _t: self._calls.pop(call_id, None))
@@ -465,6 +479,14 @@ class CloudConnector:
stream_id = str(frame.get("id") or "")
if not stream_id:
return
if len(self._streams) >= MAX_IN_FLIGHT:
logger.warning(
"Refusing stream %s: %d already open", stream_id, len(self._streams)
)
return
existing = self._streams.pop(stream_id, None)
if existing is not None:
existing.cancel()
task = asyncio.create_task(self._stream(socket, frame))
self._streams[stream_id] = task
task.add_done_callback(lambda _t: self._streams.pop(stream_id, None))
backend/fluksio/cloud/enroll.py
+15
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from urllib.parse import urlsplit
import httpx
from sqlmodel import Session, select
@@ -34,11 +35,25 @@ class AlreadyEnrolled(EnrollError):
super().__init__(409, "This installation is already connected to a portal")
def _is_local(url: str) -> bool:
"""Whether the address is this machine or a compose-internal service."""
host = urlsplit(url).hostname or ""
return host in {"localhost", "127.0.0.1", "::1"} or host.endswith(".local")
def redeem_claim(
portal_url: str, claim_code: str, *, timeout: float = 15.0
) -> dict[str, Any]:
"""Trade a claim code for this installation's credential and the portal's keys."""
base = portal_url.rstrip("/")
# The claim code and, from here on, this installation's credential go to
# this address. Over plain http both are readable by anything on the path,
# so refuse rather than enrol insecurely — bar a loopback portal, which is
# how the stack is developed against itself.
if not base.startswith("https://") and not _is_local(base):
raise EnrollError(
422, "The portal address must be https:// (or a local address)"
)
try:
response = httpx.post(
f"{base}/api/v1/enroll/",
backend/fluksio/core/config.py
+13 -3
View File
@@ -57,8 +57,12 @@ class Settings(BaseSettings):
#: secrets, artifacts and the user venv. The paths below derive from it
#: unless they are set explicitly.
DATA_DIR: Path = Path("flow-data")
#: Any SQLAlchemy URL. The default puts SQLite in the data directory, which
#: is what makes `fluksio serve` need no infrastructure at all.
#: Where the database file goes, as a SQLite URL. The default puts it in
#: the data directory, which is what makes `fluksio serve` need no
#: infrastructure at all. **SQLite only**: the upserts the metrics
#: collector and the run recorder are built on are written against that
#: dialect, so another backend would parse, migrate, serve — and then
#: silently drop every rollup and fail every run.
DATABASE_URL: str | None = None
# Flows live on disk as a git repository; secrets stay outside it.
@@ -126,6 +130,12 @@ class Settings(BaseSettings):
# node at a time.
FLOW_GPUS: int = 0
# How long the engine's own metrics, events and run records are kept.
#: The largest body `PUT /artifacts` will take, in bytes. A checkpoint or a
#: video segment is a legitimate artifact, so this is generous rather than
#: small; 0 removes the limit. Nothing capped it at all before, so any
#: account or worker credential could fill the data volume.
MAX_ARTIFACT_BYTES: int = 2 * 1024 * 1024 * 1024
OBS_RETENTION_DAYS: int = 30
# How often artifact bytes nothing refers to any more are swept away; 0
# never sweeps. A flow streaming media writes one artifact per frame, so
@@ -190,7 +200,7 @@ class Settings(BaseSettings):
@computed_field # type: ignore[prop-decorator]
@property
def SQLALCHEMY_DATABASE_URI(self) -> str:
"""SQLite in the data directory, unless a URL says otherwise.
"""SQLite in the data directory, unless a URL names another file.
One engine process owns this database — the same reason the image runs
a single uvicorn worker — so a file beside the flows is the honest
backend/fluksio/core/db.py
+40 -8
View File
@@ -10,12 +10,14 @@ carry on while that single writer works.
from __future__ import annotations
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any
from alembic import command
from alembic.config import Config
from sqlalchemy import Engine, event, inspect, make_url
from sqlalchemy import Engine, event, inspect, make_url, text
from sqlmodel import Session, SQLModel, create_engine, select
from fluksio import crud
@@ -26,12 +28,21 @@ logger = logging.getLogger(__name__)
def make_engine(url: str) -> Engine:
"""One engine for the process, configured for whichever dialect it names."""
"""One engine for the process. SQLite, and only SQLite.
Not a limitation that snuck in: `metric_minute` and every run table are
written with `sqlalchemy.dialects.sqlite.insert(...).on_conflict_do_update`
and with `max(a, b)`, neither of which another dialect has. Pointing this
at Postgres used to migrate cleanly, serve, log in — and then lose every
observability flush into the collector's hold buffer and fail every run,
which is a far worse answer than refusing at startup.
"""
if make_url(url).get_backend_name() != "sqlite":
# A connection idle across a server restart is dead but still pooled;
# the pre-ping spends a round trip to find out instead of failing the
# request.
return create_engine(url, pool_pre_ping=True)
raise RuntimeError(
f"DATABASE_URL must be a SQLite URL, not {make_url(url).get_backend_name()!r}. "
"The engine's own database is a file beside the flows; the "
"analytics Postgres in the compose stack is Umami's, not this."
)
database = make_url(url).database
if database and database != ":memory:":
@@ -48,8 +59,6 @@ engine = make_engine(settings.SQLALCHEMY_DATABASE_URI)
@event.listens_for(engine, "connect")
def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None:
"""What makes concurrent readers, cascades and waiting-out a writer work."""
if engine.dialect.name != "sqlite":
return
cursor = dbapi_connection.cursor()
# Readers do not block on the writer, which is the whole reason this is
# usable while a run is streaming metrics into it.
@@ -65,6 +74,29 @@ def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None:
cursor.close()
@contextmanager
def writing(session: Session) -> Iterator[None]:
"""Take SQLite's write lock before reading, for a read-then-write handler.
SQLAlchemy begins deferred, so a transaction that reads and then writes
has to upgrade — and if another writer committed in between, SQLite
refuses with `SQLITE_BUSY_SNAPSHOT` immediately rather than waiting out
`busy_timeout`. The engine writes constantly (the metrics flush, a run
recording nodes), so any handler that decides *whether* to write by
reading first can lose that race and 500.
Only worth it where the handler really does both: this serialises against
other writers for the length of the transaction, which is what WAL's
concurrent readers otherwise avoid.
"""
session.execute(text("BEGIN IMMEDIATE"))
try:
yield
except Exception:
session.rollback()
raise
def _alembic_config(connection: Any) -> Config:
"""Alembic driven from here, so an installed wheel needs no alembic.ini."""
config = Config()
backend/fluksio/flow/modules.py
+20
View File
@@ -29,6 +29,7 @@ import shutil
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import TYPE_CHECKING
@@ -143,6 +144,12 @@ def ensure_venv() -> None:
)
#: Held for the length of a `uv pip sync`. Two applies at once — two browser
#: tabs, or an apply landing while the boot-time reconcile is still running —
#: mutate the same venv simultaneously.
_sync_lock = threading.Lock()
def sync(requirements: str) -> tuple[bool, str]:
"""Make the venv hold exactly these packages. Returns success and uv's output.
@@ -160,6 +167,19 @@ def sync(requirements: str) -> tuple[bool, str]:
"into. Install into it with pip or uv, or set NODE_VENV=managed "
"for a venv the engine owns."
)
if not _sync_lock.acquire(timeout=0.0):
return False, (
"Another apply is already installing packages. Wait for it to "
"finish, then try again."
)
try:
return _sync(requirements)
finally:
_sync_lock.release()
def _sync(requirements: str) -> tuple[bool, str]:
"""The install itself, under `_sync_lock`."""
manifest = ""
try:
ensure_venv()
backend/fluksio/flow/panels.py
+12
View File
@@ -12,6 +12,7 @@ which screen hangs where is the deployment's concern, not any one dashboard's.
from __future__ import annotations
import threading
from pathlib import Path
from pydantic import BaseModel, Field, field_validator
@@ -49,6 +50,17 @@ class PanelsConfig(BaseModel):
panels: list[PanelDef] = Field(default_factory=list)
#: Held across a read-modify-write of the panels file.
#:
#: Both `save_panels` and `unpair_panel` are one of those, and interleaving
#: them silently undid a revocation: a save that read the file before an
#: unpair wrote it put the old nonce back, and the screen that had just been
#: unpaired kept working. The nonce carry-forward in the save handler was
#: written to make that impossible, and the window between its read and its
#: write is where it happened anyway.
edit_lock = threading.Lock()
def _path() -> Path:
return settings.PANELS_FILE
backend/fluksio/flow/queue.py
+13 -1
View File
@@ -25,6 +25,8 @@ from typing import Any, cast
import orjson
import redis
from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
logger = logging.getLogger(__name__)
# The stream is a buffer, not an archive: anything this far behind is long
@@ -407,7 +409,17 @@ class RedisWorkQueue(WorkQueue):
namespace: str = "queue",
consumer: str | None = None,
) -> None:
self._redis = redis.Redis(host=host, port=port, decode_responses=True)
self._redis = redis.Redis(
host=host,
port=port,
decode_responses=True,
# Longer than any block this asks for, so a claim still waits its
# second — but bounded, so a Redis that goes silent without
# closing the socket fails rather than hanging the consumer.
socket_timeout=SOCKET_TIMEOUT_S,
socket_connect_timeout=CONNECT_TIMEOUT_S,
health_check_interval=30,
)
self._ns = namespace
self._consumer = consumer or f"engine-{int(time.time() * 1000) % 1_000_000}"
self._stream = f"{namespace}:__queue__"
backend/fluksio/flow/state.py
+13
View File
@@ -27,6 +27,12 @@ _JSON_OPTS = orjson.OPT_NON_STR_KEYS
# A sparkline only means something for a value that can be placed on an axis,
# so the history keeps those and nothing else. 120 points fill a panel-wide
# chart while leaving Redis a cache rather than a time-series database.
#: How long a Redis call may take before it is treated as a failure, and
#: how long connecting may take. A blocking read must ask for less than the
#: first of these.
SOCKET_TIMEOUT_S = 5.0
CONNECT_TIMEOUT_S = 2.0
HISTORY_LIMIT = 120
@@ -496,6 +502,13 @@ class RedisState(StateBackend):
db=db,
password=password,
decode_responses=False,
# Without these a Redis that stops answering without closing the
# connection — a partition, a firewall dropping state — hangs the
# caller until the kernel gives up, minutes later. That takes the
# health endpoint whose job is to notice with it.
socket_timeout=SOCKET_TIMEOUT_S,
socket_connect_timeout=CONNECT_TIMEOUT_S,
health_check_interval=30,
)
self._namespace = namespace
self._ttl = ttl
backend/fluksio/main.py
+186 -144
View File
@@ -1,7 +1,8 @@
import asyncio
import contextlib
import inspect
import logging
from collections.abc import AsyncIterator, Coroutine
from collections.abc import AsyncIterator, Callable, Coroutine
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from typing import Any
@@ -129,109 +130,32 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# pays a version check for it and nothing else.
await run_in_threadpool(prepare, db_engine)
event_bus.bind(asyncio.get_running_loop())
# Node code is user code, and `print` is how it says things.
logs.install()
init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
# Connectors register their node types before any flow is built with them.
load_plugins()
alerts = AlertManager(event_bus, config=read_alerts_config())
execution = ExecutionService(
queue=_work_queue(),
max_workers=settings.FLOW_MAX_WORKERS,
events=event_bus,
max_cascades=settings.FLOW_MAX_CASCADES,
)
store = FlowStore(settings.FLOWS_DIR)
# The packages node code imports, before anything tries to import them.
await run_in_threadpool(modules.reconcile, store)
# Beside the flows rather than in them: an artifact is what a run produced,
# not something anyone wrote, so it has no business in the git repository.
artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
app.state.artifact_store = artifacts
accountant = ResourceAccountant(cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS)
app.state.resources = accountant
# Every machine a node could run on: this one, and whatever attaches.
placer = Placer(local=accountant, events=event_bus)
# Where more machines can be asked for when nothing attached will do.
placer.provisioners = load_provisioners(settings.PROVISIONERS_FILE, event_bus)
app.state.placer = placer
pool = PythonWorkerPool(
python=modules.venv_python(),
size=settings.FLOW_MAX_WORKERS,
events=event_bus,
# A worker in this container writes to the store directly; a remote one
# is given a URL instead. Node code calls the same two functions.
env={
ARTIFACT_DIR_ENV: str(artifacts.root),
# Every slot can be busy at once, so a worker left to size its own
# thread pool to the machine means as many processes as there are
# slots, each believing it has the whole of it. A node that says
# what it needs overrides this; one that says nothing gets a share.
**fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS),
},
)
pool.start()
app.state.worker_pool = pool
# Assigned rather than passed both ways: the hub tells the placer when a
# machine comes or goes, and the placer needs the hub to know what is there.
worker_hub = RemoteWorkerHub(on_change=placer.wake)
placer.hub = worker_hub
app.state.worker_hub = worker_hub
controller = FlowController(
store=store,
state=_state_backend(),
events=event_bus,
max_workers=settings.FLOW_MAX_WORKERS,
fastapi_app=app,
execution=execution,
alerts=alerts,
workers=pool,
remote=worker_hub,
resources=accountant,
placer=placer,
)
app.state.flow_controller = controller
# A "dashboard" alert channel puts its alert into the graph. Bound here
# rather than passed in: the manager is built before the controller is.
alerts.publish = lambda name, value: controller.publish_message(
name, value, ValueSource(kind="api", id="alerts", label="Alerts")
)
dashboards = DashboardStore(controller.store)
app.state.dashboard_store = dashboards
controller.dashboards = dashboards
# Charts need a deeper series than the default; tell the engine
# before it starts recording.
controller.set_history_limits(dashboards.history_requirements())
# Runs read from a stream of their own: a burst of sweep runs must not
# stand between the automations and their work, and a run that takes an
# hour must not be judged by the cascade reaper's timings.
run_service = RunService(
controller=controller,
queue=_work_queue("run"),
state_factory=_run_state,
artifacts=artifacts,
parallel=settings.FLOW_MAX_RUNS,
)
app.state.run_service = run_service
# Said out loud because it is the only way to tell that a settings file was
# read at all — the numbers are what someone raising them is looking for.
logger.info(
"Engine limits: workers=%s cascades=%s runs=%s",
settings.FLOW_MAX_WORKERS,
execution.max_cascades,
run_service.parallel,
)
watchdog = LoopWatchdog(event_bus)
app.state.watchdog = watchdog
# Everything acquired below registers how to release it, and the `finally`
# at the end walks the list backwards. It used to be a fixed block of
# shutdown steps *after* every acquisition — so a failure part-way through
# startup reached none of them and left the worker pool's subprocesses and
# every background task behind. Under `--reload` that is once per bad edit.
started: list[Callable[[], Any]] = []
def _background(coro: Coroutine[Any, Any, None], name: str) -> asyncio.Task[None]:
"""Start a long-lived task that says something if it ever stops.
async def _release() -> None:
for close in reversed(started):
try:
result = close()
if inspect.isawaitable(result):
await result
except Exception:
logger.exception("Error while shutting down")
Each of these loops catches its own exceptions *inside* the loop, so
one raised anywhere else simply ended the task — an engine that went
on serving with no metrics, no alerts or no artifact sweep and nothing
anywhere saying so.
def _background(
coro: Coroutine[Any, Any, None], name: str, once: bool = False
) -> asyncio.Task[None]:
"""Start a background task that says something if it ever stops.
The loops catch their own exceptions *inside* the loop, so one raised
anywhere else simply ended the task — an engine that went on serving
with no metrics, no alerts or no artifact sweep and nothing anywhere
saying so. ``once`` is for a task that is meant to finish.
"""
task = asyncio.create_task(coro, name=name)
@@ -240,7 +164,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
return
exc = done.exception()
if exc is None:
logger.warning("Background task '%s' stopped on its own", name)
if not once:
logger.warning("Background task '%s' stopped on its own", name)
return
logger.error("Background task '%s' died: %s", name, exc, exc_info=exc)
event_bus.publish(
@@ -248,56 +173,173 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
)
task.add_done_callback(_finished)
started.append(task.cancel)
return task
watchdog_task = _background(watchdog.run(), "loop-watchdog")
alerts_task = _background(alerts.run(), "alert-manager")
metrics_task = _background(MetricsCollector(event_bus).run(), "metrics-collector")
gc_task = _background(_sweep_artifacts(artifacts, controller), "artifact-gc")
await controller.start()
run_service.start()
# Optional, and off unless someone enrolled this installation: the
# connector dials the portal, nothing dials in.
cloud_task: asyncio.Task[None] | None = None
app.state.cloud_connector = None
app.state.cloud_task = None
from fluksio.cloud import connector as cloud_connector
if cloud_config.exists():
cloud_connector.start(app)
cloud_task = app.state.cloud_task
# Watched whether or not one exists now: enrolling from the CLI writes the
# config from another process entirely, and an engine already serving
# should pick it up rather than need restarting.
enrol_task = _background(
cloud_connector.watch_enrolment(app), "cloud-enrolment-watch"
)
# The startup itself is inside the try, so a step that raises part-way
# through still reaches the release below.
try:
# Node code is user code, and `print` is how it says things.
logs.install()
init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
# Connectors register their node types before any flow is built with them.
load_plugins()
alerts = AlertManager(event_bus, config=read_alerts_config())
execution = ExecutionService(
queue=_work_queue(),
max_workers=settings.FLOW_MAX_WORKERS,
events=event_bus,
max_cascades=settings.FLOW_MAX_CASCADES,
)
store = FlowStore(settings.FLOWS_DIR)
# The packages node code imports. In the background, because a first
# install of anything substantial takes minutes and `uv` is given five
# of them twice over — held here, the container never answers its
# healthcheck, autoheal restarts it, and it never finishes installing at
# all. A node whose import is not there yet fails and is retried, which
# is recoverable; a restart loop is not.
_background(
run_in_threadpool(modules.reconcile, store), "module-reconcile", once=True
)
# Beside the flows rather than in them: an artifact is what a run produced,
# not something anyone wrote, so it has no business in the git repository.
artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
app.state.artifact_store = artifacts
accountant = ResourceAccountant(
cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS
)
app.state.resources = accountant
# Every machine a node could run on: this one, and whatever attaches.
placer = Placer(local=accountant, events=event_bus)
# Where more machines can be asked for when nothing attached will do.
placer.provisioners = load_provisioners(settings.PROVISIONERS_FILE, event_bus)
app.state.placer = placer
pool = PythonWorkerPool(
python=modules.venv_python(),
size=settings.FLOW_MAX_WORKERS,
events=event_bus,
# A worker in this container writes to the store directly; a remote one
# is given a URL instead. Node code calls the same two functions.
env={
ARTIFACT_DIR_ENV: str(artifacts.root),
# Every slot can be busy at once, so a worker left to size its own
# thread pool to the machine means as many processes as there are
# slots, each believing it has the whole of it. A node that says
# what it needs overrides this; one that says nothing gets a share.
**fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS),
},
)
pool.start()
# On a thread: stopping a worker waits up to five seconds on each child,
# and with a pool per declared environment that is a shutdown the event
# loop should not be holding.
started.append(lambda: run_in_threadpool(pool.stop))
app.state.worker_pool = pool
# Assigned rather than passed both ways: the hub tells the placer when a
# machine comes or goes, and the placer needs the hub to know what is there.
worker_hub = RemoteWorkerHub(on_change=placer.wake)
placer.hub = worker_hub
app.state.worker_hub = worker_hub
controller = FlowController(
store=store,
state=_state_backend(),
events=event_bus,
max_workers=settings.FLOW_MAX_WORKERS,
fastapi_app=app,
execution=execution,
alerts=alerts,
workers=pool,
remote=worker_hub,
resources=accountant,
placer=placer,
)
app.state.flow_controller = controller
# A "dashboard" alert channel puts its alert into the graph. Bound here
# rather than passed in: the manager is built before the controller is.
alerts.publish = lambda name, value: controller.publish_message(
name, value, ValueSource(kind="api", id="alerts", label="Alerts")
)
dashboards = DashboardStore(controller.store)
app.state.dashboard_store = dashboards
controller.dashboards = dashboards
# Charts need a deeper series than the default; tell the engine
# before it starts recording.
controller.set_history_limits(dashboards.history_requirements())
# Runs read from a stream of their own: a burst of sweep runs must not
# stand between the automations and their work, and a run that takes an
# hour must not be judged by the cascade reaper's timings.
run_service = RunService(
controller=controller,
queue=_work_queue("run"),
state_factory=_run_state,
artifacts=artifacts,
parallel=settings.FLOW_MAX_RUNS,
)
app.state.run_service = run_service
# Said out loud because it is the only way to tell that a settings file was
# read at all — the numbers are what someone raising them is looking for.
logger.info(
"Engine limits: workers=%s cascades=%s runs=%s",
settings.FLOW_MAX_WORKERS,
execution.max_cascades,
run_service.parallel,
)
watchdog = LoopWatchdog(event_bus)
app.state.watchdog = watchdog
_background(watchdog.run(), "loop-watchdog")
_background(alerts.run(), "alert-manager")
_background(MetricsCollector(event_bus).run(), "metrics-collector")
_background(_sweep_artifacts(artifacts, controller), "artifact-gc")
await controller.start()
started.append(controller.stop)
run_service.start()
started.append(lambda: run_in_threadpool(run_service.stop))
# A machine asked for and not yet arrived would hold an allocation nobody
# is going to use.
started.append(
lambda: asyncio.gather(
*(
run_in_threadpool(provisioner.shutdown)
for provisioner in placer.provisioners
)
)
)
started.append(lambda: run_in_threadpool(close_shared_client))
# Optional, and off unless someone enrolled this installation: the
# connector dials the portal, nothing dials in.
cloud_task: asyncio.Task[None] | None = None
app.state.cloud_connector = None
app.state.cloud_task = None
from fluksio.cloud import connector as cloud_connector
if cloud_config.exists():
cloud_connector.start(app)
cloud_task = app.state.cloud_task
def _stop_cloud() -> None:
# Re-read from app.state: enrolling at runtime replaces this.
running = getattr(app.state, "cloud_task", None) or cloud_task
if running is not None:
running.cancel()
started.append(_stop_cloud)
# Watched whether or not one exists now: enrolling from the CLI writes the
# config from another process entirely, and an engine already serving
# should pick it up rather than need restarting.
_background(cloud_connector.watch_enrolment(app), "cloud-enrolment-watch")
# A mounted sub-app gets no lifespan of its own, so the MCP session
# manager is entered here; without it every /mcp request fails.
async with _mcp_sessions():
yield
except Exception:
# A startup that fails part-way lands here too, because everything
# since `event_bus.bind` registered its own release.
logger.exception("The engine could not start")
raise
finally:
watchdog_task.cancel()
alerts_task.cancel()
metrics_task.cancel()
gc_task.cancel()
enrol_task.cancel()
# Re-read from app.state: enrolling at runtime replaces this.
running_cloud = getattr(app.state, "cloud_task", None) or cloud_task
if running_cloud is not None:
running_cloud.cancel()
await run_in_threadpool(run_service.stop)
await controller.stop()
# On a thread, like the two above: stopping a worker waits up to five
# seconds on each child, and with a pool per declared environment that
# is a shutdown the event loop should not be holding.
await run_in_threadpool(pool.stop)
# A machine asked for and not yet arrived would hold an allocation
# nobody is going to use.
for provisioner in placer.provisioners:
await run_in_threadpool(provisioner.shutdown)
await run_in_threadpool(close_shared_client)
await _release()
if settings.MCP_ENABLED:
from fluksio.mcp.http import aclose
backend/fluksio/models.py
+11 -1
View File
@@ -3,7 +3,7 @@ from datetime import UTC, datetime
from typing import Any
from pydantic import EmailStr
from sqlalchemy import JSON, Column
from sqlalchemy import JSON, Column, Index
from sqlmodel import Field, SQLModel
from fluksio.core.types import UTCDateTime
@@ -259,6 +259,9 @@ class EngineEvent(SQLModel, table=True):
"""Something worth keeping after the websocket has forgotten it."""
__tablename__ = "engine_event"
# The events screen excludes audit rows and reads newest first. A `type`
# index alone serves neither half of that.
__table_args__ = (Index("ix_engine_event_type_ts", "type", "ts"),)
id: int | None = Field(default=None, primary_key=True)
ts: datetime = Field(
@@ -286,6 +289,8 @@ class FlowRun(SQLModel, table=True):
"""
__tablename__ = "flow_run"
# The history screen filters by flow, newest first.
__table_args__ = (Index("ix_flow_run_flow_started", "flow", "started_at"),)
#: The queue entry id, a `manual-` one for a run that never queued, or the
#: `Run.id` of a batch run.
@@ -327,6 +332,11 @@ class Run(SQLModel, table=True):
"""One finite execution of a flow, with what it was asked and what it made."""
__tablename__ = "run"
# The runs list filters by flow and/or status, newest first.
__table_args__ = (
Index("ix_run_flow_created", "flow", "created_at"),
Index("ix_run_status_created", "status", "created_at"),
)
id: str = Field(primary_key=True, max_length=64)
flow: str = Field(index=True, max_length=255)
backend/fluksio/utils.py
+32 -5
View File
@@ -1,3 +1,4 @@
import hashlib
import logging
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
@@ -14,6 +15,9 @@ from fluksio.core.config import settings
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
#: How long to wait on the mail server before giving up on one message.
SMTP_TIMEOUT_S = 10
@dataclass
class EmailData:
@@ -47,7 +51,14 @@ def send_email(
html=html_content,
mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
)
smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
# With no timeout smtplib inherits the socket default, which is none at
# all — an unreachable mail host then held a threadpool worker until the
# kernel gave up, from a route anybody can call.
smtp_options: dict[str, Any] = {
"host": settings.SMTP_HOST,
"port": settings.SMTP_PORT,
"timeout": SMTP_TIMEOUT_S,
}
if settings.SMTP_TLS:
smtp_options["tls"] = True
elif settings.SMTP_SSL:
@@ -105,24 +116,40 @@ def generate_new_account_email(
return EmailData(html_content=html_content, subject=subject)
def generate_password_reset_token(email: str) -> str:
def password_fingerprint(hashed_password: str) -> str:
"""A short digest of the stored hash, for binding a reset link to it.
What makes a reset token single-use without a table of spent ones: the
hash changes the moment the password does, so the link that set it stops
verifying. It was replayable for the whole 48 hours otherwise.
"""
return hashlib.sha256(hashed_password.encode()).hexdigest()[:16]
def generate_password_reset_token(email: str, hashed_password: str = "") -> str:
delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
now = datetime.now(UTC)
expires = now + delta
exp = expires.timestamp()
encoded_jwt = jwt.encode(
{"exp": exp, "nbf": now, "sub": email},
{
"exp": exp,
"nbf": now,
"sub": email,
"pwd": password_fingerprint(hashed_password),
},
settings.SECRET_KEY,
algorithm=security.ALGORITHM,
)
return encoded_jwt
def verify_password_reset_token(token: str) -> str | None:
def verify_password_reset_token(token: str) -> tuple[str, str] | None:
"""The email the token names and the password hash it was minted against."""
try:
decoded_token = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
)
return str(decoded_token["sub"])
return str(decoded_token["sub"]), str(decoded_token.get("pwd", ""))
except InvalidTokenError:
return None