consensus_encoding: add read interface for decoders

diff --git a/consensus_encoding/src/decode/decoders.rs b/consensus_encoding/src/decode/decoders.rs
index 70619dd2..6590c45d 100644
--- a/consensus_encoding/src/decode/decoders.rs
+++ b/consensus_encoding/src/decode/decoders.rs
@@ -47,6 +47,9 @@ impl<const N: usize> Decoder for ArrayDecoder<N> {
             Err(UnexpectedEofError { missing: N - self.bytes_written })
         }
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { N - self.bytes_written }
 }
 
 /// A decoder which wraps two inner decoders and returns the output of both.
@@ -165,6 +168,17 @@ where
             }
         }
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize {
+        match &self.state {
+            Decoder2State::First(first_decoder, second_decoder) =>
+                first_decoder.min_bytes_needed() + second_decoder.min_bytes_needed(),
+            Decoder2State::Second(_, second_decoder) => second_decoder.min_bytes_needed(),
+            Decoder2State::Errored => 0,
+            Decoder2State::Transitioning => 0,
+        }
+    }
 }
 
 /// A decoder which decodes three objects, one after the other.
@@ -215,6 +229,9 @@ where
         let ((first, second), third) = self.inner.end()?;
         Ok((first, second, third))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
 }
 
 /// A decoder which decodes four objects, one after the other.
@@ -268,6 +285,9 @@ where
         let ((first, second), (third, fourth)) = self.inner.end()?;
         Ok((first, second, third, fourth))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
 }
 
 /// A decoder which decodes six objects, one after the other.
@@ -346,6 +366,9 @@ where
         let ((first, second, third), (fourth, fifth, sixth)) = self.inner.end()?;
         Ok((first, second, third, fourth, fifth, sixth))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
 }
 
 /// Not enough bytes given to decoder.
diff --git a/consensus_encoding/src/decode/mod.rs b/consensus_encoding/src/decode/mod.rs
index e22e2ebb..3065036d 100644
--- a/consensus_encoding/src/decode/mod.rs
+++ b/consensus_encoding/src/decode/mod.rs
@@ -58,6 +58,14 @@ pub trait Decoder: Sized {
     /// May panic if called after a previous call to [`Self::push_bytes`] errored.
     #[must_use = "must check result to avoid panics on subsequent calls"]
     fn end(self) -> Result<Self::Output, Self::Error>;
+
+    /// Returns the minimum number of bytes needed to advance the state of the
+    /// decoder while ensuring there are no over-reads.
+    ///
+    /// Returns 0 if the decoder is complete and ready to finalize with [`Self::end`].
+    /// This is used by [`decode_from_read_unbuffered`] to optimize read sizes,
+    /// avoiding both inefficient under reads and unnecessary over-reads.
+    fn min_bytes_needed(&self) -> usize;
 }
 
 /// Decodes an object from a byte slice.
@@ -127,6 +135,89 @@ where
     }
 }
 
