From 1ec5cfb4df6f3e4a777080582a6e295be36eac8e Mon Sep 17 00:00:00 2001
From: grabowski <berwn@buildfor.life>
Date: Wed, 12 Aug 2026 11:34:12 +0700
Subject: [PATCH] perf: gzip responses; multi-worker serving with single
 collection leader
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

GZipMiddleware (min 500 bytes) compresses the dashboard HTML ~4x and
station JSON up to ~100x, end-to-end through the Caddy TLS terminator —
production load testing showed the deployment is bandwidth-bound once
the response caches hit, so compression is the capacity lever.

WEB_WORKERS (default 2) runs uvicorn multi-process via the app import
string. Every worker executes the lifespan, so a localhost lock port
(COLLECTION_LEADER_PORT, default 8901) elects exactly one
background-collection leader per machine — RID/HII polling stays
once-per-cycle instead of once-per-worker; the lock releases with the
process. Locust clients now send Accept-Encoding so future runs measure
compressed transfer, as browsers do.
---
 scripts/locustfile.py | 31 +++++++++++++++++-------------
 src/config.py         |  5 +++++
 src/main.py           | 21 +++++++++++++++------
 src/web_api.py        | 44 +++++++++++++++++++++++++++++++++++++++++--
 4 files changed, 80 insertions(+), 21 deletions(-)

diff --git a/scripts/locustfile.py b/scripts/locustfile.py
index 20a0bb1..5bb8230 100644
--- a/scripts/locustfile.py
+++ b/scripts/locustfile.py
@@ -20,6 +20,9 @@ import random
 
 from locust import FastHttpUser, between, task
 
+# Explicit so measurements reflect compressed transfer (browsers always send this)
+GZIP = {"Accept-Encoding": "gzip, deflate"}
+
 
 class DashboardVisitor(FastHttpUser):
     """A browser session: initial page load, then periodic refresh polling."""
@@ -29,23 +32,23 @@ class DashboardVisitor(FastHttpUser):
 
     def on_start(self):
         # What one real page load requests
-        self.client.get("/")
-        self.client.get("/stations")
-        self.client.get("/measurements/latest?limit=500")
-        self.client.get("/api/hii/waterlevel/latest")
-        self.client.get("/api/hii/rainfall/latest")
+        self.client.get("/", headers=GZIP)
+        self.client.get("/stations", headers=GZIP)
+        self.client.get("/measurements/latest?limit=500", headers=GZIP)
+        self.client.get("/api/hii/waterlevel/latest", headers=GZIP)
+        self.client.get("/api/hii/rainfall/latest", headers=GZIP)
 
     @task(4)
     def poll_latest(self):
-        self.client.get("/measurements/latest?limit=500")
+        self.client.get("/measurements/latest?limit=500", headers=GZIP)
 
     @task(2)
     def poll_forecast(self):
-        self.client.get("/forecast")
+        self.client.get("/forecast", headers=GZIP)
 
     @task(2)
     def poll_rain(self):
-        self.client.get("/api/hii/rainfall/latest")
+        self.client.get("/api/hii/rainfall/latest", headers=GZIP)
 
     @task(1)
     def view_history(self):
@@ -53,12 +56,13 @@ class DashboardVisitor(FastHttpUser):
         hours = random.choice([24, 168, 720])
         self.client.get(
             f"/measurements/history/{station}?hours={hours}",
+            headers=GZIP,
             name="/measurements/history/[station]",
         )
 
     @task(1)
     def stats(self):
-        self.client.get("/api/stats")
+        self.client.get("/api/stats", headers=GZIP)
 
 
 class ApiConsumer(FastHttpUser):
@@ -69,25 +73,26 @@ class ApiConsumer(FastHttpUser):
 
     @task(3)
     def latest(self):
-        self.client.get("/measurements/latest?limit=100")
+        self.client.get("/measurements/latest?limit=100", headers=GZIP)
 
     @task(3)
     def hii_feeds(self):
         self.client.get(random.choice(
             ["/api/hii/waterlevel/latest", "/api/hii/rainfall/latest"]
-        ), name="/api/hii/[feed]/latest")
+        ), headers=GZIP, name="/api/hii/[feed]/latest")
 
     @task(2)
     def forecast(self):
-        self.client.get("/forecast")
+        self.client.get("/forecast", headers=GZIP)
 
     @task(2)
     def heavy_history(self):
         self.client.get(
             "/measurements/history/P.1?hours=8760",
+            headers=GZIP,
             name="/measurements/history/P.1 [heavy]",
         )
 
     @task(1)
     def health(self):
