21. WebSocket Connection Slot Assignment

Problem

Cartesia's realtime API streams over a WebSocket where a single connection can host multiple concurrent contexts, but a service still runs a bounded pool of underlying connection slots for cost/capacity reasons.

Given a batch of sessions, each with a known start and end time (sorted by start time), assign each one the lowest-numbered free slot, reusing a slot as soon as its previous session has ended. Report which slot number each session gets.

Source: src/21_websocket_slot_assignment.py

def assign_connection_slots(sessions: list[tuple[float, float, str]]) -> dict[str, int]: ...

>>> assign_connection_slots([(0, 5, "s1"), (1, 3, "s2"), (4, 10, "s3")])
{'s1': 0, 's2': 1, 's3': 1}
>>> assign_connection_slots([(0, 1, "a"), (0, 1, "b")])
{'a': 0, 'b': 1}

Step-by-Step Approach

1. Since sessions arrive pre-sorted by start time, process them in a single forward sweep — no need to sort anything else.
2. Track two heaps: active, a min-heap of (end_time, slot) for slots currently occupied, and free, a min-heap of freed slot numbers.
3. Before assigning a new session, pop every entry off active whose end_time is at or before this session's start, pushing each freed slot number onto free — this is the "release before allocate" step.
4. If free is non-empty, pop the smallest slot number from it (this is what keeps slot numbers low and reused rather than always growing); otherwise mint a brand-new slot number.
5. Record the assignment, then push (end_time, slot) onto active for the newly occupied slot.
6. Because active is a min-heap on end time, checking "has anything freed up yet" is always an O(log n) peek/pop at the front, never a linear scan.

The key insight is that this is the same shape as the "minimum meeting rooms" problem, but instead of only counting how many rooms are needed, you also have to say which room each meeting gets — and reusing the lowest freed slot number (rather than any freed slot) is what keeps the assignment stable and minimal instead of monotonically increasing forever.

Reference solution

import heapq


def assign_connection_slots(sessions: list[tuple[float, float, str]]) -> dict[str, int]:
    # Two heaps: `active` holds (end_time, slot) for slots currently in use,
    # ordered so we can cheaply find which ones have freed up by "now";
    # `free` holds released slot numbers, ordered so we always reuse the
    # lowest-numbered one (like the "meeting rooms II" room-assignment trick).
    free: list[int] = []
    active: list[tuple[float, int]] = []
    next_slot = 0
    assignment: dict[str, int] = {}

    for start, end, session_id in sessions:
        # release every slot whose session ended at or before this one starts
        while active and active[0][0] <= start:
            _, slot = heapq.heappop(active)
            heapq.heappush(free, slot)
        if free:
            slot = heapq.heappop(free)
        else:
            slot = next_slot
            next_slot += 1
        assignment[session_id] = slot
        heapq.heappush(active, (end, slot))

    return assignment

Key Functions & Tricks

• active min-heap on end_time — lets you cheaply find and release every slot that's freed up by the current session's start.
• free min-heap of slot numbers — always hands out the lowest-numbered available slot, keeping the pool compact.
• while active and active[0][0] <= start: — releases all newly-free slots at once, not just one, before allocating.
• next_slot counter — only grows when there's truly no freed slot to reuse, so the pool size equals the true peak concurrency.

How to Recognize This Pattern