Disconnect from Bluetooth devices before the server stops

The polling threads are daemon threads, so a stop ended the server in the middle of a Bluetooth read. The read got no disconnect, and the adapter kept an open connection or a scan in progress.

The server now sets the stop event at exit, closes its socket, and waits at most 20 seconds for the polling threads. A Bluetooth read that sees the stop event cancels its current step, disconnects, and returns. A read that waits for the adapter does not connect after a stop.

README.md

@@ -104,6 +104,11 @@ first request after startup waits, until each device completes one read attempt.
 The server answers each client in its own thread, so a slow client does not
 delay the other clients. It closes a connection that stays silent for 5 seconds.
 
+The server holds a Bluetooth connection only during a read, and disconnects
+after each one. When the server stops, it ends a read in progress and
+disconnects from that device before it exits. It waits at most 20 seconds for
+the disconnect.
+
 Each Bluetooth read listens for 10 seconds. If a required report is still
 missing, the read continues until the report arrives, for at most 30 seconds in
 total. A read that reaches this limit returns a partial snapshot. Its

bmspy/ecoflow_bt.py

@@ -16,6 +16,7 @@ import contextlib
 import hashlib
 import logging
 import struct
+import threading
 from collections.abc import Callable
 from dataclasses import dataclass
 from enum import StrEnum, auto
@@ -38,6 +39,8 @@ MANUFACTURER_KEY = 0xB5B5
 
 # How often a collection checks for the required reports.
 _COMPLETION_POLL_SECONDS = 0.5
+# How often a collection looks at its stop event.
+_STOP_POLL_SECONDS = 0.2
 # Missing reports of the last collection per address, to log changes only.
 _last_missing: dict[str, set[str]] = {}
 
@@ -1052,7 +1055,9 @@ def _record_missing(address: str, snapshot: EcoFlowSnapshot, missing: set[str])
         ))
 
 
-def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool:
+def collect_data(
+    cfg: EcoFlowBTConfig, debug: int = 0, stop: threading.Event | None = None,
+) -> EcoFlowSnapshot | bool:
     """Connect to an EcoFlow device, collect push data, disconnect, return snapshot.
 
     Analogous to :func:`bmspy.jbd_bms.collect_data`.  Each call opens a fresh
@@ -1065,14 +1070,23 @@ def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool
     Each failure is logged with its reason, at every debug level.
     A snapshot from a collection that reached the limit is partial, and its
     ``ef_missing_measurements`` field names the reports that did not arrive.
+
+    If *stop* is set before the collection, no connection is made. If it is
+    set during the collection, the collection ends early, disconnects, and
+    returns ``False``.
     """
+    if stop is not None and stop.is_set():
+        return False
+    disconnecting = False
+
     def failed(reason: str) -> bool:
         # Always logged: without this line, a device that is out of reach
         # only shows as missing data.
         debugger(f"ecoflow_bt [{cfg.address}]: read failed: {reason}")
         return False
 
