Function guarded_receiver_stream 

pub fn guarded_receiver_stream<T, G>(
    rx: Receiver<T>,
    guard: G,
) -> impl Stream<Item = T> + Send + 'static
where
    T: Send + 'static,
    G: Send + 'static,

Wraps a raw mpsc::Receiver into a Stream, keeping guard alive for exactly as long as the stream is — dropped together whether the stream ends normally (overflow, see SubscriptionPush) or is cancelled mid-poll (an ordinary client disconnect just drops this whole future). This is what lets a cratestack_core::SubscriptionGuard passed as guard unsubscribe cleanly in either case without the caller needing to distinguish which one happened.