From 6765fbcb207a16937fcb181ecf5c84d287a9c110 Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 28 Sep 2026 21:28:10 +1000 Subject: [PATCH] Bundled KDE Connect: close the races the second review found - The need-packages retry only runs if no start() has taken over, so two agents can't end up running for one session. - Each copy goes to its own incoming folder; the agent removes its own once connected (and any a cancelled start left over an hour ago), so a stopped start can't delete files another is copying or still needs. - A missing package folder means need-packages, not an error. - Tests keep fake agents running, so they check ready and which agent owns the session, plus a bounded retry and the tidy rule. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_input.py | 82 +++++++++++++++++++++++++++++++---------- ui/frame_input_agent.py | 23 ++++++++---- ui/server.py | 10 ++--- 3 files changed, 84 insertions(+), 31 deletions(-) diff --git a/tests/test_input.py b/tests/test_input.py index f071ee9..4f7e66c 100644 --- a/tests/test_input.py +++ b/tests/test_input.py @@ -173,9 +173,11 @@ class Bundled(unittest.TestCase): self.deliver(frame_has=False, damaged=True) def lifecycle(self, agent_lines, deliver=None): - """An InputAgent whose ssh and agent are fakes; returns it and the folders each launch used.""" - launches = [] - test = self + """An InputAgent whose ssh and agent are fakes. Returns it, the folders each launch + used, and the fake agents. A fake agent reports its lines, then keeps running + (unless its lines end with need-packages) until the test ends.""" + launches, procs, done = [], [], threading.Event() + self.addCleanup(done.set) class Agent(self.server.InputAgent): def deliver(self, report, force=False): @@ -191,34 +193,60 @@ class Bundled(unittest.TestCase): stdin = None def __init__(self, lines): - self.stdout = iter(lines) + self.lines, self.ended = lines, threading.Event() + procs.append(self) + + @property + def stdout(self): + yield from self.lines + if b"need-packages" not in self.lines[-1]: + while not (done.is_set() or self.ended.is_set()): + self.ended.wait(0.02) + self.ended.set() def wait(self): + self.ended.wait(5) return 0 def poll(self): - return None + return 0 if self.ended.is_set() else None def terminate(self): - pass + self.ended.set() def popen(*a, **k): return Proc(agent_lines.pop(0) if agent_lines else [b'{"state": "ready"}\n']) old = self.server.ensure_master, self.server.subprocess.Popen self.server.ensure_master, self.server.subprocess.Popen = lambda: None, popen - test.addCleanup(lambda: (setattr(self.server, "ensure_master", old[0]), + self.addCleanup(lambda: (setattr(self.server, "ensure_master", old[0]), setattr(self.server.subprocess, "Popen", old[1]))) - return Agent(packages=[("a.pkg.tar.zst", "0" * 64)]), launches + return Agent(packages=[("a.pkg.tar.zst", "0" * 64)]), launches, procs + + def wait_for(self, check): + for _ in range(250): + if check(): + return + time.sleep(0.02) + self.fail("timed out") def test_agent_asking_for_packages_gets_them_and_starts_again(self): - agent, launches = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']]) - agent._launch(agent.generation) + agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']]) + agent.start() + self.wait_for(lambda: agent.status == {"state": "ready"}) self.assertEqual(launches, ["", "~/copied"]) - self.assertNotIn("reach", agent.status.get("message", "")) + self.assertIs(agent.proc, procs[1]) + + def test_asking_twice_is_an_error_not_a_loop(self): + need = [b'{"state": "need-packages"}\n'] + agent, launches, _ = self.lifecycle([need, list(need)]) + agent.start() + self.wait_for(lambda: agent.status.get("state") == "error") + self.assertEqual(launches, ["", "~/copied"]) + self.assertIn("didn't reach", agent.status["message"]) def test_start_after_stop_during_the_copy_still_starts(self): gate, entered = threading.Event(), threading.Event() - agent, launches = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5))) + agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5))) agent.start() self.assertTrue(entered.wait(5)) agent.stop() @@ -226,11 +254,17 @@ class Bundled(unittest.TestCase): agent.start() # while the first launch is still copying self.assertTrue(entered.wait(5), "the second start didn't launch") gate.set() - for _ in range(100): - if len(launches) == 2: - break - time.sleep(0.02) - self.assertEqual(len(launches), 2) # the stopped launch ran its agent too, then ended it + self.wait_for(lambda: len(procs) == 2 and agent.status == {"state": "ready"}) + # The stopped launch's agent was ended; the new one is the one in use. + self.wait_for(lambda: sum(p.ended.is_set() for p in procs) == 1) + self.assertFalse(agent.proc.ended.is_set()) + agent.stop() + self.wait_for(lambda: all(p.ended.is_set() for p in procs)) + + def test_copies_go_to_a_folder_of_their_own(self): + first, _, _ = self.deliver(frame_has=False) + second, _, _ = self.deliver(frame_has=False) + self.assertNotEqual(first, second) @unittest.skipIf(sys.platform == "win32", "the agent runs on the Frame (Linux)") @@ -296,11 +330,21 @@ class AgentInstall(unittest.TestCase): def test_asks_for_packages_it_was_not_sent(self): self.agent.listening, old = (lambda: False), self.agent.listening try: - with self.assertRaises(self.agent.NeedPackages): - self.agent.ensure_daemon("", [("new.pkg.tar.zst", "1" * 64)]) + for folder in ("", str(self.dir / "gone")): # none sent, or already tidied away + with self.assertRaises(self.agent.NeedPackages): + self.agent.ensure_daemon(folder, [("new.pkg.tar.zst", "1" * 64)]) finally: self.agent.listening = old + def test_tidy_keeps_other_starts_copies(self): + incoming = self.agent.BASE / "incoming" + mine, theirs, stale = incoming / "mine", incoming / "theirs", incoming / "stale" + for d in (mine, theirs, stale): + d.mkdir(parents=True) + os.utime(stale, (time.time() - 7200,) * 2) + self.agent.tidy_incoming(str(mine)) + self.assertEqual(sorted(p.name for p in incoming.iterdir()), ["theirs"]) + def test_server_and_agent_agree_on_the_stamp(self): import server packages = [("a.pkg.tar.zst", "1" * 64), ("b.pkg.tar.zst", "2" * 64)] diff --git a/ui/frame_input_agent.py b/ui/frame_input_agent.py index e7c1785..2c45a04 100644 --- a/ui/frame_input_agent.py +++ b/ui/frame_input_agent.py @@ -166,7 +166,7 @@ def ensure_daemon(folder, packages): """ system = SYSTEM_DAEMON.exists() if not system and not installed(packages) and not (listening() and our_daemons()): - if not folder: + if not folder or not Path(folder).is_dir(): raise NeedPackages() install(folder, packages) if listening(): @@ -346,6 +346,20 @@ def main(): stop_daemon() +def tidy_incoming(folder): + """Remove this start's copy of the packages, and any a cancelled start left over an hour ago.""" + incoming = BASE / "incoming" + if folder.startswith(str(incoming) + "/"): + shutil.rmtree(folder, ignore_errors=True) + try: + for old in incoming.iterdir(): + if time.time() - old.stat().st_mtime > 3600: + shutil.rmtree(old, ignore_errors=True) + incoming.rmdir() + except OSError: + pass # none, or another start's copy is still there + + class daemon_lock: """Installing, starting and restarting KDE Connect happen one agent at a time.""" @@ -361,12 +375,6 @@ def run(client, name, folder, packages): try: with daemon_lock(): ensure_daemon(folder, packages) - if folder.startswith(str(BASE / "incoming") + "/"): - shutil.rmtree(folder, ignore_errors=True) # unpacked; the copy isn't needed again - try: - (BASE / "incoming").rmdir() - except OSError: - pass # another device's copy is still there device, cert, key = identity(client) say("pairing") seen = our_daemons() @@ -390,6 +398,7 @@ def run(client, name, folder, packages): except (OSError, RuntimeError, subprocess.SubprocessError) as e: say("error", message=str(e)) return 1 + tidy_incoming(folder) # unpacked or not needed: the copy has done its job say("ready", keyboard=link.keyboard is not False) stdin, pending = sys.stdin.fileno(), b"" sel = selectors.DefaultSelector() diff --git a/ui/server.py b/ui/server.py index f48d49a..4ab2425 100755 --- a/ui/server.py +++ b/ui/server.py @@ -635,8 +635,8 @@ class InputAgent: " && echo yes || true", stdin=kdeconnect_stamp(self.packages).encode(), text=False, timeout=20) if have.strip() == b"yes": return "" - client = "".join(c for c in input_client() if c.isalnum() or c in "-_")[:64] or "default" - folder = f"{KDECONNECT_HOME}/incoming/{client}" + # A folder of its own: a cancelled start's agent may still be cleaning up another. + folder = f"{KDECONNECT_HOME}/incoming/{secrets.token_hex(8)}" ssh(f"mkdir -p {folder}", timeout=20) for name, sha in self.packages: path = KDECONNECT / "packages" / name @@ -645,8 +645,7 @@ class InputAgent: " (a build runs app/build/fetch-deps.js to add it)", 500) report(f"Copying KDE Connect to the Frame ({name.rsplit('-', 3)[0]})") quoted = shlex.quote(name) - # Its own temporary name: two starts can be copying at once. - ssh(f"cd {folder} && cat > {quoted}.part$$ && mv {quoted}.part$$ {quoted}", + ssh(f"cd {folder} && cat > {quoted}.part && mv {quoted}.part {quoted}", stdin=path.read_bytes(), text=False, timeout=600) return f"~/{folder}" @@ -693,7 +692,8 @@ class InputAgent: # It needed the packages after all (another device changed what's # installed after we looked): copy them and start once more. with self.lock: - if self.proc is not proc or self.generation != generation: + # Unless a start() already took over (it launches, and copies if still needed). + if self.proc is not proc or self.generation != generation or self.launching is not None: return self.proc, self.launching, self.status = None, generation, {"state": "starting"} self._launch(generation, force=True)