Define wrap-safe snapshot sequencing

server/snapshot_sequence.py

@@ -0,0 +1,60 @@
+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