ET + SM + Crypto Fixes

src/stream/buffer_pool.rs

@@ -0,0 +1,450 @@
+//! Reusable buffer pool to avoid allocations in hot paths
+//!
+//! This module provides a thread-safe pool of BytesMut buffers
+//! that can be reused across connections to reduce allocation pressure.
+
+use bytes::BytesMut;
+use crossbeam_queue::ArrayQueue;
+use std::ops::{Deref, DerefMut};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+// ============= Configuration =============
+
+/// Default buffer size (64KB - good for MTProto)
+pub const DEFAULT_BUFFER_SIZE: usize = 64 * 1024;
+
+/// Default maximum number of pooled buffers
+pub const DEFAULT_MAX_BUFFERS: usize = 1024;
+
+// ============= Buffer Pool =============
+
+/// Thread-safe pool of reusable buffers
+pub struct BufferPool {
+    /// Queue of available buffers
+    buffers: ArrayQueue<BytesMut>,
+    /// Size of each buffer
+    buffer_size: usize,
+    /// Maximum number of buffers to pool
+    max_buffers: usize,
+    /// Total allocated buffers (including in-use)
+    allocated: AtomicUsize,
+    /// Number of times we had to create a new buffer
+    misses: AtomicUsize,
+    /// Number of successful reuses
+    hits: AtomicUsize,
+}
+
+impl BufferPool {
+    /// Create a new buffer pool with default settings
+    pub fn new() -> Self {
+        Self::with_config(DEFAULT_BUFFER_SIZE, DEFAULT_MAX_BUFFERS)
+    }
+    
+    /// Create a buffer pool with custom configuration
+    pub fn with_config(buffer_size: usize, max_buffers: usize) -> Self {
+        Self {
+            buffers: ArrayQueue::new(max_buffers),
+            buffer_size,
+            max_buffers,
+            allocated: AtomicUsize::new(0),
+            misses: AtomicUsize::new(0),
+            hits: AtomicUsize::new(0),
+        }
+    }
+    
+    /// Get a buffer from the pool, or create a new one if empty
+    pub fn get(self: &Arc<Self>) -> PooledBuffer {
+        match self.buffers.pop() {
+            Some(mut buffer) => {
+                self.hits.fetch_add(1, Ordering::Relaxed);
+                buffer.clear();
+                PooledBuffer {
+                    buffer: Some(buffer),
+                    pool: Arc::clone(self),
+                }
+            }
+            None => {
+                self.misses.fetch_add(1, Ordering::Relaxed);
+                self.allocated.fetch_add(1, Ordering::Relaxed);
+                PooledBuffer {
+                    buffer: Some(BytesMut::with_capacity(self.buffer_size)),
+                    pool: Arc::clone(self),
+                }
+            }
+        }
+    }
+    
+    /// Try to get a buffer, returns None if pool is empty
+    pub fn try_get(self: &Arc<Self>) -> Option<PooledBuffer> {
+        self.buffers.pop().map(|mut buffer| {
+            self.hits.fetch_add(1, Ordering::Relaxed);
+            buffer.clear();
+            PooledBuffer {
+                buffer: Some(buffer),
+                pool: Arc::clone(self),
+            }
+        })
+    }
+    
+    /// Return a buffer to the pool
+    fn return_buffer(&self, mut buffer: BytesMut) {
+        // Clear the buffer but keep capacity
+        buffer.clear();
+        
+        // Only return if we haven't exceeded max and buffer is right size
+        if buffer.capacity() >= self.buffer_size {
+            // Try to push to pool, if full just drop
+            let _ = self.buffers.push(buffer);
+        }
+        // If buffer was dropped (pool full), decrement allocated
+        // Actually we don't decrement here because the buffer might have been
+        // grown beyond our size - we just let it go
+    }
+    
+    /// Get pool statistics
+    pub fn stats(&self) -> PoolStats {
+        PoolStats {
+            pooled: self.buffers.len(),
+            allocated: self.allocated.load(Ordering::Relaxed),
+            max_buffers: self.max_buffers,
+            buffer_size: self.buffer_size,
+            hits: self.hits.load(Ordering::Relaxed),
+            misses: self.misses.load(Ordering::Relaxed),
+        }
+    }
+    
+    /// Get buffer size
+    pub fn buffer_size(&self) -> usize {
+        self.buffer_size
+    }
+    
+    /// Preallocate buffers to fill the pool
+    pub fn preallocate(&self, count: usize) {
+        let to_alloc = count.min(self.max_buffers);
+        for _ in 0..to_alloc {
+            if self.buffers.push(BytesMut::with_capacity(self.buffer_size)).is_err() {
+                break;
+            }
+            self.allocated.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+}
+
+impl Default for BufferPool {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// ============= Pool Statistics =============
+
+/// Statistics about buffer pool usage
+#[derive(Debug, Clone)]
+pub struct PoolStats {
+    /// Current number of buffers in pool
+    pub pooled: usize,
+    /// Total buffers allocated (in-use + pooled)
+    pub allocated: usize,
+    /// Maximum buffers allowed
+    pub max_buffers: usize,
+    /// Size of each buffer
+    pub buffer_size: usize,
+    /// Number of cache hits (reused buffer)
+    pub hits: usize,
+    /// Number of cache misses (new allocation)
+    pub misses: usize,
+}
+
+impl PoolStats {
+    /// Get hit rate as percentage
+    pub fn hit_rate(&self) -> f64 {
+        let total = self.hits + self.misses;
+        if total == 0 {
+            0.0
+        } else {
+            (self.hits as f64 / total as f64) * 100.0
+        }
+    }
+}
+
+// ============= Pooled Buffer =============
+
+/// A buffer that automatically returns to the pool when dropped
+pub struct PooledBuffer {
+    buffer: Option<BytesMut>,
+    pool: Arc<BufferPool>,
+}
+
+impl PooledBuffer {
+    /// Take the inner buffer, preventing return to pool
+    pub fn take(mut self) -> BytesMut {
+        self.buffer.take().unwrap()
+    }
+    
+    /// Get the capacity of the buffer
+    pub fn capacity(&self) -> usize {
+        self.buffer.as_ref().map(|b| b.capacity()).unwrap_or(0)
+    }
+    
+    /// Check if buffer is empty
+    pub fn is_empty(&self) -> bool {
+        self.buffer.as_ref().map(|b| b.is_empty()).unwrap_or(true)
+    }
+    
+    /// Get the length of data in buffer
+    pub fn len(&self) -> usize {
+        self.buffer.as_ref().map(|b| b.len()).unwrap_or(0)
+    }
+    
+    /// Clear the buffer
+    pub fn clear(&mut self) {
+        if let Some(ref mut b) = self.buffer {
+            b.clear();
+        }
+    }
+}
+
+impl Deref for PooledBuffer {
+    type Target = BytesMut;
+    
+    fn deref(&self) -> &Self::Target {
+        self.buffer.as_ref().expect("buffer taken")
+    }
+}
+
+impl DerefMut for PooledBuffer {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.buffer.as_mut().expect("buffer taken")
+    }
+}
+
+impl Drop for PooledBuffer {
+    fn drop(&mut self) {
+        if let Some(buffer) = self.buffer.take() {
+            self.pool.return_buffer(buffer);
+        }
+    }
+}
+
+impl AsRef<[u8]> for PooledBuffer {
+    fn as_ref(&self) -> &[u8] {
+        self.buffer.as_ref().map(|b| b.as_ref()).unwrap_or(&[])
+    }
+}
+
+impl AsMut<[u8]> for PooledBuffer {
+    fn as_mut(&mut self) -> &mut [u8] {
+        self.buffer.as_mut().map(|b| b.as_mut()).unwrap_or(&mut [])
+    }
+}
+
+// ============= Scoped Buffer =============
+
+/// A buffer that can be used for a scoped operation
+/// Useful for ensuring buffer is returned even on early return
+pub struct ScopedBuffer<'a> {
+    buffer: &'a mut PooledBuffer,
+}
+
+impl<'a> ScopedBuffer<'a> {
+    /// Create a new scoped buffer
+    pub fn new(buffer: &'a mut PooledBuffer) -> Self {
+        buffer.clear();
+        Self { buffer }
+    }
+}
+
+impl<'a> Deref for ScopedBuffer<'a> {
+    type Target = BytesMut;
+    
+    fn deref(&self) -> &Self::Target {
+        self.buffer.deref()
+    }
+}
+
+impl<'a> DerefMut for ScopedBuffer<'a> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.buffer.deref_mut()
+    }
+}
+
+impl<'a> Drop for ScopedBuffer<'a> {
+    fn drop(&mut self) {
+        self.buffer.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    
+    #[test]
+    fn test_pool_basic() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        
+        // Get a buffer
+        let mut buf1 = pool.get();
+        buf1.extend_from_slice(b"hello");
+        assert_eq!(&buf1[..], b"hello");
+        
+        // Drop returns to pool
+        drop(buf1);
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 1);
+        assert_eq!(stats.hits, 0);
+        assert_eq!(stats.misses, 1);
+        
+        // Get again - should reuse
+        let buf2 = pool.get();
+        assert!(buf2.is_empty()); // Buffer was cleared
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 0);
+        assert_eq!(stats.hits, 1);
+    }
+    
+    #[test]
+    fn test_pool_multiple_buffers() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        
+        // Get multiple buffers
+        let buf1 = pool.get();
+        let buf2 = pool.get();
+        let buf3 = pool.get();
+        
+        let stats = pool.stats();
+        assert_eq!(stats.allocated, 3);
+        assert_eq!(stats.pooled, 0);
+        
+        // Return all
+        drop(buf1);
+        drop(buf2);
+        drop(buf3);
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 3);
+    }
+    
+    #[test]
+    fn test_pool_overflow() {
+        let pool = Arc::new(BufferPool::with_config(1024, 2));
+        
+        // Get 3 buffers (more than max)
+        let buf1 = pool.get();
+        let buf2 = pool.get();
+        let buf3 = pool.get();
+        
+        // Return all - only 2 should be pooled
+        drop(buf1);
+        drop(buf2);
+        drop(buf3);
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 2);
+    }
+    
+    #[test]
+    fn test_pool_take() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        
+        let mut buf = pool.get();
+        buf.extend_from_slice(b"data");
+        
+        // Take ownership, buffer should not return to pool
+        let taken = buf.take();
+        assert_eq!(&taken[..], b"data");
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 0);
+    }
+    
+    #[test]
+    fn test_pool_preallocate() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        pool.preallocate(5);
+        
+        let stats = pool.stats();
+        assert_eq!(stats.pooled, 5);
+        assert_eq!(stats.allocated, 5);
+    }
+    
+    #[test]
+    fn test_pool_try_get() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        
+        // Pool is empty, try_get returns None
+        assert!(pool.try_get().is_none());
+        
+        // Add a buffer to pool
+        pool.preallocate(1);
+        
+        // Now try_get should succeed
+        assert!(pool.try_get().is_some());
+        assert!(pool.try_get().is_none());
+    }
+    
+    #[test]
+    fn test_hit_rate() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        
+        // First get is a miss
+        let buf1 = pool.get();
+        drop(buf1);
+        
+        // Second get is a hit
+        let buf2 = pool.get();
+        drop(buf2);
+        
+        // Third get is a hit
+        let _buf3 = pool.get();
+        
+        let stats = pool.stats();
+        assert_eq!(stats.hits, 2);
+        assert_eq!(stats.misses, 1);
+        assert!((stats.hit_rate() - 66.67).abs() < 1.0);
+    }
+    
+    #[test]
+    fn test_scoped_buffer() {
+        let pool = Arc::new(BufferPool::with_config(1024, 10));
+        let mut buf = pool.get();
+        
+        {
+            let mut scoped = ScopedBuffer::new(&mut buf);
+            scoped.extend_from_slice(b"scoped data");
+            assert_eq!(&scoped[..], b"scoped data");
+        }
+        
+        // After scoped is dropped, buffer is cleared
+        assert!(buf.is_empty());
+    }
+    
+    #[test]
+    fn test_concurrent_access() {
+        use std::thread;
+        
+        let pool = Arc::new(BufferPool::with_config(1024, 100));
+        let mut handles = vec![];
+        
+        for _ in 0..10 {
+            let pool_clone = Arc::clone(&pool);
+            handles.push(thread::spawn(move || {
+                for _ in 0..100 {
+                    let mut buf = pool_clone.get();
+                    buf.extend_from_slice(b"test");
+                    // buf auto-returned on drop
+                }
+            }));
+        }
+        
+        for handle in handles {
+            handle.join().unwrap();
+        }
+        
+        let stats = pool.stats();
+        // All buffers should be returned
+        assert!(stats.pooled > 0);
+    }
+}

