From 9f8f9988b718a61a1c408107f36669c993873a3a Mon Sep 17 00:00:00 2001
From: liuhao1024 <sunsky.lau@gmail.com>
Date: Tue, 22 Sep 2026 16:02:44 +0800
Subject: [PATCH] fix(gateway): skip systemd timing check for non-loaded units

systemctl show exits 0 and reports compiled-in defaults
(TimeoutStopUSec=1min 30s) for a unit the manager does not own, so a
reachable user manager shadowed the real system-level unit with a false
90s reading and triggered a stale-unit warning. Gate each manager
iteration on LoadState=loaded so the manager that actually owns the unit
is the one consulted, and add mocked two-manager regression tests.

Fixes #36755
---
 gateway/shutdown_forensics.py            | 14 ++++++-
 tests/gateway/test_shutdown_forensics.py | 51 ++++++++++++++++++++++++
 2 files changed, 64 insertions(+), 1 deletion(-)

diff --git a/gateway/shutdown_forensics.py b/gateway/shutdown_forensics.py
index c7d337d47f23..ebbd01e32b27 100644
--- a/gateway/shutdown_forensics.py
+++ b/gateway/shutdown_forensics.py
@@ -230,11 +230,23 @@ def _systemd_timeout_stop_us(unit_name: str) -> Optional[int]:
     for flag in (["--user"], []):
         try:
             result = subprocess.run(
-                ["systemctl", *flag, "show", unit_name, "--property=TimeoutStopUSec"],
+                ["systemctl", *flag, "show", unit_name,
+                 "--property=LoadState", "--property=TimeoutStopUSec"],
                 capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2.0,
             )
         except (subprocess.TimeoutExpired, OSError):
             continue
+        # Skip managers that don't actually own the unit: ``systemctl show`` exits 0
+        # and reports compiled-in defaults (``TimeoutStopUSec=1min 30s``) for a
+        # nonexistent unit, so a reachable user manager would otherwise shadow the
+        # real system-level unit with a false 90s reading (issue #36755).
+        load_state: Optional[str] = None
+        for line in result.stdout.splitlines() if result.returncode == 0 else ():
+            if line.startswith("LoadState="):
+                load_state = line.split("=", 1)[1].strip()
+                break
+        if load_state and load_state != "loaded":
+            continue
         # Output: "TimeoutStopUSec=1min 30s" or "TimeoutStopUSec=90000000"
         for line in result.stdout.splitlines() if result.returncode == 0 else ():
             if line.startswith("TimeoutStopUSec="):
diff --git a/tests/gateway/test_shutdown_forensics.py b/tests/gateway/test_shutdown_forensics.py
index 8bd69e6380e3..5012659da34d 100644
--- a/tests/gateway/test_shutdown_forensics.py
+++ b/tests/gateway/test_shutdown_forensics.py
@@ -217,3 +217,54 @@ def test_returns_none_when_unit_undeterminable(self, monkeypatch):
         # for whatever unit pytest IS in.  Both are valid; we just ensure
         # the function doesn't raise.
         assert result is None or isinstance(result, dict)
+
+
+# ---------------------------------------------------------------------------
+# _systemd_timeout_stop_us
+# ---------------------------------------------------------------------------
+
+class TestSystemdTimeoutStopUs:
+    """Issue #36755: a reachable user manager answers ``systemctl show`` with rc=0 and
+    compiled-in defaults (``TimeoutStopUSec=1min 30s``) for a unit it does not own, so a
+    system-level deployment would read a false 90s and warn about a stale unit. The
+    LoadState gate must skip the non-loaded manager and read the one that owns the unit."""
+
+    @staticmethod
+    def _show(stdout_by_flag):
+        def fake_run(cmd, **kwargs):
+            stdout = stdout_by_flag["user" if "--user" in cmd else "system"]
+            return subprocess.CompletedProcess(cmd, 0, stdout, "")
+        return fake_run
+
+    def test_not_found_user_unit_falls_through_to_loaded_system_unit(self, monkeypatch):
+        # The pre-gate bug: this first answer (90s default) would be returned as real.
+        monkeypatch.setattr(sf.subprocess, "run", self._show({
+            "user": "LoadState=not-found\nTimeoutStopUSec=1min 30s\n",
+            "system": "LoadState=loaded\nTimeoutStopUSec=4min\n",
+        }))
+        assert sf._systemd_timeout_stop_us("hermes-gateway.service") == 240 * 1_000_000
+
+    def test_loaded_user_unit_wins_without_consulting_system_manager(self, monkeypatch):
+        calls = []
+
+        def fake_run(cmd, **kwargs):
+            calls.append(cmd)
+            return subprocess.CompletedProcess(
+                cmd, 0, "LoadState=loaded\nTimeoutStopUSec=2min\n", "")
+
+        monkeypatch.setattr(sf.subprocess, "run", fake_run)
+        assert sf._systemd_timeout_stop_us("hermes-gateway.service") == 120 * 1_000_000
+        assert len(calls) == 1 and "--user" in calls[0]
+
+    def test_no_loaded_manager_returns_none(self, monkeypatch):
+        monkeypatch.setattr(sf.subprocess, "run", self._show({
+            "user": "LoadState=not-found\nTimeoutStopUSec=1min 30s\n",
+            "system": "LoadState=not-found\nTimeoutStopUSec=1min 30s\n",
+        }))
+        assert sf._systemd_timeout_stop_us("hermes-gateway.service") is None
+
+    def test_missing_load_state_line_keeps_legacy_behaviour(self, monkeypatch):
+        # Old systemctl output without LoadState (or other managers) still parses.
+        monkeypatch.setattr(sf.subprocess, "run", self._show({
+            "user": "TimeoutStopUSec=3min\n", "system": ""}))
+        assert sf._systemd_timeout_stop_us("hermes-gateway.service") == 180 * 1_000_000