Parallelize multi-file imports and skip unreadable files

.gitignore

@@ -5,3 +5,5 @@
 !/.agents/skills/espipe/
 !/.agents/skills/espipe/**
 /target
+/.agents/skillator.yaml
+/.agents/.gitignore

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [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
+
+- 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
+
 ### Added
 
 - Added `elasticsearch:/index` and `es:/index` Elastic CLI context targets using `ELASTIC_ES_URL` and `ELASTIC_ES_API_KEY`.

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"

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_ID>    Generate deterministic IDs for local files (default: multi-source only)
       --symlinks <SYMLINKS>          Multi-source symlink policy [default: skip] [possible values: follow, fail, skip]
       --hidden <HIDDEN>              Multi-source hidden-path policy [default: skip] [possible values: include, fail, skip]
-      --batch-size <BATCH_SIZE>      Documents per Elasticsearch bulk request [default: 5000]
+      --batch-size <BATCH_SIZE>      Documents per Elasticsearch bulk request (default: 500 for multi-source local input, 5000 otherwise)
       --max-requests <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.
 
@@ -151,7 +151,9 @@ 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 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,14 +247,17 @@ 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
 - logs bulk-item error counts when Elasticsearch reports partial failures
 
 `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: `Piped 5,850 of 5,850 docs from 6,246 files ...`.
+
 ### 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.
@@ -442,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
 

openspec/changes/archive/2026-08-22-parallel-file-conversion/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-22

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.

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.