Merge pull request #264 from sheaf-project/fix/front-coalesce-scaling

Fix front-coalesce query blow-up and statement_timeout coverage gaps

Author: SiteRelEnby

CHANGELOG.md

@@ -6,6 +6,11 @@ All notable changes to Sheaf are documented here. The format is based on [Keep a
 
 ## [Unreleased]
 
+### Fixed
+
+- **"Fronting since" no longer melts the database on a large imported history.** With coalesce-contiguous-fronts enabled, the walk that computes how far back a member's unbroken fronting run extends used a recursive query that, on an imported history whose switch boundaries share timestamps (PluralKit exports round to the second, so thousands do), could enumerate hundreds of millions of intermediate rows and spill tens of gigabytes into database temporary storage - one system's genuine imported history briefly took the whole database down this way. The walk is now a single sorted pass over the member's own front entries, with cost proportional to their history size, and returns the same answers. One deliberate refinement: fronts that *overlap* for the same member now count as one continuous run (previously the run only chained when one entry ended at the exact instant the next began), so a member fronting continuously across overlapping entries gets the earlier, correct "since". As a consequence the walk-back depth cap is gone; `member_since_capped` remains in the API for compatibility but is now always empty.
+- **The realtime front-change stream's database queries are now time-capped like every other request.** The stream endpoint builds its snapshots in self-managed database sessions, which silently missed the per-request statement timeout - so a pathological query from the stream path could run unboundedly (this is how the incident above kept filling the disk for 13 minutes after the same query on the normal endpoint had been cancelled at 30 seconds). Those sessions now carry the same timeout, and the cap is re-applied on every transaction, closing a second quiet gap where a mid-request commit dropped the timeout for the remainder of that request.
+
 ## [1.3.5] - 2026-08-09
 
 ### Security

docs/SELFHOSTING.md

@@ -108,6 +108,18 @@ Sheaf refuses to start in `saas` mode if this is left at the default, and logs a
 
 **If not set**, a key is auto-generated on first startup and saved to `data/encryption.key` inside the Docker volume. **Back this file up.** Prefer setting `SHEAF_ENCRYPTION_KEY` explicitly so the key isn't tied to a single volume.
 
+### Postgres safety limits (recommended)
+
+By default Postgres lets a single query spill unlimited temporary files to disk while it works. One runaway query can therefore fill the volume Postgres lives on and take the whole database down for every user - not just fail itself. Capping it turns that failure mode into "the one query errors out":
+
+```sql
+ALTER SYSTEM SET temp_file_limit = '2GB';   -- generous for a small instance
+ALTER SYSTEM SET log_temp_files = '64MB';   -- log any query spilling more than this
+SELECT pg_reload_conf();
+```
+
+Run once via `psql` against your database (for the bundled container: `docker compose exec db psql -U sheaf`). Pick a `temp_file_limit` well below the free space on the database volume. `log_temp_files` gives you a log line naming any unusually hungry query, which is the breadcrumb you want if something ever does hit the cap.
+
 ### Compose managers that don't use a `.env` file
 
 Some stack managers keep the environment in a differently-named file: OpenMediaVault's compose plugin writes `sheaf.env`, Portainer keeps stack env in its own database, and so on. Sheaf's `docker-compose.yml` loads the app's config from a file literally named `.env`, so if your manager uses another name the app receives none of its env-file settings and falls back to defaults. The classic symptom on a brand-new stack is `password authentication failed for user "sheaf"`: Postgres was created with your `POSTGRES_PASSWORD`, but the app never saw it.

sheaf/api/v1/front_stream.py

@@ -25,7 +25,7 @@
 from sheaf.auth.dependencies import get_current_user
 from sheaf.auth.sessions import get_redis, get_session_user_id
 from sheaf.config import settings
-from sheaf.database import async_session_factory
+from sheaf.database import request_session
 from sheaf.models.user import User
 from sheaf.observability.metrics import (
     realtime_connection_duration_seconds,
@@ -122,7 +122,7 @@ async def _recheck_auth(ctx: dict) -> tuple[bool, str | None]:
 
         from sheaf.models.api_key import ApiKey
 
-        async with async_session_factory() as db:
+        async with request_session() as db:
             row = await db.get(ApiKey, ctx["api_key_id"])
         if row is None:
             return False, "auth_revoked"
@@ -215,7 +215,7 @@ async def _stream(
         # Snapshot in a short-lived session CLOSED before we start yielding, so
         # no DB connection is held while the stream is open. Building the whole
         # snapshot first also means a slow client cannot pin the connection.
-        async with async_session_factory() as db:
+        async with request_session() as db:
             snapshots = [
                 build_snapshot_payload(
                     sid,
@@ -356,7 +356,7 @@ async def stream_fronts(
     # after the response finishes, and a stream finishes only when it closes, so
     # the pooled Postgres connection would sit idle-in-transaction for the whole
     # connection and eventually exhaust the pool, blocking every other request.
-    async with async_session_factory() as db:
+    async with request_session() as db:
         system_ids = await authorized_front_system_ids(user, db)
     account_key = str(user.id)
     auth_ctx = _auth_context(request)

sheaf/api/v1/fronts.py

@@ -87,10 +87,11 @@ def _front_to_read(
     member — the literal-entry view used by history endpoints and any
     caller that doesn't want to pay for the walk-back.
 
-    `member_since_capped` lists member ids whose chain hit the
-    walk-back depth limit; the returned timestamp is a lower bound,
-    not the true chain start. Frontends should render those with a
-    "> X ago" prefix to be honest about precision.
+    `member_since_capped` is retained for API compatibility: the old
+    recursive walk-back could hit a depth limit and flag members whose
+    timestamp was only a lower bound. The set-based coalesce query has
+    no such cap, so this is now always empty; frontends that render a
+    "> X ago" prefix for flagged members simply never see one.
 
     `has_audit_history` reflects whether at least one FrontAuditEvent
     exists for this entry. Computed once per list call via a batch
@@ -141,49 +142,78 @@ async def _front_has_audit(db: AsyncSession, front_id: uuid.UUID) -> bool:
     return result.scalar_one_or_none() is not None
 
 
-# Walk-back depth cap: pathological cycles aside, real chains are
-# typically 1-3 entries. 500 is a generous bound that prevents a
-# corrupted-data edge case from running unbounded queries while still
-# covering anyone who switches every few minutes for many hours.
-# When the cap *is* hit, the response flags the affected member so the
-# UI can render "> X ago" instead of silently under-reporting. Easy to
-# raise later if the flag actually starts surfacing in real usage.
-_COALESCE_MAX_DEPTH = 500
-
-
-# Recursive CTE that walks every (seed_front, member) chain in
-# parallel. Replaces what was previously a per-(front, member) loop of
-# awaited single-row queries — fine for one open front with two
-# members on a brand-new system, ruinous for /current on a busy
-# system with coalesce_contiguous_fronts on. The CTE is bounded by
-# `_COALESCE_MAX_DEPTH` so a corrupted-data cycle can't run forever,
-# and we surface the cap-hit per member the same way the old code did.
+# Set-based gaps-and-islands "coalesce contiguous fronting" query.
+#
+# This replaced a recursive CTE that walked prev.ended_at == started_at
+# chains with UNION ALL. That formulation enumerates PATHS, not states:
+# on an imported history whose switch boundaries share timestamps
+# (PluralKit exports are second-rounded, so thousands collide), every
+# recursive step matches multiple predecessors and the intermediate set
+# grows multiplicatively per level. A depth cap bounds depth, not width.
+# On 2026-08-13 a genuine ~14k-front imported history detonated it into
+# ~19 GB of Postgres query temp, filling the data volume (see the
+# front-coalesce-query-scaling design doc). Path count is combinatorial;
+# table size is irrelevant.
+#
+# The rewrite computes each member's contiguous runs ("islands") in one
+# sorted window pass: a run breaks only where every earlier front's
+# reach (running MAX of ended_at, with an open front reaching infinity)
+# ends strictly before the next started_at. Cost is one sort of the
+# member's own fronts - O(n log n), no recursion, no width explosion.
+#
+# Contiguity is interval-merge: exact touches (ended_at == started_at)
+# chain exactly as before, and OVERLAPPING fronts for the same member -
+# the very shape that made the recursion branch - merge into one run.
+# Chain-reachability is a strict subset of interval-merge, so results
+# are identical wherever fronts don't overlap (all live-written data);
+# where they do, the merged run extends "since" through the overlap,
+# which is the correct reading of an unbroken fronting stretch.
 _COALESCED_SINCE_SQL = text(
     """
-    WITH RECURSIVE chain(seed_front_id, member_id, started_at, depth) AS (
-        SELECT f.id, fm.member_id, f.started_at, 0
+    WITH member_fronts AS (
+        SELECT fm.member_id,
+               f.id AS front_id,
+               f.started_at,
+               COALESCE(f.ended_at, 'infinity'::timestamptz) AS reach
         FROM fronts f
         JOIN front_members fm ON fm.front_id = f.id
-        WHERE f.id IN :seed_ids
-
-        UNION ALL
-
-        SELECT chain.seed_front_id, chain.member_id, prev.started_at,
-               chain.depth + 1
-        FROM chain
-        JOIN fronts prev
-            ON prev.system_id = :system_id
-            AND prev.ended_at = chain.started_at
-        JOIN front_members fm
-            ON fm.front_id = prev.id
-            AND fm.member_id = chain.member_id
-        WHERE chain.depth < :max_depth
+        WHERE f.system_id = :system_id
+          AND fm.member_id IN (
+              SELECT seed_fm.member_id
+              FROM front_members seed_fm
+              WHERE seed_fm.front_id IN :seed_ids
+          )
+    ),
+    runs AS (
+        SELECT member_id, front_id, started_at, reach,
+               MAX(reach) OVER (
+                   PARTITION BY member_id
+                   ORDER BY started_at, reach, front_id
+                   ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
+               ) AS prev_reach
+        FROM member_fronts
+    ),
+    islands AS (
+        SELECT member_id, front_id, started_at,
+               SUM(CASE WHEN prev_reach IS NULL OR prev_reach < started_at
+                        THEN 1 ELSE 0 END)
+                   OVER (PARTITION BY member_id
+                         ORDER BY started_at, reach, front_id) AS island
+        FROM runs
+    ),
+    island_starts AS (
+        SELECT member_id, island, MIN(started_at) AS run_start
+        FROM islands
+        GROUP BY member_id, island
     )
-    SELECT seed_front_id, member_id,
-           MIN(started_at) AS earliest_started,
-           MAX(depth) AS deepest
-    FROM chain
-    GROUP BY seed_front_id, member_id
+    SELECT i.front_id AS seed_front_id,
+           i.member_id,
+           s.run_start AS earliest_started
+    FROM islands i
+    JOIN island_starts s
+        ON s.member_id = i.member_id
+        AND s.island = i.island
+    WHERE i.front_id IN :seed_ids
     """
 ).bindparams(bindparam("seed_ids", expanding=True))
 
@@ -194,8 +224,14 @@ async def _build_coalesced_member_since(
     """For each given front, build (since_map, capped_member_ids).
 
     When `system.coalesce_contiguous_fronts` is False, returns the
-    literal-entry view for each member with no capped members.
-    Otherwise walks back per member to find the earliest chain start.
+    literal-entry view for each member with no capped members. Otherwise
+    resolves each member's contiguous-run start via the gaps-and-islands
+    query above.
+
+    `capped_member_ids` is always empty now: the set-based query has no
+    depth cap to hit (the old recursive walk-back did). The tuple shape
+    is kept so the schema field `member_since_capped` stays present for
+    API compatibility; it simply never flags anyone.
     """
     out: dict[uuid.UUID, tuple[dict[str, datetime], list[str]]] = {}
     if not fronts:
@@ -220,18 +256,12 @@ async def _build_coalesced_member_since(
         {
             "seed_ids": [f.id for f in fronts],
             "system_id": system.id,
-            "max_depth": _COALESCE_MAX_DEPTH,
         },
     )
 
-    for seed_id, member_id, earliest, deepest in rows:
-        per_member, capped = out[seed_id]
+    for seed_id, member_id, earliest in rows:
+        per_member, _capped = out[seed_id]
         per_member[str(member_id)] = earliest
-        # depth==max_depth means the recursive step ran the last
-        # allowed iteration and may not have found the true chain
-        # start. Same semantics as the prior per-walk cap flag.
-        if deepest is not None and deepest >= _COALESCE_MAX_DEPTH:
-            capped.append(str(member_id))
 
     return out
 

sheaf/database.py

@@ -2,8 +2,9 @@
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
-from sqlalchemy import event, text
+from sqlalchemy import event
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.orm import Session as _SyncSession
 
 from sheaf.config import settings
 
@@ -16,9 +17,10 @@
 # A connection-level cap would apply to every session drawn from the pool,
 # including the long-running background jobs (export builds, retention
 # sweeps, analytics) that legitimately outlast any request. Instead the
-# SHORT request cap is applied per-transaction inside get_db, and jobs stay
-# uncapped by default (opting into db_job_statement_timeout_ms via
-# job_session()). See _set_local_statement_timeout below.
+# SHORT request cap is applied per-transaction to sessions that opt in via
+# session.info (get_db, request_session), and jobs stay uncapped by default
+# (opting into db_job_statement_timeout_ms via job_session()). See the
+# after_begin listener below.
 engine = create_async_engine(
     settings.database_url,
     echo=False,
@@ -72,28 +74,44 @@ def _after_cursor_execute(conn, cursor, statement, parameters, context, executem
     )
 
 
-async def _set_local_statement_timeout(session: AsyncSession, timeout_ms: int) -> None:
-    """Apply a transaction-local Postgres statement_timeout to `session`.
-
-    A value <= 0 is a no-op (unlimited). SET LOCAL is scoped to the current
-    transaction, so the cap neither leaks to the next caller that checks out
-    this pooled connection nor bleeds past a commit. Postgres SET does not
-    accept bind parameters, so the value is formatted in directly - it comes
-    from a pydantic int setting, never user input, and is re-cast to int
-    here, so there is no injection surface.
+# Sessions opt into a Postgres statement_timeout by setting this key in
+# session.info; the after_begin listener below re-applies it at the start of
+# EVERY transaction the session opens. The previous design issued a single
+# SET LOCAL when the session was created, which silently evaporated at the
+# first mid-request commit (SET LOCAL is transaction-scoped), leaving the
+# rest of the request uncapped. The 2026-08-13 incident's variant of this
+# class - a request-tier session created outside get_db with no cap at all -
+# let one coalesce query spill 19 GB of query temp; see request_session().
+_TIMEOUT_INFO_KEY = "statement_timeout_ms"
+
+
+@event.listens_for(_SyncSession, "after_begin")
+def _apply_statement_timeout(session, transaction, connection) -> None:
+    """Apply the session's opted-in statement_timeout to each new transaction.
+
+    Registered on the Session class, so it fires for every session drawn from
+    the shared pool - but it is a no-op unless the session set
+    `info[_TIMEOUT_INFO_KEY]`, so background jobs that use
+    async_session_factory() directly stay uncapped as designed. SET LOCAL is
+    transaction-scoped, so nothing leaks to the next checkout of the pooled
+    connection. Postgres SET takes no bind parameters; the value is a pydantic
+    int setting (never user input) re-cast to int here, so there is no
+    injection surface.
     """
-    if timeout_ms <= 0:
+    timeout_ms = session.info.get(_TIMEOUT_INFO_KEY)
+    if not timeout_ms or timeout_ms <= 0:
         return
-    await session.execute(text(f"SET LOCAL statement_timeout = {int(timeout_ms)}"))
+    connection.exec_driver_sql(f"SET LOCAL statement_timeout = {int(timeout_ms)}")
 
 
 async def get_db() -> AsyncGenerator[AsyncSession]:
     async with async_session_factory() as session:
         # Bound the request path so a pathological O(history) query can't pin
-        # a pooled connection indefinitely. Applied per-transaction, so it
-        # never touches the background jobs, which use async_session_factory()
-        # / job_session() directly and legitimately run longer than a request.
-        await _set_local_statement_timeout(session, settings.db_statement_timeout_ms)
+        # a pooled connection indefinitely. The info key makes the after_begin
+        # listener re-apply the cap on every transaction, so it survives
+        # mid-request commits. Background jobs use async_session_factory() /
+        # job_session() directly and are not affected.
+        session.info[_TIMEOUT_INFO_KEY] = settings.db_statement_timeout_ms
         try:
             yield session
             await session.commit()
@@ -102,6 +120,29 @@ async def get_db() -> AsyncGenerator[AsyncSession]:
             raise
 
 
+@asynccontextmanager
+async def request_session() -> AsyncGenerator[AsyncSession]:
+    """Session for request-tier work that cannot use Depends(get_db).
+
+    Same engine/pool and the same request statement_timeout cap as get_db,
+    without the DI lifecycle: no auto-commit, caller manages transactions.
+    For paths that serve a user request but manage their own session - e.g.
+    the SSE front stream, which builds its snapshot in a short-lived session
+    so no DB connection is held while the stream is open.
+
+    Every session that runs queries on behalf of a user request MUST carry
+    the request timeout cap. The 2026-08-13 incident: the stream snapshot
+    used a bare async_session_factory() session, so the coalesce query it
+    ran had no statement_timeout and kept spilling query temp for ~13
+    minutes after the request-path twin of the same query had been killed
+    at 30 s - filling the disk and taking Postgres down for everyone. Use
+    this instead of async_session_factory() for anything request-shaped.
+    """
+    async with async_session_factory() as session:
+        session.info[_TIMEOUT_INFO_KEY] = settings.db_statement_timeout_ms
+        yield session
+
+
 @asynccontextmanager
 async def job_session() -> AsyncGenerator[AsyncSession]:
     """Session for background jobs that want an explicit statement_timeout
@@ -114,15 +155,11 @@ async def job_session() -> AsyncGenerator[AsyncSession]:
     ceiling that is still far above the short request timeout. Unlike get_db
     this does NOT auto-commit; the job manages its own transactions.
 
-    Note: SET LOCAL is per-transaction, so for a job that commits between
-    units of work the cap re-applies only to the first transaction after
-    entry. Jobs here follow an execute-then-commit pattern (not
-    `async with db.begin()`), so the pre-emptive SET joins the same
-    transaction as the job's first query. Left as an opt-in because the
-    default ceiling is unlimited.
+    The after_begin listener re-applies the cap on every transaction, so a
+    job that commits between units of work keeps its ceiling for each one
+    (this previously only covered the first transaction after entry). Left
+    as an opt-in because the default ceiling is unlimited.
     """
     async with async_session_factory() as session:
-        await _set_local_statement_timeout(
-            session, settings.db_job_statement_timeout_ms
-        )
+        session.info[_TIMEOUT_INFO_KEY] = settings.db_job_statement_timeout_ms
         yield session