Binary Search Off-by-One

Problem

A log-tailing job keeps a sorted list of event timestamps. On every tick it must find the index of the first event at or after a watermark t, so that it only processes events it has not seen. This is the helper in production:

def first_at_or_after(timestamps: list[int], t: int) -> int:
    """Index of the first element >= t. If every element is < t,
    return len(timestamps) so the caller processes nothing."""
    lo, hi = 0, len(timestamps) - 1
    while lo < hi:
        mid = (lo + hi) // 2
        if timestamps[mid] < t:
            lo = mid + 1
        else:
            hi = mid
    return lo

Symptom: during quiet minutes, when no new event has arrived since the last tick, the job re-processes the most recent event once per tick and produces duplicate downstream writes. The unit tests all pass. Find the bug, explain why the tests did not catch it, and fix it.

Examples

• timestamps = [10, 20, 30], t = 25 → 2. Correct: 30 is the first value at or after 25.
• timestamps = [10, 20, 30], t = 31 → returns 2, expected 3. Nothing is ≥ 31, so the answer must be len(timestamps); the function points at 30 instead, which is exactly the "re-process the last event" symptom.
• timestamps = [], t = 5 → 0. Correct, but only by accident: the loop never runs and 0 == len([]).

Hints

1. Write one sentence each for what is true of every index below lo and every index at or above hi at the top of each iteration.
2. The function is supposed to be able to return len(timestamps). Trace which values lo can ever take. Can it?
3. A lower-bound search has n + 1 possible answers (insert before element 0 ... insert after element n-1), so the search space must have n + 1 candidates.