From 6634f285aeb3e27ad475cb9c3d1ac55065eb58b6 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 1 Oct 2026 07:04:12 +0000 Subject: [PATCH] Route live domains pushes through the connection queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 ++-- paskia/fastapi/admin/satellites.py | 2 +- paskia/fastapi/sync.py | 27 +++++++++++++++++---------- tests/test_satellite_e2e.py | 6 +++--- 4 files changed, 23 insertions(+), 16 deletions(-) diff --git a/paskia/fastapi/admin/domains.py b/paskia/fastapi/admin/domains.py index 111d588..036507a 100644 --- a/paskia/fastapi/admin/domains.py +++ b/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"} diff --git a/paskia/fastapi/admin/satellites.py b/paskia/fastapi/admin/satellites.py index 417087a..d69084c 100644 --- a/paskia/fastapi/admin/satellites.py +++ b/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"} diff --git a/paskia/fastapi/sync.py b/paskia/fastapi/sync.py index 7fc6301..b0b2c18 100644 --- a/paskia/fastapi/sync.py +++ b/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: diff --git a/tests/test_satellite_e2e.py b/tests/test_satellite_e2e.py index db75f8c..94c596d 100644 --- a/tests/test_satellite_e2e.py +++ b/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: (