src/stream/mod.rs

@@ -1,10 +1,22 @@
 //! Stream wrappers for MTProto protocol layers
 
+pub mod state;
+pub mod buffer_pool;
 pub mod traits;
 pub mod crypto_stream;
 pub mod tls_stream;
 pub mod frame_stream;
 
+// Re-export state machine types
+pub use state::{
+    StreamState, Transition, PollResult,
+    ReadBuffer, WriteBuffer, HeaderBuffer, YieldBuffer,
+};
+
+// Re-export buffer pool
+pub use buffer_pool::{BufferPool, PooledBuffer, PoolStats};
+
+// Re-export stream implementations
 pub use crypto_stream::{CryptoReader, CryptoWriter, PassthroughStream};
 pub use tls_stream::{FakeTlsReader, FakeTlsWriter};
 pub use frame_stream::*;

src/stream/state.rs

@@ -0,0 +1,571 @@
+//! State machine foundation types for async streams
+//!
+//! This module provides core types and traits for implementing
+//! stateful async streams with proper partial read/write handling.
+
+use bytes::{Bytes, BytesMut};
+use std::io;
+
+// ============= Core Traits =============
+
+/// Trait for stream states
+pub trait StreamState: Sized {
+    /// Check if this is a terminal state (no more transitions possible)
+    fn is_terminal(&self) -> bool;
+    
+    /// Check if stream is in poisoned/error state
+    fn is_poisoned(&self) -> bool;
+    
+    /// Get human-readable state name for debugging
+    fn state_name(&self) -> &'static str;
+}
+
+// ============= Transition Types =============
+
+/// Result of a state transition
+#[derive(Debug)]
+pub enum Transition<S, O> {
+    /// Stay in the same state, no output
+    Same,
+    /// Transition to a new state, no output
+    Next(S),
+    /// Complete with output, typically transitions to Idle
+    Complete(O),
+    /// Yield output and transition to new state
+    Yield(O, S),
+    /// Error occurred, transition to error state
+    Error(io::Error),
+}
+
+impl<S, O> Transition<S, O> {
+    /// Check if transition produces output
+    pub fn has_output(&self) -> bool {
+        matches!(self, Transition::Complete(_) | Transition::Yield(_, _))
+    }
+    
+    /// Map the output value
+    pub fn map_output<U, F: FnOnce(O) -> U>(self, f: F) -> Transition<S, U> {
+        match self {
+            Transition::Same => Transition::Same,
+            Transition::Next(s) => Transition::Next(s),
+            Transition::Complete(o) => Transition::Complete(f(o)),
+            Transition::Yield(o, s) => Transition::Yield(f(o), s),
+            Transition::Error(e) => Transition::Error(e),
+        }
+    }
+    
+    /// Map the state value
+    pub fn map_state<T, F: FnOnce(S) -> T>(self, f: F) -> Transition<T, O> {
+        match self {
+            Transition::Same => Transition::Same,
+            Transition::Next(s) => Transition::Next(f(s)),
+            Transition::Complete(o) => Transition::Complete(o),
+            Transition::Yield(o, s) => Transition::Yield(o, f(s)),
+            Transition::Error(e) => Transition::Error(e),
+        }
+    }
+}
+
+// ============= Poll Result Types =============
+
+/// Result of polling for more data
+#[derive(Debug)]
+pub enum PollResult<T> {
+    /// Data is ready
+    Ready(T),
+    /// Operation would block, need to poll again
+    Pending,
+    /// Need more input data (minimum bytes required)
+    NeedInput(usize),
+    /// End of stream reached
+    Eof,
+    /// Error occurred
+    Error(io::Error),
+}
+
+impl<T> PollResult<T> {
+    /// Check if result is ready
+    pub fn is_ready(&self) -> bool {
+        matches!(self, PollResult::Ready(_))
+    }
+    
+    /// Check if result indicates EOF
+    pub fn is_eof(&self) -> bool {
+        matches!(self, PollResult::Eof)
+    }
+    
+    /// Convert to Option, discarding non-ready states
+    pub fn ok(self) -> Option<T> {
+        match self {
+            PollResult::Ready(t) => Some(t),
+            _ => None,
+        }
+    }
+    
+    /// Map the value
+    pub fn map<U, F: FnOnce(T) -> U>(self, f: F) -> PollResult<U> {
+        match self {
+            PollResult::Ready(t) => PollResult::Ready(f(t)),
+            PollResult::Pending => PollResult::Pending,
+            PollResult::NeedInput(n) => PollResult::NeedInput(n),
+            PollResult::Eof => PollResult::Eof,
+            PollResult::Error(e) => PollResult::Error(e),
+        }
+    }
+}
+
+impl<T> From<io::Result<T>> for PollResult<T> {
+    fn from(result: io::Result<T>) -> Self {
+        match result {
+            Ok(t) => PollResult::Ready(t),
+            Err(e) if e.kind() == io::ErrorKind::WouldBlock => PollResult::Pending,
+            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => PollResult::Eof,
+            Err(e) => PollResult::Error(e),
+        }
+    }
+}
+
+// ============= Buffer State =============
+
+/// State for buffered reading operations
+#[derive(Debug)]
+pub struct ReadBuffer {
+    /// The buffer holding data
+    buffer: BytesMut,
+    /// Target number of bytes to read (if known)
+    target: Option<usize>,
+}
+
+impl ReadBuffer {
+    /// Create new empty read buffer
+    pub fn new() -> Self {
+        Self {
+            buffer: BytesMut::with_capacity(8192),
+            target: None,
+        }
+    }
+    
+    /// Create with specific capacity
+    pub fn with_capacity(capacity: usize) -> Self {
+        Self {
+            buffer: BytesMut::with_capacity(capacity),
+            target: None,
+        }
+    }
+    
+    /// Create with target size
+    pub fn with_target(target: usize) -> Self {
+        Self {
+            buffer: BytesMut::with_capacity(target),
+            target: Some(target),
+        }
+    }
+    
+    /// Get current buffer length
+    pub fn len(&self) -> usize {
+        self.buffer.len()
+    }
+    
+    /// Check if buffer is empty
+    pub fn is_empty(&self) -> bool {
+        self.buffer.is_empty()
+    }
+    
+    /// Check if target is reached
+    pub fn is_complete(&self) -> bool {
+        match self.target {
+            Some(t) => self.buffer.len() >= t,
+            None => false,
+        }
+    }
+    
+    /// Get remaining bytes needed
+    pub fn remaining(&self) -> usize {
+        match self.target {
+            Some(t) => t.saturating_sub(self.buffer.len()),
+            None => 0,
+        }
+    }
+    
+    /// Append data to buffer
+    pub fn extend(&mut self, data: &[u8]) {
+        self.buffer.extend_from_slice(data);
+    }
+    
+    /// Take all data from buffer
+    pub fn take(&mut self) -> Bytes {
+        self.target = None;
+        self.buffer.split().freeze()
+    }
+    
+    /// Take exactly n bytes
+    pub fn take_exact(&mut self, n: usize) -> Option<Bytes> {
+        if self.buffer.len() >= n {
+            Some(self.buffer.split_to(n).freeze())
+        } else {
+            None
+        }
+    }
+    
+    /// Get a slice of the buffer
+    pub fn as_slice(&self) -> &[u8] {
+        &self.buffer
+    }
+    
+    /// Get mutable access to underlying BytesMut
+    pub fn as_bytes_mut(&mut self) -> &mut BytesMut {
+        &mut self.buffer
+    }
+    
+    /// Clear the buffer
+    pub fn clear(&mut self) {
+        self.buffer.clear();
+        self.target = None;
+    }
+    
+    /// Set new target
+    pub fn set_target(&mut self, target: usize) {
+        self.target = Some(target);
+    }
+}
+
+impl Default for ReadBuffer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// State for buffered writing operations
+#[derive(Debug)]
+pub struct WriteBuffer {
+    /// The buffer holding data to write
+    buffer: BytesMut,
+    /// Position of next byte to write
+    position: usize,
+    /// Maximum buffer size
+    max_size: usize,
+}
+
+impl WriteBuffer {
+    /// Create new write buffer with default max size (256KB)
+    pub fn new() -> Self {
+        Self::with_max_size(256 * 1024)
+    }
+    
+    /// Create with specific max size
+    pub fn with_max_size(max_size: usize) -> Self {
+        Self {
+            buffer: BytesMut::with_capacity(8192),
+            position: 0,
+            max_size,
+        }
+    }
+    
+    /// Get pending bytes count
+    pub fn len(&self) -> usize {
+        self.buffer.len() - self.position
+    }
+    
+    /// Check if buffer is empty (all written)
+    pub fn is_empty(&self) -> bool {
+        self.position >= self.buffer.len()
+    }
+    
+    /// Check if buffer is full
+    pub fn is_full(&self) -> bool {
+        self.buffer.len() >= self.max_size
+    }
+    
+    /// Get remaining capacity
+    pub fn remaining_capacity(&self) -> usize {
+        self.max_size.saturating_sub(self.buffer.len())
+    }
+    
+    /// Append data to buffer
+    pub fn extend(&mut self, data: &[u8]) -> Result<(), ()> {
+        if self.buffer.len() + data.len() > self.max_size {
+            return Err(());
+        }
+        self.buffer.extend_from_slice(data);
+        Ok(())
+    }
+    
+    /// Get slice of data to write
+    pub fn pending(&self) -> &[u8] {
+        &self.buffer[self.position..]
+    }
+    
+    /// Advance position by n bytes (after successful write)
+    pub fn advance(&mut self, n: usize) {
+        self.position += n;
+        
+        // If all data written, reset buffer
+        if self.position >= self.buffer.len() {
+            self.buffer.clear();
+            self.position = 0;
+        }
+    }
+    
+    /// Clear the buffer
+    pub fn clear(&mut self) {
+        self.buffer.clear();
+        self.position = 0;
+    }
+}
+
+impl Default for WriteBuffer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// ============= Fixed-Size Buffer States =============
+
+/// State for reading a fixed-size header
+#[derive(Debug, Clone)]
+pub struct HeaderBuffer<const N: usize> {
+    /// The buffer
+    data: [u8; N],
+    /// Bytes filled so far
+    filled: usize,
+}
+
+impl<const N: usize> HeaderBuffer<N> {
+    /// Create new empty header buffer
+    pub fn new() -> Self {
+        Self {
+            data: [0u8; N],
+            filled: 0,
+        }
+    }
+    
+    /// Get slice for reading into
+    pub fn unfilled_mut(&mut self) -> &mut [u8] {
+        &mut self.data[self.filled..]
+    }
+    
+    /// Advance filled count
+    pub fn advance(&mut self, n: usize) {
+        self.filled = (self.filled + n).min(N);
+    }
+    
+    /// Check if completely filled
+    pub fn is_complete(&self) -> bool {
+        self.filled >= N
+    }
+    
+    /// Get remaining bytes needed
+    pub fn remaining(&self) -> usize {
+        N - self.filled
+    }
+    
+    /// Get filled bytes as slice
+    pub fn as_slice(&self) -> &[u8] {
+        &self.data[..self.filled]
+    }
+    
+    /// Get complete buffer (panics if not complete)
+    pub fn as_array(&self) -> &[u8; N] {
+        assert!(self.is_complete());
+        &self.data
+    }
+    
+    /// Take the buffer, resetting state
+    pub fn take(&mut self) -> [u8; N] {
+        let data = self.data;
+        self.data = [0u8; N];
+        self.filled = 0;
+        data
+    }
+    
+    /// Reset to empty state
+    pub fn reset(&mut self) {
+        self.filled = 0;
+    }
+}
+
+impl<const N: usize> Default for HeaderBuffer<N> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// ============= Yield Buffer =============
+
+/// Buffer for yielding data to caller in chunks
+#[derive(Debug)]
+pub struct YieldBuffer {
+    data: Bytes,
+    position: usize,
+}
+
+impl YieldBuffer {
+    /// Create new yield buffer
+    pub fn new(data: Bytes) -> Self {
+        Self { data, position: 0 }
+    }
+    
+    /// Check if all data has been yielded
+    pub fn is_empty(&self) -> bool {
+        self.position >= self.data.len()
+    }
+    
+    /// Get remaining bytes
+    pub fn remaining(&self) -> usize {
+        self.data.len() - self.position
+    }
+    
+    /// Copy data to output slice, return bytes copied
+    pub fn copy_to(&mut self, dst: &mut [u8]) -> usize {
+        let available = &self.data[self.position..];
+        let to_copy = available.len().min(dst.len());
+        dst[..to_copy].copy_from_slice(&available[..to_copy]);
+        self.position += to_copy;
+        to_copy
+    }
+    
+    /// Get remaining data as slice
+    pub fn as_slice(&self) -> &[u8] {
+        &self.data[self.position..]
+    }
+}
+
+// ============= Macros =============
+
+/// Macro to simplify state transitions in poll methods
+#[macro_export]
+macro_rules! transition {
+    (same) => {
+        $crate::stream::state::Transition::Same
+    };
+    (next $state:expr) => {
+        $crate::stream::state::Transition::Next($state)
+    };
+    (complete $output:expr) => {
+        $crate::stream::state::Transition::Complete($output)
+    };
+    (yield $output:expr, $state:expr) => {
+        $crate::stream::state::Transition::Yield($output, $state)
+    };
+    (error $err:expr) => {
+        $crate::stream::state::Transition::Error($err)
+    };
+}
+
+/// Macro to match poll ready or return pending
+#[macro_export]
+macro_rules! ready_or_pending {
+    ($poll:expr) => {
+        match $poll {
+            std::task::Poll::Ready(t) => t,
+            std::task::Poll::Pending => return std::task::Poll::Pending,
+        }
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    
+    #[test]
+    fn test_read_buffer_basic() {
+        let mut buf = ReadBuffer::with_target(10);
+        assert_eq!(buf.remaining(), 10);
+        assert!(!buf.is_complete());
+        
+        buf.extend(b"hello");
+        assert_eq!(buf.len(), 5);
+        assert_eq!(buf.remaining(), 5);
+        assert!(!buf.is_complete());
+        
+        buf.extend(b"world");
+        assert_eq!(buf.len(), 10);
+        assert!(buf.is_complete());
+    }
+    
+    #[test]
+    fn test_read_buffer_take() {
+        let mut buf = ReadBuffer::new();
+        buf.extend(b"test data");
+        
+        let data = buf.take();
+        assert_eq!(&data[..], b"test data");
+        assert!(buf.is_empty());
+    }
+    
+    #[test]
+    fn test_write_buffer_basic() {
+        let mut buf = WriteBuffer::with_max_size(100);
+        assert!(buf.is_empty());
+        
+        buf.extend(b"hello").unwrap();
+        assert_eq!(buf.len(), 5);
+        assert!(!buf.is_empty());
+        
+        buf.advance(3);
+        assert_eq!(buf.len(), 2);
+        assert_eq!(buf.pending(), b"lo");
+    }
+    
+    #[test]
+    fn test_write_buffer_overflow() {
+        let mut buf = WriteBuffer::with_max_size(10);
+        assert!(buf.extend(b"short").is_ok());
+        assert!(buf.extend(b"toolong").is_err());
+    }
+    
+    #[test]
+    fn test_header_buffer() {
+        let mut buf = HeaderBuffer::<5>::new();
+        assert!(!buf.is_complete());
+        assert_eq!(buf.remaining(), 5);
+        
+        buf.unfilled_mut()[..3].copy_from_slice(b"hel");
+        buf.advance(3);
+        assert_eq!(buf.remaining(), 2);
+        
+        buf.unfilled_mut()[..2].copy_from_slice(b"lo");
+        buf.advance(2);
+        assert!(buf.is_complete());
+        assert_eq!(buf.as_array(), b"hello");
+    }
+    
+    #[test]
+    fn test_yield_buffer() {
+        let mut buf = YieldBuffer::new(Bytes::from_static(b"hello world"));
+        
+        let mut dst = [0u8; 5];
+        assert_eq!(buf.copy_to(&mut dst), 5);
+        assert_eq!(&dst, b"hello");
+        
+        assert_eq!(buf.remaining(), 6);
+        
+        let mut dst = [0u8; 10];
+        assert_eq!(buf.copy_to(&mut dst), 6);
+        assert_eq!(&dst[..6], b" world");
+        
+        assert!(buf.is_empty());
+    }
+    
+    #[test]
+    fn test_transition_map() {
+        let t: Transition<i32, String> = Transition::Complete("hello".to_string());
+        let t = t.map_output(|s| s.len());
+        
+        match t {
+            Transition::Complete(5) => {}
+            _ => panic!("Expected Complete(5)"),
+        }
+    }
+    
+    #[test]
+    fn test_poll_result() {
+        let r: PollResult<i32> = PollResult::Ready(42);
+        assert!(r.is_ready());
+        assert_eq!(r.ok(), Some(42));
+        
+        let r: PollResult<i32> = PollResult::Eof;
+        assert!(r.is_eof());
+        assert_eq!(r.ok(), None);
+    }
+}