From f229f91632e4aaeb2ed867415f44e43ef7214476 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 23 Jul 2026 10:52:07 +0200 Subject: [PATCH] Add server-push (SSE) notifications for round/bet/balance state changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Frontend dashboards previously found out about state changes only on their next poll tick (up to 15s, or 3s during a draw) — this adds a push channel so updates land as soon as they happen instead. - app/rounds/events.py: a small in-process pub/sub (RoundEventBroadcaster). The message carries no payload — it's just a "something changed, go refetch" signal, so it needs no auth and no knowledge of who's allowed to see what; personalization stays entirely in the existing REST endpoints. - GET /rounds/stream: an SSE endpoint exposing that channel, with keep-alive comments so it survives idle periods behind a reverse proxy, and a defensive MAX_SUBSCRIBERS cap (well above the ~100 concurrent users expected) — past it, the endpoint returns 503 instead of opening a stream, and callers just keep working off polling. - publish() calls added at every point that actually changes what a dashboard would want to know: new round opened, round status transitions (closing/drawing/paying_out/closed), a bet or withdrawal broadcast, any pending tx confirming (bet/withdrawal/payout), a deposit credited, and a new block tip arriving (the exact moment the "drawing" phase is waiting on). - Caddyfile: excludes /rounds/stream from gzip encoding, since compression would buffer output and defeat the point of a live stream. Single-process only by design for now (no cross-worker fan-out) and the notification is a generic broadcast rather than a per-user channel — both are deliberate scope decisions for the current ~100-user, single-container deployment, not oversights. Polling is left fully in place as a fallback; this is purely additive. Co-Authored-By: Claude Sonnet 5 --- Caddyfile | 8 ++++- app/api/routes/rounds.py | 62 +++++++++++++++++++++++++++++++++ app/bets/service.py | 2 ++ app/deposits/service.py | 2 ++ app/electrum/listener.py | 5 +++ app/rounds/events.py | 50 ++++++++++++++++++++++++++ app/rounds/scheduler.py | 5 +++ app/rounds/service.py | 6 ++++ app/tx/confirmation.py | 2 ++ app/withdrawals/service.py | 2 ++ tests/unit/test_round_events.py | 62 +++++++++++++++++++++++++++++++++ 11 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 app/rounds/events.py create mode 100644 tests/unit/test_round_events.py diff --git a/Caddyfile b/Caddyfile index 2274cef..b51229c 100644 --- a/Caddyfile +++ b/Caddyfile @@ -6,6 +6,12 @@ # instead, via its internal CA. Browsers will still warn on first visit # unless that CA is explicitly trusted — expected for local/dev use. {$SITE_ADDRESS:localhost} { - encode gzip + # gzip buffers output, which would delay delivery on the SSE stream + # (/rounds/stream, app/api/routes/rounds.py) — it needs each event flushed + # to the client immediately, not batched. Everything else still compresses. + @not_sse { + not path /rounds/stream + } + encode @not_sse gzip reverse_proxy app:8123 } diff --git a/app/api/routes/rounds.py b/app/api/routes/rounds.py index acddd07..f48d591 100644 --- a/app/api/routes/rounds.py +++ b/app/api/routes/rounds.py @@ -1,6 +1,9 @@ +import asyncio +import json from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, Request +from fastapi.responses import JSONResponse, Response, StreamingResponse from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession @@ -9,10 +12,69 @@ from app.auth.dependencies import get_optional_user from app.db.models import RoundParticipant, User from app.db.session import get_session from app.rounds.config import get_round_config +from app.rounds.events import RoundEventCapacityError, broadcaster from app.rounds.service import get_active_round router = APIRouter(prefix="/rounds", tags=["rounds"]) +# How often request.is_disconnected() gets (re-)checked while idle — bounds +# how long a subscriber slot lingers after a client goes away without a clean +# TCP close (e.g. the network just vanishes). Kept short since the check +# itself is cheap; it does NOT control how often anything is sent on the wire. +_SSE_DISCONNECT_CHECK_SECONDS = 5 + +# How often to send an SSE keep-alive comment on an otherwise-idle connection — +# well under any reasonable reverse-proxy/load-balancer idle-connection timeout +# (Caddy's default is 5 minutes) so the stream isn't silently dropped. Expressed +# as a multiple of the disconnect-check interval above. +_SSE_KEEPALIVE_TICKS = 4 # 4 * 5s = 20s between keep-alive comments + + +@router.get("/stream") +async def round_stream(request: Request) -> Response: + """Server-Sent Events channel: pushes a content-free "update" notification + the instant round/bet/balance state changes anywhere (see app/rounds/events.py + for the publish() call sites), instead of clients only finding out on their + next poll. No payload and no auth: it's a public "go refetch" signal, and + the actual data still comes from the normal per-user REST endpoints, which + is where authorization and personalization (e.g. user_played) already live. + + Frontend polling (app/static/index.html) is left in place as a fallback — + this is purely additive, so a dropped/blocked SSE connection degrades to + the pre-existing polling behavior rather than losing updates outright. + That's also what happens past MAX_SUBSCRIBERS (app/rounds/events.py): this + returns 503 rather than opening a stream, and the browser's EventSource + just retries later while the frontend keeps working off polling meanwhile. + """ + try: + queue = broadcaster.subscribe() + except RoundEventCapacityError: + return JSONResponse(status_code=503, content={"detail": "too many concurrent update streams"}) + + async def event_generator(): + ticks_since_keepalive = 0 + try: + while True: + if await request.is_disconnected(): + break + try: + await asyncio.wait_for(queue.get(), timeout=_SSE_DISCONNECT_CHECK_SECONDS) + yield "event: update\ndata: {}\n\n".format(json.dumps({})) + ticks_since_keepalive = 0 + except asyncio.TimeoutError: + ticks_since_keepalive += 1 + if ticks_since_keepalive >= _SSE_KEEPALIVE_TICKS: + yield ": keep-alive\n\n" + ticks_since_keepalive = 0 + finally: + broadcaster.unsubscribe(queue) + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}, + ) + class CurrentRoundResponse(BaseModel): server_time: str diff --git a/app/bets/service.py b/app/bets/service.py index 5f0301e..50a31a5 100644 --- a/app/bets/service.py +++ b/app/bets/service.py @@ -8,6 +8,7 @@ from app.audit.log import write_audit_log from app.db.models import PendingTransaction, RoundParticipant, User, UtxoEvent from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config +from app.rounds.events import broadcaster from app.rounds.service import open_new_round_if_needed, round_accepts_bets from app.wallet.balance import recompute_balance from app.wallet.hd import derive_pool_address, derive_user_key @@ -91,6 +92,7 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) - await session.commit() await session.refresh(participant) + broadcaster.publish() # participant_count/jackpot changed — nudge every dashboard to refetch return participant diff --git a/app/deposits/service.py b/app/deposits/service.py index 142b74f..d515d91 100644 --- a/app/deposits/service.py +++ b/app/deposits/service.py @@ -3,6 +3,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.audit.log import write_audit_log from app.db.models import UtxoEvent +from app.rounds.events import broadcaster from app.wallet.balance import recompute_balance @@ -50,5 +51,6 @@ async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: l await session.flush() await recompute_balance(session, user_id) await session.commit() + broadcaster.publish() # nudges this user's dashboard to refetch its balance instantly return newly_credited diff --git a/app/electrum/listener.py b/app/electrum/listener.py index 48a7d7a..32082db 100644 --- a/app/electrum/listener.py +++ b/app/electrum/listener.py @@ -9,6 +9,7 @@ from app.db.models import User from app.deposits.service import credit_confirmed_utxos from app.electrum.client import ElectrumClient from app.electrum.scripthash import address_to_scripthash +from app.rounds.events import broadcaster logger = logging.getLogger(__name__) @@ -94,6 +95,10 @@ class ElectrumListener: for header in params: self.tip_height = header["height"] self.tip_header_hex = header.get("hex") + # A new block is exactly what the "drawing" phase is waiting on + # (rounds/scheduler.py:_wait_for_next_block) — nudge dashboards to + # refetch instead of waiting for their next poll. + broadcaster.publish() async def _consume_scripthash(self, queue: asyncio.Queue) -> None: while True: diff --git a/app/rounds/events.py b/app/rounds/events.py new file mode 100644 index 0000000..d648e4a --- /dev/null +++ b/app/rounds/events.py @@ -0,0 +1,50 @@ +import asyncio + +# Defensive cap on concurrent SSE subscribers. Expected load is on the order of +# ~100 concurrent users; this is set well above that so it never engages under +# normal use — it exists purely so a runaway/DoS-y number of open connections +# degrades (new connections fall back to polling, see round_stream()) instead +# of growing the in-memory subscriber set without bound. Revisit this number if +# expected concurrency grows well past it. +MAX_SUBSCRIBERS = 500 + + +class RoundEventCapacityError(Exception): + """Raised by subscribe() when MAX_SUBSCRIBERS is already reached.""" + + +class RoundEventBroadcaster: + """In-process pub/sub so SSE clients (GET /rounds/stream) get pushed a + notification the instant round/bet/balance state changes, instead of only + finding out on their next poll. The message carries no payload — it's just + a "something changed, go refetch" signal; the client re-hits the existing + per-user REST endpoints (/rounds/current, /users/me, ...) for the actual + data, so this never needs to know what changed or who's allowed to see it. + + Single-process only (no cross-worker fan-out) — fine for this deployment + (one uvicorn process, see docker-compose.yml). A multi-worker deployment + would need a shared channel (e.g. Redis pub/sub) instead. + """ + + def __init__(self, max_subscribers: int = MAX_SUBSCRIBERS): + self._subscribers: set[asyncio.Queue] = set() + self.max_subscribers = max_subscribers + + def subscribe(self) -> asyncio.Queue: + if len(self._subscribers) >= self.max_subscribers: + raise RoundEventCapacityError(f"already at the {self.max_subscribers}-subscriber cap") + queue: asyncio.Queue = asyncio.Queue(maxsize=1) + self._subscribers.add(queue) + return queue + + def unsubscribe(self, queue: asyncio.Queue) -> None: + self._subscribers.discard(queue) + + def publish(self) -> None: + for queue in self._subscribers: + if queue.full(): + continue # a not-yet-delivered notification already covers this one + queue.put_nowait(None) + + +broadcaster = RoundEventBroadcaster() diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py index b6bf4eb..1993f7d 100644 --- a/app/rounds/scheduler.py +++ b/app/rounds/scheduler.py @@ -12,6 +12,7 @@ from app.electrum.listener import ElectrumListener from app.electrum.scripthash import address_to_scripthash from app.rounds.config import get_round_config from app.rounds.draw import draw_winner, header_hex_to_block_hash +from app.rounds.events import broadcaster from app.rounds.service import open_new_round_if_needed from app.wallet.hd import derive_pool_key from app.wallet.plm_network import PLM_MAINNET @@ -71,6 +72,7 @@ class RoundScheduler: round_.status = "closing" round_.closed_at = datetime.now(timezone.utc) await session.commit() + broadcaster.publish() async with self._session_factory() as session: pending_count = await session.scalar( @@ -99,6 +101,7 @@ class RoundScheduler: round_.status = "closed" await write_audit_log(session, "round_closed", {"participants": 0}, round_id=round_id) await session.commit() + broadcaster.publish() logger.info("round %s closed with no participants", round_id) return @@ -112,6 +115,7 @@ class RoundScheduler: round_.status = "drawing" await session.commit() + broadcaster.publish() tip_at_close = self._listener.tip_height block_height, block_hash = await self._wait_for_next_block(tip_at_close) @@ -139,6 +143,7 @@ class RoundScheduler: round_id=round_id, ) await session.commit() + broadcaster.publish() logger.info("round %s: winner=%s pool=%s", round_id, winner_address, pool_amount) await self._trigger_payout(round_id) diff --git a/app/rounds/service.py b/app/rounds/service.py index 5454a25..c6c25ef 100644 --- a/app/rounds/service.py +++ b/app/rounds/service.py @@ -5,6 +5,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Round from app.rounds.config import get_round_config +from app.rounds.events import broadcaster _ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out") @@ -57,4 +58,9 @@ async def open_new_round_if_needed(session: AsyncSession) -> Round | None: round_ = Round(status="open") session.add(round_) await session.flush() + # Published pre-commit (the caller commits right after) — acceptable: this + # only tells subscribers "go refetch", and by the time an SSE client's + # refetch request actually lands, this in-process commit (microseconds + # away) has essentially always already happened. + broadcaster.publish() return round_ diff --git a/app/tx/confirmation.py b/app/tx/confirmation.py index edba732..61c4c69 100644 --- a/app/tx/confirmation.py +++ b/app/tx/confirmation.py @@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.db.models import PendingTransaction from app.electrum.client import ElectrumClient +from app.rounds.events import broadcaster logger = logging.getLogger(__name__) @@ -44,6 +45,7 @@ async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) if handler is not None: await handler(session, row) await session.commit() + broadcaster.publish() # a bet/withdrawal/payout just confirmed — balance and/or round state changed confirmed += 1 return confirmed diff --git a/app/withdrawals/service.py b/app/withdrawals/service.py index 561b2c0..578ba53 100644 --- a/app/withdrawals/service.py +++ b/app/withdrawals/service.py @@ -6,6 +6,7 @@ from app.audit.log import write_audit_log from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal from app.electrum.client import ElectrumClient from app.rounds.config import get_round_config +from app.rounds.events import broadcaster from app.wallet.balance import recompute_balance from app.wallet.hd import derive_user_key from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_signed_transaction @@ -84,4 +85,5 @@ async def request_withdrawal( await session.commit() await session.refresh(withdrawal) + broadcaster.publish() # balance just went "pending" — nudge the dashboard to refetch return withdrawal diff --git a/tests/unit/test_round_events.py b/tests/unit/test_round_events.py new file mode 100644 index 0000000..6425d06 --- /dev/null +++ b/tests/unit/test_round_events.py @@ -0,0 +1,62 @@ +import asyncio + +import pytest + +from app.rounds.events import RoundEventBroadcaster, RoundEventCapacityError + + +async def test_publish_wakes_up_subscriber(): + broadcaster = RoundEventBroadcaster() + queue = broadcaster.subscribe() + + broadcaster.publish() + + await asyncio.wait_for(queue.get(), timeout=1) + + +async def test_publish_with_no_subscribers_does_not_raise(): + broadcaster = RoundEventBroadcaster() + broadcaster.publish() # no subscribers yet — must be a no-op, not an error + + +async def test_publish_coalesces_when_subscriber_has_not_drained(): + """The queue is maxsize=1: a second publish() before the subscriber reads + the first notification must not block or raise — it's fine to drop it, + since the subscriber will refetch full state anyway on the first one.""" + broadcaster = RoundEventBroadcaster() + queue = broadcaster.subscribe() + + broadcaster.publish() + broadcaster.publish() # would raise QueueFull if not guarded + + assert queue.qsize() == 1 + + +async def test_unsubscribe_stops_delivery(): + broadcaster = RoundEventBroadcaster() + queue = broadcaster.subscribe() + broadcaster.unsubscribe(queue) + + broadcaster.publish() + + assert queue.empty() + + +async def test_subscribe_rejects_past_the_cap(): + broadcaster = RoundEventBroadcaster(max_subscribers=3) + for _ in range(3): + broadcaster.subscribe() + + with pytest.raises(RoundEventCapacityError): + broadcaster.subscribe() + + +async def test_unsubscribe_frees_a_capacity_slot(): + broadcaster = RoundEventBroadcaster(max_subscribers=1) + queue = broadcaster.subscribe() + + with pytest.raises(RoundEventCapacityError): + broadcaster.subscribe() + + broadcaster.unsubscribe(queue) + broadcaster.subscribe() # no longer at capacity