[PATCH v6 proxmox-backup 12/18] client: don't poll terminated source in FixedChunkStream

[Robert Obkircher <r.obkircher@proxmox.com>]

Fix incorrect chunking when the input file is a pts character device
(stdin in an interactive terminal). Previously, the stream could
produce multiple smaller chunks from interactive input, and a single
Ctrl+D was not enough to terminate it. This was because a None from
the source resulted in Some partial chunk, but the next time the input
was polled again and it suddenly had more input.

The documentation of Stream::poll_next clearly states that calling it
on a terminated stream may panic, block forever, or cause other kinds
of problems, so this should be fixed even it it appeared to work with
normal files.

Using the fuse method or adding a FusedStream trait bound might be
nicer that seemed like a larger change compared to the boolean.

Signed-off-by: Robert Obkircher <r.obkircher@proxmox.com>
---
 pbs-client/src/chunk_stream.rs | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/pbs-client/src/chunk_stream.rs b/pbs-client/src/chunk_stream.rs
index e3f0980c..94a45ac7 100644
--- a/pbs-client/src/chunk_stream.rs
+++ b/pbs-client/src/chunk_stream.rs
@@ -187,6 +187,7 @@ pub struct FixedChunkStream<S: Unpin> {
     input: S,
     chunk_size: usize,
     buffer: BytesMut,
+    done: bool,
 }
 
 impl<S: Unpin> FixedChunkStream<S> {
@@ -195,6 +196,7 @@ impl<S: Unpin> FixedChunkStream<S> {
             input,
             chunk_size,
             buffer: BytesMut::new(),
+            done: false,
         }
     }
 }
@@ -213,6 +215,9 @@ where
         cx: &mut Context,
     ) -> Poll<Option<Result<BytesMut, S::Error>>> {
         let this = self.get_mut();
+        if this.done {
+            return Poll::Ready(None);
+        }
         loop {
             if this.buffer.len() >= this.chunk_size {
                 return Poll::Ready(Some(Ok(this.buffer.split_to(this.chunk_size))));
@@ -223,6 +228,9 @@ where
                     return Poll::Ready(Some(Err(err)));
                 }
                 None => {
+                    // Must not call input.try_poll_next again!
+                    this.done = true;
+
                     // last chunk can have any size
                     if !this.buffer.is_empty() {
                         return Poll::Ready(Some(Ok(this.buffer.split())));
-- 
2.47.3