Ordered integer stream

Items carry sequence numbers 1, 2, 3, … but arrive out of order, as with chunks of a streamed model reply sent over several connections, or messages from a queue with retries. The consumer must see them strictly in order. Operation:

• insert(seq, value) -> list: store the item and return every value that is now ready: the longest run of consecutive sequence numbers starting at the next one the consumer has not seen. Usually empty; sometimes many at once when a gap fills.

Ask:

• Is the total count known? In the classic version, n is known and ids are 1..n. In the streaming version it is unbounded. Support unbounded; it covers both.
• Duplicates? A retried item may arrive twice. Ignore one that is already buffered or already emitted.
• Gaps that never fill. A lost item would block the stream forever. Part 2 adds a bounded buffer and a timeout that skips the gap.

Keep next_seq (the first sequence number not yet emitted) and a dict pending from sequence number to value for items that arrived early.

• insert(seq, value): if seq < next_seq or seq in pending, it is a duplicate: return []. Otherwise store it. Then, while next_seq in pending, pop it into the output and increment next_seq.

Each item is stored once and popped once, so the total work over the whole stream is O(n), and each call costs O(1 + number of values it returns). Memory is the number of items waiting behind the gap.

The classic fixed-size version often uses an array of size n and a pointer, which is the same idea with a list instead of a dict.

Part 2: real networks lose things. Two protections:

• Bounded window. Refuse (or tell the sender to slow down) items more than window ahead of next_seq. This caps memory at window items, like a TCP receive window.
• Gap timeout. Record when the oldest gap started. If it is still open after timeout, give up on the missing sequence numbers: report them as lost and jump next_seq to the smallest buffered number. Use an injected clock so this is testable.

from typing import Any, Callable, Dict, List, Optional, Tuple


class OrderedStream:
    """Releases values in sequence order as out-of-order inserts arrive."""

    def __init__(self, start: int = 1) -> None:
        self.next_seq = start
        self.pending: Dict[int, Any] = {}

    def insert(self, seq: int, value: Any) -> List[Any]:
        if seq < self.next_seq or seq in self.pending:
            return []                      # duplicate or already delivered
        self.pending[seq] = value
        return self._drain()

    def _drain(self) -> List[Any]:
        out = []
        while self.next_seq in self.pending:
            out.append(self.pending.pop(self.next_seq))
            self.next_seq += 1
        return out


class TooFarAhead(Exception):
    pass


class ReorderBuffer(OrderedStream):
    """Part 2: bounded memory, and gaps are skipped after a timeout."""

    def __init__(self, window: int, timeout: float, clock: Callable[[], float], start: int = 1) -> None:
        super().__init__(start)
        self.window = window
        self.timeout = timeout
        self.clock = clock
        self.gap_since: Optional[float] = None
        self.lost: List[int] = []

    def insert(self, seq: int, value: Any) -> List[Any]:
        if seq >= self.next_seq + self.window:
            raise TooFarAhead(seq)
        out = super().insert(seq, value)
        if out:
            self.gap_since = None          # the old gap closed; time the next one afresh
        self._track_gap()
        return out

    def poll(self) -> Tuple[List[Any], List[int]]:
        """Call periodically: skip a gap that has been open too long."""
        if self.gap_since is None or self.clock() - self.gap_since < self.timeout:
            return [], []
        first = min(self.pending)
        skipped = list(range(self.next_seq, first))
        self.lost += skipped
        self.next_seq = first
        out = self._drain()
        self.gap_since = None
        self._track_gap()
        return out, skipped

    def _track_gap(self) -> None:
        if not self.pending:
            self.gap_since = None
        elif self.gap_since is None:
            self.gap_since = self.clock()

When a gap closes and another one is behind it, the timer restarts at that moment, so each gap gets its own full timeout.

• insert: O(1) amortized plus O(r) for the r values it releases. Over the stream, O(n) total.
• poll: O(p) for p pending items (the min); keep a heap of pending sequence numbers if p is large and polls are frequent.
• Memory: O(items waiting), at most window in part 2.

1. Streaming model output. When a reply is generated in parallel chunks (or relayed through several servers), each chunk carries a sequence number and the client renders through a buffer like this. A lost chunk is re-requested by number instead of being skipped, so the window doubles as the list of chunks to retry.
2. Acknowledgements. The sender needs to know what can be freed. Return next_seq − 1 as a cumulative acknowledgement, plus the buffered numbers above it (selective acknowledgements) so only the gaps are re-sent.
3. Per-key ordering. Messages only need to be ordered per conversation, not globally. Keep one buffer per key, so one stalled conversation does not block the rest, and expire idle buffers.
4. Persisted progress. If the consumer restarts, it must resume at next_seq. Store it with the consumer's output in the same transaction, so a crash cannot deliver an item twice or skip one.
5. Sequence wraparound. With 32-bit sequence numbers on a long-lived stream, compare with serial-number arithmetic (difference modulo 2^32 interpreted as signed) instead of <.

Mid-level (L4)

• A pointer plus a buffer that releases the longest consecutive run on each insert.
• Correct on the classic example and when a gap fills and releases many values at once.
• States the O(n) total cost.

Senior (L5)

• Handles duplicates and late arrivals, and supports an unbounded stream with a dict.
• Bounds memory with a window and skips stuck gaps with a timeout driven by an injected clock.
• Tests every permutation of a small input, plus the timeout edge cases.

Staff (L6+)

• Relates it to transport design: acknowledgements, selective re-sends, receive windows.
• Orders per key so one slow stream does not block others, and persists progress for exactly-once delivery.
• Raises wraparound and when to retry a gap versus skip it.