Add portable bzip2 format scanning primitives

DEV.md

@@ -2,14 +2,19 @@
 
 `fastbz2` is a mixed Rust/PyO3 project. The Rust crate is the implementation and public Rust API; `python/fastbz2/` is the public Python package over the private `fastbz2._core` extension.
 
-## Initial architecture
+## Architecture
 
 ```text
-src/                  portable decoder, scanner, index, scheduler, CLI support
+src/bitreader.rs      bounded MSB-first in-memory bit reads
+src/crc.rs            bzip2 block and combined-stream CRC primitives
+src/format.rs         cheap structural scan for header and marker candidates
+src/lib.rs            public Rust API and private PyO3 binding
 python/fastbz2/       thin Python I/O wrapper over fastbz2._core
 tests/                Python API and integration tests
 ```
 
+The current scanner deliberately does not treat 48-bit marker matches or later `BZh` headers as validated structure. Full decoding must establish the exact block chain and validate every block CRC plus the combined stream CRC before marker candidates can become trusted index entries. Python integration tests use standard-library `bz2`/libbz2 as an independent fixture generator.
+
 The decoder remains independent of files, threads, Python, and the CLI. Parallel scanning/decoding and indexed seeking are layered over it. Native workers never call Python. Large offsets use explicit 64-bit bit/byte types, and speculative block-marker hits are accepted only when they form an exact stream chain with valid block and combined stream CRCs.
 
 Start with safe scalar Rust designed for LLVM auto-vectorisation. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling, with the safe implementation retained as a differential oracle.

README.md

@@ -6,7 +6,16 @@ Fast parallel and indexed bzip2 decompression for Rust and Python.
 
 The first performance target is end-to-end throughput within 20% of `librapidarchive`'s `indexed_bzip2` on the same host and input. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it.
 
-The decoder is not implemented yet.
+The first implemented layer is a safe, portable MSB-first bit reader, bzip2 CRC primitives, and a structural scanner for stream headers and non-byte-aligned block/end markers. The scanner reports candidates rather than claiming validation: the decoder, exact stream-chain validation, indexing, parallel scheduler, CLI, and seekable Python file API are not implemented yet.
+
+```python
+import bz2
+from fastbz2 import scan
+
+result = scan(bz2.compress(b"hello"))
+result.blocks[0].bit_offset
+# 32
+```
 
 ## Inspiration and credit
 

python/fastbz2/__init__.py

@@ -1,3 +1,24 @@
-from ._core import __version__, hello
+from collections import namedtuple
 
-__all__ = ["__version__", "hello"]
+from ._core import __version__, _scan, bz2_crc32
+
+StreamHeaderCandidate = namedtuple("StreamHeaderCandidate", "byte_offset block_size_100k")
+BlockCandidate = namedtuple("BlockCandidate", "bit_offset expected_crc randomized orig_ptr")
+EndCandidate = namedtuple("EndCandidate", "bit_offset expected_stream_crc")
+ScanResult = namedtuple("ScanResult", "streams blocks stream_ends")
+
+def scan(data: bytes) -> ScanResult:
+    """Find candidate stream headers and bit-level block markers in *data*.
+
+    This is a structural scan rather than full validation. Marker patterns after
+    the first header remain candidates until a decoder validates the stream.
+    """
+    streams, blocks, stream_ends = _scan(data)
+    streams = [StreamHeaderCandidate(*item) for item in streams]
+    blocks = [BlockCandidate(*item) for item in blocks]
+    stream_ends = [EndCandidate(*item) for item in stream_ends]
+    return ScanResult(streams, blocks, stream_ends)
+
+__all__ = [
+    "__version__", "BlockCandidate", "EndCandidate", "ScanResult", "StreamHeaderCandidate", "bz2_crc32", "scan"
+]

src/bitreader.rs

