db/log_reader: handle modified CRCs

If we detect a CRC error on the first record of a log file, check to see
whether the CRC has the log_number mixed in.  If so, we set the recycled_
flag, and thereafter assume that the CRC will have log_number mixed in.
This raises our probability of a false-positive CRC to 2/2^32 only
for the first record, but leaves it at 1/2^32 for the rest of each file.

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/db/log_reader.cc b/db/log_reader.cc
index d9f681f312af09dca5e6b191f6cc876a05994d1c..1ba35ac7162fa1cf3866d54ead1dc6d12e7e6d9c 100644 (file)
--- a/db/log_reader.cc
+++ b/db/log_reader.cc
@@ -33,6 +33,7 @@ Reader::Reader(const DBOptions *opt,
       buffer_(),
       eof_(false),
       read_error_(false),
+      recycled_(false),
       eof_offset_(0),
       last_record_offset_(0),
       end_of_buffer_offset_(0),
@@ -281,7 +282,9 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) {
       size_t drop_size = buffer_.size();
       buffer_.clear();
       if (!eof_) {
-        ReportCorruption(drop_size, "bad record length");
+       if (!recycled_) {
+         ReportCorruption(drop_size, "bad record length");
+       }
         return kBadRecord;
       }
       // If the end of the file has been reached without reading |length| bytes
@@ -302,16 +305,39 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) {
 
     // Check crc
     if (checksum_) {
-      uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header));
-      uint32_t actual_crc = crc32c::Value(header + 6, 1 + length);
+      uint32_t expected_crc, actual_crc;
+      bool try_recycled = false;
+      if (!recycled_) {
+       // Simple CRC (no log number)
+       expected_crc = crc32c::Unmask(DecodeFixed32(header));
+       actual_crc = crc32c::Value(header + 6, 1 + length);
+       if (last_record_offset_ == 0 && actual_crc != expected_crc) {
+         try_recycled = true;
+       }
+      }
+      if (recycled_ || try_recycled) {
+       uint32_t stored_crc = crc32c::Unmask(DecodeFixed32(header));
+       expected_crc = stored_crc ^ log_number_;
+       actual_crc = crc32c::Value(header + 6, 1 + length);
+       if (try_recycled && db_options_)
+         Log(InfoLogLevel::INFO_LEVEL, db_options_->info_log,
+             "ReadPhysicalRecord file is recycled; using alt CRC\n");
+       if (try_recycled && actual_crc == expected_crc) {
+         // We failed the normal CRC but we matched a recycled CRC.. this
+         // must be a recycled file.
+         recycled_ = true;
+       }
+      }
       if (actual_crc != expected_crc) {
         // Drop the rest of the buffer since "length" itself may have
         // been corrupted and if we trust it, we could find some
         // fragment of a real log record that just happens to look
         // like a valid log record.
-        size_t drop_size = buffer_.size();
         buffer_.clear();
-        ReportCorruption(drop_size, "checksum mismatch");
+        if (!recycled_) {
+         size_t drop_size = buffer_.size();
+         ReportCorruption(drop_size, "checksum mismatch");
+       }
         return kBadRecord;
       }
     }

diff --git a/db/log_reader.h b/db/log_reader.h
index f24083139424a9157a5b9371fd316acf4c7d9ede..45e945d2b28ac3aca71837cfd0b20d54e08d21c2 100644 (file)
--- a/db/log_reader.h
+++ b/db/log_reader.h
@@ -94,6 +94,7 @@ class Reader {
   Slice buffer_;
   bool eof_;   // Last Read() indicated EOF by returning < kBlockSize
   bool read_error_;   // Error occurred while reading from file
+  bool recycled_;  // true if current file was recycled
 
   // Offset of the file position indicator within the last block when an
   // EOF was detected.