Error Recovery & Graceful Deployment

from contextlib import asynccontextmanager
import asyncio, signal

shutdown_event = asyncio.Event()

@asynccontextmanager
async def lifespan(app):
    # Trap SIGTERM so we can drain instead of crashing.
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, shutdown_event.set)
    yield
    # Drain phase
    log.info('draining %d connections', manager.total_connections)
    for room in list(manager.rooms):
        await manager.broadcast(room, {
            'type': 'system.reconnect',
            'data': {'reason': 'deploy'},
        })
    await asyncio.sleep(2.0)  # let clients reconnect on their own
    close_tasks = [
        ws.close(code=1001, reason='deploy')
        for ws in list(manager.ws_meta)
    ]
    await asyncio.gather(*close_tasks, return_exceptions=True)
    log.info('drain complete')

app = FastAPI(lifespan=lifespan)

class BufferedSocket {
  constructor(url) {
    this.url = url;
    this.buffer = [];
    this._connect();
  }
  _connect() {
    this.ws = new WebSocket(this.url);
    this.ws.addEventListener('open', () => {
      while (this.buffer.length) this.ws.send(this.buffer.shift());
    });
    this.ws.addEventListener('close', () => {
      // schedule reconnect (Track 2 backoff pattern)
      setTimeout(() => this._connect(), 1_000);
    });
  }
  send(msg) {
    const wire = JSON.stringify(msg);
    if (this.ws?.readyState === WebSocket.OPEN) this.ws.send(wire);
    else this.buffer.push(wire);
  }
}