+/// Decodes an object from an unbuffered reader using a fixed-size buffer.
+///
+/// For most use cases, prefer [`decode_from_read`] with a [`std::io::BufReader`].
+/// This function is only needed when you have an unbuffered reader which you
+/// cannot wrap. It will probably have worse performance.
+///
+/// # Buffer
+///
+/// Uses a fixed 4KB (4096 bytes) stack-allocated buffer that is reused across
+/// read operations. This size is a good balance between memory usage and
+/// system call efficiency for most use cases.
+///
+/// For different buffer sizes, use [`decode_from_read_unbuffered_with`].
+///
+/// # Errors
+///
+/// Returns [`ReadError::Decode`] if the decoder encounters an error while parsing
+/// the data, or [`ReadError::Io`] if an I/O error occurs while reading.
+#[cfg(feature = "std")]
+pub fn decode_from_read_unbuffered<T, R>(
+    reader: R,
+) -> Result<T, ReadError<<T::Decoder as Decoder>::Error>>
+where
+    T: Decodable,
+    R: std::io::Read,
+{
+    decode_from_read_unbuffered_with::<T, R, 4096>(reader)
+}
+
+/// Decodes an object from an unbuffered reader using a custom-sized buffer.
+///
+/// For most use cases, prefer [`decode_from_read`] with a [`std::io::BufReader`].
+/// This function is only needed when you have an unbuffered reader which you
+/// cannot wrap. It will probably have worse performance.
+///
+/// # Buffer
+///
+/// The `BUFFER_SIZE` parameter controls the intermediate buffer size used for
+/// reading. The buffer is allocated on the stack (not heap) and reused across
+/// read operations. Larger buffers reduce the number of system calls, but use
+/// more memory.
+///
+/// # Errors
+///
+/// Returns [`ReadError::Decode`] if the decoder encounters an error while parsing
+/// the data, or [`ReadError::Io`] if an I/O error occurs while reading.
+#[cfg(feature = "std")]
+pub fn decode_from_read_unbuffered_with<T, R, const BUFFER_SIZE: usize>(
+    mut reader: R,
+) -> Result<T, ReadError<<T::Decoder as Decoder>::Error>>
+where
+    T: Decodable,
+    R: std::io::Read,
+{
+    let mut decoder = T::decoder();
+    let mut buffer = [0u8; BUFFER_SIZE];
+
+    while decoder.min_bytes_needed() > 0 {
+        // Only read what we need, up to buffer size.
+        let clamped_buffer = &mut buffer[..decoder.min_bytes_needed().min(BUFFER_SIZE)];
+        match reader.read(clamped_buffer) {
+            Ok(0) => {
+                // EOF, but still try to finalize the decoder.
+                return decoder.end().map_err(ReadError::Decode);
+            }
+            Ok(bytes_read) => {
+                if !decoder
+                    .push_bytes(&mut &clamped_buffer[..bytes_read])
+                    .map_err(ReadError::Decode)?
+                {
+                    return decoder.end().map_err(ReadError::Decode);
+                }
+            }
+            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {
+                // Auto retry read for non-fatal error.
+            }
+            Err(e) => return Err(ReadError::Io(e)),
+        }
+    }
+
+    decoder.end().map_err(ReadError::Decode)
+}
+
 /// An error that can occur when reading and decoding from a buffered reader.
 #[cfg(feature = "std")]
 #[derive(Debug)]
@@ -196,6 +287,8 @@ mod tests {
         }
 
         fn end(self) -> Result<Self::Output, Self::Error> { self.inner.end().map(TestArray) }
+
+        fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
     }
 
     #[test]
@@ -270,4 +363,44 @@ mod tests {
         let mut buf = Vec::new();
         let _ = cursor.read_to_end(&mut buf);
     }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn decode_from_read_unbuffered_success() {
+        let data = [1, 2, 3, 4];
+        let cursor = Cursor::new(&data);
+        let result: Result<TestArray, _> = decode_from_read_unbuffered(cursor);
+        assert!(result.is_ok());
+        let decoded = result.unwrap();
+        assert_eq!(decoded.0, [1, 2, 3, 4]);
+    }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn decode_from_read_unbuffered_unexpected_eof() {
+        let data = [1, 2, 3];
+        let cursor = Cursor::new(&data);
+        let result: Result<TestArray, _> = decode_from_read_unbuffered(cursor);
+        assert!(matches!(result, Err(ReadError::Decode(_))));
+    }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn decode_from_read_unbuffered_empty() {
+        let data = [];
+        let cursor = Cursor::new(&data);
+        let result: Result<TestArray, _> = decode_from_read_unbuffered(cursor);
+        assert!(matches!(result, Err(ReadError::Decode(_))));
+    }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn decode_from_read_unbuffered_extra_data() {
+        let data = [1, 2, 3, 4, 5, 6];
+        let cursor = Cursor::new(&data);
+        let result: Result<TestArray, _> = decode_from_read_unbuffered(cursor);
+        assert!(result.is_ok());
+        let decoded = result.unwrap();
+        assert_eq!(decoded.0, [1, 2, 3, 4]);
+    }
 }
diff --git a/consensus_encoding/src/lib.rs b/consensus_encoding/src/lib.rs
index 74993433..da8649d1 100644
--- a/consensus_encoding/src/lib.rs
+++ b/consensus_encoding/src/lib.rs
@@ -26,7 +26,9 @@ pub use self::decode::decoders::{
     ArrayDecoder, Decoder2, Decoder3, Decoder4, Decoder6, UnexpectedEofError,
 };
 #[cfg(feature = "std")]
