diff --git a/tests/test_sync_error_classification.py b/tests/test_sync_error_classification.py index dc4bff3..0381275 100644 --- a/tests/test_sync_error_classification.py +++ b/tests/test_sync_error_classification.py @@ -129,18 +129,21 @@ class SyncErrorClassificationTest(unittest.TestCase): def test_transport_timeout_has_a_stable_code_and_bounded_diagnostic(self): timeout = subprocess.TimeoutExpired(["adapter", "push"], 30) - with mock.patch.object(sync_peer_worker.subprocess, "run", side_effect=timeout): - with self.assertRaisesRegex( - RuntimeError, - r"^transport_timeout: MTP push did not finish within 30 seconds$", - ) as caught: - sync_peer_worker.run_adapter( - Path("/adapter"), - "push", - "peer_00000000-0000-4000-8000-000000000001", - Path("/transport"), - sync_peer_worker.time.monotonic() + 60, - ) + with tempfile.TemporaryDirectory() as directory: + transport = Path(directory) / "transport" + transport.mkdir() + with mock.patch.object(sync_peer_worker.subprocess, "run", side_effect=timeout): + with self.assertRaisesRegex( + RuntimeError, + r"^transport_timeout: MTP push did not finish within 30 seconds$", + ) as caught: + sync_peer_worker.run_adapter( + Path("/adapter"), + "push", + "peer_00000000-0000-4000-8000-000000000001", + transport, + sync_peer_worker.time.monotonic() + 60, + ) self.assertEqual(sync_orchestrator.failure_code(caught.exception), "transport_timeout") def test_known_operational_failures_remain_distinct(self): diff --git a/tests/test_sync_peer_worker.py b/tests/test_sync_peer_worker.py index 624b678..c328796 100644 --- a/tests/test_sync_peer_worker.py +++ b/tests/test_sync_peer_worker.py @@ -17,6 +17,14 @@ SPEC.loader.exec_module(worker) class SyncPeerWorkerTest(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory(prefix="trainlog-peer-worker-test-") + self.transport = Path(self.temporary.name) / "transport" + self.transport.mkdir() + + def tearDown(self): + self.temporary.cleanup() + def test_drive_adapter_uses_remaining_conversation_deadline(self): completed = subprocess.CompletedProcess([], 0, b"", b"") with mock.patch.object(worker.subprocess, "run", return_value=completed) as run: @@ -24,7 +32,7 @@ class SyncPeerWorkerTest(unittest.TestCase): Path("/fixed/adapter.py"), "pull", "remote:Trainlog/Sync-Test/run", - Path("/private/transport"), + self.transport, time.monotonic() + 180, ) timeout = run.call_args.kwargs["timeout"] @@ -36,10 +44,25 @@ class SyncPeerWorkerTest(unittest.TestCase): 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.transport, 100.0, clock=lambda: 50.0, ) self.assertEqual(worker.MTP_OPERATION_TIMEOUT_SECONDS, run.call_args.kwargs["timeout"]) + def test_mtp_adapter_uses_and_releases_shared_transport_lock(self): + completed = subprocess.CompletedProcess([], 0, b"", b"") + real_flock = worker.fcntl.flock + with mock.patch.object(worker.subprocess, "run", return_value=completed), mock.patch.object( + worker.fcntl, "flock", wraps=real_flock + ) as flock: + worker.run_adapter( + Path("/fixed/adapter"), "pull", "peer_fixed", + self.transport, 100.0, clock=lambda: 50.0, + ) + self.assertTrue((self.transport.parent / "mtp.lock").is_file()) + operations = [call.args[1] for call in flock.call_args_list] + self.assertIn(worker.fcntl.LOCK_EX | worker.fcntl.LOCK_NB, operations) + self.assertEqual(worker.fcntl.LOCK_UN, operations[-1]) + def test_mtp_pull_pump_is_throttled(self): now = [0.0] pulls = [] @@ -55,7 +78,7 @@ class SyncPeerWorkerTest(unittest.TestCase): with self.assertRaises(worker.ConversationDeadlineExpired): worker.run_adapter( Path("/fixed/adapter"), "pull", "peer_fixed", - Path("/private/transport"), 10.0, clock=lambda: 10.0, + self.transport, 10.0, clock=lambda: 10.0, ) def test_deadline_limited_pull_becomes_protocol_timeout(self): @@ -65,7 +88,7 @@ class SyncPeerWorkerTest(unittest.TestCase): with self.assertRaises(worker.ConversationDeadlineExpired): worker.run_adapter( Path("/fixed/adapter"), "pull", "peer_fixed", - Path("/private/transport"), 10.5, clock=lambda: 10.0, + self.transport, 10.5, clock=lambda: 10.0, ) def test_blocked_full_budget_pull_is_transport_timeout(self): @@ -75,7 +98,7 @@ class SyncPeerWorkerTest(unittest.TestCase): with self.assertRaisesRegex(RuntimeError, "transport_timeout"): worker.run_adapter( Path("/fixed/adapter"), "pull", "peer_fixed", - Path("/private/transport"), 100.0, clock=lambda: 50.0, + self.transport, 100.0, clock=lambda: 50.0, ) def test_delayed_correlated_generation_survives_multiple_polls(self): diff --git a/tests/test_syncd_routing.py b/tests/test_syncd_routing.py index fc6e866..5f99f12 100644 --- a/tests/test_syncd_routing.py +++ b/tests/test_syncd_routing.py @@ -126,7 +126,7 @@ class SyncdRoutingTest(unittest.TestCase): self.assertEqual(1, len(commands)) self.assertEqual("sy_old", json.loads(state.read_text())["run_id"]) - def test_daemon_skips_mtp_probe_while_canonical_sync_lock_is_busy(self): + def test_daemon_skips_mtp_probe_while_transport_lock_is_busy(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) transport = root / "transport" diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py index 103374f..c0020b3 100644 --- a/tools/sync_peer_worker.py +++ b/tools/sync_peer_worker.py @@ -9,6 +9,7 @@ always produced and consumed by the existing generation services. from __future__ import annotations import argparse +import fcntl import hashlib import json import os @@ -123,28 +124,46 @@ def run_adapter( allow_missing_peer: bool = False, clock=time.monotonic, ) -> bool: - remaining = deadline - clock() - if remaining <= 0: - raise ConversationDeadlineExpired() - timeout = min(remaining, MTP_OPERATION_TIMEOUT_SECONDS) - deadline_limited = remaining < MTP_OPERATION_TIMEOUT_SECONDS + lock_path = root.parent / "mtp.lock" + lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) try: - result = subprocess.run( - [str(adapter), operation, peer, str(root)], - stdin=subprocess.DEVNULL, - capture_output=True, - timeout=timeout, - 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 " - f"{MTP_OPERATION_TIMEOUT_SECONDS:g} seconds" - ) from error + while True: + remaining = deadline - clock() + if remaining <= 0: + raise ConversationDeadlineExpired() + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + time.sleep(min(0.05, remaining)) + 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)], + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=timeout, + 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 " + f"{MTP_OPERATION_TIMEOUT_SECONDS:g} seconds" + ) from error + finally: + try: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + finally: + os.close(lock_fd) if result.returncode != 0: diagnostic = ( result.stderr.decode("utf-8", "replace")[:1024] or "MTP adapter failed" diff --git a/tools/trainlog_syncd.py b/tools/trainlog_syncd.py index 116d8e6..59239bd 100755 --- a/tools/trainlog_syncd.py +++ b/tools/trainlog_syncd.py @@ -170,13 +170,13 @@ def run_full_generation(args: argparse.Namespace, config_path: Path) -> int: expected_peer = config["expected_peer_id"] adapter = args.mtp_adapter # WHY: the daemon must probe MTP to discover Android-origin requests, but - # Web/TUI orchestration owns the same physical transport while it holds the - # canonical sync lock. Probing outside that lock lets libmtp race itself - # and fail with "device is busy". CONTRACT: idle discovery takes the same - # lock non-blockingly and simply skips this poll when another conversation - # is active. INVARIANT: the lock is released before launching the - # orchestrator, which remains the sole owner of business-sync locking. - lock_path = args.database.parent / "sync.lock" + # Web/TUI orchestration can use the same libmtp device concurrently. A + # business sync.lock is too broad here: holding it during discovery makes + # a user Web click fail with sync_in_progress. CONTRACT: discovery instead + # shares a dedicated physical-transport lock with every worker MTP adapter + # call and skips this poll when that transport is busy. INVARIANT: no MTP + # adapter processes overlap, while business-sync admission stays independent. + lock_path = root.parent / "mtp.lock" lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) try: