diff --git a/docs/RemoteProxy.md b/docs/RemoteProxy.md
index 515cfcf..da53206 100644
--- a/docs/RemoteProxy.md
+++ b/docs/RemoteProxy.md
@@ -36,7 +36,11 @@ In the stored config (`DomainConfig.remote`):
- `url` — the remote instance's base URL. The satellite connects to
`{url}/auth/api/sync/ws`; the connection is server-to-server and not
- host-dispatched, so internal addresses work.
+ host-dispatched, so internal addresses work. The admin dialog probes
+ `{url}/auth/api/settings` as you type: it must be reachable, serve this
+ very rp-id, and not be a satellite for it itself (nested satellites are
+ not supported — always point at the authoritative instance). Probe
+ findings are warnings; the server-side validation is authoritative.
- `token` — bearer token for the sync channel. The **remote accepts tokens
via its `PASKIA_SYNC_TOKENS` environment variable** (comma-separated);
nothing is stored in the remote's database, and with the variable unset
@@ -44,9 +48,13 @@ In the stored config (`DomainConfig.remote`):
API (an empty field keeps the stored one).
- `cache_ttl` — seconds the replica remains trusted after the sync channel
goes down; then checks fail closed (503). Set it large (up to the 24 h
- session lifetime) for fail-open behavior.
+ session lifetime) for fail-open behavior. Minimum 1.
- `refresh_interval` — seconds between reconnects; every connect starts
- from a full snapshot, which reconciles any drift.
+ from a full snapshot, which reconciles any drift. Minimum 30.
+
+Several domains may share one remote (they share a single replica); their
+remote settings must then be identical — conflicting configurations are
+rejected at validation and aligned with a warning at load.
A remote domain **must mark an auth host** (validated cross-domain and in
the UI): the profile, admin and sign-in pages live there, so browsers and
@@ -86,6 +94,13 @@ signing key is never replicated.
| `/auth/oidc/*` | proxied (signing key and OIDC sessions stay on the remote) |
| `/auth/ws/*`, `/auth/remote-auth/*`, admin, profile | not served — the auth host requirement means these are reached on the remote directly |
+Proxied calls answer 503 when the remote is unreachable. `/auth/api/settings`
+reports `remote: true` for a satellite-served domain (this is what the admin
+dialog's uplink probe checks). Should a remote domain's auth host be
+misrouted to the satellite anyway, profile/avatar requests
+(`/auth/api/user/*`) are forwarded to the remote rather than answered from
+local storage — the satellite never takes writes for a remote domain.
+
Freshness hierarchy:
1. Changes made **through** the satellite: immediate (write-behind,
diff --git a/frontend/src/admin/dialogs/DomainEditDialog.vue b/frontend/src/admin/dialogs/DomainEditDialog.vue
index 26682ac..4eb7745 100644
--- a/frontend/src/admin/dialogs/DomainEditDialog.vue
+++ b/frontend/src/admin/dialogs/DomainEditDialog.vue
@@ -45,6 +45,44 @@ const remoteMissingAuthHost = computed(
() => !!props.dialog.data?.remote && !props.dialog.data?.auth_host
)
+// --- Uplink verification ---
+// Fetch the remote's /auth/api/settings (CORS-open), like the origin
+// connectivity probes: proves reachability, that the peer is a paskia
+// instance, and that it serves this very domain authoritatively — not as
+// a satellite itself (nested satellites are not supported). Results are
+// warnings, never submit blockers (the remote may not be deployed yet).
+const uplinkCheck = ref(null) // null|validating|valid|unreachable|mismatch|nested
+let uplinkTimer = null
+let uplinkSeq = 0
+watch(
+ () => [props.dialog.data?.remote?.url, dialogRpId.value],
+ () => {
+ clearTimeout(uplinkTimer)
+ uplinkCheck.value = null
+ const url = props.dialog.data?.remote?.url?.trim().replace(/\/+$/, '')
+ if (!url || remoteUrlInvalid.value) return
+ uplinkTimer = setTimeout(() => checkUplink(url), 600)
+ },
+ { immediate: true }
+)
+
+async function checkUplink(url) {
+ const seq = ++uplinkSeq
+ uplinkCheck.value = 'validating'
+ try {
+ const response = await fetch(`${url}/auth/api/settings`, {
+ headers: { Accept: 'application/json' },
+ })
+ if (seq !== uplinkSeq) return // url/rp-id changed while fetching
+ if (!response.ok) throw new Error('not ok')
+ const data = await response.json()
+ uplinkCheck.value =
+ data.rp_id !== dialogRpId.value ? 'mismatch' : data.remote ? 'nested' : 'valid'
+ } catch {
+ if (seq === uplinkSeq) uplinkCheck.value = 'unreachable'
+ }
+}
+
// Block submit on hard errors: malformed entries, an over-cap related
// list (the server rejects the save), a save that would lock the admin
// out of the domain they are using, or validation still in flight.
@@ -406,6 +444,7 @@ function onDocumentClick(e) {
onMounted(() => document.addEventListener('click', onDocumentClick))
onBeforeUnmount(() => {
document.removeEventListener('click', onDocumentClick)
+ clearTimeout(uplinkTimer)
for (const t of originValidateTimers.values()) clearTimeout(t)
originValidateTimers.clear()
})
@@ -556,6 +595,11 @@ function onRemoveOrigin(i) {
Must be an http(s) URL.
+ Checking the uplink…
+ ✓ Uplink verified — the remote serves this domain authoritatively.
+ The remote is unreachable — the uplink cannot be verified. Save anyway only if it is not deployed yet.
+ The remote does not serve {{ dialogRpId }} — check the URL. Connect directly to the authoritative instance.
+ The remote is itself a satellite for this domain — nested satellites are not supported. Point directly at the authoritative instance.
A remote domain must mark an auth host above — profile, admin and sign-in pages live there (typically the remote's own site).
Sync token
diff --git a/paskia/domains.py b/paskia/domains.py
index 3a56b6d..d284c13 100644
--- a/paskia/domains.py
+++ b/paskia/domains.py
@@ -299,6 +299,21 @@ def validate_config(
f"domain '{related_hosts[hn]}'"
)
+ # Several domains may share one remote instance (and its replica), but
+ # then its settings must be identical — a conflict would otherwise be
+ # resolved silently and arbitrarily.
+ remotes: dict[str, tuple[str, RemoteConfig]] = {}
+ for rp_id, domain in config.domains.items():
+ if domain.remote is None:
+ continue
+ prior = remotes.setdefault(domain.remote.url, (rp_id, domain.remote))
+ if prior[1] != domain.remote:
+ raise ValueError(
+ f"Domains '{prior[0]}' and '{rp_id}' share remote "
+ f"'{domain.remote.url}' with conflicting settings — a shared "
+ "remote must have one configuration"
+ )
+
def sanitize_config(
config: Config, *, related_origin_cap: int = DEFAULT_RELATED_ORIGIN_CAP
@@ -317,6 +332,7 @@ def sanitize_config(
warnings.append(msg)
domains: dict[str, DomainConfig] = {}
+ remotes_seen: dict[str, RemoteConfig] = {}
for rp_id, domain in config.domains.items():
try:
hostutil.validate_rp_id(rp_id)
@@ -384,6 +400,14 @@ def sanitize_config(
remote = None
else:
remote = domain.remote
+ prior = remotes_seen.setdefault(remote.url, remote)
+ if prior != remote:
+ warn(
+ f"Domain '{rp_id}': remote '{remote.url}' settings "
+ "conflict with another domain's — aligned to the "
+ "first configuration"
+ )
+ remote = prior
if not auth_seen:
warn(
f"Domain '{rp_id}': remote domain without an auth host — "
@@ -493,6 +517,7 @@ def _derive_site(
_registry: DomainRegistry | None = None
_listen: list[str] | None = None
_rebuild_listeners: list = []
+_rebuild_tasks: set[asyncio.Task] = set()
def add_rebuild_listener(fn) -> None:
@@ -505,6 +530,34 @@ def remove_rebuild_listener(fn) -> None:
_rebuild_listeners.remove(fn)
+def _rebuild_done(task: asyncio.Task) -> None:
+ _rebuild_tasks.discard(task)
+ if not task.cancelled() and (exc := task.exception()) is not None:
+ logger.error("Registry rebuild listener failed: %s", exc)
+
+
+def _notify_rebuild(reg: DomainRegistry) -> None:
+ """Run rebuild listeners; schedule coroutines on the running loop.
+
+ Every serving context (lifespan, admin rebuild, tests) runs a loop.
+ Task references are held until completion so listeners are neither
+ garbage-collected mid-run nor silently failing.
+ """
+ for fn in _rebuild_listeners:
+ result = fn(reg)
+ if not asyncio.iscoroutine(result):
+ continue
+ try:
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ result.close()
+ logger.warning("Rebuild listener %r skipped: no running event loop", fn)
+ continue
+ task = loop.create_task(result)
+ _rebuild_tasks.add(task)
+ task.add_done_callback(_rebuild_done)
+
+
def configure(*, listen: list[str] | None = None) -> None:
"""Record process-global serve parameters for site URL derivation."""
global _listen
@@ -543,12 +596,7 @@ def init_registry(config: Config) -> DomainRegistry:
"""Build and install the global registry from a combined configuration."""
global _registry
_registry = build(config)
- for fn in _rebuild_listeners:
- result = fn(_registry)
- if asyncio.iscoroutine(result):
- # init_registry runs within a running loop in every serving
- # context (lifespan, tests, admin rebuild).
- asyncio.get_running_loop().create_task(result)
+ _notify_rebuild(_registry)
return _registry
diff --git a/paskia/fastapi/admin/domains.py b/paskia/fastapi/admin/domains.py
index b2530c4..9b1fe32 100644
--- a/paskia/fastapi/admin/domains.py
+++ b/paskia/fastapi/admin/domains.py
@@ -62,8 +62,8 @@ def _normalize_remote(
return RemoteConfig(
url=value["url"].rstrip("/"),
token=token,
- cache_ttl=int(value.get("cache_ttl") or 60),
- refresh_interval=int(value.get("refresh_interval") or 300),
+ cache_ttl=max(1, int(value.get("cache_ttl") or 60)),
+ refresh_interval=max(30, int(value.get("refresh_interval") or 300)),
)
diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py
index bee3b3c..85dc992 100644
--- a/paskia/fastapi/api.py
+++ b/paskia/fastapi/api.py
@@ -312,6 +312,7 @@ async def get_settings():
auth_site_url=domain.auth_site_url,
session_cookie=AUTH_COOKIE_NAME,
version=__version__,
+ remote=domain.remote is not None,
),
headers={"Access-Control-Allow-Origin": "*", "Vary": "Origin"},
)
diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py
index d05a654..0b017da 100644
--- a/paskia/fastapi/sync.py
+++ b/paskia/fastapi/sync.py
@@ -92,3 +92,5 @@ async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None:
await _send(ws, await queue.get())
except WebSocketDisconnect, RuntimeError, asyncio.CancelledError:
pass
+ except Exception:
+ _logger.debug("Sync event pump ended", exc_info=True)
diff --git a/paskia/fastapi/user.py b/paskia/fastapi/user.py
index c775116..7148604 100644
--- a/paskia/fastapi/user.py
+++ b/paskia/fastapi/user.py
@@ -11,7 +11,7 @@ from fastapi import (
)
from fastapi.responses import FileResponse, JSONResponse
-from paskia import db
+from paskia import db, satellite
from paskia.authsession import (
delete_credential,
expires,
@@ -27,6 +27,19 @@ from paskia.util.apistructs import ApiCreateLinkResponse
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+@app.middleware("http")
+async def proxy_remote_domain(request: Request, call_next):
+ """Profile and avatar data live on the remote for remote domains.
+
+ A remote domain's auth host normally resolves to the remote itself;
+ if traffic for it reaches the satellite anyway (misrouted DNS), the
+ satellite forwards rather than writing to its local database.
+ """
+ if (proxied := await satellite.forward_request(request)) is not None:
+ return proxied
+ return await call_next(request)
+
+
def _can_manage_avatar(ctx, target_user) -> bool:
if ctx.user.uuid == target_user.uuid:
return True
diff --git a/paskia/satellite.py b/paskia/satellite.py
index c5d62db..52a671e 100644
--- a/paskia/satellite.py
+++ b/paskia/satellite.py
@@ -78,7 +78,7 @@ class RemoteReplica:
)
def refresh_session(
- self, key: str, validated, ip: str | None, user_agent: str | None
+ self, key: str, validated: datetime, ip: str | None, user_agent: str | None
) -> None:
"""Apply a /validate refresh locally and queue it for the remote."""
session = self.db.sessions.get(key)
@@ -227,6 +227,8 @@ class SatelliteManager:
for replica in self.replicas.values():
await replica.stop()
self.replicas.clear()
+ for url in list(_clients):
+ await _drop_client(url)
async def reconcile(self, registry: domains.DomainRegistry) -> None:
"""Start/stop replicas to match the configured remote domains."""
@@ -237,6 +239,7 @@ class SatelliteManager:
for url in list(self.replicas):
if url not in wanted:
await self.replicas.pop(url).stop()
+ await _drop_client(url)
for url, remote in wanted.items():
replica = self.replicas.get(url)
if replica is None or replica.remote != remote:
@@ -275,7 +278,12 @@ def store_for_host(host: str | None) -> DB:
def refresh_session(
- key, host: str | None, ip: str, user_agent: str, validated, ctx=None
+ key: str,
+ host: str | None,
+ ip: str | None,
+ user_agent: str | None,
+ validated: datetime,
+ ctx=None,
):
"""/validate refresh: write-behind for remote domains, else local DB."""
replica = replica_for_host(host)
@@ -313,6 +321,13 @@ _HOP_BY_HOP = {
_clients: dict[str, httpx.AsyncClient] = {}
+async def _drop_client(url: str) -> None:
+ """Close and forget the forwarding client for a remote URL."""
+ client = _clients.pop(url, None)
+ if client is not None:
+ await client.aclose()
+
+
async def forward_request(request: Request) -> Response | None:
"""Forward the request to its domain's remote, or None when local.
@@ -327,15 +342,18 @@ async def forward_request(request: Request) -> Response | None:
client = _clients.get(url)
if client is None:
client = _clients[url] = httpx.AsyncClient(base_url=url, timeout=_TIMEOUT)
- upstream = await client.request(
- request.method,
- request.url.path,
- params=request.url.query,
- content=await request.body(),
- headers={
- k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP
- },
- )
+ try:
+ upstream = await client.request(
+ request.method,
+ request.url.path,
+ params=request.url.query,
+ content=await request.body(),
+ headers={
+ k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP
+ },
+ )
+ except httpx.TransportError as e:
+ raise HTTPException(503, "Remote authentication service unavailable") from e
response = Response(content=upstream.content, status_code=upstream.status_code)
# Raw headers to preserve repeated Set-Cookie
response.raw_headers = [
diff --git a/paskia/syncfeed.py b/paskia/syncfeed.py
index 2ce9fb2..8c3d3cd 100644
--- a/paskia/syncfeed.py
+++ b/paskia/syncfeed.py
@@ -23,6 +23,8 @@ _subscribers: set[asyncio.Queue] = set()
def emit(table: str, key: str, obj) -> None:
"""Publish an upsert (obj given) or delete (obj None) to subscribers."""
+ if not _subscribers:
+ return
event = {
"type": "event",
"table": table,
diff --git a/paskia/util/apistructs.py b/paskia/util/apistructs.py
index 376cb35..b95b6b2 100644
--- a/paskia/util/apistructs.py
+++ b/paskia/util/apistructs.py
@@ -185,6 +185,7 @@ class ApiSettings(msgspec.Struct):
auth_site_url: str
session_cookie: str
version: str
+ remote: bool = False # this instance satellite-serves the domain
class ApiDomain(msgspec.Struct):
diff --git a/tests/test_remote.py b/tests/test_remote.py
index b878d61..0d6bad5 100644
--- a/tests/test_remote.py
+++ b/tests/test_remote.py
@@ -1,5 +1,6 @@
"""Tests for remote (satellite) domains: config, replica application, feed."""
+import asyncio
import secrets
import time
from datetime import UTC, datetime
@@ -375,7 +376,10 @@ async def test_admin_configures_remote_domain(client, session_token, test_db):
assert entry["remote"]["url"] == "http://remote.test"
assert "token" not in entry["remote"] # write-only
- headers = {"Host": "localhost:4401", "Cookie": f"{AUTH_COOKIE_NAME}={session_token}"}
+ headers = {
+ "Host": "localhost:4401",
+ "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+ }
origins = {"**.example.com": True, "auth.example.com": {"auth_host": True}}
# PATCH without the remote key preserves it (and its token)
@@ -391,8 +395,11 @@ async def test_admin_configures_remote_domain(client, session_token, test_db):
# PATCH with a new URL but no token keeps the stored token
r = await client.patch(
"/auth/api/admin/domains/example.com",
- json={"rp_name": "Ex", "origins": origins,
- "remote": {"url": "http://other.test", "cache_ttl": 30}},
+ json={
+ "rp_name": "Ex",
+ "origins": origins,
+ "remote": {"url": "http://other.test", "cache_ttl": 30},
+ },
headers=headers,
)
assert r.status_code == 200, r.text
@@ -425,3 +432,140 @@ async def test_admin_remote_domain_requires_auth_host(client, session_token):
)
assert r.status_code == 400
assert "auth host" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_admin_remote_values_clamped(client, session_token, test_db):
+ """Nonsensical TTL/interval values are clamped to the supported floor."""
+ r = await client.post(
+ "/auth/api/admin/domains/",
+ json={
+ "rp_id": "example.com",
+ "origins": {
+ "**.example.com": True,
+ "auth.example.com": {"auth_host": True},
+ },
+ "remote": {
+ "url": "http://remote.test",
+ "token": "t",
+ "cache_ttl": -5,
+ "refresh_interval": 1,
+ },
+ },
+ headers={
+ "Host": "localhost:4401",
+ "Cookie": f"{AUTH_COOKIE_NAME}={session_token}",
+ },
+ )
+ assert r.status_code == 200, r.text
+ remote = test_db.config.domains["example.com"].remote
+ assert remote.cache_ttl == 1
+ assert remote.refresh_interval == 30
+
+
+def _add_shared_remote(config: Config, token: str) -> None:
+ config.domains["sub.example.com"] = DomainConfig(
+ origins={
+ "**.sub.example.com": True,
+ "auth.sub.example.com": OriginEntry(auth_host=True),
+ },
+ remote=RemoteConfig(url=REMOTE_URL, token=token),
+ )
+
+
+def test_shared_remote_same_settings_ok():
+ config = _remote_domain_config()
+ _add_shared_remote(config, token="t")
+ domains.validate_config(config)
+
+
+def test_shared_remote_conflicting_settings_rejected():
+ config = _remote_domain_config()
+ _add_shared_remote(config, token="different")
+ with pytest.raises(ValueError, match="conflicting"):
+ domains.validate_config(config)
+
+
+def test_sanitize_aligns_conflicting_shared_remote():
+ config = _remote_domain_config()
+ _add_shared_remote(config, token="different")
+ sanitized, warnings = domains.sanitize_config(config)
+ assert any("conflict" in w for w in warnings)
+ # Aligned to the first configuration, and the result validates
+ assert sanitized.domains["sub.example.com"].remote.token == "t"
+ domains.validate_config(sanitized)
+
+
+def test_rebuild_listener_skipped_without_running_loop():
+ """A registry rebuild outside an event loop warns instead of crashing."""
+
+ async def listener(_registry):
+ raise AssertionError("must not run") # pragma: no cover
+
+ domains.add_rebuild_listener(listener)
+ try:
+ domains.init_registry(_remote_domain_config())
+ finally:
+ domains.remove_rebuild_listener(listener)
+
+
+@pytest.mark.asyncio
+async def test_rebuild_listener_runs_with_loop():
+ ran = asyncio.Event()
+
+ async def listener(_registry):
+ ran.set()
+
+ domains.add_rebuild_listener(listener)
+ try:
+ domains.init_registry(_remote_domain_config())
+ await asyncio.wait_for(ran.wait(), 2)
+ finally:
+ domains.remove_rebuild_listener(listener)
+
+
+@pytest.mark.asyncio
+async def test_settings_reports_remote_flag(remote_client):
+ client, _, _ = remote_client
+ r = await client.get("/auth/api/settings", headers={"Host": "app2.example.com"})
+ assert r.status_code == 200
+ assert r.json()["remote"] is True
+ r = await client.get("/auth/api/settings", headers={"Host": "localhost:4401"})
+ assert r.json()["remote"] is False
+
+
+@pytest.mark.asyncio
+async def test_logout_remote_down_returns_503(remote_client, monkeypatch):
+ """A transport failure on a proxied call is a deliberate 503, not a 500."""
+ client, secret, replica = remote_client
+
+ class Down:
+ async def request(self, *args, **kwargs):
+ raise httpx.ConnectError("down")
+
+ monkeypatch.setitem(satellite._clients, REMOTE_URL, Down())
+ r = await client.post(
+ "/auth/api/logout",
+ headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},
+ )
+ assert r.status_code == 503
+ # Not evicted: the remote never saw the logout, the session is still valid
+ assert replica.db.sessions
+
+
+@pytest.mark.asyncio
+async def test_user_api_proxied_on_remote_auth_host(remote_client, monkeypatch):
+ """Profile traffic for a remote domain's auth host is forwarded, never
+ written to the satellite's local database."""
+ client, _, _ = remote_client
+
+ async def fake_forward(request):
+ return Response(status_code=200, content=b"proxied")
+
+ monkeypatch.setattr(satellite, "forward_request", fake_forward)
+ r = await client.get(
+ f"/auth/api/user/{UUID(int=104)}/profile.webp",
+ headers={"Host": "auth.example.com"},
+ )
+ assert r.status_code == 200
+ assert r.content == b"proxied"
diff --git a/tests/test_satellite_e2e.py b/tests/test_satellite_e2e.py
new file mode 100644
index 0000000..43b3d71
--- /dev/null
+++ b/tests/test_satellite_e2e.py
@@ -0,0 +1,252 @@
+"""Live end-to-end tests for satellite domains over a real sync channel.
+
+A real uvicorn server plays the remote (its sync WebSocket fed from the
+process-global database), while the satellite side runs through the ASGI
+app with a real RemoteReplica started by the real SatelliteManager
+reconcile. Unlike test_remote.py (which injects a warm replica by hand),
+nothing here is stubbed: the snapshot, live events, write-behind, dead-peer
+TTL handling and reconnect reconciliation all cross a TCP socket.
+
+One in-process distortion remains: both sides share the module-level
+database and registry, so endpoints the satellite *forwards* to the remote
+(logout, set-session, OIDC) would re-forward and loop here — those paths
+are covered in test_remote.py with a mocked forward_request.
+"""
+
+import asyncio
+import socket
+import time
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import httpx
+import pytest
+import pytest_asyncio
+import uvicorn
+
+import paskia.db.operations as ops_db
+from paskia import domains, satellite
+from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig
+from paskia.fastapi.mainapp import app
+from paskia.fastapi.session import AUTH_COOKIE_NAME
+
+from .conftest import TEST_RP_ID, create_test_session
+
+REMOTE_DOMAIN = "example.com"
+APP_HOST = "app2.example.com"
+
+
+def _free_port() -> int:
+ with socket.socket() as s:
+ s.bind(("127.0.0.1", 0))
+ return s.getsockname()[1]
+
+
+async def wait_for(desc: str, condition, timeout: float = 10) -> None:
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ if condition():
+ return
+ await asyncio.sleep(0.02)
+ raise AssertionError(f"timed out waiting for {desc}")
+
+
+@pytest_asyncio.fixture
+async def live_remote(test_db, monkeypatch):
+ """A uvicorn-served 'remote' and the satellite manager wired against it.
+
+ The manager is started for real: reconcile builds the replica from the
+ domain config and its sync client connects over TCP. Yielded namespace:
+ client (ASGI, satellite side), replica, db (the remote's database),
+ url/port of the remote, start_server (for restarts), servers.
+ """
+ monkeypatch.setenv("PASKIA_SYNC_TOKENS", "live-tok")
+ servers: list[tuple[uvicorn.Server, asyncio.Task]] = []
+
+ async def start_server(port: int) -> uvicorn.Server:
+ config = uvicorn.Config(
+ app, host="127.0.0.1", port=port, log_level="error", lifespan="off"
+ )
+ server = uvicorn.Server(config)
+ task = asyncio.create_task(server.serve())
+ servers.append((server, task))
+ while not server.started:
+ await asyncio.sleep(0.02)
+ return server
+
+ port = _free_port()
+ url = f"http://127.0.0.1:{port}"
+ await start_server(port)
+
+ domains.configure(listen=["localhost:4401"])
+ domains.init_registry(
+ Config(
+ domains={
+ TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
+ REMOTE_DOMAIN: DomainConfig(
+ origins={
+ f"**.{REMOTE_DOMAIN}": True,
+ f"auth.{REMOTE_DOMAIN}": OriginEntry(auth_host=True),
+ },
+ remote=RemoteConfig(url=url, token="live-tok"),
+ ),
+ }
+ )
+ )
+ await satellite.manager.start()
+ replica = satellite.manager.replicas[url]
+ await wait_for("initial snapshot", lambda: replica.available())
+
+ transport = httpx.ASGITransport(app=app)
+ async with httpx.AsyncClient(
+ transport=transport, base_url="http://localhost:4401"
+ ) as client:
+ try:
+ yield SimpleNamespace(
+ client=client,
+ replica=replica,
+ db=test_db,
+ url=url,
+ port=port,
+ start_server=start_server,
+ servers=servers,
+ )
+ finally:
+ await satellite.manager.stop()
+ for server, task in servers:
+ server.should_exit = True
+ await task
+
+
+def _app_headers(secret: str) -> dict[str, str]:
+ return {"Host": APP_HOST, "Cookie": f"{AUTH_COOKIE_NAME}={secret}"}
+
+
+@pytest.mark.asyncio
+async def test_live_snapshot_and_user_events(live_remote, test_user):
+ """Bootstrap data arrives with the snapshot; later mutations as events."""
+ assert test_user.uuid in live_remote.replica.db.users
+ ops_db.update_user_display_name(test_user.uuid, "Live Rename")
+ await wait_for(
+ "rename to propagate",
+ lambda: (
+ live_remote.replica.db.users[test_user.uuid].display_name == "Live Rename"
+ ),
+ )
+
+
+@pytest.mark.asyncio
+async def test_live_forward_and_writebehind(live_remote, test_user, test_credential):
+ """A session created on the remote authorizes locally; the /validate
+ refresh travels back over the wire into the remote's database."""
+ live = live_remote
+ key, secret = create_test_session(
+ user_uuid=test_user.uuid,
+ credential_uuid=test_credential.uuid,
+ host=APP_HOST,
+ rp_id=REMOTE_DOMAIN,
+ )
+ await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+ r = await live.client.get(
+ "/auth/api/forward?perm=auth:admin", headers=_app_headers(secret)
+ )
+ assert r.status_code == 204
+ assert r.headers["remote-name"] == "Test Admin"
+
+ # Age the session on the remote (propagates), then validate on the satellite
+ stale = datetime.now(UTC) - timedelta(minutes=10)
+ ops_db.update_session(key, validated=stale)
+ await wait_for(
+ "staleness to propagate",
+ lambda: live.replica.db.sessions[key].validated == stale,
+ )
+ r = await live.client.post("/auth/api/validate", headers=_app_headers(secret))
+ assert r.status_code == 200
+ assert r.json()["renewed"] is True
+ await wait_for(
+ "write-behind to land on the remote",
+ lambda: live.db.sessions[key].validated > stale,
+ )
+
+
+@pytest.mark.asyncio
+async def test_live_session_delete_propagates(live_remote, test_user, test_credential):
+ live = live_remote
+ key, secret = create_test_session(
+ user_uuid=test_user.uuid,
+ credential_uuid=test_credential.uuid,
+ host=APP_HOST,
+ rp_id=REMOTE_DOMAIN,
+ )
+ await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+ ops_db.delete_session(key)
+ await wait_for(
+ "session delete to propagate", lambda: key not in live.replica.db.sessions
+ )
+ r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+ assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_live_wrong_token_rejected(live_remote):
+ bad = satellite.RemoteReplica(RemoteConfig(url=live_remote.url, token="wrong"))
+ await bad.start()
+ try:
+ await asyncio.sleep(1.5)
+ assert not bad.available()
+ finally:
+ await bad.stop()
+
+
+@pytest.mark.asyncio
+async def test_live_disconnect_ttl_and_reconnect(
+ live_remote, test_user, test_credential
+):
+ """Channel down: replica keeps serving within cache_ttl, then fails
+ closed (503); reconnect starts from a snapshot and reconciles drift."""
+ live = live_remote
+ key, secret = create_test_session(
+ user_uuid=test_user.uuid,
+ credential_uuid=test_credential.uuid,
+ host=APP_HOST,
+ rp_id=REMOTE_DOMAIN,
+ )
+ await wait_for("session sync", lambda: key in live.replica.db.sessions)
+
+ server, task = live.servers[0]
+ server.should_exit = True
+ await task
+ await wait_for(
+ "dead peer detection", lambda: not live.replica.connected, timeout=15
+ )
+
+ r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+ assert r.status_code == 204 # still authoritative within cache_ttl
+
+ live.replica.remote.cache_ttl = 0 # expire the trust window
+ r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+ assert r.status_code == 503
+
+ # Drift while the channel is down, then let the client auto-reconnect
+ ops_db.update_user_display_name(test_user.uuid, "While Down")
+ await live.start_server(live.port)
+ await wait_for("reconnect", lambda: live.replica.available(), timeout=20)
+ assert (
+ live.replica.db.users[test_user.uuid].display_name == "While Down"
+ ) # snapshot reconciled the drift
+ r = await live.client.get("/auth/api/forward", headers=_app_headers(secret))
+ assert r.status_code == 204
+
+
+@pytest.mark.asyncio
+async def test_reconcile_stops_replica_when_remote_removed(live_remote):
+ """A registry rebuild without the remote config stops the replica."""
+ live = live_remote
+ domains.init_registry(
+ Config(domains={TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True})})
+ )
+ await wait_for(
+ "replica shutdown", lambda: live.url not in satellite.manager.replicas
+ )