perf(tenants,permissions): single-flight auth cache fills, unknown-subdomain cache, paginated members

CHANGELOG.md

@@ -67,6 +67,7 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Changed
 
+- **Breaking:** tenant member lists now use server pagination (`page` / `per_page`, default 25, maximum 200). The current-members API returns `{items, total, page, per_page, pages}` instead of an array; the Members page provides navigation and retains its page after mutations. `TenantService.list_members(tenant_id, params)` now takes `PageParams` and returns `Page[Membership]` instead of a list.
 - Cache env-supplied settings fields per class and default API version discovery per process; changes to environment, dotenv, or secret sources require a restart.
 - Pre-serialize shared Inertia i18n catalogs per locale, audience, and catalog load, preserving page JSON bytes and HTML escaping.
 - Reduce background task polling and list reads to their required columns; count task history separately from paginated reads and index newest-first, nulls-last ordering (migration `c5d8e2f1a903`, 2026-11-05 performance review).
@@ -85,6 +86,7 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Fixed
 
+- Coalesce concurrent tenant membership and permission grant cache misses. In-flight invalidation retries against fresh authorization data; failures and cancellation release waiting readers. Cache unknown tenant subdomains for five seconds and drop the new slug's host lookups (on a new `tenants.hosts` invalidation channel, so no user id is read as a slug) and the owner's memberships after committed tenant creation; other users' caches are not cleared. Services built outside request dependencies pass `invalidate_hosts=make_host_invalidator(app)` beside `invalidate=make_invalidator(app)`. Retries after concurrent invalidation are bounded to three reads (2026-11-05 local performance review).
 - Offload S3 server-upload spool I/O, wait for in-flight I/O during cancellation, and reuse one client per multipart upload. Index the default tenant file browse order (migration `f6c2d4e8b109`; local performance review, 2026-11-05).
 - Users password creation, forgot-password fingerprints, reset verification and
   password updates offload hashing to worker threads, keeping unrelated async

docs/framework/invalidation.md

@@ -110,3 +110,41 @@ A standalone worker never builds the FastAPI app, so it installs no transport: i
 ## Still per-process
 
 `file_storage`'s aggregate cache remains per-process. Role overlays (`permissions.role_grants`), module settings (`settings.modules`, keyed by package) and feature flags (`feature_flags.overrides`, both scopes) subscribe through their modules' `register_invalidations` hooks. Their refresh tasks start and stop with module lifespan.
+
+## Concurrent cache fills
+
+`simple_module_core.singleflight.SingleFlight` serializes in-process fills for
+one cache key. Keep one helper per app/cache and use `async with flight.hold(key)`
+on a miss. Recheck the cache after acquiring the lock: an earlier reader may
+already have filled it. Different keys proceed independently; idle keys are
+removed after the final caller leaves. Failure or cancellation releases the
+lock, so a waiter can retry without a cached failed task.
+
+The lock does not version or invalidate cached data. Authorization caches must
+also compare their invalidation epoch before storing a DB read, and retry
+against a fresh session if the epoch changed. `simple_module_core.singleflight.fill`
+is that loop:
+
+```python
+return await fill(
+    lambda: read_from_db(key),
+    epoch=lambda: _epoch,  # bumped by every forget
+    store=lambda value: cache.__setitem__(key, value),
+    cached=lambda: cache.get(key, MISS),  # MISS: nothing cached; None is a value
+)
+```
+
+It reads at most `MAX_FILL_ATTEMPTS` (3) times. When every read raced an
+invalidation it returns the last read **uncached** rather than loop while
+invalidations keep arriving. That answer is not guaranteed to be newer than an
+invalidation that landed during it: it is what an uncached read racing a
+concurrent change would return, and it is never stored. Tenant memberships,
+tenant subdomain lookups and direct permission grants use it inside their
+per-key `SingleFlight` (subdomain lookups without one); their existing TTLs
+still bound staleness from missed cross-worker invalidations. The users
+session-version cache has not yet adopted the helper.
+
+Give each kind of key its own channel. A handler that dispatches on a key
+prefix misreads any key of the other kind that happens to start with it (the
+tenants module keeps user ids on `tenants.membership` and slugs on
+`tenants.hosts` for this reason).

docs/modules/tenants.md

@@ -141,6 +141,50 @@ Providers own persistence, period boundaries, idempotency and hold expiry.
 choose to raise it and map it to HTTP 402; the protocol's `consume` returns a
 decision rather than requiring an exception.
 
