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>
This commit is contained in:
@@ -0,0 +1,377 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
End-to-end test for wedge detection and recovery (app/recovery.py).
|
||||
|
||||
Spins up a fake NL-43 (control + FTP listeners speaking just enough of the
|
||||
ASCII protocol) that can be "wedged" — control listener killed while FTP
|
||||
stays up, exactly the failure signature from the Feb 2026 investigation.
|
||||
A local HTTP server stands in for the relay webhook; hitting it "power
|
||||
cycles" the fake device.
|
||||
|
||||
Runs against a throwaway SQLite DB in a temp directory — does not touch
|
||||
data/slmm.db.
|
||||
|
||||
Usage:
|
||||
python3 test_wedge_recovery.py
|
||||
|
||||
Exit code 0 = all tests passed.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from pathlib import Path
|
||||
|
||||
# --- Run from a temp dir so app.database creates a throwaway DB ------------
|
||||
REPO_ROOT = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
WORKDIR = tempfile.mkdtemp(prefix="slmm-wedge-test-")
|
||||
os.chdir(WORKDIR)
|
||||
|
||||
# Confirm fast in tests
|
||||
os.environ.setdefault("WEDGE_CONFIRM_FAILURES", "2")
|
||||
|
||||
from app.database import Base, engine, SessionLocal # noqa: E402
|
||||
from app.models import NL43Config, NL43Status # noqa: E402
|
||||
from app.recovery import classify_failure, probe_tcp, recovery_manager # noqa: E402
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
PASS = 0
|
||||
FAIL = 0
|
||||
|
||||
|
||||
def check(name: str, condition: bool, detail: str = ""):
|
||||
global PASS, FAIL
|
||||
if condition:
|
||||
PASS += 1
|
||||
print(f" ✓ {name}")
|
||||
else:
|
||||
FAIL += 1
|
||||
print(f" ✗ {name} {('— ' + detail) if detail else ''}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake NL-43 device
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class FakeNL43:
|
||||
"""Minimal NL-43: control port speaks the ASCII protocol, FTP port
|
||||
just accepts. Can be wedged (control listener killed, FTP alive)."""
|
||||
|
||||
def __init__(self):
|
||||
self.control_server = None
|
||||
self.ftp_server = None
|
||||
self.control_port = None
|
||||
self.ftp_port = None
|
||||
self.measurement_state = "Stop"
|
||||
self.index = 1
|
||||
self.commands = [] # every command line received
|
||||
|
||||
async def start(self):
|
||||
self.ftp_server = await asyncio.start_server(self._handle_ftp, "127.0.0.1", 0)
|
||||
self.ftp_port = self.ftp_server.sockets[0].getsockname()[1]
|
||||
await self.start_control()
|
||||
|
||||
async def start_control(self):
|
||||
self.control_server = await asyncio.start_server(
|
||||
self._handle_control, "127.0.0.1", 0 if self.control_port is None else self.control_port
|
||||
)
|
||||
self.control_port = self.control_server.sockets[0].getsockname()[1]
|
||||
|
||||
async def wedge(self):
|
||||
"""Kill the control listener (existing + new connections die). FTP stays up."""
|
||||
if self.control_server:
|
||||
self.control_server.close()
|
||||
await self.control_server.wait_closed()
|
||||
self.control_server = None
|
||||
|
||||
async def power_cycle(self, boot_delay: float = 2.0):
|
||||
"""Simulate a power cycle: everything down, then back up after boot_delay.
|
||||
Power loss stops any running measurement."""
|
||||
await self.wedge()
|
||||
self.measurement_state = "Stop"
|
||||
await asyncio.sleep(boot_delay)
|
||||
await self.start_control()
|
||||
|
||||
async def stop(self):
|
||||
for srv in (self.control_server, self.ftp_server):
|
||||
if srv:
|
||||
srv.close()
|
||||
await srv.wait_closed()
|
||||
self.control_server = None
|
||||
self.ftp_server = None
|
||||
|
||||
async def _handle_ftp(self, reader, writer):
|
||||
try:
|
||||
writer.write(b"220 Connection Ready\r\n")
|
||||
await writer.drain()
|
||||
await reader.read(1024)
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
writer.close()
|
||||
|
||||
async def _handle_control(self, reader, writer):
|
||||
try:
|
||||
while True:
|
||||
line = await reader.readuntil(b"\n")
|
||||
cmd = line.decode(errors="ignore").strip()
|
||||
if not cmd:
|
||||
continue
|
||||
self.commands.append(cmd)
|
||||
writer.write(self._respond(cmd))
|
||||
await writer.drain()
|
||||
except (asyncio.IncompleteReadError, ConnectionResetError):
|
||||
pass
|
||||
finally:
|
||||
try:
|
||||
writer.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _respond(self, cmd: str) -> bytes:
|
||||
ok = b"R+0000\r\n"
|
||||
if cmd == "Measure?":
|
||||
return ok + f"{self.measurement_state}\r\n".encode()
|
||||
if cmd == "Measure,Start":
|
||||
self.measurement_state = "Start"
|
||||
return ok
|
||||
if cmd == "Measure,Stop":
|
||||
self.measurement_state = "Stop"
|
||||
return ok
|
||||
if cmd.startswith("Clock,"):
|
||||
return ok
|
||||
if cmd == "Store Name?":
|
||||
return ok + f"{self.index:04d}\r\n".encode()
|
||||
if cmd.startswith("Store Name,"):
|
||||
self.index = int(cmd.split(",")[1])
|
||||
return ok
|
||||
if cmd == "Overwrite?":
|
||||
return ok + b"None\r\n"
|
||||
if cmd == "DOD?":
|
||||
return ok + b"1,55.5,54.2,60.1,50.3,72.8\r\n"
|
||||
if cmd.startswith("Sleep Mode"):
|
||||
return ok + (b"Off\r\n" if cmd.endswith("?") else b"")
|
||||
return b"R+0001\r\n" # command error
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake relay webhook
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class FakeRelay:
|
||||
"""HTTP server standing in for a relay controller. A request to /cycle
|
||||
power-cycles the fake device."""
|
||||
|
||||
def __init__(self, device: FakeNL43, loop: asyncio.AbstractEventLoop):
|
||||
self.device = device
|
||||
self.loop = loop
|
||||
self.hits = 0
|
||||
relay = self
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def _handle(self):
|
||||
relay.hits += 1
|
||||
# Schedule the device power cycle on the asyncio loop
|
||||
asyncio.run_coroutine_threadsafe(relay.device.power_cycle(boot_delay=2.0), relay.loop)
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b"cycling")
|
||||
|
||||
do_GET = _handle
|
||||
do_POST = _handle
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
self.server = HTTPServer(("127.0.0.1", 0), Handler)
|
||||
self.port = self.server.server_address[1]
|
||||
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
return f"http://127.0.0.1:{self.port}/cycle"
|
||||
|
||||
def stop(self):
|
||||
self.server.shutdown()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def test_classifier(device: FakeNL43):
|
||||
print("\n[1] Failure classification")
|
||||
|
||||
# Healthy device
|
||||
c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
|
||||
check("healthy device classifies as ok", c["state"] == "ok", str(c))
|
||||
|
||||
# Wedged: control refused, FTP alive — the definitive signature
|
||||
await device.wedge()
|
||||
c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
|
||||
check("wedged device classifies as wedged", c["state"] == "wedged", str(c))
|
||||
check("wedge confidence is high (FTP alive)", c["confidence"] == "high", str(c))
|
||||
|
||||
# Wedged without FTP probe available
|
||||
c = await classify_failure("127.0.0.1", device.control_port, None)
|
||||
check("wedge without FTP probe is medium confidence",
|
||||
c["state"] == "wedged" and c["confidence"] == "medium", str(c))
|
||||
|
||||
# Offline: unroutable address times out (TEST-NET-1)
|
||||
c = await classify_failure("192.0.2.1", 2255, None)
|
||||
check("unreachable host classifies as offline", c["state"] == "offline", str(c))
|
||||
|
||||
await device.start_control()
|
||||
probe = await probe_tcp("127.0.0.1", device.control_port)
|
||||
check("control port back open after un-wedge", probe == "open", probe)
|
||||
|
||||
|
||||
async def test_webhook_recovery(device: FakeNL43, relay: FakeRelay):
|
||||
print("\n[2] End-to-end webhook recovery (wedge → relay cycle → reconnect → resume)")
|
||||
|
||||
db = SessionLocal()
|
||||
cfg = NL43Config(
|
||||
unit_id="TEST-NL43",
|
||||
host="127.0.0.1",
|
||||
tcp_port=device.control_port,
|
||||
ftp_port=device.ftp_port,
|
||||
tcp_enabled=True,
|
||||
ftp_enabled=True,
|
||||
auto_recovery_enabled=True,
|
||||
reset_backend="webhook",
|
||||
reset_webhook_url=relay.url,
|
||||
reset_webhook_method="POST",
|
||||
recovery_boot_wait_seconds=10, # validator min; device boots in 2s
|
||||
recovery_reconnect_timeout=60,
|
||||
recovery_auto_resume=True,
|
||||
recovery_max_attempts=3,
|
||||
recovery_window_minutes=360,
|
||||
)
|
||||
db.add(cfg)
|
||||
# Device was measuring before the wedge
|
||||
status = NL43Status(unit_id="TEST-NL43", measurement_state="Start", consecutive_failures=2)
|
||||
db.add(status)
|
||||
db.commit()
|
||||
|
||||
# Wedge it
|
||||
device.measurement_state = "Start"
|
||||
await device.wedge()
|
||||
|
||||
classification = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
|
||||
check("pre-recovery classification is wedged", classification["state"] == "wedged")
|
||||
|
||||
started = recovery_manager.start_recovery(cfg, classification)
|
||||
check("recovery task started", started)
|
||||
|
||||
# Wait for the recovery task to finish (boot wait 10s + commands ≈ 20s)
|
||||
deadline = time.time() + 90
|
||||
while recovery_manager.is_active("TEST-NL43") and time.time() < deadline:
|
||||
await asyncio.sleep(1)
|
||||
check("recovery task completed", not recovery_manager.is_active("TEST-NL43"))
|
||||
|
||||
check("relay webhook was hit", relay.hits == 1, f"hits={relay.hits}")
|
||||
|
||||
db.expire_all()
|
||||
status = db.query(NL43Status).filter_by(unit_id="TEST-NL43").first()
|
||||
check("recovery_state is idle", status.recovery_state == "idle", status.recovery_state)
|
||||
check("connection_state is ok", status.connection_state == "ok", status.connection_state)
|
||||
check("consecutive_failures reset", status.consecutive_failures == 0)
|
||||
check("last_recovery_result records success",
|
||||
status.last_recovery_result and "Recovered" in status.last_recovery_result,
|
||||
str(status.last_recovery_result))
|
||||
|
||||
check("measurement resumed on device", device.measurement_state == "Start", device.measurement_state)
|
||||
check("device received Measure,Start", "Measure,Start" in device.commands)
|
||||
check("start cycle used overwrite protection", "Overwrite?" in device.commands)
|
||||
|
||||
db.close()
|
||||
|
||||
|
||||
async def test_gating(device: FakeNL43):
|
||||
print("\n[3] Recovery gating (confirmation threshold + attempt limit)")
|
||||
|
||||
db = SessionLocal()
|
||||
cfg = NL43Config(
|
||||
unit_id="TEST-GATE",
|
||||
host="127.0.0.1",
|
||||
tcp_port=device.control_port,
|
||||
ftp_port=device.ftp_port,
|
||||
tcp_enabled=True,
|
||||
auto_recovery_enabled=True,
|
||||
reset_backend="webhook",
|
||||
reset_webhook_url="http://127.0.0.1:1/unreachable", # fails fast
|
||||
recovery_max_attempts=1,
|
||||
recovery_window_minutes=360,
|
||||
)
|
||||
db.add(cfg)
|
||||
status = NL43Status(unit_id="TEST-GATE", consecutive_failures=1)
|
||||
db.add(status)
|
||||
db.commit()
|
||||
|
||||
classification = {"state": "wedged", "confidence": "high", "tcp_probe": "refused", "ftp_probe": "open"}
|
||||
|
||||
# Below confirmation threshold (1 < 2) — no recovery
|
||||
started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
|
||||
check("below threshold does not trigger", not started)
|
||||
|
||||
# At threshold — triggers (and fails fast on the dead webhook)
|
||||
status.consecutive_failures = 2
|
||||
db.commit()
|
||||
started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
|
||||
check("at threshold triggers recovery", started)
|
||||
|
||||
deadline = time.time() + 30
|
||||
while recovery_manager.is_active("TEST-GATE") and time.time() < deadline:
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
db.expire_all()
|
||||
status = db.query(NL43Status).filter_by(unit_id="TEST-GATE").first()
|
||||
check("failed webhook marks recovery failed", status.recovery_state == "failed", status.recovery_state)
|
||||
check("failure recorded in result",
|
||||
status.last_recovery_result and "FAILED" in status.last_recovery_result,
|
||||
str(status.last_recovery_result))
|
||||
|
||||
# Attempt limit (max 1 per window) — second attempt blocked
|
||||
started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
|
||||
check("attempt limit blocks repeat recovery", not started)
|
||||
|
||||
# Disabled flag blocks everything
|
||||
cfg.auto_recovery_enabled = False
|
||||
db.commit()
|
||||
started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
|
||||
check("auto_recovery_enabled=False blocks recovery", not started)
|
||||
|
||||
db.close()
|
||||
|
||||
|
||||
async def main():
|
||||
print(f"Wedge recovery test — temp workdir: {WORKDIR}")
|
||||
|
||||
device = FakeNL43()
|
||||
await device.start()
|
||||
relay = FakeRelay(device, asyncio.get_running_loop())
|
||||
print(f"Fake NL-43: control=127.0.0.1:{device.control_port}, ftp=127.0.0.1:{device.ftp_port}")
|
||||
print(f"Fake relay webhook: {relay.url}")
|
||||
|
||||
try:
|
||||
await test_classifier(device)
|
||||
await test_webhook_recovery(device, relay)
|
||||
await test_gating(device)
|
||||
finally:
|
||||
relay.stop()
|
||||
await device.stop()
|
||||
|
||||
print(f"\n{'='*50}\nResults: {PASS} passed, {FAIL} failed")
|
||||
return FAIL == 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
ok = asyncio.run(main())
|
||||
sys.exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user