feat(auth): auth.json 跨进程写锁 + runtime 真身损坏回滚

orzmc/orzmc/infra/auth_store.py

@@ -10,6 +10,7 @@
 import os
 
 from orzmc.domain.auth import AuthSession
+from orzmc.infra.filelock import DEFAULT_TIMEOUT, STALE_AFTER_SECONDS, FileLock, lock_path
 from orzmc.infra.fs import FileStore, default_state_dir
 
 FILENAME = "auth.json"
@@ -46,3 +47,11 @@ def save(self, session: AuthSession, client_id: str) -> None:
 
     def clear(self) -> None:
         self._fs.remove(self._path)
+
+    def lock(self, *, timeout: float = DEFAULT_TIMEOUT, stale_after: float = STALE_AFTER_SECONDS) -> FileLock:
+        """跨进程锁 (``frontends.md`` F4.1): 刷新令牌的整个「读->换新->写」要持它.
+
+        与 Swift 侧算法逐字一致 (``<stateDir>/auth.lock`` 目录锁); 拿不到锁会抛
+        :class:`~orzmc.infra.filelock.LockTimeout` (``E_AUTH_LOCKED``).
+        """
+        return FileLock(lock_path(os.path.dirname(self._path)), timeout=timeout, stale_after=stale_after)

orzmc/orzmc/infra/filelock.py

