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).

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:

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: (