consensus_encoding: add decoder composition

diff --git a/consensus_encoding/src/decode/decoders.rs b/consensus_encoding/src/decode/decoders.rs
index 56534a9d..ba0041d6 100644
--- a/consensus_encoding/src/decode/decoders.rs
+++ b/consensus_encoding/src/decode/decoders.rs
@@ -4,6 +4,42 @@
 
 use super::Decoder;
 
+/// A sum type representing one of two possible decoder errors.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Either<F, S> {
+    /// The first variant.
+    First(F),
+    /// The second variant.
+    Second(S),
+}
+
+impl<F, S> core::fmt::Display for Either<F, S>
+where
+    F: core::fmt::Display,
+    S: core::fmt::Display,
+{
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            Either::First(first) => first.fmt(f),
+            Either::Second(second) => second.fmt(f),
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl<F, S> std::error::Error for Either<F, S>
+where
+    F: std::error::Error,
+    S: std::error::Error,
+{
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Either::First(first) => first.source(),
+            Either::Second(second) => second.source(),
+        }
+    }
+}
+
 /// Not enough bytes given to decoder.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct UnexpectedEof {
@@ -63,3 +99,258 @@ impl<const N: usize> Decoder for ArrayDecoder<N> {
         }
     }
 }
+
+/// A decoder which decodes two objects, one after the other.
+pub struct Decoder2<A, B>
+where
+    A: Decoder,
+    B: Decoder,
+{
+    state: Decoder2State<A, B>,
+}
+
+enum Decoder2State<A: Decoder, B: Decoder> {
+    /// Decoding the first decoder, with second decoder waiting.
+    First(A, B),
+    /// Decoding the second decoder, with the first result stored.
+    Second(A::Output, B),
+    /// Decoder has failed and cannot be used again.
+    Errored,
+    /// Temporary state during transitions from First to Second, should never be observed.
+    Transitioning,
+}
+
+impl<A: Decoder, B: Decoder> Decoder2State<A, B> {
+    /// Transitions from the first state to second by extracting both decoders.
+    ///
+    /// We use `mem::replace` to atomically swap the entire state, giving us
+    /// ownership of the decoders so we can consume the first decoder while
+    /// holding a mutable reference to the state.
+    ///
+    /// If this method is called when not in the `First` state, we panic
+    /// with `#[track_caller]` to show where the bug occurred.
+    #[track_caller]
+    fn transition(&mut self) -> (A, B) {
+        match core::mem::replace(self, Decoder2State::Transitioning) {
+            Decoder2State::First(first, second) => (first, second),
+            _ => panic!("transition called on invalid state"),
+        }
+    }
+}
+
+impl<A, B> Decoder2<A, B>
+where
+    A: Decoder,
+    B: Decoder,
+{
+    /// Constructs a new composite decoder.
+    pub fn new(first: A, second: B) -> Self { Self { state: Decoder2State::First(first, second) } }
+}
+
+impl<A, B> Decoder for Decoder2<A, B>
+where
+    A: Decoder,
+    B: Decoder,
+{
+    type Output = (A::Output, B::Output);
+    type Error = Either<A::Error, B::Error>;
+
+    fn push_bytes(&mut self, bytes: &mut &[u8]) -> Result<bool, Self::Error> {
+        loop {
+            match &mut self.state {
+                Decoder2State::First(first_decoder, _) => {
+                    if first_decoder.push_bytes(bytes).map_err(Either::First)? {
+                        // First decoder wants more data.
+                        return Ok(true);
+                    }
+
+                    // First decoder is complete, transition to second.
+                    let (first, second) = self.state.transition();
+                    let first_result = first.end().map_err(|error| {
+                        self.state = Decoder2State::Errored;
+                        Either::First(error)
+                    })?;
+                    self.state = Decoder2State::Second(first_result, second);
+                }
+                Decoder2State::Second(_, second_decoder) => {
+                    return second_decoder.push_bytes(bytes).map_err(|error| {
+                        self.state = Decoder2State::Errored;
+                        Either::Second(error)
+                    });
+                }
+                Decoder2State::Errored => {
+                    panic!("use of failed decoder");
+                }
+                Decoder2State::Transitioning => {
+                    panic!("use of decoder in transitioning state");
+                }
+            }
+        }
+    }
+
+    fn end(self) -> Result<Self::Output, Self::Error> {
+        match self.state {
+            Decoder2State::First(first_decoder, second_decoder) => {
+                // This branch is most likely an error since the decoder
+                // never got to the second one. But letting the error bubble
+                // up naturally from the child decoders.
+                let first_result = first_decoder.end().map_err(Either::First)?;
+                let second_result = second_decoder.end().map_err(Either::Second)?;
+                Ok((first_result, second_result))
+            }
+            Decoder2State::Second(first_result, second_decoder) => {
+                let second_result = second_decoder.end().map_err(Either::Second)?;
+                Ok((first_result, second_result))
+            }
+            Decoder2State::Errored => {
+                panic!("use of failed decoder");
+            }
+            Decoder2State::Transitioning => {
+                panic!("use of decoder in transitioning state");
+            }
+        }
+    }
+}
+
+/// A decoder which decodes three objects, one after the other.
+pub struct Decoder3<A, B, C>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+{
+    inner: Decoder2<Decoder2<A, B>, C>,
+}
+
+impl<A, B, C> Decoder3<A, B, C>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+{
+    /// Constructs a new composite decoder.
+    pub fn new(dec_1: A, dec_2: B, dec_3: C) -> Self {
+        Self { inner: Decoder2::new(Decoder2::new(dec_1, dec_2), dec_3) }
+    }
+}
+
+impl<A, B, C> Decoder for Decoder3<A, B, C>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+{
+    type Output = (A::Output, B::Output, C::Output);
+    type Error = Either<Either<A::Error, B::Error>, C::Error>;
+
+    fn push_bytes(&mut self, bytes: &mut &[u8]) -> Result<bool, Self::Error> {
+        self.inner.push_bytes(bytes)
+    }
+
+    fn end(self) -> Result<Self::Output, Self::Error> {
+        let ((first, second), third) = self.inner.end()?;
+        Ok((first, second, third))
+    }
+}
+
+/// A decoder which decodes four objects, one after the other.
+pub struct Decoder4<A, B, C, D>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+{
+    inner: Decoder2<Decoder2<A, B>, Decoder2<C, D>>,
+}
+
+impl<A, B, C, D> Decoder4<A, B, C, D>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+{
+    /// Constructs a new composite decoder.
+    pub fn new(dec_1: A, dec_2: B, dec_3: C, dec_4: D) -> Self {
+        Self { inner: Decoder2::new(Decoder2::new(dec_1, dec_2), Decoder2::new(dec_3, dec_4)) }
+    }
+}
+
+impl<A, B, C, D> Decoder for Decoder4<A, B, C, D>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+{
+    type Output = (A::Output, B::Output, C::Output, D::Output);
+    type Error = Either<Either<A::Error, B::Error>, Either<C::Error, D::Error>>;
+
+    fn push_bytes(&mut self, bytes: &mut &[u8]) -> Result<bool, Self::Error> {
+        self.inner.push_bytes(bytes)
+    }
+
+    fn end(self) -> Result<Self::Output, Self::Error> {
+        let ((first, second), (third, fourth)) = self.inner.end()?;
+        Ok((first, second, third, fourth))
+    }
+}
+
+/// A decoder which decodes six objects, one after the other.
+pub struct Decoder6<A, B, C, D, E, F>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+    E: Decoder,
+    F: Decoder,
+{
+    inner: Decoder2<Decoder3<A, B, C>, Decoder3<D, E, F>>,
+}
+
+impl<A, B, C, D, E, F> Decoder6<A, B, C, D, E, F>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+    E: Decoder,
+    F: Decoder,
+{
+    /// Constructs a new composite decoder.
+    pub fn new(dec_1: A, dec_2: B, dec_3: C, dec_4: D, dec_5: E, dec_6: F) -> Self {
+        Self {
+            inner: Decoder2::new(
+                Decoder3::new(dec_1, dec_2, dec_3),
+                Decoder3::new(dec_4, dec_5, dec_6),
+            ),
+        }
+    }
+}
+
+impl<A, B, C, D, E, F> Decoder for Decoder6<A, B, C, D, E, F>
+where
+    A: Decoder,
+    B: Decoder,
+    C: Decoder,
+    D: Decoder,
+    E: Decoder,
+    F: Decoder,
+{
+    type Output = (A::Output, B::Output, C::Output, D::Output, E::Output, F::Output);
+    type Error = Either<
+        Either<Either<A::Error, B::Error>, C::Error>,
+        Either<Either<D::Error, E::Error>, F::Error>,
+    >;
+
+    fn push_bytes(&mut self, bytes: &mut &[u8]) -> Result<bool, Self::Error> {
+        self.inner.push_bytes(bytes)
+    }
+
+    fn end(self) -> Result<Self::Output, Self::Error> {
+        let ((first, second, third), (fourth, fifth, sixth)) = self.inner.end()?;
+        Ok((first, second, third, fourth, fifth, sixth))
+    }
+}
diff --git a/consensus_encoding/src/lib.rs b/consensus_encoding/src/lib.rs
index 9663f05b..3246d1c3 100644
--- a/consensus_encoding/src/lib.rs
+++ b/consensus_encoding/src/lib.rs
@@ -22,7 +22,9 @@ extern crate std;
 mod decode;
 mod encode;
 
