Route live domains pushes through the connection queue

The pump stays the single sender on a connection — a direct send from the
admin path could interleave frames with an in-flight event. Queuing also
makes the push helpers synchronous, and a full queue drops the subscriber
like syncfeed does (the reconnect's welcome re-pushes the domain set).

README.md

@@ -236,3 +236,4 @@ See [Multi-Site documentation](docs/MultiSite.md) for details.
 - [Trusted Headers for Backend Apps](https://git.zi.fi/LeoVasanko/paskia/src/branch/main/docs/Headers.md)
 - [Frontend integration](https://git.zi.fi/LeoVasanko/paskia/src/branch/main/docs/Integration.md)
 - [Paskia API](https://git.zi.fi/LeoVasanko/paskia/src/branch/main/docs/API.md)
+- [Remote Proxy / Satellite](https://git.zi.fi/LeoVasanko/paskia/src/branch/main/docs/RemoteProxy.md) — serve domains from a local satellite instance

docs/API.md

@@ -76,8 +76,15 @@ E.g. Org admin cannot see anything of the other orgs that he has no admin access
 | POST | /auth/api/admin/domains/ | Create domain `{rp_id, rp_name?, origins?, related?}` | 200/400/401/403 |
 | PATCH | /auth/api/admin/domains/{rp_id} | Update domain rp_name/origins/related | 200/400/401/403 |
 | DELETE | /auth/api/admin/domains/{rp_id} | Delete domain (refused while credentials remain) | 200/400/401/403 |
+| GET | /auth/api/admin/satellites/ | List registered satellite links with live state | 200/401/403 |
+| POST | /auth/api/admin/satellites/ | Mint a satellite registration offer `{rp_ids}` — wss link URL, RAM-only, 10 min, single-use | 200/400/401/403 |
+| PATCH | /auth/api/admin/satellites/{id} | Adjust a link's domain scope (pushed over the live link) | 200/400/401/403 |
+| DELETE | /auth/api/admin/satellites/{id} | Revoke a satellite link (drops its connections) | 200/400/401/403 |
+| GET | /auth/api/admin/uplinks/ | List this instance's uplinks with served domains and live state | 200/401/403 |
+| POST | /auth/api/admin/uplinks/ | Register an uplink from a pasted link URL `{url}` (online handshake) | 200/400/401/403 |
+| DELETE | /auth/api/admin/uplinks/{id} | Unlink an uplink (drops it and its domains) | 200/400/401/403 |
 
-Domain endpoints require the `auth:admin` permission; writes additionally require recent authentication (5 minutes). Changes are validated cross-domain and apply immediately.
+Domain endpoints require the `auth:admin` permission; writes additionally require recent authentication (5 minutes). Changes are validated cross-domain and apply immediately. Remote (satellite-served) domains are uplink-managed and cannot be written here; see [Remote Proxy / Satellite](RemoteProxy.md) for the link model. The same applies to the satellite/uplink endpoints.
 
 `origins` is a single object keyed by all of the domain's sign-in sites. Entries *within* the rp-id domain are in-domain sites: bare hosts, wildcards under the rp-id following the shell-glob convention (`**.example.com` covers the apex and subdomains at any depth, `*.example.com` exactly one subdomain level — https only, any scheme and port under localhost; plain `*` is not accepted), or full origins when not https. Entries *outside* the rp-id domain are related origins that may assert this domain's rp-id (WebAuthn Related Origin Requests, max 5, no wildcards); those are published at `/.well-known/webauthn` on the rp-id host. An empty object allows nothing of the domain itself. A value of `true` marks presence; `{"auth_host": true}` additionally marks an in-domain entry as the domain's authentication host.
 

frontend/src/admin/AdminSatellites.vue

@@ -16,6 +16,7 @@
 import { computed, onBeforeUnmount, onMounted, ref } from 'vue'
 import AdminDialog from './dialogs/AdminDialog.vue'
 import { useAuthStore } from '@/stores/auth'
+import { formatDate } from '@/utils/helpers'
 import { apiJson } from 'paskia'
 
 const props = defineProps({
@@ -234,7 +235,7 @@ function uplinkStatus(u) {
         <tr v-for="link in links" :key="link.id">
           <td class="perm-name-cell">
             <div class="perm-title">{{ link.label || 'satellite' }}</div>
-            <div class="perm-id-info"><span class="id-text">{{ link.id }}</span></div>
+            <div class="perm-id-info"><span class="id-text">{{ link.id }}</span><span class="muted">·</span><span>{{ formatDate(link.created) }}</span></div>
           </td>
           <td>
             <span v-if="link.rp_ids.length">{{ link.rp_ids.join(', ') }}</span>
@@ -274,7 +275,7 @@ function uplinkStatus(u) {
         <tr v-for="uplink in uplinks" :key="uplink.id">
           <td class="perm-name-cell">
             <div class="perm-title">{{ uplink.url }}</div>
-            <div class="perm-id-info"><span class="id-text">{{ uplink.id }}</span></div>
+            <div class="perm-id-info"><span class="id-text">{{ uplink.id }}</span><span class="muted">·</span><span>{{ formatDate(uplink.created) }}</span></div>
           </td>
           <td>
             <span v-if="uplink.domains.length">{{ uplink.domains.join(', ') }}</span>

paskia/__main__.py

@@ -243,8 +243,11 @@ def cmd_satellite(args: argparse.Namespace) -> None:
     print(f"🛰  Linked to {info['url']} — serving: {domains_list}")
     for rp_id, reason in (info.get("rejected") or {}).items():
         print(f"⚠️  Domain {rp_id} rejected: {reason}")
+    # A successful CLI run means the server was not holding the database
+    # open — the linked domains start serving on the next start.
+    print("Start the server to serve them: paskia")
     if fresh:
-        print("Created the database. Serve with: paskia")
+        print("Created the database.")
         configure_domains(listen=None)
         registry = build_registry(_load_stored_config(db_path))
         log_reset_link(

paskia/fastapi/admin/domains.py

@@ -196,7 +196,7 @@ async def admin_update_domain(
         ctx=ctx,
     )
     _rebuild_registry()
-    await sync.refresh_links_serving(rp_id)
+    sync.refresh_links_serving(rp_id)
     return {"status": "ok"}
 
 
@@ -219,5 +219,5 @@ async def admin_delete_domain(
         )
     db.delete_domain(rp_id, ctx=ctx)
     _rebuild_registry()
-    await sync.drop_domain_from_links(rp_id, ctx=ctx)
+    sync.drop_domain_from_links(rp_id, ctx=ctx)
     return {"status": "ok"}

paskia/fastapi/admin/satellites.py

@@ -114,7 +114,7 @@ async def admin_update_link_scope(
     )
     rp_ids = _check_scope(payload.get("rp_ids"))
     db.set_satellite_link_scope(link_id, rp_ids, ctx=ctx)
-    await sync.push_link_domains(link_id)
+    sync.push_link_domains(link_id)
     return {"status": "ok"}
 
 

paskia/fastapi/sync.py

@@ -136,37 +136,44 @@ def domain_infos(rp_ids) -> list[dict]:
     return infos
 
 
-async def push_link_domains(link_id: str) -> None:
-    """Send the link's current domain set to its live connections and
-    update their feed scope (after scope edits or origin changes)."""
+def push_link_domains(link_id: str) -> None:
+    """Queue the link's current domain set to its live connections and
+    update their feed scope (after scope edits or origin changes). Frames
+    flow through the connection's queue: the event pump stays the single
+    sender, so frames never interleave with a concurrent send."""
     link = db.data().satellite_links.get(link_id)
     if link is None:
         return
     message = {"type": "domains", "domains": domain_infos(link.rp_ids)}
-    for conn in list(_connections.values()):
+    for queue, conn in list(_connections.items()):
         if conn.link_id != link_id:
             continue
         conn.scope.clear()
         conn.scope.update(link.rp_ids)
-        with contextlib.suppress(Exception):
-            await _send(conn.ws, message)
+        try:
+            queue.put_nowait(message)
+        except asyncio.QueueFull:
+            # Slow consumer: drop it like syncfeed does — the reconnect's
+            # welcome re-pushes the domain set anyway.
+            _connections.pop(queue, None)
+            syncfeed.unsubscribe(queue)
 
 
-async def refresh_links_serving(rp_id: str) -> None:
+def refresh_links_serving(rp_id: str) -> None:
     """A domain changed upstream: push its updated info to links serving it."""
     for link_id, link in db.data().satellite_links.items():
         if rp_id in link.rp_ids:
-            await push_link_domains(link_id)
+            push_link_domains(link_id)
 
 
-async def drop_domain_from_links(rp_id: str, *, ctx=None) -> None:
+def drop_domain_from_links(rp_id: str, *, ctx=None) -> None:
     """A domain was deleted upstream: shrink every link's scope and push."""
     for link_id, link in db.data().satellite_links.items():
         if rp_id in link.rp_ids:
             db.set_satellite_link_scope(
                 link_id, [r for r in link.rp_ids if r != rp_id], ctx=ctx
             )
-            await push_link_domains(link_id)
+            push_link_domains(link_id)
 
 
 async def _send(ws: WebSocket, message: dict) -> None:
@@ -256,7 +263,10 @@ async def _handshake(ws: WebSocket) -> _Connection | None:
             await ws.close(code=1008)
             return None
         db.create_satellite_link(
-            link_id, rp_ids, server_part + client_part, label=hello.get("label") or ""
+            link_id,
+            rp_ids,
+            server_part + client_part,
+            label=(hello.get("label") or "")[:80],
         )
     return _Connection(
         ws=ws,

paskia/satellite.py

@@ -233,18 +233,22 @@ def _drop_replica_rows(replica: ReplicaDB, removed_rp_ids: set[str]) -> None:
 
 def apply_domains_push(
     uplink_id: str, infos: list[dict], replica_db: ReplicaDB | None = None
-) -> dict[str, str]:
+) -> tuple[dict[str, str], bool]:
     """Apply a domain set pushed by the uplink: persist the accepted set
     (adding, updating and removing the link's domains) and rebuild the
-    registry. Returns the {rp_id: reason} rejections to report upstream."""
+    registry. Returns ({rp_id: rejection reason}, changed) — the rejections
+    to report upstream, and whether the accepted set moved (the caller then
+    resyncs: a newly added domain's existing sessions only arrive with a
+    fresh snapshot)."""
     accepted, rejected = _accepted_domains(uplink_id, infos)
     current = _uplink_domains(uplink_id)
-    if accepted != current:
+    changed = accepted != current
+    if changed:
         db.apply_uplink_domains(uplink_id, accepted)
         domains.init_registry(db.data().config)
     if replica_db is not None and (removed := set(current) - set(accepted)):
         _drop_replica_rows(replica_db, removed)
-    return rejected
+    return rejected, changed
 
 
 async def register_uplink(url: str) -> dict:
@@ -437,7 +441,9 @@ class UplinkReplica:
                 raise ValueError("sync: protocol error (expected welcome)")
             # The upstream re-pushes the link's domain set on every connect
             # (the scope may have changed while offline) — apply and ack.
-            rejected = apply_domains_push(
+            # The snapshot that follows covers any added domains, so no
+            # reconnect is needed here.
+            rejected, _ = apply_domains_push(
                 self.uplink.uplink_id, welcome.get("domains") or [], replica_db=self.db
             )
             await ws.send(syncfeed.encode({"type": "ready", "rejected": rejected}))
@@ -477,8 +483,11 @@ class UplinkReplica:
                             message.get("fields"),
                         )
                     elif mtype == "domains":
-                        # Live scope/origin change: apply and acknowledge.
-                        rejected = apply_domains_push(
+                        # Live scope/origin change: apply and acknowledge. A
+                        # changed domain set means a resync: a newly added
+                        # domain's existing sessions only arrive with a fresh
+                        # snapshot, so reconnect for one.
+                        rejected, changed = apply_domains_push(
                             self.uplink.uplink_id,
                             message.get("domains") or [],
                             replica_db=self.db,
@@ -488,6 +497,8 @@ class UplinkReplica:
                                 {"type": "domains_ack", "rejected": rejected}
                             )
                         )
+                        if changed:
+                            return
                     elif mtype == "ready":
                         if staging is not None:
                             self.db = staging
@@ -578,7 +589,8 @@ class SatelliteManager:
                 replica = self.replicas.pop(uplink_id)
                 await replica.stop()
                 await _drop_client(replica.uplink.url)
-        for uplink_id, uplink in wanted.items():
+        # Snapshot: a revocation task may mutate the uplinks mid-loop
+        for uplink_id, uplink in list(wanted.items()):
             uplink.uplink_id = uplink_id  # fresh objects lack it until load
             replica = self.replicas.get(uplink_id)
             if replica is None or replica.uplink != uplink:

paskia/syncfeed.py

@@ -25,10 +25,10 @@ from uuid import UUID
 import msgspec
 
 # Tables mirrored by satellites, all filtered to the subscriber's rp-id
-# (reset tokens, sync tokens, OIDC data and domain config are
+# scope set (reset tokens, link tables, OIDC data and domain config are
 # instance-local and never replicated).
 TABLES = ("credentials", "sessions")
-SCOPED_TABLES = TABLES
+
 
 # The user graph is never published; upserts of these tables drive the
 # eager re-push of affected session contexts instead.
@@ -102,7 +102,7 @@ def emit(table: str, key: str, obj) -> None:
     for queue, scope in list(_subscribers.items()):
         if (
             scope is not None
-            and table in SCOPED_TABLES
+            and table in TABLES
             and fields is not None
             and fields.get("rp_id") not in scope
         ):

paskia/util/startupbox.py

@@ -125,7 +125,11 @@ def print_startup_config(
         header_urls = [f"{w}{vite_url}{r} (vite dev)"]
     else:
         header_urls = [
-            f"{w}{url}{r}" for d in domains if (url := auth_host_url(d.config))
+            f"{w}{url}{r}"
+            for d in domains
+            # A satellite-served domain's auth host lives on the upstream —
+            # the header links this instance's own account pages.
+            if d.config.remote is None and (url := auth_host_url(d.config))
         ]
         if not header_urls:
             guess = min(
@@ -164,6 +168,8 @@ def print_startup_config(
         rp_name = domain.rp_name
         suffix = f" ({rp_name})" if rp_name and rp_name != domain.rp_id else ""
         head = f"{w}{domain.rp_id}{r}{suffix}"
+        if domain.remote is not None:
+            head += " 🛰"  # satellite-served via an uplink
         if not domain.config.origins:
             rows.append(f"{head} — no sign-in sites")
             continue

tests/test_remote.py

@@ -310,19 +310,22 @@ _PUSHED = {
 async def test_apply_domains_push_accept_and_remove(test_db, domain_registry):
     """A push persists the accepted set (marked remote via the link); a
     later push without the domain removes it."""
-    rejected = satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
+    rejected, changed = satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
     assert not rejected
+    assert changed
     domain = test_db.config.domains["example.com"]
     assert domain.remote.link == UPLINK_ID
     assert domain.rp_name == "Example"
     assert domains.registry().get("example.com").remote is not None
 
     # Idempotent: the same push again changes nothing
-    assert not satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
+    rejected, changed = satellite.apply_domains_push(UPLINK_ID, [_PUSHED])
+    assert not rejected and not changed
 
     # The uplink's scope shrank: the domain leaves the local config
-    rejected = satellite.apply_domains_push(UPLINK_ID, [])
+    rejected, changed = satellite.apply_domains_push(UPLINK_ID, [])
     assert not rejected
+    assert changed
     assert "example.com" not in test_db.config.domains
     assert domains.registry().get("example.com") is None
 
@@ -330,10 +333,11 @@ async def test_apply_domains_push_accept_and_remove(test_db, domain_registry):
 @pytest.mark.asyncio
 async def test_apply_domains_push_rejects_local_conflict(test_db, domain_registry):
     """A pushed rp-id that this instance serves locally is rejected."""
-    rejected = satellite.apply_domains_push(
+    rejected, changed = satellite.apply_domains_push(
         UPLINK_ID, [{**_PUSHED, "rp_id": TEST_RP_ID}]
     )
     assert TEST_RP_ID in rejected
+    assert not changed
     assert test_db.config.domains[TEST_RP_ID].remote is None
 
 
@@ -341,7 +345,7 @@ async def test_apply_domains_push_rejects_local_conflict(test_db, domain_registr
 async def test_apply_domains_push_rejects_auth_host_collision(test_db, domain_registry):
     """A pushed auth host that collides with a local rp-id is rejected
     (the merged config must stay servable)."""
-    rejected = satellite.apply_domains_push(
+    rejected, _ = satellite.apply_domains_push(
         UPLINK_ID,
         [
             {

tests/test_satellite_e2e.py

@@ -542,7 +542,7 @@ async def test_live_scope_shrink_stops_feed(live_remote, test_user, remote_crede
     connection: sessions of removed domains stop flowing."""
     live = live_remote
     ops_db.set_satellite_link_scope(LINK_ID, [])
-    await sync.push_link_domains(LINK_ID)
+    sync.push_link_domains(LINK_ID)
 
     removed_key, _ = create_test_session(
         user_uuid=test_user.uuid,
@@ -553,7 +553,7 @@ async def test_live_scope_shrink_stops_feed(live_remote, test_user, remote_crede
     # Positive control: widen the scope again; a session created now does
     # arrive, so the earlier one demonstrably passed a closed pipe.
     ops_db.set_satellite_link_scope(LINK_ID, [REMOTE_DOMAIN])
-    await sync.push_link_domains(LINK_ID)
+    sync.push_link_domains(LINK_ID)
     remote_key, _ = create_test_session(
         user_uuid=test_user.uuid,
         credential_uuid=remote_credential.uuid,
@@ -582,7 +582,7 @@ async def test_live_domain_conflict_reported_upstream(live_remote, test_db):
             },
         ),
     )
-    await sync.push_link_domains(LINK_ID)
+    sync.push_link_domains(LINK_ID)
     await wait_for(
         "rejection to arrive",
         lambda: (