10 Commits

Author SHA1 Message Date
serversdown b9bf4ea9e6 quick nl52 usb probe 2026-06-08 00:45:07 +00:00
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
serversdown 450509d210 stop tracking dev runtime data 2026-03-12 22:46:37 +00:00
serversdown fefa9eace8 chore: gitignore clean up 2026-03-12 21:34:14 +00:00
serversdown 98a8d357e5 chore: data-dev folder added to gitignore 2026-03-12 21:33:43 +00:00
claude 0a7422eceb Merge branch 'dev-persistent' of ssh://10.0.0.2:2222/serversdown/slmm into dev-persistent 2026-03-12 20:26:56 +00:00
claude 996b993cb9 chore: gitignore dev data 2026-03-12 20:26:53 +00:00
claude 01337696b3 feat: add connection pool status logging every 15 minutes 2026-02-19 15:09:50 +00:00
claude a302fd15d4 fix: change debug logs to info level for connection pool events 2026-02-19 06:04:34 +00:00
claude af5ecc1a92 fix: improve connection pool idle and max age checks to allow disabling 2026-02-19 01:25:01 +00:00
11 changed files with 1798 additions and 5 deletions
+1
View File
@@ -1,5 +1,6 @@
/manuals/ /manuals/
/data/ /data/
/data-dev/
/SLM-stress-test/stress_test_logs/ /SLM-stress-test/stress_test_logs/
/SLM-stress-test/tcpdump-runs/ /SLM-stress-test/tcpdump-runs/
+20
View File
@@ -5,6 +5,26 @@ All notable changes to SLMM (Sound Level Meter Manager) will be documented in th
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
#### Wedge Detection & Automatic Recovery
- **Failure classification** - Poll failures are now probed and classified: `wedged` (control port REFUSES — device stack alive, listener gone), `offline` (timeout — power/cellular loss), or `ok` (transient). The wedge signature matches the Feb 2026 investigation: refused on 2255 while FTP/21 accepts is definitive (high confidence); refused alone is medium confidence.
- **RecoveryManager** (`app/recovery.py`) - Per-device recovery state machine: confirm wedge → trigger reset → wait for boot → reconnect → resume measurement (via existing `start_cycle` with overwrite protection) → log the full incident to device logs (category `RECOVERY`).
- **Pluggable reset backends**:
- `manual` (default) - logs/flags the wedge for a human; still auto-resumes measurement when the device returns
- `webhook` - HTTP GET/POST to a relay controller (Pi GPIO relay, Shelly, Tasmota, etc.) that power-cycles the NL-43
- **Safety guards** - per-device `auto_recovery_enabled` master switch (default off), confirmation threshold (`WEDGE_CONFIRM_FAILURES`, default 2 consecutive failures), windowed attempt limit (default 2 per 6 h), one recovery at a time per device.
- **New endpoints**:
- `GET /api/nl43/{unit_id}/recovery/status` - connection/recovery state, wedge count, last result
- `POST /api/nl43/{unit_id}/recovery/probe` - diagnostic port probe + classification (no action taken)
- `POST /api/nl43/{unit_id}/recovery/trigger` - manually start recovery (bypasses confirmation threshold)
- **Config fields** (via `PUT /{unit_id}/config`): `auto_recovery_enabled`, `reset_backend`, `reset_webhook_url`, `reset_webhook_method`, `recovery_boot_wait_seconds`, `recovery_reconnect_timeout`, `recovery_auto_resume`, `recovery_max_attempts`, `recovery_window_minutes`
- **Status fields**: `connection_state`, `last_wedge_at`, `wedge_count`, `recovery_state`, `last_recovery_at`, `last_recovery_result`
- **Migration**: `migrate_add_recovery_fields.py`
- **Test suite**: `test_wedge_recovery.py` - fake wedge-able NL-43 + fake relay webhook; covers classification, end-to-end recovery with measurement resume, and gating (23 assertions)
## [0.3.0] - 2026-02-17 ## [0.3.0] - 2026-02-17
### Added ### Added
@@ -0,0 +1,403 @@
# NL-43 + RX55 TCP “Wedge” Investigation (2255 Refusal) — Full Log & Next Steps
**Last updated:** 2026-02-18
**Owner:** Brian / serversdown
**Context:** Terra-View / SLMM / field-deployed Rion NL-43 behind Sierra Wireless RX55
---
## 0) What this document is
This is a **comprehensive, chronological** record of the debugging we did to isolate a failure where the **NL-43s TCP control port (2255) eventually stops accepting connections** (“wedges”), while other services (notably FTP/21) remain reachable.
This is written to be fed back into future troubleshooting, so it intentionally includes the **full reasoning chain, experiments, commands, packet evidence, and conclusions**.
---
## 1) Architecture (as tested)
### Network path
- **Server (SLMM host):** `10.0.0.40`
- **RX55 WAN IP:** `63.45.161.30`
- **RX55 LAN subnet:** `192.168.1.0/24`
- **RX55 LAN gateway:** `192.168.1.1`
- **NL-43 LAN IP:** `192.168.1.10` (confirmed via ARP OUI + ping; see LAN validation)
### RX55 details
- **Sierra Wireless RX55**
- **OS:** 5.2
- **Firmware:** `01.14.24.00`
- **Carrier:** Verizon LTE (Band 66)
### Port forwarding rules (RX55)
- **WAN:2255 → NL-43:2255** (NL-43 TCP control)
- **WAN:21 → NL-43:21** (NL-43 FTP control)
You also experimented with additional forwards:
- **WAN:2253 → NL-43:2255** (test)
- **WAN:2253 → NL-43:2253** (test)
- **WAN:4450 → NL-43:4450** (test)
**Important:** Rule “Input zone / interface” was set to **WAN-NAT**, and Source IP left as **Any IPv4**. This is correct for inbound port-forward behavior on Sierra OS 5.x.
---
## 2) Original problem statement (the “wedge”)
After running for hours, the NL-43 becomes unreachable over TCP control.
### Symptom signature (WAN-side)
- Client attempts to connect to `63.45.161.30:2255`
- Instead of timing out, the client gets **connection refused** quickly.
- Packet-level: SYN from client → **RST,ACK** back (meaning active refusal vs silent drop)
### Critical operational behavior
- **Power cycling the NL-43 fixes it.**
- **Power cycling the RX55 does NOT fix it.**
- FTP sometimes remains available even while TCP control (2255) is dead.
This combination is what forced us to determine whether:
- The RX55 is rejecting connections, OR
- The NL-43 is no longer listening on 2255, OR
- Something about the RX55 path triggers the NL-43s control listener to die.
---
## 3) Event timeline evidence (SLMM logs)
A concrete wedge window was observed on **2026-02-18**:
- 10:55:46 AM — Poll success (Start)
- 11:00:28 AM — Measurement STOPPED (scheduled stop/download cycle succeeded)
- 11:55:50 AM — Poll success (Stop)
- 12:55:55 PM — Poll success (Stop)
- **1:55:58 PM — Poll failed (attempt 1/3): Errno 111 (connection refused)**
- 2:56:02 PM — Poll failed (attempt 2/3): Errno 111 (connection refused)
Key interpretation:
- The wedge occurred sometime between **12:55 and 1:55**.
- The failure type is **refused**, not timeout.
---
## 4) Early hypotheses (before proof)
We considered two main buckets:
### A) NL-43-side failure (most suspicious)
- NL-43 TCP control service crashes / exits / unbinds from 2255
- socket leak / accept backlog exhaustion
- “single control session allowed” and it gets stuck thinking a session is active
- mode/service manager bug (service restart fails after other activities)
- firmware bug in TCP daemon
### B) RX55-side failure (possible trigger / less likely once FTP works)
- NAT/forwarding table corruption
- firewall behavior
- helper/ALG interference
- MSS/MTU weirdness causing edge-case behavior
- session churn behavior causing downstream issues
---
## 5) Key experiments and what they proved
### 5.1) LAN-only stability test (No RX55 path)
**Test:** NL-43 tested directly on LAN (no modem path involved).
- Ran **24+ hours**
- Scheduler start/stop cycles worked
- Stress test: **500 commands @ 1/sec** → no failure
- Response time trend decreased (not degrading)
**Result:** The NL-43 appears stable in a “pure LAN” environment.
**Interpretation:** The trigger is likely related to the RX55/WAN environment, connection patterns, or service switching patterns—not just simple uptime.
---
### 5.2) Port-forward behavior: timeout vs refused (RX55 behavior characterization)
You observed:
- **If a WAN port is NOT forwarded (no rule):** connecting to that port **times out** (silent drop)
- **If a WAN port IS forwarded to NL-43 but nothing listens:** it **actively refuses** (RST)
Concrete example:
- Port **4450** with no rule → timeout
- Port **4450 → NL-43:4450** rule created → connection refused
**Interpretation:** This confirms the RX55 is actually forwarding packets to the NL-43 when a rule exists. “Refused” is consistent with the NL-43 (or RX55 relay behavior) responding quickly because the packet reached the target.
Important nuance:
- A “refused” on forwarded ports does **not** automatically prove the NL-43 is the one generating RST, because NAT hides the inside host and the RX55 could reject on behalf of an unreachable target. We needed a LAN-side proof test to close the loop.
---
### 5.3) UDP test confusion (and resolution)
You ran:
```bash
nc -vzu 63.45.161.30 2255
nc -vz 63.45.161.30 2255
```
Observed:
- UDP: “succeeded”
- TCP: “connection refused”
Resolution:
- UDP has **no handshake**. netcat prints “succeeded” if it doesnt immediately receive an ICMP unreachable. It does **not** mean a UDP service exists.
- TCP refused is meaningful: a RST implies “no listener” or “actively rejected.”
**Net effect:** UDP test did not change the diagnosis.
---
### 5.4) Packet capture proof (WAN-side)
You captured a Wireshark/tcpdump summary with these key patterns:
#### Port 2255 (TCP control)
Example:
- `10.0.0.40 → 63.45.161.30:2255` SYN
- `63.45.161.30 → 10.0.0.40` **RST, ACK** within ~50ms
This happened repeatedly.
#### Port 2253 (test port)
Multiple SYN attempts to 2253 showed **retransmissions and no response**, i.e., **silent drop** (consistent with no rule or not forwarded at that moment).
#### Port 21 (FTP)
Clean 3-way handshake:
- SYN → SYN/ACK → ACK
Then:
- FTP server banner: `220 Connection Ready`
Then:
- `530 Not logged in` (because SLMM was sending non-FTP “requests” as an experiment)
Session closes cleanly.
**Key takeaway from capture:**
- TCP transport to NL-43 via RX55 is definitely working (port 21 proves it).
- Port 2255 is being actively refused.
This strongly suggested “2255 listener is gone,” but still didnt fully prove whether the refusal was generated internally by NL-43 or by RX55 on behalf of NL-43.
---
## 6) The decisive experiment: LAN-side test while wedged (final proof)
Because the RX55 does not offer SSH, the plan was to test from **inside the LAN behind the RX55**.
### 6.1) Physical LAN tap setup
Constraint:
- NL-43 has only one Ethernet port.
Solution:
- Insert an unmanaged switch:
- RX55 LAN → switch
- NL-43 → switch
- Windows 10 laptop → switch
This creates a shared L2 segment where the laptop can test NL-43 directly.
### 6.2) Windows LAN validation
On the Windows laptop:
- `ipconfig` showed:
- IP: `192.168.1.100`
- Gateway: `192.168.1.1` (RX55)
- Initial `arp -a` only showed RX55, not NL-43.
You then:
- pinged likely host addresses and discovered NL-43 responds on **192.168.1.10**
- `arp -a` then showed:
- `192.168.1.10 → 00-10-50-14-0a-d8`
- OUI `00-10-50` recognized as **Rion** (matches NL-43)
So LAN identities were confirmed:
- RX55: `192.168.1.1`
- NL-43: `192.168.1.10`
### 6.3) The LAN port tests (the smoking gun)
From Windows:
```powershell
Test-NetConnection -ComputerName 192.168.1.10 -Port 2255
Test-NetConnection -ComputerName 192.168.1.10 -Port 21
```
Results (while the unit was “wedged” from the WAN perspective):
- **2255:** `TcpTestSucceeded : False`
- **21:** `TcpTestSucceeded : True`
**Conclusion (PROVEN):**
- The NL-43 is reachable on the LAN
- FTP port 21 is alive
- **The NL-43 is NOT listening on TCP port 2255**
- Therefore the RX55 is not the root cause of the refusal. The WAN refusal is consistent with the NL-43 having no listener on 2255.
This is now settled.
---
## 7) What we learned (final conclusions)
### 7.1) RX55 innocence (for this failure mode)
The RX55 is not “randomly rejecting” or “breaking TCP” in the way originally feared.
It successfully forwards and supports TCP to the NL-43 on port 21, and the LAN-side test proves the 2255 failure exists *even without NAT/WAN involvement*.
### 7.2) NL-43 control listener failure
The NL-43s TCP control service (port 2255) stops listening while:
- the device remains alive
- the LAN stack remains alive (ping)
- FTP remains alive (port 21)
This looks like one of:
- control daemon crash/exit
- service unbind
- stuck service state (e.g., “busy” / “session active forever”)
- resource leak (sockets/file descriptors) specific to the control service
- firmware service manager bug (start/stop of services fails after certain sequences)
---
## 8) Additional constraint discovered: “Web App mode” conflicts
You noted an important operational constraint:
> Turning on the web app disables other interfaces like TCP and FTP.
Meaning the NL-43 appears to have mutually exclusive service/mode behavior (or at least serious conflicts). That matters because:
- If any workflow toggles modes (explicitly or implicitly), it could destabilize the service lifecycle.
- It reduces the possibility of using “web UI toggle” as an easy remote recovery mechanism **if** it disables the services needed.
We have not yet run a controlled long test to determine whether:
- mode switching contributes directly to the 2255 listener dying, OR
- it happens even in a pure TCP-only mode with no switching.
---
## 9) Immediate operational decision (field tomorrow)
Because the device is needed in the field immediately, you chose:
- **Old-school manual deployment**
- **Manual SD card downloads**
- Avoid reliance on 2255/TCP control and remote workflows for now.
**Important operational note:**
The 2255 listener dying does not necessarily stop the NL-43 from measuring; it primarily breaks remote control/polling. Manual SD workflow sidesteps the entire remote control dependency.
---
## 10) Whats next (future work — when the unit is back)
Because long tests cant be run before tomorrow, the plan is to resume in a few weeks with controlled experiments designed to isolate the trigger and develop an operational mitigation.
### 10.1) Controlled experiment matrix (recommended)
Run each test for 2472 hours, or until wedge occurs, and record:
- number of TCP connects
- whether connections are persistent
- whether FTP is used
- whether any mode toggling is performed
- time-to-wedge
#### Test A — TCP-only (ideal baseline)
- TCP control only (2255)
- **True persistent connection** (open once, keep forever)
- No FTP
- No web mode toggling
Outcome interpretation:
- If stable: connection churn and/or FTP/mode switching is the trigger.
- If wedges anyway: pure 2255 daemon leak/bug.
#### Test B — TCP with connection churn
- Same as A but intentionally reconnect on a schedule (current SLMM behavior)
- No FTP
Outcome:
- If this wedges but A doesnt: churn is the trigger.
#### Test C — FTP activity + TCP
- Introduce scheduled FTP sessions (downloads) while using TCP control
- Observe whether wedge correlates with FTP use or with post-download periods.
Outcome:
- If wedge correlates with FTP, suspect internal service lifecycle conflict.
#### Test D — Web mode interaction (only if safe/possible)
- Evaluate what toggling web mode does to TCP/FTP services.
- Determine if any remote-safe “soft reset” exists.
---
## 11) Mitigation options (ranked)
### Option 1 — Make SLMM truly persistent (highest probability of success)
If the NL-43 wedges due to session churn or leaked socket states, the best mitigation is:
- Open one TCP socket per device
- Keep it open indefinitely
- Use OS keepalive
- Do **not** rotate connections on timers
- Reconnect only when the socket actually dies
This reduces:
- connect/close cycles
- NAT edge-case exposure
- resource churn inside NL-43
### Option 2 — Service “soft reset” (if possible without disabling required services)
If there exists any way to restart the 2255 service without power cycling:
- LAN TCP toggle (if it doesnt require web mode)
- any “restart comms” command (unknown)
- any maintenance menu sequence
then SLMM could:
- detect wedge
- trigger soft reset
- recover automatically
Current constraint: web app mode appears to disable other services, so this may not be viable.
### Option 3 — Hardware watchdog power cycle (industrial but reliable)
If this is a firmware bug with no clean workaround:
- Add a remotely controlled relay/power switch
- On wedge detection, power-cycle NL-43 automatically
- Optionally schedule a nightly power cycle to prevent leak accumulation
This is “field reality” and often the only long-term move with embedded devices.
### Option 4 — Vendor escalation (Rion)
You now have excellent evidence:
- LAN-side proof: 2255 dead while 21 alive
- WAN packet evidence
- clear isolation of RX55 innocence
This is strong enough to send to Rion support as a firmware defect report.
---
## 12) Repro “wedge bundle” checklist (for future captures)
When the wedge happens again, capture these before power cycling:
1) From server:
- `nc -vz 63.45.161.30 2255` (expect refused)
- `nc -vz 63.45.161.30 21` (expect success if FTP alive)
2) From LAN side (via switch/laptop):
- `Test-NetConnection 192.168.1.10 -Port 2255`
- `Test-NetConnection 192.168.1.10 -Port 21`
3) Optional: packet capture around the refused attempt.
4) Record:
- last successful poll timestamp
- last FTP session timestamp
- any scheduled start/stop/download cycles near wedge time
- SLMM connection reuse/rotation settings in effect
---
## 13) Final, current-state summary (as of 2026-02-18)
- The issue is **NOT** the RX55 rejecting inbound connections.
- The NL-43 is **alive**, reachable on LAN, and FTP works.
- The NL-43s **TCP control listener on 2255 stops listening** while the device remains otherwise healthy.
- The wedge can occur hours after successful operations.
- The unit is needed in the field immediately, so investigation pauses.
- Next phase: controlled tests to isolate trigger + implement mitigation (persistent socket or watchdog reset).
---
## 14) Notes / misc observations
- The Wireshark trace showed repeated FTP sessions were opened and closed cleanly, but SLMMs “FTP requests” were not valid FTP (causing `530 Not logged in`). That was part of experimentation, not a normal workflow.
- UDP “success” via netcat is not meaningful because UDP has no handshake; it simply indicates no ICMP unreachable was returned.
---
**End of document.**
+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python3
"""
RION NL-42 / NL-52 USB serial probe — zero dependencies (stdlib termios only).
The NL-52's USB port enumerates as a virtual COM port ("RION USB to RS232C
Converter"). On Linux that is almost always handled by an in-kernel USB-serial
driver (ftdi_sio / cp210x / ch341) which creates /dev/ttyUSB*. This script
opens that port and sends a few harmless REQUEST commands (no settings are
changed) to confirm two-way communication.
Protocol (NL-42/NL-52 Serial Interface Manual 55779):
Request: "<Command>?" + CRLF
Setting: "$<Command>,<param>" + CRLF (NOT used here — read-only probe)
Reply: result code "R+0000" + CRLF, then data line(s) for requests.
Before running:
1. On the meter: MENU -> I/O -> Communication Interface -> "USB"
(set this BEFORE plugging in the cable).
2. Connect a generic USB-A -> mini-B cable directly (no hub).
3. Find the port: ls -l /dev/ttyUSB* (and `dmesg | tail` after plugging in)
Usage:
python3 nl52_usb_probe.py # defaults to /dev/ttyUSB0
python3 nl52_usb_probe.py --port /dev/ttyUSB0 --baud 115200
python3 nl52_usb_probe.py --cmd "System Version?" --cmd "Clock?"
"""
import argparse
import os
import select
import sys
import termios
import time
# Safe, read-only probe commands (all are pure requests).
DEFAULT_COMMANDS = [
"System Version?", # firmware version — proves the link end-to-end
"Clock?", # current date/time
"SD Card Free Size?",
"DOD?", # snapshot of currently displayed values
]
BAUD_CONSTANTS = {
9600: termios.B9600,
19200: termios.B19200,
38400: termios.B38400,
57600: termios.B57600,
115200: termios.B115200,
}
def open_serial(port: str, baud: int) -> int:
"""Open a serial port in raw 8N1, no flow control. Returns an fd."""
if baud not in BAUD_CONSTANTS:
raise ValueError(f"Unsupported baud {baud}; choose from {sorted(BAUD_CONSTANTS)}")
fd = os.open(port, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
attrs = termios.tcgetattr(fd)
iflag, oflag, cflag, lflag, ispeed, ospeed, cc = attrs
# Raw mode
iflag = 0
oflag = 0
lflag = 0
# 8 data bits, enable receiver, ignore modem control lines
cflag = termios.CS8 | termios.CREAD | termios.CLOCAL
# (no PARENB = no parity, no CSTOPB = 1 stop bit, no CRTSCTS = no flow control)
bconst = BAUD_CONSTANTS[baud]
ispeed = bconst
ospeed = bconst
termios.tcsetattr(fd, termios.TCSANOW, [iflag, oflag, cflag, lflag, ispeed, ospeed, cc])
termios.tcflush(fd, termios.TCIOFLUSH)
return fd
def send(fd: int, line: str):
os.write(fd, (line + "\r\n").encode("ascii"))
def read_reply(fd: int, timeout: float = 3.0) -> bytes:
"""Read whatever arrives within `timeout` seconds (idle-gap terminated)."""
buf = bytearray()
deadline = time.time() + timeout
while time.time() < deadline:
r, _, _ = select.select([fd], [], [], 0.3)
if r:
try:
chunk = os.read(fd, 4096)
except BlockingIOError:
continue
if chunk:
buf.extend(chunk)
# Once we've seen a CRLF and there's a brief idle, stop early
deadline = min(deadline, time.time() + 0.4)
return bytes(buf)
def main():
ap = argparse.ArgumentParser(description="RION NL-42/NL-52 USB serial probe")
ap.add_argument("--port", default="/dev/ttyUSB0")
ap.add_argument("--baud", type=int, default=115200,
help="USB CDC usually ignores baud, but RS-232C needs a match")
ap.add_argument("--cmd", action="append", dest="cmds",
help="Override probe command(s); repeatable")
ap.add_argument("--timeout", type=float, default=3.0)
args = ap.parse_args()
commands = args.cmds or DEFAULT_COMMANDS
if not os.path.exists(args.port):
print(f"[!] {args.port} does not exist.")
print(" Plug in the meter (Comm Interface = USB) and check: ls -l /dev/ttyUSB*")
print(" Also check the kernel saw it: dmesg | tail -20")
return 2
try:
fd = open_serial(args.port, args.baud)
except PermissionError:
print(f"[!] Permission denied on {args.port}.")
print(" Add yourself to the 'dialout' group, or run with sudo:")
print(f" sudo usermod -aG dialout $USER (then log out/in)")
return 2
except Exception as e:
print(f"[!] Could not open {args.port}: {e}")
return 2
print(f"[*] Opened {args.port} @ {args.baud} 8N1 (raw, no flow control)")
print(f"[*] Sending {len(commands)} read-only request command(s)\n")
ok = 0
try:
for cmd in commands:
send(fd, cmd)
reply = read_reply(fd, args.timeout)
decoded = reply.decode("ascii", errors="replace").replace("\r", "\\r").replace("\n", "\\n\n")
if reply:
ok += 1
print(f" > {cmd}")
for line in decoded.splitlines():
print(f" {line}")
else:
print(f" > {cmd}")
print(f" (no response within {args.timeout}s)")
print()
time.sleep(1.0) # NL-series likes >=1s between commands
finally:
os.close(fd)
if ok == 0:
print("[!] No responses. Things to check:")
print(" - Meter's Communication Interface is set to USB (not RS-232C)")
print(" - ECO / Sleep mode is OFF (both disable the comm interface)")
print(" - Right port (try other /dev/ttyUSB* or /dev/ttyACM*)")
print(" - For RS-232C path, baud must match the meter's setting")
return 1
print(f"[OK] {ok}/{len(commands)} commands answered — two-way comms confirmed.")
return 0
if __name__ == "__main__":
sys.exit(main())
+76
View File
@@ -17,6 +17,7 @@ from app.database import SessionLocal
from app.models import NL43Config, NL43Status from app.models import NL43Config, NL43Status
from app.services import NL43Client, persist_snapshot, sync_measurement_start_time_from_ftp from app.services import NL43Client, persist_snapshot, sync_measurement_start_time_from_ftp
from app.device_logger import log_device_event, cleanup_old_logs from app.device_logger import log_device_event, cleanup_old_logs
from app.recovery import classify_failure, recovery_manager
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -38,6 +39,7 @@ class BackgroundPoller:
self._running = False self._running = False
self._logger = logger self._logger = logger
self._last_cleanup = None # Track last log cleanup time self._last_cleanup = None # Track last log cleanup time
self._last_pool_log = None # Track last connection pool heartbeat log
async def start(self): async def start(self):
"""Start the background polling task.""" """Start the background polling task."""
@@ -89,6 +91,24 @@ class BackgroundPoller:
except Exception as e: except Exception as e:
self._logger.warning(f"Log cleanup failed: {e}") self._logger.warning(f"Log cleanup failed: {e}")
# Log connection pool status every 15 minutes
try:
now = datetime.utcnow()
if self._last_pool_log is None or (now - self._last_pool_log).total_seconds() > 900:
from app.services import _connection_pool
stats = _connection_pool.get_stats()
conns = stats.get("connections", {})
if conns:
for key, c in conns.items():
self._logger.info(
f"[POOL] {key} — age={c['age_seconds']}s idle={c['idle_seconds']}s alive={c['alive']}"
)
else:
self._logger.info("[POOL] No active connections in pool")
self._last_pool_log = now
except Exception as e:
self._logger.warning(f"Pool status log failed: {e}")
# Calculate dynamic sleep interval # Calculate dynamic sleep interval
sleep_time = self._calculate_sleep_interval() sleep_time = self._calculate_sleep_interval()
self._logger.debug(f"Sleeping for {sleep_time} seconds until next poll cycle") self._logger.debug(f"Sleeping for {sleep_time} seconds until next poll cycle")
@@ -212,6 +232,7 @@ class BackgroundPoller:
status.consecutive_failures = 0 status.consecutive_failures = 0
status.last_success = datetime.utcnow() status.last_success = datetime.utcnow()
status.last_error = None status.last_error = None
status.connection_state = "ok"
db.commit() db.commit()
self._logger.info(f"✓ Successfully polled {unit_id}") self._logger.info(f"✓ Successfully polled {unit_id}")
@@ -303,6 +324,61 @@ class BackgroundPoller:
db.commit() db.commit()
# Classify the failure: wedge (listener dead, host alive) vs offline
await self._classify_and_recover(cfg, status, db)
async def _classify_and_recover(self, cfg: NL43Config, status: NL43Status, db: Session):
"""Probe the device to classify a poll failure and trigger wedge recovery.
The wedge signature (see SLM-stress-test investigation doc): control
port REFUSES connections (RST = host stack alive, listener gone) while
the device is otherwise up. A timeout means offline (power/cell loss).
"""
unit_id = cfg.unit_id
# Skip while a recovery for this device is already running — its own
# probes manage state until it finishes.
if recovery_manager.is_active(unit_id):
return
try:
classification = await classify_failure(
cfg.host,
cfg.tcp_port,
ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
)
except Exception as e:
self._logger.warning(f"Failure classification error for {unit_id}: {e}")
return
new_state = classification["state"]
prev_state = status.connection_state or "ok"
if new_state != prev_state:
status.connection_state = new_state
if new_state == "wedged":
status.last_wedge_at = datetime.utcnow()
status.wedge_count = (status.wedge_count or 0) + 1
log_device_event(
unit_id, "ERROR", "RECOVERY",
f"WEDGE detected: control port {cfg.tcp_port} refused "
f"(tcp_probe={classification['tcp_probe']}, ftp_probe={classification['ftp_probe']}, "
f"confidence={classification['confidence']})",
db,
)
self._logger.error(f"Device {unit_id} classified as WEDGED ({classification})")
elif new_state == "offline":
log_device_event(
unit_id, "WARNING", "RECOVERY",
f"Device classified OFFLINE (tcp_probe={classification['tcp_probe']}) - "
f"not a wedge, no reset will be triggered",
db,
)
db.commit()
if new_state == "wedged":
recovery_manager.maybe_start_recovery(cfg, status, classification, db)
def _calculate_sleep_interval(self) -> int: def _calculate_sleep_interval(self) -> int:
""" """
Calculate the next sleep interval based on all device poll intervals. Calculate the next sleep interval based on all device poll intervals.
+19
View File
@@ -23,6 +23,17 @@ class NL43Config(Base):
poll_interval_seconds = Column(Integer, nullable=True, default=60) # Polling interval (10-3600 seconds) poll_interval_seconds = Column(Integer, nullable=True, default=60) # Polling interval (10-3600 seconds)
poll_enabled = Column(Boolean, default=True) # Enable/disable background polling for this device poll_enabled = Column(Boolean, default=True) # Enable/disable background polling for this device
# Wedge recovery configuration
auto_recovery_enabled = Column(Boolean, default=False) # Master switch for automatic wedge recovery
reset_backend = Column(String, default="manual") # Reset actuation: "manual" (notify only) or "webhook"
reset_webhook_url = Column(String, nullable=True) # URL that triggers a power cycle (relay controller)
reset_webhook_method = Column(String, default="POST") # HTTP method for webhook (GET or POST)
recovery_boot_wait_seconds = Column(Integer, default=90) # Wait after reset before reconnect attempts
recovery_reconnect_timeout = Column(Integer, default=180) # Max seconds to wait for device after boot wait
recovery_auto_resume = Column(Boolean, default=True) # Restart measurement if it was running pre-wedge
recovery_max_attempts = Column(Integer, default=2) # Max recovery attempts per window
recovery_window_minutes = Column(Integer, default=360) # Attempt-limit window (default 6 hours)
class NL43Status(Base): class NL43Status(Base):
""" """
@@ -57,6 +68,14 @@ class NL43Status(Base):
# FTP start time sync tracking # FTP start time sync tracking
start_time_sync_attempted = Column(Boolean, default=False) # True if FTP sync was attempted for current measurement start_time_sync_attempted = Column(Boolean, default=False) # True if FTP sync was attempted for current measurement
# Wedge detection and recovery tracking
connection_state = Column(String, default="ok") # ok | wedged | degraded | offline
last_wedge_at = Column(DateTime, nullable=True) # When the current/most recent wedge was detected
wedge_count = Column(Integer, default=0) # Lifetime count of detected wedges
recovery_state = Column(String, default="idle") # idle | resetting | waiting_boot | reconnecting | resuming | awaiting_manual_reset | failed
last_recovery_at = Column(DateTime, nullable=True) # When the last recovery attempt started
last_recovery_result = Column(Text, nullable=True) # Outcome summary of the last recovery attempt
class DeviceLog(Base): class DeviceLog(Base):
""" """
+482
View File
@@ -0,0 +1,482 @@
"""
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()
+143
View File
@@ -13,6 +13,7 @@ import asyncio
from app.database import get_db from app.database import get_db
from app.models import NL43Config, NL43Status from app.models import NL43Config, NL43Status
from app.services import NL43Client, persist_snapshot from app.services import NL43Client, persist_snapshot
from app.recovery import classify_failure, recovery_manager
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -53,6 +54,38 @@ class ConfigPayload(BaseModel):
poll_enabled: bool | None = None poll_enabled: bool | None = None
poll_interval_seconds: int | None = None poll_interval_seconds: int | None = None
# Wedge recovery configuration
auto_recovery_enabled: bool | None = None
reset_backend: str | None = None
reset_webhook_url: str | None = None
reset_webhook_method: str | None = None
recovery_boot_wait_seconds: int | None = Field(None, ge=10, le=600)
recovery_reconnect_timeout: int | None = Field(None, ge=30, le=3600)
recovery_auto_resume: bool | None = None
recovery_max_attempts: int | None = Field(None, ge=1, le=10)
recovery_window_minutes: int | None = Field(None, ge=30, le=1440)
@field_validator("reset_backend")
@classmethod
def validate_reset_backend(cls, v):
if v is not None and v.lower() not in ("manual", "webhook"):
raise ValueError("reset_backend must be 'manual' or 'webhook'")
return v.lower() if v else v
@field_validator("reset_webhook_method")
@classmethod
def validate_reset_webhook_method(cls, v):
if v is not None and v.upper() not in ("GET", "POST"):
raise ValueError("reset_webhook_method must be 'GET' or 'POST'")
return v.upper() if v else v
@field_validator("reset_webhook_url")
@classmethod
def validate_reset_webhook_url(cls, v):
if v is not None and v != "" and not v.lower().startswith(("http://", "https://")):
raise ValueError("reset_webhook_url must start with http:// or https://")
return v
@field_validator("host") @field_validator("host")
@classmethod @classmethod
def validate_host(cls, v): def validate_host(cls, v):
@@ -348,6 +381,15 @@ def get_config(unit_id: str, db: Session = Depends(get_db)):
"ftp_username": cfg.ftp_username, "ftp_username": cfg.ftp_username,
"ftp_password": cfg.ftp_password, "ftp_password": cfg.ftp_password,
"web_enabled": cfg.web_enabled, "web_enabled": cfg.web_enabled,
"auto_recovery_enabled": cfg.auto_recovery_enabled,
"reset_backend": cfg.reset_backend,
"reset_webhook_url": cfg.reset_webhook_url,
"reset_webhook_method": cfg.reset_webhook_method,
"recovery_boot_wait_seconds": cfg.recovery_boot_wait_seconds,
"recovery_reconnect_timeout": cfg.recovery_reconnect_timeout,
"recovery_auto_resume": cfg.recovery_auto_resume,
"recovery_max_attempts": cfg.recovery_max_attempts,
"recovery_window_minutes": cfg.recovery_window_minutes,
}, },
} }
@@ -407,6 +449,24 @@ async def upsert_config(unit_id: str, payload: ConfigPayload, db: Session = Depe
cfg.poll_enabled = payload.poll_enabled cfg.poll_enabled = payload.poll_enabled
if payload.poll_interval_seconds is not None: if payload.poll_interval_seconds is not None:
cfg.poll_interval_seconds = payload.poll_interval_seconds cfg.poll_interval_seconds = payload.poll_interval_seconds
if payload.auto_recovery_enabled is not None:
cfg.auto_recovery_enabled = payload.auto_recovery_enabled
if payload.reset_backend is not None:
cfg.reset_backend = payload.reset_backend
if payload.reset_webhook_url is not None:
cfg.reset_webhook_url = payload.reset_webhook_url or None
if payload.reset_webhook_method is not None:
cfg.reset_webhook_method = payload.reset_webhook_method
if payload.recovery_boot_wait_seconds is not None:
cfg.recovery_boot_wait_seconds = payload.recovery_boot_wait_seconds
if payload.recovery_reconnect_timeout is not None:
cfg.recovery_reconnect_timeout = payload.recovery_reconnect_timeout
if payload.recovery_auto_resume is not None:
cfg.recovery_auto_resume = payload.recovery_auto_resume
if payload.recovery_max_attempts is not None:
cfg.recovery_max_attempts = payload.recovery_max_attempts
if payload.recovery_window_minutes is not None:
cfg.recovery_window_minutes = payload.recovery_window_minutes
db.commit() db.commit()
db.refresh(cfg) db.refresh(cfg)
@@ -803,6 +863,89 @@ async def get_measurement_state(unit_id: str, db: Session = Depends(get_db)):
raise HTTPException(status_code=502, detail=str(e)) raise HTTPException(status_code=502, detail=str(e))
# ============================================================================
# WEDGE RECOVERY ENDPOINTS
# ============================================================================
@router.get("/{unit_id}/recovery/status")
def get_recovery_status(unit_id: str, db: Session = Depends(get_db)):
"""Get wedge detection and recovery status for a device."""
cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
if not cfg:
raise HTTPException(status_code=404, detail="NL43 config not found")
status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
manager_status = recovery_manager.get_status(unit_id)
return {
"status": "ok",
"data": {
"unit_id": unit_id,
"connection_state": status.connection_state if status else "unknown",
"recovery_state": status.recovery_state if status else "idle",
"last_wedge_at": status.last_wedge_at.isoformat() + "Z" if status and status.last_wedge_at else None,
"wedge_count": status.wedge_count if status else 0,
"last_recovery_at": status.last_recovery_at.isoformat() + "Z" if status and status.last_recovery_at else None,
"last_recovery_result": status.last_recovery_result if status else None,
"auto_recovery_enabled": cfg.auto_recovery_enabled,
"reset_backend": cfg.reset_backend,
**manager_status,
},
}
@router.post("/{unit_id}/recovery/probe")
async def probe_device(unit_id: str, db: Session = Depends(get_db)):
"""Probe the device's control and FTP ports and classify its state.
Does not trigger recovery — diagnostic only. Useful for confirming a
wedge before manually triggering a reset.
"""
cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
if not cfg:
raise HTTPException(status_code=404, detail="NL43 config not found")
classification = await classify_failure(
cfg.host,
cfg.tcp_port,
ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
)
return {"status": "ok", "unit_id": unit_id, "data": classification}
@router.post("/{unit_id}/recovery/trigger")
async def trigger_recovery(unit_id: str, db: Session = Depends(get_db)):
"""Manually trigger the recovery flow for a device.
Bypasses the consecutive-failure confirmation threshold but still
respects the one-recovery-at-a-time guard. Use after confirming a wedge
via /recovery/probe, or to test the reset hardware end-to-end.
"""
cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
if not cfg:
raise HTTPException(status_code=404, detail="NL43 config not found")
if recovery_manager.is_active(unit_id):
raise HTTPException(status_code=409, detail="Recovery already in progress for this device")
classification = await classify_failure(
cfg.host,
cfg.tcp_port,
ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
)
started = recovery_manager.start_recovery(cfg, classification)
if not started:
raise HTTPException(status_code=409, detail="Could not start recovery (already in progress)")
return {
"status": "ok",
"unit_id": unit_id,
"message": f"Recovery started (backend={cfg.reset_backend or 'manual'})",
"classification": classification,
}
@router.post("/{unit_id}/sleep") @router.post("/{unit_id}/sleep")
async def sleep_device(unit_id: str, db: Session = Depends(get_db)): async def sleep_device(unit_id: str, db: Session = Depends(get_db)):
"""Put the device into sleep mode for battery conservation.""" """Put the device into sleep mode for battery conservation."""
+5 -5
View File
@@ -338,14 +338,14 @@ class ConnectionPool:
if self._is_alive(conn): if self._is_alive(conn):
self._drain_buffer(conn.reader) self._drain_buffer(conn.reader)
conn.last_used_at = time.time() conn.last_used_at = time.time()
logger.debug(f"Pool hit for {device_key} (age={time.time() - conn.created_at:.0f}s)") logger.info(f"Pool hit for {device_key} (age={time.time() - conn.created_at:.0f}s)")
return conn.reader, conn.writer, True return conn.reader, conn.writer, True
else: else:
await self._close_connection(conn, reason="stale") await self._close_connection(conn, reason="stale")
# Open fresh connection # Open fresh connection
reader, writer = await self._open_connection(host, port, timeout) reader, writer = await self._open_connection(host, port, timeout)
logger.debug(f"New connection opened for {device_key}") logger.info(f"New connection opened for {device_key}")
return reader, writer, False return reader, writer, False
async def release(self, device_key: str, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, host: str, port: int): async def release(self, device_key: str, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, host: str, port: int):
@@ -454,11 +454,11 @@ class ConnectionPool:
"""Check whether a cached connection is still usable.""" """Check whether a cached connection is still usable."""
now = time.time() now = time.time()
# Age / idle checks # Age / idle checks (value of -1 disables the check)
if now - conn.last_used_at > self._idle_ttl: if self._idle_ttl >= 0 and now - conn.last_used_at > self._idle_ttl:
logger.debug(f"Connection {conn.device_key} idle too long ({now - conn.last_used_at:.0f}s > {self._idle_ttl}s)") logger.debug(f"Connection {conn.device_key} idle too long ({now - conn.last_used_at:.0f}s > {self._idle_ttl}s)")
return False return False
if now - conn.created_at > self._max_age: if self._max_age >= 0 and now - conn.created_at > self._max_age:
logger.debug(f"Connection {conn.device_key} too old ({now - conn.created_at:.0f}s > {self._max_age}s)") logger.debug(f"Connection {conn.device_key} too old ({now - conn.created_at:.0f}s > {self._max_age}s)")
return False return False
+107
View File
@@ -0,0 +1,107 @@
#!/usr/bin/env python3
"""
Migration script to add wedge-recovery fields to nl43_config and nl43_status tables.
Adds to nl43_config:
- auto_recovery_enabled (BOOLEAN, default 0/False)
- reset_backend (TEXT, default 'manual')
- reset_webhook_url (TEXT, nullable)
- reset_webhook_method (TEXT, default 'POST')
- recovery_boot_wait_seconds (INTEGER, default 90)
- recovery_reconnect_timeout (INTEGER, default 180)
- recovery_auto_resume (BOOLEAN, default 1/True)
- recovery_max_attempts (INTEGER, default 2)
- recovery_window_minutes (INTEGER, default 360)
Adds to nl43_status:
- connection_state (TEXT, default 'ok')
- last_wedge_at (DATETIME, nullable)
- wedge_count (INTEGER, default 0)
- recovery_state (TEXT, default 'idle')
- last_recovery_at (DATETIME, nullable)
- last_recovery_result (TEXT, nullable)
Usage:
python migrate_add_recovery_fields.py
"""
import sqlite3
import sys
from pathlib import Path
CONFIG_COLUMNS = [
("auto_recovery_enabled", "BOOLEAN DEFAULT 0"),
("reset_backend", "TEXT DEFAULT 'manual'"),
("reset_webhook_url", "TEXT"),
("reset_webhook_method", "TEXT DEFAULT 'POST'"),
("recovery_boot_wait_seconds", "INTEGER DEFAULT 90"),
("recovery_reconnect_timeout", "INTEGER DEFAULT 180"),
("recovery_auto_resume", "BOOLEAN DEFAULT 1"),
("recovery_max_attempts", "INTEGER DEFAULT 2"),
("recovery_window_minutes", "INTEGER DEFAULT 360"),
]
STATUS_COLUMNS = [
("connection_state", "TEXT DEFAULT 'ok'"),
("last_wedge_at", "DATETIME"),
("wedge_count", "INTEGER DEFAULT 0"),
("recovery_state", "TEXT DEFAULT 'idle'"),
("last_recovery_at", "DATETIME"),
("last_recovery_result", "TEXT"),
]
def migrate():
db_path = Path("data/slmm.db")
if not db_path.exists():
print(f"❌ Database not found at {db_path}")
print(" Run this script from the slmm directory")
return False
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(nl43_config)")
config_columns = [row[1] for row in cursor.fetchall()]
cursor.execute("PRAGMA table_info(nl43_status)")
status_columns = [row[1] for row in cursor.fetchall()]
changes_made = False
for name, ddl in CONFIG_COLUMNS:
if name not in config_columns:
print(f"Adding {name} to nl43_config...")
cursor.execute(f"ALTER TABLE nl43_config ADD COLUMN {name} {ddl}")
changes_made = True
else:
print(f"{name} already exists in nl43_config")
for name, ddl in STATUS_COLUMNS:
if name not in status_columns:
print(f"Adding {name} to nl43_status...")
cursor.execute(f"ALTER TABLE nl43_status ADD COLUMN {name} {ddl}")
changes_made = True
else:
print(f"{name} already exists in nl43_status")
if changes_made:
conn.commit()
print("\n✓ Migration completed successfully")
print(" Added wedge-recovery fields to nl43_config and nl43_status")
else:
print("\n✓ All recovery fields already exist - no changes needed")
conn.close()
return True
except Exception as e:
print(f"❌ Migration failed: {e}")
return False
if __name__ == "__main__":
success = migrate()
sys.exit(0 if success else 1)
+377
View File
@@ -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)