{"record":{"id":"fb9a2a373bc2f93b","repo":"windmill-labs/windmill","slug":"channel-send-error","errorCode":null,"errorMessage":"Channel send error: {}","messagePattern":"Channel send error: (.+?)","errorType":"exception","errorClass":"std::io::Error (BrokenPipe)","httpStatus":null,"severity":"warning","filePath":"backend/windmill-object-store/src/lib.rs","lineNumber":1325,"sourceCode":"    if row_count > DEFAULT_SCHEMA_INFER_MAX_RECORD as u64\n        && schema.fields().iter().any(|f| is_untyped(f.data_type()))\n    {\n        return infer(None);\n    }\n    Ok(schema)\n}\n\n#[cfg(feature = \"parquet\")]\nstruct ChannelWriter {\n    sender: tokio::sync::mpsc::Sender<anyhow::Result<Bytes>>,\n}\n\n#[cfg(feature = \"parquet\")]\nimpl Write for ChannelWriter {\n    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {\n        let data: Bytes = buf.to_vec().into();\n        self.sender.blocking_send(Ok(data)).map_err(|e| {\n            std::io::Error::new(\n                std::io::ErrorKind::BrokenPipe,\n                format!(\"Channel send error: {}\", e),\n            )\n        })?;\n        Ok(buf.len())\n    }\n\n    fn flush(&mut self) -> std::io::Result<()> {\n        Ok(())\n    }\n}\n\n#[cfg(not(feature = \"parquet\"))]\n#[derive(Debug, Clone, Copy, Default)]\npub struct IngestStats {\n    pub rows: u64,\n    pub bytes: u64,\n    pub elapsed: std::time::Duration,","sourceCodeStart":1307,"sourceCodeEnd":1343,"githubUrl":"https://github.com/windmill-labs/windmill/blob/e474e8803ce2ff5c2df09a58dab51d45f5c922ca/backend/windmill-object-store/src/lib.rs#L1307-L1343","documentation":"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: {}'.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Treat as a cancelled stream: stop producing and drop the writer; retry the export if the data is still needed.","Fix the consumer-side error that dropped the receiver first (check its logs/panic).","Increase consumer timeouts or stream in pages so slow clients don't hit cancellation.","If the receiver is dropped intentionally on early exit, ensure the writer loop checks for that condition instead of continuing to write."],"exampleFix":"// before\nfor row in rows {\n    writer.write(&serialize(row))?; // keeps failing once receiver is gone\n}\n// after\nfor row in rows {\n    if sender.is_closed() { break; } // or propagate a clean cancellation\n    writer.write(&serialize(row))?;\n}","handlingStrategy":"try-catch","validationCode":"// check the consumer is still alive before writing\nif channel_writer.sender_is_closed() { /* stop producing, release resources */ }","typeGuard":null,"tryCatchPattern":"match writer.write(&chunk) {\n    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {\n        tracing::info!(\"consumer gone, aborting parquet stream: {e}\");\n        break; // treat as cancellation, not a data error\n    }\n    Ok(n) => { /* continue */ }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["Handle client disconnects on the consumer side so receivers are dropped deliberately","Add consumer-side timeouts generous enough for the expected export size","Check sender closed-status in long write loops and exit early","Log consumer-side panics/errors — the BrokenPipe is usually a downstream symptom"],"tags":["parquet","channel","broken-pipe","streaming"],"backgroundTag":"broken-pipe-stream-cancelled","analyzedSha":"e474e8803ce2ff5c2df09a58dab51d45f5c922ca","analyzedAt":"2026-09-03T12:38:19.024Z","contentChangedAt":"2026-09-03T12:38:19.024Z","schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}