@@ -0,0 +1,154 @@
+"""跨进程文件锁(`frontends.md` F4.1):`<stateDir>/auth.lock` 目录锁。
+
+**算法必须与 Swift 侧逐字一致**,否则两边互不加锁。核心只有三点:
+
+1. **获取**:`os.mkdir(lock)` —— 在 POSIX 与 Windows 上都是原子的;
+2. **陈旧检测**:目录 `mtime` 超过 `stale_after`(默认 30s)视为持锁进程已崩,删掉重试;
+3. **超时**:最多等 `timeout`(默认 10s),拿不到就报 ``E_AUTH_LOCKED``——
+   **绝不**在没拿到锁的情况下进临界区(令牌刷新会在网络请求期间持有它)。
+
+为什么不用 `flock`:`flock` 与 `fcntl` 锁在 Linux/macOS 上属于不同命名空间,跨语言极易"看起来加了锁,
+其实互不可见";Windows 又没有 `flock`。`mkdir` 三平台语义一致,代价是崩溃后要陈旧检测兜底。
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from types import TracebackType
+
+STALE_AFTER_SECONDS = 30.0
+DEFAULT_TIMEOUT = 10.0
+POLL_INTERVAL = 0.05
+
+
+class LockTimeout(RuntimeError):
+    """拿不到锁(规范错误码 ``E_AUTH_LOCKED``)。"""
+
+    code = "E_AUTH_LOCKED"
+
+
+def lock_path(state_dir: str) -> str:
+    """锁目录 `<stateDir>/auth.lock`(**目录**,靠 `mkdir` 原子性)。"""
+    return os.path.join(state_dir, "auth.lock")
+
+
+def _age_seconds(path: str, now: float | None = None) -> float:
+    try:
+        mtime = os.stat(path).st_mtime
+    except OSError:
+        return 0.0
+    return (now if now is not None else time.time()) - mtime
+
+
+def _is_stale(path: str, stale_after: float, now: float | None = None) -> bool:
+    return _age_seconds(path, now) > stale_after
+
+
+def _break_stale(path: str) -> None:
+    """清掉陈旧锁目录(先删 owner,再 rmdir;失败就算了,下轮还会重试)。"""
+    try:
+        owner = os.path.join(path, "owner")
+        if os.path.exists(owner):
+            os.remove(owner)
+        os.rmdir(path)
+    except OSError:
+        pass
+
+
+class FileLock:
+    """目录锁上下文管理器(`with FileLock(path): ...`)。"""
+
+    def __init__(
+        self,
+        path: str,
+        *,
+        timeout: float = DEFAULT_TIMEOUT,
+        stale_after: float = STALE_AFTER_SECONDS,
+        sleep=time.sleep,
+        clock=time.time,
+    ) -> None:
+        self.path = path
+        self.timeout = timeout
+        self.stale_after = stale_after
+        self._sleep = sleep
+        self._clock = clock
+        self._held = False
+
+    # ── 获取 / 释放 ────────────────────────────────────────────────────────
+
+    def acquire(self) -> None:
+        deadline = self._clock() + self.timeout
+        while True:
+            try:
+                os.mkdir(self.path)
+            except FileExistsError:
+                if _is_stale(self.path, self.stale_after, self._clock()):
+                    _break_stale(self.path)
+                    continue
+                if self._clock() >= deadline:
+                    raise LockTimeout(
+                        f"拿不到认证锁 {self.path}(另一进程正在刷新令牌);稍后重试,或确认没有卡住的 orzmc 进程",
+                    ) from None
+                self._sleep(POLL_INTERVAL)
+                continue
+            except OSError as exc:  # 权限 / 磁盘满等:不该静默继续
+                raise LockTimeout(f"无法创建认证锁 {self.path}: {exc}") from exc
+            self._write_owner()
+            self._held = True
+            return
+
+    def _write_owner(self) -> None:
+        """写持有者信息(pid + 起始时间)——仅供排查,不参与判定。"""
+        try:
+            with open(os.path.join(self.path, "owner"), "w", encoding="utf-8") as handle:
+                handle.write(f"pid={os.getpid()}\nsince={_iso_now()}\n")
+        except OSError:
+            pass
+
+    def release(self) -> None:
+        if not self._held:
+            return
+        self._held = False
+        try:
+            owner = os.path.join(self.path, "owner")
+            if os.path.exists(owner):
+                os.remove(owner)
+        except OSError:
+            pass
+        try:
+            os.rmdir(self.path)
+        except OSError:
+            # 释放失败不抛:陈旧检测会在 30s 后兜底。
+            pass
+
+    # ── 上下文管理器 ──────────────────────────────────────────────────────
+
+    def __enter__(self) -> FileLock:
+        self.acquire()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.release()
+
+
+def _iso_now() -> str:
+    """UTC ISO8601(秒精度)——写进 ``owner`` 供排查,不参与判定。"""
+    from datetime import datetime, timezone
+
+    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+__all__ = [
+    "DEFAULT_TIMEOUT",
+    "POLL_INTERVAL",
+    "STALE_AFTER_SECONDS",
+    "FileLock",
+    "LockTimeout",
+    "lock_path",
+]

orzmc/orzmc/services/auth.py

@@ -52,16 +52,26 @@ def __init__(
 
     # ── 对外入口 ────────────────────────────────────────────────────────────
     def ensure(self) -> AuthSession:
-        """返回一个 MC 令牌有效的会话:缓存 → 刷新 → 设备码登录。"""
+        """返回一个 MC 令牌有效的会话:缓存 → 刷新 → 设备码登录。
+
+        **刷新必须持跨进程锁, 并在锁内先重读** (`frontends.md` F4/F4.1):
+        刷新会轮换 refresh token,两个前端同时刷会让后者的令牌立刻失效(用户被登出)。
+        """
         session = self._store.load()
         if session is not None and auth_map.token_state(session.expires_at, self._clock()).state == "valid":
             return replace(session, source="cache")
         if session is not None:
-            try:
-                return self.refresh(session.msa)
-            except AuthError:
-                # refresh token 被拒(如用户改密/撤销授权)→ 清掉重登,不吞错。
-                self._store.clear()
+            with self._store.lock():
+                # 锁内**重读**:另一个前端可能刚刷完,直接用它的结果,别再刷一次。
+                current = self._store.load()
+                if current is not None and auth_map.token_state(current.expires_at, self._clock()).state == "valid":
+                    return replace(current, source="cache")
+                target = current if current is not None else session
+                try:
+                    return self.refresh(target.msa)
+                except AuthError:
+                    # refresh token 被拒(如用户改密/撤销授权)→ 清掉重登,不吞错。
+                    self._store.clear()
         return self.login()
 
     def login(self) -> AuthSession:

orzmc/orzmc/services/runtime.py

@@ -286,13 +286,68 @@ def resolve(self) -> tuple[str, str]:
         record = self._record()
         recorded = record.get("binary", "")
         source = record.get("source", "")
-        if recorded and _is_swift_payload(source, record.get("channel", "")) and self._runnable(recorded):
-            return recorded, source
         own = self.path
+        if recorded and _is_swift_payload(source, record.get("channel", "")) and self._runnable(recorded):
+            if recorded != own:
+                # 别人装的 Swift 原生版:直接用(它的备份与更新由那条通道负责,我们不插手)。
+                return recorded, source
+            # 是**我们自己的副本**:跑之前做魔数预检,坏了就用 .bak 回滚(§10.1)。
+            usable = self._verified_payload(recorded)
+            if usable is not None:
+                return usable, source
+            return own, source  # 坏且无备份:交给 forward 报 E_RUNTIME 并提示重装
         if self._fs.is_file(own):
-            return own, (source if recorded == own else f"github:{REPO}")
+            usable = self._verified_payload(own)
+            if usable is not None:
+                return usable, (source if recorded == own else f"github:{REPO}")
         return own, f"github:{REPO}"
 
+    @property
+    def backup_path(self) -> str:
+        """旧真身的备份路径 (同一目录 rename, 跨盘不会发生)."""
+        return self.path + ".bak"
+
+    def _replace_payload(self, staged: str) -> None:
+        """把新真身原子换上去; **先把旧的改名成 .bak**, 失败则还原 (规范 §10.1 回滚)."""
+        backup = self.backup_path
+        had_old = self._fs.is_file(self.path)
+        if had_old:
+            try:
+                if self._fs.is_file(backup):
+                    os.remove(backup)
+                os.replace(self.path, backup)
+            except OSError as exc:  # 备份失败就别动现有的 (安全优先)
+                raise RuntimeChannelError(f"无法备份旧真身: {exc}", "E_RUNTIME") from exc
+        try:
+            os.replace(staged, self.path)
+        except OSError as exc:
+            if had_old:
+                try:
+                    os.replace(backup, self.path)
+                except OSError:
+                    pass
+            raise RuntimeChannelError(f"写入真身失败: {exc}", "E_RUNTIME") from exc
+
+    def _recover_from_backup(self, reason: str) -> str | None:
+        """用 `.bak` 还原真身; 成功返回路径, 否则 ``None`` (规范 §10.1 回滚)."""
+        backup = self.backup_path
+        if not (self._fs.is_file(backup) and looks_like_executable(backup, self._windows)):
+            return None
+        try:
+            os.replace(backup, self.path)
+        except OSError:
+            return None
+        self._reporter.warn(f"真身不可用({reason}),已回滚到上一份: {self.path}")
+        return self.path
+
+    def _verified_payload(self, path: str) -> str | None:
+        """跑之前的魔数预检: 坏文件 + 好 `.bak` -> 回滚, 返回可用路径."""
+        if not self._fs.is_file(path):
+            return None
+        if looks_like_executable(path, self._windows):
+            return path
+        return self._recover_from_backup("魔数校验失败")
+
     def _runnable(self, path: str) -> bool:
         if not self._fs.is_file(path):
             return False
@@ -358,7 +413,11 @@ def install(self, version: str | None = None) -> RuntimeStatus:
         """
         recorded = self._record()
         native = recorded.get("binary", "")
