from __future__ import annotations MAX_SEQUENCE = 0xFFFFFFFF _HALF_RANGE = 0x80000000 def next_sequence(current: int) -> int: if not isinstance(current, int) or isinstance(current, bool) or not 0 <= current <= MAX_SEQUENCE: raise ValueError("snapshot sequence must be an unsigned 32-bit integer") value = (current + 1) & MAX_SEQUENCE return 1 if value == 0 else value def is_newer(candidate: int, baseline: int) -> bool: if not isinstance(candidate, int) or isinstance(candidate, bool): return False if not isinstance(baseline, int) or isinstance(baseline, bool): return False if not 0 <= candidate <= MAX_SEQUENCE or not 0 <= baseline <= MAX_SEQUENCE: return False if candidate == 0: return False if baseline == 0: return True delta = (candidate - baseline) & MAX_SEQUENCE return 0 < delta < _HALF_RANGE class SequenceCounter: def __init__(self) -> None: self._value = 0 @property def current(self) -> int: return self._value def advance(self) -> int: self._value = next_sequence(self._value) return self._value def reset(self) -> None: self._value = 0 class SequenceWindow: def __init__(self) -> None: self._last_accepted = 0 @property def last_accepted(self) -> int: return self._last_accepted def accept(self, sequence: int) -> bool: if not is_newer(sequence, self._last_accepted): return False self._last_accepted = sequence return True def reset(self) -> None: self._last_accepted = 0