windmill-labs/windmill · warning · std::io::Error (BrokenPipe)

Channel send error: {}

Error message

Channel send error: {}

What it means

ChannelWriter wraps an async mpsc receiver for streaming Parquet data and uses `blocking_send` in its sync `Write` impl. If the receiver side has been dropped (consumer cancelled, connection closed, task aborted), the send fails and is converted into an io::Error with kind BrokenPipe reading 'Channel send error: {}'.

Source

Thrown at backend/windmill-object-store/src/lib.rs:1325

    if row_count > DEFAULT_SCHEMA_INFER_MAX_RECORD as u64
        && schema.fields().iter().any(|f| is_untyped(f.data_type()))
    {
        return infer(None);
    }
    Ok(schema)
}

#[cfg(feature = "parquet")]
struct ChannelWriter {
    sender: tokio::sync::mpsc::Sender<anyhow::Result<Bytes>>,
}

#[cfg(feature = "parquet")]
impl Write for ChannelWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let data: Bytes = buf.to_vec().into();
        self.sender.blocking_send(Ok(data)).map_err(|e| {
            std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                format!("Channel send error: {}", e),
            )
        })?;
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[cfg(not(feature = "parquet"))]
#[derive(Debug, Clone, Copy, Default)]
pub struct IngestStats {
    pub rows: u64,
    pub bytes: u64,
    pub elapsed: std::time::Duration,

View on GitHub (pinned to e474e8803c)

Solutions

1. Treat as a cancelled stream: stop producing and drop the writer; retry the export if the data is still needed.
2. Fix the consumer-side error that dropped the receiver first (check its logs/panic).
3. Increase consumer timeouts or stream in pages so slow clients don't hit cancellation.
4. If the receiver is dropped intentionally on early exit, ensure the writer loop checks for that condition instead of continuing to write.

Example fix

// before
for row in rows {
    writer.write(&serialize(row))?; // keeps failing once receiver is gone
}
// after
for row in rows {
    if sender.is_closed() { break; } // or propagate a clean cancellation
    writer.write(&serialize(row))?;
}

// check the consumer is still alive before writing
if channel_writer.sender_is_closed() { /* stop producing, release resources */ }

match writer.write(&chunk) {
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
        tracing::info!("consumer gone, aborting parquet stream: {e}");
        break; // treat as cancellation, not a data error
    }
    Ok(n) => { /* continue */ }
    Err(e) => return Err(e.into()),
}

When it happens

Trigger: Writing to a ChannelWriter (parquet feature) after the receiving end of the channel was dropped — e.g. the HTTP response/stream consumer was cancelled, the client disconnected, or the receiving task panicked/finished early.

Common situations: Client aborts a large Parquet download mid-stream; downstream task erroring out while the writer keeps producing; timeouts cancelling the consumer while a query still streams results.

Related errors

• No response body for SSE stream
• AI response was empty
• SSE error: ${previewJobUpdates}
• Parquet S3 input requires the `parquet` feature to be enable
• Error streaming get request from authed http client to {url}

AI-assisted analysis of windmill-labs/windmill@e474e8803c (2026-09-03). Data as JSON: /api/errors/fb9a2a373bc2f93b. Report an issue: GitHub.