mvlog: errc::short_read for entry stream

A short read can be indicative of corruption and shouldn't be ignored.
Adds a specific error code for this.

src/v/storage/mvlog/entry_stream.cc

@@ -26,7 +26,10 @@ ss::future<result<std::optional<entry>, errc>> entry_stream::next() {
     auto header_buf = co_await read_iobuf_exactly(
       stream_, packed_entry_header_size);
     if (header_buf.size_bytes() != packed_entry_header_size) {
-        co_return std::nullopt;
+        if (header_buf.size_bytes() == 0) {
+            co_return std::nullopt;
+        }
+        co_return errc::short_read;
     }
 
     auto entry_hdr = entry_header_from_iobuf(std::move(header_buf));
@@ -37,7 +40,7 @@ ss::future<result<std::optional<entry>, errc>> entry_stream::next() {
 
     auto body_buf = co_await read_iobuf_exactly(stream_, entry_hdr.body_size);
     if (body_buf.size_bytes() != static_cast<size_t>(entry_hdr.body_size)) {
-        co_return std::nullopt;
+        co_return errc::short_read;
     }
     co_return entry{entry_hdr, std::move(body_buf)};
 }

src/v/storage/mvlog/errc.h

@@ -20,6 +20,10 @@ enum class errc {
 
     // Checksum error, likely corruption.
     checksum_mismatch,
+
+    // Not enough data in a stream, potentially corruption.
+    short_read,
+
 };
 
 struct errc_category final : public std::error_category {
@@ -35,6 +39,8 @@ struct errc_category final : public std::error_category {
             return "broken_data_invariant";
         case errc::checksum_mismatch:
             return "checksum_mismatch";
+        case errc::short_read:
+            return "short_read";
         }
     }
 };

src/v/storage/mvlog/tests/entry_stream_utils_test.cc

@@ -104,8 +104,8 @@ TEST(EntryStream, TestShortHeader) {
         auto short_stream = make_iobuf_input_stream(std::move(copy_buf));
         entry_stream entries(std::move(short_stream));
         auto entry = entries.next().get();
-        ASSERT_FALSE(entry.has_error()) << entry.error();
-        ASSERT_FALSE(entry.value().has_value()) << "Expected no entry";
+        ASSERT_TRUE(entry.has_error());
+        ASSERT_EQ(entry.error(), errc::short_read);
     }
 }
 
@@ -137,8 +137,8 @@ TEST(EntryStream, TestShortBody) {
         auto short_stream = make_iobuf_input_stream(std::move(copy_buf));
         entry_stream entries(std::move(short_stream));
         auto entry = entries.next().get();
-        ASSERT_FALSE(entry.has_error()) << entry.error();
-        ASSERT_FALSE(entry.value().has_value()) << "Expected no entry";
+        ASSERT_TRUE(entry.has_error());
+        ASSERT_EQ(entry.error(), errc::short_read);
     }
 }