Compare commits
10 Commits
v0.3.0
...
b9bf4ea9e6
| Author | SHA1 | Date | |
|---|---|---|---|
| b9bf4ea9e6 | |||
| 3c22f1d70c | |||
| 450509d210 | |||
| fefa9eace8 | |||
| 98a8d357e5 | |||
| 0a7422eceb | |||
| 996b993cb9 | |||
| 01337696b3 | |||
| a302fd15d4 | |||
| af5ecc1a92 |
@@ -1,5 +1,6 @@
|
||||
/manuals/
|
||||
/data/
|
||||
/data-dev/
|
||||
/SLM-stress-test/stress_test_logs/
|
||||
/SLM-stress-test/tcpdump-runs/
|
||||
|
||||
|
||||
@@ -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/),
|
||||
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
|
||||
|
||||
### 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-43’s 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-43’s 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 doesn’t 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 didn’t 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-43’s 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) What’s next (future work — when the unit is back)
|
||||
Because long tests can’t 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 24–72 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 doesn’t: 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 doesn’t 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-43’s **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 SLMM’s “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.**
|
||||
@@ -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())
|
||||
@@ -17,6 +17,7 @@ from app.database import SessionLocal
|
||||
from app.models import NL43Config, NL43Status
|
||||
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.recovery import classify_failure, recovery_manager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -38,6 +39,7 @@ class BackgroundPoller:
|
||||
self._running = False
|
||||
self._logger = logger
|
||||
self._last_cleanup = None # Track last log cleanup time
|
||||
self._last_pool_log = None # Track last connection pool heartbeat log
|
||||
|
||||
async def start(self):
|
||||
"""Start the background polling task."""
|
||||
@@ -89,6 +91,24 @@ class BackgroundPoller:
|
||||
except Exception as 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
|
||||
sleep_time = self._calculate_sleep_interval()
|
||||
self._logger.debug(f"Sleeping for {sleep_time} seconds until next poll cycle")
|
||||
@@ -212,6 +232,7 @@ class BackgroundPoller:
|
||||
status.consecutive_failures = 0
|
||||
status.last_success = datetime.utcnow()
|
||||
status.last_error = None
|
||||
status.connection_state = "ok"
|
||||
|
||||
db.commit()
|
||||
self._logger.info(f"✓ Successfully polled {unit_id}")
|
||||
@@ -303,6 +324,61 @@ class BackgroundPoller:
|
||||
|
||||
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:
|
||||
"""
|
||||
Calculate the next sleep interval based on all device poll intervals.
|
||||
|
||||
@@ -23,6 +23,17 @@ class NL43Config(Base):
|
||||
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
|
||||
|
||||
# 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):
|
||||
"""
|
||||
@@ -57,6 +68,14 @@ class NL43Status(Base):
|
||||
# FTP start time sync tracking
|
||||
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):
|
||||
"""
|
||||
|
||||
+482
@@ -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
@@ -13,6 +13,7 @@ import asyncio
|
||||
from app.database import get_db
|
||||
from app.models import NL43Config, NL43Status
|
||||
from app.services import NL43Client, persist_snapshot
|
||||
from app.recovery import classify_failure, recovery_manager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -53,6 +54,38 @@ class ConfigPayload(BaseModel):
|
||||
poll_enabled: bool | 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")
|
||||
@classmethod
|
||||
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_password": cfg.ftp_password,
|
||||
"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
|
||||
if payload.poll_interval_seconds is not None:
|
||||
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.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))
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 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")
|
||||
async def sleep_device(unit_id: str, db: Session = Depends(get_db)):
|
||||
"""Put the device into sleep mode for battery conservation."""
|
||||
|
||||
+5
-5
@@ -338,14 +338,14 @@ class ConnectionPool:
|
||||
if self._is_alive(conn):
|
||||
self._drain_buffer(conn.reader)
|
||||
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
|
||||
else:
|
||||
await self._close_connection(conn, reason="stale")
|
||||
|
||||
# Open fresh connection
|
||||
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
|
||||
|
||||
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."""
|
||||
now = time.time()
|
||||
|
||||
# Age / idle checks
|
||||
if now - conn.last_used_at > self._idle_ttl:
|
||||
# Age / idle checks (value of -1 disables the check)
|
||||
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)")
|
||||
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)")
|
||||
return False
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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