-    async def _run() -> EcoFlowSnapshot | bool:
+    async def _collect() -> EcoFlowSnapshot | bool:
+        nonlocal disconnecting
         device = EcoFlowBTDevice(
             address=cfg.address,
             serial_number=cfg.serial_number,
@@ -1093,8 +1107,25 @@ def collect_data(cfg: EcoFlowBTConfig, debug: int = 0) -> EcoFlowSnapshot | bool
             _record_missing(cfg.address, device.snapshot, missing)
             return device.snapshot
         finally:
+            disconnecting = True
             await device.disconnect()
 
+    async def _run() -> EcoFlowSnapshot | bool:
+        if stop is None:
+            return await _collect()
+        task = asyncio.ensure_future(_collect())
+        while not task.done() and not stop.is_set():
+            await asyncio.wait({task}, timeout=_STOP_POLL_SECONDS)
+        # Cancel once, and never during the disconnect: the cancellation ends
+        # the current step, and the disconnect in _collect then runs in full.
+        if not task.done() and not disconnecting:
+            task.cancel()
+        try:
+            return await task
+        except asyncio.CancelledError:
+            debugger(f"ecoflow_bt [{cfg.address}]: read stopped, device disconnected")
+            return False
+
     try:
         return asyncio.run(_run())
     except Exception as exc:

bmspy/server.py

@@ -44,6 +44,8 @@ SERIAL_RETRY_SECONDS = 1
 SERIAL_STALE_SECONDS = 30
 BLUETOOTH_POLL_SECONDS = 60
 FIRST_POLL_TIMEOUT = 120
+# Time limit at exit for the reads in progress to end and to disconnect.
+POLL_STOP_TIMEOUT = 20
 # Time limit for each read from a client and each write to it.
 CLIENT_TIMEOUT_SECONDS = 5
 # A request is a short JSON object. A larger length prefix is an error.
@@ -189,13 +191,18 @@ def parse_ecoflow_spec(spec: str) -> tuple[str, str, str]:
     return parts[0].upper(), parts[1], parts[2]
 
 
-def collect_device_data(device: DeviceState, debug: int = 0) -> UPS | None:
+def collect_device_data(
-    """Collect fresh data from whichever device type is in *device*."""
+    device: DeviceState, debug: int = 0, stop: threading.Event | None = None,
+) -> UPS | None:
+    """Collect fresh data from whichever device type is in *device*.
+
+    A Bluetooth read ends early and disconnects when *stop* is set.
+    """
     if device.ser is not None:
         return collect_data(device.ser, debug) or None
     if device.ecoflow is not None:
         with bluetooth_lock:
-            return ecoflow_collect_data(device.ecoflow, debug) or None
+            return ecoflow_collect_data(device.ecoflow, debug, stop) or None
     return None
 
 
@@ -209,7 +216,7 @@ def poll_device(device: DeviceState, stop: threading.Event, debug: int = 0) -> N
     last_success = 0.0
     while not stop.is_set():
         try:
-            data = collect_device_data(device, debug)
+            data = collect_device_data(device, debug, stop)
         except Exception as e:
             # A failed read must not end polling for this device.
             debugger("server: collection error: {}".format(e))
@@ -247,6 +254,24 @@ def start_polling(
     return threads
 
 
+def stop_polling(threads: list[threading.Thread], stop: threading.Event) -> None:
+    """End the polling threads, and wait for a Bluetooth read to disconnect.
+
+    The threads are daemon threads, so the process does not wait for them. A
+    Bluetooth read that ends without a disconnect leaves the adapter with an
+    open connection or a scan in progress.
+    """
+    stop.set()
+    deadline = time.monotonic() + POLL_STOP_TIMEOUT
+    for thread in threads:
+        thread.join(max(0.0, deadline - time.monotonic()))
+    alive = [thread.name for thread in threads if thread.is_alive()]
+    if alive:
+        debugger("server: no clean stop in {} seconds for {}".format(
+            POLL_STOP_TIMEOUT, ", ".join(alive),
+        ))
+
+
 def handle_client(
     connection: socket.socket, ups_devices: dict[str, DeviceState], debug: int = 0,
 ) -> None:
@@ -491,10 +516,10 @@ def main():
 
     sock.listen(1)
 
-    stop_polling = threading.Event()
+    stop = threading.Event()
-    atexit.register(stop_polling.set)
+    threads = start_polling(ups_devices, stop, debug)
-    start_polling(ups_devices, stop_polling, debug)
 
+    try:
         while True:
             if debug > 2:
                 debugger("socket: waiting for a connection")
@@ -505,6 +530,11 @@ def main():
             threading.Thread(
                 target=handle_client, args=(connection, ups_devices, debug), name="client", daemon=True,
             ).start()
+    finally:
+        # SIGTERM arrives here as SystemExit. Refuse new clients at once, then
+        # let each device finish: a Bluetooth read must disconnect first.
+        sock.close()
+        stop_polling(threads, stop)
 
 
 if __name__ == "__main__":

tests/test_ecoflow_bt.py

@@ -9,6 +9,8 @@ import asyncio
 import importlib.util
 import struct
 import sys
+import threading
+import time
 from argparse import Namespace
 from dataclasses import asdict
 from itertools import permutations
@@ -538,6 +540,78 @@ def test_successful_collection_logs_no_failure(monkeypatch, capsys):
     assert capsys.readouterr().out == ""
 
 
+def _stoppable_device(monkeypatch, **steps):
+    """Stand-in device that records whether its disconnect ran in full."""
+    events = []
+
+    async def disconnect():
+        events.append("disconnect started")
+        await asyncio.sleep(0.05)
+        events.append("disconnect done")
+
+    fake = Mock(
+        connect=AsyncMock(return_value=True), wait_authenticated=AsyncMock(return_value=True),
+        wait_for_measurements=AsyncMock(return_value=set()), disconnect=disconnect,
+        _pkt_count=1, last_error=None,
+    )
+    fake.configure_mock(**steps)
+    monkeypatch.setattr("bmspy.ecoflow_bt.EcoFlowBTDevice", Mock(return_value=fake))
+    monkeypatch.setattr("bmspy.ecoflow_bt._STOP_POLL_SECONDS", 0.01)
+    monkeypatch.setattr("bmspy.ecoflow_bt._last_missing", {})
+    return fake, events
+
+
+@pytest.mark.parametrize("step", ["connect", "wait_authenticated", "wait_for_measurements"])
+def test_stop_during_a_collection_disconnects_before_it_returns(monkeypatch, capsys, step):
+    """A server that stops in the middle of a read must not leave the link open."""
+    stop = threading.Event()
+
+    async def interrupted(*args, **kwargs):
+        stop.set()
+        await asyncio.sleep(30)
+
+    fake, events = _stoppable_device(monkeypatch, **{step: interrupted})
+    config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0)
+    started = time.monotonic()
+    assert collect_data(config, stop=stop) is False
+    assert time.monotonic() - started < 5
+    assert events == ["disconnect started", "disconnect done"]
+    lines = capsys.readouterr().out.splitlines()
+    assert len(lines) == 1
+    assert lines[0].endswith(f"ecoflow_bt [{ADDRESS}]: read stopped, device disconnected")
+
+
+def test_stop_during_the_disconnect_lets_the_disconnect_finish(monkeypatch):
+    stop = threading.Event()
+    fake, events = _stoppable_device(monkeypatch)
+    slow_disconnect = fake.disconnect
+
+    async def disconnect():
+        stop.set()
+        await slow_disconnect()
+
+    fake.disconnect = disconnect
+    config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0)
+    assert collect_data(config, stop=stop) is fake.snapshot
+    assert events == ["disconnect started", "disconnect done"]
+
+
+def test_collection_does_not_connect_after_stop(monkeypatch):
+    stop = threading.Event()
+    stop.set()
+    device = Mock()
+    monkeypatch.setattr("bmspy.ecoflow_bt.EcoFlowBTDevice", device)
+    assert collect_data(initialise_device(ADDRESS, SERIAL, "test-user"), stop=stop) is False
+    device.assert_not_called()
+
+
+def test_collection_with_an_unset_stop_event_returns_the_snapshot(monkeypatch):
+    fake, events = _stoppable_device(monkeypatch)
+    config = initialise_device(ADDRESS, SERIAL, "test-user", listen_duration=0)
+    assert collect_data(config, stop=threading.Event()) is fake.snapshot
+    assert events == ["disconnect started", "disconnect done"]
+
+
 def test_connect_records_the_reason_for_a_failure(monkeypatch):
     client = Mock(connect=AsyncMock(side_effect=TimeoutError("no answer")), disconnect=AsyncMock(), is_connected=False)
     monkeypatch.setitem(sys.modules, "bleak", SimpleNamespace(BleakClient=Mock(return_value=client)))

