From 949272778c1d978405a4b174268fdc8301249016 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sun, 13 Sep 2026 10:19:52 +0200 Subject: [PATCH] gatelock 0.8.2: wait_for_x_server waits for as long as it takes The 60 s deadline was the 2026-09-13 leetcode-guard miss with a smaller number: a Persistent=true catch-up starts one second into a boot, before X exists, and a bounded wait that expires exits the unit while systemd has already stamped the slot as fired. No X server means the user cannot use the machine either, so an unbounded wait is not a bypass. The wait now re-states itself at WARNING every five minutes so a long wait never reads as a healthy, silent unit. `timeout_s` stays as an opt-in for callers that want a bound. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01BxkZDHDmoVSxUT4w72xSiE --- gatelock/gatelock/_guards.py | 41 ++++++++-- gatelock/gatelock/tests/test_guards.py | 105 ++++++++++++++++++++++--- gatelock/pyproject.toml | 2 +- 3 files changed, 130 insertions(+), 18 deletions(-) diff --git a/gatelock/gatelock/_guards.py b/gatelock/gatelock/_guards.py index 887ad32..3e4cabc 100644 --- a/gatelock/gatelock/_guards.py +++ b/gatelock/gatelock/_guards.py @@ -15,11 +15,21 @@ So the rule now is: **output count never gates arming.** Zero live outputs means lock without showing, not decline to lock. This function may only answer "can we talk to an X server at all", because without one there is no Tk and genuinely nothing to do. It must never ask how many outputs are live. + +The clock does not gate arming either (2026-09-13). The wait used to give up +after 60 s, and a boot-time ``Persistent=true`` catch-up starts one second +after the user manager, long before X exists: on a slow boot the deadline +passed, the unit exited, and nothing locked until the next timer slot -- which +systemd had already marked as fired. No X server means the user cannot use the +machine either, so waiting forever costs nothing and cannot be a bypass. The +wait now polls until the server answers and re-states itself at WARNING every +five minutes, so ``systemctl status`` never shows a healthy, silent unit. """ from __future__ import annotations import logging +import os import sys import time import tkinter as tk @@ -30,8 +40,11 @@ if TYPE_CHECKING: _logger = logging.getLogger(__name__) -_DEFAULT_TIMEOUT_S = 60.0 _DEFAULT_INTERVAL_S = 1.0 +# How often a still-missing X server is re-stated at WARNING. Same reasoning +# as ``_queue.QUEUE_HEARTBEAT_SECONDS``: a long silent wait is +# indistinguishable from a hung unit. +X_WAIT_HEARTBEAT_S = 300.0 def assert_not_under_pytest(what: str) -> None: @@ -66,7 +79,7 @@ def assert_not_under_pytest(what: str) -> None: def wait_for_x_server( *, - timeout_s: float = _DEFAULT_TIMEOUT_S, + timeout_s: float | None = None, interval_s: float = _DEFAULT_INTERVAL_S, sleep: Callable[[float], None] = time.sleep, monotonic: Callable[[], float] = time.monotonic, @@ -80,26 +93,42 @@ def wait_for_x_server( is a reason to lock silently, never a reason to skip locking. Args: - timeout_s: Give up after this long. + timeout_s: Give up after this long. ``None`` (the default) waits for + as long as it takes; see the module docstring for why a bound is + the wrong tool here. interval_s: Seconds between attempts. sleep: Injected for tests. monotonic: Injected for tests. probe: Injected for tests; defaults to opening a throwaway Tk root. Returns: - True if an X server answered, False on timeout. + True once an X server answered, False only if ``timeout_s`` was set + and passed. """ attempt = probe if probe is not None else _probe_x_server - deadline = monotonic() + timeout_s + started = monotonic() + next_heartbeat = X_WAIT_HEARTBEAT_S while True: if attempt(): + elapsed = monotonic() - started + if elapsed >= interval_s: + _logger.warning("X server answered after %.0fs; arming now", elapsed) return True - if monotonic() >= deadline: + elapsed = monotonic() - started + if timeout_s is not None and elapsed >= timeout_s: _logger.error( "no X server answered within %.0fs; cannot build a lock window", timeout_s, ) return False + if elapsed >= next_heartbeat: + _logger.warning( + "no X server on %s after %.0fs -- still waiting, nothing is locked", + os.environ.get("DISPLAY", ""), + elapsed, + ) + while next_heartbeat <= elapsed: + next_heartbeat += X_WAIT_HEARTBEAT_S sleep(interval_s) diff --git a/gatelock/gatelock/tests/test_guards.py b/gatelock/gatelock/tests/test_guards.py index ad5f69a..0c86c77 100644 --- a/gatelock/gatelock/tests/test_guards.py +++ b/gatelock/gatelock/tests/test_guards.py @@ -3,18 +3,46 @@ from __future__ import annotations +import logging import sys import tkinter as tk +from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from gatelock._guards import ( + X_WAIT_HEARTBEAT_S, _probe_x_server, assert_not_under_pytest, wait_for_x_server, ) +if TYPE_CHECKING: + from collections.abc import Iterator + + +class FakeClock: + """A monotonic clock that only moves when something sleeps.""" + + def __init__(self) -> None: + self.value = 0.0 + self.slept: list[float] = [] + + def now(self) -> float: + return self.value + + def sleep(self, seconds: float) -> None: + self.slept.append(seconds) + self.value += seconds + + +def _false_then_true(after: int) -> Iterator[bool]: + for _ in range(after): + yield False + while True: + yield True + class TestAssertNotUnderPytest: """Refuse to build a real grabbing window inside a test run.""" @@ -59,18 +87,73 @@ class TestWaitForXServer: ) assert sleeps == [1.0, 1.0] - def test_times_out(self) -> None: - """A display that never arrives gives up and says so.""" - clock = iter([0.0, 100.0]) - assert ( - wait_for_x_server( - probe=lambda: False, - sleep=lambda _s: None, - monotonic=lambda: next(clock), - timeout_s=1.0, + def test_an_explicit_timeout_gives_up_and_says_so( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Opt-in only: a caller that asks for a bound gets one.""" + clock = FakeClock() + with caplog.at_level(logging.ERROR): + assert ( + wait_for_x_server( + probe=lambda: False, + sleep=clock.sleep, + monotonic=clock.now, + timeout_s=2.5, + ) + is False ) - is False - ) + assert clock.slept == [1.0, 1.0, 1.0] + assert "cannot build a lock window" in caplog.text + + def test_the_default_wait_outlives_the_old_sixty_second_deadline( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The 2026-09-13 rule: time never gates arming. A server that takes + ten minutes to appear is still waited for, and the wait says so.""" + clock = FakeClock() + probe = _false_then_true(after=600) + with caplog.at_level(logging.WARNING): + assert ( + wait_for_x_server( + probe=lambda: next(probe), + sleep=clock.sleep, + monotonic=clock.now, + ) + is True + ) + assert len(clock.slept) == 600 + assert "X server answered after 600s; arming now" in caplog.text + + def test_a_long_wait_heartbeats_at_warning( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """A silent wait looks like a hung unit; it must re-state itself.""" + clock = FakeClock() + probe = _false_then_true(after=int(X_WAIT_HEARTBEAT_S * 2) + 1) + with caplog.at_level(logging.WARNING): + wait_for_x_server( + probe=lambda: next(probe), sleep=clock.sleep, monotonic=clock.now + ) + heartbeats = [r for r in caplog.records if "still waiting" in r.message] + assert len(heartbeats) == 2 + assert all(r.levelno == logging.WARNING for r in heartbeats) + assert "nothing is locked" in heartbeats[0].message + + def test_a_heartbeat_skips_ahead_after_a_long_sleep( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """One coarse sleep past several heartbeat slots logs once, not once + per missed slot, and the next heartbeat lands on a future slot.""" + clock = FakeClock() + probe = _false_then_true(after=2) + with caplog.at_level(logging.WARNING): + wait_for_x_server( + probe=lambda: next(probe), + sleep=clock.sleep, + monotonic=clock.now, + interval_s=X_WAIT_HEARTBEAT_S * 3, + ) + assert sum("still waiting" in r.message for r in caplog.records) == 1 def test_zero_live_outputs_does_not_block_arming(self) -> None: """THE 2b rule: a connected X server with dark monitors still arms. diff --git a/gatelock/pyproject.toml b/gatelock/pyproject.toml index 94071a4..be2be15 100644 --- a/gatelock/pyproject.toml +++ b/gatelock/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "gatelock" -version = "0.8.1" +version = "0.8.2" description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps" requires-python = ">=3.14" dependencies = [] # pure stdlib — tkinter is bundled with Python