From f2c9075d16d17a14598246cfc6e620d60135849d Mon Sep 17 00:00:00 2001
From: qiuchengxuan <qiuchengxuan@gmail.com>
Date: Wed, 25 Oct 2023 09:16:24 -0700
Subject: [PATCH] Fix dead loop with kSkipAnyCorruptedRecords mode selected in
 some cases (#11955) (#11979)

Summary:
With fragmented record span across multiple blocks, if any following blocks corrupted with arbitary data, and intepreted log number less than the current log number, program will fall into infinite loop due to
not skipping buffer leading bytes

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11979

Test Plan: existing unit tests

Reviewed By: ajkr

Differential Revision: D50604408

Pulled By: jowlyzhang

fbshipit-source-id: e50a0c7e7c3d293fb9d5afec0a3eb4a1835b7a3b
---
 db/log_reader.cc | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/db/log_reader.cc b/db/log_reader.cc
index 5ec262dcd3..4e470616f0 100644
--- a/db/log_reader.cc
+++ b/db/log_reader.cc
@@ -469,12 +469,14 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result, size_t* drop_size,
     const unsigned int type = header[6];
     const uint32_t length = a | (b << 8);
     int header_size = kHeaderSize;
-    if ((type >= kRecyclableFullType && type <= kRecyclableLastType) ||
-        type == kRecyclableUserDefinedTimestampSizeType) {
+    const bool is_recyclable_type =
+        ((type >= kRecyclableFullType && type <= kRecyclableLastType) ||
+         type == kRecyclableUserDefinedTimestampSizeType);
+    if (is_recyclable_type) {
+      header_size = kRecyclableHeaderSize;
       if (end_of_buffer_offset_ - buffer_.size() == 0) {
         recycled_ = true;
       }
-      header_size = kRecyclableHeaderSize;
       // We need enough for the larger header
       if (buffer_.size() < static_cast<size_t>(kRecyclableHeaderSize)) {
         int r = kEof;
@@ -483,11 +485,8 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result, size_t* drop_size,
         }
         continue;
       }
-      const uint32_t log_num = DecodeFixed32(header + 7);
-      if (log_num != log_number_) {
-        return kOldRecord;
-      }
     }
+
     if (header_size + length > buffer_.size()) {
       assert(buffer_.size() >= static_cast<size_t>(header_size));
       *drop_size = buffer_.size();
@@ -499,6 +498,14 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result, size_t* drop_size,
       return kBadRecordLen;
     }
 
+    if (is_recyclable_type) {
+      const uint32_t log_num = DecodeFixed32(header + 7);
+      if (log_num != log_number_) {
+        buffer_.remove_prefix(header_size + length);
+        return kOldRecord;
+      }
+    }
+
     if (type == kZeroType && length == 0) {
       // Skip zero length record without reporting any drops since
       // such records are produced by the mmap based writing code in