tests/test_server.py

@@ -25,6 +25,7 @@ def _poll_once(devices, stop, debug=0):
     for device in devices.values():
         device.data = collect_device_data(device, debug)
         device.polled.set()
+    return []
 
 
 # ---------------------------------------------------------------------------
@@ -217,6 +218,39 @@ listener.accept()
         assert result.stderr.strip() == "terminating"
         assert not socket_path.exists()
 
+    def test_sigterm_during_a_bluetooth_read_waits_for_the_disconnect(self, tmp_path):
+        """The server process itself, with a stand-in for the Bluetooth read."""
+        socket_path = tmp_path / "bms.sock"
+        script = """
+import os
+import signal
+import sys
+import threading
+from bmspy import server
+
+def bluetooth_read(config, debug=0, stop=None):
+    if stop.is_set():
+        return False
+    print("read started", flush=True)
+    threading.Timer(0.1, os.kill, args=(os.getpid(), signal.SIGTERM)).start()
+    stop.wait(10)
+    print("disconnected", flush=True)
+    return False
+
+server.ecoflow_collect_data = bluetooth_read
+sys.argv = ["bmspy-server", "--socket", sys.argv[1],
+            "-d", "river@02:00:00:00:00:01/R601TEST00000001/test-user"]
+server.main()
+"""
+        result = subprocess.run(
+            [sys.executable, "-c", script, str(socket_path)],
+            capture_output=True, text=True, timeout=30,
+        )
+        assert result.stdout.splitlines() == ["read started", "disconnected"]
+        assert result.returncode == 1
+        assert result.stderr.strip() == "terminating"
+        assert not socket_path.exists()
+
 
 # ---------------------------------------------------------------------------
 # socket_cleanup