-pub use self::decode::{decode_from_read, ReadError};
+pub use self::decode::{
+    decode_from_read, decode_from_read_unbuffered, decode_from_read_unbuffered_with, ReadError,
+};
 pub use self::decode::{decode_from_slice, Decodable, Decoder};
 #[cfg(feature = "alloc")]
 pub use self::encode::encode_to_vec;
diff --git a/consensus_encoding/tests/composition.rs b/consensus_encoding/tests/composition.rs
index 611aee1a..6f5ca237 100644
--- a/consensus_encoding/tests/composition.rs
+++ b/consensus_encoding/tests/composition.rs
@@ -68,6 +68,8 @@ impl Decoder for CompositeDataDecoder {
         let (first, second) = self.inner.end()?;
         Ok(CompositeData { first, second })
     }
+
+    fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
 }
 
 impl Decodable for CompositeData {
@@ -217,6 +219,8 @@ fn composition_error_unification() {
             let (first, second) = self.inner.end()?;
             Ok((first, second))
         }
+
+        fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
     }
 
     /// Another test composite decoder.
@@ -240,6 +244,8 @@ fn composition_error_unification() {
             let result = self.inner.end()?;
             Ok(result)
         }
+
+        fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
     }
 
     /// A decoder which can fail.
@@ -267,6 +273,8 @@ fn composition_error_unification() {
                 self.inner.end().map_err(NestedError::from)
             }
         }
+
+        fn min_bytes_needed(&self) -> usize { self.inner.min_bytes_needed() }
     }
 
     // A multi-layer, nested, decoder structure with a unified top level error type.
diff --git a/units/src/amount/unsigned.rs b/units/src/amount/unsigned.rs
index 301ee8ee..9d20674a 100644
--- a/units/src/amount/unsigned.rs
+++ b/units/src/amount/unsigned.rs
@@ -608,6 +608,9 @@ impl encoding::Decoder for AmountDecoder {
         let a = u64::from_le_bytes(self.0.end().map_err(AmountDecoderError::eof)?);
         Amount::from_sat(a).map_err(AmountDecoderError::out_of_range)
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.0.min_bytes_needed() }
 }
 
 #[cfg(feature = "encoding")]
diff --git a/units/src/block.rs b/units/src/block.rs
index c7ff5fb0..6f9f2342 100644
--- a/units/src/block.rs
+++ b/units/src/block.rs
@@ -191,6 +191,9 @@ impl encoding::Decoder for BlockHeightDecoder {
         let n = u32::from_le_bytes(self.0.end().map_err(BlockHeightDecoderError)?);
         Ok(BlockHeight::from_u32(n))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.0.min_bytes_needed() }
 }
 
 #[cfg(feature = "encoding")]
diff --git a/units/src/locktime/absolute/mod.rs b/units/src/locktime/absolute/mod.rs
index b626726a..ee3f7dd0 100644
--- a/units/src/locktime/absolute/mod.rs
+++ b/units/src/locktime/absolute/mod.rs
@@ -449,6 +449,9 @@ impl encoding::Decoder for LockTimeDecoder {
         let n = u32::from_le_bytes(self.0.end().map_err(LockTimeDecoderError)?);
         Ok(LockTime::from_consensus(n))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.0.min_bytes_needed() }
 }
 
 #[cfg(feature = "encoding")]
diff --git a/units/src/sequence.rs b/units/src/sequence.rs
index 938502f2..98a382e7 100644
--- a/units/src/sequence.rs
+++ b/units/src/sequence.rs
@@ -314,6 +314,9 @@ impl encoding::Decoder for SequenceDecoder {
         let n = u32::from_le_bytes(self.0.end().map_err(SequenceDecoderError)?);
         Ok(Sequence::from_consensus(n))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.0.min_bytes_needed() }
 }
 
 #[cfg(feature = "encoding")]
diff --git a/units/src/time.rs b/units/src/time.rs
index 8e6954ed..e16a3b7c 100644
--- a/units/src/time.rs
+++ b/units/src/time.rs
@@ -124,6 +124,9 @@ impl encoding::Decoder for BlockTimeDecoder {
         let t = u32::from_le_bytes(self.0.end().map_err(BlockTimeDecoderError)?);
         Ok(BlockTime::from_u32(t))
     }
+
+    #[inline]
+    fn min_bytes_needed(&self) -> usize { self.0.min_bytes_needed() }
 }
 
 #[cfg(feature = "encoding")]