From e692ddc7f6677d5416916fc28d957c715847a2e6 Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 12:36:14 -0700 Subject: [PATCH 1/6] Skip unreadable files in batch imports --- .gitignore | 2 + CHANGELOG.md | 8 ++ Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 2 +- openspec/specs/anydoc-input/spec.md | 19 +++-- openspec/specs/file-document-import/spec.md | 13 +++ src/input.rs | 94 +++++++++++++++++---- tests/file_output.rs | 81 +++++++++++++++++- 9 files changed, 195 insertions(+), 28 deletions(-) diff --git a/.gitignore b/.gitignore index 41c9616..58589c9 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ !/.agents/skills/espipe/ !/.agents/skills/espipe/** /target +/.agents/skillator.yaml +/.agents/.gitignore diff --git a/CHANGELOG.md b/CHANGELOG.md index 4627978..efbabf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.6.1] - 2026-08-22 + +### Fixed + +- Fixed multi-file and glob imports aborting when an individual file could not be read or converted; failures now log warnings and the remaining files continue. + +## [0.6.0] - 2026-08-21 + ### Added - Added `elasticsearch:/index` and `es:/index` Elastic CLI context targets using `ELASTIC_ES_URL` and `ELASTIC_ES_API_KEY`. diff --git a/Cargo.lock b/Cargo.lock index fc6b86f..d5b3506 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -785,7 +785,7 @@ dependencies = [ [[package]] name = "espipe" -version = "0.6.0" +version = "0.6.1" dependencies = [ "anydoc", "base64 0.23.1", diff --git a/Cargo.toml b/Cargo.toml index a4bc71e..544c3d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ description = "A command-line utility to quickly and easily bulk ingest document repository = "https://github.com/VimCommando/espipe" homepage = "https://github.com/VimCommando/espipe" documentation = "https://docs.rs/crate/espipe" -version = "0.6.0" +version = "0.6.1" edition = "2024" rust-version = "1.88" license = "Apache-2.0" diff --git a/README.md b/README.md index 0e73b37..1db6ed3 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,7 @@ Local files with these extensions are converted to GitHub-Flavored Markdown thro `.doc`, `.docx`, `.docm`, `.odt`, `.pdf`, `.ppt`, `.pps`, `.pot`, `.pptx`, `.pptm`, `.ppsx`, `.ppsm`, `.rtf`, `.epub`, `.xls`, `.xlsx`, `.xlsm`, `.xlsb`, `.ods`, and `.odp`. -Converted content is stored in `content.body` by default. Use `--content markdown` to store it in `content.markdown`. Every local file-document input adds an `origin` object with `scheme: file`, a working-directory-relative `path`, and `filename`; root-level files use `./` as the path. Remote CSV, NDJSON, and Toon inputs preserve the same components from their source URI. Anydoc conversion remains local-only. Scanned or image-only PDFs require OCR outside espipe and are not converted. +Converted content is stored in `content.body` by default. Use `--content markdown` to store it in `content.markdown`. Every local file-document input adds an `origin` object with `scheme: file`, a working-directory-relative `path`, and `filename`; root-level files use `./` as the path. Remote CSV, NDJSON, and Toon inputs preserve the same components from their source URI. Anydoc conversion remains local-only. Per-file read or conversion errors in multi-file imports are logged as warnings and skipped so later files continue. Scanned or image-only PDFs require OCR outside espipe and are skipped with a warning when they occur in a multi-file import. ### Supported output forms diff --git a/openspec/specs/anydoc-input/spec.md b/openspec/specs/anydoc-input/spec.md index 35d9af3..bb2851a 100644 --- a/openspec/specs/anydoc-input/spec.md +++ b/openspec/specs/anydoc-input/spec.md @@ -78,18 +78,25 @@ The system SHALL process supported anydoc files supplied as direct local paths, - **THEN** the system combines and de-duplicates the resolved paths using existing file discovery rules - **AND** converts each supported path according to its extension -### Requirement: Anydoc conversion failures identify the source file +### Requirement: File conversion failures identify and recover per-file -The system SHALL report anydoc conversion failures through the existing file-input error path, including the source path and the underlying conversion reason when available. It SHALL not emit a synthetic document for a file that anydoc cannot convert. +The system SHALL report anydoc conversion failures with the source path and the underlying conversion reason when available. It SHALL not emit a synthetic document for a file that anydoc cannot convert. When a batch of local file-document inputs encounters a per-file read or conversion failure, the system SHALL log a warning and continue with the remaining files. A direct single-file import SHALL retain its fatal error behavior. -#### Scenario: An unsupported document is encountered +#### Scenario: An unsupported document is encountered in a batch -- **WHEN** anydoc reports that a supported-extension file is encrypted, malformed, unsupported, or exceeds a conversion limit -- **THEN** ingestion fails with a diagnostic identifying the source path -- **AND** the diagnostic is written to stderr +- **WHEN** anydoc reports that a supported-extension file is encrypted, malformed, unsupported, or exceeds a conversion limit during a multi-file import +- **THEN** ingestion logs a warning identifying the source path and reason +- **AND** the failed file is skipped without emitting a synthetic document +- **AND** remaining files continue to be imported #### Scenario: An image-only PDF is encountered - **WHEN** anydoc cannot extract meaningful text from a scanned or image-only PDF +- **THEN** a multi-file import logs a path-specific warning and skips that PDF +- **AND** the system does not claim to perform OCR + +#### Scenario: A single image-only PDF is encountered + +- **WHEN** anydoc cannot extract meaningful text from a scanned or image-only PDF supplied as the only input - **THEN** ingestion fails with a path-specific unsupported-conversion diagnostic - **AND** the system does not claim to perform OCR diff --git a/openspec/specs/file-document-import/spec.md b/openspec/specs/file-document-import/spec.md index 9c290b3..d4ef894 100644 --- a/openspec/specs/file-document-import/spec.md +++ b/openspec/specs/file-document-import/spec.md @@ -49,6 +49,19 @@ The system SHALL accept local glob input patterns, including recursive `**` patt - **THEN** the system imports the matched regular files - **AND** it does not emit documents for matched directories +### Requirement: Batch file errors do not abort later imports +When a batch of local file-document inputs encounters a per-file read or conversion error, the system SHALL log a warning identifying the source and error, skip that file, and continue importing the remaining files. A direct single-file import SHALL retain its fatal error behavior. + +#### Scenario: A file fails during a batch import +- **WHEN** one file cannot be read or converted during a multi-file or glob import +- **THEN** the system logs a warning identifying the failed source +- **AND** it does not emit a synthetic document for that file +- **AND** it continues importing later files + +#### Scenario: A direct file fails during import +- **WHEN** the only direct file input cannot be read or converted +- **THEN** ingestion fails with a diagnostic identifying the source and error + ### Requirement: File document import order is deterministic The system SHALL process file-document inputs in deterministic lexicographic path order after combining concrete file inputs and glob matches. diff --git a/src/input.rs b/src/input.rs index 741d17a..6db4e63 100644 --- a/src/input.rs +++ b/src/input.rs @@ -65,6 +65,7 @@ pub enum Input { path_index: usize, split: SplitPath, generate_id: bool, + skip_errors: bool, active: Option, }, Stdin { @@ -79,6 +80,7 @@ pub enum Input { document_index: usize, content_field: String, generate_id: bool, + skip_errors: bool, bundle_id: String, }, } @@ -425,6 +427,7 @@ fn read_local_split_line(input: &mut Input) -> Result { path_index, split, generate_id, + skip_errors, active, } = input else { @@ -442,15 +445,40 @@ fn read_local_split_line(input: &mut Input) -> Result { .ok_or_else(|| eyre!("Local split origin cursor is invalid"))?; *path_index += 1; let source = path.display().to_string(); - let file = File::open(path)?; - let receiver = - start_split_reader(local_file_reader(file, path), source.clone(), split.clone())?; + let file = match File::open(path) { + Ok(file) => file, + Err(error) if *skip_errors => { + log::warn!("skipping file {}: {error}", path.display()); + continue; + } + Err(error) => return Err(error.into()), + }; + let receiver = match start_split_reader( + local_file_reader(file, path), + source.clone(), + split.clone(), + ) { + Ok(receiver) => receiver, + Err(error) if *skip_errors => { + log::warn!("skipping file {}: {error}", path.display()); + continue; + } + Err(error) => return Err(error), + }; + let file_identity = match FileInputIdentity::new(path, *generate_id) { + Ok(file_identity) => file_identity, + Err(error) if *skip_errors => { + log::warn!("skipping file {}: {error}", path.display()); + continue; + } + Err(error) => return Err(error), + }; *active = Some(ActiveLocalSplit { source, receiver, pending_documents: VecDeque::new(), origin, - file_identity: FileInputIdentity::new(path, *generate_id)?, + file_identity, }); } @@ -485,6 +513,10 @@ fn read_local_split_line(input: &mut Input) -> Result { .map(|state| state.source.clone()) .unwrap_or_else(|| "local split".to_string()); *active = None; + if *skip_errors { + log::warn!("skipping file {source}: {error}"); + continue; + } return Err(eyre!("{source}: {error}")); } Ok(SplitEvent::Complete) => { @@ -496,6 +528,10 @@ fn read_local_split_line(input: &mut Input) -> Result { .map(|state| state.source.clone()) .unwrap_or_else(|| "local split".to_string()); *active = None; + if *skip_errors { + log::warn!("skipping file {source}: JSON split parser stopped unexpectedly"); + continue; + } return Err(eyre!("{source}: JSON split parser stopped unexpectedly")); } } @@ -582,8 +618,12 @@ fn open_input_values_with_generate_id_and_options( let (paths, origins) = resolve_file_document_paths_with_options(uris.clone(), discovery_options)?; let generate_id = effective_generate_id(generate_id, paths.len()); + let skip_errors = uris.len() > 1 + || uris + .iter() + .any(|uri| has_glob_metachar(uri.path().as_str())); - if paths.len() == 1 && uris.len() == 1 { + if paths.len() == 1 && uris.len() == 1 && !skip_errors { let uri = uris.into_iter().next().unwrap(); let path_str = uri.path().as_str(); let path = paths.into_iter().next().unwrap(); @@ -601,10 +641,16 @@ fn open_input_values_with_generate_id_and_options( if is_unsupported_compressed_input(path_str) { return Err(eyre!("Unsupported compressed input format: {path_str}")); } - return open_file_documents_from_paths(vec![path], origins, content_field, generate_id); + return open_file_documents_from_paths( + vec![path], + origins, + content_field, + generate_id, + skip_errors, + ); } - open_file_documents_from_paths(paths, origins, content_field, generate_id) + open_file_documents_from_paths(paths, origins, content_field, generate_id, skip_errors) } fn effective_generate_id(mode: Option, source_count: usize) -> bool { @@ -644,6 +690,10 @@ fn open_split_inputs_with_options( ); } + let skip_errors = uris.len() > 1 + || uris + .iter() + .any(|uri| has_glob_metachar(uri.path().as_str())); let (paths, origins) = resolve_file_document_paths_with_options(uris, discovery_options)?; for path in &paths { if !matches!(local_input_kind(path)?, InputKind::Json) { @@ -651,7 +701,7 @@ fn open_split_inputs_with_options( } } let effective_generate_id = effective_generate_id(generate_id, paths.len()); - if paths.len() == 1 { + if paths.len() == 1 && !skip_errors { let path = paths.into_iter().next().unwrap(); let origin = origins.into_iter().next(); return open_local_split_input(path, origin, split, effective_generate_id); @@ -663,6 +713,7 @@ fn open_split_inputs_with_options( path_index: 0, split, generate_id: effective_generate_id, + skip_errors, active: None, }) } @@ -846,7 +897,7 @@ fn open_file_documents( generate_id: bool, ) -> Result { let (paths, origins) = resolve_file_document_paths(values)?; - open_file_documents_from_paths(paths, origins, content_field, generate_id) + open_file_documents_from_paths(paths, origins, content_field, generate_id, false) } fn open_file_documents_from_paths( @@ -854,6 +905,7 @@ fn open_file_documents_from_paths( origins: Vec, content_field: &str, generate_id: bool, + skip_errors: bool, ) -> Result { let source = format!("{} file document(s)", paths.len()); Ok(Input::FileDocuments { @@ -865,6 +917,7 @@ fn open_file_documents_from_paths( document_index: 0, content_field: content_field.to_string(), generate_id, + skip_errors, bundle_id: if generate_id { bundle_identifier()? } else { @@ -882,6 +935,7 @@ fn read_file_document_line(input: &mut Input) -> Result { document_index, content_field, generate_id, + skip_errors, bundle_id, .. } = input @@ -918,7 +972,14 @@ fn read_file_document_line(input: &mut Input) -> Result { }; let origin = origins.get(*path_index); *path_index += 1; - *documents = read_file_documents(path, content_field, origin)?; + *documents = match read_file_documents(path, content_field, origin) { + Ok(documents) => documents, + Err(error) if *skip_errors => { + log::warn!("skipping file {}: {error}", path.display()); + Vec::new() + } + Err(error) => return Err(error), + }; *document_index = 0; } } @@ -2670,7 +2731,7 @@ mod tests { } #[test] - fn single_file_glob_uses_streaming_parser_for_structured_input() { + fn single_file_glob_uses_batch_file_import_for_structured_input() { let dir = workspace_tempdir(); let input = dir.path().join("records.ndjson"); fs::write(&input, "{\"message\":\"hello\"}\n").unwrap(); @@ -2678,7 +2739,7 @@ mod tests { let input = open_input_values(vec![uri(&pattern)], "body").unwrap(); - assert!(matches!(input, Input::FileJson { .. })); + assert!(matches!(input, Input::FileDocuments { .. })); } #[test] @@ -3191,12 +3252,11 @@ mod tests { assert_eq!(actual["content"]["body"], "alpha"); line.clear(); - let err = input.read_line(&mut line).unwrap_err(); - assert!(err.to_string().contains("not valid UTF-8")); + assert!(input.read_next(&mut line).unwrap().is_none()); } #[test] - fn json_file_document_requires_whole_object() { + fn json_file_import_rejects_non_object() { let dir = workspace_tempdir(); let path = dir.path().join("doc.json"); fs::write(&path, "{\"a\":1}").unwrap(); @@ -3209,8 +3269,8 @@ mod tests { assert_eq!(values[0]["origin"]["filename"], "doc.json"); fs::write(&path, "[1,2]").unwrap(); - let err = read_err(open_input_values(vec![uri(&path), uri(&path)], "body")); - assert!(err.contains("must contain one JSON object")); + let err = read_err(open_input_values(vec![uri(&path)], "body")); + assert_eq!(err, JSON_LINE_OPENING_ERROR); } #[test] diff --git a/tests/file_output.rs b/tests/file_output.rs index 4187ac4..ab5b7b1 100644 --- a/tests/file_output.rs +++ b/tests/file_output.rs @@ -463,6 +463,72 @@ fn cli_reports_image_only_pdf_requires_ocr_on_stderr() { assert!(stderr.contains("OCR is required")); } +#[test] +fn cli_skips_image_only_pdf_and_continues_with_later_files() { + let (_workspace, image_path) = temp_workspace_path("image-only.pdf"); + let sample_path = image_path.with_file_name("sample.pdf"); + write_base64_fixture("anydoc/image-only.pdf.base64", &image_path); + write_base64_fixture("anydoc/sample.pdf.base64", &sample_path); + let output_path = temp_output_path("skipped-image-only.ndjson"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&image_path) + .arg(&sample_path) + .arg(&output_path) + .output() + .expect("run espipe"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("image-only.pdf")); + assert!(stderr.contains("skipping file")); + + let documents = json_lines(&fs::read(&output_path).expect("read output")); + assert_eq!(documents.len(), 1); + assert!( + documents[0]["content"]["body"] + .as_str() + .is_some_and(|body| body.contains("Hello PDF")) + ); +} + +#[test] +fn cli_skips_malformed_pdf_and_continues_with_later_files() { + let (_workspace, invalid_path) = temp_workspace_path("invalid.pdf"); + let sample_path = invalid_path.with_file_name("sample.pdf"); + fs::write(&invalid_path, b"not a PDF").expect("write invalid input"); + write_base64_fixture("anydoc/sample.pdf.base64", &sample_path); + let output_path = temp_output_path("skipped-invalid-pdf.ndjson"); + + let output = Command::new(env!("CARGO_BIN_EXE_espipe")) + .arg(&invalid_path) + .arg(&sample_path) + .arg(&output_path) + .output() + .expect("run espipe"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("invalid.pdf")); + assert!(stderr.contains("skipping file")); + + let documents = json_lines(&fs::read(&output_path).expect("read output")); + assert_eq!(documents.len(), 1); + assert!( + documents[0]["content"]["body"] + .as_str() + .is_some_and(|body| body.contains("Hello PDF")) + ); +} + #[test] fn cli_rejects_multi_file_input_to_non_ndjson_file_output_before_writing() { let first_input = fixture_path("glob_docs").join("alpha.md"); @@ -518,7 +584,7 @@ fn cli_preserves_remote_input_error_for_multi_https_inputs() { } #[test] -fn cli_exits_with_error_when_later_file_document_read_fails() { +fn cli_warns_and_skips_when_later_file_document_read_fails() { let first_input = fixture_path("glob_docs").join("alpha.md"); let (_workspace, bad_input) = temp_workspace_path("bad.txt"); fs::write(&bad_input, [0xff]).expect("write invalid utf8 input"); @@ -531,12 +597,23 @@ fn cli_exits_with_error_when_later_file_document_read_fails() { .output() .expect("run espipe"); - assert!(!output.status.success(), "espipe should reject bad input"); + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); let stderr = String::from_utf8_lossy(&output.stderr); assert!( stderr.contains("not valid UTF-8"), "stderr should report read failure: {stderr}" ); + assert!(stderr.contains("skipping file")); + let documents = json_lines(&fs::read(&output_path).expect("read output")); + assert_eq!(documents.len(), 1); + assert_eq!( + documents[0]["content"]["body"], + "# Alpha\n\nFirst document.\n" + ); } #[test] From c7181285a253a5c7ffb19dc51b110dbfcf82a48e Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 12:48:50 -0700 Subject: [PATCH 2/6] Address Copilot review feedback --- openspec/specs/anydoc-input/spec.md | 2 +- src/input.rs | 119 +++++++++++++++++++++++++--- 2 files changed, 111 insertions(+), 10 deletions(-) diff --git a/openspec/specs/anydoc-input/spec.md b/openspec/specs/anydoc-input/spec.md index bb2851a..9cbdf1c 100644 --- a/openspec/specs/anydoc-input/spec.md +++ b/openspec/specs/anydoc-input/spec.md @@ -78,7 +78,7 @@ The system SHALL process supported anydoc files supplied as direct local paths, - **THEN** the system combines and de-duplicates the resolved paths using existing file discovery rules - **AND** converts each supported path according to its extension -### Requirement: File conversion failures identify and recover per-file +### Requirement: Per-file conversion failures are recoverable The system SHALL report anydoc conversion failures with the source path and the underlying conversion reason when available. It SHALL not emit a synthetic document for a file that anydoc cannot convert. When a batch of local file-document inputs encounters a per-file read or conversion failure, the system SHALL log a warning and continue with the remaining files. A direct single-file import SHALL retain its fatal error behavior. diff --git a/src/input.rs b/src/input.rs index 6db4e63..f8f5e81 100644 --- a/src/input.rs +++ b/src/input.rs @@ -68,6 +68,15 @@ pub enum Input { skip_errors: bool, active: Option, }, + LocalFileDocuments { + path: PathBuf, + origin: OriginMetadata, + content_field: String, + generate_id: bool, + skip_errors: bool, + active: Option>, + complete: bool, + }, Stdin { reader: Box>, }, @@ -376,6 +385,7 @@ impl Input { read_json_line(reader, line_buffer, false).map(InputDocument::from_raw) } Input::FileDocuments { .. } => read_file_document_line(self), + Input::LocalFileDocuments { .. } => read_local_file_document_line(self, line_buffer), Input::LocalSplitDocuments { .. } => read_local_split_line(self), } } @@ -448,7 +458,7 @@ fn read_local_split_line(input: &mut Input) -> Result { let file = match File::open(path) { Ok(file) => file, Err(error) if *skip_errors => { - log::warn!("skipping file {}: {error}", path.display()); + log_skipped_file(path, error); continue; } Err(error) => return Err(error.into()), @@ -460,7 +470,7 @@ fn read_local_split_line(input: &mut Input) -> Result { ) { Ok(receiver) => receiver, Err(error) if *skip_errors => { - log::warn!("skipping file {}: {error}", path.display()); + log_skipped_file(path, error); continue; } Err(error) => return Err(error), @@ -468,7 +478,7 @@ fn read_local_split_line(input: &mut Input) -> Result { let file_identity = match FileInputIdentity::new(path, *generate_id) { Ok(file_identity) => file_identity, Err(error) if *skip_errors => { - log::warn!("skipping file {}: {error}", path.display()); + log_skipped_file(path, error); continue; } Err(error) => return Err(error), @@ -514,7 +524,7 @@ fn read_local_split_line(input: &mut Input) -> Result { .unwrap_or_else(|| "local split".to_string()); *active = None; if *skip_errors { - log::warn!("skipping file {source}: {error}"); + log_skipped_file(Path::new(&source), error); continue; } return Err(eyre!("{source}: {error}")); @@ -529,7 +539,7 @@ fn read_local_split_line(input: &mut Input) -> Result { .unwrap_or_else(|| "local split".to_string()); *active = None; if *skip_errors { - log::warn!("skipping file {source}: JSON split parser stopped unexpectedly"); + log_skipped_file(Path::new(&source), "JSON split parser stopped unexpectedly"); continue; } return Err(eyre!("{source}: JSON split parser stopped unexpectedly")); @@ -559,6 +569,7 @@ impl std::fmt::Display for Input { Input::LocalSplitDocuments { paths, .. } => { write!(f, "{} split file(s)", paths.len()) } + Input::LocalFileDocuments { .. } => write!(f, "1 file document(s)"), Input::Stdin { .. } => write!(f, "stdin"), Input::FileDocuments { source, .. } => write!(f, "{source}"), } @@ -650,6 +661,18 @@ fn open_input_values_with_generate_id_and_options( ); } + if paths.len() == 1 && skip_errors && is_streaming_local_input(&paths[0]) { + return Ok(Input::LocalFileDocuments { + path: paths.into_iter().next().unwrap(), + origin: origins.into_iter().next().unwrap(), + content_field: content_field.to_string(), + generate_id, + skip_errors, + active: None, + complete: false, + }); + } + open_file_documents_from_paths(paths, origins, content_field, generate_id, skip_errors) } @@ -975,7 +998,7 @@ fn read_file_document_line(input: &mut Input) -> Result { *documents = match read_file_documents(path, content_field, origin) { Ok(documents) => documents, Err(error) if *skip_errors => { - log::warn!("skipping file {}: {error}", path.display()); + log_skipped_file(path, error); Vec::new() } Err(error) => return Err(error), @@ -984,6 +1007,76 @@ fn read_file_document_line(input: &mut Input) -> Result { } } +fn read_local_file_document_line( + input: &mut Input, + line_buffer: &mut String, +) -> Result { + let Input::LocalFileDocuments { + path, + origin, + content_field, + generate_id, + skip_errors, + active, + complete, + } = input + else { + return Err(eyre!("Input is not a local file document import")); + }; + + loop { + if let Some(current) = active.as_mut() { + match current.read_next(line_buffer) { + Ok(Some(document)) => return Ok(document), + Ok(None) => { + *active = None; + continue; + } + Err(error) if *skip_errors => { + *active = None; + log_skipped_file(path, error); + continue; + } + Err(error) => return Err(error), + } + } + + if *complete { + return Err(eyre!("No file document")); + } + *complete = true; + + let next = if is_streaming_local_input(path) { + open_local_file(path.clone(), *generate_id) + } else { + open_file_documents_from_paths( + vec![path.clone()], + vec![origin.clone()], + content_field, + *generate_id, + false, + ) + }; + match next { + Ok(next) => *active = Some(Box::new(next)), + Err(error) if *skip_errors => { + log_skipped_file(path, error); + } + Err(error) => return Err(error), + } + } +} + +fn log_skipped_file(path: &Path, error: impl std::fmt::Display) { + let error = error.to_string(); + let path_prefix = format!("{}: ", path.display()); + if error.starts_with(&path_prefix) { + log::warn!("skipping file {error}"); + } else { + log::warn!("skipping file {}: {error}", path.display()); + } +} + fn resolve_file_document_paths( values: Vec>, ) -> Result<(Vec, Vec)> { @@ -1169,6 +1262,14 @@ fn should_use_file_document(path: &Path) -> bool { ) } +fn is_streaming_local_input(path: &Path) -> bool { + match local_input_kind(path) { + Ok(InputKind::Csv | InputKind::Ndjson | InputKind::Toon) => true, + Ok(InputKind::Json) => !should_use_file_document(path), + Ok(InputKind::FileDocument) | Err(_) => false, + } +} + fn read_file_documents( path: &Path, content_field: &str, @@ -2731,7 +2832,7 @@ mod tests { } #[test] - fn single_file_glob_uses_batch_file_import_for_structured_input() { + fn single_file_glob_preserves_streaming_for_structured_input() { let dir = workspace_tempdir(); let input = dir.path().join("records.ndjson"); fs::write(&input, "{\"message\":\"hello\"}\n").unwrap(); @@ -2739,7 +2840,7 @@ mod tests { let input = open_input_values(vec![uri(&pattern)], "body").unwrap(); - assert!(matches!(input, Input::FileDocuments { .. })); + assert!(matches!(input, Input::LocalFileDocuments { .. })); } #[test] @@ -3237,7 +3338,7 @@ mod tests { } #[test] - fn file_document_import_reads_files_lazily() { + fn batch_file_import_skips_unreadable_files_after_reading_earlier_files() { let dir = workspace_tempdir(); let first = dir.path().join("a.txt"); let second = dir.path().join("b.txt"); From 98f6e1dfcfb933f79bec287791835b3551f54e41 Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 12:56:58 -0700 Subject: [PATCH 3/6] Clarify batch import diagnostics --- README.md | 2 +- src/input.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 1db6ed3..4c85b6b 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,7 @@ Local files with these extensions are converted to GitHub-Flavored Markdown thro `.doc`, `.docx`, `.docm`, `.odt`, `.pdf`, `.ppt`, `.pps`, `.pot`, `.pptx`, `.pptm`, `.ppsx`, `.ppsm`, `.rtf`, `.epub`, `.xls`, `.xlsx`, `.xlsm`, `.xlsb`, `.ods`, and `.odp`. -Converted content is stored in `content.body` by default. Use `--content markdown` to store it in `content.markdown`. Every local file-document input adds an `origin` object with `scheme: file`, a working-directory-relative `path`, and `filename`; root-level files use `./` as the path. Remote CSV, NDJSON, and Toon inputs preserve the same components from their source URI. Anydoc conversion remains local-only. Per-file read or conversion errors in multi-file imports are logged as warnings and skipped so later files continue. Scanned or image-only PDFs require OCR outside espipe and are skipped with a warning when they occur in a multi-file import. +Converted content is stored in `content.body` by default. Use `--content markdown` to store it in `content.markdown`. Every local file-document input adds an `origin` object with `scheme: file`, a working-directory-relative `path`, and `filename`; root-level files use `./` as the path. Remote CSV, NDJSON, and Toon inputs preserve the same components from their source URI. Anydoc conversion remains local-only. Per-file read or conversion errors in multi-file or glob imports, including globs that resolve to one file, are logged as warnings and skipped so later files continue. Scanned or image-only PDFs require OCR outside espipe and are skipped with a warning when they occur in a multi-file or glob import. ### Supported output forms diff --git a/src/input.rs b/src/input.rs index f8f5e81..7402317 100644 --- a/src/input.rs +++ b/src/input.rs @@ -569,7 +569,7 @@ impl std::fmt::Display for Input { Input::LocalSplitDocuments { paths, .. } => { write!(f, "{} split file(s)", paths.len()) } - Input::LocalFileDocuments { .. } => write!(f, "1 file document(s)"), + Input::LocalFileDocuments { path, .. } => write!(f, "{}", path.display()), Input::Stdin { .. } => write!(f, "stdin"), Input::FileDocuments { source, .. } => write!(f, "{source}"), } From e94d806c9e10eccbb497e8d5e60d40e2a6b35183 Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 20:40:23 -0700 Subject: [PATCH 4/6] Fix file completion counters --- README.md | 2 ++ src/input.rs | 69 ++++++++++++++++++++++++++++++++++++++++++-- src/main.rs | 24 +++++++++++---- tests/file_output.rs | 5 ++++ 4 files changed, 91 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 4c85b6b..9dfc5b8 100644 --- a/README.md +++ b/README.md @@ -253,6 +253,8 @@ For Elasticsearch targets, `espipe`: `400 Bad Request` bulk responses are logged and counted as zero successful documents for that batch. +For local file imports, the completion summary reports all discovered files separately from documents sent and documents evaluated/read. Skipped files contribute to the file count but do not contribute documents. For example: `From 6,246 files, piped 5,850 of 5,850 docs ...`. + ### File and stdout output For file and `stdout` targets, `espipe` writes one raw JSON document per line. It does not emit Elasticsearch bulk action metadata lines for these outputs. diff --git a/src/input.rs b/src/input.rs index 7402317..b35641c 100644 --- a/src/input.rs +++ b/src/input.rs @@ -67,6 +67,8 @@ pub enum Input { generate_id: bool, skip_errors: bool, active: Option, + evaluated_documents: usize, + file_count: usize, }, LocalFileDocuments { path: PathBuf, @@ -76,6 +78,8 @@ pub enum Input { skip_errors: bool, active: Option>, complete: bool, + evaluated_documents: usize, + file_count: usize, }, Stdin { reader: Box>, @@ -91,6 +95,8 @@ pub enum Input { generate_id: bool, skip_errors: bool, bundle_id: String, + evaluated_documents: usize, + file_count: usize, }, } @@ -397,6 +403,39 @@ impl Input { Err(err) => Err(err), } } + + pub(crate) fn evaluated_document_count(&self, successful_documents: usize) -> usize { + match self { + Input::LocalSplitDocuments { + evaluated_documents, + .. + } + | Input::LocalFileDocuments { + evaluated_documents, + .. + } + | Input::FileDocuments { + evaluated_documents, + .. + } => (*evaluated_documents).max(successful_documents), + _ => successful_documents, + } + } + + pub(crate) fn file_count(&self, successful_documents: usize) -> Option { + match self { + Input::FileJson { origin, .. } + | Input::FileCsv { origin, .. } + | Input::FileToon { origin, .. } + | Input::JsonSplit { origin, .. } => origin + .as_ref() + .map(|_| usize::from(successful_documents > 0)), + Input::LocalSplitDocuments { file_count, .. } + | Input::LocalFileDocuments { file_count, .. } + | Input::FileDocuments { file_count, .. } => Some(*file_count), + Input::Stdin { .. } => None, + } + } } fn finalize_file_input_document( @@ -439,6 +478,8 @@ fn read_local_split_line(input: &mut Input) -> Result { generate_id, skip_errors, active, + evaluated_documents, + file_count: _, } = input else { return Err(eyre!("Input is not a local split import")); @@ -497,11 +538,15 @@ fn read_local_split_line(input: &mut Input) -> Result { .and_then(|state| state.pending_documents.pop_front()) { let state = active.as_mut().expect("active split state disappeared"); - return finalize_split_document( + let result = finalize_split_document( document, Some(&state.origin), Some(&mut state.file_identity), ); + if result.is_ok() { + *evaluated_documents += 1; + } + return result; } let event = active @@ -670,6 +715,8 @@ fn open_input_values_with_generate_id_and_options( skip_errors, active: None, complete: false, + evaluated_documents: 0, + file_count: 1, }); } @@ -730,6 +777,7 @@ fn open_split_inputs_with_options( return open_local_split_input(path, origin, split, effective_generate_id); } + let file_count = paths.len(); Ok(Input::LocalSplitDocuments { paths, origins, @@ -738,6 +786,8 @@ fn open_split_inputs_with_options( generate_id: effective_generate_id, skip_errors, active: None, + evaluated_documents: 0, + file_count, }) } @@ -930,6 +980,7 @@ fn open_file_documents_from_paths( generate_id: bool, skip_errors: bool, ) -> Result { + let file_count = paths.len(); let source = format!("{} file document(s)", paths.len()); Ok(Input::FileDocuments { source, @@ -946,6 +997,8 @@ fn open_file_documents_from_paths( } else { String::new() }, + evaluated_documents: 0, + file_count, }) } @@ -960,6 +1013,8 @@ fn read_file_document_line(input: &mut Input) -> Result { generate_id, skip_errors, bundle_id, + evaluated_documents, + file_count: _, .. } = input else { @@ -996,7 +1051,10 @@ fn read_file_document_line(input: &mut Input) -> Result { let origin = origins.get(*path_index); *path_index += 1; *documents = match read_file_documents(path, content_field, origin) { - Ok(documents) => documents, + Ok(documents) => { + *evaluated_documents += documents.len(); + documents + } Err(error) if *skip_errors => { log_skipped_file(path, error); Vec::new() @@ -1019,6 +1077,8 @@ fn read_local_file_document_line( skip_errors, active, complete, + evaluated_documents, + file_count: _, } = input else { return Err(eyre!("Input is not a local file document import")); @@ -1027,7 +1087,10 @@ fn read_local_file_document_line( loop { if let Some(current) = active.as_mut() { match current.read_next(line_buffer) { - Ok(Some(document)) => return Ok(document), + Ok(Some(document)) => { + *evaluated_documents += 1; + return Ok(document); + } Ok(None) => { *active = None; continue; diff --git a/src/main.rs b/src/main.rs index 3323520..dc2ebd2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -298,12 +298,24 @@ async fn main() -> ExitCode { Err(err) => return exit_with_error(err), }; if !quiet { - println!( - "Piped {} of {} docs to {output_name} in {:.3} seconds", - comma_formatted(output_line), - comma_formatted(input_line), - start_time.elapsed().as_secs_f32() - ); + let evaluated_line = input.evaluated_document_count(input_line); + if let Some(file_count) = input.file_count(input_line) { + let file_label = if file_count == 1 { "file" } else { "files" }; + println!( + "From {} {file_label}, piped {} of {} docs to {output_name} in {:.3} seconds", + comma_formatted(file_count), + comma_formatted(output_line), + comma_formatted(evaluated_line), + start_time.elapsed().as_secs_f32() + ); + } else { + println!( + "Piped {} of {} docs to {output_name} in {:.3} seconds", + comma_formatted(output_line), + comma_formatted(evaluated_line), + start_time.elapsed().as_secs_f32() + ); + } } ExitCode::SUCCESS } diff --git a/tests/file_output.rs b/tests/file_output.rs index ab5b7b1..e248382 100644 --- a/tests/file_output.rs +++ b/tests/file_output.rs @@ -608,6 +608,11 @@ fn cli_warns_and_skips_when_later_file_document_read_fails() { "stderr should report read failure: {stderr}" ); assert!(stderr.contains("skipping file")); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("From 2 files, piped 1 of 1 docs to"), + "summary should count the skipped file as evaluated: {stdout}" + ); let documents = json_lines(&fs::read(&output_path).expect("read output")); assert_eq!(documents.len(), 1); assert_eq!( From d8da8006da203af28eb2a5b02b83e14a458afb44 Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 22:12:22 -0700 Subject: [PATCH 5/6] Parallelize multi-source file conversion --- CHANGELOG.md | 8 +- README.md | 28 +- .../.openspec.yaml | 2 + .../design.md | 70 +++ .../proposal.md | 28 + .../specs/anydoc-input/spec.md | 54 ++ .../specs/file-document-import/spec.md | 59 ++ .../specs/rawvalue-document-pipeline/spec.md | 35 ++ .../tasks.md | 30 + openspec/specs/anydoc-input/spec.md | 31 +- openspec/specs/file-document-import/spec.md | 28 +- .../specs/rawvalue-document-pipeline/spec.md | 17 +- src/input.rs | 523 +++++++++++++++--- src/main.rs | 120 ++-- src/output/elasticsearch.rs | 2 + src/output/mod.rs | 10 + tests/file_output.rs | 2 +- 17 files changed, 896 insertions(+), 151 deletions(-) create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/.openspec.yaml create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/design.md create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/proposal.md create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/anydoc-input/spec.md create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/file-document-import/spec.md create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/rawvalue-document-pipeline/spec.md create mode 100644 openspec/changes/archive/2026-08-22-parallel-file-conversion/tasks.md diff --git a/CHANGELOG.md b/CHANGELOG.md index efbabf4..116e0a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,9 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [0.6.1] - 2026-08-22 +### Changed + +- Multi-source local imports now convert files in a bounded worker pool and emit documents as each file finishes. Generated IDs remain deterministic. +- Multi-source local imports now use 500-document Elasticsearch bulk requests by default. Single-file streams and other inputs keep the 5,000-document default. +- Local import summaries now report piped and evaluated document counts before the discovered file count. + ### Fixed -- Fixed multi-file and glob imports aborting when an individual file could not be read or converted; failures now log warnings and the remaining files continue. +- Multi-file and glob imports no longer abort when one file cannot be read or converted. They log a warning and continue with the remaining files. ## [0.6.0] - 2026-08-21 diff --git a/README.md b/README.md index 9dfc5b8..098ef21 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ espipe docs.ndjson my-cluster:/new_index Being multi-threaded and unthrottled, `espipe` is capable of fully saturating the CPU of the sending host and can potentially overwhelm the target cluster, so use with caution. It will gracefully handle backpressure and `http 429` responses to ensure at-least-once delivery. -Documents are batched into `_bulk` requests of 5,000 documents and sent with the `index` action by default. Use `--action` to select `create`, `index`, `update`, or `upsert`. Multi-source file-document inputs receive deterministic IDs based on their bundle and working-directory-relative path by default; single-source inputs require `--generate-id=true` to generate IDs. Use `--generate-id=false` to let Elasticsearch assign IDs for `create` and `index`, or to require explicit IDs for `update` and `upsert`. Use `--batch-size` and `--max-requests` to tune bulk request size and concurrency at runtime. +Documents are batched into `_bulk` requests of 500 documents for multi-source local imports and 5,000 documents for single-file streaming and other inputs. They use the `index` action by default. Use `--action` to select `create`, `index`, `update`, or `upsert`. Multi-source file-document inputs receive deterministic IDs based on their bundle and working-directory-relative path by default; single-source inputs require `--generate-id=true` to generate IDs. Use `--generate-id=false` to let Elasticsearch assign IDs for `create` and `index`, or to require explicit IDs for `update` and `upsert`. Use `--batch-size` and `--max-requests` to tune bulk request size and concurrency at runtime. ## Installation @@ -77,7 +77,7 @@ It writes records to: - a local `.ndjson` or `.ndjson.gz` file - `stdout` -When writing to Elasticsearch, `espipe` batches documents into groups of 5,000 records by default, enables request body gzip compression by default, and sends multiple bulk requests concurrently. Use `--batch-size` to change the number of documents per bulk request and `--max-requests` to change the number of in-flight bulk requests. File gzip compression is selected only for supported `.csv.gz`, `.ndjson.gz`, and output `.ndjson.gz` suffixes, and is separate from Elasticsearch request body compression. +When writing to Elasticsearch, `espipe` uses 500-record batches for multi-source local imports and 5,000-record batches otherwise. It enables request body gzip compression by default and sends multiple bulk requests concurrently. Use `--batch-size` to override the source-aware default and `--max-requests` to change the number of in-flight bulk requests. File gzip compression is selected only for supported `.csv.gz`, `.ndjson.gz`, and output `.ndjson.gz` suffixes, and is separate from Elasticsearch request body compression. ## CLI Reference @@ -100,7 +100,7 @@ Options: --generate-id Generate deterministic IDs for local files (default: multi-source only) --symlinks Multi-source symlink policy [default: skip] [possible values: follow, fail, skip] --hidden Multi-source hidden-path policy [default: skip] [possible values: include, fail, skip] - --batch-size Documents per Elasticsearch bulk request [default: 5000] + --batch-size Documents per Elasticsearch bulk request (default: 500 for multi-source local input, 5000 otherwise) --max-requests Maximum concurrent Elasticsearch bulk requests [default: 16] -h, --help Print help ``` @@ -141,7 +141,7 @@ are followed by the output URI. - `'docs/**/*.pdf'` Recursively finds local PDFs and converts each one to a file document. - `path/to/file.pdf path/to/file.xlsx output.ndjson` - Imports multiple local file inputs in deterministic path order. + Imports multiple local file inputs and emits each source as its conversion finishes. HTTP and HTTPS input URIs are supported for unauthenticated remote `.csv`, `.ndjson`, and `.json` sources. URLs without a supported file extension can still be accepted when the response `Content-Type` maps to CSV or NDJSON-oriented JSON input. @@ -153,6 +153,8 @@ Local files with these extensions are converted to GitHub-Flavored Markdown thro Converted content is stored in `content.body` by default. Use `--content markdown` to store it in `content.markdown`. Every local file-document input adds an `origin` object with `scheme: file`, a working-directory-relative `path`, and `filename`; root-level files use `./` as the path. Remote CSV, NDJSON, and Toon inputs preserve the same components from their source URI. Anydoc conversion remains local-only. Per-file read or conversion errors in multi-file or glob imports, including globs that resolve to one file, are logged as warnings and skipped so later files continue. Scanned or image-only PDFs require OCR outside espipe and are skipped with a warning when they occur in a multi-file or glob import. +Multi-source local file documents are read and converted by a bounded worker pool with up to eight workers. Each source is emitted when its conversion finishes, so output order is unspecified. Generated IDs remain stable because they use source paths rather than output positions. + ### Supported output forms - `-` @@ -233,7 +235,7 @@ For all Elasticsearch actions, a top-level string `_id` is used as the transport For Elasticsearch targets: - `--batch-size` - Sets the number of documents included in each `_bulk` request. + Sets the number of documents included in each `_bulk` request. Without this option, multi-source local input uses 500 and other input modes use 5,000. - `--max-requests` Sets the maximum number of concurrent in-flight bulk requests. @@ -245,7 +247,8 @@ The internal channel capacity always matches `--batch-size`. For Elasticsearch targets, `espipe`: -- batches documents into 5,000-document `_bulk` requests by default +- batches multi-source local documents into 500-document `_bulk` requests by default +- retains 5,000-document `_bulk` requests for single-file streaming and other input modes - keeps up to 16 bulk requests in flight by default - enables gzip request body compression by default - retries `429 Too Many Requests` responses with exponential backoff @@ -253,7 +256,7 @@ For Elasticsearch targets, `espipe`: `400 Bad Request` bulk responses are logged and counted as zero successful documents for that batch. -For local file imports, the completion summary reports all discovered files separately from documents sent and documents evaluated/read. Skipped files contribute to the file count but do not contribute documents. For example: `From 6,246 files, piped 5,850 of 5,850 docs ...`. +For local file imports, the completion summary reports all discovered files separately from documents sent and documents evaluated/read. Skipped files contribute to the file count but do not contribute documents. For example: `Piped 5,850 of 5,850 docs from 6,246 files ...`. ### File and stdout output @@ -444,14 +447,15 @@ One current limitation is that input parsing errors and end-of-input are handled `espipe` is intentionally aggressive enough to saturate a local or small remote cluster. -Current bulk worker settings: +Current worker settings: -- batch size: 5,000 documents -- channel capacity: 5,000 documents +- multi-source local bulk batch size: 500 documents +- other bulk batch size: 5,000 documents +- channel capacity: the effective bulk batch size - max in-flight bulk requests: 16 -- Tokio worker threads: 3 +- multi-source file conversion workers: up to 8, bounded by available parallelism and source count -This is fast for local ingestion and test data loading, but it can overwhelm smaller clusters or shared environments. +File conversion results are emitted in completion order, which avoids waiting for slower earlier paths. Explicit `--batch-size` values override the source-aware defaults. These settings can overwhelm smaller clusters or shared environments. ## Troubleshooting diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/.openspec.yaml b/openspec/changes/archive/2026-08-22-parallel-file-conversion/.openspec.yaml new file mode 100644 index 0000000..6529e83 --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-22 diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/design.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/design.md new file mode 100644 index 0000000..fc16ebe --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/design.md @@ -0,0 +1,70 @@ +## Context + +Multi-source local files currently share one synchronous `FileDocuments` iterator. Each call converts one complete source before the next path starts, and the async output handoff occurs only after conversion. Stable generated IDs depend on source paths and per-source document indexes rather than cross-source output order. + +The CLI currently resolves `--batch-size` to 5,000 during argument parsing. Source cardinality becomes authoritative only after local path discovery and filtering. + +## Goals / Non-Goals + +**Goals:** + +- Keep several CPU-bound file conversions active on machines with available cores. +- Bound active work and completed results. +- Emit completed sources without waiting for slower earlier paths while preserving existing file-level diagnostics. +- Select the implicit Elasticsearch batch size after input discovery. + +**Non-Goals:** + +- Parallelize records within one streaming file. +- Add a user-facing conversion-worker option in this change. +- Add OCR or change anydoc extraction behavior. +- Guarantee linear speedup across storage devices or document formats. + +## Decisions + +### Use a dedicated standard-library worker pool + +Multi-source `FileDocuments` input will own a fixed set of worker threads. Each worker receives one source job at a time and returns the path and conversion result. Blocking conversion will not occupy a Tokio runtime worker. + +The pool size will be the smaller of the source count, available parallelism, and eight workers. Eight allows the reported 500 to 600 percent CPU target while limiting simultaneous whole-file reads and nested parser work. A dedicated pool also avoids Tokio's much larger general blocking-thread limit. Adding Rayon directly was considered, but it would add a dependency without improving the existing channel-based output handoff. + +### Emit results in completion order + +The coordinator initially gives one source to each worker. When a result arrives, it immediately schedules the next source on that worker and returns the completed result to the output pipeline. The result channel is bounded by the worker count, so active and completed work remains bounded without an ordered result map. + +This removes `BTreeMap` operations and retained out-of-order documents. More importantly, a slow early PDF cannot stop other workers after a fixed look-ahead window. File and stdout consumers that need order can sort by `origin.path` and `origin.filename`. + +### Keep identity and error decisions on the consumer + +Workers only read and convert a source into raw documents. The consumer logs failed sources, updates evaluated-document counters, and derives generated IDs from the source path and per-file document index. Completion order therefore cannot affect IDs, warning policy, or summary counts. + +### Resolve the implicit bulk size from constructed input + +`--batch-size` will become optional in the parsed CLI model. After local discovery constructs `Input`, the program will select 500 when the input reports more than one local source and 5,000 otherwise. An explicit value bypasses this selection. + +Constructing local input before Elasticsearch output is safe because path discovery does not convert or emit documents. Remote and single-file streams retain the 5,000 default. The Elasticsearch output configuration remains explicit after selection, so its channel capacity continues to equal the effective batch size. + +## Risks / Trade-offs + +- [Nested parser parallelism can oversubscribe cores] -> Cap outer conversion workers at eight and benchmark the representative PDF collection. +- [Concurrent whole-file conversion increases peak memory and disk traffic] -> Bound active jobs and completed results by the worker count. +- [File and stdout output order changes across runs] -> Preserve source identity in `origin` and document IDs so consumers can sort when needed. +- [Input construction now precedes Elasticsearch preflight for local sources] -> Keep construction limited to validation and discovery; no conversion or output starts before preflight succeeds. + +## Migration Plan + +The behavior changes automatically for multi-source local imports. Users who need the old request size can pass `--batch-size 5000`. Rollback consists of restoring serial `FileDocuments` iteration and the static 5,000 default; document formats and generated IDs remain compatible. + +## Validation + +The release build was measured on 2026-08-22 against the 6,246-file NASA STI abstracts collection: + +```bash +cd /Users/reno/Development/elastic-notes/samples +LOG_LEVEL=error /usr/bin/time -p /Users/reno/Development/espipe/target/release/espipe \ + 'nasa-sti-abstracts/**/*.pdf' /tmp/espipe-nasa-parallel.ndjson +``` + +The ordered worker-pool candidate emitted 5,850 of 5,850 eligible documents in 8.768 seconds. Process timing reported 9.43 seconds real, 24.46 seconds user, and 6.31 seconds system. After removing ordered result buffering, the same command emitted the same document count in 5.185 seconds, with 5.68 seconds real, 24.37 seconds user, and 6.49 seconds system. Completion-order emission reduced application elapsed time by 40.9% while process CPU time stayed nearly flat, confirming that ordered delivery caused head-of-line waiting rather than useful work. + +The document counts match the user's 23.388-second localhost Elasticsearch run. The outputs differ, so these measurements isolate conversion and local serialization rather than claiming a strict end-to-end Elasticsearch speedup. Repeating the exact Elasticsearch command would mutate the existing `localhost:/nasa-sti-abstracts` index and was left for an explicitly authorized run. diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/proposal.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/proposal.md new file mode 100644 index 0000000..ba23942 --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/proposal.md @@ -0,0 +1,28 @@ +## Why + +Multi-source document imports convert files serially, so large PDF collections leave most CPU cores idle and delay the first Elasticsearch bulk request. The default 5,000-document bulk size compounds the delay for one-document-per-file imports. + +## What Changes + +- Convert multi-source file documents with a bounded worker pool sized for the local machine. +- Emit multi-source file documents as conversion workers finish while preserving generated IDs, bounded memory use, and per-file warn-and-skip behavior. +- Default Elasticsearch bulk requests to 500 documents for multi-source local input and retain 5,000 for single-file streaming input. +- Keep an explicit `--batch-size` value authoritative in every input mode. +- Change file-import completion summaries to report documents before source-file counts. +- Add regression tests for concurrent conversion, completion-order results, skipped files, and source-aware bulk defaults. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `anydoc-input`: Multi-source local document conversion becomes bounded and concurrent, with results emitted in completion order without changing identity or error recovery. +- `file-document-import`: Multi-source file output order becomes unspecified while deterministic discovery and de-duplication remain intact. +- `rawvalue-document-pipeline`: The default Elasticsearch bulk batch size depends on whether input is a multi-source local import or a single-file stream. + +## Impact + +The change affects local input construction and iteration in `src/input.rs`, completion summaries and Elasticsearch configuration selection in `src/main.rs`, related integration and unit tests, CLI documentation, and performance notes. It does not add a dependency or change explicit CLI option behavior. diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/anydoc-input/spec.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/anydoc-input/spec.md new file mode 100644 index 0000000..50c9ff4 --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/anydoc-input/spec.md @@ -0,0 +1,54 @@ +## MODIFIED Requirements + +### Requirement: Supported local non-text files are converted through anydoc + +The system SHALL use the Rust `anydoc` processor for local regular files with these extensions: `.doc`, `.docx`, `.docm`, `.odt`, `.pdf`, `.ppt`, `.pps`, `.pot`, `.pptx`, `.pptm`, `.ppsx`, `.ppsm`, `.rtf`, `.epub`, `.xls`, `.xlsx`, `.xlsm`, `.xlsb`, `.ods`, and `.odp`. The processor SHALL convert each file to GitHub-Flavored Markdown before file-document construction. + +#### Scenario: A PDF file is imported + +- **WHEN** the user runs `espipe` with a local text-based `.pdf` input +- **THEN** the system converts the PDF to Markdown through anydoc +- **AND** emits one JSON document containing the converted Markdown in the configured content field + +#### Scenario: An office document is imported + +- **WHEN** the user runs `espipe` with a local supported Word, PowerPoint, Excel, or OpenDocument input +- **THEN** the system converts the file to Markdown through anydoc +- **AND** emits one JSON document for the source file + +#### Scenario: A supported file is imported from a mixed local collection + +- **WHEN** file-document input resolves supported anydoc files together with Markdown or text files +- **THEN** each anydoc file is converted by the bounded worker pool +- **AND** documents may be emitted in conversion completion order +- **AND** existing Markdown and text files continue through their existing readers + +## ADDED Requirements + +### Requirement: Multi-source document conversion uses bounded concurrency + +The system SHALL convert files from a multi-source local document import concurrently. It SHALL bound the number of active conversions and completed conversion results retained in memory. Concurrent conversion SHALL emit source results as workers complete while preserving generated document identity and per-file error recovery behavior. + +#### Scenario: Multiple PDFs are converted concurrently + +- **WHEN** a multi-source local import contains more convertible PDFs than the conversion worker limit +- **THEN** the system permits multiple PDF conversions to execute at the same time +- **AND** it does not start an unbounded number of conversion operations + +#### Scenario: Concurrent conversions finish in a different order from discovery + +- **WHEN** a later file finishes conversion before an earlier file +- **THEN** the system emits the completed result without waiting for the earlier file +- **AND** output order is not guaranteed to match source-path order + +#### Scenario: A concurrent conversion fails + +- **WHEN** one file fails to read or convert while other file conversions are active +- **THEN** the system logs the same path-specific warning used by batch file recovery +- **AND** emits no document for the failed file +- **AND** continues emitting successful documents as conversions finish + +#### Scenario: Generated IDs are produced by concurrent conversion + +- **WHEN** multi-source conversion completes files in a different order across two runs +- **THEN** each emitted document receives the same generated ID in both runs diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/file-document-import/spec.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/file-document-import/spec.md new file mode 100644 index 0000000..ef6ca2a --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/file-document-import/spec.md @@ -0,0 +1,59 @@ +## ADDED Requirements + +### Requirement: Multi-source file output order is unspecified + +The system SHALL combine and de-duplicate concrete file inputs and glob matches deterministically before scheduling conversion. It SHALL emit each source result as conversion completes and SHALL NOT delay completed sources solely to restore lexicographic output order. + +#### Scenario: Later file completes first + +- **WHEN** a later source path finishes conversion before an earlier source path +- **THEN** the later source document may reach the output first + +#### Scenario: Same file appears more than once + +- **WHEN** a file is provided directly and also matched by a glob pattern +- **THEN** the system processes the resolved source once +- **AND** emits each document from that source at most once + +### Requirement: Local import summaries distinguish documents from files + +The system SHALL report piped documents, evaluated documents, and discovered source files as `Piped X of Y docs from Z files ...`. A skipped source SHALL count toward the discovered file count but SHALL NOT add a document to the evaluated count. + +#### Scenario: Some discovered files are skipped + +- **WHEN** a local import discovers 10 source files +- **AND** 3 files are skipped without producing documents +- **AND** the remaining files produce 7 documents that are sent successfully +- **THEN** the completion summary begins `Piped 7 of 7 docs from 10 files` + +## MODIFIED Requirements + +### Requirement: Toon files stream one document per Toon document + +The system SHALL import `.toon` files as structured Toon input where each decoded Toon object emits one JSON object document. Documents within one Toon source SHALL preserve their source order. A Toon source within a multi-source file import MAY be emitted before or after other sources according to file conversion completion order. + +#### Scenario: Toon file is imported + +- **WHEN** a file-document input resolves to a `.toon` file +- **THEN** the system parses the file using the Toon input reader +- **AND** each decoded Toon object emits one document + +#### Scenario: Toon file is included with multiple file inputs + +- **WHEN** file-document input contains a `.toon` file and other supported files +- **THEN** documents within the Toon file retain their source order +- **AND** the Toon file may be emitted before or after other sources according to conversion completion order + +#### Scenario: Toon file contains a non-object document + +- **WHEN** a `.toon` file contains a document that decodes to a non-object value +- **THEN** importing that file fails +- **AND** the invalid Toon document is not sent to any output + +## REMOVED Requirements + +### Requirement: File document import order is deterministic + +**Reason**: Restoring lexicographic output order adds head-of-line blocking and retained conversion results without affecting Elasticsearch identity or correctness. + +**Migration**: Consumers that require ordering must sort emitted documents by `origin.path` and `origin.filename`. diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/rawvalue-document-pipeline/spec.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/rawvalue-document-pipeline/spec.md new file mode 100644 index 0000000..f95e34c --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/specs/rawvalue-document-pipeline/spec.md @@ -0,0 +1,35 @@ +## MODIFIED Requirements + +### Requirement: Elasticsearch bulk output preserves raw document buffering + +The system SHALL buffer raw documents for Elasticsearch output and emit valid `_bulk` request bodies without requiring `Value` in the steady-state queue. Unless the user supplies `--batch-size`, the system SHALL use a default batch size of 500 documents for multi-source local input and 5,000 documents for single-file streaming input and other input modes. + +#### Scenario: Multi-source local input uses smaller default batches + +- **WHEN** local input resolves to more than one source file +- **AND** the user does not pass `--batch-size` +- **THEN** Elasticsearch bulk output targets 500 documents per request + +#### Scenario: Single-file streaming input retains large default batches + +- **WHEN** the user imports one streaming NDJSON, CSV, JSON, or Toon source +- **AND** the user does not pass `--batch-size` +- **THEN** Elasticsearch bulk output targets 5,000 documents per request + +#### Scenario: Explicit batch size overrides the source-aware default + +- **WHEN** the user passes `--batch-size 750` +- **THEN** Elasticsearch bulk output targets 750 documents per request regardless of input source count + +#### Scenario: Bulk queue flushes to Elasticsearch + +- **WHEN** the Elasticsearch output flushes one or more buffered documents +- **THEN** it constructs a valid `_bulk` request body for `create` operations using the buffered raw JSON documents +- **AND** Elasticsearch accepts the request without document-shape regressions + +#### Scenario: Buffered documents are large + +- **WHEN** the queue contains many large documents +- **THEN** the queue retains raw JSON payloads instead of cloned `Value` trees +- **AND** the implementation avoids additional whole-document copies beyond what is required to build the outbound request + diff --git a/openspec/changes/archive/2026-08-22-parallel-file-conversion/tasks.md b/openspec/changes/archive/2026-08-22-parallel-file-conversion/tasks.md new file mode 100644 index 0000000..657df95 --- /dev/null +++ b/openspec/changes/archive/2026-08-22-parallel-file-conversion/tasks.md @@ -0,0 +1,30 @@ +## 1. Parallel file conversion + +- [x] 1.1 Add a bounded multi-source file conversion worker pool with machine-aware worker sizing and bounded scheduling. +- [x] 1.2 Integrate worker results with generated IDs, evaluated-document counts, and warn-and-skip error handling. +- [x] 1.3 Add tests proving conversions overlap while generated IDs remain deterministic. + +## 2. Source-aware bulk defaults + +- [x] 2.1 Defer implicit batch-size selection until input source cardinality is known. +- [x] 2.2 Default multi-source local input to 500 documents and retain 5,000 for single-file streaming and other inputs. +- [x] 2.3 Add CLI and configuration tests for implicit and explicit batch sizes. + +## 3. Documentation and verification + +- [x] 3.1 Update README and changelog text for parallel conversion and source-aware bulk defaults. +- [x] 3.2 Run formatting, static checks, targeted tests, and the full test suite. +- [x] 3.3 Benchmark the representative NASA PDF collection or record any environment blocker and the reproducible command. + +## 4. Completion-order output + +- [x] 4.1 Remove ordered result buffering and emit multi-source documents as conversions finish. +- [x] 4.2 Change the file-import summary to `Piped X of Y docs from Z files ...`. +- [x] 4.3 Update ordering tests, documentation, and changelog wording. +- [x] 4.4 Run validation and repeat the NASA benchmark to measure ordering overhead. + +## 5. Verification fixes + +- [x] 5.1 Correct duplicate-source semantics and specify the local import completion summary. +- [x] 5.2 Add a worker-level test for a failed conversion while another conversion is active. +- [x] 5.3 Run the full test suite and strict OpenSpec validation. diff --git a/openspec/specs/anydoc-input/spec.md b/openspec/specs/anydoc-input/spec.md index 9cbdf1c..f9d5e32 100644 --- a/openspec/specs/anydoc-input/spec.md +++ b/openspec/specs/anydoc-input/spec.md @@ -23,9 +23,38 @@ The system SHALL use the Rust `anydoc` processor for local regular files with th #### Scenario: A supported file is imported from a mixed local collection - **WHEN** file-document input resolves supported anydoc files together with Markdown or text files -- **THEN** each anydoc file is converted at its position in the deterministic file order +- **THEN** each anydoc file is converted by the bounded worker pool +- **AND** documents may be emitted in conversion completion order - **AND** existing Markdown and text files continue through their existing readers +### Requirement: Multi-source document conversion uses bounded concurrency + +The system SHALL convert files from a multi-source local document import concurrently. It SHALL bound the number of active conversions and completed conversion results retained in memory. Concurrent conversion SHALL emit source results as workers complete while preserving generated document identity and per-file error recovery behavior. + +#### Scenario: Multiple PDFs are converted concurrently + +- **WHEN** a multi-source local import contains more convertible PDFs than the conversion worker limit +- **THEN** the system permits multiple PDF conversions to execute at the same time +- **AND** it does not start an unbounded number of conversion operations + +#### Scenario: Concurrent conversions finish in a different order from discovery + +- **WHEN** a later file finishes conversion before an earlier file +- **THEN** the system emits the completed result without waiting for the earlier file +- **AND** output order is not guaranteed to match source-path order + +#### Scenario: A concurrent conversion fails + +- **WHEN** one file fails to read or convert while other file conversions are active +- **THEN** the system logs the same path-specific warning used by batch file recovery +- **AND** emits no document for the failed file +- **AND** continues emitting successful documents as conversions finish + +#### Scenario: Generated IDs are produced by concurrent conversion + +- **WHEN** multi-source conversion completes files in a different order across two runs +- **THEN** each emitted document receives the same generated ID in both runs + ### Requirement: Anydoc conversion preserves the existing file-document shape The system SHALL construct converted documents using the existing file-document semantics after Markdown conversion. It SHALL store converted Markdown under `content.`, apply the configured `--content` value, and add one `origin` metadata object for multi-file or glob-resolved local imports. The object SHALL contain `scheme`, `authority`, `path`, `query`, `fragment`, and `filename` fields. It SHALL NOT emit the legacy `file.path` or `file.name` metadata. diff --git a/openspec/specs/file-document-import/spec.md b/openspec/specs/file-document-import/spec.md index d4ef894..063a51d 100644 --- a/openspec/specs/file-document-import/spec.md +++ b/openspec/specs/file-document-import/spec.md @@ -62,16 +62,26 @@ When a batch of local file-document inputs encounters a per-file read or convers - **WHEN** the only direct file input cannot be read or converted - **THEN** ingestion fails with a diagnostic identifying the source and error -### Requirement: File document import order is deterministic -The system SHALL process file-document inputs in deterministic lexicographic path order after combining concrete file inputs and glob matches. +### Requirement: Multi-source file output order is unspecified +The system SHALL combine and de-duplicate concrete file inputs and glob matches deterministically before scheduling conversion. It SHALL emit each source result as conversion completes and SHALL NOT delay completed sources solely to restore lexicographic output order. -#### Scenario: Multiple files are imported -- **WHEN** file-document input contains multiple files -- **THEN** the emitted documents follow lexicographic order by file path +#### Scenario: Later file completes first +- **WHEN** a later source path finishes conversion before an earlier source path +- **THEN** the later source document may reach the output first #### Scenario: Same file appears more than once - **WHEN** a file is provided directly and also matched by a glob pattern -- **THEN** the system emits at most one document for that file +- **THEN** the system processes the resolved source once +- **AND** emits each document from that source at most once + +### Requirement: Local import summaries distinguish documents from files +The system SHALL report piped documents, evaluated documents, and discovered source files as `Piped X of Y docs from Z files ...`. A skipped source SHALL count toward the discovered file count but SHALL NOT add a document to the evaluated count. + +#### Scenario: Some discovered files are skipped +- **WHEN** a local import discovers 10 source files +- **AND** 3 files are skipped without producing documents +- **AND** the remaining files produce 7 documents that are sent successfully +- **THEN** the completion summary begins `Piped 7 of 7 docs from 10 files` ### Requirement: File documents store content in a configurable field The system SHALL store imported text content in the `content.` field named by the `--content ` command-line argument, defaulting to `content.body`. @@ -296,7 +306,7 @@ The system SHALL import `.ndjson` and `.jsonl` files as line-delimited JSON wher - **AND** the error identifies the file and line as invalid ### Requirement: Toon files stream one document per Toon document -The system SHALL import `.toon` files as structured Toon input where each decoded Toon object emits one JSON object document. +The system SHALL import `.toon` files as structured Toon input where each decoded Toon object emits one JSON object document. Documents within one Toon source SHALL preserve their source order. A Toon source within a multi-source file import MAY be emitted before or after other sources according to file conversion completion order. #### Scenario: Toon file is imported - **WHEN** a file-document input resolves to a `.toon` file @@ -305,8 +315,8 @@ The system SHALL import `.toon` files as structured Toon input where each decode #### Scenario: Toon file is included with multiple file inputs - **WHEN** file-document input contains a `.toon` file and other supported files -- **THEN** the `.toon` file participates in the existing deterministic file input order -- **AND** documents decoded from that `.toon` file are emitted at that file's position in the ordered input sequence +- **THEN** documents within the Toon file retain their source order +- **AND** the Toon file may be emitted before or after other sources according to conversion completion order #### Scenario: Toon file contains a non-object document - **WHEN** a `.toon` file contains a document that decodes to a non-object value diff --git a/openspec/specs/rawvalue-document-pipeline/spec.md b/openspec/specs/rawvalue-document-pipeline/spec.md index b935321..e801cc8 100644 --- a/openspec/specs/rawvalue-document-pipeline/spec.md +++ b/openspec/specs/rawvalue-document-pipeline/spec.md @@ -59,12 +59,25 @@ The system SHALL write raw JSON directly for stdout and file outputs. - **AND** it does not serialize the document through `serde_json::to_writer` ### Requirement: Elasticsearch bulk output preserves raw document buffering -The system SHALL buffer raw documents for Elasticsearch output and emit valid `_bulk` request bodies without requiring `Value` in the steady-state queue. +The system SHALL buffer raw documents for Elasticsearch output and emit valid `_bulk` request bodies without requiring `Value` in the steady-state queue. Unless the user supplies `--batch-size`, the system SHALL use a default batch size of 500 documents for multi-source local input and 5,000 documents for single-file streaming input and other input modes. + +#### Scenario: Multi-source local input uses smaller default batches +- **WHEN** local input resolves to more than one source file +- **AND** the user does not pass `--batch-size` +- **THEN** Elasticsearch bulk output targets 500 documents per request + +#### Scenario: Single-file streaming input retains large default batches +- **WHEN** the user imports one streaming NDJSON, CSV, JSON, or Toon source +- **AND** the user does not pass `--batch-size` +- **THEN** Elasticsearch bulk output targets 5,000 documents per request + +#### Scenario: Explicit batch size overrides the source-aware default +- **WHEN** the user passes `--batch-size 750` +- **THEN** Elasticsearch bulk output targets 750 documents per request regardless of input source count #### Scenario: Bulk queue flushes to Elasticsearch - **WHEN** the Elasticsearch output flushes one or more buffered documents - **THEN** it constructs a valid `_bulk` request body for `create` operations using the buffered raw JSON documents -- **AND** the implementation targets a batch size of `5,000` documents per bulk request - **AND** Elasticsearch accepts the request without document-shape regressions #### Scenario: Buffered documents are large diff --git a/src/input.rs b/src/input.rs index b35641c..497ade6 100644 --- a/src/input.rs +++ b/src/input.rs @@ -18,7 +18,11 @@ use std::{ fs::{self, File}, io::{BufRead, BufReader, Read, Seek, SeekFrom, Stdin, Write, stdin}, path::{Component, Path, PathBuf}, - sync::{OnceLock, mpsc::Receiver}, + sync::{ + Arc, OnceLock, + mpsc::{self, Receiver, SyncSender}, + }, + thread::{self, JoinHandle}, time::Duration, }; use tempfile::{Builder, NamedTempFile}; @@ -86,12 +90,10 @@ pub enum Input { }, FileDocuments { source: String, - paths: Vec, - origins: Vec, - path_index: usize, + workers: FileDocumentWorkers, + current_path: Option, documents: Vec>, document_index: usize, - content_field: String, generate_id: bool, skip_errors: bool, bundle_id: String, @@ -100,6 +102,36 @@ pub enum Input { }, } +const MAX_FILE_DOCUMENT_WORKERS: usize = 8; + +type FileDocumentConverter = + Arc) -> Result>> + Send + Sync>; + +struct FileDocumentJob { + path: PathBuf, + origin: OriginMetadata, +} + +struct FileDocumentResult { + worker_index: usize, + path: PathBuf, + documents: Result>>, +} + +struct FileDocumentWorker { + sender: SyncSender, + _handle: JoinHandle<()>, +} + +pub(crate) struct FileDocumentWorkers { + sources: Vec<(PathBuf, OriginMetadata)>, + workers: Vec, + results: Receiver, + idle_workers: VecDeque, + next_job_index: usize, + remaining_results: usize, +} + #[derive(Debug)] pub(crate) struct InputDocument { pub(crate) raw: Box, @@ -143,6 +175,115 @@ pub(crate) struct ActiveLocalSplit { file_identity: FileInputIdentity, } +impl FileDocumentWorkers { + fn start( + paths: Vec, + origins: Vec, + content_field: &str, + ) -> Result { + let available_workers = thread::available_parallelism() + .map(usize::from) + .unwrap_or(1); + let worker_limit = available_workers.min(MAX_FILE_DOCUMENT_WORKERS); + Self::start_with_converter( + paths, + origins, + content_field, + worker_limit, + Arc::new(read_file_documents), + ) + } + + fn start_with_converter( + paths: Vec, + origins: Vec, + content_field: &str, + worker_limit: usize, + converter: FileDocumentConverter, + ) -> Result { + let sources: Vec<_> = paths.into_iter().zip(origins).collect(); + let source_count = sources.len(); + let worker_count = worker_limit.max(1).min(sources.len()); + let (result_sender, results) = mpsc::sync_channel(worker_count); + let mut workers = Vec::with_capacity(worker_count); + + for worker_index in 0..worker_count { + let (sender, receiver) = mpsc::sync_channel::(1); + let result_sender = result_sender.clone(); + let content_field = content_field.to_string(); + let converter = Arc::clone(&converter); + let handle = thread::Builder::new() + .name(format!("espipe-convert-{worker_index}")) + .spawn(move || { + while let Ok(job) = receiver.recv() { + let documents = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + converter(&job.path, &content_field, Some(&job.origin)) + })) + .unwrap_or_else(|_| Err(eyre!("file conversion worker panicked"))); + let result = FileDocumentResult { + worker_index, + path: job.path, + documents, + }; + if result_sender.send(result).is_err() { + break; + } + } + }) + .map_err(|error| eyre!("Could not start file conversion worker: {error}"))?; + workers.push(FileDocumentWorker { + sender, + _handle: handle, + }); + } + drop(result_sender); + + Ok(Self { + sources, + workers, + results, + idle_workers: (0..worker_count).collect(), + next_job_index: 0, + remaining_results: source_count, + }) + } + + fn schedule_available(&mut self) -> Result<()> { + while self.next_job_index < self.sources.len() { + let Some(worker_index) = self.idle_workers.pop_front() else { + break; + }; + let (path, origin) = &self.sources[self.next_job_index]; + let job = FileDocumentJob { + path: path.clone(), + origin: origin.clone(), + }; + self.workers[worker_index] + .sender + .send(job) + .map_err(|_| eyre!("File conversion worker stopped unexpectedly"))?; + self.next_job_index += 1; + } + Ok(()) + } + + fn next_completed(&mut self) -> Result> { + if self.remaining_results == 0 { + return Ok(None); + } + self.schedule_available()?; + let result = self + .results + .recv() + .map_err(|_| eyre!("File conversion workers stopped unexpectedly"))?; + self.remaining_results -= 1; + self.idle_workers.push_back(result.worker_index); + self.schedule_available()?; + Ok(Some(result)) + } +} + impl InputDocument { fn from_raw(raw: Box) -> Self { Self { @@ -436,6 +577,16 @@ impl Input { Input::Stdin { .. } => None, } } + + pub(crate) fn is_multi_source_local(&self) -> bool { + matches!( + self, + Input::LocalSplitDocuments { file_count, .. } + | Input::LocalFileDocuments { file_count, .. } + | Input::FileDocuments { file_count, .. } + if *file_count > 1 + ) + } } fn finalize_file_input_document( @@ -979,17 +1130,43 @@ fn open_file_documents_from_paths( content_field: &str, generate_id: bool, skip_errors: bool, +) -> Result { + open_file_documents_from_paths_with_converter( + paths, + origins, + content_field, + generate_id, + skip_errors, + None, + ) +} + +fn open_file_documents_from_paths_with_converter( + paths: Vec, + origins: Vec, + content_field: &str, + generate_id: bool, + skip_errors: bool, + worker_settings: Option<(usize, FileDocumentConverter)>, ) -> Result { let file_count = paths.len(); let source = format!("{} file document(s)", paths.len()); + let workers = match worker_settings { + Some((worker_limit, converter)) => FileDocumentWorkers::start_with_converter( + paths, + origins, + content_field, + worker_limit, + converter, + )?, + None => FileDocumentWorkers::start(paths, origins, content_field)?, + }; Ok(Input::FileDocuments { source, - paths, - origins, - path_index: 0, + workers, + current_path: None, documents: Vec::new(), document_index: 0, - content_field: content_field.to_string(), generate_id, skip_errors, bundle_id: if generate_id { @@ -1004,17 +1181,14 @@ fn open_file_documents_from_paths( fn read_file_document_line(input: &mut Input) -> Result { let Input::FileDocuments { - paths, - origins, - path_index, + workers, + current_path, documents, document_index, - content_field, generate_id, skip_errors, bundle_id, evaluated_documents, - file_count: _, .. } = input else { @@ -1031,8 +1205,8 @@ fn read_file_document_line(input: &mut Input) -> Result { .and_then(|value| value.as_object().map(|object| object.contains_key("_id"))) .unwrap_or(false); let generated_id = if *generate_id && !has_explicit_id { - let path = paths - .get(path_index.saturating_sub(1)) + let path = current_path + .as_ref() .ok_or_else(|| eyre!("File document path cursor is invalid"))?; Some(file_document_id( bundle_id, @@ -1045,18 +1219,17 @@ fn read_file_document_line(input: &mut Input) -> Result { return Ok(InputDocument { raw, generated_id }); } - let Some(path) = paths.get(*path_index) else { + let Some(result) = workers.next_completed()? else { return Err(eyre!("No file document")); }; - let origin = origins.get(*path_index); - *path_index += 1; - *documents = match read_file_documents(path, content_field, origin) { + *current_path = Some(result.path.clone()); + *documents = match result.documents { Ok(documents) => { *evaluated_documents += documents.len(); documents } Err(error) if *skip_errors => { - log_skipped_file(path, error); + log_skipped_file(&result.path, error); Vec::new() } Err(error) => return Err(error), @@ -2286,10 +2459,11 @@ mod tests { JSON_LINE_OPENING_ERROR, REMOTE_NDJSON_ERROR, SymlinkMode, bundle_identifier, fetch_remote_input_with_client, fetch_remote_input_with_client_and_split, file_document_id, file_document_id_for_relative_path, input_kind_from_path, local_input_kind, - normalize_local_path, open_file_documents, open_input_values, - open_input_values_with_generate_id, open_input_values_with_generate_id_and_options, - open_split_inputs, origin_from_local_path, origin_from_uri, relative_path_from_working_dir, - validate_content_field, validate_ndjson_file, + normalize_local_path, open_file_documents, open_file_documents_from_paths_with_converter, + open_input_values, open_input_values_with_generate_id, + open_input_values_with_generate_id_and_options, open_split_inputs, origin_from_local_path, + origin_from_uri, relative_path_from_working_dir, validate_content_field, + validate_ndjson_file, }; use crate::json_split::SplitPath; use base64::Engine as _; @@ -2303,13 +2477,18 @@ mod tests { #[cfg(unix)] use std::os::unix::fs::symlink; use std::{ + collections::BTreeMap, fs, io::{Read, Write}, net::{TcpListener, TcpStream}, path::{Path, PathBuf}, - sync::{Arc, mpsc}, + sync::{ + Arc, Barrier, + atomic::{AtomicBool, AtomicUsize, Ordering}, + mpsc, + }, thread, - time::{SystemTime, UNIX_EPOCH}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use tempfile::NamedTempFile; @@ -3079,7 +3258,7 @@ mod tests { } #[test] - fn anydoc_mixed_file_import_sorts_paths_and_preserves_origin_metadata() { + fn anydoc_mixed_file_import_preserves_origin_metadata() { let dir = workspace_tempdir(); let pdf = dir.path().join("sample.pdf"); write_base64_fixture("anydoc/sample.pdf.base64", &pdf); @@ -3088,20 +3267,216 @@ mod tests { let values = collect_values(open_input_values(vec![uri(&rtf), uri(&pdf)], "body").unwrap()); assert_eq!(values.len(), 2); - assert_eq!(values[0]["origin"]["scheme"], "file"); - assert!(values[0]["origin"].get("authority").is_none()); - assert_eq!(values[0]["origin"]["filename"], "sample.pdf"); + let mut filenames = values + .iter() + .map(|value| value["origin"]["filename"].as_str().unwrap()) + .collect::>(); + filenames.sort(); + assert_eq!(filenames, ["sample.pdf", "sample.rtf"]); + for value in values { + assert_eq!(value["origin"]["scheme"], "file"); + assert!(value["origin"].get("authority").is_none()); + assert!(!value["origin"]["path"].as_str().unwrap().starts_with('/')); + assert!(value["origin"].get("query").is_none()); + assert!(value["origin"].get("fragment").is_none()); + assert!(value["content"]["body"].is_string()); + } + } + + #[test] + fn multi_source_file_workers_emit_as_completed_with_stable_ids() { + let dir = workspace_tempdir(); + let paths: Vec<_> = (0..6) + .map(|index| { + let path = dir.path().join(format!("{index}.txt")); + fs::write(&path, format!("document {index}")).unwrap(); + path + }) + .collect(); + let origins = paths + .iter() + .map(|path| origin_from_local_path(path)) + .collect::>(); + let repeat_paths = paths.clone(); + let repeat_origins = origins.clone(); + let active = Arc::new(AtomicUsize::new(0)); + let maximum_active = Arc::new(AtomicUsize::new(0)); + let converter = { + let active = Arc::clone(&active); + let maximum_active = Arc::clone(&maximum_active); + Arc::new( + move |path: &Path, _: &str, _: Option<&super::OriginMetadata>| { + let active_now = active.fetch_add(1, Ordering::SeqCst) + 1; + maximum_active.fetch_max(active_now, Ordering::SeqCst); + let filename = path.file_name().unwrap().to_string_lossy().into_owned(); + if filename == "0.txt" { + thread::sleep(Duration::from_millis(60)); + } else { + thread::sleep(Duration::from_millis(10)); + } + active.fetch_sub(1, Ordering::SeqCst); + let raw = serde_json::value::RawValue::from_string(format!( + r#"{{"filename":"{filename}"}}"# + ))?; + Ok(vec![raw]) + }, + ) as super::FileDocumentConverter + }; + + let documents = collect_documents( + open_file_documents_from_paths_with_converter( + paths, + origins, + "body", + true, + false, + Some((3, converter)), + ) + .unwrap(), + ); + + assert!(maximum_active.load(Ordering::SeqCst) >= 2); + let filenames: Vec<_> = documents + .iter() + .map(|document| { + serde_json::from_str::(document.get()).unwrap()["filename"] + .as_str() + .unwrap() + .to_string() + }) + .collect(); + assert_ne!(filenames[0], "0.txt"); + let mut sorted_filenames = filenames.clone(); + sorted_filenames.sort(); + assert_eq!( + sorted_filenames, + ["0.txt", "1.txt", "2.txt", "3.txt", "4.txt", "5.txt"] + ); assert!( - !values[0]["origin"]["path"] - .as_str() - .unwrap() - .starts_with('/') + documents + .iter() + .all(|document| document.generated_id.is_some()) ); - assert!(values[0]["origin"].get("query").is_none()); - assert!(values[0]["origin"].get("fragment").is_none()); - assert_eq!(values[1]["origin"]["filename"], "sample.rtf"); - assert!(values[0]["content"]["body"].is_string()); - assert!(values[1]["content"]["body"].is_string()); + + let repeat_converter = + Arc::new(|path: &Path, _: &str, _: Option<&super::OriginMetadata>| { + let filename = path.file_name().unwrap().to_string_lossy(); + let raw = serde_json::value::RawValue::from_string(format!( + r#"{{"filename":"{filename}"}}"# + ))?; + Ok(vec![raw]) + }) as super::FileDocumentConverter; + let repeated = collect_documents( + open_file_documents_from_paths_with_converter( + repeat_paths, + repeat_origins, + "body", + true, + false, + Some((3, repeat_converter)), + ) + .unwrap(), + ); + let ids_by_filename = |documents: &[InputDocument]| { + documents + .iter() + .map(|document| { + let value: serde_json::Value = serde_json::from_str(document.get()).unwrap(); + ( + value["filename"].as_str().unwrap().to_string(), + document.generated_id.clone().unwrap(), + ) + }) + .collect::>() + }; + assert_eq!(ids_by_filename(&documents), ids_by_filename(&repeated)); + } + + #[test] + fn multi_source_file_workers_recover_from_failure_during_active_conversion() { + let dir = workspace_tempdir(); + let paths: Vec<_> = (0..3) + .map(|index| { + let path = dir.path().join(format!("{index}.txt")); + fs::write(&path, format!("document {index}")).unwrap(); + path + }) + .collect(); + let origins = paths + .iter() + .map(|path| origin_from_local_path(path)) + .collect::>(); + let first_jobs_started = Arc::new(Barrier::new(2)); + let active = Arc::new(AtomicUsize::new(0)); + let failure_saw_parallel_work = Arc::new(AtomicBool::new(false)); + let converter = { + let first_jobs_started = Arc::clone(&first_jobs_started); + let active = Arc::clone(&active); + let failure_saw_parallel_work = Arc::clone(&failure_saw_parallel_work); + Arc::new( + move |path: &Path, _: &str, _: Option<&super::OriginMetadata>| { + active.fetch_add(1, Ordering::SeqCst); + let filename = path.file_name().unwrap().to_string_lossy().into_owned(); + if filename == "0.txt" || filename == "1.txt" { + first_jobs_started.wait(); + } + if filename == "0.txt" { + failure_saw_parallel_work + .store(active.load(Ordering::SeqCst) >= 2, Ordering::SeqCst); + active.fetch_sub(1, Ordering::SeqCst); + return Err(eyre::eyre!("forced conversion failure")); + } + if filename == "1.txt" { + thread::sleep(Duration::from_millis(30)); + } + active.fetch_sub(1, Ordering::SeqCst); + let raw = serde_json::value::RawValue::from_string(format!( + r#"{{"filename":"{filename}"}}"# + ))?; + Ok(vec![raw]) + }, + ) as super::FileDocumentConverter + }; + + let documents = collect_documents( + open_file_documents_from_paths_with_converter( + paths, + origins, + "body", + false, + true, + Some((2, converter)), + ) + .unwrap(), + ); + + assert!(failure_saw_parallel_work.load(Ordering::SeqCst)); + let mut filenames = documents + .iter() + .map(|document| { + serde_json::from_str::(document.get()).unwrap()["filename"] + .as_str() + .unwrap() + .to_string() + }) + .collect::>(); + filenames.sort(); + assert_eq!(filenames, ["1.txt", "2.txt"]); + } + + #[test] + fn local_source_count_selects_multi_source_mode_after_discovery() { + let dir = workspace_tempdir(); + let first = dir.path().join("a.md"); + let second = dir.path().join("b.md"); + fs::write(&first, "a").unwrap(); + fs::write(&second, "b").unwrap(); + + let single = open_input_values(vec![uri(&first)], "body").unwrap(); + let multiple = open_input_values(vec![uri(&first), uri(&second)], "body").unwrap(); + + assert!(!single.is_multi_source_local()); + assert!(multiple.is_multi_source_local()); } #[test] @@ -3133,7 +3508,7 @@ mod tests { } #[test] - fn anydoc_multiple_extension_globs_combine_and_sort_inputs() { + fn anydoc_multiple_extension_globs_combine_inputs() { let dir = workspace_tempdir(); let pdf = dir.path().join("a.pdf"); let rtf = dir.path().join("b.rtf"); @@ -3154,8 +3529,12 @@ mod tests { ); assert_eq!(values.len(), 2); - assert_eq!(values[0]["origin"]["filename"], "a.pdf"); - assert_eq!(values[1]["origin"]["filename"], "b.rtf"); + let mut filenames = values + .iter() + .map(|value| value["origin"]["filename"].as_str().unwrap()) + .collect::>(); + filenames.sort(); + assert_eq!(filenames, ["a.pdf", "b.rtf"]); } #[test] @@ -3183,7 +3562,7 @@ mod tests { } #[test] - fn shell_expanded_files_are_sorted_deduplicated_and_include_origin_metadata() { + fn shell_expanded_files_are_deduplicated_and_include_origin_metadata() { let dir = workspace_tempdir(); let b = dir.path().join("b.txt"); let a = dir.path().join("a.txt"); @@ -3194,10 +3573,17 @@ mod tests { let values = collect_values(input); assert_eq!(values.len(), 2); - assert_eq!(values[0]["content"]["body"], "alpha"); - assert_eq!(values[1]["content"]["body"], "bravo"); - assert_eq!(values[0]["origin"]["filename"], "a.txt"); - assert_eq!(values[1]["origin"]["filename"], "b.txt"); + let mut documents = values + .iter() + .map(|value| { + ( + value["origin"]["filename"].as_str().unwrap(), + value["content"]["body"].as_str().unwrap(), + ) + }) + .collect::>(); + documents.sort(); + assert_eq!(documents, [("a.txt", "alpha"), ("b.txt", "bravo")]); } #[test] @@ -3218,21 +3604,21 @@ mod tests { let values = collect_values(input); assert_eq!(values.len(), 2); - assert_eq!(values[0]["content"]["body"], "child"); - assert_eq!(values[1]["content"]["body"], "root"); - assert_eq!(values[0]["origin"]["filename"], "child.md"); - assert!( - !values[0]["origin"]["path"] - .as_str() - .unwrap() - .starts_with('/') - ); - assert_eq!(values[1]["origin"]["filename"], "root.md"); + let mut documents = values + .iter() + .map(|value| { + ( + value["origin"]["filename"].as_str().unwrap(), + value["content"]["body"].as_str().unwrap(), + ) + }) + .collect::>(); + documents.sort(); + assert_eq!(documents, [("child.md", "child"), ("root.md", "root")]); assert!( - !values[1]["origin"]["path"] - .as_str() - .unwrap() - .starts_with('/') + values + .iter() + .all(|value| !value["origin"]["path"].as_str().unwrap().starts_with('/')) ); } @@ -3558,9 +3944,16 @@ mod tests { collect_values(open_input_values(vec![uri(&text), uri(&toon)], "body").unwrap()); assert_eq!(values.len(), 2); - assert_eq!(values[0]["content"]["body"], "alpha"); - assert_eq!(values[1]["id"], 2); - assert_eq!(values[1]["origin"]["filename"], "b.toon"); + let text_value = values + .iter() + .find(|value| value["origin"]["filename"] == "a.txt") + .unwrap(); + let toon_value = values + .iter() + .find(|value| value["origin"]["filename"] == "b.toon") + .unwrap(); + assert_eq!(text_value["content"]["body"], "alpha"); + assert_eq!(toon_value["id"], 2); } #[test] diff --git a/src/main.rs b/src/main.rs index dc2ebd2..d034bd3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -113,12 +113,11 @@ struct Cli { hidden: HiddenMode, /// Documents per Elasticsearch bulk request #[arg( - help = "Documents per Elasticsearch bulk request", + help = "Documents per Elasticsearch bulk request (default: 500 for multi-source local input, 5000 otherwise)", long, - default_value_t = ElasticsearchOutputConfig::DEFAULT_BATCH_SIZE, value_parser = parse_nonzero_usize )] - batch_size: usize, + batch_size: Option, /// Maximum concurrent Elasticsearch bulk requests #[arg( help = "Maximum concurrent Elasticsearch bulk requests", @@ -201,11 +200,6 @@ async fn main() -> ExitCode { Ok(auth) => auth, Err(err) => return exit_with_error(err), }; - let elasticsearch_config = match ElasticsearchOutputConfig::try_new(batch_size, max_requests) { - Ok(config) => config, - Err(err) => return exit_with_error(err), - }; - let preflight = OutputPreflightConfig { pipeline, pipeline_name, @@ -216,60 +210,39 @@ async fn main() -> ExitCode { if let Err(err) = preflight.validate() { return exit_with_error(err); } + if let Err(err) = Output::validate_preflight_target(&output, &preflight) { + return exit_with_error(err); + } - let (mut input, mut output) = if preflight.has_elasticsearch_options() { - let output = match Output::try_new( - insecure, - auth, - output, - elastic_cli_url(), - action, - !uncompressed, - elasticsearch_config, - preflight, - ) - .await - { - Ok(output) => output, + let discovery_options = DiscoveryOptions { symlinks, hidden }; + let mut input = + match Input::try_new(inputs, content, split, generate_id, discovery_options).await { + Ok(input) => input, Err(err) => return exit_with_error(err), }; - log::debug!("output: {output}"); - - let discovery_options = DiscoveryOptions { symlinks, hidden }; - let input = - match Input::try_new(inputs, content, split, generate_id, discovery_options).await { - Ok(input) => input, - Err(err) => return exit_with_error(err), - }; - log::debug!("input: {input}"); - (input, output) - } else { - let discovery_options = DiscoveryOptions { symlinks, hidden }; - let input = - match Input::try_new(inputs, content, split, generate_id, discovery_options).await { - Ok(input) => input, - Err(err) => return exit_with_error(err), - }; - log::debug!("input: {input}"); + log::debug!("input: {input}"); - let output = match Output::try_new( - insecure, - auth, - output, - elastic_cli_url(), - action, - !uncompressed, - elasticsearch_config, - preflight, - ) - .await - { - Ok(output) => output, - Err(err) => return exit_with_error(err), - }; - log::debug!("output: {output}"); - (input, output) + let batch_size = effective_batch_size(batch_size, input.is_multi_source_local()); + let elasticsearch_config = match ElasticsearchOutputConfig::try_new(batch_size, max_requests) { + Ok(config) => config, + Err(err) => return exit_with_error(err), }; + let mut output = match Output::try_new( + insecure, + auth, + output, + elastic_cli_url(), + action, + !uncompressed, + elasticsearch_config, + preflight, + ) + .await + { + Ok(output) => output, + Err(err) => return exit_with_error(err), + }; + log::debug!("output: {output}"); let mut input_line: usize = 0; let mut output_line: usize = 0; @@ -302,10 +275,10 @@ async fn main() -> ExitCode { if let Some(file_count) = input.file_count(input_line) { let file_label = if file_count == 1 { "file" } else { "files" }; println!( - "From {} {file_label}, piped {} of {} docs to {output_name} in {:.3} seconds", - comma_formatted(file_count), + "Piped {} of {} docs from {} {file_label} to {output_name} in {:.3} seconds", comma_formatted(output_line), comma_formatted(evaluated_line), + comma_formatted(file_count), start_time.elapsed().as_secs_f32() ); } else { @@ -390,6 +363,14 @@ fn parse_nonzero_usize(value: &str) -> Result { Ok(parsed) } +fn effective_batch_size(explicit: Option, multi_source_local: bool) -> usize { + explicit.unwrap_or(if multi_source_local { + ElasticsearchOutputConfig::MULTI_SOURCE_DEFAULT_BATCH_SIZE + } else { + ElasticsearchOutputConfig::DEFAULT_BATCH_SIZE + }) +} + fn elastic_cli_url() -> Option { env::var("ELASTIC_ES_URL").ok() } @@ -413,7 +394,8 @@ fn resolve_api_key( #[cfg(test)] mod tests { - use super::resolve_api_key; + use super::{effective_batch_size, resolve_api_key}; + use crate::output::ElasticsearchOutputConfig; #[test] fn elastic_cli_api_key_is_used_without_explicit_authentication() { @@ -444,4 +426,22 @@ mod tests { None ); } + + #[test] + fn batch_size_default_depends_on_local_source_count() { + assert_eq!( + effective_batch_size(None, true), + ElasticsearchOutputConfig::MULTI_SOURCE_DEFAULT_BATCH_SIZE + ); + assert_eq!( + effective_batch_size(None, false), + ElasticsearchOutputConfig::DEFAULT_BATCH_SIZE + ); + } + + #[test] + fn explicit_batch_size_overrides_input_default() { + assert_eq!(effective_batch_size(Some(750), true), 750); + assert_eq!(effective_batch_size(Some(750), false), 750); + } } diff --git a/src/output/elasticsearch.rs b/src/output/elasticsearch.rs index 6e2aa37..ee618aa 100644 --- a/src/output/elasticsearch.rs +++ b/src/output/elasticsearch.rs @@ -23,6 +23,7 @@ use tokio::{sync::mpsc, task::JoinHandle, time::sleep}; use url::Url; const DEFAULT_BATCH_SIZE: usize = 5_000; +const MULTI_SOURCE_DEFAULT_BATCH_SIZE: usize = 500; const DEFAULT_MAX_INFLIGHT_REQUESTS: usize = 16; #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -64,6 +65,7 @@ impl TemplateConfig { impl ElasticsearchOutputConfig { pub const DEFAULT_BATCH_SIZE: usize = DEFAULT_BATCH_SIZE; + pub const MULTI_SOURCE_DEFAULT_BATCH_SIZE: usize = MULTI_SOURCE_DEFAULT_BATCH_SIZE; pub const DEFAULT_MAX_INFLIGHT_REQUESTS: usize = DEFAULT_MAX_INFLIGHT_REQUESTS; pub fn try_new(batch_size: usize, max_inflight_requests: usize) -> Result { diff --git a/src/output/mod.rs b/src/output/mod.rs index b1c4be9..9adc557 100644 --- a/src/output/mod.rs +++ b/src/output/mod.rs @@ -80,6 +80,16 @@ impl OutputPreflightConfig { } impl Output { + pub fn validate_preflight_target( + uri: &UriRef, + preflight: &OutputPreflightConfig, + ) -> Result<()> { + match uri.scheme().map(|scheme| scheme.as_str()) { + Some("file") | None => reject_elasticsearch_options(preflight), + _ => Ok(()), + } + } + pub async fn try_new( insecure: bool, auth: Auth, diff --git a/tests/file_output.rs b/tests/file_output.rs index e248382..9e0d535 100644 --- a/tests/file_output.rs +++ b/tests/file_output.rs @@ -610,7 +610,7 @@ fn cli_warns_and_skips_when_later_file_document_read_fails() { assert!(stderr.contains("skipping file")); let stdout = String::from_utf8_lossy(&output.stdout); assert!( - stdout.contains("From 2 files, piped 1 of 1 docs to"), + stdout.contains("Piped 1 of 1 docs from 2 files to"), "summary should count the skipped file as evaluated: {stdout}" ); let documents = json_lines(&fs::read(&output_path).expect("read output")); From 9b8a74210c6d6d21cd9bef4028f7e188aac7759e Mon Sep 17 00:00:00 2001 From: Ryan Eno Date: Sat, 22 Aug 2026 22:59:05 -0700 Subject: [PATCH 6/6] Fix import preflight ordering and file counts --- src/input.rs | 33 +++++++++++++-- src/main.rs | 116 +++++++++++++++++++++++++++++++++++++++------------ 2 files changed, 120 insertions(+), 29 deletions(-) diff --git a/src/input.rs b/src/input.rs index 497ade6..01013c8 100644 --- a/src/input.rs +++ b/src/input.rs @@ -563,14 +563,15 @@ impl Input { } } - pub(crate) fn file_count(&self, successful_documents: usize) -> Option { + pub(crate) fn file_count(&self) -> Option { match self { Input::FileJson { origin, .. } | Input::FileCsv { origin, .. } | Input::FileToon { origin, .. } | Input::JsonSplit { origin, .. } => origin .as_ref() - .map(|_| usize::from(successful_documents > 0)), + .filter(|origin| origin.scheme == "file") + .map(|_| 1), Input::LocalSplitDocuments { file_count, .. } | Input::LocalFileDocuments { file_count, .. } | Input::FileDocuments { file_count, .. } => Some(*file_count), @@ -2479,7 +2480,7 @@ mod tests { use std::{ collections::BTreeMap, fs, - io::{Read, Write}, + io::{Cursor, Read, Write}, net::{TcpListener, TcpStream}, path::{Path, PathBuf}, sync::{ @@ -3479,6 +3480,32 @@ mod tests { assert!(multiple.is_multi_source_local()); } + #[test] + fn empty_local_streaming_input_still_reports_one_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("empty.ndjson"); + fs::write(&path, "").unwrap(); + + let input = Input::try_from(uri(&path)).unwrap(); + + assert_eq!(input.file_count(), Some(1)); + } + + #[test] + fn remote_streaming_input_does_not_report_a_local_file_count() { + let remote = UriRef::parse("https://example.com/docs.ndjson".to_string()).unwrap(); + let input = Input::FileJson { + source: remote.to_string(), + reader: Box::new(std::io::BufReader::new(Box::new(Cursor::new(Vec::new())))), + first_record: true, + origin: Some(origin_from_uri(&remote)), + file_identity: None, + _temp_file: None, + }; + + assert_eq!(input.file_count(), None); + } + #[test] fn anydoc_recursive_glob_imports_pdf_files() { let dir = workspace_tempdir(); diff --git a/src/main.rs b/src/main.rs index d034bd3..27a8e07 100644 --- a/src/main.rs +++ b/src/main.rs @@ -215,34 +215,70 @@ async fn main() -> ExitCode { } let discovery_options = DiscoveryOptions { symlinks, hidden }; - let mut input = - match Input::try_new(inputs, content, split, generate_id, discovery_options).await { - Ok(input) => input, + let discover_before_output = should_discover_input_before_output(batch_size, &inputs); + let (mut input, mut output) = if discover_before_output { + let input = + match Input::try_new(inputs, content, split, generate_id, discovery_options).await { + Ok(input) => input, + Err(err) => return exit_with_error(err), + }; + log::debug!("input: {input}"); + + let batch_size = effective_batch_size(batch_size, input.is_multi_source_local()); + let elasticsearch_config = + match ElasticsearchOutputConfig::try_new(batch_size, max_requests) { + Ok(config) => config, + Err(err) => return exit_with_error(err), + }; + let output = match Output::try_new( + insecure, + auth, + output, + elastic_cli_url(), + action, + !uncompressed, + elasticsearch_config, + preflight, + ) + .await + { + Ok(output) => output, + Err(err) => return exit_with_error(err), + }; + log::debug!("output: {output}"); + (input, output) + } else { + let batch_size = effective_batch_size(batch_size, false); + let elasticsearch_config = + match ElasticsearchOutputConfig::try_new(batch_size, max_requests) { + Ok(config) => config, + Err(err) => return exit_with_error(err), + }; + let output = match Output::try_new( + insecure, + auth, + output, + elastic_cli_url(), + action, + !uncompressed, + elasticsearch_config, + preflight, + ) + .await + { + Ok(output) => output, Err(err) => return exit_with_error(err), }; - log::debug!("input: {input}"); + log::debug!("output: {output}"); - let batch_size = effective_batch_size(batch_size, input.is_multi_source_local()); - let elasticsearch_config = match ElasticsearchOutputConfig::try_new(batch_size, max_requests) { - Ok(config) => config, - Err(err) => return exit_with_error(err), - }; - let mut output = match Output::try_new( - insecure, - auth, - output, - elastic_cli_url(), - action, - !uncompressed, - elasticsearch_config, - preflight, - ) - .await - { - Ok(output) => output, - Err(err) => return exit_with_error(err), + let input = + match Input::try_new(inputs, content, split, generate_id, discovery_options).await { + Ok(input) => input, + Err(err) => return exit_with_error(err), + }; + log::debug!("input: {input}"); + (input, output) }; - log::debug!("output: {output}"); let mut input_line: usize = 0; let mut output_line: usize = 0; @@ -272,7 +308,7 @@ async fn main() -> ExitCode { }; if !quiet { let evaluated_line = input.evaluated_document_count(input_line); - if let Some(file_count) = input.file_count(input_line) { + if let Some(file_count) = input.file_count() { let file_label = if file_count == 1 { "file" } else { "files" }; println!( "Piped {} of {} docs from {} {file_label} to {output_name} in {:.3} seconds", @@ -371,6 +407,13 @@ fn effective_batch_size(explicit: Option, multi_source_local: bool) -> us }) } +fn should_discover_input_before_output( + explicit_batch_size: Option, + inputs: &[UriRef], +) -> bool { + inputs.len() > 1 || (explicit_batch_size.is_none() && inputs.iter().all(is_local_file_input)) +} + fn elastic_cli_url() -> Option { env::var("ELASTIC_ES_URL").ok() } @@ -394,8 +437,9 @@ fn resolve_api_key( #[cfg(test)] mod tests { - use super::{effective_batch_size, resolve_api_key}; + use super::{effective_batch_size, resolve_api_key, should_discover_input_before_output}; use crate::output::ElasticsearchOutputConfig; + use fluent_uri::UriRef; #[test] fn elastic_cli_api_key_is_used_without_explicit_authentication() { @@ -444,4 +488,24 @@ mod tests { assert_eq!(effective_batch_size(Some(750), true), 750); assert_eq!(effective_batch_size(Some(750), false), 750); } + + #[test] + fn multi_input_validation_and_implicit_local_batch_sizes_precede_output() { + let local = UriRef::parse("docs/**/*.pdf".to_string()).unwrap(); + let remote = UriRef::parse("https://example.com/docs.ndjson".to_string()).unwrap(); + let second_remote = UriRef::parse("https://example.com/more.ndjson".to_string()).unwrap(); + let stdin = UriRef::parse("-".to_string()).unwrap(); + + assert!(should_discover_input_before_output(None, &[local.clone()])); + assert!(!should_discover_input_before_output(Some(750), &[local])); + assert!(!should_discover_input_before_output( + None, + &[remote.clone()] + )); + assert!(should_discover_input_before_output( + Some(750), + &[remote, second_remote] + )); + assert!(!should_discover_input_before_output(None, &[stdin])); + } }