Harden poller from Codex review: close residual storm/robustness gaps

Adversarial review (gpt-5.5) of the new architecture surfaced real gaps: Hardware-safety / storm prevention: - Restart-storm guard: on startup, defer the first sweep if a sweep ran < POLL_SWEEP_INTERVAL ago (persisted in Redis), so deploy-cycling can no longer trigger back-to-back full hardware sweeps (poller.py). - Centralize pacing in the hardware gate: POLL_DRIVE_GAP is now held after EVERY hardware op (SMART, SES, host, MegaRAID, ledctl), not just the enclosure-drive loop (hwgate.py); removed the now-redundant per-loop sleeps. - Single-instance lock made atomic (Lua compare-and-set / compare-and-expire) and the refresher is now fatal on any error + has a done-callback, so a dropped lock can never leave two pollers sweeping concurrently (store.py, poller.py). - Warn loudly when POLL_CONCURRENCY > 1. Correctness: - Heartbeat now actually writes: cache_set omits EX when ttl<=0 (Redis rejects EX 0), so poller meta/last_sweep_ts persists — this also enables the restart-storm guard and the poller_fresh health flag (cache.py). - Web image runs a single uvicorn worker so the MQTT publisher is a true singleton (two workers shared a client_id and flapped the broker) (Dockerfile). - LED enqueue catches Redis errors -> API returns 503, not 500 (store.py). Deploy independence: - build.sh takes a target (web|poller|all) so web iterations never rebuild or repush the poller image. Nits: drop dead SMART_CACHE_TTL constant.
2026-06-16 15:58:43 +00:00
parent cccc2bc004
commit 7878e17e55
7 changed files with 123 additions and 44 deletions
--- a/5
+++ b/5
@@ -50,4 +50,7 @@ EXPOSE 8000
 HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/api/health')" || exit 1

-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]
+# Single worker: the MQTT publisher is a per-process singleton (multiple
+# workers would connect with the same client_id and flap the broker). The
+# API is a low-traffic internal monitor, so one worker is plenty.
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]
--- a/build.sh
+++ b/build.sh
@@ -1,24 +1,38 @@
 #!/usr/bin/env bash
 set -euo pipefail

+# Usage: ./build.sh [poller|web|all] [commit message]
+#   web   — build/push ONLY the web image (use this for web iterations so the
+#           poller image is never rebuilt/repushed)
+#   poller— build/push ONLY the poller image (touch deliberately)
+#   all   — both (default)
+#
+# Deploy pins images by SHA (compose.*.yml ship :latest as a convenience only).
+
 REG="docker.adamksmith.xyz"

 cd "$(dirname "$0")"

+TARGET="${1:-all}"
+case "$TARGET" in
+    poller) TARGETS="poller" ;;
+    web)    TARGETS="web" ;;
+    all)    TARGETS="poller web" ;;
+    *) echo "usage: $0 [poller|web|all] [message]" >&2; exit 1 ;;
+esac
+MSG="${2:-Build and push images}"
+
 # Stage and commit all changes
 git add -A
 if git diff --cached --quiet; then
    echo "No changes to commit, using HEAD"
 else
-    git commit -m "${1:-Build and push images}"
+    git commit -m "${MSG}"
 fi

 SHA=$(git rev-parse --short HEAD)

-# Two images from one Dockerfile (shared frontend build stage is cached):
-#   jbod-poller — privileged hardware producer
-#   jbod-web    — unprivileged read-only consumer (REST/UI/MQTT)
-for target in poller web; do
+for target in ${TARGETS}; do
    img="${REG}/jbod-${target}"
    echo "Building ${img}:${SHA}"
    docker build --target "${target}" -t "${img}:${SHA}" -t "${img}:latest" .
@@ -27,4 +41,4 @@ for target in poller web; do
    docker push "${img}:latest"
 done