@@ -765,7 +799,7 @@ class TestServerMain:
              patch("bmspy.server.signal.signal"), \
              patch("bmspy.server.atexit.register"), \
              patch("bmspy.server.initialise_serial"), \
-             patch("bmspy.server.start_polling"), \
+             patch("bmspy.server.start_polling", return_value=[]), \
              patch("bmspy.server.os.path.isdir", return_value=True), \
              patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt):
             from bmspy.server import main as server_main
@@ -773,6 +807,31 @@ class TestServerMain:
                 server_main()
         assert exit_info.value.code == 1
 
+    def test_exit_closes_the_socket_and_stops_polling(self, tmp_path):
+        sock_path = str(tmp_path / "server_stop.sock")
+        threads = [object()]
+        listeners = []
+        original_listen = socket_module.socket.listen
+
+        def listen(self, backlog=1):
+            listeners.append(self)
+            return original_listen(self, backlog)
+
+        with patch("sys.argv", ["bmspy-server", "--socket", sock_path, "--device", "/dev/ttyUSB0"]), \
+             patch("bmspy.server.signal.signal"), \
+             patch("bmspy.server.atexit.register"), \
+             patch("bmspy.server.initialise_serial"), \
+             patch("bmspy.server.start_polling", return_value=threads) as start, \
+             patch("bmspy.server.stop_polling") as stop_polling, \
+             patch("bmspy.server.os.path.isdir", return_value=True), \
+             patch.object(socket_module.socket, "listen", listen), \
+             patch.object(socket_module.socket, "accept", side_effect=KeyboardInterrupt):
+            from bmspy.server import main as server_main
+            with pytest.raises(SystemExit):
+                server_main()
+        stop_polling.assert_called_once_with(threads, start.call_args.args[1])
+        assert listeners[0].fileno() == -1
+
     def test_socket_read_error_logs_and_continues(self, tmp_path, capsys):
         """Test that read_request errors are caught and logged."""
         sock_path = str(tmp_path / "server_err.sock")
@@ -1098,7 +1157,7 @@ class TestServerMain:
              patch("bmspy.server.signal.signal"), \
              patch("bmspy.server.atexit.register"), \
              patch("bmspy.server.initialise_serial"), \
