risingwavelabs/risingwave · error · BatchError

broken broadcast_channel

Error message

broken broadcast_channel

What it means

The broadcast channel between a batch task's sender(s) and receiver was closed while the receiver was still waiting for data. `BroadcastReceiver::recv` returns `None` when the underlying mpsc channel is dropped, and because a healthy exchange always terminates with an explicit barrier/Done signal rather than closing, an early close indicates a sender (or the task that owns it) failed or was torn down abruptly. The receiver converts this into an `Internal` error, "broken broadcast_channel", so the batch task fails fast instead of hanging or silently returning partial results.

Source

Thrown at src/batch/src/task/broadcast_channel.rs:79

                .await
                .map_err(|_| SenderError)?
        }

        Ok(())
    }
}

/// One or more `BroadcastReceiver`s corresponds to a single `BroadcastReceiver`
pub struct BroadcastReceiver {
    receiver: mpsc::Receiver<SharedResult<Option<DataChunkInChannel>>>,
}

impl ChanReceiver for BroadcastReceiver {
    async fn recv(&mut self) -> SharedResult<Option<DataChunkInChannel>> {
        match self.receiver.recv().await {
            Some(data_chunk) => data_chunk,
            // Early close should be treated as an error.
            None => Err(Arc::new(Internal(anyhow!("broken broadcast_channel")))),
        }
    }
}

pub fn new_broadcast_channel(
    shuffle: &ExchangeInfo,
    output_channel_size: usize,
) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {
    let broadcast_info = match shuffle.distribution {
        Some(exchange_info::Distribution::BroadcastInfo(ref v)) => *v,
        _ => BroadcastInfo::default(),
    };

    let output_count = broadcast_info.count as usize;
    let mut senders = Vec::with_capacity(output_count);
    let mut receivers = Vec::with_capacity(output_count);
    for _ in 0..output_count {
        let (s, r) = mpsc::channel(output_channel_size);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the logs of the task that owned the broadcast channel's senders for a preceding panic or error (the broken channel is almost always a secondary symptom).
2. Ensure every producer path sends the final barrier/Done chunk before dropping the sender, including error and cancellation paths.
3. Verify the upstream exchange/connection (e.g. gRPC exchange stream) is not being dropped prematurely due to network or timeout settings.
4. If writing new executor code, hold the sender until the loop completes and never use channel close as the completion signal.

Example fix

// before: producer drops sender early on error without signalling
if err.is_some() {
    return Err(err.unwrap()); // sender dropped here -> receiver sees broken channel
}
// after: propagate error through the channel as a chunk or barrier so the receiver gets a real signal
if let Err(e) = produce_chunk(&mut sender).await {
    sender.send(Err(e)).await?;
    return Ok(());
}

// Task output is SharedResult; treat channel errors as query failure and surface the root cause
match receiver.recv().await {
    Ok(Some(chunk)) => { /* process */ }
    Ok(None) => { /* graceful end */ }
    Err(e) => return Err(anyhow!("broadcast exchange terminated early: {}", e)),
}

When it happens

Trigger: Raised inside `BroadcastReceiver::recv` (src/batch/src/task/broadcast_channel.rs:79) when `mpsc::Receiver::recv().await` yields `None`, i.e. every `ChanSender` clone for the broadcast channel has been dropped before a `Done`/barrier chunk was sent. Typical calls: a sender task panicked or returned Err and dropped its sender without sending a final barrier; the executor upstream was cancelled; the channel was constructed with zero senders.

Common situations: A query fails midway (e.g. OOM, connection reset to an upstream exchange) causing the sender side of the exchange to be dropped; a developer adds an early `return`/`?` in a producer executor that skips sending the final barrier chunk; a worker process dies while a distributed query's shuffle exchange is still streaming; tests that drop the sender to 'finish' the channel instead of sending Done.

Related errors

• broken hash_shuffle_channel
• broken fifo_channel
• broken hash_shuffle_channel
• Unexpected exchange detected. We are either converting a sin
• ExchangeChannelClosed

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/fc53484e836e5323. Report an issue: GitHub.