t8y2/dbx · warning · std::io::Error

Excel import consumer closed

Error message

Excel import consumer closed

What it means

This is not a real io::Error from Excel parsing: when the async consumer of the streaming XLSX import channel has dropped (hung up), blocking_send on the progress channel fails, and the code converts that into a std::io::ErrorKind::BrokenPipe with the message "Excel import consumer closed" before propagating it with `?`. The library uses it as an internal signal that nobody is listening on the import stream anymore, so the reader thread should stop.

Solutions

1. Treat BrokenPipe with this message as a normal cancellation/early-close signal, not a data error — stop the import and clean up.
2. Keep the progress-channel receiver alive until the reader task has finished, or join/await the reader before dropping the consumer.
3. If closing early is intentional, filter/handle io::ErrorKind::BrokenPipe from the import call and report 'cancelled' instead of 'failed'.
4. Check for a cancel flag (is_cancelled) being honored promptly so the reader exits itself before the consumer disappears.

Example fix

// before
let mut rx = importer.progress_receiver();
if first_error { drop(rx); } // reader then panics/errors with BrokenPipe
// after
let rx = importer.progress_receiver();
let handle = tokio::spawn(importer.run());
// drain rx until handle completes, then:
drop(rx);
let result = handle.await; // handle BrokenPipe as cancellation if it surfaces

// Rust: check the receiver is still connected before starting
assert!(!progress_rx.is_closed(), "progress consumer already dropped; aborting import early");

fn is_consumer_gone(err: &std::io::Error) -> bool {
    err.kind() == std::io::ErrorKind::BrokenPipe
        && err.to_string().contains("Excel import consumer closed")
}

match table_import::run_xlsx_import(...).await {
    Ok(res) => res,
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
        // consumer closed: treat as user cancellation, not failure
        ImportOutcome::Cancelled
    }
    Err(e) => return Err(e.into()),
}

When it happens

Trigger: Calling the streaming XLSX import API (table_import.rs, open_xlsx_shared_strings_with_control path) and dropping/closing the receiver side of the progress channel before the import finishes — e.g. the UI consumer task is cancelled, aborted, or the future holding the receiver is dropped while the reader still emits periodic Progress messages (every IMPORT_PROGRESS_INTERVAL or when shared-strings bytes are fully read).

Common situations: User cancels a long Excel import mid-stream; a web request/aborted connection drops the consumer future; a timeout cancels the receiving task while the background reader thread continues; the importer is driven with a channel that is closed early by error handling on the consumer side.

Related errors

• a batch cancellation token is always available
• a cancellation token is always available
• driver token registered
• failed to create agent runtime
• TDengine operation was cancelled

AI-assisted analysis of t8y2/dbx@c0390bff16 (2026-09-05). Data as JSON: /api/errors/18b044c2f7e29620. Report an issue: GitHub.

Appendix: source

Thrown at crates/dbx-core/src/table_import.rs:3093

        .ok_or_else(|| format!("Workbook sheet not found: {sheet_name}"))?;
    let shared_strings_bytes = match zip.by_name("xl/sharedStrings.xml") {
        Ok(file) => file.size(),
        Err(zip::result::ZipError::FileNotFound) => 0,
        Err(error) => return Err(error.to_string()),
    };
    let shared_progress_end = if shared_strings_bytes > 0 { total_bytes / 2 } else { 0 };
    let progress_sender = sender.clone();
    let mut last_shared_progress = Instant::now() - IMPORT_PROGRESS_INTERVAL;
    let mut on_shared_progress = |bytes_read: u64| {
        let progress = bytes_read
            .saturating_mul(shared_progress_end)
            .checked_div(shared_strings_bytes.max(1))
            .unwrap_or_default()
            .min(shared_progress_end);
        if last_shared_progress.elapsed() >= IMPORT_PROGRESS_INTERVAL || bytes_read >= shared_strings_bytes {
            progress_sender
                .blocking_send(Ok(XlsxStreamMessage::Progress(progress)))
                .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "Excel import consumer closed"))?;
            last_shared_progress = Instant::now();
        }
        Ok(())
    };
    let is_cancelled = || cancelled.load(Ordering::Acquire);
    let mut shared_strings = open_xlsx_shared_strings_with_control(
        &mut zip,
        MAX_IN_MEMORY_XLSX_SHARED_STRINGS_BYTES,
        &is_cancelled,
        &mut on_shared_progress,
    )?;
    let row_range = effective_import_row_range(options)?;
    let sheet = zip.by_name(&sheet_path).map_err(|error| error.to_string())?;
    let uncompressed_sheet_bytes = sheet.size().max(1);
    let mut reader = XmlReader::from_reader(BufReader::new(sheet));
    reader.config_mut().trim_text(false);
    let mut rows = XlsxStreamRowsState::new(sender, row_range, None, expected_columns, batch_size);
    let mut buffer = Vec::new();

View on GitHub (pinned to c0390bff16)