Coverage for src/lilbee/parent_monitor.py: 100%

1"""Watch ``LILBEE_PARENT_PID`` and shut down when the parent process exits.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import logging 

7import os 

8import threading 

9import time 

10from collections.abc import Callable 

11 

12import psutil 

13 

14log = logging.getLogger(__name__) 

15 

16POLL_INTERVAL_SECS = 2.0 

17PARENT_PID_ENV = "LILBEE_PARENT_PID" 

18 

19 

20def parse_parent_pid(env: dict[str, str] | None = None) -> int | None: 

21 """Return a valid parent PID from the env, or None if unset/garbage.""" 

22 src = env if env is not None else os.environ 

23 raw = src.get(PARENT_PID_ENV) 

24 if not raw: 

25 return None 

26 try: 

27 pid = int(raw) 

28 except ValueError: 

29 log.warning("%s=%r is not an integer; skipping parent-death monitor", PARENT_PID_ENV, raw) 

30 return None 

31 if pid <= 0: 

32 log.warning("%s=%d is non-positive; skipping parent-death monitor", PARENT_PID_ENV, pid) 

33 return None 

34 return pid 

35 

36 

37def _parent_start_time(pid: int) -> float | None: 

38 """Process create-time for *pid*, or None if it is gone or unreadable.""" 

39 try: 

40 return float(psutil.Process(pid).create_time()) 

41 except (psutil.NoSuchProcess, psutil.AccessDenied): 

42 return None 

43 

44 

45def _same_process(pid: int, start_time: float | None) -> bool: 

46 """False when *pid* now belongs to a different process than *start_time*.""" 

47 if start_time is None: 

48 return True 

49 try: 

50 return bool(psutil.Process(pid).create_time() == start_time) 

51 except (psutil.NoSuchProcess, psutil.AccessDenied): 

52 return False 

53 

54 

55def _parent_alive(pid: int, start_time: float | None) -> bool: 

56 """True while *pid* still refers to the original parent process.""" 

57 return psutil.pid_exists(pid) and _same_process(pid, start_time) 

58 

59 

60async def watch_parent_async( 

61 parent_pid: int, 

62 on_death: Callable[[], None], 

63 *, 

64 poll_interval_secs: float = POLL_INTERVAL_SECS, 

65) -> None: 

66 """Poll *parent_pid* until it exits or its PID is recycled, then call *on_death* once.""" 

67 start_time = _parent_start_time(parent_pid) 

68 while _parent_alive(parent_pid, start_time): 

69 await asyncio.sleep(poll_interval_secs) 

70 log.info("%s=%d is no longer alive; triggering shutdown", PARENT_PID_ENV, parent_pid) 

71 on_death() 

72 

73 

74def watch_parent_thread( 

75 parent_pid: int, 

76 on_death: Callable[[], None], 

77 *, 

78 poll_interval_secs: float = POLL_INTERVAL_SECS, 

79) -> threading.Thread: 

80 """Daemon thread that fires *on_death* once *parent_pid* exits or its PID is recycled.""" 

81 

82 def _loop() -> None: 

83 start_time = _parent_start_time(parent_pid) 

84 while _parent_alive(parent_pid, start_time): 

85 time.sleep(poll_interval_secs) 

86 log.info("%s=%d is no longer alive; triggering shutdown", PARENT_PID_ENV, parent_pid) 

87 on_death() 

88 

89 thread = threading.Thread(target=_loop, daemon=True, name="lilbee-parent-monitor") 

90 thread.start() 

91 return thread