fix(sync): throttle MTP polling and preserve timeout cause

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()

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")

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'),