@@ -0,0 +1,117 @@
+use crate::{Error, Result};
+
+/// An MSB-first reader over an in-memory bzip2 bitstream.
+#[derive(Clone, Debug)]
+pub struct BitReader<'a> {
+    data: &'a [u8],
+    bit_offset: u64,
+}
+
+impl<'a> BitReader<'a> {
+    pub fn new(data: &'a [u8]) -> Self {
+        Self { data, bit_offset: 0 }
+    }
+
+    pub fn at(data: &'a [u8], bit_offset: u64) -> Result<Self> {
+        let reader = Self { data, bit_offset };
+        if bit_offset > reader.len_bits() {
+            return Err(Error::InvalidBitOffset { bit_offset, len_bits: reader.len_bits() });
+        }
+        Ok(reader)
+    }
+
+    #[inline]
+    pub fn position(&self) -> u64 {
+        self.bit_offset
+    }
+
+    #[inline]
+    pub fn len_bits(&self) -> u64 {
+        u64::try_from(self.data.len()).unwrap_or(u64::MAX / 8).saturating_mul(8)
+    }
+
+    #[inline]
+    pub fn remaining(&self) -> u64 {
+        self.len_bits().saturating_sub(self.bit_offset)
+    }
+
+    #[inline]
+    pub fn read_bit(&mut self) -> Result<bool> {
+        if self.bit_offset >= self.len_bits() {
+            return Err(self.eof(1));
+        }
+        let byte = self.data[(self.bit_offset / 8) as usize];
+        let shift = 7 - (self.bit_offset & 7);
+        self.bit_offset += 1;
+        Ok((byte >> shift) & 1 != 0)
+    }
+
+    /// Read up to 64 bits, returning them in the low bits of a `u64`.
+    #[inline]
+    pub fn read_bits(&mut self, count: u8) -> Result<u64> {
+        if count > 64 {
+            return Err(Error::InvalidBitCount(count));
+        }
+        if u64::from(count) > self.remaining() {
+            return Err(self.eof(u64::from(count)));
+        }
+        if count == 0 {
+            return Ok(0);
+        }
+
+        let first_byte = (self.bit_offset / 8) as usize;
+        let skipped = (self.bit_offset & 7) as u32;
+        let byte_count = (skipped + u32::from(count)).div_ceil(8) as usize;
+        let mut word = 0_u128;
+        for &byte in &self.data[first_byte..first_byte + byte_count] {
+            word = (word << 8) | u128::from(byte);
+        }
+        let available = (byte_count * 8) as u32;
+        let trailing = available - skipped - u32::from(count);
+        let mask = if count == 64 { u128::from(u64::MAX) } else { (1_u128 << count) - 1 };
+        self.bit_offset += u64::from(count);
+        Ok(((word >> trailing) & mask) as u64)
+    }
+
+    #[inline]
+    pub fn skip(&mut self, count: u64) -> Result<()> {
+        if count > self.remaining() {
+            return Err(self.eof(count));
+        }
+        self.bit_offset += count;
+        Ok(())
+    }
+
+    fn eof(&self, requested: u64) -> Error {
+        Error::UnexpectedEof { bit_offset: self.bit_offset, requested, remaining: self.remaining() }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn reference(data: &[u8], offset: usize, count: usize) -> u64 {
+        (offset..offset + count).fold(0, |result, bit| (result << 1) | u64::from((data[bit / 8] >> (7 - bit % 8)) & 1))
+    }
+
+    #[test]
+    fn reads_every_alignment_and_width() {
+        let data = [0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc, 0xde, 0xf0, 0x55];
+        for offset in 0..8 {
+            for count in 0..=64.min(data.len() * 8 - offset) {
+                let mut reader = BitReader::at(&data, offset as u64).unwrap();
+                assert_eq!(reader.read_bits(count as u8).unwrap(), reference(&data, offset, count));
+                assert_eq!(reader.position(), (offset + count) as u64);
+            }
+        }
+    }
+
+    #[test]
+    fn reads_single_bits_msb_first() {
+        let mut reader = BitReader::new(&[0b1010_0001]);
+        let bits: Vec<_> = (0..8).map(|_| reader.read_bit().unwrap()).collect();
+        assert_eq!(bits, [true, false, true, false, false, false, false, true]);
+        assert!(matches!(reader.read_bit(), Err(Error::UnexpectedEof { .. })));
+    }
+}

src/crc.rs

@@ -0,0 +1,51 @@
+const POLYNOMIAL: u32 = 0x04c1_1db7;
+
+const fn make_table() -> [u32; 256] {
+    let mut table = [0; 256];
+    let mut byte = 0;
+    while byte < 256 {
+        let mut crc = (byte as u32) << 24;
+        let mut bit = 0;
+        while bit < 8 {
+            crc = if crc & 0x8000_0000 != 0 { (crc << 1) ^ POLYNOMIAL } else { crc << 1 };
+            bit += 1;
+        }
+        table[byte] = crc;
+        byte += 1;
+    }
+    table
+}
+
+const TABLE: [u32; 256] = make_table();
+
+/// Compute the CRC used for an uncompressed bzip2 block.
+pub fn bz2_crc32(data: &[u8]) -> u32 {
+    let mut crc = u32::MAX;
+    for &byte in data {
+        let index = ((crc >> 24) as u8 ^ byte) as usize;
+        crc = (crc << 8) ^ TABLE[index];
+    }
+    !crc
+}
+
+/// Add a block CRC to bzip2's combined stream CRC.
+#[inline]
+pub fn combine_stream_crc(stream_crc: u32, block_crc: u32) -> u32 {
+    stream_crc.rotate_left(1) ^ block_crc
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn known_values() {
+        assert_eq!(bz2_crc32(b""), 0);
+        assert_eq!(bz2_crc32(b"123456789"), 0xfc89_1918);
+    }
+
+    #[test]
+    fn combines_by_rotating_then_xoring() {
+        assert_eq!(combine_stream_crc(0x8000_0001, 0x1234_5678), 0x1234_567b);
+    }
+}

src/error.rs

@@ -0,0 +1,28 @@
+use std::{error, fmt};
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Error {
+    InvalidBitCount(u8),
+    InvalidBitOffset { bit_offset: u64, len_bits: u64 },
+    UnexpectedEof { bit_offset: u64, requested: u64, remaining: u64 },
+    InvalidStreamHeader,
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::InvalidBitCount(count) => write!(f, "cannot read {count} bits at once; maximum is 64"),
+            Self::InvalidBitOffset { bit_offset, len_bits } => {
+                write!(f, "bit offset {bit_offset} is beyond the {len_bits}-bit input")
+            }
+            Self::UnexpectedEof { bit_offset, requested, remaining } => {
+                write!(f, "unexpected end of input at bit {bit_offset}: requested {requested} bits, {remaining} remain")
+            }
+            Self::InvalidStreamHeader => write!(f, "input does not start with a bzip2 BZh1-BZh9 header"),
+        }
+    }
+}
+
+impl error::Error for Error {}