serversdown/slmm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
b9bf4ea9e69088908d0de1b89beae46e3a2701eb
slmm/app/recovery.py
T
serversdown 3c22f1d70c feat: wedge detection and automatic recovery orchestration 
Adds the recovery layer for the NL-43 TCP wedge failure mode (control
listener on 2255 dies while device/FTP stay alive — see
SLM-stress-test/NL43_RX55_TCP_Wedge_Investigation_2026-02-18.md).

- Failure classification: poll failures probed raw and classified as
  wedged (refused = stack alive, listener gone) vs offline (timeout).
  FTP-port probe upgrades wedge confidence to definitive.
- RecoveryManager state machine: confirm wedge -> trigger reset ->
  wait boot -> reconnect -> resume measurement via start_cycle (keeps
  overwrite protection) -> full incident trail in device logs.
- Pluggable reset backends: manual (notify, default) and webhook
  (HTTP to relay controller: Pi GPIO, Shelly, Tasmota, etc.).
- Safety: auto_recovery_enabled off by default, 2-failure confirmation
  threshold, windowed attempt limit (2 per 6h), one recovery per device.
- Endpoints: GET recovery/status, POST recovery/probe (diagnostic),
  POST recovery/trigger (manual, bypasses threshold).
- Migration: migrate_add_recovery_fields.py
- Tests: test_wedge_recovery.py — fake wedge-able NL-43 + fake relay
  webhook; 23 assertions covering classification, end-to-end recovery
  with resume, and gating. All passing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 05:03:25 +00:00

483 lines
19 KiB
Python
Raw Blame History

