From 245672d3f2681b8181b570df1ac6fac1ad23add0 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sun, 19 Jul 2026 06:49:48 +0200 Subject: [PATCH] =?UTF-8?q?sync-v2=20step=204:=20event-driven=20account=20?= =?UTF-8?q?lifecycle=20(=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Coordinator subscribes to account.state_changed over Redis pub/sub: going live marks the account claimable at once; disabled/deleted/error tears the owning runner down immediately (≤30s) instead of waiting for the lease TTL. - delete_account now publishes account.state_changed:deleted after dropping the account from sync:live, so teardown is event-driven, not just fencing. - AccountRunner gains a verify loop (defense in depth): it periodically confirms the account is still live and flags itself dead if not; the coordinator reaps dead runners and releases their lease. Zombie loops are structurally impossible. - New core event user.active (SPEC §8.3), published on login and on every WS connect; consumed for tier promotion in step 5. - Acceptance: added event-driven teardown test (isolated from fencing/verify). Co-Authored-By: Claude Opus 4.8 --- backend/app/accounts/service.py | 7 ++++ backend/app/events.py | 3 ++ backend/app/identity/routes.py | 3 ++ backend/app/sync/coordinator.py | 47 +++++++++++++++++++++++++++ backend/app/sync/runner.py | 32 ++++++++++++++++++ backend/app/ws.py | 3 ++ backend/tests/sync/test_acceptance.py | 32 ++++++++++++++++++ 7 files changed, 127 insertions(+) diff --git a/backend/app/accounts/service.py b/backend/app/accounts/service.py index 4818195..ae7d50b 100644 --- a/backend/app/accounts/service.py +++ b/backend/app/accounts/service.py @@ -82,10 +82,17 @@ async def delete_account( # Deregister from sync:live and drop any lease so the owning sync worker # stops the account's runner on its next renewal (refactor §2 fencing). + from app.events import get_event_bus from app.redis_client import get_redis from app.sync import leases await leases.mark_not_live(get_redis(), account_id) + # Sync v2 §3: signal the owning worker to tear the runner down now (≤30s), + # rather than waiting for the lease to lapse. + await get_event_bus().publish( + "account.state_changed", + {"user_id": str(user_id), "account_id": str(account_id), "state": "deleted"}, + ) # Content-addressed blobs are shared across the user's accounts (dedup): only # remove a blob when no remaining message/attachment references it. diff --git a/backend/app/events.py b/backend/app/events.py index f080534..e1ea717 100644 --- a/backend/app/events.py +++ b/backend/app/events.py @@ -38,6 +38,9 @@ ALLOWED_EVENTS: frozenset[str] = frozenset( "account.state_changed", "draft.saved", "outbox.state_changed", + # Sync v2 §3: published on login and every WS connect; drives tier + # promotion in the sync workers (§5). Payload: {user_id}. + "user.active", } ) diff --git a/backend/app/identity/routes.py b/backend/app/identity/routes.py index 18dc8ef..353cf54 100644 --- a/backend/app/identity/routes.py +++ b/backend/app/identity/routes.py @@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.errors import raise_api_error from app.config import get_settings from app.db import get_session +from app.events import get_event_bus from app.identity import totp as totp_mod from app.identity.cookies import ( clear_auth_cookies, @@ -116,6 +117,8 @@ async def login(body: LoginRequest, request: Request, response: Response, db: Db if user is None: raise_api_error(401, "invalid_credentials", "incorrect handle or password") await _establish_session(request, response, user) + # Sync v2 §3: a login marks the user active, promoting their accounts' tier. + await get_event_bus().publish("user.active", {"user_id": str(user.id)}) return _user_out(user) diff --git a/backend/app/sync/coordinator.py b/backend/app/sync/coordinator.py index cf86f87..11bcc0f 100644 --- a/backend/app/sync/coordinator.py +++ b/backend/app/sync/coordinator.py @@ -16,6 +16,7 @@ account because claims are ``SET NX`` and renew/release are owner-checked. from __future__ import annotations import asyncio +import contextlib import json import logging import os @@ -29,6 +30,7 @@ from sqlalchemy import select from app.config import get_settings from app.db import session_scope +from app.events import channel_for from app.models import Account from app.redis_client import aw, get_redis from app.sync import leases @@ -78,6 +80,7 @@ class SyncCoordinator: asyncio.create_task(self._loop(self._reconcile_once, self.reconcile_interval)), asyncio.create_task(self._loop(self._heartbeat_once, self.heartbeat_refresh)), asyncio.create_task(self._loop(self._rebalance_once, self.rebalance_interval)), + asyncio.create_task(self._event_subscriber()), ] logger.info("sync coordinator %s started", self.worker_id) @@ -132,6 +135,14 @@ class SyncCoordinator: async def _renew_once(self) -> None: for account_id in list(self._runners): + runner = self._runners.get(account_id) + if runner is None: + continue + # Reap a runner that self-verified it is no longer live (§3). + if runner.dead.is_set(): + logger.info("reaping dead runner for account %s", account_id) + await self._release(account_id) + continue if not await self._scripts.renew(account_id, self.worker_id, self.lease_ttl): logger.warning("lost lease for account %s; fencing runner", account_id) await self._drop(account_id) @@ -203,3 +214,39 @@ class SyncCoordinator: for account_id in list(self._runners)[-excess:]: await self._release(account_id) logger.info("worker %s shed %d account(s) to rebalance", self.worker_id, excess) + + # --- event-driven lifecycle (§3) ----------------------------------------- + + async def _event_subscriber(self) -> None: + """React to ``account.state_changed`` across processes (Redis pub/sub). + + Going live makes an account claimable at once; going away tears the + runner down immediately (≤30s from the API call) instead of waiting for + the lease to lapse. + """ + pubsub = self._redis.pubsub() + await pubsub.subscribe(channel_for("account.state_changed")) + try: + async for message in pubsub.listen(): + if self._stop.is_set(): + break + if message.get("type") != "message": + continue + try: + payload = json.loads(message["data"]).get("payload", {}) + except (ValueError, TypeError): + continue + await self._on_state_changed(payload) + finally: + with contextlib.suppress(Exception): + await pubsub.aclose() # type: ignore[no-untyped-call] + + async def _on_state_changed(self, payload: dict[str, object]) -> None: + raw = payload.get("account_id") + if not isinstance(raw, str): + return + account_id = uuid.UUID(raw) + if payload.get("state") == "live": + await leases.mark_live(self._redis, account_id) # claimable next cycle + elif account_id in self._runners: + await self._release(account_id) # disabled/deleted/error -> stop now diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 220fb26..c06977b 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -14,11 +14,16 @@ import logging import uuid from collections.abc import Awaitable, Callable +from sqlalchemy import select + +from app.db import session_scope +from app.models import Account from app.sync.engine import live_loop, secondary_poll_loop logger = logging.getLogger(__name__) _RECONNECT_BACKOFF = 30.0 +_VERIFY_INTERVAL = 60.0 class AccountRunner: @@ -28,6 +33,10 @@ class AccountRunner: self.account_id = account_id self._stop = asyncio.Event() self._tasks: list[asyncio.Task[None]] = [] + # Set when the account is found no longer live (§3 defense in depth); the + # coordinator reaps such runners and releases their lease. + self.dead = asyncio.Event() + self.verify_interval = _VERIFY_INTERVAL def start(self) -> None: """Launch the account's sync loops (idempotent-ish: call once).""" @@ -35,6 +44,7 @@ class AccountRunner: self._tasks = [ asyncio.create_task(self._supervise(self._run_live, "live")), asyncio.create_task(self._supervise(self._run_secondary, "secondary")), + asyncio.create_task(self._verify_loop()), ] def running(self) -> bool: @@ -55,6 +65,28 @@ class AccountRunner: async def _run_secondary(self) -> None: await secondary_poll_loop(self.account_id, self._stop) + async def _verify_loop(self) -> None: + """Periodically confirm the account is still live; flag ``dead`` if not. + + This makes zombie loops structurally impossible even if a lifecycle event + is lost: an account that was disabled/deleted stops being synced within a + verify cycle regardless of the lease/event paths. + """ + while not self._stop.is_set(): + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(self._stop.wait(), timeout=self.verify_interval) + if self._stop.is_set(): + return + if not await self._account_is_live(): + logger.info("account %s no longer live; flagging runner dead", self.account_id) + self.dead.set() + return + + async def _account_is_live(self) -> bool: + async with session_scope() as session: + state = await session.scalar(select(Account.state).where(Account.id == self.account_id)) + return state == "live" + async def _supervise(self, run: Callable[[], Awaitable[None]], label: str) -> None: """Run a loop, restarting it (with backoff) until stopped.""" while not self._stop.is_set(): diff --git a/backend/app/ws.py b/backend/app/ws.py index 4731835..dc58749 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -13,6 +13,7 @@ import logging from fastapi import APIRouter, WebSocket, WebSocketDisconnect +from app.events import get_event_bus from app.identity.cookies import SESSION_COOKIE_INSECURE, SESSION_COOKIE_SECURE from app.identity.sessions import resolve_session from app.redis_client import get_redis @@ -34,6 +35,8 @@ async def ws_endpoint(websocket: WebSocket) -> None: await websocket.accept() user_id = str(session.user_id) + # Sync v2 §3: an open WS means the user is active -> promote their tier. + await get_event_bus().publish("user.active", {"user_id": user_id}) pubsub = get_redis().pubsub() await pubsub.psubscribe("evt:*") diff --git a/backend/tests/sync/test_acceptance.py b/backend/tests/sync/test_acceptance.py index bf12a46..cd02c8b 100644 --- a/backend/tests/sync/test_acceptance.py +++ b/backend/tests/sync/test_acceptance.py @@ -26,6 +26,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.config import get_settings +from app.events import get_event_bus from app.identity.passwords import hash_password from app.models import Account, User from app.redis_client import get_redis @@ -222,5 +223,36 @@ async def test_acceptance_3_create_delete(connections, make_accounts) -> None: await c.stop() +async def test_event_driven_teardown(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + """§3: an account.state_changed event tears the runner down at once. + + Leases are kept long (no fencing) and the verify loop slow, so the only path + that can stop the runner in time is the cross-process event subscriber. + """ + (aid,) = await make_accounts(1) + user_id = uuid.uuid4() + await _seed({aid}) + c = _fast(SyncCoordinator("w1")) + c.lease_ttl = 60 # lease stays valid: fencing cannot explain the teardown + c.lease_renew = 30 + await c.start() + try: + await _until(lambda: aid in c.owned) + assert connections[aid] == 1 + + # As delete_account does: drop it from sync:live (so it isn't re-claimed), + # then emit the event. The lease is left intact, so only the event — not + # fencing — can explain the teardown. + await get_redis().srem(leases.LIVE_SET, str(aid)) + await get_event_bus().publish( + "account.state_changed", + {"user_id": str(user_id), "account_id": str(aid), "state": "deleted"}, + ) + await _until(lambda: aid not in c.owned, timeout_s=3.0) + await _until(lambda: connections[aid] == 0, timeout_s=3.0) + finally: + await c.stop() + + def test_worker_id_is_unique() -> None: assert coordinator_mod._new_worker_id() != coordinator_mod._new_worker_id()