-        reusable = _is_swift_payload(recorded.get("source", ""), recorded.get("channel", ""))
+        # 只在「**别人装的** Swift 原生安装」时 no-op(别在同一台机器上再存一份);
+        # 本通道自己的副本(channel=wrapper)要允许重装/升级,否则永远换不了新版本。
+        reusable = recorded.get("channel") != WRAPPER_CHANNEL and _is_swift_payload(
+            recorded.get("source", ""), recorded.get("channel", "")
+        )
         if native and reusable and self._runnable(native):
             self._reporter.info(f"复用已有的 Swift 安装: {native}")
             return self.status()
@@ -392,7 +451,7 @@ def install(self, version: str | None = None) -> RuntimeStatus:
                 raise RuntimeChannelError(f"{self._asset} 解出来的不是可执行文件", "E_VERIFY")
             if not self._windows:
                 os.chmod(staged, 0o755)
-            os.replace(staged, self.path)
+            self._replace_payload(staged)
         finally:
             shutil.rmtree(staging, ignore_errors=True)
 
@@ -461,13 +520,29 @@ def forward(self, argv: list[str], binary: str | None = None) -> int:
         """
         target = binary or self.ensure()
         if os.name != "nt":
-            os.execv(target, [target, *argv])  # pragma: no cover - 进程被替换,after 不可达
+            try:
+                os.execv(target, [target, *argv])  # pragma: no cover - 进程被替换,after 不可达
+            except OSError as exc:
+                # 文件本身不可执行 (被截断 / 权限坏了 / 不是本机架构) -> 用 .bak 回滚后重试一次.
+                recovered = self._recover_from_backup(f"执行失败: {exc}")
+                if recovered is None:
+                    raise RuntimeChannelError(
+                        f"无法执行真身 {target}: {exc};可运行 `orzmc runtime install` 重装",
+                        "E_RUNTIME",
+                    ) from exc
+                os.execv(recovered, [recovered, *argv])  # pragma: no cover
         try:
             return subprocess.call([target, *argv])
         except KeyboardInterrupt:  # pragma: no cover - 交互中断
             return 130
         except OSError as exc:
-            raise RuntimeChannelError(f"无法执行 {target}: {exc}", "E_RUNTIME") from exc
+            recovered = self._recover_from_backup(f"执行失败: {exc}")
+            if recovered is None:
+                raise RuntimeChannelError(
+                    f"无法执行真身 {target}: {exc};可运行 `orzmc runtime install` 重装",
+                    "E_RUNTIME",
+                ) from exc
+            return subprocess.call([recovered, *argv])
 
 
 def _stderr_reporter() -> tuple[Reporter, ProgressSink]: