Host-keyed dispatch in the satellite module; snapshot-only sync

Callers never see stores: session_ctx/verify/user-info resolve the store
from the request host via satellite.store_for_host; session refresh and
logout eviction are dispatch functions too (satellite.refresh_session /
evict_session). API handlers keep one code path plus forward_request
one-liners; proxy.py folds into satellite.py; Domain.store and the
store parameters are gone; 503 comes from the dispatch point as a plain
HTTPException.

The sync protocol drops replay/generation/seq: snapshots are small, so
every connect starts from a full snapshot and a single ordered WebSocket
cannot gap; a slow subscriber is dropped and resyncs. The satellite
reconnects every refresh_interval to reconcile drift.

tests/test_remote.py

@@ -1,6 +1,5 @@
 """Tests for remote (satellite) domains: config, replica application, feed."""
 
-import collections
 import secrets
 import time
 from datetime import UTC, datetime
@@ -79,9 +78,9 @@ def test_apply_upsert_and_delete():
     replica = DB()
     user = User.create(display_name="U", role=UUID(int=1))
     user.uuid = UUID(int=2)
-    satellite._apply(replica, "users", str(user.uuid), "upsert", _builtins(user))
+    satellite._apply(replica, "users", str(user.uuid), _builtins(user))
     assert replica.users[user.uuid].display_name == "U"
-    satellite._apply(replica, "users", str(user.uuid), "delete", None)
+    satellite._apply(replica, "users", str(user.uuid), None)
     assert not replica.users
 
 
@@ -102,7 +101,7 @@ def test_apply_session_roundtrip():
         validated=datetime.now(UTC),
         rp_id="example.com",
     )
-    satellite._apply(replica, "sessions", session.key, "upsert", _builtins(session))
+    satellite._apply(replica, "sessions", session.key, _builtins(session))
     stored = replica.sessions[session.key]
     assert stored.host == "app2.example.com"
     assert stored.validated == session.validated
@@ -123,47 +122,50 @@ def test_apply_credential_bytes_roundtrip():
     cred.uuid = UUID(int=9)
     # Simulate the full wire path: builtins -> JSON -> builtins
     wire = msgspec.json.decode(msgspec.json.encode(_builtins(cred)))
-    satellite._apply(replica, "credentials", str(cred.uuid), "upsert", wire)
+    satellite._apply(replica, "credentials", str(cred.uuid), wire)
     stored = replica.credentials[cred.uuid]
     assert stored.credential_id == cred.credential_id
     assert stored.public_key == cred.public_key
     assert stored.sign_count == 3
 
 
-def test_feed_emit_and_replay():
-    feed = syncfeed.SyncFeed()
-    user = User.create(display_name="A", role=UUID(int=1))
-    feed.emit("users", "k1", user)
-    feed.emit("users", "k1", None)
-    assert feed.seq == 2
-    assert feed.replay_since(0)[0]["op"] == "upsert"
-    assert feed.replay_since(1)[0]["op"] == "delete"
-    assert feed.replay_since(2) == []
-    assert feed.replay_since(99) is None
+def test_feed_emit_to_subscribers():
+    queue = syncfeed.subscribe()
+    try:
+        user = User.create(display_name="A", role=UUID(int=1))
+        syncfeed.emit("users", "k1", user)
+        syncfeed.emit("users", "k1", None)
+        assert queue.get_nowait()["fields"]["display_name"] == "A"
+        assert queue.get_nowait()["fields"] is None
+    finally:
+        syncfeed.unsubscribe(queue)
 
 
-def test_feed_ring_overflow_replay_none():
-    feed = syncfeed.SyncFeed()
-    feed.events = collections.deque(maxlen=3)
-    for i in range(5):
-        feed.emit("users", f"k{i}", None)
-    assert feed.replay_since(0) is None  # fell off the ring
-    assert [e["seq"] for e in feed.replay_since(4)] == [5]
-    assert feed.replay_since(5) == []
+def test_feed_drops_full_queue():
+    queue = syncfeed.subscribe()
+    try:
+        for i in range(1001):
+            syncfeed.emit("users", f"k{i}", None)
+        assert queue.qsize() == 1000
+        syncfeed.emit("users", "k1001", None)  # subscriber already dropped
+        assert queue.qsize() == 1000
+    finally:
+        syncfeed.unsubscribe(queue)
 
 
 @pytest.mark.asyncio
 async def test_operations_emit_events(test_db):
     """Writes through db.operations land on the sync feed."""
-    syncfeed.feed.events.clear()
-    syncfeed.feed.seq = 0
-    user = next(iter(test_db.users.values()))
-    ops_db.update_user_display_name(user.uuid, "Renamed")
-    tables = {e["table"] for e in syncfeed.feed.events}
-    assert "users" in tables
-    key = syncfeed.feed.events[-1]["key"]
-    assert syncfeed.feed.events[-1]["fields"]["display_name"] == "Renamed"
-    assert key == str(user.uuid)
+    queue = syncfeed.subscribe()
+    try:
+        user = next(iter(test_db.users.values()))
+        ops_db.update_user_display_name(user.uuid, "Renamed")
+        event = queue.get_nowait()
+        assert event["table"] == "users"
+        assert event["key"] == str(user.uuid)
+        assert event["fields"]["display_name"] == "Renamed"
+    finally:
+        syncfeed.unsubscribe(queue)
 
 
 @pytest.mark.asyncio
@@ -188,8 +190,13 @@ async def test_replica_refresh_and_evict():
     assert queued["type"] == "session_refresh"
     assert queued["ip"] == "2.2.2.2"
 
-    replica.evict_session(token)
+    # Host-keyed dispatch eviction (the replica's domain is resolved by host)
+    domains.configure(listen=["localhost:4401"])
+    domains.init_registry(_remote_domain_config())
+    satellite.manager.replicas[REMOTE_URL] = replica
+    satellite.evict_session(token, "app2.example.com")
     assert not replica.db.sessions
+    satellite.manager.replicas.pop(REMOTE_URL)
 
 
 def test_availability_gate():
@@ -259,7 +266,6 @@ async def remote_client(test_db):
     replica.last_contact = time.monotonic()
     replica.connected = True
     satellite.manager.replicas[REMOTE_URL] = replica
-    satellite.attach_stores()
     transport = httpx.ASGITransport(app=app)
     async with httpx.AsyncClient(
         transport=transport, base_url="http://localhost:4401"
@@ -322,10 +328,10 @@ async def test_remote_domain_503_when_replica_stale(remote_client):
 async def test_logout_proxied_and_evicted(remote_client, monkeypatch):
     client, secret, replica = remote_client
 
-    async def fake_proxy(request, remote):
+    async def fake_forward(request):
         return Response(status_code=200, content=b'{"message": "Logged out"}')
 
-    monkeypatch.setattr("paskia.fastapi.proxy.proxy_to_remote", fake_proxy)
+    monkeypatch.setattr(satellite, "forward_request", fake_forward)
     r = await client.post(
         "/auth/api/logout",
         headers={"Host": "app2.example.com", "Cookie": f"{AUTH_COOKIE_NAME}={secret}"},