diff --git a/tests/test_syncd_routing.py b/tests/test_syncd_routing.py index 6d90c3b..fc6e866 100644 --- a/tests/test_syncd_routing.py +++ b/tests/test_syncd_routing.py @@ -126,6 +126,45 @@ 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): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + transport = root / "transport" + transport.mkdir() + config = root / "config.json" + config.write_text( + json.dumps( + { + "format": "trainlog-sync-orchestrator-config", + "version": 1, + "enabled": True, + "mode": "mtp", + "expected_peer_id": "peer_11111111-1111-4111-8111-111111111111", + "transport_root": str(transport), + "owned_root": str(root / "owned"), + "timeout_seconds": 30, + } + ) + ) + args = argparse.Namespace( + mtp_adapter=root / "adapter", + state=root / "trainlog.db.sync-run.json", + trigger="android", + orchestrator=root / "sync_orchestrator.py", + database=root / "trainlog.db", + ) + + def flock(_fd, operation): + if operation == syncd.fcntl.LOCK_EX | syncd.fcntl.LOCK_NB: + raise BlockingIOError + self.assertEqual(syncd.fcntl.LOCK_UN, operation) + + with mock.patch.object(syncd.fcntl, "flock", side_effect=flock), mock.patch.object( + syncd.subprocess, "run" + ) as run: + self.assertEqual(3, syncd.run_full_generation(args, config)) + run.assert_not_called() + def test_android_production_publishes_full_generation_before_optional_legacy(self): source = (ROOT / "android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt").read_text() production = source.split("internal suspend fun publishBundleAndRequest", 1)[1].split( diff --git a/tools/trainlog_syncd.py b/tools/trainlog_syncd.py index c35c2f8..116d8e6 100755 --- a/tools/trainlog_syncd.py +++ b/tools/trainlog_syncd.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import fcntl import json import os import re @@ -168,12 +169,32 @@ def run_full_generation(args: argparse.Namespace, config_path: Path) -> int: root = Path(config["transport_root"]) expected_peer = config["expected_peer_id"] adapter = args.mtp_adapter - pull = subprocess.run( - [str(adapter), "pull", expected_peer, str(root)], - capture_output=True, - text=True, - check=False, - ) + # 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" + 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: + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return 3 + pull = subprocess.run( + [str(adapter), "pull", expected_peer, str(root)], + capture_output=True, + text=True, + check=False, + ) + finally: + try: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + finally: + os.close(lock_fd) if pull.returncode != 0: return 3 _, seen = load_request_state(args.state)