LeoVasanko
5ecb10166d
DomainConfig.remote {url, token, cache_ttl, refresh_interval} marks a
domain as backed by a remote paskia instance (auth host required). The
remote publishes committed changes via struct store()/delete() hooks and
explicit emits in field-mutating operations into syncfeed, an in-RAM
sequenced ring buffer served over a token-gated WebSocket
(/auth/api/sync/ws, tokens from PASKIA_SYNC_TOKENS env). The satellite
keeps a plain DB replica per remote URL, applies snapshots/events,
enforces expiry locally, and writes session refreshes back over the same
channel. /validate refreshes locally with write-behind; /logout,
/set-session, /token-info and /auth/oidc/* are proxied to the remote
with the original Host header; logout also evicts from the replica.
Replicas go fail-closed (503) after cache_ttl of silence.
249 lines
8.7 KiB
Python
249 lines
8.7 KiB
Python
"""Satellite side of remote domains: RAM-only read replicas.
|
|
|
|
For each domain configured with ``DomainConfig.remote`` a replica of the
|
|
remote's tables (another plain DB instance, never persisted) is attached to
|
|
the runtime Domain as its store, fed by a sync WebSocket to the remote and
|
|
refreshed by periodic full snapshots. Session refreshes from /validate are
|
|
written back over the same channel.
|
|
|
|
While the sync channel has been silent for longer than the domain's
|
|
cache_ttl the replica is considered unavailable (fail-closed; set a large
|
|
cache_ttl for fail-open behavior bounded by session expiry).
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from datetime import UTC, datetime
|
|
from uuid import UUID
|
|
|
|
import msgspec
|
|
import websockets
|
|
|
|
from paskia import domains
|
|
from paskia.authsession import EXPIRES
|
|
from paskia.db.structs import (
|
|
DB,
|
|
Credential,
|
|
Org,
|
|
Permission,
|
|
RemoteConfig,
|
|
Role,
|
|
Session,
|
|
User,
|
|
)
|
|
from paskia.util.crypto import hash_secret
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_TABLES = {
|
|
"permissions": (Permission, True),
|
|
"orgs": (Org, True),
|
|
"roles": (Role, True),
|
|
"users": (User, True),
|
|
"credentials": (Credential, True),
|
|
"sessions": (Session, False),
|
|
}
|
|
|
|
_RECONNECT_DELAY = 5
|
|
_SWEEP_INTERVAL = 60
|
|
|
|
|
|
def _apply(replica: DB, table: str, key: str, op: str, fields: dict | None) -> None:
|
|
cls, uuid_key = _TABLES[table]
|
|
store = getattr(replica, table)
|
|
store_key = UUID(key) if uuid_key else key
|
|
if op == "delete":
|
|
store.pop(store_key, None)
|
|
return
|
|
obj = msgspec.convert(fields, cls)
|
|
if uuid_key:
|
|
obj.uuid = store_key
|
|
else:
|
|
obj.key = key
|
|
store[store_key] = obj
|
|
|
|
|
|
class RemoteReplica:
|
|
"""One remote instance's replica, its sync client and write-behind queue."""
|
|
|
|
def __init__(self, remote: RemoteConfig):
|
|
self.remote = remote
|
|
self.db = DB()
|
|
self.generation: str | None = None
|
|
self.seq = 0
|
|
self.last_contact = 0.0 # monotonic time of last snapshot/event
|
|
self._pending_refresh: dict[str, dict] = {}
|
|
self._refresh_signal = asyncio.Event()
|
|
self._task: asyncio.Task | None = None
|
|
self._sweeper: asyncio.Task | None = None
|
|
self._stopped = True
|
|
|
|
def available(self) -> bool:
|
|
return (
|
|
self.last_contact > 0
|
|
and time.monotonic() - self.last_contact <= self.remote.cache_ttl
|
|
)
|
|
|
|
def refresh_session(self, key: str, validated, ip: str, user_agent: str) -> None:
|
|
"""Apply a /validate refresh locally and queue it for the remote."""
|
|
session = self.db.sessions.get(key)
|
|
if session is not None:
|
|
session.validated = validated
|
|
session.ip = ip
|
|
session.user_agent = user_agent
|
|
self._pending_refresh[key] = {
|
|
"type": "session_refresh",
|
|
"key": key,
|
|
"validated": msgspec.to_builtins(validated),
|
|
"ip": ip,
|
|
"user_agent": user_agent,
|
|
}
|
|
self._refresh_signal.set()
|
|
|
|
def evict_session(self, secret: str) -> None:
|
|
self.db.sessions.pop(hash_secret("cookie", secret), None)
|
|
|
|
async def start(self) -> None:
|
|
self._stopped = False
|
|
self._task = asyncio.create_task(self._run())
|
|
self._sweeper = asyncio.create_task(self._sweep())
|
|
|
|
async def stop(self) -> None:
|
|
self._stopped = True
|
|
for task in (self._task, self._sweeper):
|
|
if task:
|
|
task.cancel()
|
|
with asyncio.suppress(asyncio.CancelledError):
|
|
await task
|
|
|
|
async def _sweep(self) -> None:
|
|
while True:
|
|
await asyncio.sleep(_SWEEP_INTERVAL)
|
|
limit = datetime.now(UTC) - EXPIRES
|
|
for key in [k for k, s in self.db.sessions.items() if s.validated < limit]:
|
|
del self.db.sessions[key]
|
|
|
|
async def _run(self) -> None:
|
|
while not self._stopped:
|
|
try:
|
|
await self._connect()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
_logger.info("Sync to %s failed: %s", self.remote.url, e)
|
|
if not self._stopped:
|
|
await asyncio.sleep(_RECONNECT_DELAY)
|
|
|
|
async def _connect(self) -> None:
|
|
ws_url = self.remote.url.replace("http", "ws", 1) + "/auth/api/sync/ws"
|
|
# Periodic full snapshots reconcile any drift; resume is cheaper.
|
|
full_resync = self.generation is None or (
|
|
time.monotonic() - self.last_contact > self.remote.refresh_interval
|
|
)
|
|
resume = {} if full_resync else {"generation": self.generation, "seq": self.seq}
|
|
async with websockets.connect(
|
|
ws_url, additional_headers={"Authorization": f"Bearer {self.remote.token}"}
|
|
) as ws:
|
|
hello = msgspec.json.decode(await ws.recv())
|
|
if hello.get("type") != "hello":
|
|
raise ValueError("sync: expected hello")
|
|
await ws.send(msgspec.json.encode({"type": "resume", **resume}))
|
|
sender = asyncio.create_task(self._send_loop(ws))
|
|
staging: DB | None = None
|
|
try:
|
|
while True:
|
|
message = msgspec.json.decode(await ws.recv())
|
|
self.last_contact = time.monotonic()
|
|
mtype = message.get("type")
|
|
if mtype == "snapshot":
|
|
if staging is None:
|
|
staging = DB()
|
|
for key, fields in message["items"]:
|
|
_apply(staging, message["table"], key, "upsert", fields)
|
|
elif mtype == "event":
|
|
if staging is not None or (
|
|
self.generation is not None
|
|
and message["seq"] != self.seq + 1
|
|
):
|
|
raise ValueError("sync: event out of order")
|
|
self.seq = message["seq"]
|
|
_apply(
|
|
self.db,
|
|
message["table"],
|
|
message["key"],
|
|
message["op"],
|
|
message.get("fields"),
|
|
)
|
|
elif mtype == "ready":
|
|
if staging is not None:
|
|
self.db = staging
|
|
staging = None
|
|
attach_stores()
|
|
self.generation = hello["generation"]
|
|
self.seq = message["seq"]
|
|
finally:
|
|
sender.cancel()
|
|
with asyncio.suppress(asyncio.CancelledError):
|
|
await sender
|
|
|
|
async def _send_loop(self, ws) -> None:
|
|
while True:
|
|
self._refresh_signal.clear()
|
|
while self._pending_refresh:
|
|
_, message = self._pending_refresh.popitem()
|
|
await ws.send(msgspec.json.encode(message))
|
|
await self._refresh_signal.wait()
|
|
|
|
|
|
class SatelliteManager:
|
|
"""Replicas keyed by remote URL; domains sharing a remote share one."""
|
|
|
|
def __init__(self):
|
|
self.replicas: dict[str, RemoteReplica] = {}
|
|
|
|
def replica_for(self, domain: domains.Domain) -> RemoteReplica | None:
|
|
if domain.remote is None:
|
|
return None
|
|
return self.replicas.get(domain.remote.url)
|
|
|
|
async def start(self) -> None:
|
|
domains.add_rebuild_listener(self.reconcile)
|
|
await self.reconcile(domains.registry())
|
|
|
|
async def stop(self) -> None:
|
|
domains.remove_rebuild_listener(self.reconcile)
|
|
for replica in self.replicas.values():
|
|
await replica.stop()
|
|
self.replicas.clear()
|
|
|
|
async def reconcile(self, registry: domains.DomainRegistry) -> None:
|
|
"""Attach stores and start/stop replicas to match the config."""
|
|
wanted = {}
|
|
for domain in registry.domains:
|
|
if domain.remote is not None:
|
|
wanted.setdefault(domain.remote.url, domain.remote)
|
|
for url in list(self.replicas):
|
|
if url not in wanted:
|
|
await self.replicas.pop(url).stop()
|
|
for url, remote in wanted.items():
|
|
replica = self.replicas.get(url)
|
|
if replica is None or replica.remote != remote:
|
|
if replica is not None:
|
|
await replica.stop()
|
|
replica = RemoteReplica(remote)
|
|
self.replicas[url] = replica
|
|
await replica.start()
|
|
attach_stores()
|
|
|
|
|
|
manager = SatelliteManager()
|
|
|
|
|
|
def attach_stores() -> None:
|
|
"""Attach each remote domain's store to its replica."""
|
|
for domain in domains.registry().domains:
|
|
replica = manager.replica_for(domain)
|
|
if replica is not None:
|
|
domain.store = replica.db
|