sync::broadcast returns Lagged error when capacity is not exceeded

[iffyio]

Version

rustc 1.41.0 (5e1a79984 2020-01-27)
tokio: 0.2.18

Platform

Linux 5.3.0-45-generic #37~18.04.1-Ubuntu

Description

Hi! I ran into an issue on the sync::broadcast::channel where buffering events up to the channel's capacity triggers a RecvError::Lagged error when the channel is finally read from. I expected this error to only be triggered if the channel's capacity is exceeded.

As an example the following code errors out with Err(Lagged(1)).

#[tokio::main]
async fn main() {
    let (tx, mut rx) = tokio::sync::broadcast::channel(1);
    let _1 = tokio::spawn(async move {
        tx.send(0).unwrap();
    });
    let _2 = tokio::spawn(async move {
        rx.recv().await.unwrap(); // panic
    });
    _1.await.unwrap();
    _2.await.unwrap();
}

It seems to be a threading issue? compared to the following that works as expected.

#[tokio::main]
async fn main() {
    let (tx, mut rx) = tokio::sync::broadcast::channel(1);
    tx.send(0).unwrap();
    let _2 = tokio::spawn(async move {
        rx.recv().await.unwrap(); // OK
    });
    _2.await.unwrap();
}

[Darksonn]

This appears to be a bug. I will take a look at it when I have time.

[Darksonn]

This is because the stream sends a None message when it has been closed. It would be nice to not count that None message as part of the message limit, though.