diff --git a/crdt-sync/requirements.txt b/crdt-sync/requirements.txt index cdad590..90c8c69 100644 --- a/crdt-sync/requirements.txt +++ b/crdt-sync/requirements.txt @@ -4,7 +4,7 @@ bandit[toml]==1.9.4 mypy==2.3.1 pre-commit==4.6.2 -pylint==4.0.7 +pylint==4.0.8 pytest==9.1.1 pytest-cov==7.1.0 requests==2.34.2 diff --git a/gatelock/gatelock/__init__.py b/gatelock/gatelock/__init__.py index 299970f..1118765 100644 --- a/gatelock/gatelock/__init__.py +++ b/gatelock/gatelock/__init__.py @@ -39,6 +39,7 @@ from gatelock._queue import ( QUEUE_HEARTBEAT_SECONDS, QUEUE_POLL_SECONDS, QueueResult, + Timebase, stronger_claims, wait_for_turn, ) @@ -112,6 +113,7 @@ __all__ = [ "SurfaceDelta", "SurfaceInfo", "SurfaceSet", + "Timebase", "TypeRole", "WidgetGroup", "assert_not_under_pytest", diff --git a/gatelock/gatelock/_queue.py b/gatelock/gatelock/_queue.py index 1c4bf6c..19041d6 100644 --- a/gatelock/gatelock/_queue.py +++ b/gatelock/gatelock/_queue.py @@ -75,14 +75,111 @@ def stronger_claims(arbiter: Arbiter) -> tuple[Claim, ...]: ) +@dataclass(frozen=True) +class Timebase: + """Injected clock and sleep, so a wait can be tested without waiting. + + Bundled rather than passed as two separate arguments: ``wait_for_turn`` + is a public entry point and its knobs are already at the limit ruff + allows, and these two only ever vary together. + """ + + sleep: Callable[[float], None] = time.sleep + now: Callable[[], float] = time.monotonic + + +class _Wait: + """One in-progress wait: who is ahead of us, and for how long. + + A class rather than closures inside ``wait_for_turn`` because the + heartbeat needs state that outlives a single loop iteration, and threading + that through as locals pushed the function past ruff's complexity limit. + """ + + def __init__( + self, + arbiter: Arbiter, + on_state: Callable[[tuple[str, ...], float], None] | None, + started: float, + ) -> None: + """Start tracking a wait. + + Args: + arbiter: Our own published arbiter. + on_state: Optional observer, see :func:`wait_for_turn`. + started: The monotonic timestamp the wait began at. + """ + self._arbiter = arbiter + self._on_state = on_state + self._started = started + self._next_heartbeat = QUEUE_HEARTBEAT_SECONDS + self.seen: list[str] = [] + + def publish(self, blocked: tuple[str, ...], elapsed: float) -> None: + """Tell the observer what this wait looks like right now. + + Args: + blocked: Apps ahead of us; empty means the wait has ended. + elapsed: Seconds waited so far. + """ + if self._on_state is not None: + self._on_state(blocked, elapsed) + + def note(self, blockers: tuple[Claim, ...], elapsed: float) -> None: + """Record any blocker not seen before, and publish if that changed. + + Args: + blockers: The live claims currently ahead of us. + elapsed: Seconds waited so far. + """ + newly_seen = False + for claim in blockers: + if claim.app in self.seen: + continue + self.seen.append(claim.app) + newly_seen = True + _logger.info( + "%s is queued behind %s (rank %d) -- waiting with no window", + self._arbiter.claim.app, + claim.app, + claim.rank, + ) + if newly_seen: + self.publish(tuple(self.seen), elapsed) + + def heartbeat(self, elapsed: float, deadline: float) -> None: + """Re-state a still-blocked wait, periodically and at WARNING. + + The wait used to announce itself once, at INFO, and then go silent: + a 2h58m block on 2026-08-30 was indistinguishable from a healthy idle + unit in ``systemctl status``. + + Args: + elapsed: Seconds waited so far. + deadline: The runaway backstop, for context in the message. + """ + if elapsed < self._next_heartbeat: + return + _logger.warning( + "%s has been queued behind %s for %.0fs with no window -- " + "still waiting (deadline %.0fs)", + self._arbiter.claim.app, + ", ".join(self.seen), + elapsed, + deadline, + ) + self.publish(tuple(self.seen), elapsed) + while self._next_heartbeat <= elapsed: + self._next_heartbeat += QUEUE_HEARTBEAT_SECONDS + + def wait_for_turn( arbiter: Arbiter, *, poll: float = QUEUE_POLL_SECONDS, deadline: float = QUEUE_DEADLINE_SECONDS, on_state: Callable[[tuple[str, ...], float], None] | None = None, - sleep: Callable[[float], None] | None = None, - now: Callable[[], float] | None = None, + timebase: Timebase | None = None, ) -> QueueResult: """Block, showing nothing, until no stronger locker holds the screen. @@ -92,73 +189,40 @@ def wait_for_turn( poll: Seconds between checks. Liveness comes from ``flock``, so a SIGKILLed incumbent is noticed on the very next tick. deadline: Runaway backstop in seconds. - on_state: Called with ``(blocked_by, elapsed)`` each time the - blocked set is first seen or changes, and once with an empty - tuple when the wait ends. Lets a caller publish the wait + on_state: Called with ``(blocked_by, elapsed)`` when the blocked set + first appears or changes, on every heartbeat, and once with an + empty tuple when the wait ends. Lets a caller publish the wait somewhere observable -- from outside, a blocked process is indistinguishable from a hung one. - sleep: Injected for tests. - now: Injected monotonic clock, for tests. + timebase: Injected clock and sleep, for tests. Returns: A record of the wait. """ - rest = sleep if sleep is not None else time.sleep - clock = now if now is not None else time.monotonic + time_source = timebase if timebase is not None else Timebase() + clock = time_source.now started = clock() - seen: list[str] = [] - next_heartbeat = QUEUE_HEARTBEAT_SECONDS - - def _publish(blocked: tuple[str, ...], elapsed: float) -> None: - if on_state is not None: - on_state(blocked, elapsed) + wait = _Wait(arbiter, on_state, started) while True: blockers = stronger_claims(arbiter) + elapsed = clock() - started if not blockers: - waited = clock() - started - if seen: + if wait.seen: _logger.warning( "%s waited %.0fs behind %s; arming now", arbiter.claim.app, - waited, - ", ".join(seen), + elapsed, + ", ".join(wait.seen), ) - _publish((), waited) + wait.publish((), elapsed) return QueueResult( - waited_seconds=waited, blocked_by=tuple(seen), timed_out=False + waited_seconds=elapsed, blocked_by=tuple(wait.seen), timed_out=False ) - newly_seen = False - for claim in blockers: - if claim.app not in seen: - seen.append(claim.app) - newly_seen = True - _logger.info( - "%s is queued behind %s (rank %d) -- waiting with no window", - arbiter.claim.app, - claim.app, - claim.rank, - ) - - elapsed = clock() - started - if newly_seen: - _publish(tuple(seen), elapsed) - # Re-state a long wait periodically. Silence here is what made a - # 2h58m block look identical to a healthy idle unit. - if elapsed >= next_heartbeat: - _logger.warning( - "%s has been queued behind %s for %.0fs with no window -- " - "still waiting (deadline %.0fs)", - arbiter.claim.app, - ", ".join(seen), - elapsed, - deadline, - ) - _publish(tuple(seen), elapsed) - while next_heartbeat <= elapsed: - next_heartbeat += QUEUE_HEARTBEAT_SECONDS + wait.note(blockers, elapsed) + wait.heartbeat(elapsed, deadline) if elapsed >= deadline: _logger.error( @@ -166,11 +230,11 @@ def wait_for_turn( "anyway rather than leaving the machine unlocked", arbiter.claim.app, elapsed, - ", ".join(seen), + ", ".join(wait.seen), ) - _publish((), elapsed) + wait.publish((), elapsed) return QueueResult( - waited_seconds=elapsed, blocked_by=tuple(seen), timed_out=True + waited_seconds=elapsed, blocked_by=tuple(wait.seen), timed_out=True ) - rest(poll) + time_source.sleep(poll) diff --git a/gatelock/gatelock/tests/test_queue.py b/gatelock/gatelock/tests/test_queue.py index f818968..018c1c1 100644 --- a/gatelock/gatelock/tests/test_queue.py +++ b/gatelock/gatelock/tests/test_queue.py @@ -6,7 +6,12 @@ import logging from typing import TYPE_CHECKING from gatelock import RANK_DIET_GUARD, RANK_SCREEN_LOCKER, RANK_WAKE_ALARM, Arbiter -from gatelock._queue import stronger_claims, wait_for_turn +from gatelock._queue import ( + QUEUE_HEARTBEAT_SECONDS, + Timebase, + stronger_claims, + wait_for_turn, +) if TYPE_CHECKING: from pathlib import Path @@ -39,7 +44,7 @@ def test_an_empty_ladder_arms_immediately(tmp_path: Path) -> None: arbiter = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path) clock = FakeClock() - result = wait_for_turn(arbiter, sleep=clock.sleep, now=clock.now) + result = wait_for_turn(arbiter, timebase=Timebase(sleep=clock.sleep, now=clock.now)) assert not result.queued assert not result.timed_out @@ -65,7 +70,7 @@ def test_we_queue_behind_the_workout_lock_and_arm_when_it_finishes( released["done"] = True with caplog.at_level(logging.INFO): - result = wait_for_turn(ours, sleep=sleep, now=clock.now) + result = wait_for_turn(ours, timebase=Timebase(sleep=sleep, now=clock.now)) assert result.queued assert result.blocked_by == ("screen_locker",) @@ -80,7 +85,7 @@ def test_a_lower_ranked_locker_is_not_waited_for(tmp_path: Path) -> None: ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path) clock = FakeClock() - result = wait_for_turn(ours, sleep=clock.sleep, now=clock.now) + result = wait_for_turn(ours, timebase=Timebase(sleep=clock.sleep, now=clock.now)) assert not result.queued diet.release() @@ -98,7 +103,7 @@ def test_every_higher_ranked_app_is_named_once(tmp_path: Path) -> None: alarm.release() workout.release() - result = wait_for_turn(ours, sleep=sleep, now=clock.now) + result = wait_for_turn(ours, timebase=Timebase(sleep=sleep, now=clock.now)) assert set(result.blocked_by) == {"wake_alarm", "screen_locker"} ours.release() @@ -114,7 +119,10 @@ def test_the_deadline_arms_anyway_rather_than_leaving_the_pc_unlocked( with caplog.at_level(logging.ERROR): result = wait_for_turn( - ours, poll=10.0, deadline=30.0, sleep=clock.sleep, now=clock.now + ours, + poll=10.0, + deadline=30.0, + timebase=Timebase(sleep=clock.sleep, now=clock.now), ) assert result.timed_out @@ -141,3 +149,93 @@ def test_a_dead_holder_stops_blocking_us(tmp_path: Path) -> None: workout.release() assert stronger_claims(ours) == () ours.release() + + +def test_a_long_wait_is_restated_periodically( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A block used to announce itself once, at INFO, then go silent. + + On 2026-08-30 screen_locker sat behind wake_alarm for 10716s and + ``systemctl status`` showed an active, quiet, apparently healthy unit. + """ + theirs = make_arbiter("wake_alarm", RANK_WAKE_ALARM, tmp_path) + ours = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path) + clock = FakeClock() + released = 3 * QUEUE_HEARTBEAT_SECONDS + 10.0 + + def sleep(seconds: float) -> None: + clock.sleep(seconds) + if clock.value >= released: + theirs.release() + + with caplog.at_level(logging.WARNING): + result = wait_for_turn(ours, timebase=Timebase(sleep=sleep, now=clock.now)) + + heartbeats = [r for r in caplog.records if "still waiting" in r.message] + assert len(heartbeats) == 3 + assert all(r.levelname == "WARNING" for r in heartbeats) + assert result.blocked_by == ("wake_alarm",) + ours.release() + + +def test_the_wait_is_published_while_it_happens(tmp_path: Path) -> None: + """A blocked process cannot answer for itself, so it must be observable.""" + theirs = make_arbiter("wake_alarm", RANK_WAKE_ALARM, tmp_path) + ours = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path) + clock = FakeClock() + seen: list[tuple[tuple[str, ...], float]] = [] + + def sleep(seconds: float) -> None: + clock.sleep(seconds) + if clock.value >= QUEUE_HEARTBEAT_SECONDS + 10.0: + theirs.release() + + wait_for_turn( + ours, + on_state=lambda blocked, elapsed: seen.append((blocked, elapsed)), + timebase=Timebase(sleep=sleep, now=clock.now), + ) + + # First sighting, then at least one heartbeat, then the clear. + assert seen[0][0] == ("wake_alarm",) + assert any(blocked == ("wake_alarm",) for blocked, _ in seen[1:-1]) + assert seen[-1][0] == () + ours.release() + + +def test_an_unblocked_wait_publishes_only_the_clear(tmp_path: Path) -> None: + """The common case must not spam an observer that has nothing to show.""" + ours = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path) + clock = FakeClock() + seen: list[tuple[str, ...]] = [] + + wait_for_turn( + ours, + on_state=lambda blocked, _elapsed: seen.append(blocked), + timebase=Timebase(sleep=clock.sleep, now=clock.now), + ) + + assert seen == [()] + ours.release() + + +def test_hitting_the_deadline_publishes_the_clear_too(tmp_path: Path) -> None: + """Arming anyway must not leave an observer thinking we are still queued.""" + theirs = make_arbiter("wake_alarm", RANK_WAKE_ALARM, tmp_path) + ours = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path) + clock = FakeClock() + seen: list[tuple[str, ...]] = [] + + result = wait_for_turn( + ours, + poll=10.0, + deadline=30.0, + on_state=lambda blocked, _elapsed: seen.append(blocked), + timebase=Timebase(sleep=clock.sleep, now=clock.now), + ) + + assert result.timed_out + assert seen[-1] == () + theirs.release() + ours.release() diff --git a/gatelock/requirements.txt b/gatelock/requirements.txt index 2222fa1..973acd0 100644 --- a/gatelock/requirements.txt +++ b/gatelock/requirements.txt @@ -4,7 +4,7 @@ bandit[toml]==1.9.4 mypy==2.3.1 pre-commit==4.6.2 -pylint==4.0.7 +pylint==4.0.8 pytest==9.1.1 pytest-cov==7.1.0 ruff==0.16.5