mux

BlockMuxStream

BlockMuxStream(reader, start=None, timeout=0, on_gap=ONGAP_DEFAULT, stride=None, label=None)

Bases: Generic[T]

A time-aware, gap-handling multiplexer for SeriesBlock streams.

Given SeriesBlocks from multiple named streams with monotonically increasing integer timestamps, this data structure can be used to pull out sets of synchronized blocks, all with the same timestamps. If data on the streams is not available before timeouts are reached, gap blocks will be returned.

The oldest items will be held until either all named streams are available or until the timeout has been reached. If a start time has been set, any items with an older timestamp will be rejected.

Parameters:

Name	Type	Description	Default
reader	StreamReader	StreamReader object producing multiple stream to multiplex.	required
start	int	GPS start time of stream, in nanoseconds.	None
timeout	int = 0	Overall timeout for the muxer, in nanoseconds. Overrides individual queue timeouts. If not specified the mux timeout will be the max of the individual queue timeouts.	0
on_gap	str	Policy for strides in which no stream has any data. 'fill' (default) emits masked gap blocks so the output timeline is continuous; 'skip' emits nothing for such strides (sparse output, no cost for absence); 'raise' raises GapError. Strides where at least one stream has data are always emitted (with the missing channels masked), whatever the policy.	ONGAP_DEFAULT
stride	int	Duration of output blocks, in nanoseconds. Must be an integer multiple of the native stride — the least common multiple of the individual stream strides — which is also the default. Larger strides aggregate multiple native blocks into each output block.	None

Source code in arrakis/mux.py

def __init__(
    self,
    reader: HasStreams,
    start: int | None = None,
    timeout: int = 0,
    on_gap: str = ONGAP_DEFAULT,
    stride: int | None = None,
    label: str | None = None,
):
    self.reader = reader
    self.start = start
    self.on_gap = OnGap[on_gap.upper()]
    # context prefix for the queues' names in log messages
    self.label = label
    # streams currently dropping stride-mismatched blocks, so the
    # recovery transition can be logged when good blocks resume
    self._stride_mismatched: set[str] = set()
    # extract stream info
    self._queues: dict[str, TimedQueue] = {}
    self._stream_channels: dict[str, list[Channel]] = {}
    self.channels: list[Channel] = []
    has_latency_constraint = True
    for stream_name, channels in self.reader.streams.items():
        # gather stride and latency
        # done this way to satisfy type checking consistency
        strides = []
        timeouts = []
        for channel in channels:
            assert channel.stride
            strides.append(channel.stride)
            if channel.max_latency is not None:
                timeouts.append(channel.max_latency)
        # the stride for a stream is the least common multiple of
        # the stride of the individual channels
        qstride = math.lcm(*strides)
        # timeout is max of all expected latencies, or 0 if none
        # have a latency constraint (e.g. historical data)
        if timeouts:
            # one stride of headroom prevents gap-fill from racing
            # data that is in flight between the gRPC thread and
            # the TimedQueue
            qtimeout = max(timeouts) + qstride
        else:
            qtimeout = 0
            has_latency_constraint = False
        self._queues[stream_name] = TimedQueue(
            stride=qstride,
            # mux timeout overrides individual timeouts
            timeout=timeout or qtimeout,
            start=start,
            name=describe_stream(stream_name, channels, label),
        )
        self._stream_channels[stream_name] = list(channels)
        self.channels.extend(channels)
    # the overall stride for multiple streams is the least common
    # multiple of the stride of the individual streams
    self._base_stride = math.lcm(*(q.stride for q in self._queues.values()))
    if stride is None:
        stride = self._base_stride
    elif stride <= 0 or stride % self._base_stride:
        msg = (
            f"stride ({stride:_} ns) must be a positive integer "
            f"multiple of the stream's native stride "
            f"({self._base_stride:_} ns)"
        )
        raise ValueError(msg)
    self.stride = stride
    # a full output stride must be buffered before it can be
    # served; keep the queue backpressure caps above that
    for q in self._queues.values():
        q.ensure_buffer(self.stride)
    self.timeout = max(q.timeout for q in self._queues.values())
    # The timeout deadline is floored to the stride grid, so the
    # timeout and alignment pass over every queue in ready() can
    # only change anything once per slot of wall-clock time (of the
    # finest queue stride); ready() is called after every push and
    # would otherwise cost O(queues) per push.  The pass also runs
    # whenever a queue has just realigned to its first data, the
    # one event that moves queues out of lockstep.
    self._min_stride = min(q.stride for q in self._queues.values())
    self._pass_slot = -1
    self._needs_align = True
    # Disabled when streams have no latency constraint (e.g.
    # historical data from frames) since there is no wall-clock
    # deadline to gap-fill against.
    self._use_timeouts = has_latency_constraint