-echo "Done: ${REG}/jbod-{poller,web}:${SHA}"
+echo "Done (${TARGET}): ${REG}/jbod-{${TARGETS// /,}}:${SHA}"
--- a/poller.py
+++ b/poller.py
@@ -35,9 +35,10 @@ logging.basicConfig(
 logger = logging.getLogger("poller")

 SWEEP_INTERVAL = int(os.environ.get("POLL_SWEEP_INTERVAL", "300"))
-DRIVE_GAP = float(os.environ.get("POLL_DRIVE_GAP", "0.5"))
 STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10"))
 LOCK_TTL = 30
+# Pacing between hardware ops (POLL_DRIVE_GAP) is enforced centrally by the
+# hardware gate, so every op — SMART, SES, host, MegaRAID, ledctl — is spaced.

 TOKEN = f"{socket.gethostname()}-{os.getpid()}"

@@ -66,16 +67,15 @@ async def sweep() -> None:
    except Exception as e:
        errors.append(f"zfs:{e}")

-    # 3. Per-drive SMART, one at a time with a gap between each.
+    # 3. Per-drive SMART (gate serializes + paces each call).
    for dev in drive_devices:
        try:
            data = await _run_smartctl(dev)
            await store.set_smart(dev, data)
        except Exception as e:
            errors.append(f"smart:{dev}:{e}")
-        await asyncio.sleep(DRIVE_GAP)

-    # 4. Per-enclosure SES status (gated), paced.
+    # 4. Per-enclosure SES status (gate serializes + paces each call).
    for enc in enclosures:
        if not enc.get("sg_device"):
            continue
@@ -85,7 +85,6 @@ async def sweep() -> None:
                await store.set_ses(enc["id"], ses)
        except Exception as e:
            errors.append(f"ses:{enc['id']}:{e}")
-        await asyncio.sleep(DRIVE_GAP)

    # 5. Host (non-enclosure) drives + MegaRAID (gated, reuses pool_map).
    try:
@@ -132,11 +131,27 @@ async def led_consumer() -> None:
 async def lock_refresher() -> None:
    while True:
        await asyncio.sleep(LOCK_TTL / 3)
-        if not await store.refresh_lock(TOKEN, LOCK_TTL):
+        # Any failure to confirm we still hold the lock is fatal — keeping the
+        # lock alive is what prevents a second poller from sweeping concurrently.
+        try:
+            ok = await store.refresh_lock(TOKEN, LOCK_TTL)
+        except Exception as e:
+            logger.error("lock refresh error (%s) — exiting to avoid double-polling", e)
+            os._exit(1)
+        if not ok:
            logger.error("lost poller lock — exiting to avoid double-polling")
            os._exit(1)


+def _critical_task_done(task: asyncio.Task) -> None:
+    """Fail-safe: if a critical task ends for any reason other than a clean
+    cancel (shutdown), exit so we never run unguarded."""
+    if task.cancelled():
+        return
+    logger.error("critical task ended unexpectedly — exiting")
+    os._exit(1)
+
+
 async def main() -> None:
    await init_cache()
    while not redis_available():
@@ -162,7 +177,19 @@ async def main() -> None:
    logger.info("startup jitter %.1fs", jitter)
    await asyncio.sleep(jitter)

+    # Restart-storm guard: if a sweep ran recently (persisted in Redis across
+    # restarts), wait out the remainder of the interval instead of immediately
+    # re-sweeping. This is the key protection against deploy-cycling storms.
+    meta = await store.get_meta()
+    if meta and meta.get("last_sweep_ts"):
+        since = store.now() - meta["last_sweep_ts"]
+        if 0 <= since < SWEEP_INTERVAL:
+            wait = SWEEP_INTERVAL - since
+            logger.info("last sweep %.0fs ago — deferring first sweep %.0fs", since, wait)
+            await asyncio.sleep(wait)
+
    refresher = asyncio.create_task(lock_refresher())
+    refresher.add_done_callback(_critical_task_done)
    leds = asyncio.create_task(led_consumer())
    try:
        while True:
--- a/services/cache.py
+++ b/services/cache.py
@@ -58,10 +58,13 @@ async def cache_get(key: str) -> Any | None:


 async def cache_set(key: str, value: Any, ttl: int = 120) -> None:
-    """SET key in Redis with expiry, silently catches errors."""
+    """SET key in Redis. ttl<=0 stores with no expiry (Redis rejects EX 0)."""
    if _redis is None:
        return
    try:
+        if ttl and ttl > 0:
            await _redis.set(key, json.dumps(value), ex=ttl)
+        else:
+            await _redis.set(key, json.dumps(value))
    except Exception as e:
        logger.warning("Redis SET %s failed: %s", key, e)
--- a/services/hwgate.py
+++ b/services/hwgate.py
@@ -1,27 +1,49 @@
-"""Global hardware-access gate.
+"""Global hardware-access gate + pacing.

-Every command that touches the SAS bus — smartctl, sg_ses, zpool, ledctl —
-must run inside this gate. It serializes hardware I/O so the poller can
-never fire a thundering herd of subprocesses at fragile SAS expanders
-(which on some shelves drop out and suspend pools under concurrent load).
+Every command that touches the SAS bus — smartctl, sg_ses, zpool, lsblk,
+ledctl — must run inside this gate. It does two things:

-Concurrency defaults to 1 (fully serialized). It can be raised via
-POLL_CONCURRENCY for healthier backplanes, but 1 is the safe default.
+  1. Serializes hardware I/O (semaphore, default concurrency 1) so the
+     poller can never fire a thundering herd at fragile SAS expanders.
+  2. Paces it: after every hardware op it holds the gate for POLL_DRIVE_GAP
+     seconds, so back-to-back ops (gathered SMART, MegaRAID scans, SES,
+     ledctl) are always spaced — not just the per-drive loop.

-The gate is a per-process asyncio.Semaphore — only the poller process
-calls hardware, so a per-process gate is sufficient. The semaphore is
-created lazily on first use so it binds to the running event loop.
+This is a per-process gate; only the poller process calls hardware, so
+per-process serialization is sufficient. Created lazily so it binds to the
+running event loop.
 """
 import asyncio
+import contextlib
+import logging
 import os

+logger = logging.getLogger(__name__)
+
 _sem: asyncio.Semaphore | None = None


-def gate() -> asyncio.Semaphore:
-    """Return the shared hardware semaphore (use as `async with gate():`)."""
+def _semaphore() -> asyncio.Semaphore:
    global _sem
    if _sem is None:
        n = max(1, int(os.environ.get("POLL_CONCURRENCY", "1")))
+        if n > 1:
+            logger.warning(
+                "POLL_CONCURRENCY=%d (>1) — hardware ops will run concurrently; "
+                "only do this on backplanes known to tolerate it", n,
+            )
        _sem = asyncio.Semaphore(n)
    return _sem
+
+
+@contextlib.asynccontextmanager
+async def gate():
+    """Acquire the hardware gate for one op, then hold it for the pacing gap."""
+    sem = _semaphore()
+    async with sem:
+        try:
+            yield
+        finally:
+            gap = float(os.environ.get("POLL_DRIVE_GAP", "0.5"))
+            if gap > 0:
+                await asyncio.sleep(gap)
--- a/services/smart.py
+++ b/services/smart.py
@@ -18,8 +18,6 @@ ATTR_PENDING = 197
 ATTR_UNCORRECTABLE = 198
 ATTR_WEAR_LEVELING = 177  # SSD wear leveling

-SMART_CACHE_TTL = int(os.environ.get("SMART_CACHE_TTL", "120"))
-

 def smartctl_available() -> bool:
    return shutil.which("smartctl") is not None
--- a/services/store.py
+++ b/services/store.py
@@ -82,11 +82,15 @@ async def get_meta() -> dict | None:
 # ── locate-LED command queue ────────────────────────────────────────────
 async def enqueue_led(device: str, state: str) -> bool:
    """API side: push a LED request for the poller to execute. Returns False
-    if Redis is unavailable (so the API can surface 503)."""
+    if Redis is unavailable or the push fails (so the API can surface 503)."""
    client = get_client()
    if client is None:
        return False
+    try:
        await client.rpush(K_LED_QUEUE, json.dumps({"device": device, "state": state}))
+    except Exception as e:
+        logger.warning("LED enqueue failed: %s", e)
+        return False
    return True


@@ -105,29 +109,37 @@ async def pop_led(timeout: int = 5) -> dict | None:
        return None


-# ── poller single-instance lock ─────────────────────────────────────────
+# ── poller single-instance lock (atomic compare-and-set) ────────────────
+# Acquire if free OR already ours, refreshing the TTL — all in one round trip
+# so there's no GET/SET race that could let two pollers run concurrently.
+_ACQUIRE_LUA = (
+    "local v = redis.call('get', KEYS[1]) "
+    "if v == false or v == ARGV[1] then "
+    "  redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) return 1 "
+    "else return 0 end"
+)
+# Extend only if we still own it (atomic compare-and-expire).
+_REFRESH_LUA = (
+    "if redis.call('get', KEYS[1]) == ARGV[1] then "
+    "  return redis.call('expire', KEYS[1], ARGV[2]) "
+    "else return 0 end"
+)
+
+
 async def acquire_lock(token: str, ttl: int = 30) -> bool:
-    """Try to claim the poller lock. Returns True if acquired/owned."""
+    """Atomically claim the poller lock (or re-own it). True if held."""
    client = get_client()
    if client is None:
        return True  # no Redis → no coordination possible; run anyway
-    got = await client.set(K_LOCK, token, nx=True, ex=ttl)
-    if got:
-        return True
-    current = await client.get(K_LOCK)
-    return current == token
+    return bool(await client.eval(_ACQUIRE_LUA, 1, K_LOCK, token, ttl))


 async def refresh_lock(token: str, ttl: int = 30) -> bool:
-    """Extend the lock if we still own it."""
+    """Atomically extend the lock iff we still own it."""
    client = get_client()
    if client is None:
        return True
-    current = await client.get(K_LOCK)
-    if current == token:
-        await client.expire(K_LOCK, ttl)
-        return True
-    return False
+    return bool(await client.eval(_REFRESH_LUA, 1, K_LOCK, token, ttl))


 def now() -> float: