risingwavelabs/risingwave · error · BatchError

broken fifo_channel

Error message

broken fifo_channel

What it means

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.

Source

Thrown at src/batch/src/task/fifo_channel.rs:61

        let data = DataChunkInChannel::new(chunk);
        self.sender
            .send(Ok(Some(data)))
            .await
            .map_err(|_| SenderError)
    }

    async fn close(self, error: Option<Arc<BatchError>>) -> BatchResult<()> {
        let result = error.map(Err).unwrap_or(Ok(None));
        self.sender.send(result).await.map_err(|_| SenderError)
    }
}

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

pub fn new_fifo_channel(output_channel_size: usize) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {
    let (s, r) = mpsc::channel(output_channel_size);
    (
        ChanSenderImpl::Fifo(FifoSender { sender: s }),
        vec![ChanReceiverImpl::Fifo(FifoReceiver { receiver: r })],
    )
}

mod tests {
    #[tokio::test]
    async fn test_recv_not_fail_on_closed_channel() {
        use crate::task::fifo_channel::new_fifo_channel;

        let (sender, mut receivers) = new_fifo_channel(64);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the producing executor's logs for the original panic or error that caused its sender to drop; fix that root cause.
2. Make every executor send the final barrier/Done chunk before returning, including on error and cancellation paths.
3. If the producer can fail, send the error through the channel (chunks are `SharedResult`) rather than dropping the sender.
4. For tests, assert that producers terminate via Done chunks, not by dropping senders.

Example fix

// before
let rows = scan().await?; // ? drops sender on error -> broken fifo_channel downstream
sender.send(Ok(done_chunk())).await?;
// after
match scan().await {
    Ok(rows) => sender.send(Ok(rows)).await?,
    Err(e) => {
        sender.send(Err(e)).await?; // receiver gets the real error
        return Ok(());
    }
}
sender.send(Ok(done_chunk())).await?;

match receiver.recv().await {
    Ok(Some(chunk)) => { /* process */ }
    Ok(None) => return Err(anyhow!("fifo channel closed before Done")),
    Err(e) => return Err(e), // real producer error sent through the channel
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• broken broadcast_channel
• broken hash_shuffle_channel
• broken hash_shuffle_channel
• query_epoch not set in distributed lookup join
• Storage error: {0}

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