fix: Don't give up on reading local header when given short reads

rc-zip-sync/src/entry_reader.rs

@@ -30,38 +30,45 @@
     R: io::Read,
 {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let mut fsm = match self.fsm.take() {
-            Some(fsm) => fsm,
-            None => return Ok(0),
-        };
+        loop {
+            let mut fsm = match self.fsm.take() {
+                Some(fsm) => fsm,
+                None => return Ok(0),
+            };
 
-        if fsm.wants_read() {
-            trace!("fsm wants read");
-            let n = self.rd.read(fsm.space())?;
-            trace!("giving fsm {} bytes", n);
-            fsm.fill(n);
-        } else {
-            trace!("fsm does not want read");
-        }
+            #[allow(clippy::needless_late_init)] // don't tell me what to do
+            let filled_bytes;
+            if fsm.wants_read() {
+                tracing::trace!(space_avail = fsm.space().len(), "fsm wants read");
+                let n = self.rd.read(fsm.space())?;
+                fsm.fill(n);
+                filled_bytes = n;
+            } else {
+                trace!("fsm does not want read");
+                filled_bytes = 0;
+            }
 
-        match fsm.process(buf)? {
-            FsmResult::Continue((fsm, outcome)) => {
-                self.fsm = Some(fsm);
+            match fsm.process(buf)? {
+                FsmResult::Continue((fsm, outcome)) => {
+                    self.fsm = Some(fsm);
 
-                if outcome.bytes_written > 0 {
-                    Ok(outcome.bytes_written)
-                } else if outcome.bytes_read == 0 {
-                    // that's EOF, baby!
-                    Ok(0)
-                } else {
-                    // loop, it happens
-                    self.read(buf)
+                    if outcome.bytes_written > 0 {
+                        tracing::trace!("wrote {} bytes", outcome.bytes_written);
+                        return Ok(outcome.bytes_written);
+                    } else if filled_bytes > 0 || outcome.bytes_read > 0 {
+                        // progress was made, keep reading
+                        continue;
+                    } else {
+                        return Err(io::Error::new(
+                            io::ErrorKind::Other,
+                            "entry reader: no progress",
+                        ));
+                    }
+                }
+                FsmResult::Done(_) => {
+                    // neat!
+                    return Ok(0);
                 }
-            }
-            FsmResult::Done(_) => {
-                // neat!
-                trace!("fsm done");
-                Ok(0)
             }
         }
     }

rc-zip-tokio/src/entry_reader.rs

@@ -1,4 +1,4 @@
-use std::{pin::Pin, task};
+use std::{io, pin::Pin, task};
 
 use pin_project_lite::pin_project;
 use rc_zip::{
@@ -42,48 +42,57 @@
         cx: &mut task::Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> task::Poll<std::io::Result<()>> {
-        let this = self.as_mut().project();
+        let mut this = self.as_mut().project();
 
-        let mut fsm = match this.fsm.take() {
-            Some(fsm) => fsm,
-            None => return Ok(()).into(),
-        };
+        loop {
+            let mut fsm = match this.fsm.take() {
+                Some(fsm) => fsm,
+                None => return Ok(()).into(),
+            };
 
-        if fsm.wants_read() {
-            tracing::trace!(space_avail = fsm.space().len(), "fsm wants read");
-            let mut buf = ReadBuf::new(fsm.space());
-            match this.rd.poll_read(cx, &mut buf) {
-                task::Poll::Ready(res) => res?,
-                task::Poll::Pending => {
-                    *this.fsm = Some(fsm);
-                    return task::Poll::Pending;
+            let filled_bytes;
+            if fsm.wants_read() {
+                tracing::trace!(space_avail = fsm.space().len(), "fsm wants read");
+                let mut buf = ReadBuf::new(fsm.space());
+                match this.rd.as_mut().poll_read(cx, &mut buf) {
+                    task::Poll::Ready(res) => res?,
+                    task::Poll::Pending => {
+                        *this.fsm = Some(fsm);
+                        return task::Poll::Pending;
+                    }
                 }
-            }
-            let n = buf.filled().len();
+                let n = buf.filled().len();
 
-            tracing::trace!("read {} bytes", n);
-            fsm.fill(n);
-        } else {
-            tracing::trace!("fsm does not want read");
-        }
+                tracing::trace!("read {} bytes", n);
+                fsm.fill(n);
+                filled_bytes = n;
+            } else {
+                tracing::trace!("fsm does not want read");
+                filled_bytes = 0;
+            }
 
-        match fsm.process(buf.initialize_unfilled())? {
-            FsmResult::Continue((fsm, outcome)) => {
-                *this.fsm = Some(fsm);
-                if outcome.bytes_written > 0 {
-                    tracing::trace!("wrote {} bytes", outcome.bytes_written);
-                    buf.advance(outcome.bytes_written);
-                } else if outcome.bytes_read == 0 {
-                    // that's EOF, baby!
-                } else {
-                    // loop, it happens
-                    return self.poll_read(cx, buf);
+            match fsm.process(buf.initialize_unfilled())? {
+                FsmResult::Continue((fsm, outcome)) => {
+                    *this.fsm = Some(fsm);
+                    if outcome.bytes_written > 0 {
+                        tracing::trace!("wrote {} bytes", outcome.bytes_written);
+                        buf.advance(outcome.bytes_written);
+                    } else if filled_bytes > 0 || outcome.bytes_read > 0 {
+                        // progress was made, keep reading
+                        continue;
+                    } else {
+                        return Err(io::Error::new(
+                            io::ErrorKind::Other,
+                            "entry reader: no progress",
+                        ))
+                        .into();
+                    }
+                }
+                FsmResult::Done(_) => {
+                    // neat!
                 }
             }
-            FsmResult::Done(_) => {
-                // neat!
-            }
+            return Ok(()).into();
         }
-        Ok(()).into()
     }
 }

rc-zip-tokio/src/read_zip.rs

@@ -61,13 +61,10 @@
         let mut fsm = ArchiveFsm::new(size);
         loop {
             if let Some(offset) = fsm.wants_read() {
-                trace!(%offset, "read_zip_with_size: wants_read, space len = {}", fsm.space().len());
-
                 let mut cstate_next = match cstate.take() {
                     Some(cstate) => {
                         if cstate.offset == offset {
                             // all good, re-using
-                            trace!(%offset, "read_zip_with_size: re-using cursor");
                             cstate
                         } else {
                             trace!(%offset, %cstate.offset, "read_zip_with_size: making new cursor (had wrong offset)");
@@ -91,7 +88,7 @@
                         cstate_next.offset += read_bytes as u64;
                         cstate = Some(cstate_next);
 
-                        trace!(%read_bytes, "read_zip_with_size: read");
+                        trace!(%read_bytes, "filling fsm");
                         if read_bytes == 0 {
                             return Err(Error::IO(io::ErrorKind::UnexpectedEof.into()));
                         }
@@ -308,23 +305,18 @@
         match &mut self.state {
             ARAFCState::Idle(core) => {
                 if core.inner_buf_offset < core.inner_buf_len {
-                    trace!(inner_buf_offset = %core.inner_buf_offset, inner_buf_len = %core.inner_buf_len, avail = %(core.inner_buf_len - core.inner_buf_offset), "poll_read: have data in inner buffer");
-
                     // we have data in the inner buffer, don't even need
                     // to spawn a blocking task
                     let read_len =
                         cmp::min(buf.remaining(), core.inner_buf_len - core.inner_buf_offset);
-                    trace!(%read_len, "poll_read: putting slice");
 
                     buf.put_slice(&core.inner_buf[core.inner_buf_offset..][..read_len]);
                     core.inner_buf_offset += read_len;
-                    trace!(inner_buf_offset = %core.inner_buf_offset, inner_buf_len = %core.inner_buf_len, "poll_read: after put_slice");
+                    trace!(inner_buf_offset = %core.inner_buf_offset, inner_buf_len = %core.inner_buf_len, "read from inner buffer");
 
                     return Poll::Ready(Ok(()));
                 }
 
-                trace!("will have to issue a read call");
-
                 // this is just used to shadow core
                 #[allow(unused_variables, clippy::let_unit_value)]
                 let core = ();
@@ -339,7 +331,7 @@
 
                 let fut = Box::pin(tokio::task::spawn_blocking(move || {
                     let read_bytes = file.read_at(file_offset, &mut inner_buf)?;
-                    trace!("read {} bytes", read_bytes);
+                    trace!(%read_bytes, "read from file");
                     Ok(ARAFCCore {
                         file_offset: file_offset + read_bytes as u64,
                         file,

rc-zip/src/fsm/entry/lzma_dec.rs

@@ -40,75 +40,92 @@
 impl Decompressor for LzmaDec {
     fn decompress(
         &mut self,
-        in_buf: &[u8],
+        mut in_buf: &[u8],
         out: &mut [u8],
         has_more_input: HasMoreInput,
     ) -> Result<DecompressOutcome, Error> {
-        tracing::trace!(
-            in_buf_len = in_buf.len(),
-            out_len = out.len(),
-            remain_in_internal_buffer = self.internal_buf_mut().len(),
-            "decompress",
-        );
-
         let mut outcome: DecompressOutcome = Default::default();
 
-        self.copy_to_out(out, &mut outcome);
-        if outcome.bytes_written > 0 {
-            trace!(
-                "still draining internal buffer, just copied {} bytes",
-                outcome.bytes_written
+        loop {
+            tracing::trace!(
+                in_buf_len = in_buf.len(),
+                out_len = out.len(),
+                remain_in_internal_buffer = self.internal_buf_mut().len(),
+                ?outcome,
+                "decompress",
             );
-            return Ok(outcome);
-        }
 
-        match &mut self.state {
-            State::Writing(stream) => {
-                let n = stream.write(in_buf).map_err(dec_err)?;
+            self.copy_to_out(out, &mut outcome);
+            if outcome.bytes_written > 0 {
                 trace!(
-                    "wrote {} bytes to decompressor (of {} available)",
-                    n,
-                    in_buf.len()
+                    "still draining internal buffer, just copied {} bytes",
+                    outcome.bytes_written
                 );
-                outcome.bytes_read = n;
-
-                // if we haven't written all the input, and we haven't gotten
-                // any output, then we need to keep going
-                if n != 0 && n < in_buf.len() && self.internal_buf_mut().is_empty() {
-                    // note: the n != 0 here is because apparently there can be a 10-byte
-                    // trailer after LZMA compressed data? and the decoder will _refuse_
-                    // to let us write them, so when we have just these 10 bytes left,
-                    // it's good to just let the decoder finish up.
-                    trace!("didn't write all output AND no output yet, so keep going");
-                    return self.decompress(&in_buf[n..], out, has_more_input);
-                }
+                return Ok(outcome);
+            }
 
-                match has_more_input {
-                    HasMoreInput::Yes => {
-                        // keep going
-                        trace!("more input to come");
+            match &mut self.state {
+                State::Writing(stream) => {
+                    let n = stream.write(in_buf).map_err(dec_err)?;
+                    trace!(
+                        "wrote {} bytes to decompressor (of {} available)",
+                        n,
+                        in_buf.len()
+                    );
+                    outcome.bytes_read += n;
+                    in_buf = &in_buf[n..];
+
+                    // if we wrote some of the input, and we haven't gotten any
+                    // output, then we need to loop
+                    if n > 0 && n < in_buf.len() && self.internal_buf_mut().is_empty() {
+                        trace!("fed _some_ to the decoder and no output yet, keep going");
+                        continue;
                     }
-                    HasMoreInput::No => {
-                        trace!("no more input to come");
-                        match std::mem::take(&mut self.state) {
-                            State::Writing(stream) => {
-                                trace!("finishing...");
-                                self.state = State::Draining(stream.finish().map_err(dec_err)?);
+
+                    match has_more_input {
+                        HasMoreInput::Yes => {
+                            trace!("more input to come");
+                        }
+                        HasMoreInput::No => {
+                            trace!("no more input to come");
+
+                            // this happens when we hit the 10-byte trailer mentioned above
+                            // in this case, we just pretend we wrote everything
+                            match in_buf.len() {
+                                0 => {
+                                    // trailer is not present, that's okay
+                                }
+                                10 => {
+                                    trace!("eating LZMA trailer");
+                                    outcome.bytes_read += 10;
+                                }
+                                _ => {
+                                    return Err(Error::Decompression { method: Method::Lzma, msg: format!("expected LZMA trailer or no LZMA trailer, but not a {}-byte trailer", in_buf.len()) });
+                                }
+                            }
+
+                            match std::mem::take(&mut self.state) {
+                                State::Writing(stream) => {
+                                    trace!("finishing...");
+                                    self.state = State::Draining(stream.finish().map_err(dec_err)?);
+                                    continue;
+                                }
+                                _ => unreachable!(),
                             }
-                            _ => unreachable!(),
                         }
                     }
                 }
+                State::Draining(_) => {
+                    // keep going
+                    trace!("draining");
+                }
+                State::Transition => unreachable!(),
             }
-            State::Draining(_) => {
-                // keep going
-            }
-            State::Transition => unreachable!(),
-        }
 
-        self.copy_to_out(out, &mut outcome);
-        trace!("decompressor gave us {} bytes", outcome.bytes_written);
-        Ok(outcome)
+            self.copy_to_out(out, &mut outcome);
+            trace!("decompressor gave us {} bytes", outcome.bytes_written);
+            return Ok(outcome);
+        }
     }
 }