From b580b6388602a954e9f51b5bdda621c5ab2d477b Mon Sep 17 00:00:00 2001 From: fy59 Date: Tue, 22 Sep 2026 14:10:47 +0200 Subject: [PATCH] fix(sync): serialize auto mtp probe --- tests/test_sync_orchestrator.py | 64 ++++++++++++++++++++++++++++++++- tools/sync_orchestrator.py | 45 +++++++++++++++++------ 2 files changed, 97 insertions(+), 12 deletions(-) diff --git a/tests/test_sync_orchestrator.py b/tests/test_sync_orchestrator.py index 7ba5245..6c9b2f5 100644 --- a/tests/test_sync_orchestrator.py +++ b/tests/test_sync_orchestrator.py @@ -1,4 +1,4 @@ -import json, os, signal, subprocess, tempfile, time, unittest, uuid +import fcntl, json, os, signal, subprocess, tempfile, time, unittest, uuid from pathlib import Path ROOT = Path(__file__).resolve().parents[1] @@ -134,6 +134,68 @@ class OrchestratorTest(unittest.TestCase): os.kill(pid, 0) self.assertEqual(json.loads(state.read_text())["phase"], "interrupted") + def test_auto_transport_waits_for_mtp_lock_instead_of_falling_back_to_drive(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + sandbox = root / "program" + sandbox.mkdir() + (sandbox / "sync_orchestrator.py").write_bytes(ORCHESTRATOR.read_bytes()) + peer, desktop = identity("peer_"), identity("peer_") + inbound, outbound = identity("gen_"), identity("gen_") + report = { + "format": "trainlog-sync-worker-report", "version": 1, + "run_id": "RUN", "producer_peer_id": peer, "consumer_peer_id": desktop, + "inbound_generation_id": inbound, "outbound_generation_id": outbound, + "manifest_sha256": "a" * 64, "result": "completed", + "sessions_reconciled": 0, "domains": {}, "drafts": [], + "ai_midpoint": {"result": "not_configured"}, + "ai_post_sync": {"result": "not_configured"}, + } + (sandbox / "sync_peer_worker.py").write_text( + "import json,sys\nv=" + repr(report) + + "\nv['run_id']=sys.argv[sys.argv.index('--run-id')+1]\nprint(json.dumps(v))\n" + ) + adapter = root / "build/tui/trainlog-generation-mtp-adapter" + adapter.parent.mkdir(parents=True) + marker = root / "mtp-probe" + adapter.write_text(f"#!/bin/sh\necho probe >> {marker}\nexit 0\n") + adapter.chmod(0o755) + values = self.fixture( + root, + version=2, + mode="auto", + drive_enabled=True, + drive_remote="fake:Trainlog/Sync/v1", + ) + db, state, config, run, request = values + mtp_lock = root / "mtp.lock" + lock_fd = os.open(mtp_lock, os.O_RDWR | os.O_CREAT, 0o600) + fcntl.flock(lock_fd, fcntl.LOCK_EX) + process = subprocess.Popen( + [ + "python3", str(sandbox / "sync_orchestrator.py"), + "--database", str(db), "--state", str(state), + "--config", str(config), "--run-id", run, "--request-id", request, + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + time.sleep(0.2) + self.assertIsNone(process.poll()) + self.assertFalse(marker.exists()) + finally: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + os.close(lock_fd) + stdout, stderr = process.communicate(timeout=10) + self.assertEqual(process.returncode, 0, stderr or stdout) + self.assertTrue(marker.is_file()) + final = json.loads(state.read_text()) + self.assertEqual("completed", final["phase"]) + self.assertEqual("mtp", final["transport"]) + self.assertEqual("success", final["usb_state"]) + def test_committed_primary_run_survives_drive_mirror_failure(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/tools/sync_orchestrator.py b/tools/sync_orchestrator.py index 0d1abba..d89b29a 100644 --- a/tools/sync_orchestrator.py +++ b/tools/sync_orchestrator.py @@ -403,17 +403,40 @@ def run(args: argparse.Namespace) -> int: if selected_mode == "auto": # WHY: cable presence is not peer availability. A successful pull # is the bounded proof that the configured Trainlog peer is usable. - try: - probe = subprocess.run( - [str(adapter), "pull", config["expected_peer_id"], config["transport_root"]], - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - timeout=min(timeout, 30), - check=False, - ) if adapter.is_absolute() and os.access(adapter, os.X_OK) else None - except (OSError, subprocess.TimeoutExpired): - probe = None + # The daemon performs the same libmtp operation for Android-origin + # discovery, so auto probing must share the dedicated transport + # lock rather than racing libmtp and falsely falling back to Drive. + probe = None + if adapter.is_absolute() and os.access(adapter, os.X_OK): + mtp_lock_path = Path(config["transport_root"]).parent / "mtp.lock" + mtp_lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + mtp_lock_fd = os.open(mtp_lock_path, os.O_RDWR | os.O_CREAT, 0o600) + probe_deadline = time.monotonic() + min(timeout, 30) + try: + while time.monotonic() < probe_deadline: + try: + fcntl.flock(mtp_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + time.sleep(0.05) + else: + raise subprocess.TimeoutExpired([str(adapter), "pull"], min(timeout, 30)) + remaining = max(0.01, probe_deadline - time.monotonic()) + probe = subprocess.run( + [str(adapter), "pull", config["expected_peer_id"], config["transport_root"]], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=remaining, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + probe = None + finally: + try: + fcntl.flock(mtp_lock_fd, fcntl.LOCK_UN) + finally: + os.close(mtp_lock_fd) selected_mode = "mtp" if probe is not None and probe.returncode == 0 else "drive" command = [ sys.executable,