Close Panama recovery race windows

config/dot/quickshell/scripts/panama-doctor

@@ -5,6 +5,8 @@
 from __future__ import annotations
 
 import argparse
+import ctypes
+import errno
 import json
 import os
 import re
@@ -23,6 +25,11 @@ from typing import Callable, Literal
 Status = Literal["ok", "warning", "error", "unconfigured"]
 Group = Literal["desktop-foundation", "input-media", "integrations", "panama-tools"]
 ActionKind = Literal["repair", "open", "instructions"]
+InhibitorRow = tuple[str, str, str, str, str, str, str, str]
+
+AT_FDCWD = -100
+RENAME_NOREPLACE = 1
+RENAME_EXCHANGE = 2
 
 
 @dataclass(frozen=True)
@@ -452,10 +459,10 @@ def check_caffeine(config: DoctorConfig) -> Check:
     result = run_command(("systemd-inhibit", "--list", "--no-pager", "--no-legend"), config)
     if result.state != "ok":
         return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe is unavailable.")
-    inhibitor_pids = parse_caffeine_pids(result.stdout, str(os.getuid()))
-    if inhibitor_pids is None:
+    inhibitor_rows = parse_caffeine_rows(result.stdout, str(os.getuid()))
+    if inhibitor_rows is None:
         return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.")
