{"record":{"id":"a5f5e652d9057c59","repo":"risingwavelabs/risingwave","slug":"broken-fifo-channel","errorCode":null,"errorMessage":"broken fifo_channel","messagePattern":"broken fifo_channel","errorType":"error_code","errorClass":"BatchError","httpStatus":null,"severity":"error","filePath":"src/batch/src/task/fifo_channel.rs","lineNumber":61,"sourceCode":"        let data = DataChunkInChannel::new(chunk);\n        self.sender\n            .send(Ok(Some(data)))\n            .await\n            .map_err(|_| SenderError)\n    }\n\n    async fn close(self, error: Option<Arc<BatchError>>) -> BatchResult<()> {\n        let result = error.map(Err).unwrap_or(Ok(None));\n        self.sender.send(result).await.map_err(|_| SenderError)\n    }\n}\n\nimpl ChanReceiver for FifoReceiver {\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 error.\n            None => Err(Arc::new(Internal(anyhow!(\"broken fifo_channel\")))),\n        }\n    }\n}\n\npub fn new_fifo_channel(output_channel_size: usize) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {\n    let (s, r) = mpsc::channel(output_channel_size);\n    (\n        ChanSenderImpl::Fifo(FifoSender { sender: s }),\n        vec![ChanReceiverImpl::Fifo(FifoReceiver { receiver: r })],\n    )\n}\n\nmod tests {\n    #[tokio::test]\n    async fn test_recv_not_fail_on_closed_channel() {\n        use crate::task::fifo_channel::new_fifo_channel;\n\n        let (sender, mut receivers) = new_fifo_channel(64);","sourceCodeStart":43,"sourceCodeEnd":79,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/src/task/fifo_channel.rs#L43-L79","documentation":"The FIFO (single-producer/single-consumer) channel feeding a batch executor was closed before a proper end-of-stream signal was delivered. `FifoReceiver::recv` maps `None` from the underlying mpsc receiver — meaning the sender was dropped without sending the final Done/barrier chunk — into the `Internal` error \"broken fifo_channel\", so the consuming task fails instead of treating closure as graceful termination.","triggerScenarios":"Raised inside `FifoReceiver::recv` (src/batch/src/task/fifo_channel.rs:61) when `mpsc::Receiver::recv().await` returns `None` after the channel created by `new_fifo_channel(output_channel_size)` has had its sole sender dropped without a terminal chunk. Happens when the producing executor returns early, panics, or is cancelled.","commonSituations":"An upstream executor in the same batch task errors out (scan failure, OOM) and its sender is dropped during task teardown; task cancellation races with normal completion; new executor implementations forget to emit the final Done chunk on early-return paths.","solutions":["Inspect the producing executor's logs for the original panic or error that caused its sender to drop; fix that root cause.","Make every executor send the final barrier/Done chunk before returning, including on error and cancellation paths.","If the producer can fail, send the error through the channel (chunks are `SharedResult`) rather than dropping the sender.","For tests, assert that producers terminate via Done chunks, not by dropping senders."],"exampleFix":"// before\nlet rows = scan().await?; // ? drops sender on error -> broken fifo_channel downstream\nsender.send(Ok(done_chunk())).await?;\n// after\nmatch scan().await {\n    Ok(rows) => sender.send(Ok(rows)).await?,\n    Err(e) => {\n        sender.send(Err(e)).await?; // receiver gets the real error\n        return Ok(());\n    }\n}\nsender.send(Ok(done_chunk())).await?;","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match receiver.recv().await {\n    Ok(Some(chunk)) => { /* process */ }\n    Ok(None) => return Err(anyhow!(\"fifo channel closed before Done\")),\n    Err(e) => return Err(e), // real producer error sent through the channel\n}","preventionTips":["Structure executors so the sender is dropped only after the final Done chunk is sent","Send errors via `sender.send(Err(e))` instead of early-returning with the sender alive-drop","Write executor tests asserting receivers always see Done, never None","Check cancellation handling: cancelled producers should still signal, not vanish"],"tags":["batch","channel","fifo"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}