-        self.client.get("/health")
+        self.client.get("/health", headers=GZIP)
diff --git a/src/config.py b/src/config.py
index 7f78b88..35f82d9 100644
--- a/src/config.py
+++ b/src/config.py
@@ -91,6 +91,11 @@ class Config:
     # TTL for the /measurements/latest response cache (hottest endpoint)
     LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
 
+    # Web server worker processes. Above 1, uvicorn forks workers and a
+    # localhost lock port elects a single background-collection leader.
+    WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2"))
+    COLLECTION_LEADER_PORT = int(os.getenv("COLLECTION_LEADER_PORT", "8901"))
+
     # Umami analytics (self-hosted). The website id is public (it ships in the
     # dashboard <script> tag); server-side API tracking posts to /api/send.
     UMAMI_API_URL = os.getenv("UMAMI_API_URL", "https://stats.buildfor.life/api/send")
diff --git a/src/main.py b/src/main.py
index 62fa47e..54ac005 100644
--- a/src/main.py
+++ b/src/main.py
@@ -344,15 +344,24 @@ def run_web_api():
     try:
         import uvicorn
 
-        from .web_api import app
-
         # Validate configuration
         Config.validate_config()
 
-        # Run the server
-        uvicorn.run(
-            app, host="0.0.0.0", port=8000, log_config=None  # Use our custom logging
-        )
+        workers = max(1, Config.WEB_WORKERS)
+        if workers > 1:
+            # Multi-worker needs the app as an import string; a localhost lock
+            # port keeps background collection in exactly one worker.
+            uvicorn.run(
+                "src.web_api:app",
+                host="0.0.0.0",
+                port=8000,
+                workers=workers,
+                log_config=None,
+            )
+        else:
+            from .web_api import app
+
+            uvicorn.run(app, host="0.0.0.0", port=8000, log_config=None)
 
     except ImportError:
         logger.error("FastAPI not installed. Run: pip install fastapi uvicorn")
diff --git a/src/web_api.py b/src/web_api.py
index c718f44..73b987d 100644
--- a/src/web_api.py
+++ b/src/web_api.py
@@ -24,6 +24,7 @@ from fastapi import (
     Response,
 )
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, HTMLResponse
 from fastapi.staticfiles import StaticFiles
 
@@ -110,6 +111,26 @@ app_state = {
 }
 
 
+def _acquire_collection_leadership(port: int):
+    """Elect one background-collection leader per machine via a localhost bind.
+
+    With multiple uvicorn workers every process runs this lifespan; only the
+    worker holding the lock port runs the scraper/HII loops, so external APIs
+    are polled once per cycle instead of once per worker. The socket is held
+    for the process lifetime and releases automatically if the worker dies.
+    """
+    import socket
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        sock.bind(("127.0.0.1", port))
+        sock.listen(1)
+        return sock
+    except OSError:
+        sock.close()
+        return None
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Application lifespan manager"""
@@ -148,8 +169,18 @@ async def lifespan(app: FastAPI):
     health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000))
     app_state["health_manager"] = health_manager
 
-    # Start background scraping task
-    app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
+    # Start background scraping in exactly one worker per machine
+    app_state["leader_lock"] = _acquire_collection_leadership(
+        Config.COLLECTION_LEADER_PORT
+    )
+    if app_state["leader_lock"]:
+        app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
+        logger.info("This worker is the background-collection leader")
+    else:
+        logger.info(
+            "Another worker holds collection leadership; "
+            "background scraping disabled in this process"
+        )
 
     logger.info("Water Monitor API started successfully")
 
@@ -164,6 +195,8 @@ async def lifespan(app: FastAPI):
             await app_state["scraping_task"]
         except asyncio.CancelledError:
             pass
+    if app_state.get("leader_lock"):
+        app_state["leader_lock"].close()
 
     logger.info("Water Monitor API shutdown complete")
 
@@ -185,6 +218,13 @@ app.mount(
 # Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured
 # we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination);
 # credentials are only enabled when explicit origins are provided.
+# Compress large responses end-to-end: the dashboard HTML and station JSON
+# payloads shrink ~5-6x, which matters both on the internal VPN hop to the
+# Caddy TLS terminator and on the public leg (Caddy passes Content-Encoding
+# through). Load testing showed the deployment is bandwidth-bound, not
+# compute-bound, once the response caches are hit.
+app.add_middleware(GZipMiddleware, minimum_size=500)
+
 _cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"]
 _cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS)
 app.add_middleware(