Reconnect the WiFi mOTA seeder when the source closes its link
A WiFi OTA source drops the seeder socket whenever its WiFi blips, and motatool serve then exits 0. The destination keeps its staged partial download and resumes when the same mOTA is served again, so restart the TCP seeder (up to 3 times per run) instead of failing the transfer. Serial seeders and non-zero exits still fail as before.
This commit is contained in:
1 parent
e5b96d2ee0
commit
6831520d69
2 files changed
+104
-5
No files matched your search
@@ -4047,12 +4047,20 @@ class SeederProcess:
|
||||
r"folder\s+(?:is\s+)?already\s+(?:owned|attached)|"
|
||||
r"could not (?:attach|enter).*folder"
|
||||
)
|
||||
# A WiFi source closes the seeder socket whenever its WiFi drops, even
|
||||
# briefly, and `motatool serve` then exits 0 ("bye"). The destination keeps
|
||||
# its staged partial download and resumes once the same mOTA is served
|
||||
# again, so a clean exit on TCP is reconnected instead of ending the run.
|
||||
MAX_RECONNECTS = 3
|
||||
RECONNECT_DELAY_SECONDS = 5.0
|
||||
|
||||
def __init__(self, args: argparse.Namespace, served_dir: Path, work_dir: Path):
|
||||
self.args = args
|
||||
self.log_path = work_dir / "motatool-serve.log"
|
||||
self.log_file = None
|
||||
self.process: subprocess.Popen[str] | None = None
|
||||
self.reconnects = 0
|
||||
self._log_offset = 0 # startup checks read only the current session's output
|
||||
command = [args.motatool, "serve", "--dir", str(served_dir), "-v"]
|
||||
if args.source_serial:
|
||||
command.extend([
|
||||
@@ -4066,6 +4074,11 @@ class SeederProcess:
|
||||
def start(self) -> None:
|
||||
print(f"[seeder] log: {self.log_path}")
|
||||
self.log_file = self.log_path.open("w", encoding="utf-8")
|
||||
self._launch()
|
||||
self._wait_until_attached()
|
||||
print("[seeder] running (device COUNT confirmed)")
|
||||
|
||||
def _launch(self) -> None:
|
||||
creationflags = 0
|
||||
if os.name == "nt":
|
||||
creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
||||
@@ -4080,8 +4093,6 @@ class SeederProcess:
|
||||
except FileNotFoundError as exc:
|
||||
self._close_log()
|
||||
raise OtaError(f"required command was not found: {self.args.motatool}") from exc
|
||||
self._wait_until_attached()
|
||||
print("[seeder] running (device COUNT confirmed)")
|
||||
|
||||
def _close_log(self) -> None:
|
||||
log_file = self.log_file
|
||||
@@ -4110,9 +4121,34 @@ class SeederProcess:
|
||||
detail = self._read_log()
|
||||
except OSError as exc:
|
||||
return f"could not read seeder log: {exc}"
|
||||
return detail[-4096:].strip() or "no seeder log output"
|
||||
return detail[self._log_offset:][-4096:].strip() or "no seeder log output"
|
||||
|
||||
def ensure_running(self, context: str = "") -> None:
|
||||
process = self.process
|
||||
if (process is not None and process.poll() == 0 and not self.args.source_serial
|
||||
and self.reconnects < self.MAX_RECONNECTS):
|
||||
self._reconnect(context)
|
||||
return
|
||||
self._require_running(context)
|
||||
|
||||
def _reconnect(self, context: str) -> None:
|
||||
"""Serve the same folder again after the source closed the TCP link."""
|
||||
self.reconnects += 1
|
||||
suffix = f" {context}" if context else ""
|
||||
print(
|
||||
f"[seeder] the source closed the seeder link{suffix}; reconnecting "
|
||||
f"({self.reconnects}/{self.MAX_RECONNECTS})"
|
||||
)
|
||||
time.sleep(self.RECONNECT_DELAY_SECONDS)
|
||||
try:
|
||||
self._log_offset = len(self._read_log())
|
||||
except OSError:
|
||||
self._log_offset = 0
|
||||
self._launch()
|
||||
self._wait_until_attached()
|
||||
print("[seeder] reconnected (device COUNT confirmed)")
|
||||
|
||||
def _require_running(self, context: str = "") -> None:
|
||||
if self.process is None:
|
||||
raise OtaError("motatool seeder is not running")
|
||||
return_code = self.process.poll()
|
||||
@@ -4135,7 +4171,7 @@ class SeederProcess:
|
||||
last_progress = None
|
||||
next_notice = started + 30
|
||||
while True:
|
||||
self.ensure_running("during startup")
|
||||
self._require_running("during startup")
|
||||
detail = self._log_tail()
|
||||
if self.ATTACH_ERROR_PATTERN.search(detail):
|
||||
raise OtaError(
|
||||
@@ -5721,7 +5757,7 @@ def monitor_download(
|
||||
last_status_activity = -1
|
||||
last_reported_quarter = -1
|
||||
while time.monotonic() < deadline:
|
||||
ensure_seeder_running(seeder, "before first transfer status check")
|
||||
ensure_seeder_running(seeder, "during transfer")
|
||||
seen, total, activity = seeder.payload_read_progress(package)
|
||||
now = time.monotonic()
|
||||
if activity > last_activity:
|
||||
|
||||
@@ -244,6 +244,69 @@ class SeederStartupTests(unittest.TestCase):
|
||||
self.assertIn("child diagnostic", str(failure.exception))
|
||||
|
||||
|
||||
class SeederReconnectTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.output = io.StringIO()
|
||||
self.addCleanup(mock.patch.stopall)
|
||||
mock.patch.object(ota.time, "sleep").start()
|
||||
self.redirect = contextlib.redirect_stdout(self.output)
|
||||
self.redirect.__enter__()
|
||||
self.addCleanup(self.redirect.__exit__, None, None, None)
|
||||
|
||||
def seeder(self, return_code, source_serial=None):
|
||||
args = argparse.Namespace(
|
||||
motatool="motatool", source_serial=source_serial,
|
||||
source_tcp="10.0.0.2:5001", source_baud=115200,
|
||||
)
|
||||
seeder = ota.SeederProcess(args, Path("/served"), Path("/work"))
|
||||
seeder.process = mock.Mock()
|
||||
seeder.process.poll.return_value = return_code
|
||||
seeder._read_log = mock.Mock(return_value="first session\nbye\n")
|
||||
seeder._launch = mock.Mock()
|
||||
seeder._wait_until_attached = mock.Mock()
|
||||
return seeder
|
||||
|
||||
def test_clean_tcp_exit_reconnects_the_same_folder(self):
|
||||
seeder = self.seeder(0)
|
||||
seeder.ensure_running("during transfer")
|
||||
seeder._launch.assert_called_once_with()
|
||||
seeder._wait_until_attached.assert_called_once_with()
|
||||
self.assertEqual(seeder.reconnects, 1)
|
||||
self.assertEqual(seeder._log_offset, len("first session\nbye\n"))
|
||||
self.assertIn("closed the seeder link during transfer; reconnecting (1/3)",
|
||||
self.output.getvalue())
|
||||
self.assertIn("reconnected (device COUNT confirmed)", self.output.getvalue())
|
||||
|
||||
def test_running_seeder_is_left_alone(self):
|
||||
seeder = self.seeder(None)
|
||||
seeder.ensure_running("during transfer")
|
||||
seeder._launch.assert_not_called()
|
||||
|
||||
def test_reconnects_are_bounded(self):
|
||||
seeder = self.seeder(0)
|
||||
seeder.reconnects = ota.SeederProcess.MAX_RECONNECTS
|
||||
with self.assertRaisesRegex(ota.OtaError, "exited with status 0 during transfer"):
|
||||
seeder.ensure_running("during transfer")
|
||||
seeder._launch.assert_not_called()
|
||||
|
||||
def test_failed_or_serial_exit_still_fails_the_run(self):
|
||||
for return_code, serial in ((1, None), (0, "/dev/source")):
|
||||
with self.subTest(return_code=return_code, serial=serial):
|
||||
seeder = self.seeder(return_code, serial)
|
||||
with self.assertRaisesRegex(ota.OtaError, f"exited with status {return_code}"):
|
||||
seeder.ensure_running("during transfer")
|
||||
seeder._launch.assert_not_called()
|
||||
|
||||
def test_startup_reads_only_the_current_session(self):
|
||||
seeder = ota.SeederProcess(
|
||||
argparse.Namespace(motatool="motatool", source_serial=None,
|
||||
source_tcp="10.0.0.2:5001", source_baud=115200),
|
||||
Path("/served"), Path("/work"))
|
||||
seeder._read_log = mock.Mock(return_value=" [dev] COUNT -> 1\nbye\nserving again\n")
|
||||
seeder._log_offset = len(" [dev] COUNT -> 1\nbye\n")
|
||||
self.assertEqual(seeder._log_tail(), "serving again")
|
||||
|
||||
|
||||
class SeederStopTests(unittest.TestCase):
|
||||
def seeder(self, return_code=0):
|
||||
args = argparse.Namespace(
|
||||
|
||||
Reference in new issue
Block a user