{"record":{"id":"fc53484e836e5323","repo":"risingwavelabs/risingwave","slug":"broken-broadcast-channel","errorCode":null,"errorMessage":"broken broadcast_channel","messagePattern":"broken broadcast_channel","errorType":"error_code","errorClass":"BatchError","httpStatus":null,"severity":"error","filePath":"src/batch/src/task/broadcast_channel.rs","lineNumber":79,"sourceCode":"                .await\n                .map_err(|_| SenderError)?\n        }\n\n        Ok(())\n    }\n}\n\n/// One or more `BroadcastReceiver`s corresponds to a single `BroadcastReceiver`\npub struct BroadcastReceiver {\n    receiver: mpsc::Receiver<SharedResult<Option<DataChunkInChannel>>>,\n}\n\nimpl ChanReceiver for BroadcastReceiver {\n    async fn recv(&mut self) -> SharedResult<Option<DataChunkInChannel>> {\n        match self.receiver.recv().await {\n            Some(data_chunk) => data_chunk,\n            // Early close should be treated as an error.\n            None => Err(Arc::new(Internal(anyhow!(\"broken broadcast_channel\")))),\n        }\n    }\n}\n\npub fn new_broadcast_channel(\n    shuffle: &ExchangeInfo,\n    output_channel_size: usize,\n) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {\n    let broadcast_info = match shuffle.distribution {\n        Some(exchange_info::Distribution::BroadcastInfo(ref v)) => *v,\n        _ => BroadcastInfo::default(),\n    };\n\n    let output_count = broadcast_info.count as usize;\n    let mut senders = Vec::with_capacity(output_count);\n    let mut receivers = Vec::with_capacity(output_count);\n    for _ in 0..output_count {\n        let (s, r) = mpsc::channel(output_channel_size);","sourceCodeStart":61,"sourceCodeEnd":97,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/src/task/broadcast_channel.rs#L61-L97","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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).","Ensure every producer path sends the final barrier/Done chunk before dropping the sender, including error and cancellation paths.","Verify the upstream exchange/connection (e.g. gRPC exchange stream) is not being dropped prematurely due to network or timeout settings.","If writing new executor code, hold the sender until the loop completes and never use channel close as the completion signal."],"exampleFix":"// before: producer drops sender early on error without signalling\nif err.is_some() {\n    return Err(err.unwrap()); // sender dropped here -> receiver sees broken channel\n}\n// after: propagate error through the channel as a chunk or barrier so the receiver gets a real signal\nif let Err(e) = produce_chunk(&mut sender).await {\n    sender.send(Err(e)).await?;\n    return Ok(());\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Task output is SharedResult; treat channel errors as query failure and surface the root cause\nmatch receiver.recv().await {\n    Ok(Some(chunk)) => { /* process */ }\n    Ok(None) => { /* graceful end */ }\n    Err(e) => return Err(anyhow!(\"broadcast exchange terminated early: {}\", e)),\n}","preventionTips":["Always send an explicit Done/barrier chunk instead of dropping senders to signal completion","Send error results through the channel rather than dropping senders on failure paths","Keep senders alive for the whole producer loop, even in early-return branches","Monitor batch task logs for panics — broken channels are usually secondary errors"],"tags":["batch","channel","exchange"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}