From f5bfe72c7b668f7da93cfe11648ca732e9a6837e Mon Sep 17 00:00:00 2001
From: Fraser Savage <fsavage@influxdata.com>
Date: Thu, 14 Sep 2023 14:59:57 +0100
Subject: [PATCH] refactor(wal): Remove `IncompleteEntry` reader error

---
 ingester/src/init/wal_replay.rs | 20 +++++++++++---------
 wal/src/blocking/reader.rs      | 23 +++++++++--------------
 2 files changed, 20 insertions(+), 23 deletions(-)

diff --git a/ingester/src/init/wal_replay.rs b/ingester/src/init/wal_replay.rs
index 4362a9a296..d789db800d 100644
--- a/ingester/src/init/wal_replay.rs
+++ b/ingester/src/init/wal_replay.rs
@@ -194,9 +194,10 @@ impl SegmentedWalOpBatchReader for wal::ClosedSegmentFileReader {
 ///
 /// # Warnings
 ///
-/// This function relies on the [`wal::Error::UnableToReadNextOps`] error
-/// meaning that there are no more valid completed writes which can be read
-/// from the provided `batches` and that it is safe to ignore them.
+/// This function relies on the [`wal::blocking::ReaderError::UnableToReadData`]
+/// error sourced from an unexpected eof error to mean that there are no more
+/// valid completed writes which can be read from the provided `batches` and
+/// that it is safe to ignore them.
 async fn replay_file<T, F>(
     file: F,
     sink: &T,
@@ -214,14 +215,15 @@ where
     for batch in file {
         if let Err(
             err @ wal::Error::UnableToReadNextOps {
-                source: wal::blocking::ReaderError::IncompleteEntry { .. },
+                source: wal::blocking::ReaderError::UnableToReadData { source: io_err },
             },
-        ) = batch
+        ) = &batch
         {
-            warn!(%err, ?segment_id, "detected truncated WAL write, ending replay for file early");
-            break;
+            if io_err.kind() == std::io::ErrorKind::UnexpectedEof {
+                warn!(%err, ?segment_id, "detected truncated WAL write, ending replay for file early");
+                break;
+            }
         }
-
         let ops = batch.map_err(WalReplayError::ReadEntry)?;
 
         for op in ops {
@@ -703,7 +705,7 @@ mod tests {
                     .into_iter()
                     .collect()),
                 Err(wal::Error::UnableToReadNextOps {
-                    source: wal::blocking::ReaderError::IncompleteEntry {
+                    source: wal::blocking::ReaderError::UnableToReadData {
                         source: std::io::Error::new(
                             std::io::ErrorKind::UnexpectedEof,
                             "gremlins in the drive",
diff --git a/wal/src/blocking/reader.rs b/wal/src/blocking/reader.rs
index 07f9411fd2..dc4b5cf92a 100644
--- a/wal/src/blocking/reader.rs
+++ b/wal/src/blocking/reader.rs
@@ -60,12 +60,8 @@ where
         let mut decompressing_read = FrameDecoder::new(hashing_read);
 
         let mut data = Vec::with_capacity(100);
-        match decompressing_read.read_to_end(&mut data) {
-            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
-                Err(e).context(IncompleteEntrySnafu)
-            }
-            other => other.context(UnableToReadDataSnafu),
-        }?;
+        let other = decompressing_read.read_to_end(&mut data);
+        other.context(UnableToReadDataSnafu)?;
 
         let (actual_compressed_len, actual_checksum) = decompressing_read.into_inner().checksum();
 
@@ -164,12 +160,6 @@ pub enum Error {
         source: io::Error,
     },
 
-    /// An [`Error::IncompleteEntry`] error is returned when the reader is unable to
-    /// read an entry because of an unexpected end of file.
-    IncompleteEntry {
-        source: io::Error,
-    },
-
     LengthMismatch {
         expected: u64,
         actual: u64,
@@ -199,6 +189,7 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
 mod tests {
     use super::*;
     use crate::{SegmentId, FILE_TYPE_IDENTIFIER};
+    use assert_matches::assert_matches;
     use byteorder::WriteBytesExt;
     use std::io::Write;
     use test_helpers::assert_error;
@@ -265,7 +256,9 @@ mod tests {
         assert_eq!(uuid, segment_file.id.as_bytes());
 
         let read_fail = reader.one_entry();
-        assert_error!(read_fail, Error::IncompleteEntry { .. });
+        assert_matches!(read_fail, Err(Error::UnableToReadData { source: e }) => {
+            assert_matches!(e.kind(), std::io::ErrorKind::UnexpectedEof);
+        });
         // Trying to continue reading will fail as well, see:
         // <https://github.com/influxdata/influxdb_iox/issues/6222>
         assert_error!(reader.one_entry(), Error::UnableToReadData { .. });
@@ -290,7 +283,9 @@ mod tests {
         assert_eq!(uuid, segment_file.id.as_bytes());
 
         let read_fail = reader.one_entry();
-        assert_error!(read_fail, Error::IncompleteEntry { .. });
+        assert_matches!(read_fail, Err(Error::UnableToReadData { source: e }) => {
+            assert_matches!(e.kind(), std::io::ErrorKind::UnexpectedEof);
+        });
         // Trying to continue reading will fail as well, see:
         // <https://github.com/influxdata/influxdb_iox/issues/6222>
         assert_error!(reader.one_entry(), Error::UnableToReadData { .. });