Process multiple stream decoder quads

Author: eldryoth

CHANGELOG.md

@@ -12,6 +12,9 @@
   pending quantum continues through the current slice when buffer capacity
   allows, including preserving final 1-2 byte tails instead of forcing an
   early short write.
+- Hardened `stream::Decoder::write` so direct writes process multiple complete
+  Base64 quads per call, continue after completing pending input, and preserve
+  final partial quads as pending input when those bytes are accepted.
 
 ## 0.12.0 - 2026-05-17
 

src/lib.rs

@@ -1347,25 +1347,57 @@ pub mod stream {
                 }
                 self.clear_pending();
                 consumed += needed;
-                return Ok(consumed);
+
+                if self.finished {
+                    return Ok(consumed);
+                }
             }
 
-            let remaining = &input[consumed..];
-            let full_len = remaining.len() / 4 * 4;
-            if full_len > 0 {
-                let quad = [remaining[0], remaining[1], remaining[2], remaining[3]];
+            while input.len() - consumed >= 4 {
+                if self.output.available_capacity() < 3 {
+                    return Ok(consumed);
+                }
+
+                let quad = [
+                    input[consumed],
+                    input[consumed + 1],
+                    input[consumed + 2],
+                    input[consumed + 3],
+                ];
                 let mut quad = quad;
-                let result = self.queue_full_quad(quad);
+                let mut decoded = [0u8; 3];
+                let written = match self.engine.decode_slice(&quad, &mut decoded) {
+                    Ok(written) => written,
+                    Err(err) => {
+                        crate::wipe_bytes(&mut quad);
+                        crate::wipe_bytes(&mut decoded);
+                        if consumed > 0 {
+                            return Ok(consumed);
+                        }
+
+                        self.failed = true;
+                        return Err(decode_error_to_io(err));
+                    }
+                };
+
+                let result = self.output.push_slice(&decoded[..written]);
                 crate::wipe_bytes(&mut quad);
+                crate::wipe_bytes(&mut decoded);
                 result?;
-                return Ok(consumed + 4);
+                consumed += 4;
+
+                if written < 3 {
+                    self.finished = true;
+                    return Ok(consumed);
+                }
             }
 
-            let tail = &remaining[full_len..];
+            let tail = &input[consumed..];
             self.pending[..tail.len()].copy_from_slice(tail);
             self.pending_len = tail.len();
+            consumed += tail.len();
 
-            Ok(input.len())
+            Ok(consumed)
         }
 
         fn flush(&mut self) -> io::Result<()> {

tests/rfc4648.rs

@@ -3399,6 +3399,56 @@ fn stream_decoder_write_failure_preserves_pending_input() {
     assert_eq!(decoder.get_ref().output, b"hi");
 }
 
+#[cfg(feature = "stream")]
+#[test]
+fn stream_decoder_direct_write_processes_multiple_quads() {
+    let mut decoder = Decoder::new(Vec::new(), STANDARD);
+
+    let written = decoder.write(b"aGVsbG8=").unwrap();
+    assert_eq!(written, 8);
+    assert_eq!(decoder.buffered_output_len(), 5);
+    assert!(decoder.has_terminal_padding());
+
+    let decoded = decoder.finish().unwrap();
+    assert_eq!(decoded, b"hello");
+}
+
+#[cfg(feature = "stream")]
+#[test]
+fn stream_decoder_direct_write_continues_after_pending_quad() {
+    let mut decoder = Decoder::new(Vec::new(), STANDARD);
+
+    assert_eq!(decoder.write(b"a").unwrap(), 1);
+    assert_eq!(decoder.pending_len(), 1);
+
+    let written = decoder.write(b"GVsbG8=").unwrap();
+    assert_eq!(written, 7);
+    assert_eq!(decoder.pending_len(), 0);
+    assert_eq!(decoder.buffered_output_len(), 5);
+    assert!(decoder.has_terminal_padding());
+
+    let decoded = decoder.finish().unwrap();
+    assert_eq!(decoded, b"hello");
+}
+
+#[cfg(feature = "stream")]
+#[test]
+fn stream_decoder_direct_write_reports_partial_progress_for_large_input() {
+    let input = vec![b'a'; 1500];
+    let encoded = STANDARD.encode_vec(&input).unwrap();
+    let mut decoder = Decoder::new(Vec::new(), STANDARD);
+
+    let written = decoder.write(&encoded).unwrap();
+    assert_eq!(written, 1364);
+    assert_eq!(decoder.buffered_output_len(), 1023);
+    assert_eq!(decoder.pending_len(), 0);
+    assert!(decoder.has_buffered_output());
+
+    decoder.write_all(&encoded[written..]).unwrap();
+    let decoded = decoder.finish().unwrap();
+    assert_eq!(decoded, input);
+}
+
 #[cfg(feature = "stream")]
 #[test]
 fn stream_decoder_direct_write_reports_partial_progress() {