+## Member pagination and resolver caches
+
+`GET /api/tenants/current/members` accepts `page` (1-based) and `per_page`
+(default 25, maximum 200). It returns `{items, total, page, per_page, pages}`;
+clients previously reading a bare array must read `items` and traverse the pages.
+Invalid API parameters return 422. The Members page accepts the same parameters,
+normalizes malformed values, and provides Previous/Next controls. Both order by
+`(created_at, id)` and explicitly scope rows and totals to the active tenant.
+Requests past the end clamp to the last page, including after a member removal;
+mutations reload the current page and page size. Offset pagination reflects the
+current roster, so concurrent additions/removals can shift page boundaries.
+
+Membership cache fills serialize concurrent misses per app/user, recheck the cache
+when waiting ends, and retry if invalidated during a DB read. After three reads
+that each raced an invalidation, the request answers with the last read without
+caching it. That read is not guaranteed to be newer than an invalidation that
+landed during it: it is equivalent to an uncached read racing a concurrent
+change, and it is never cached. The 60-second membership TTL and invalidation
+bus remain the fallback for changes in other workers. Permission grants and
+subdomain lookups use the same `simple_module_core.singleflight.fill` loop;
+grants keep their 30-second TTL.
+
+Unknown subdomain slugs are negatively cached for five seconds (bounded to 10,000
+entries); known slugs keep their 60-second TTL. Two invalidation channels carry
+the evictions, both published after commit and delivered to every worker when a
+transport is installed:
+
+| Channel | Key | Drops |
+|---|---|---|
+| `tenants.membership` (`INVALIDATION_CHANNEL`) | a user id | that user's memberships |
+| | `None` | every membership **and** every host lookup |
+| `tenants.hosts` (`HOSTS_INVALIDATION_CHANNEL`) | a slug | that slug's host lookups |
+| | `None` | every host lookup |
+
+Membership keys are always user ids, whatever their shape (an OIDC subject can
+be any string), so slugs never share their channel. Committed tenant creation
+drops the owner's memberships and the new slug's host lookups; other users'
+caches stay warm. Status changes and deletion publish `None` on the membership
+channel. Construct services with both invalidators outside request
+dependencies: `TenantService(db, invalidate=make_invalidator(app),
+invalidate_hosts=make_host_invalidator(app))` (from `tenants.resolver` and
+`tenants.host_resolver`). Direct SQL writes must publish `tenants.membership`
+with `key=None` after commit or wait for TTL expiry.
+
 ## Models
 
 `tenants_tenant` (id = the value stored in every `tenant_id`, slug, name,

framework/core/simple_module_core/singleflight.py

@@ -0,0 +1,96 @@
+"""Serialize cache fills per key, and fill caches without storing stale reads.
+
+``SingleFlight`` serializes concurrent misses for one key; callers recheck their
+cache inside the lock. Own one instance per application/cache. No background
+task outlives its caller: failed or cancelled fills release the lock and the
+next waiter retries normally.
+
+``fill`` is the epoch-checked read loop shared by caches that an invalidation
+can drop mid-read: a read only gets stored if no invalidation landed while it
+ran.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable, Hashable
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Final, Literal
+
+# Epoch-checked fills retry at most this many reads; continuous invalidation
+# must not keep a request looping.
+MAX_FILL_ATTEMPTS = 3
+
+
+class Miss(Enum):
+    """Type of :data:`MISS`; a cache lookup's "nothing cached" answer."""
+
+    MISS = "miss"
+
+
+MISS: Final = Miss.MISS
+
+
+def _nothing_cached() -> Literal[Miss.MISS]:
+    return MISS
+
+
+async def fill[T](
+    read: Callable[[], Awaitable[T]],
+    *,
+    epoch: Callable[[], int],
+    store: Callable[[T], None],
+    cached: Callable[[], T | Literal[Miss.MISS]] = _nothing_cached,
+) -> T:
+    """Read through *read* and *store* the result unless invalidated meanwhile.
+
+    *epoch* returns the cache's invalidation counter (bumped by every forget);
+    a read whose epoch changed while it ran is not stored and is retried, after
+    rechecking *cached* (return :data:`MISS` when nothing is cached; ``None``
+    is a valid cached value). At most :data:`MAX_FILL_ATTEMPTS` reads.
+
+    After the last attempt the last read is returned uncached. It is **not**
+    guaranteed to be newer than an invalidation that landed during it: it is
+    equivalent to an uncached read racing a concurrent change, so the caller
+    answers this one request with it, and it is never cached.
+    """
+    attempts = 0
+    while True:
+        hit = cached()
+        if hit is not MISS:
+            return hit
+        started = epoch()
+        value = await read()
+        attempts += 1
+        if epoch() == started:
+            store(value)
+            return value
+        if attempts >= MAX_FILL_ATTEMPTS:
+            return value
+
+
+@dataclass
+class _Flight:
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+    users: int = 0
+
+
+class SingleFlight:
+    def __init__(self) -> None:
+        self._locks: dict[Hashable, _Flight] = {}
+
+    @asynccontextmanager
+    async def hold(self, key: Hashable) -> AsyncGenerator[None]:
+        flight = self._locks.get(key)
+        if flight is None:
+            flight = self._locks[key] = _Flight()
+        flight.users += 1
+        try:
+            async with flight.lock:
+                yield
+        finally:
+            flight.users -= 1
+            if not flight.users:
+                del self._locks[key]

framework/core/tests/test_core_singleflight.py

@@ -0,0 +1,100 @@
+"""Single-flight locks survive failures/cancellation and release idle keys."""
+
+import asyncio
+
+import pytest
+from simple_module_core.singleflight import MAX_FILL_ATTEMPTS, MISS, SingleFlight, fill
+
+
+@pytest.mark.parametrize("cancel_leader", [False, True])
+async def test_failure_or_cancellation_allows_next_reader(cancel_leader):
+    flight = SingleFlight()
+    started, release = asyncio.Event(), asyncio.Event()
+
+    async def leader():
+        async with flight.hold("key"):
+            started.set()
+            await release.wait()
+            raise ValueError("failed read")
+
+    async def follower():
+        async with flight.hold("key"):
+            return "fresh"
+
+    first = asyncio.create_task(leader())
+    await started.wait()
+    second = asyncio.create_task(follower())
+    await asyncio.sleep(0)
+    if cancel_leader:
+        first.cancel()
+    else:
+        release.set()
+    with pytest.raises(asyncio.CancelledError if cancel_leader else ValueError):
+        await first
+    assert await second == "fresh"
+    assert not flight._locks
+
+
+async def test_cancelled_waiter_does_not_unlock_leader_or_leak_key():
+    flight = SingleFlight()
+    entered = asyncio.Event()
+
+    async def waiter():
+        async with flight.hold("key"):
+            entered.set()
+
+    async with flight.hold("key"):
+        task = asyncio.create_task(waiter())
+        await asyncio.sleep(0)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert not entered.is_set()
+        async with flight.hold("different"):
+            pass
+    assert not flight._locks
+
+
+async def test_fill_stores_a_read_no_invalidation_raced():
+    stored = []
+
+    async def read():
+        return "value"
+
+    assert await fill(read, epoch=lambda: 0, store=stored.append) == "value"
+    assert stored == ["value"]
+
+
+async def test_fill_returns_a_cached_none_without_reading():
+    async def read():
+        raise AssertionError("must not read")
+
+    assert await fill(read, epoch=lambda: 0, store=lambda _value: None, cached=lambda: None) is None
+
+
+async def test_fill_retries_then_answers_uncached_under_continuous_invalidation():
+    epoch, stored, reads = [0], [], []
+
+    async def read():
+        reads.append(len(reads))
+        epoch[0] += 1  # every read races an invalidation
+        return len(reads)
+
+    result = await fill(read, epoch=lambda: epoch[0], store=stored.append, cached=lambda: MISS)
+    assert result == MAX_FILL_ATTEMPTS
+    assert len(reads) == MAX_FILL_ATTEMPTS
+    assert stored == []
+
+
+async def test_fill_rechecks_the_cache_before_retrying():
+    epoch, cache = [0], {}
+
+    async def read():
+        epoch[0] += 1
+        cache["k"] = "filled-by-another-reader"
+        return "raced"
+
+    result = await fill(
+        read, epoch=lambda: epoch[0], store=lambda _value: None, cached=lambda: cache.get("k", MISS)
+    )
+    assert result == "filled-by-another-reader"

modules/api_tokens/tests/credentials/resolver/test_api_tokens_hardening.py

@@ -32,7 +32,7 @@
 
 async def _member_ids(resp: httpx.Response) -> set[str]:
     assert resp.status_code == 200, resp.text
-    return {str(m["user_id"]) for m in resp.json()}
+    return {str(m["user_id"]) for m in resp.json()["items"]}
 
 
 async def test_pat_is_pinned_to_its_tenant_whatever_the_header_says(app, actor, bearer):