src/leveldb/db/corruption_test.cc

   1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4
   5 #include "leveldb/db.h"
   6
   7 #include <errno.h>
   8 #include <fcntl.h>
   9 #include <sys/stat.h>
  10 #include <sys/types.h>
  11 #include "leveldb/cache.h"
  12 #include "leveldb/env.h"
  13 #include "leveldb/table.h"
  14 #include "leveldb/write_batch.h"
  15 #include "db/db_impl.h"
  16 #include "db/filename.h"
  17 #include "db/log_format.h"
  18 #include "db/version_set.h"
  19 #include "util/logging.h"
  20 #include "util/testharness.h"
  21 #include "util/testutil.h"
  22
  23 namespace leveldb {
  24
  25 static const int kValueSize = 1000;
  26
  27 class CorruptionTest {
  28  public:
  29   test::ErrorEnv env_;
  30   std::string dbname_;
  31   Cache* tiny_cache_;
  32   Options options_;
  33   DB* db_;
  34
  35   CorruptionTest() {
  36     tiny_cache_ = NewLRUCache(100);
  37     options_.env = &env_;
  38     options_.block_cache = tiny_cache_;
  39     dbname_ = test::TmpDir() + "/db_test";
  40     DestroyDB(dbname_, options_);
  41
  42     db_ = NULL;
  43     options_.create_if_missing = true;
  44     Reopen();
  45     options_.create_if_missing = false;
  46   }
  47
  48   ~CorruptionTest() {
  49      delete db_;
  50      DestroyDB(dbname_, Options());
  51      delete tiny_cache_;
  52   }
  53
  54   Status TryReopen() {
  55     delete db_;
  56     db_ = NULL;
  57     return DB::Open(options_, dbname_, &db_);
  58   }
  59
  60   void Reopen() {
  61     ASSERT_OK(TryReopen());
  62   }
  63
  64   void RepairDB() {
  65     delete db_;
  66     db_ = NULL;
  67     ASSERT_OK(::leveldb::RepairDB(dbname_, options_));
  68   }
  69
  70   void Build(int n) {
  71     std::string key_space, value_space;
  72     WriteBatch batch;
  73     for (int i = 0; i < n; i++) {
  74       //if ((i % 100) == 0) fprintf(stderr, "@ %d of %d\n", i, n);
  75       Slice key = Key(i, &key_space);
  76       batch.Clear();
  77       batch.Put(key, Value(i, &value_space));
  78       WriteOptions options;
  79       // Corrupt() doesn't work without this sync on windows; stat reports 0 for
  80       // the file size.
  81       if (i == n - 1) {
  82         options.sync = true;
  83       }
  84       ASSERT_OK(db_->Write(options, &batch));
  85     }
  86   }
  87
  88   void Check(int min_expected, int max_expected) {
  89     int next_expected = 0;
  90     int missed = 0;
  91     int bad_keys = 0;
  92     int bad_values = 0;
  93     int correct = 0;
  94     std::string value_space;
  95     Iterator* iter = db_->NewIterator(ReadOptions());
  96     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
  97       uint64_t key;
  98       Slice in(iter->key());
  99       if (in == "" || in == "~") {
 100         // Ignore boundary keys.
 101         continue;
 102       }
 103       if (!ConsumeDecimalNumber(&in, &key) ||
 104           !in.empty() ||
 105           key < next_expected) {
 106         bad_keys++;
 107         continue;
 108       }
 109       missed += (key - next_expected);
 110       next_expected = key + 1;
 111       if (iter->value() != Value(key, &value_space)) {
 112         bad_values++;
 113       } else {
 114         correct++;
 115       }
 116     }
 117     delete iter;
 118
 119     fprintf(stderr,
 120             "expected=%d..%d; got=%d; bad_keys=%d; bad_values=%d; missed=%d\n",
 121             min_expected, max_expected, correct, bad_keys, bad_values, missed);
 122     ASSERT_LE(min_expected, correct);
 123     ASSERT_GE(max_expected, correct);
 124   }
 125
 126   void Corrupt(FileType filetype, int offset, int bytes_to_corrupt) {
 127     // Pick file to corrupt
 128     std::vector<std::string> filenames;
 129     ASSERT_OK(env_.GetChildren(dbname_, &filenames));
 130     uint64_t number;
 131     FileType type;
 132     std::string fname;
 133     int picked_number = -1;
 134     for (size_t i = 0; i < filenames.size(); i++) {
 135       if (ParseFileName(filenames[i], &number, &type) &&
 136           type == filetype &&
 137           int(number) > picked_number) {  // Pick latest file
 138         fname = dbname_ + "/" + filenames[i];
 139         picked_number = number;
 140       }
 141     }
 142     ASSERT_TRUE(!fname.empty()) << filetype;
 143
 144     struct stat sbuf;
 145     if (stat(fname.c_str(), &sbuf) != 0) {
 146       const char* msg = strerror(errno);
 147       ASSERT_TRUE(false) << fname << ": " << msg;
 148     }
 149
 150     if (offset < 0) {
 151       // Relative to end of file; make it absolute
 152       if (-offset > sbuf.st_size) {
 153         offset = 0;
 154       } else {
 155         offset = sbuf.st_size + offset;
 156       }
 157     }
 158     if (offset > sbuf.st_size) {
 159       offset = sbuf.st_size;
 160     }
 161     if (offset + bytes_to_corrupt > sbuf.st_size) {
 162       bytes_to_corrupt = sbuf.st_size - offset;
 163     }
 164
 165     // Do it
 166     std::string contents;
 167     Status s = ReadFileToString(Env::Default(), fname, &contents);
 168     ASSERT_TRUE(s.ok()) << s.ToString();
 169     for (int i = 0; i < bytes_to_corrupt; i++) {
 170       contents[i + offset] ^= 0x80;
 171     }
 172     s = WriteStringToFile(Env::Default(), contents, fname);
 173     ASSERT_TRUE(s.ok()) << s.ToString();
 174   }
 175
 176   int Property(const std::string& name) {
 177     std::string property;
 178     int result;
 179     if (db_->GetProperty(name, &property) &&
 180         sscanf(property.c_str(), "%d", &result) == 1) {
 181       return result;
 182     } else {
 183       return -1;
 184     }
 185   }
 186
 187   // Return the ith key
 188   Slice Key(int i, std::string* storage) {
 189     char buf[100];
 190     snprintf(buf, sizeof(buf), "%016d", i);
 191     storage->assign(buf, strlen(buf));
 192     return Slice(*storage);
 193   }
 194
 195   // Return the value to associate with the specified key
 196   Slice Value(int k, std::string* storage) {
 197     Random r(k);
 198     return test::RandomString(&r, kValueSize, storage);
 199   }
 200 };
 201
 202 TEST(CorruptionTest, Recovery) {
 203   Build(100);
 204   Check(100, 100);
 205   Corrupt(kLogFile, 19, 1);      // WriteBatch tag for first record
 206   Corrupt(kLogFile, log::kBlockSize + 1000, 1);  // Somewhere in second block
 207   Reopen();
 208
 209   // The 64 records in the first two log blocks are completely lost.
 210   Check(36, 36);
 211 }
 212
 213 TEST(CorruptionTest, RecoverWriteError) {
 214   env_.writable_file_error_ = true;
 215   Status s = TryReopen();
 216   ASSERT_TRUE(!s.ok());
 217 }
 218
 219 TEST(CorruptionTest, NewFileErrorDuringWrite) {
 220   // Do enough writing to force minor compaction
 221   env_.writable_file_error_ = true;
 222   const int num = 3 + (Options().write_buffer_size / kValueSize);
 223   std::string value_storage;
 224   Status s;
 225   for (int i = 0; s.ok() && i < num; i++) {
 226     WriteBatch batch;
 227     batch.Put("a", Value(100, &value_storage));
 228     s = db_->Write(WriteOptions(), &batch);
 229   }
 230   ASSERT_TRUE(!s.ok());
 231   ASSERT_GE(env_.num_writable_file_errors_, 1);
 232   env_.writable_file_error_ = false;
 233   Reopen();
 234 }
 235
 236 TEST(CorruptionTest, TableFile) {
 237   Build(100);
 238   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 239   dbi->TEST_CompactMemTable();
 240   dbi->TEST_CompactRange(0, NULL, NULL);
 241   dbi->TEST_CompactRange(1, NULL, NULL);
 242
 243   Corrupt(kTableFile, 100, 1);
 244   Check(90, 99);
 245 }
 246
 247 TEST(CorruptionTest, TableFileRepair) {
 248   options_.block_size = 2 * kValueSize;  // Limit scope of corruption
 249   options_.paranoid_checks = true;
 250   Reopen();
 251   Build(100);
 252   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 253   dbi->TEST_CompactMemTable();
 254   dbi->TEST_CompactRange(0, NULL, NULL);
 255   dbi->TEST_CompactRange(1, NULL, NULL);
 256
 257   Corrupt(kTableFile, 100, 1);
 258   RepairDB();
 259   Reopen();
 260   Check(95, 99);
 261 }
 262
 263 TEST(CorruptionTest, TableFileIndexData) {
 264   Build(10000);  // Enough to build multiple Tables
 265   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 266   dbi->TEST_CompactMemTable();
 267
 268   Corrupt(kTableFile, -2000, 500);
 269   Reopen();
 270   Check(5000, 9999);
 271 }
 272
 273 TEST(CorruptionTest, MissingDescriptor) {
 274   Build(1000);
 275   RepairDB();
 276   Reopen();
 277   Check(1000, 1000);
 278 }
 279
 280 TEST(CorruptionTest, SequenceNumberRecovery) {
 281   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v1"));
 282   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v2"));
 283   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v3"));
 284   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v4"));
 285   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v5"));
 286   RepairDB();
 287   Reopen();
 288   std::string v;
 289   ASSERT_OK(db_->Get(ReadOptions(), "foo", &v));
 290   ASSERT_EQ("v5", v);
 291   // Write something.  If sequence number was not recovered properly,
 292   // it will be hidden by an earlier write.
 293   ASSERT_OK(db_->Put(WriteOptions(), "foo", "v6"));
 294   ASSERT_OK(db_->Get(ReadOptions(), "foo", &v));
 295   ASSERT_EQ("v6", v);
 296   Reopen();
 297   ASSERT_OK(db_->Get(ReadOptions(), "foo", &v));
 298   ASSERT_EQ("v6", v);
 299 }
 300
 301 TEST(CorruptionTest, CorruptedDescriptor) {
 302   ASSERT_OK(db_->Put(WriteOptions(), "foo", "hello"));
 303   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 304   dbi->TEST_CompactMemTable();
 305   dbi->TEST_CompactRange(0, NULL, NULL);
 306
 307   Corrupt(kDescriptorFile, 0, 1000);
 308   Status s = TryReopen();
 309   ASSERT_TRUE(!s.ok());
 310
 311   RepairDB();
 312   Reopen();
 313   std::string v;
 314   ASSERT_OK(db_->Get(ReadOptions(), "foo", &v));
 315   ASSERT_EQ("hello", v);
 316 }
 317
 318 TEST(CorruptionTest, CompactionInputError) {
 319   Build(10);
 320   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 321   dbi->TEST_CompactMemTable();
 322   const int last = config::kMaxMemCompactLevel;
 323   ASSERT_EQ(1, Property("leveldb.num-files-at-level" + NumberToString(last)));
 324
 325   Corrupt(kTableFile, 100, 1);
 326   Check(5, 9);
 327
 328   // Force compactions by writing lots of values
 329   Build(10000);
 330   Check(10000, 10000);
 331 }
 332
 333 TEST(CorruptionTest, CompactionInputErrorParanoid) {
 334   options_.paranoid_checks = true;
 335   options_.write_buffer_size = 512 << 10;
 336   Reopen();
 337   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 338
 339   // Make multiple inputs so we need to compact.
 340   for (int i = 0; i < 2; i++) {
 341     Build(10);
 342     dbi->TEST_CompactMemTable();
 343     Corrupt(kTableFile, 100, 1);
 344     env_.SleepForMicroseconds(100000);
 345   }
 346   dbi->CompactRange(NULL, NULL);
 347
 348   // Write must fail because of corrupted table
 349   std::string tmp1, tmp2;
 350   Status s = db_->Put(WriteOptions(), Key(5, &tmp1), Value(5, &tmp2));
 351   ASSERT_TRUE(!s.ok()) << "write did not fail in corrupted paranoid db";
 352 }
 353
 354 TEST(CorruptionTest, UnrelatedKeys) {
 355   Build(10);
 356   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
 357   dbi->TEST_CompactMemTable();
 358   Corrupt(kTableFile, 100, 1);
 359
 360   std::string tmp1, tmp2;
 361   ASSERT_OK(db_->Put(WriteOptions(), Key(1000, &tmp1), Value(1000, &tmp2)));
 362   std::string v;
 363   ASSERT_OK(db_->Get(ReadOptions(), Key(1000, &tmp1), &v));
 364   ASSERT_EQ(Value(1000, &tmp2).ToString(), v);
 365   dbi->TEST_CompactMemTable();
 366   ASSERT_OK(db_->Get(ReadOptions(), Key(1000, &tmp1), &v));
 367   ASSERT_EQ(Value(1000, &tmp2).ToString(), v);
 368 }
 369
 370 }  // namespace leveldb
 371
 372 int main(int argc, char** argv) {
 373   return leveldb::test::RunAllTests();
 374 }