"""
Wedge detection and automatic recovery for NL-43 devices.
The NL-43's TCP control listener (port 2255) can stop listening while the
device otherwise remains healthy (LAN stack alive, FTP alive) — the "wedge"
failure mode documented in SLM-stress-test/NL43_RX55_TCP_Wedge_Investigation_2026-02-18.md.
Only a power cycle of the NL-43 has been proven to restore the listener.
This module provides:
1. Failure classification — distinguishes a wedge (connection REFUSED, i.e.
the device's IP stack answered with RST so the host is alive but the
listener is gone) from the device being offline (connect timeout — power
loss, cellular outage). An FTP-port probe adds confidence when available.
2. RecoveryManager — a per-device state machine that, once a wedge is
confirmed, triggers a reset through a pluggable backend, waits for the
device to boot, reconnects, and resumes measurement if one was running.
Reset backends:
- "manual" — log/flag only; a human power-cycles the device. The manager
keeps watching and still auto-resumes measurement when the
device returns.
- "webhook" — HTTP request to a relay controller (Pi GPIO relay, Shelly,
Tasmota, RX55 GPIO bridge, etc.) that power-cycles the device.
Recovery states (NL43Status.recovery_state):
idle -> resetting -> waiting_boot -> reconnecting -> resuming -> idle
\\-> awaiting_manual_reset -> reconnecting -> ...
any step may end in "failed" (recorded in last_recovery_result).
"""
import asyncio
import json
import logging
import os
import time
import urllib.request
from datetime import datetime
from typing import Dict, List, Optional
from sqlalchemy.orm import Session
from app.database import SessionLocal
from app.models import NL43Config, NL43Status
from app.device_logger import log_device_event
logger = logging.getLogger(__name__)
# Number of consecutive wedge-classified poll failures required before
# automatic recovery is triggered (manual trigger bypasses this).
WEDGE_CONFIRM_FAILURES = int(os.getenv("WEDGE_CONFIRM_FAILURES", "2"))
# Timeout for individual TCP probes during classification/reconnection
PROBE_TIMEOUT = float(os.getenv("RECOVERY_PROBE_TIMEOUT", "5.0"))
# Seconds between reconnect probes after a reset
RECONNECT_PROBE_INTERVAL = 10
# Seconds between probes while awaiting a manual reset
MANUAL_WATCH_PROBE_INTERVAL = 30
# ---------------------------------------------------------------------------
# TCP probing and failure classification
# ---------------------------------------------------------------------------
async def probe_tcp(host: str, port: int, timeout: float = PROBE_TIMEOUT) -> str:
"""Probe a TCP port and report what happened.
Returns one of:
"open" — connect succeeded (closed again immediately)
"refused" — RST received: host stack alive, no listener
"timeout" — no response: host down / network unreachable / filtered
"unreachable" — OS-level error (no route, etc.)
"""
try:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=timeout
)
writer.close()
try:
await writer.wait_closed()
except Exception:
pass
return "open"
except asyncio.TimeoutError:
return "timeout"
except ConnectionRefusedError:
return "refused"
except OSError:
return "unreachable"
async def classify_failure(host: str, tcp_port: int, ftp_port: Optional[int] = 21) -> dict:
"""Classify a poll failure as wedged / offline / ok.
The wedge signature (from the Feb 2026 investigation): the control port
actively REFUSES (RST) while the device is otherwise alive. A refused
connection requires a live IP stack at the far end — a powered-off device
behind the modem produces a timeout instead. The FTP probe upgrades
confidence when it accepts, but refusal of the control port alone is
already a strong wedge indicator.
Returns dict:
state — "ok" | "wedged" | "offline"
confidence — "high" | "medium" (only meaningful for "wedged")
tcp_probe — raw probe result for the control port
ftp_probe — raw probe result for the FTP port (None if not probed)
"""
tcp_result = await probe_tcp(host, tcp_port)
result = {
"state": "ok",
"confidence": "high",
"tcp_probe": tcp_result,
"ftp_probe": None,
}
if tcp_result == "open":
# Transient failure — the listener is there now
return result
if tcp_result in ("timeout", "unreachable"):
result["state"] = "offline"
return result
# tcp_result == "refused": host alive, control listener gone
result["state"] = "wedged"
if ftp_port:
ftp_result = await probe_tcp(host, ftp_port)
result["ftp_probe"] = ftp_result
# FTP accepting while control refuses is the definitive signature
result["confidence"] = "high" if ftp_result == "open" else "medium"
else:
result["confidence"] = "medium"
return result
# ---------------------------------------------------------------------------
# Reset backends
# ---------------------------------------------------------------------------
class ResetBackend:
"""Base class for reset actuation backends.
Backends receive a plain dict of snapshotted config values (never a live
ORM row — recovery runs as a detached task after the originating DB
session has closed).
"""
name = "base"
async def trigger(self, params: dict) -> dict:
"""Attempt to power-cycle the device.
Returns dict: {"triggered": bool, "detail": str}
"triggered" True means a reset is expected to be in progress.
"""
raise NotImplementedError
class ManualNotifyBackend(ResetBackend):
"""No actuation — flags the wedge for a human to power-cycle the device."""
name = "manual"
async def trigger(self, params: dict) -> dict:
return {
"triggered": False,
"detail": "No reset hardware configured - manual power cycle required",
}
class WebhookBackend(ResetBackend):
"""Calls an HTTP endpoint that power-cycles the device.
The endpoint is expected to perform the full cycle itself (e.g. a Shelly
relay with `?turn=off&timer=10`, a Tasmota PulseTime rule, or a small
HTTP service on a Pi driving a relay GPIO). Any 2xx response counts as
triggered.
"""
name = "webhook"
async def trigger(self, params: dict) -> dict:
url = params.get("webhook_url")
if not url:
return {"triggered": False, "detail": "reset_backend is 'webhook' but reset_webhook_url is not set"}
method = (params.get("webhook_method") or "POST").upper()
def _call() -> dict:
req = urllib.request.Request(url, method=method)
req.add_header("User-Agent", "SLMM-wedge-recovery")
with urllib.request.urlopen(req, timeout=15) as resp:
body = resp.read(500).decode(errors="ignore")
return {"status": resp.status, "body": body}
try:
resp = await asyncio.to_thread(_call)
if 200 <= resp["status"] < 300:
return {"triggered": True, "detail": f"Webhook {method} {url} -> {resp['status']}"}
return {"triggered": False, "detail": f"Webhook {method} {url} -> HTTP {resp['status']}"}
except Exception as e:
return {"triggered": False, "detail": f"Webhook {method} {url} failed: {e}"}
BACKENDS: Dict[str, ResetBackend] = {
"manual": ManualNotifyBackend(),
"webhook": WebhookBackend(),
}
def get_backend(name: Optional[str]) -> ResetBackend:
return BACKENDS.get((name or "manual").lower(), BACKENDS["manual"])
# ---------------------------------------------------------------------------
# Recovery manager
# ---------------------------------------------------------------------------
class RecoveryManager:
"""Orchestrates wedge recovery, one concurrent recovery per device."""
def __init__(self):
self._active: Dict[str, asyncio.Task] = {}
# In-memory attempt timestamps per unit for windowed rate limiting.
# Resets on service restart; last_recovery_at in the DB provides a
# coarse cross-restart guard via the same window check.
self._attempts: Dict[str, List[float]] = {}
# -- public API ---------------------------------------------------------
def is_active(self, unit_id: str) -> bool:
task = self._active.get(unit_id)
return task is not None and not task.done()
def get_status(self, unit_id: str) -> dict:
now = time.time()
attempts = self._attempts.get(unit_id, [])
return {
"recovery_in_progress": self.is_active(unit_id),
"attempts_this_session": len(attempts),
"last_attempt_seconds_ago": round(now - attempts[-1]) if attempts else None,
}
def maybe_start_recovery(self, cfg: NL43Config, status: NL43Status, classification: dict, db: Session) -> bool:
"""Start a recovery task if conditions are met. Called from the poller.
Returns True if a recovery task was started.
"""
unit_id = cfg.unit_id
if not cfg.auto_recovery_enabled:
logger.info(f"[RECOVERY] {unit_id}: wedge detected but auto_recovery_enabled is off")
return False
if self.is_active(unit_id):
logger.debug(f"[RECOVERY] {unit_id}: recovery already in progress")
return False
if status.consecutive_failures < WEDGE_CONFIRM_FAILURES:
logger.info(
f"[RECOVERY] {unit_id}: wedge suspected "
f"({status.consecutive_failures}/{WEDGE_CONFIRM_FAILURES} confirmations)"
)
return False
if not self._attempt_allowed(cfg, status):
logger.warning(
f"[RECOVERY] {unit_id}: attempt limit reached "
f"({cfg.recovery_max_attempts} per {cfg.recovery_window_minutes} min) - standing down"
)
log_device_event(
unit_id, "ERROR", "RECOVERY",
f"Wedge persists but recovery attempt limit reached "
f"({cfg.recovery_max_attempts} per {cfg.recovery_window_minutes} min). Manual intervention required.",
db,
)
return False
return self.start_recovery(cfg, classification)
def start_recovery(self, cfg: NL43Config, classification: Optional[dict] = None) -> bool:
"""Launch the recovery task (also used by the manual-trigger endpoint)."""
unit_id = cfg.unit_id
if self.is_active(unit_id):
return False
self._attempts.setdefault(unit_id, []).append(time.time())
# Snapshot config values so the task doesn't depend on a live DB row
params = {
"unit_id": unit_id,
"host": cfg.host,
"tcp_port": cfg.tcp_port,
"ftp_port": cfg.ftp_port or 21,
"ftp_enabled": bool(cfg.ftp_enabled),
"ftp_username": cfg.ftp_username,
"ftp_password": cfg.ftp_password,
"backend": (cfg.reset_backend or "manual").lower(),
"webhook_url": cfg.reset_webhook_url,
"webhook_method": cfg.reset_webhook_method,
"boot_wait": cfg.recovery_boot_wait_seconds or 90,
"reconnect_timeout": cfg.recovery_reconnect_timeout or 180,
"auto_resume": bool(cfg.recovery_auto_resume),
"window_minutes": cfg.recovery_window_minutes or 360,
"classification": classification or {},
}
task = asyncio.create_task(self._run_recovery(params))
self._active[unit_id] = task
logger.info(f"[RECOVERY] {unit_id}: recovery task started (backend={params['backend']})")
return True
# -- internals ----------------------------------------------------------
def _attempt_allowed(self, cfg: NL43Config, status: NL43Status) -> bool:
"""Windowed attempt limit: max N attempts per window."""
window_seconds = (cfg.recovery_window_minutes or 360) * 60
max_attempts = cfg.recovery_max_attempts or 2
now = time.time()
recent = [t for t in self._attempts.get(cfg.unit_id, []) if now - t < window_seconds]
self._attempts[cfg.unit_id] = recent
if len(recent) >= max_attempts:
return False
# Cross-restart guard: if the in-memory history is empty (service
# restarted) but the DB shows a very recent attempt, respect a
# cooldown of one window-fraction to avoid rapid-fire cycling.
if not recent and status.last_recovery_at is not None:
elapsed = (datetime.utcnow() - status.last_recovery_at).total_seconds()
if elapsed < window_seconds / max_attempts / 2:
return False
return True
def _set_state(self, unit_id: str, recovery_state: str, result: Optional[str] = None,
connection_state: Optional[str] = None, mark_recovered: bool = False):
"""Persist recovery progress on the status row (own short-lived session)."""
db = SessionLocal()
try:
status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
if not status:
status = NL43Status(unit_id=unit_id)
db.add(status)
status.recovery_state = recovery_state
if result is not None:
status.last_recovery_result = result[:1000]
if connection_state is not None:
status.connection_state = connection_state
if mark_recovered:
status.is_reachable = True
status.consecutive_failures = 0
db.commit()
except Exception as e:
logger.warning(f"[RECOVERY] {unit_id}: failed to persist state '{recovery_state}': {e}")
finally:
db.close()
async def _run_recovery(self, p: dict):
"""The recovery state machine. Runs as a detached task."""
unit_id = p["unit_id"]
backend = get_backend(p["backend"])
started = time.time()
def _log(level: str, message: str):
log_device_event(unit_id, level, "RECOVERY", message)
try:
# Record attempt start
db = SessionLocal()
try:
status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
want_resume = bool(status and status.measurement_state == "Start")
if status:
status.last_recovery_at = datetime.utcnow()
status.recovery_state = "resetting"
db.commit()
finally:
db.close()
cls = p.get("classification") or {}
_log("ERROR",
f"Wedge recovery started (backend={backend.name}, "
f"tcp_probe={cls.get('tcp_probe', '?')}, ftp_probe={cls.get('ftp_probe', '?')}, "
f"was_measuring={want_resume})")
# --- Step 1: trigger reset ---------------------------------------
trigger_result = await backend.trigger(p)
_log("INFO", f"Reset trigger: {trigger_result['detail']}")
if trigger_result["triggered"]:
# --- Step 2: wait for the device to boot ---------------------
self._set_state(unit_id, "waiting_boot")
_log("INFO", f"Reset triggered - waiting {p['boot_wait']}s for device boot")
await asyncio.sleep(p["boot_wait"])
reconnect_deadline = time.time() + p["reconnect_timeout"]
probe_interval = RECONNECT_PROBE_INTERVAL
else:
if backend.name == "manual":
# Watch for a human to power-cycle the device
self._set_state(unit_id, "awaiting_manual_reset")
_log("ERROR",
f"WEDGE DETECTED on {p['host']}:{p['tcp_port']} - "
f"manual power cycle required. Watching for device return "
f"(up to {p['window_minutes']} min).")
reconnect_deadline = time.time() + p["window_minutes"] * 60
probe_interval = MANUAL_WATCH_PROBE_INTERVAL
else:
raise RuntimeError(f"Reset trigger failed: {trigger_result['detail']}")
# --- Step 3: wait for the control port to come back --------------
self._set_state(unit_id, "reconnecting")
port_open = False
while time.time() < reconnect_deadline:
probe = await probe_tcp(p["host"], p["tcp_port"])
if probe == "open":
port_open = True
break
await asyncio.sleep(probe_interval)
if not port_open:
raise RuntimeError(
f"Device did not return within "
f"{round(reconnect_deadline - started)}s of recovery start"
)
_log("INFO", f"Control port {p['tcp_port']} is accepting connections again")
# --- Step 4: flush stale pooled connection, verify protocol ------
from app.services import NL43Client, _connection_pool
device_key = f"{p['host']}:{p['tcp_port']}"
await _connection_pool.discard(device_key)
client = NL43Client(
p["host"], p["tcp_port"], timeout=10.0,
ftp_username=p["ftp_username"], ftp_password=p["ftp_password"],
ftp_port=p["ftp_port"],
)
state = await client.get_measurement_state()
_log("INFO", f"Device responding to commands - measurement state: {state}")
# --- Step 5: resume measurement if needed -------------------------
resumed = False
if want_resume and p["auto_resume"] and state != "Start":
self._set_state(unit_id, "resuming")
_log("INFO", "Measurement was running before the wedge - executing start cycle")
cycle = await client.start_cycle(sync_clock=True)
resumed = True
_log("INFO",
f"Measurement restarted (index {cycle.get('old_index')} -> {cycle.get('new_index')})")
elif want_resume and state == "Start":
_log("INFO", "Device is already measuring after reset - no resume needed")
resumed = True
# --- Done ---------------------------------------------------------
elapsed = round(time.time() - started)
summary = (
f"Recovered in {elapsed}s via {backend.name}"
+ (", measurement resumed" if resumed else
(", measurement NOT resumed" if want_resume else ""))
)
self._set_state(unit_id, "idle", result=summary, connection_state="ok", mark_recovered=True)
_log("INFO", f"Wedge recovery SUCCEEDED: {summary}")
except Exception as e:
elapsed = round(time.time() - started)
summary = f"Recovery FAILED after {elapsed}s: {e}"
self._set_state(unit_id, "failed", result=summary)
_log("ERROR", summary)
logger.error(f"[RECOVERY] {unit_id}: {summary}")
finally:
self._active.pop(unit_id, None)
# Global singleton
recovery_manager = RecoveryManager()
Reference in New Issue View Git Blame Copy Permalink