| """Per-slot window bookkeeping of CONTRACT.md 7.2 (v1.3), the worker's copy. |
| |
| The rule (same as lane O3's remote_dspark/state.py WindowRange, which the Spark proxy uses for its rank-0 ring): a |
| request's rows form one contiguous position range [lo, hi], at most W long, stored at pos mod W. Applying rows |
| [p0, p0+n): empty or gap (p0 > hi+1) -> lo = p0; else roll back rows >= p0 and lo = min(lo, p0); then hi = p0+n-1, |
| lo = max(lo, hi-W+1). n == 0 is a pure rollback to anchor_pos-1 (requires hi+1 >= anchor_pos). |
| Because [lo, hi] never exceeds W consecutive positions, its pos-mod-W slots are distinct, and every position in it was |
| written by the current generation (lane O3's argument, CONTRACT 7.2); stale ring entries lie outside [lo, hi]. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
|
|
|
|
| @dataclass |
| class WindowRange: |
| window: int |
| lo: int | None = None |
| hi: int | None = None |
|
|
| @property |
| def empty(self) -> bool: |
| return self.lo is None |
|
|
| def clear(self) -> None: |
| self.lo = self.hi = None |
|
|
| def apply(self, first_pos: int, n_rows: int, anchor_pos: int) -> None: |
| """Raises ValueError (-> BAD_ENTRY) when the entry is inconsistent (CONTRACT 7.2 steps 2-4).""" |
| if not 0 <= n_rows <= self.window: |
| raise ValueError(f"n_rows {n_rows} outside 0..{self.window}") |
| if anchor_pos < 0: |
| raise ValueError(f"anchor_pos {anchor_pos} < 0") |
| if n_rows > 0: |
| if first_pos < 0: |
| raise ValueError(f"first_pos {first_pos} < 0") |
| if anchor_pos != first_pos + n_rows: |
| raise ValueError(f"anchor_pos {anchor_pos} != first_pos {first_pos} + n_rows {n_rows}") |
| p0 = first_pos |
| if self.empty or p0 > self.hi + 1: |
| self.lo = p0 |
| else: |
| self.lo = min(self.lo, p0) |
| self.hi = p0 + n_rows - 1 |
| else: |
| if self.empty or self.hi + 1 < anchor_pos: |
| raise ValueError(f"no rows sent and the window ends at {self.hi}, anchor_pos {anchor_pos}") |
| self.hi = anchor_pos - 1 |
| if self.hi < self.lo: |
| self.clear() |
| return |
| self.lo = max(self.lo, self.hi - self.window + 1) |
| if self.hi != anchor_pos - 1: |
| raise ValueError(f"window ends at {self.hi}, anchor_pos {anchor_pos}") |
|
|
| def read_lo(self, anchor_pos: int) -> int: |
| """Lowest position an anchor at anchor_pos reads: max(lo, anchor_pos - W, 0).""" |
| return max(self.lo if self.lo is not None else anchor_pos, anchor_pos - self.window, 0) |
|
|
| def rows_for_anchor(self, anchor_pos: int) -> int: |
| if self.empty or self.hi != anchor_pos - 1: |
| return 0 |
| return max(0, self.hi - self.read_lo(anchor_pos) + 1) |
|
|
| def expected_rows(self, anchor_pos: int) -> int: |
| return min(self.window, max(anchor_pos, 0)) |
|
|
| def positions(self, anchor_pos: int | None = None) -> range: |
| if self.empty: |
| return range(0) |
| start = self.lo if anchor_pos is None else self.read_lo(anchor_pos) |
| return range(start, self.hi + 1) |
|
|