-pub use self::decode::decoders::{ArrayDecoder, UnexpectedEof};
+pub use self::decode::decoders::{
+    ArrayDecoder, Decoder2, Decoder3, Decoder4, Decoder6, Either, UnexpectedEof,
+};
 pub use self::decode::{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
new file mode 100644
index 00000000..aec4306d
--- /dev/null
+++ b/consensus_encoding/tests/composition.rs
@@ -0,0 +1,139 @@
+// SPDX-License-Identifier: CC0-1.0
+
+//! Test composition of encoders and decoders.
+
+use consensus_encoding::{
+    ArrayDecoder, ArrayEncoder, Decodable, Decoder, Decoder2, Decoder6, Encodable, Encoder,
+    Encoder2, Encoder6,
+};
+
+const EMPTY: &[u8] = &[];
+
+// A simple composite type that encodes as [4 bytes] + [2 bytes].
+#[derive(Debug, PartialEq, Eq)]
+struct CompositeData {
+    first: [u8; 4],
+    second: [u8; 2],
+}
+
+impl Encodable for CompositeData {
+    type Encoder<'e> = Encoder2<ArrayEncoder<4>, ArrayEncoder<2>>;
+
+    fn encoder(&self) -> Self::Encoder<'_> {
+        Encoder2::new(
+            ArrayEncoder::without_length_prefix(self.first),
+            ArrayEncoder::without_length_prefix(self.second),
+        )
+    }
+}
+
+/// A wrapper decoder that converts the tuple output to [`CompositeData`].
+struct CompositeDataDecoder {
+    inner: Decoder2<ArrayDecoder<4>, ArrayDecoder<2>>,
+}
+
+impl CompositeDataDecoder {
+    fn new() -> Self {
+        Self { inner: Decoder2::new(ArrayDecoder::<4>::new(), ArrayDecoder::<2>::new()) }
+    }
+}
+
+impl Decoder for CompositeDataDecoder {
+    type Output = CompositeData;
+    type Error = <Decoder2<ArrayDecoder<4>, ArrayDecoder<2>> as Decoder>::Error;
+
+    fn push_bytes(&mut self, bytes: &mut &[u8]) -> Result<bool, Self::Error> {
+        self.inner.push_bytes(bytes)
+    }
+
+    fn end(self) -> Result<Self::Output, Self::Error> {
+        let (first, second) = self.inner.end()?;
+        Ok(CompositeData { first, second })
+    }
+}
+
+impl Decodable for CompositeData {
+    type Decoder = CompositeDataDecoder;
+
+    fn decoder() -> Self::Decoder { CompositeDataDecoder::new() }
+}
+
+#[test]
+fn composition_chain() {
+    let original = CompositeData { first: [0x01, 0x02, 0x03, 0x04], second: [0x05, 0x06] };
+    // Encode using the pull encoder.
+    let mut encoder = original.encoder();
+    let mut encoded_bytes = Vec::new();
+    while let Some(chunk) = encoder.current_chunk() {
+        encoded_bytes.extend_from_slice(chunk);
+        encoder.advance();
+    }
+    // Decode using the push decoder.
+    let mut decoder = CompositeData::decoder();
+    let mut bytes = &encoded_bytes[..];
+    let needs_more = decoder.push_bytes(&mut bytes).unwrap();
+    assert!(!needs_more, "CompositeData decoder should be ready to end");
+    assert_eq!(bytes, EMPTY);
+    let decoded = decoder.end().unwrap();
+    assert_eq!(original, decoded);
+}
+
+#[test]
+fn composition_nested() {
+    let data = b"abcdef";
+    let mut encoder6 = Encoder6::new(
+        ArrayEncoder::without_length_prefix([data[0]]),
+        ArrayEncoder::without_length_prefix([data[1]]),
+        ArrayEncoder::without_length_prefix([data[2]]),
+        ArrayEncoder::without_length_prefix([data[3]]),
+        ArrayEncoder::without_length_prefix([data[4]]),
+        ArrayEncoder::without_length_prefix([data[5]]),
+    );
+
+    let mut encoded_bytes = Vec::new();
+    while let Some(chunk) = encoder6.current_chunk() {
+        encoded_bytes.extend_from_slice(chunk);
+        encoder6.advance();
+    }
+    assert_eq!(encoded_bytes, data);
+
+    let mut decoder6 = Decoder6::new(
+        ArrayDecoder::<1>::new(),
+        ArrayDecoder::<1>::new(),
+        ArrayDecoder::<1>::new(),
+        ArrayDecoder::<1>::new(),
+        ArrayDecoder::<1>::new(),
+        ArrayDecoder::<1>::new(),
+    );
+    let mut bytes = &encoded_bytes[..];
+    let needs_more = decoder6.push_bytes(&mut bytes).unwrap();
+    assert!(!needs_more, "Decoder6 should be ready to end");
+    assert_eq!(bytes, EMPTY);
+    let (first, second, third, fourth, fifth, sixth) = decoder6.end().unwrap();
+    assert_eq!(first, [data[0]]);
+    assert_eq!(second, [data[1]]);
+    assert_eq!(third, [data[2]]);
+    assert_eq!(fourth, [data[3]]);
+    assert_eq!(fifth, [data[4]]);
+    assert_eq!(sixth, [data[5]]);
+}
+
+#[test]
+fn composition_extra_bytes() {
+    // Test that Decoder2 consumes exactly what it needs and leaves extra bytes unconsumed.
+    let mut decoder2 = Decoder2::new(ArrayDecoder::<2>::new(), ArrayDecoder::<3>::new());
+    let mut bytes = &[0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08][..];
+    let original_len = bytes.len();
+
+    let needs_more = decoder2.push_bytes(&mut bytes).unwrap();
+    assert!(!needs_more, "Decoder2 should be ready to end after consuming all needed bytes");
+
+    let consumed = original_len - bytes.len();
+    assert_eq!(consumed, 5, "Decoder2 should consume exactly 5 bytes");
+    assert_eq!(bytes.len(), 3, "3 bytes should remain unconsumed");
+    assert_eq!(bytes, &[0x06, 0x07, 0x08], "Remaining bytes should be the last 3");
+
+    let (first, second) = decoder2.end().unwrap();
+    assert_eq!(first, [0x01, 0x02], "First decoder should get first 2 bytes");
+    assert_eq!(second, [0x03, 0x04, 0x05], "Second decoder should get next 3 bytes");
+}