-             patch("bmspy.server.start_polling"), \
+             patch("bmspy.server.start_polling", return_value=[]), \
              patch("bmspy.server.os.path.isdir", return_value=True), \
              patch.object(socket_module.socket, "accept", side_effect=OSError("accept failed")) as accept:
             from bmspy.server import main as server_main

tests/test_server_ecoflow.py

@@ -1,6 +1,7 @@
 """Server routing and caching tests with synthetic device identifiers."""
 
 import threading
+import time
 from unittest.mock import Mock
 
 import pytest
@@ -71,7 +72,7 @@ def test_collection_dispatches_to_the_selected_backend(monkeypatch):
     assert server.collect_device_data(server.DeviceState(ser=port), 2) is serial_data
     serial_read.assert_called_once_with(port, 2)
     assert server.collect_device_data(server.DeviceState(ecoflow=config), 3) is bluetooth_data
-    bluetooth_read.assert_called_once_with(config, 3)
+    bluetooth_read.assert_called_once_with(config, 3, None)
     assert server.collect_device_data(server.DeviceState()) is None
 
 
@@ -104,7 +105,7 @@ def test_poll_interval_depends_on_backend(monkeypatch, bluetooth, interval):
     assert state.timestamp == 100
     assert state.polled.is_set()
     assert stop.waits == [interval]
-    read.assert_called_once_with(state, 2)
+    read.assert_called_once_with(state, 2, stop)
 
 
 def test_offline_bluetooth_does_not_retry_at_once_or_serve_stale_data(monkeypatch):
@@ -151,7 +152,7 @@ def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch):
     offline_started, release = threading.Event(), threading.Event()
     working, offline = server.DeviceState(ser=Mock()), server.DeviceState(ser=Mock())
 
-    def read(device, debug=0):
+    def read(device, debug=0, stop=None):
         if device is offline:
             offline_started.set()
             release.wait(5)
@@ -172,3 +173,51 @@ def test_polling_threads_do_not_wait_for_each_other_and_stop(monkeypatch):
         for thread in threads:
             thread.join(5)
     assert not any(thread.is_alive() for thread in threads)
+
+
+def test_stop_polling_waits_for_the_bluetooth_disconnect(monkeypatch):
+    events = []
+
+    def bluetooth_read(config, debug=0, stop=None):
+        if stop.is_set():
+            return False
+        events.append("read started")
+        assert stop.wait(5)
+        events.append("disconnected")
+        return False
+
+    monkeypatch.setattr(server, "ecoflow_collect_data", bluetooth_read)
+    # The second device waits for the first one. It must not start a read.
+    devices = {
+        name: server.DeviceState(ecoflow=initialise_device(ADDRESS, SERIAL, "test-user"))
+        for name in ("first", "second")
+    }
+    stop = threading.Event()
+    threads = server.start_polling(devices, stop)
+    while not events:
+        time.sleep(0.01)
+    server.stop_polling(threads, stop)
+    assert events == ["read started", "disconnected"]
+    assert not any(thread.is_alive() for thread in threads)
+
+
+def test_stop_polling_reports_a_read_that_does_not_end(monkeypatch, capsys):
+    started, release = threading.Event(), threading.Event()
+
+    def stuck_read(device, debug=0, stop=None):
+        started.set()
+        release.wait(5)
+
+    monkeypatch.setattr(server, "collect_device_data", stuck_read)
+    monkeypatch.setattr(server, "POLL_STOP_TIMEOUT", 0.05)
+    stop = threading.Event()
+    threads = server.start_polling({"stuck": server.DeviceState(ser=Mock())}, stop)
+    try:
+        assert started.wait(5)
+        server.stop_polling(threads, stop)
+        assert "server: no clean stop in 0.05 seconds for poll-stuck" in capsys.readouterr().out
+    finally:
+        release.set()
+        for thread in threads:
+            thread.join(5)
+    assert not any(thread.is_alive() for thread in threads)