From 60b08f225adb70443ac44365298d47ba89b67c81 Mon Sep 17 00:00:00 2001 From: fy59 Date: Sun, 20 Sep 2026 15:30:06 +0200 Subject: [PATCH] fix(sync): throttle MTP polling and preserve timeout cause --- tests/test_sync_peer_worker.py | 103 ++++++++++++++++++++++++++++++ tools/sync_peer_worker.py | 111 ++++++++++++++++++++++++--------- tui/meson.build | 5 ++ 3 files changed, 190 insertions(+), 29 deletions(-) diff --git a/tests/test_sync_peer_worker.py b/tests/test_sync_peer_worker.py index 6afab48..624b678 100644 --- a/tests/test_sync_peer_worker.py +++ b/tests/test_sync_peer_worker.py @@ -1,5 +1,6 @@ import importlib.util import subprocess +import tempfile import time import unittest from pathlib import Path @@ -30,6 +31,108 @@ class SyncPeerWorkerTest(unittest.TestCase): self.assertGreaterEqual(timeout, 178) self.assertLessEqual(timeout, 180) + def test_mtp_adapter_uses_fixed_operation_budget(self): + completed = subprocess.CompletedProcess([], 0, b"", b"") + with mock.patch.object(worker.subprocess, "run", return_value=completed) as run: + worker.run_adapter( + Path("/fixed/adapter"), "pull", "peer_fixed", + Path("/private/transport"), 100.0, clock=lambda: 50.0, + ) + self.assertEqual(worker.MTP_OPERATION_TIMEOUT_SECONDS, run.call_args.kwargs["timeout"]) + + def test_mtp_pull_pump_is_throttled(self): + now = [0.0] + pulls = [] + pump = worker.MtpPullPump( + lambda: pulls.append(now[0]), interval=3.0, clock=lambda: now[0] + ) + for timestamp in (0.05, 0.5, 2.99, 3.0, 3.1, 5.99, 6.0): + now[0] = timestamp + pump() + self.assertEqual([3.0, 6.0], pulls) + + def test_global_deadline_is_not_reported_as_transport_timeout(self): + with self.assertRaises(worker.ConversationDeadlineExpired): + worker.run_adapter( + Path("/fixed/adapter"), "pull", "peer_fixed", + Path("/private/transport"), 10.0, clock=lambda: 10.0, + ) + + def test_deadline_limited_pull_becomes_protocol_timeout(self): + with mock.patch.object( + worker.subprocess, "run", side_effect=subprocess.TimeoutExpired(["adapter"], 0.5) + ): + with self.assertRaises(worker.ConversationDeadlineExpired): + worker.run_adapter( + Path("/fixed/adapter"), "pull", "peer_fixed", + Path("/private/transport"), 10.5, clock=lambda: 10.0, + ) + + def test_blocked_full_budget_pull_is_transport_timeout(self): + with mock.patch.object( + worker.subprocess, "run", side_effect=subprocess.TimeoutExpired(["adapter"], 30) + ): + with self.assertRaisesRegex(RuntimeError, "transport_timeout"): + worker.run_adapter( + Path("/fixed/adapter"), "pull", "peer_fixed", + Path("/private/transport"), 100.0, clock=lambda: 50.0, + ) + + def test_delayed_correlated_generation_survives_multiple_polls(self): + with tempfile.TemporaryDirectory() as directory: + reference = Path(directory) / "android-generation-v1.json" + now = [0.0] + pulls = [] + + def pump(): + pulls.append(now[0]) + if len(pulls) == 3: + reference.write_text('{"run_id":"sy_fresh"}', encoding="utf-8") + + value = worker.wait_json( + reference, 10.0, + lambda candidate: candidate.get("run_id") == "sy_fresh", + pump, clock=lambda: now[0], + sleeper=lambda duration: now.__setitem__(0, now[0] + duration), + ) + self.assertEqual("sy_fresh", value["run_id"]) + self.assertEqual(3, len(pulls)) + + def test_stale_reference_is_ignored_until_correlated_replacement(self): + with tempfile.TemporaryDirectory() as directory: + reference = Path(directory) / "android-generation-v1.json" + reference.write_text('{"run_id":"sy_stale"}', encoding="utf-8") + now = [0.0] + polls = [0] + + def pump(): + polls[0] += 1 + if polls[0] == 2: + reference.write_text('{"run_id":"sy_fresh"}', encoding="utf-8") + + value = worker.wait_json( + reference, 1.0, + lambda candidate: candidate.get("run_id") == "sy_fresh", + pump, clock=lambda: now[0], + sleeper=lambda duration: now.__setitem__(0, now[0] + duration), + ) + self.assertEqual("sy_fresh", value["run_id"]) + + def test_missing_generation_reports_correlated_protocol_timeout(self): + with tempfile.TemporaryDirectory() as directory: + reference = Path(directory) / "android-generation-v1.json" + reference.write_text('{"run_id":"sy_stale"}', encoding="utf-8") + now = [0.0] + with self.assertRaisesRegex( + RuntimeError, "timeout waiting for correlated android-generation-v1.json" + ): + worker.wait_json( + reference, 0.2, + lambda candidate: candidate.get("run_id") == "sy_fresh", + clock=lambda: now[0], + sleeper=lambda duration: now.__setitem__(0, now[0] + duration), + ) + if __name__ == "__main__": unittest.main() diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py index 8bad60c..103374f 100644 --- a/tools/sync_peer_worker.py +++ b/tools/sync_peer_worker.py @@ -34,6 +34,30 @@ CAPS = { "session-preparations-v2", } +MTP_OPERATION_TIMEOUT_SECONDS = 30.0 +MTP_POLL_INTERVAL_SECONDS = 3.0 + + +class ConversationDeadlineExpired(RuntimeError): + """Internal signal that the protocol deadline, not the transport, expired.""" + + +class MtpPullPump: + """Throttle complete MTP refreshes while local correlation is polled.""" + + def __init__(self, pull, interval=MTP_POLL_INTERVAL_SECONDS, clock=time.monotonic): + self._pull = pull + self._interval = interval + self._clock = clock + self._last_pull = clock() + + def __call__(self, force: bool = False) -> None: + now = self._clock() + if not force and now - self._last_pull < self._interval: + return + self._pull() + self._last_pull = self._clock() + def emit(run_id: str, phase: str, **values: object) -> None: value = { @@ -97,9 +121,13 @@ def run_adapter( root: Path, deadline: float, allow_missing_peer: bool = False, + clock=time.monotonic, ) -> bool: - remaining = max(1, int(deadline - time.monotonic())) - timeout = min(remaining, 30) + remaining = deadline - clock() + if remaining <= 0: + raise ConversationDeadlineExpired() + timeout = min(remaining, MTP_OPERATION_TIMEOUT_SECONDS) + deadline_limited = remaining < MTP_OPERATION_TIMEOUT_SECONDS try: result = subprocess.run( [str(adapter), operation, peer, str(root)], @@ -109,10 +137,13 @@ def run_adapter( check=False, ) except subprocess.TimeoutExpired as error: + if deadline_limited: + raise ConversationDeadlineExpired() from error # CONTRACT: an adapter timeout is an ambiguous transport failure. It # never implies rollback, replay, or permission to retry a mutation. raise RuntimeError( - f"transport_timeout: MTP {operation} did not finish within {timeout} seconds" + f"transport_timeout: MTP {operation} did not finish within " + f"{MTP_OPERATION_TIMEOUT_SECONDS:g} seconds" ) from error if result.returncode != 0: diagnostic = ( @@ -248,29 +279,44 @@ def push_drive( run_drive_adapter(adapter, "push", remote, outbox, deadline) -def wait_file(path: Path, deadline: float, pump=None) -> None: +def wait_file(path: Path, deadline: float, pump=None, clock=time.monotonic, sleeper=time.sleep) -> None: while not path.is_file(): - if time.monotonic() >= deadline: + if clock() >= deadline: raise RuntimeError(f"timeout waiting for {path.name}") if pump is not None: - pump() - time.sleep(0.05) + try: + pump() + except ConversationDeadlineExpired as error: + raise RuntimeError(f"timeout waiting for {path.name}") from error + sleeper(0.05) -def wait_json(path: Path, deadline: float, predicate, pump=None) -> dict: +def wait_json( + path: Path, + deadline: float, + predicate, + pump=None, + clock=time.monotonic, + sleeper=time.sleep, +) -> dict: while True: - wait_file(path, deadline, pump) + wait_file(path, deadline, pump, clock, sleeper) try: value = json.loads(bounded(path)) except (OSError, json.JSONDecodeError): value = None if isinstance(value, dict) and predicate(value): return value - if time.monotonic() >= deadline: + if clock() >= deadline: raise RuntimeError(f"timeout waiting for correlated {path.name}") if pump is not None: - pump() - time.sleep(0.05) + try: + pump() + except ConversationDeadlineExpired as error: + raise RuntimeError( + f"timeout waiting for correlated {path.name}" + ) from error + sleeper(0.05) def refresh_mtp_peer( @@ -291,19 +337,24 @@ def refresh_mtp_peer( """ peer_path = root / "android-peer-v1.json" while True: - pulled = run_adapter( - adapter, - "pull", - peer, - root, - deadline, - True, - ) + try: + pulled = run_adapter( + adapter, + "pull", + peer, + root, + deadline, + True, + ) + except ConversationDeadlineExpired as error: + raise RuntimeError("timeout waiting for android-peer-v1.json") from error if pulled and peer_path.is_file(): return if time.monotonic() >= deadline: raise RuntimeError("timeout waiting for android-peer-v1.json") - time.sleep(0.05) + # A complete libmtp discovery is the expensive poll. Keep retries + # interactive without continuously reopening the USB/MTP session. + time.sleep(min(MTP_POLL_INTERVAL_SECONDS, max(0.0, deadline - time.monotonic()))) def correlated_peer_error(root: Path, run_id: str) -> None: @@ -360,20 +411,22 @@ def main() -> int: if args.mode == "mtp": if args.mtp_adapter is None: raise RuntimeError("MTP adapter is required") - pump = lambda: run_adapter( - args.mtp_adapter, - "pull", - args.expected_peer, - args.transport_root, - deadline, - True, - ) refresh_mtp_peer( args.mtp_adapter, args.expected_peer, args.transport_root, deadline, ) + pump = MtpPullPump( + lambda: run_adapter( + args.mtp_adapter, + "pull", + args.expected_peer, + args.transport_root, + deadline, + True, + ) + ) elif args.mode == "drive": if args.drive_adapter is None or not args.drive_remote: raise RuntimeError("Drive adapter and remote are required") diff --git a/tui/meson.build b/tui/meson.build index e97f20e..706b5e3 100644 --- a/tui/meson.build +++ b/tui/meson.build @@ -876,6 +876,11 @@ test( find_program('python3'), args: [meson.project_source_root() / 'tests/test_sync_orchestrator.py'], ) +test( + 'sync_peer_worker', + find_program('python3'), + args: [meson.project_source_root() / 'tests/test_sync_peer_worker.py'], +) test( 'sync_drive_transport', find_program('python3'),