(name: str, lock_path: str)
| 111 | |
| 112 | |
| 113 | def _hammer(name: str, lock_path: str) -> tuple[list[tuple[float, float]], str | None]: |
| 114 | lock = _build(name, lock_path) |
| 115 | intervals: list[tuple[float, float]] = [] |
| 116 | for _ in range(_HOLDS): |
| 117 | if (reason := _resiliently(lambda: lock.acquire(timeout=_ACQUIRE_TIMEOUT))) is not None: |
| 118 | return intervals, reason |
| 119 | enter = time.monotonic() |
| 120 | # Hold briefly so a broken lock lets a second holder in during an observable window; a correct lock serializes |
| 121 | # the holds regardless. enter is stamped after acquire and leave before release, so a correct hand-off can never |
| 122 | # look like an overlap even though release and the next acquire race. |
| 123 | time.sleep(_HOLD_SECONDS) |
| 124 | leave = time.monotonic() |
| 125 | if (reason := _resiliently(lock.release)) is not None: |
| 126 | intervals.append((enter, leave)) |
| 127 | return intervals, reason |
| 128 | intervals.append((enter, leave)) |
| 129 | # A randomized gap outside the lock breaks the lock-step herd, so a poll-based lock hands off fairly and no |
| 130 | # contender is starved out of finishing its holds. |
| 131 | time.sleep(random.uniform(0, _GAP_MAX_SECONDS)) # ruff:ignore[suspicious-non-cryptographic-random-usage] - test dispersion, not cryptographic |
| 132 | return intervals, None |
| 133 | |
| 134 | |
| 135 | def _resiliently(action: Callable[[], object]) -> str | None: |
nothing calls this directly
no test coverage detected