-    inhibitors = len(dict.fromkeys(inhibitor_pids))
+    inhibitors = len(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
     if inhibitors > 1:
         return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Duplicate Panama Caffeine inhibitors detected.", Action("repair", "Release duplicate inhibitors"))
     if inhibitors == 1:
@@ -542,29 +549,120 @@ def lexical_link_target(destination: Path) -> Path:
     return lexical_path(target if target.is_absolute() else destination.parent / target)
 
 
-def atomic_symlink_replace(destination: Path, source: Path) -> None:
-    """Install an authored sibling symlink without first removing destination."""
+def renameat2(source: Path, destination: Path, flags: int) -> None:
+    """Call Linux renameat2 with fixed flags selected by authored code."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    function = getattr(libc, "renameat2", None)
+    if function is None:
+        raise OSError(errno.ENOSYS, "renameat2 is unavailable")
+    function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
+    function.restype = ctypes.c_int
+    result = function(
+        AT_FDCWD,
+        os.fsencode(source),
+        AT_FDCWD,
+        os.fsencode(destination),
+        flags,
+    )
+    if result != 0:
+        error = ctypes.get_errno()
+        raise OSError(error, os.strerror(error), destination)
+
+
+def rename_exchange(source: Path, destination: Path) -> None:
+    renameat2(source, destination, RENAME_EXCHANGE)
+
+
+def rename_noreplace(source: Path, destination: Path) -> None:
+    renameat2(source, destination, RENAME_NOREPLACE)
+
+
+def create_symlink_candidate(destination: Path, source: Path) -> Path:
+    """Create one unpredictable authored sibling candidate symlink."""
     for _ in range(32):
-        temporary = destination.with_name(
+        candidate = destination.with_name(
             f".panama-link-{destination.name}-{os.getpid()}-{secrets.token_hex(8)}"
         )
-        created = False
         try:
-            os.symlink(source, temporary, target_is_directory=True)
-            created = True
-            os.replace(temporary, destination)
-            return
+            os.symlink(source, candidate, target_is_directory=True)
+            return candidate
         except FileExistsError:
             continue
-        finally:
-            if created:
-                try:
-                    temporary.unlink()
-                except FileNotFoundError:
-                    pass
     raise OSError("Could not allocate an authored temporary link")
 
 
+def cleanup_candidate(candidate: Path) -> None:
+    try:
+        candidate.unlink()
+    except FileNotFoundError:
+        pass
+
+
+def install_absent_symlink(destination: Path, source: Path) -> Literal["repaired", "blocked", "failed"]:
+    candidate = create_symlink_candidate(destination, source)
+    try:
+        try:
+            rename_noreplace(candidate, destination)
+        except FileExistsError:
+            return "blocked"
+        except OSError:
+            return "failed"
+        return "repaired"
+    finally:
+        cleanup_candidate(candidate)
+
+
+def exchange_owned_symlink(
+    destination: Path,
+    source: Path,
+    authored_sources: frozenset[Path],
+) -> Literal["repaired", "blocked", "failed"]:
+    """Exchange first, then validate the exact object removed from destination."""
+    candidate = create_symlink_candidate(destination, source)
+    exchanged = False
+    rolled_back = False
+    try:
+        try:
+            rename_exchange(candidate, destination)
+            exchanged = True
+        except OSError:
+            return "failed"
+
+        try:
+            old_is_authored = candidate.is_symlink() \
+                and lexical_link_target(candidate) in authored_sources
+        except OSError:
+            old_is_authored = False
+        if old_is_authored:
+            cleanup_candidate(candidate)
+            return "repaired"
+
+        try:
+            rename_exchange(candidate, destination)
+            rolled_back = True
+        except OSError:
+            # The displaced object remains at the unpredictable candidate path;
+            # never unlink it when rollback could not restore ownership.
+            return "failed"
+
+        try:
+            restored_candidate_is_ours = candidate.is_symlink() \
+                and lexical_link_target(candidate) == source
+        except OSError:
+            restored_candidate_is_ours = False
+        if not restored_candidate_is_ours:
+            return "failed"
+        cleanup_candidate(candidate)
+        return "blocked"
+    finally:
+        if not exchanged or rolled_back:
+            try:
+                if candidate.is_symlink() and lexical_link_target(candidate) == source:
+                    cleanup_candidate(candidate)
+            except OSError:
+                pass
+
+
 def repair_runtime_links(config: DoctorConfig) -> RepairResult:
     root = lexical_path(config.root)
     sources = [(name, lexical_path(config.root / relative_source)) for name, relative_source in RUNTIME_LINK_TARGETS]
@@ -591,13 +689,17 @@ def repair_runtime_links(config: DoctorConfig) -> RepairResult:
                 if current_target not in authored_sources:
                     blocked = True
                     continue
-                atomic_symlink_replace(destination, source)
+                outcome = exchange_owned_symlink(destination, source, authored_sources)
+                blocked = blocked or outcome == "blocked"
+                failed = failed or outcome == "failed"
             elif destination.exists():
                 # A regular file or directory is user-owned unless proven
                 # otherwise. Report it, but never replace it.
                 blocked = True
             else:
-                atomic_symlink_replace(destination, source)
+                outcome = install_absent_symlink(destination, source)
+                blocked = blocked or outcome == "blocked"
+                failed = failed or outcome == "failed"
         except OSError:
             failed = True
 
@@ -618,8 +720,8 @@ def repair_vicinae_commands(config: DoctorConfig) -> RepairResult:
     return RepairResult("panama.vicinae-commands", True, exit_code, message)
 
 
-def parse_caffeine_pids(output: str, uid: str) -> list[int] | None:
-    inhibitor_pids: list[int] = []
+def parse_caffeine_rows(output: str, uid: str) -> list[InhibitorRow] | None:
+    inhibitor_rows: list[InhibitorRow] = []
     for line in output.splitlines():
         parts = line.split()
         if len(parts) < 2 or parts[0] != "Panama" or parts[1] != uid:
@@ -632,8 +734,8 @@ def parse_caffeine_pids(output: str, uid: str) -> list[int] | None:
             continue
         if not parts[3].isdecimal():
             return None
-        inhibitor_pids.append(int(parts[3]))
-    return inhibitor_pids
+        inhibitor_rows.append(tuple(parts))
+    return inhibitor_rows
 
 
 def close_pidfds(pidfds: list[int]) -> None:
@@ -644,6 +746,31 @@ def close_pidfds(pidfds: list[int]) -> None:
             pass
 
 
+def signal_caffeine_pidfds(
+    pidfds: list[int],
+    sender: Callable[..., None] | None = None,
+) -> Literal["released", "preflight-failed", "incomplete"]:
+    send = sender or signal.pidfd_send_signal
+    for pidfd in pidfds:
+        try:
+            send(pidfd, 0, None, 0)
+        except (OSError, ValueError):
+            return "preflight-failed"
+
+    incomplete = False
+    for pidfd in pidfds:
+        try:
+            send(pidfd, signal.SIGTERM, None, 0)
+        except ProcessLookupError:
+            continue
+        except OSError as error:
+            if error.errno != errno.ESRCH:
+                incomplete = True
+        except ValueError:
+            incomplete = True
+    return "incomplete" if incomplete else "released"
+
+
 def repair_caffeine(config: DoctorConfig) -> RepairResult:
     list_command = ("systemd-inhibit", "--list", "--no-pager", "--no-legend")
     list_exit, output = run_repair_command(list_command, config)
@@ -651,10 +778,10 @@ def repair_caffeine(config: DoctorConfig) -> RepairResult:
         return RepairResult("panama.caffeine", True, list_exit, "Caffeine inhibitors could not be inspected.")
 
     uid = str(os.getuid())
-    parsed_pids = parse_caffeine_pids(output, uid)
-    if parsed_pids is None:
+    inhibitor_rows = parse_caffeine_rows(output, uid)
+    if inhibitor_rows is None:
         return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
-    inhibitor_pids = list(dict.fromkeys(parsed_pids))
+    inhibitor_pids = list(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
 
     if len(inhibitor_pids) <= 1:
         return RepairResult("panama.caffeine", True, 0, "No duplicate Panama Caffeine inhibitors needed release.")
@@ -673,15 +800,15 @@ def repair_caffeine(config: DoctorConfig) -> RepairResult:
         second_exit, second_output = run_repair_command(list_command, config)
         if second_exit != 0:
             return RepairResult("panama.caffeine", True, second_exit, "Caffeine inhibitors could not be revalidated; nothing was released.")
-        second_parsed = parse_caffeine_pids(second_output, uid)
-        if second_parsed is None or any(pid not in set(second_parsed) for pid in duplicates):
+        second_rows = parse_caffeine_rows(second_output, uid)
+        if second_rows is None or second_rows != inhibitor_rows:
             return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata changed; nothing was released.")
 
-        try:
-            for pidfd in pidfds:
-                signal.pidfd_send_signal(pidfd, signal.SIGTERM, None, 0)
-        except (OSError, ValueError):
+        signal_outcome = signal_caffeine_pidfds(pidfds)
+        if signal_outcome == "preflight-failed":
             return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
+        if signal_outcome == "incomplete":
+            return RepairResult("panama.caffeine", True, 1, "One or more duplicate inhibitors could not be released.")
     finally:
         close_pidfds(pidfds)
     return RepairResult("panama.caffeine", True, 0, "Duplicate Panama Caffeine inhibitors were released. A fresh health check will verify recovery.")