From 7878e17e550d8a7a3aee1970a2a1210293c08c64 Mon Sep 17 00:00:00 2001 From: adam Date: Tue, 16 Jun 2026 15:58:43 +0000 Subject: [PATCH] Harden poller from Codex review: close residual storm/robustness gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- Dockerfile | 5 ++++- build.sh | 26 ++++++++++++++++++++------ poller.py | 39 +++++++++++++++++++++++++++++++++------ services/cache.py | 7 +++++-- services/hwgate.py | 46 ++++++++++++++++++++++++++++++++++------------ services/smart.py | 2 -- services/store.py | 42 +++++++++++++++++++++++++++--------------- 7 files changed, 123 insertions(+), 44 deletions(-) diff --git a/Dockerfile b/Dockerfile index ef11d8d..c09ea5e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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"] diff --git a/build.sh b/build.sh index 0725d9e..30ff9d2 100755 --- 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}" diff --git a/poller.py b/poller.py index 5bff8a5..e107597 100644 --- 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: diff --git a/services/cache.py b/services/cache.py index de56b4e..a7c9970 100644 --- 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: - await _redis.set(key, json.dumps(value), ex=ttl) + 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) diff --git a/services/hwgate.py b/services/hwgate.py index 8547c75..67d7cd9 100644 --- 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) diff --git a/services/smart.py b/services/smart.py index 13f0aa3..0051ef1 100644 --- 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 diff --git a/services/store.py b/services/store.py index 0c5046f..ee349d9 100644 --- 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 - await client.rpush(K_LED_QUEUE, json.dumps({"device": device, "state": state})) + 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: