Add filename to corruption errors

When a corruption happens, report the filename of the file where
corruptions happens.  This will aid in diagnosing database corruption
issues.

Adds a GetName() to all the environment file classes to make it possible
to report a filename downstream.

Author: laanwj

db/db_impl.cc

@@ -432,7 +432,7 @@ Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,
   while (reader.ReadRecord(&record, &scratch) && status.ok()) {
     if (record.size() < 12) {
       reporter.Corruption(record.size(),
-                          Status::Corruption("log record too small"));
+                          Status::Corruption("log record too small", fname));
       continue;
     }
     WriteBatchInternal::SetContents(&batch, record);

db/db_test.cc

@@ -190,6 +190,7 @@ class SpecialEnv : public EnvWrapper {
         }
         return base_->Sync();
       }
+      std::string GetName() const override { return ""; }
     };
     class ManifestFile : public WritableFile {
      private:
@@ -215,6 +216,7 @@ class SpecialEnv : public EnvWrapper {
           return base_->Sync();
         }
       }
+      std::string GetName() const override { return ""; }
     };
 
     if (non_writable_.load(std::memory_order_acquire)) {
@@ -247,6 +249,7 @@ class SpecialEnv : public EnvWrapper {
         counter_->Increment();
         return target_->Read(offset, n, result, scratch);
       }
+      std::string GetName() const override { return ""; }
     };
 
     Status s = target()->NewRandomAccessFile(f, r);

db/fault_injection_test.cc

@@ -114,6 +114,7 @@ class TestWritableFile : public WritableFile {
   Status Close() override;
   Status Flush() override;
   Status Sync() override;
+  std::string GetName() const override { return ""; }
 
  private:
   FileState state_;

db/leveldbutil.cc

@@ -20,6 +20,7 @@ class StdoutPrinter : public WritableFile {
   Status Close() override { return Status::OK(); }
   Status Flush() override { return Status::OK(); }
   Status Sync() override { return Status::OK(); }
+  std::string GetName() const override { return "[stdout]"; }
 };
 
 bool HandleDumpCommand(Env* env, char** files, int num) {

db/log_reader.cc

@@ -176,7 +176,7 @@ bool Reader::ReadRecord(Slice* record, std::string* scratch) {
 uint64_t Reader::LastRecordOffset() { return last_record_offset_; }
 
 void Reader::ReportCorruption(uint64_t bytes, const char* reason) {
-  ReportDrop(bytes, Status::Corruption(reason));
+  ReportDrop(bytes, Status::Corruption(reason, file_->GetName()));
 }
 
 void Reader::ReportDrop(uint64_t bytes, const Status& reason) {

db/log_test.cc

@@ -168,6 +168,7 @@ class LogTest : public testing::Test {
       contents_.append(slice.data(), slice.size());
       return Status::OK();
     }
+    std::string GetName() const override { return ""; }
 
     std::string contents_;
   };
@@ -204,6 +205,7 @@ class LogTest : public testing::Test {
 
       return Status::OK();
     }
+    std::string GetName() const { return ""; }
 
     Slice contents_;
     bool force_error_;

db/repair.cc

@@ -183,7 +183,7 @@ class Repairer {
     while (reader.ReadRecord(&record, &scratch)) {
       if (record.size() < 12) {
         reporter.Corruption(record.size(),
-                            Status::Corruption("log record too small"));
+                            Status::Corruption("log record too small", logname));
         continue;
       }
       WriteBatchInternal::SetContents(&batch, record);

helpers/memenv/memenv.cc

@@ -177,6 +177,7 @@ class SequentialFileImpl : public SequentialFile {
     return Status::OK();
   }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
   uint64_t pos_;
@@ -193,6 +194,7 @@ class RandomAccessFileImpl : public RandomAccessFile {
     return file_->Read(offset, n, result, scratch);
   }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
 };
@@ -209,6 +211,7 @@ class WritableFileImpl : public WritableFile {
   Status Flush() override { return Status::OK(); }
   Status Sync() override { return Status::OK(); }
 
+  virtual std::string GetName() const override { return "[memenv]"; }
  private:
   FileState* file_;
 };

include/leveldb/env.h

@@ -246,6 +246,9 @@ class LEVELDB_EXPORT SequentialFile {
   //
   // REQUIRES: External synchronization
   virtual Status Skip(uint64_t n) = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // A file abstraction for randomly reading the contents of a file.
@@ -269,6 +272,9 @@ class LEVELDB_EXPORT RandomAccessFile {
   // Safe for concurrent use by multiple threads.
   virtual Status Read(uint64_t offset, size_t n, Slice* result,
                       char* scratch) const = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // A file abstraction for sequential writing.  The implementation
@@ -287,6 +293,9 @@ class LEVELDB_EXPORT WritableFile {
   virtual Status Close() = 0;
   virtual Status Flush() = 0;
   virtual Status Sync() = 0;
+
+  // Get a name for the file, only for error reporting
+  virtual std::string GetName() const = 0;
 };
 
 // An interface for writing log messages.

table/format.cc

@@ -84,7 +84,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
   }
   if (contents.size() != n + kBlockTrailerSize) {
     delete[] buf;
-    return Status::Corruption("truncated block read");
+    return Status::Corruption("truncated block read", file->GetName());
   }
 
   // Check the crc of the type and the block contents
@@ -94,7 +94,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
     const uint32_t actual = crc32c::Value(data, n + 1);
     if (actual != crc) {
       delete[] buf;
-      s = Status::Corruption("block checksum mismatch");
+      s = Status::Corruption("block checksum mismatch", file->GetName());
       return s;
     }
   }
@@ -121,7 +121,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
       size_t ulength = 0;
       if (!port::Snappy_GetUncompressedLength(data, n, &ulength)) {
         delete[] buf;
-        return Status::Corruption("corrupted snappy compressed block length");
+        return Status::Corruption("corrupted snappy compressed block length", file->GetName());
       }
       char* ubuf = new char[ulength];
       if (!port::Snappy_Uncompress(data, n, ubuf)) {
@@ -145,7 +145,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
       if (!port::Zstd_Uncompress(data, n, ubuf)) {
         delete[] buf;
         delete[] ubuf;
-        return Status::Corruption("corrupted zstd compressed block contents");
+        return Status::Corruption("corrupted zstd compressed block contents", file->GetName());
       }
       delete[] buf;
       result->data = Slice(ubuf, ulength);
@@ -155,7 +155,7 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
     }
     default:
       delete[] buf;
-      return Status::Corruption("bad block type");
+      return Status::Corruption("bad block type", file->GetName());
   }
 
   return Status::OK();

table/table_test.cc

@@ -103,6 +103,7 @@ class StringSink : public WritableFile {
     return Status::OK();
   }
 
+  std::string GetName() const override { return ""; }
  private:
   std::string contents_;
 };
@@ -129,6 +130,7 @@ class StringSource : public RandomAccessFile {
     return Status::OK();
   }
 
+  std::string GetName() const { return ""; }
  private:
   std::string contents_;
 };

util/env_posix.cc

@@ -163,6 +163,8 @@ class PosixSequentialFile final : public SequentialFile {
     return Status::OK();
   }
 
+  virtual std::string GetName() const override { return filename_; }
+
  private:
   const int fd_;
   const std::string filename_;
@@ -223,6 +225,8 @@ class PosixRandomAccessFile final : public RandomAccessFile {
     return status;
   }
 
+  virtual std::string GetName() const override { return filename_; }
+
  private:
   const bool has_permanent_fd_;  // If false, the file is opened on every read.
   const int fd_;                 // -1 if has_permanent_fd_ is false.
@@ -267,6 +271,8 @@ class PosixMmapReadableFile final : public RandomAccessFile {
     return Status::OK();
   }
 
+  virtual std::string GetName() const override { return filename_; }
+
  private:
   char* const mmap_base_;
   const size_t length_;
@@ -454,6 +460,8 @@ class PosixWritableFile final : public WritableFile {
     return Basename(filename).starts_with("MANIFEST");
   }
 
+  virtual std::string GetName() const override { return filename_; }
+
   // buf_[0, pos_ - 1] contains data to be written to fd_.
   char buf_[kWritableFileBufferSize];
   size_t pos_;

util/env_windows.cc

@@ -193,6 +193,8 @@ class WindowsSequentialFile : public SequentialFile {
     return Status::OK();
   }
 
+  std::string GetName() const override { return filename_; }
+
  private:
   const ScopedHandle handle_;
   const std::string filename_;
@@ -225,6 +227,8 @@ class WindowsRandomAccessFile : public RandomAccessFile {
     return Status::OK();
   }
 
+  std::string GetName() const override { return filename_; }
+
  private:
   const ScopedHandle handle_;
   const std::string filename_;
@@ -256,6 +260,8 @@ class WindowsMmapReadableFile : public RandomAccessFile {
     return Status::OK();
   }
 
+  std::string GetName() const override { return filename_; }
+
  private:
   char* const mmap_base_;
   const size_t length_;
@@ -325,6 +331,8 @@ class WindowsWritableFile : public WritableFile {
     return Status::OK();
   }
 
+  std::string GetName() const override { return filename_; }
+
  private:
   Status FlushBuffer() {
     Status status = WriteUnbuffered(buf_, pos_);