__getitem__

__getitem__(key)

Access an individual queue.

Source code in arrakis/mux.py

def __getitem__(self, key: str) -> TimedQueue:
    """Access an individual queue."""
    return self._queues[key]

declare_complete

declare_complete(time_ns)

Declare all data before time_ns final: absence is gap.

Advances every queue's known horizon to at least time_ns (rounded up to each queue's stride) without storing anything, so pulls can serve the span with missing data synthesized as gaps. For a bounded request whose source is exhausted, this lets the stream finish with explicit gaps instead of waiting forever for data that cannot arrive; data older than the declared horizon arriving afterwards is dropped as too old.

Source code in arrakis/mux.py

def declare_complete(self, time_ns: int) -> None:
    """Declare all data before ``time_ns`` final: absence is gap.

    Advances every queue's known horizon to at least ``time_ns``
    (rounded up to each queue's stride) without storing anything,
    so pulls can serve the span with missing data synthesized as
    gaps.  For a bounded request whose source is exhausted, this
    lets the stream finish with explicit gaps instead of waiting
    forever for data that cannot arrive; data older than the
    declared horizon arriving afterwards is dropped as too old.
    """
    for q in self._queues.values():
        # push() treats its time as an element slot start and
        # advances the horizon one stride past it
        last_slot = -(-time_ns // q.stride) * q.stride - q.stride
        if last_slot >= q.horizon:
            q.push(last_slot, None)

pull

pull(stride=None)

Pull synchronized, concatenated, combined blocks from all streams covering the overall specified stride.

Parameters:

Name	Type	Description	Default
stride	int	Duration to pull, in nanoseconds. Defaults to the output stride; the end-of-stream tail flush pulls at the native stride.	None

Returns:

Type	Description
SeriesBlock or None	The combined block for the next stride, or None if the stride held no data and was skipped (on_gap='skip').

Source code in arrakis/mux.py

def pull(self, stride: int | None = None) -> SeriesBlock | None:
    """Pull synchronized, concatenated, combined blocks from all
    streams covering the overall specified stride.

    Parameters
    ----------
    stride : int, optional
        Duration to pull, in nanoseconds.  Defaults to the output
        stride; the end-of-stream tail flush pulls at the native
        stride.

    Returns
    -------
    SeriesBlock or None
        The combined block for the next stride, or None if the
        stride held no data and was skipped (``on_gap='skip'``).

    """
    if stride is None:
        stride = self.stride
    if self.on_gap is not OnGap.FILL and self._next_stride_is_all_gap(stride):
        if self.on_gap is OnGap.RAISE:
            front = min(q.cursor for q in self._queues.values())
            msg = f"gap in stream at {front:_} ns"
            raise GapError(msg)
        self._skip_gap_span(stride)
        return None
    blocks = []
    for stream_name, queue in self._queues.items():
        q_blocks = []
        for time_ns, block in queue.pull(stride, update_timeout=False):
            if block is None:
                block = SeriesBlock.full_gap(
                    time_ns,
                    queue.stride,
                    self._stream_channels[stream_name],
                )
            q_blocks.append(block)
        q_block = concatenate_blocks(*q_blocks)
        blocks.append(q_block)
    return combine_blocks(*blocks)

push

push(stream_name, block)

Push an element for time into a particular queue.

Source code in arrakis/mux.py

def push(self, stream_name: str, block: SeriesBlock):
    """Push an element for time into a particular queue."""
    q = self._queues[stream_name]
    if block.duration_ns != q.stride:
        # allow partial-stride edge blocks for non-aligned requests
        if q.start is not None and block.duration_ns < q.stride:
            pass
        else:
            self._stride_mismatched.add(stream_name)
            q.count_drop(
                "stride-mismatched",
                f"block at {block.time_ns} ns for stream {stream_name}: "
                f"got {block.duration_ns} ns, expected {q.stride} ns",
            )
            return
    if stream_name in self._stride_mismatched:
        self._stride_mismatched.discard(stream_name)
        logger.info(
            "stream %s: block stride matches again at %d ns "
            "(%d unreported drop(s))",
            stream_name,
            block.time_ns,
            q.finish_drop("stride-mismatched"),
        )
    was_initialized = q.initialized
    q.push(block.time_ns, block)
    if not was_initialized and q.initialized:
        self._needs_align = True

ready

ready()

True if all queues have the expected number of elements

covering the overall muxer stride.

Source code in arrakis/mux.py

def ready(self) -> bool:
    """True if all queues have the expected number of elements

    covering the overall muxer stride.

    """
    # Skip timeouts for muxes without a latency constraint: no
    # synchronization needed, and timeouts would race incoming
    # data causing spurious drops.
    if self._use_timeouts:
        now_ns = time_as_ns(gpsnow())
        slot = now_ns // self._min_stride
        if slot != self._pass_slot or self._needs_align:
            self._pass_slot = slot
            self._needs_align = False
            # 1. Update timeouts on ALL queues first so gap-filling
            #    is consistent before any alignment happens.
            for q in self._queues.values():
                q.update_timeout(now_ns)
            # 2. Align queues to the same front timestamp.  Only
            #    live streams can diverge (queues realign to their
            #    first data); bounded requests share the requested
            #    start, where the stride ceiling would instead trim
            #    data after a non-stride-aligned start.
            if self.start is None:
                self._align_queues()
    # 3. Check readiness WITHOUT re-triggering timeouts — the
    #    alignment established in step 2 must not be disturbed.
    #    Short-circuits at the first queue still waiting for data.
    stride = self.stride
    queues = self._queues.values()
    if not all(q.horizon - q.cursor >= stride for q in queues):
        return False
    if self.start is None and len(self._queues) > 1:
        # pull() combines one slot from every queue, so they must
        # serve the same time; realign if anything has diverged
        # since the last pass
        cursor = next(iter(queues)).cursor
        if any(q.cursor != cursor for q in queues):
            self._align_queues()
            return all(q.horizon - q.cursor >= stride for q in queues)
    return True

GapError

Bases: Exception

Raised when a stream encounters a gap and on_gap='raise'.

HasStreams

Bases: Protocol

Minimal protocol for objects that provide stream channel metadata.

TimedQueue

TimedQueue(stride, timeout, start=None, *, name='')

Bases: Generic[T]

A sequential, time-stamped queue handling gaps and timeouts.

The queue stores only real elements; the spans between them are gaps, synthesized lazily when pulled. Two integer cursors define the queue state:

• cursor: the next timestamp to be served by :meth:pull.
• horizon: the end (exclusive) of the known span. Every stride in [cursor, horizon) is either a stored element or a gap. Elements older than the horizon are rejected.

The horizon advances when an element is pushed (per-stream delivery is assumed to be in time order, so an element at time T implies nothing older is still coming) and, for live streams, when the timeout deadline passes (absence past the deadline is declared to be a gap). A discontinuity of any size therefore costs no memory: it is served as synthesized gaps between the two stored elements that surround it.

Parameters:

Name	Type	Description	Default
start	int	GPS start time of queue, in nanoseconds.	None
stride	int	Time step for elements in the queue, in nanoseconds.	required
timeout	int	Timeout for elements in the queue, in nanoseconds.	required

Source code in arrakis/mux.py

def __init__(
    self,
    stride: int,
    timeout: int,
    start: int | None = None,
    *,
    name: str = "",
):
    self.stride = stride
    self.timeout = timeout
    self.start = start
    # label for log messages (the stream this queue serves)
    self.name = name
    # gap-episode state for transition logging: the slot at which
    # the current run of synthesized gaps began, or None while
    # data is being served
    self._gap_start: int | None = None
    self._gap_warned = False
    self._gap_warn_after = max(timeout, int(GAP_WARN_MIN * Time.SECONDS))
    self._served_data = False
    # Track if queue has received data yet (only relevant when start is None)
    self._initialized = start is not None
    if start is None:
        start = time_as_ns(gpsnow())
    # serve from the stride boundary at or before start, so that a
    # non-aligned edge element at start falls in the first slot
    self.cursor = (start // self.stride) * self.stride
    self.horizon = self.cursor
    # real elements only, strictly increasing times
    self._queue: deque = deque()
    # Backpressure cap on *buffered real elements* (gap spans cost
    # nothing).  Scaled to 2x the timeout window, with a floor of
    # 1000 for historical/no-timeout streams.
    self._max_queue = max(1000, int(2 * self.timeout / self.stride))
    # True once any slot has been served (or the cursor committed
    # by alignment): from then on the serving position must only
    # move forward, so the first-data realignment is disallowed
    self._committed = False
    # True once any slot has actually been served: distinguishes
    # expected startup-alignment trims from genuine late drops
    self._served = False
    self._drop_counts: dict[str, int] = {}
    self._drop_log_times: dict[str, float] = {}
    self._lock = RLock()

initialized property

initialized

Whether the queue has aligned to its first element (live queues) or was given a start time.

__len__

__len__()

Number of buffered real elements (gap spans are implicit).

Source code in arrakis/mux.py

def __len__(self):
    """Number of buffered real elements (gap spans are implicit)."""
    return len(self._queue)

count_drop

count_drop(kind, detail)

Count a dropped element; log a summary at most once per DROP_LOG_INTERVAL seconds per kind (first drop logs immediately), so sustained loss leaves a fingerprint in the logs without flooding them.

Source code in arrakis/mux.py

def count_drop(self, kind: str, detail: str) -> None:
    """Count a dropped element; log a summary at most once per
    DROP_LOG_INTERVAL seconds per kind (first drop logs
    immediately), so sustained loss leaves a fingerprint in the
    logs without flooding them."""
    self._drop_counts[kind] = self._drop_counts.get(kind, 0) + 1
    now = time.monotonic()
    last = self._drop_log_times.get(kind)
    if last is None or now - last >= DROP_LOG_INTERVAL:
        logger.warning(
            "dropped %d %s element(s) since last report; latest: %s",
            self._drop_counts[kind],
            kind,
            detail,
        )
        self._drop_counts[kind] = 0
        self._drop_log_times[kind] = now

drain_until

drain_until(target_time)

Advance the cursor so serving resumes at or after target_time.

Source code in arrakis/mux.py

def drain_until(self, target_time: int) -> None:
    """Advance the cursor so serving resumes at or after target_time."""
    # Snap to stride boundary so the cursor stays stride-aligned.
    target_time = (target_time // self.stride) * self.stride
    with self._lock:
        if target_time <= self.cursor:
            return
        self.cursor = target_time
        self._committed = True
        while self._queue and self._queue[0][0] < self.cursor:
            self._queue.popleft()
        self.horizon = max(self.horizon, self.cursor)

ensure_buffer

ensure_buffer(duration)

Raise the backpressure cap to hold at least the given span of elements, with the usual 2x headroom (used when the muxer serves multiples of the queue stride).

Source code in arrakis/mux.py

def ensure_buffer(self, duration: int) -> None:
    """Raise the backpressure cap to hold at least the given
    span of elements, with the usual 2x headroom (used when the
    muxer serves multiples of the queue stride)."""
    self._max_queue = max(self._max_queue, int(2 * duration / self.stride))

finish_drop

finish_drop(kind)

End a drop episode for kind: return and clear the unreported count and the rate-limit state, so the next drop of this kind warns immediately again.

Source code in arrakis/mux.py

def finish_drop(self, kind: str) -> int:
    """End a drop episode for *kind*: return and clear the
    unreported count and the rate-limit state, so the next drop
    of this kind warns immediately again."""
    self._drop_log_times.pop(kind, None)
    return self._drop_counts.pop(kind, 0)

front_time

front_time()

Return the next timestamp to be served, or None if nothing is known.

Source code in arrakis/mux.py

def front_time(self) -> int | None:
    """Return the next timestamp to be served, or None if nothing is known."""
    with self._lock:
        if self.horizon > self.cursor or self._queue:
            return self.cursor
        return None

next_element_time

next_element_time()

Return the timestamp of the oldest buffered element, or None.

Unlike :meth:front_time this ignores gap spans: it reports where the next real element is, however far ahead.

Source code in arrakis/mux.py

def next_element_time(self) -> int | None:
    """Return the timestamp of the oldest buffered element, or None.

    Unlike :meth:`front_time` this ignores gap spans: it reports
    where the next real element is, however far ahead.
    """
    with self._lock:
        if self._queue:
            return self._queue[0][0]
        return None

pull

pull(duration=None, *, update_timeout=True)

Drain the queue.

Gaps are represented by None elements, synthesized on the fly.

Parameters:

Name	Type	Description	Default
duration	int	Duration to extract, in nanoseconds. If the specified duration is not available, no elements will be returned. If not specified, the known span will be drained.	None
update_timeout	bool	Whether to trigger timeout gap-filling before pulling. Default is True. Set to False when the caller has already updated timeouts (e.g. BlockMuxStream.pull()).	True

Yields:

Type	Description
tuple[time, element]	The element from the queue and it's associated timestamp.

Source code in arrakis/mux.py

def pull(
    self, duration: int | None = None, *, update_timeout: bool = True
) -> Iterator[tuple[int, Any]]:
    """Drain the queue.

    Gaps are represented by None elements, synthesized on the fly.

    Parameters
    ----------
    duration : int
        Duration to extract, in nanoseconds.  If the specified
        duration is not available, no elements will be returned.
        If not specified, the known span will be drained.
    update_timeout : bool, optional
        Whether to trigger timeout gap-filling before pulling.
        Default is True.  Set to False when the caller has already
        updated timeouts (e.g. BlockMuxStream.pull()).

    Yields
    ------
    tuple[time, element]
        The element from the queue and it's associated timestamp.

    """
    if update_timeout:
        self.update_timeout()

    with self._lock:
        # if duration specified, pull the requested number of elements
        if duration:
            if n_elements := self.ready(duration, update_timeout=False):
                for _ in range(n_elements):
                    yield self._serve()

        # else drain the known span
        else:
            while self.cursor < self.horizon or self._queue:
                yield self._serve()

push

push(time, element, on_drop=ONDROP_DEFAULT)

Push an element into the queue.

The time being pushed into the queue must be a multiple of the time stride specified at initialization of the queue.

If the time associated with the pushed element is older than the queue's horizon (already served or declared gap), the element will be dropped and this operation will be a no-op.

Pushing None declares the span up to time to be a gap without storing anything.

Parameters:

Name	Type	Description	Default
time	int	GPS time associated with the element, in nanoseconds	required
element	Any	element being pushed into the queue.	required
on_drop	str	Per-event behavior when the item is dropped as too old (e.g. it arrived after the timeout declared its span a gap, or it is a duplicate). Options are 'ignore', 'raise', or 'warn'. Default is 'ignore': drops are expected under the latency contract, and are always counted and summarized in the log at most once per DROP_LOG_INTERVAL regardless of this policy.	ONDROP_DEFAULT

Source code in arrakis/mux.py

def push(self, time: int, element: Any, on_drop: str = ONDROP_DEFAULT) -> None:
    """Push an element into the queue.

    The time being pushed into the queue must be a multiple of the
    time stride specified at initialization of the queue.

    If the time associated with the pushed element is older than
    the queue's horizon (already served or declared gap), the
    element will be dropped and this operation will be a no-op.

    Pushing ``None`` declares the span up to ``time`` to be a gap
    without storing anything.

    Parameters
    ----------
    time : int
        GPS time associated with the element, in nanoseconds
    element : Any
        element being pushed into the queue.
    on_drop : str, optional
        Per-event behavior when the item is dropped as too old
        (e.g. it arrived after the timeout declared its span a
        gap, or it is a duplicate).  Options are 'ignore', 'raise',
        or 'warn'.  Default is 'ignore': drops are expected under
        the latency contract, and are always counted and
        summarized in the log at most once per DROP_LOG_INTERVAL
        regardless of this policy.

    """
    assert time % self.stride == 0 or time == self.start, (
        f"time {time} is not a multiple of queue stride {self.stride:_}"
        f" (and does not match start {self.start})"
    )
    # If this is the first real element on a live queue, realign to
    # it: serving starts at its stride boundary, and any span the
    # timeout may have declared gap in the meantime is discarded.
    # Only allowed while nothing has been served yet — once a slot
    # has been yielded (e.g. a timeout gap emitted just before the
    # first element landed), rewinding would duplicate it, so the
    # element falls through to the normal too-old handling instead.
    if not self._initialized and element is not None:
        if not self._committed:
            with self._lock:
                self.cursor = (time // self.stride) * self.stride
                self.horizon = self.cursor
        self._initialized = True

    # if time is older than the horizon, drop it
    if time < self.horizon:
        msg = f"item's timestamp is too old: ({time:_} < {self.horizon:_})"
        match OnDrop[on_drop.upper()]:
            case OnDrop.IGNORE:
                if element is not None:
                    # on live streams, elements trimmed by the
                    # startup alignment (before anything has been
                    # served) are expected, not data loss
                    if self.start is None and not self._served:
                        logger.debug("startup alignment trim: %s", msg)
                    else:
                        self.count_drop("late", msg)
                return
            case OnDrop.RAISE:
                raise ValueError(msg)
            case OnDrop.WARN:
                # the rate-limited drop summary covers the log; the
                # python warning is what distinguishes this policy
                if element is not None:
                    self.count_drop("late", msg)
                warnings.warn(msg, stacklevel=2)
                return
    with self._lock:
        if element is not None:
            # Never raise on overflow: this runs inside long-lived
            # server poll threads, where an exception silently
            # kills the stream for every subscriber.  Bound memory
            # by dropping the oldest element instead; the dropped
            # span is served as a gap.
            if len(self._queue) >= self._max_queue:
                self._queue.popleft()
                self.count_drop(
                    "overflow",
                    f"queue at capacity ({self._max_queue} elements)",
                )
            self._queue.append((time, element))
        # the element's slot is now known; for a non-aligned edge
        # element this is the enclosing stride boundary
        self.horizon = (time // self.stride) * self.stride + self.stride

ready

ready(duration, *, update_timeout=True)

Check if queue holds duration worth of elements

Parameters:

Name	Type	Description	Default
duration	int	Duration to check for, in nanoseconds.	required
update_timeout	bool	Whether to trigger timeout gap-filling before checking. Default is True. Set to False when the caller has already updated timeouts (e.g. BlockMuxStream.ready()).	True

Returns:

Type	Description
int or None	Returns either the number of elements that span duration, or None if the known span does not cover duration.

Source code in arrakis/mux.py

def ready(self, duration: int, *, update_timeout: bool = True) -> int | None:
    """Check if queue holds duration worth of elements

    Parameters
    ----------
    duration : int
        Duration to check for, in nanoseconds.
    update_timeout : bool, optional
        Whether to trigger timeout gap-filling before checking.
        Default is True.  Set to False when the caller has already
        updated timeouts (e.g. BlockMuxStream.ready()).

    Returns
    -------
    int or None
        Returns either the number of elements that span duration,
        or None if the known span does not cover duration.

    """
    if update_timeout:
        self.update_timeout()
    assert duration % self.stride == 0, (
        f"duration {duration:_} is not a multiple of queue stride {self.stride:_}"
    )
    if self.horizon - self.cursor >= duration:
        return int(duration / self.stride)
    return None

update_timeout

update_timeout(now_ns=None)

Declare absence past the timeout deadline a gap.

now_ns is the current GPS time in nanoseconds; a caller updating many queues at once passes one reading to all of them.

Source code in arrakis/mux.py

def update_timeout(self, now_ns: int | None = None):
    """Declare absence past the timeout deadline a gap.

    ``now_ns`` is the current GPS time in nanoseconds; a caller
    updating many queues at once passes one reading to all of them.
    """
    # only do timeouts for live streams (i.e. start is None)
    if self.start is not None:
        return
    if now_ns is None:
        now_ns = time_as_ns(gpsnow())
    # absence past the timeout deadline is declared to be a gap:
    # advance the horizon so pulls can serve the span.  Data
    # arriving later than this is dropped as too old.
    deadline = now_ns - self.timeout
    deadline = (deadline // self.stride) * self.stride
    with self._lock:
        self.horizon = max(self.horizon, deadline + self.stride)

describe_stream

describe_stream(stream_name, channels, label=None)

A readable name for a stream's queue in log messages.

Stream keys are transport identifiers (a partition id, a Flight endpoint), so the name leads with the channels the stream carries, keeps the key for correlation, and takes an optional caller label for context (e.g. which reader or request the queue belongs to).

Source code in arrakis/mux.py

def describe_stream(
    stream_name: str, channels: Sequence[Channel], label: str | None = None
) -> str:
    """A readable name for a stream's queue in log messages.

    Stream keys are transport identifiers (a partition id, a Flight
    endpoint), so the name leads with the channels the stream carries,
    keeps the key for correlation, and takes an optional caller *label*
    for context (e.g. which reader or request the queue belongs to).
    """
    names = sorted(channel.name for channel in channels)
    if not names:
        chans = "no channels"
    elif len(names) == 1:
        chans = names[0]
    else:
        chans = f"{names[0]} +{len(names) - 1} more"
    desc = f"{chans} ({stream_name})"
    return f"{label} {desc}" if label else desc