Add parallel streaming Reader and crates.io publishing

.github/workflows/ci.yml

@@ -17,6 +17,7 @@ jobs:
       - uses: dtolnay/rust-toolchain@stable
       - run: cargo test --release
       - run: cargo check --all-features
+      - run: cargo package
       - uses: actions/setup-python@v7
         with:
           python-version: '3.12'
@@ -49,31 +50,9 @@ jobs:
           name: wheels-${{ matrix.os }}
           path: dist
 
-  sdist:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v7
-      - uses: dtolnay/rust-toolchain@stable
-      - uses: actions/setup-python@v7
-        with:
-          python-version: '3.14'
-      - run: mkdir -p target/wheel-data
-      - uses: PyO3/maturin-action@v1
-        with:
-          command: sdist
-          args: -o dist
-      - run: |
-          python -m venv /tmp/fbz-sdist-test
-          /tmp/fbz-sdist-test/bin/pip install pytest dist/*.tar.gz
-          /tmp/fbz-sdist-test/bin/pytest tests/test_install.py
-      - uses: actions/upload-artifact@v7
-        with:
-          name: wheels-sdist
-          path: dist
-
   publish:
     if: startsWith(github.ref, 'refs/tags/v')
-    needs: [build, sdist]
+    needs: build
     runs-on: ubuntu-latest
     permissions:
       id-token: write
@@ -91,3 +70,21 @@ jobs:
       - uses: pypa/gh-action-pypi-publish@release/v1
         with:
           packages-dir: dist/
+
+  publish-crate:
+    if: startsWith(github.ref, 'refs/tags/v')
+    needs: test
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+    steps:
+      - uses: actions/checkout@v7
+      - uses: dtolnay/rust-toolchain@stable
+      - id: auth
+        uses: rust-lang/crates-io-auth-action@v1
+      - name: Publish crate when its version is new
+        run: |
+          version=$(cargo metadata --format-version 1 --no-deps | jq -r '.packages[] | select(.name=="fbz") | .version')
+          if curl -sf -A "fbz-ci (https://github.com/AnswerDotAI/fbz)" "https://crates.io/api/v1/crates/fbz/$version" >/dev/null; then echo "fbz $version is already published"; else cargo publish; fi
+        env:
+          CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }}

Cargo.toml

@@ -7,7 +7,8 @@ license = "Apache-2.0"
 description = "Compression-format research workbench with fast bzip2, gzip, LZ4, and ZIP decompression"
 repository = "https://github.com/AnswerDotAI/fbz"
 homepage = "https://github.com/AnswerDotAI/fbz"
-documentation = "https://github.com/AnswerDotAI/fbz"
+documentation = "https://docs.rs/fbz"
+include = ["/src/**", "/Cargo.toml", "/README.md", "/LICENSE"]
 
 [lib]
 name = "fbz"

DEV.md

@@ -49,6 +49,10 @@ ZIP scheduling deliberately uses one parallelism level at a time. A sole DEFLATE
 
 The shared `pipeline.rs` scheduler provides ordered results, byte-budgeted admission, cancellation, and a staged priority queue. Bzip2 uses the rolling candidate path: workers reserve the maximum possible decoded block size, then shrink that reservation to actual retained output until ordered validation consumes or rejects it. Gzip uses the staged path: native workers alternate speculative DEFLATE decoding with higher-priority marker resolution, while the coordinator advances only the 32 KiB dependency windows and emits resolved chunks in order. LZ4 independent blocks use the simpler ordered-job path and charge each retained allocation against the budget. Decode results and outstanding resolution results have bounded horizons, preventing dependency stalls from causing unbounded memory.
 
+The shared `OutputSink` boundary accepts owned decoder chunks. Direct writer APIs adapt those chunks to `Write::write_all` without a channel or allocation copy. `Reader` and streaming tar extraction instead use the same zero-capacity owned-chunk pipe, so completed bzip2 blocks and resolved parallel-gzip segments move into the consumer rather than being copied into an intermediate pipe buffer. The pipe adds at most the consumer's current chunk and the producer's next blocked chunk beyond the scheduler budget. Its receiver owns a cancellation flag; parallel bzip2 scanning checks that flag between bounded waves.
+
+The non-indexing parallel bzip2 path decodes and emits its first CRC-validated block before scanning the remainder of the compressed source. That candidate is then reused by normal ordered assembly rather than decoded twice. Index construction retains the full scan-first path because it produces no output. `Reader` sends decoder success or failure separately from the byte pipe and treats a missing terminal status as an error, so a worker panic or corrupt trailer cannot appear as EOF. Reader errors are sticky across subsequent calls, and drop disconnects the pipe before joining the decoder thread.
+
 The bzip2 decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Gzip uses full canonical lookup tables packed into `u16`, a branch-free 64 KiB marker-resolution lookup for large chunks, and `crc32fast::Hasher::combine` so CRC scanning runs with the resolution workers rather than serially in the coordinator. Gzip byte/marker output and LZ4 share `history::extend_match`, whose doubling copies handle overlapping matches in logarithmically many operations; format-specific history validation remains at each call site. The only unsafe codec operation marks a just-initialized `Vec` result as initialized after writing every spare-capacity byte. Add architecture-specific SIMD or further cross-codec abstraction only after profiling; `libbz2-rs-sys`, `flate2`, and `lz4_flex` remain dev-only differential oracles.
 
 ## Commands
@@ -58,6 +62,7 @@ cargo test
 cargo test --release
 cargo check --all-features
 cargo build --release --bins
+cargo package
 python tools/stage_binaries.py
 uv pip install --reinstall --no-deps -e .
 pytest -q
@@ -74,6 +79,21 @@ The normal release test path decodes selected valid and corrupt cases from the m
 
 The normal release path contains warmed end-to-end performance regression gates capped at 1.3 times each oracle, allowing for noise on shared runners. The gzip gates independently exercise a highly compressible LZ77-heavy shape and an incompressible literal-heavy shape against `flate2`; the bzip2 gate uses `libbz2-rs-sys`. Representative local acceptance remains 1.2 times the corresponding oracle. The ignored full-wiki gzip test applies that threshold to rapidgzip-rust. Keep the whole release test suite below five seconds on the primary development laptop; individual timed workloads should normally be about 0.1 seconds or less.
 
+### Local Reader benchmark
+
+`tests/reader_perf.rs` compares the public `Read` adapter with the direct writer path on both 84.4 MiB SimpleWiki fixtures. Each row is one release-mode run, and both timings include opening the file. The Reader path necessarily copies into the caller's buffer but transfers decoder-owned chunks into its rendezvous pipe without another copy:
+
+```bash
+cargo test --release --test reader_perf reader_writer_comparison -- --ignored --exact --nocapture
+```
+
+| Format | First byte | Reader | Direct writer | Reader/writer |
+|---|---:|---:|---:|---:|
+| bzip2 | 36.627 ms | 205.081 ms | 183.465 ms | 1.118x |
+| gzip | 11.509 ms | 28.119 ms | 24.515 ms | 1.147x |
+
+The bzip2 first-byte measurement includes opening, mapping, and decoding its first block, but not the subsequent whole-input parallel marker scan. The gzip path parses only the current member header before beginning DEFLATE work. Keep this diagnostic single-run.
+
 ### Local LZ4 benchmark reproduction and diagnostics
 
 `tests/lz4_perf.rs` decodes `meta/simplewiki-first-5pct.xml.bz2` before timing, then uses dev-only `lz4_flex` to create a standard Max4MB independent-block LZ4 frame with a content checksum. Fixture construction and both warm-ups are outside the measured intervals. One test then measures exactly one validation run of each CLI and records child-process memory without task-inspection permissions:
@@ -276,13 +296,14 @@ CI tests and builds Linux on x86-64 and ARM64, and macOS on ARM64. macOS Intel r
 
 The canonical version lives in `Cargo.toml`. `pyproject.toml` gets the Python package version from Cargo via `dynamic = ["version"]`.
 
-The thin PEP 517 backend delegates to Maturin after building and staging the native executable. This makes wheels built from the sdist contain the same native CLI as CI-built wheels. Release CI verifies that path from a fresh Python 3.14 environment before publishing.
+The thin PEP 517 backend delegates to Maturin after building and staging the native executable, so repository and editable builds contain the same native CLI as CI-built wheels. Releases publish platform wheels rather than a Python sdist; the Rust source package is published separately to crates.io.
 
 ## Release
 
 1. Run `cargo build --release --bins && python tools/stage_binaries.py`.
 2. Run `uv pip install --reinstall --no-deps -e . && pytest -q` so the custom backend installs both the extension and native CLI.
 3. Confirm the release version in `Cargo.toml` (`[package].version`).
-4. Run `ship-release`.
+4. For the first crates.io release only, run `cargo publish`, then configure the `ci.yml` trusted publisher for `AnswerDotAI/fbz`; crates.io requires the crate to exist before trusted publishing can be configured.
+5. Run `ship-release`.
 
-Fastship pushes the version tag for GitHub Actions, then bumps and pushes `Cargo.toml`.
+Fastship pushes the version tag for GitHub Actions, then bumps and pushes `Cargo.toml`. Tagged CI publishes the crate through crates.io trusted publishing as well as building the GitHub release and PyPI packages.

README.md

@@ -16,11 +16,11 @@ pip install fbz
 
 Python 3.10 and later are supported. Prebuilt wheels target Linux on x86-64 and ARM64, and macOS on ARM64. macOS Intel is best-effort and can build from source.
 
-The Rust crate is not yet published separately on crates.io. Install the CLI from the repository, or add the library as a Git dependency:
+Install the native CLI or add the Rust library from crates.io:
 
 ```bash
-cargo install --git https://github.com/AnswerDotAI/fbz
-cargo add fbz --git https://github.com/AnswerDotAI/fbz
+cargo install fbz
+cargo add fbz
 ```
 
 ## CLI
@@ -157,6 +157,27 @@ let plain = fbz::lz4::decompress(&compressed_lz4)?;
 
 It accepts standard independent or linked blocks, stored blocks, all four standard block maxima, optional block/content checksums and sizes, concatenated frames, and skippable frames. External dictionaries and the obsolete legacy frame format are intentionally unsupported.
 
+### Streaming reads
+
+`fbz::Reader` provides a normal `std::io::Read` over bzip2 or gzip files without a preliminary indexing or validation pass:
+
+```rust
+use std::io::{BufReader, Read};
+use fbz::{DecodeOptions, Reader};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let reader = Reader::open("dump.xml.bz2", DecodeOptions::default())?;
+    let mut reader = BufReader::new(reader);
+    let mut header = [0; 4096];
+    reader.read_exact(&mut header)?;
+    Ok(())
+}
+```
+
+Magic takes priority over the filename extension, with the extension used as a fallback for damaged headers. The decoder runs on an owned worker thread and transfers completed decoder allocations through a zero-capacity pipe; it neither materializes the plaintext nor writes an intermediate file. `DecodeOptions` controls decoder threads and speculative memory. Dropping early disconnects the pipe, cancels outstanding work, and joins the worker.
+
+Checksum errors discovered after output has begun are returned by a later `read()` call. Therefore only successful EOF establishes that the complete stream was valid; dropping early deliberately does not finish validation. Compressed tar inputs yield the decoded tar byte stream rather than extracting it. LZ4 and ZIP are not currently exposed through `Reader`: LZ4's frame-layout pass must first become incremental, while ZIP has no single decoded byte stream.
+
 ## Performance
 
 These are single local release-mode CLI runs on the primary Apple Silicon development machine. The gzip comparison warms each executable with the 5% fixture, then measures exactly one full validation run; peak physical footprint comes from a separate sampled run because process inspection can perturb such a short workload. ZIP and tar likewise warm each CLI once and then measure one extraction. The LZ4 comparison uses its automatic four-worker limit. These are observations rather than statistical aggregates. In-process codec and library-oracle comparisons live in [DEV.md](DEV.md), not this user-facing section.

src/bin/fbz.rs

@@ -15,8 +15,8 @@ use std::{
 
 use clap::{ArgGroup, Parser};
 use fbz::{
-    DecodeOptions, DecodeProgress, Error, Index, OutputSink, Source, WriterSink, build_index_with_progress, decode_to_writer_with_progress,
-    decompress_to_sink_with_progress, gzip, lz4,
+    DecodeOptions, DecodeProgress, Error, Format, Index, OutputSink, Source, WriterSink, build_index_with_progress, decode_stream_to_sink_with_progress,
+    decode_to_writer_with_progress, gzip, lz4,
 };
 use serde_json::{Value, json};
 use tempfile::NamedTempFile;
@@ -85,14 +85,6 @@ struct Cli {
     json: bool,
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-enum Format {
-    Bzip2,
-    Gzip,
-    Lz4,
-    Zip,
-}
-
 fn main() -> ExitCode {
     match run(Cli::parse()) {
         Ok(()) => ExitCode::SUCCESS,
@@ -326,12 +318,7 @@ fn decode_data_to_sink(
 ) -> fbz::Result<()> {
     let mut output = LimitedOutput::new(output, max_output);
     let mut display = ProgressDisplay::new(label, data.len() as u64, quiet);
-    match select_format(label, data)? {
-        Format::Bzip2 => decompress_to_sink_with_progress(data, &mut output, options, |progress| display.update(progress)),
-        Format::Gzip => gzip::decompress_to_sink_with_options_and_progress(data, &mut output, options, |progress| display.update(progress)).map(|_| ()),
-        Format::Lz4 => lz4::decompress_to_sink_with_options_and_progress(data, &mut output, options, |progress| display.update(progress)).map(|_| ()),
-        Format::Zip => Err(invalid("ZIP archives extract to a directory and cannot be decoded to one output stream")),
-    }
+    decode_stream_to_sink_with_progress(select_format(label, data)?, data, &mut output, options, |progress| display.update(progress))
 }
 
 fn build_gzip_report_data(data: &[u8], label: &str, options: DecodeOptions, max_output: Option<usize>, quiet: bool) -> fbz::Result<gzip::Report> {
@@ -406,6 +393,10 @@ impl<W: OutputSink> OutputSink for LimitedOutput<W> {
     fn flush(&mut self) -> io::Result<()> {
         self.inner.flush()
     }
+
+    fn is_cancelled(&self) -> bool {
+        self.inner.is_cancelled()
+    }
 }
 
 struct ProgressDisplay {
@@ -540,22 +531,7 @@ fn default_output(input: &Path) -> PathBuf {
 }
 
 fn select_format(input: &str, data: &[u8]) -> fbz::Result<Format> {
-    if let Some((format, _)) = format_extension(Path::new(input)) {
-        return Ok(format);
-    }
-    if data.starts_with(b"BZh") {
-        Ok(Format::Bzip2)
-    } else if data.starts_with(&[0x1f, 0x8b]) {
-        Ok(Format::Gzip)
-    } else if data.starts_with(&[0x04, 0x22, 0x4d, 0x18])
-        || data.get(..4).is_some_and(|magic| (0x184d_2a50..=0x184d_2a5f).contains(&u32::from_le_bytes(magic.try_into().unwrap())))
-    {
-        Ok(Format::Lz4)
-    } else if data.starts_with(b"PK\x03\x04") || data.starts_with(b"PK\x05\x06") || data.starts_with(b"PK\x07\x08") {
-        Ok(Format::Zip)
-    } else {
-        Err(invalid(format!("cannot determine compression format for {input}; expected a bzip2, gzip, LZ4, or ZIP extension or magic")))
-    }
+    Format::detect(input, data)
 }
 
 fn print_index(input: Option<&String>, index: &Index) {
@@ -805,7 +781,13 @@ fn exit_status(error: &Error) -> u8 {
         Error::Io(source) if source.kind() == io::ErrorKind::InvalidData => 3,
         Error::Io(_) => 1,
         Error::InvalidConfiguration(_) => 2,
-        Error::InvalidStreamHeader | Error::InvalidGzip(_) | Error::InvalidLz4(_) | Error::InvalidZip(_) | Error::Decode { .. } | Error::InvalidIndex(_) => 3,
+        Error::InvalidStreamHeader
+        | Error::InvalidGzip(_)
+        | Error::InvalidLz4(_)
+        | Error::InvalidZip(_)
+        | Error::UnsupportedFormat(_)
+        | Error::Decode { .. }
+        | Error::InvalidIndex(_) => 3,
         _ => 4,
     }
 }

src/bin/fbz/tar_extract.rs

@@ -1,76 +1,9 @@
-use std::{
-    cmp,
-    io::{self, Read},
-    path::Path,
-    sync::mpsc::{Receiver, SyncSender, sync_channel},
-    thread,
-};
+use std::{io, path::Path, thread};
 
-use fbz::{Error, OutputSink, Result};
+use fbz::{Error, PipeWriter, Result, output_pipe};
 
 use super::archive_extract;
 
-struct Chunk {
-    bytes: Vec<u8>,
-    offset: usize,
-}
-
-pub(super) struct PipeWriter {
-    sender: SyncSender<Chunk>,
-}
-
-impl PipeWriter {
-    fn send(&self, bytes: Vec<u8>, offset: usize) -> io::Result<()> {
-        let suffix = bytes.get(offset..).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?;
-        if suffix.is_empty() {
-            return Ok(());
-        }
-        self.sender.send(Chunk { bytes, offset }).map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "tar extractor stopped reading"))
-    }
-}
-
-impl OutputSink for PipeWriter {
-    fn write_borrowed(&mut self, buffer: &[u8]) -> io::Result<()> {
-        self.send(buffer.to_vec(), 0)
-    }
-
-    fn write_owned_from(&mut self, buffer: Vec<u8>, start: usize) -> io::Result<()> {
-        self.send(buffer, start)
-    }
-}
-
-struct PipeReader {
-    receiver: Receiver<Chunk>,
-    chunk: Vec<u8>,
-    offset: usize,
-}
-
-impl Read for PipeReader {
-    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
-        if buffer.is_empty() {
-            return Ok(0);
-        }
-        if self.offset == self.chunk.len() {
-            match self.receiver.recv() {
-                Ok(chunk) => {
-                    self.chunk = chunk.bytes;
-                    self.offset = chunk.offset;
-                }
-                Err(_) => return Ok(0),
-            }
-        }
-        let count = cmp::min(buffer.len(), self.chunk.len() - self.offset);
-        buffer[..count].copy_from_slice(&self.chunk[self.offset..self.offset + count]);
-        self.offset += count;
-        Ok(count)
-    }
-}
-
-fn pipe() -> (PipeWriter, PipeReader) {
-    let (sender, receiver) = sync_channel(0);
-    (PipeWriter { sender }, PipeReader { receiver, chunk: Vec::new(), offset: 0 })
-}
-
 fn broken_pipe(error: &Error) -> bool {
     matches!(error, Error::Io(source) if source.kind() == io::ErrorKind::BrokenPipe)
 }
@@ -81,7 +14,7 @@ where
 {
     let staging = archive_extract::staging(destination)?;
     thread::scope(|scope| {
-        let (mut writer, mut reader) = pipe();
+        let (mut writer, mut reader) = output_pipe();
         let decoder = scope.spawn(move || decode(&mut writer));
         let extracted = {
             let mut archive = tar::Archive::new(&mut reader);

src/bin/fbz/zip_extract.rs

@@ -210,6 +210,10 @@ impl<W: OutputSink> OutputSink for ExpectedOutput<W> {
     fn flush(&mut self) -> io::Result<()> {
         self.inner.flush()
     }
+
+    fn is_cancelled(&self) -> bool {
+        self.inner.is_cancelled()
+    }
 }
 
 fn entry_error(entry: &Entry, error: Error) -> Error {