dbwrapper.cpp raw

   1  // Copyright (c) The Bitcoin Core developers
   2  // Distributed under the MIT software license, see the accompanying
   3  // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4  
   5  #include <dbwrapper.h>
   6  #include <compat/byteswap.h>
   7  #include <random.h>
   8  #include <sync.h>
   9  #include <test/fuzz/FuzzedDataProvider.h>
  10  #include <test/fuzz/fuzz.h>
  11  #include <test/fuzz/util.h>
  12  #include <test/util/random.h>
  13  #include <test/util/setup_common.h>
  14  #include <util/byte_units.h>
  15  #include <util/check.h>
  16  #include <util/threadpool.h>
  17  
  18  #include <leveldb/env.h>
  19  #include <leveldb/helpers/memenv/memenv.h>
  20  
  21  #include <algorithm>
  22  #include <cassert>
  23  #include <cstdint>
  24  #include <deque>
  25  #include <functional>
  26  #include <future>
  27  #include <latch>
  28  #include <map>
  29  #include <memory>
  30  #include <numeric>
  31  #include <optional>
  32  #include <set>
  33  #include <span>
  34  #include <string>
  35  #include <tuple>
  36  #include <vector>
  37  
  38  namespace {
  39  
  40  /**
  41   * A leveldb::Env that wraps a memenv and captures scheduled background
  42   * work (compaction) instead of dispatching to a real thread. The fuzz
  43   * harness calls RunOne() or DrainWork() at fuzzer-chosen points to
  44   * execute it, giving deterministic control over when compaction
  45   * interleaves with foreground operations.
  46   *
  47   * Deadlock prevention: LevelDB's MakeRoomForWrite blocks on a condition
  48   * variable when the previous immutable memtable is still awaiting compaction,
  49   * or when the L0 file count hits kL0_StopWritesTrigger. Since both conditions
  50   * can only be resolved by the (deferred) background work, the harness drains
  51   * all pending work before every write to avoid a single-threaded deadlock.
  52   * Callers must also DrainWork() before destroying the CDBWrapper, since the
  53   * leveldb destructor waits for any pending background work to complete.
  54   *
  55   * The same reasoning rules out exercising DBOptions::force_compact under
  56   * this env, because CompactRange(nullptr, nullptr) blocks waiting for
  57   * background work that is queued on the (blocked) foreground thread. The
  58   * sibling dbwrapper_threaded target covers that path.
  59   */
  60  class DeterministicEnv final : public leveldb::EnvWrapper
  61  {
  62      using WorkFunction = void (*)(void*);
  63  
  64      struct Work {
  65          WorkFunction function;
  66          void* arg;
  67      };
  68  
  69      Mutex m_mutex;
  70      std::deque<Work> m_queue GUARDED_BY(m_mutex);
  71  
  72  public:
  73      explicit DeterministicEnv(leveldb::Env* base) : EnvWrapper(base) {}
  74  
  75      void Schedule(WorkFunction function, void* arg) override EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
  76      {
  77          LOCK(m_mutex);
  78          m_queue.push_back({function, arg});
  79      }
  80  
  81      /** Execute one pending background task. The task may schedule a
  82       *  successor which is left pending for a later call. */
  83      bool RunOne() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
  84      {
  85          Work work;
  86          {
  87              LOCK(m_mutex);
  88              if (m_queue.empty()) return false;
  89              work = m_queue.front();
  90              m_queue.pop_front();
  91          }
  92          work.function(work.arg);
  93          return true;
  94      }
  95  
  96      /** Execute pending background tasks until none remain. */
  97      void DrainWork() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) { while (RunOne()) {} }
  98  };
  99  
 100  constexpr size_t MAX_VALUE_LEN{4096};
 101  constexpr uint8_t MAX_VALUE_MULTIPLIER{8};
 102  constexpr size_t WRITE_BATCH_HEADER{12}; // See kHeader in db/write_batch.cc
 103  
 104  /** Mirror of CDBWrapper::OBFUSCATION_KEY, the fixed key under which leveldb
 105   *  stores the obfuscation metadata entry when obfuscation is enabled. */
 106  const std::string OBFUSCATION_KEY{"\000obfuscate_key", 14};
 107  
 108  /** Generate a deterministic value from key and size. The fuzz input picks
 109   *  a 16-bit length (up to MAX_VALUE_LEN) and an 8-bit multiplier so that a
 110   *  small amount of fuzz input can produce a wide range of value sizes. */
 111  std::vector<uint8_t> MakeValue(uint16_t key, uint32_t size)
 112  {
 113      std::vector<uint8_t> v(size);
 114      std::iota(v.begin(), v.end(), static_cast<uint8_t>(key ^ (key >> 8)));
 115      return v;
 116  }
 117  
 118  /** Equivalent to leveldb::BytewiseComparator() on 2-byte little-endian
 119   *  serialized uint16_t keys, while keeping the oracle keyed by uint16_t. */
 120  struct LevelDBBytewiseU16Cmp {
 121      bool operator()(uint16_t a, uint16_t b) const { return internal_bswap_16(a) < internal_bswap_16(b); }
 122  };
 123  
 124  /** key → value-size map ordered by LevelDB's bytewise comparator. */
 125  using Oracle = std::map<uint16_t, uint32_t, LevelDBBytewiseU16Cmp>;
 126  
 127  struct FailUnserialize {
 128      template <typename Stream>
 129      void Unserialize(Stream&) { throw std::ios_base::failure{"always fail"}; }
 130  };
 131  
 132  uint16_t ConsumeKey(FuzzedDataProvider& provider) { return provider.ConsumeIntegral<uint16_t>(); }
 133  uint32_t ConsumeValueSize(FuzzedDataProvider& provider)
 134  {
 135      const uint16_t len{provider.ConsumeIntegralInRange<uint16_t>(0, MAX_VALUE_LEN)};
 136      const uint8_t multiplier{provider.ConsumeIntegralInRange<uint8_t>(1, MAX_VALUE_MULTIPLIER)};
 137      return static_cast<uint32_t>(len) * multiplier;
 138  }
 139  
 140  /** Verify that the DB iterator matches the oracle, handling the obfuscation
 141   *  metadata entry (stored under a non-uint16_t key) when obfuscation is on. */
 142  void VerifyIterator(CDBWrapper& dbw, const Oracle& oracle,
 143                      bool obfuscate, std::optional<uint16_t> seek_key = std::nullopt)
 144  {
 145      const std::unique_ptr<CDBIterator> it{dbw.NewIterator()};
 146      auto oracle_it{seek_key ? oracle.lower_bound(*seek_key) : oracle.begin()};
 147      if (seek_key) {
 148          it->Seek(*seek_key);
 149      } else {
 150          it->SeekToFirst();
 151      }
 152      for (; it->Valid(); it->Next()) {
 153          uint16_t db_key;
 154          assert(it->GetKey(db_key));
 155          if (oracle_it != oracle.end() && db_key == oracle_it->first) {
 156              std::vector<uint8_t> db_value;
 157              assert(it->GetValue(db_value));
 158              assert(db_value == MakeValue(db_key, oracle_it->second));
 159              ++oracle_it;
 160          } else {
 161              assert(obfuscate);
 162              std::string key_str;
 163              assert(it->GetKey(key_str));
 164              assert(key_str == OBFUSCATION_KEY);
 165          }
 166      }
 167      assert(oracle_it == oracle.end());
 168  }
 169  
 170  /** Maximum number of concurrent reader threads in dbwrapper_concurrent_reads. */
 171  constexpr size_t MAX_READ_WORKERS{8};
 172  
 173  /** Maximum number of queries each worker executes in dbwrapper_concurrent_reads. */
 174  constexpr size_t MAX_READ_QUERIES_PER_WORKER{128};
 175  
 176  ThreadPool g_read_pool{"dbfuzz"};
 177  
 178  void StartReadPoolIfNeeded()
 179  {
 180      if (!g_read_pool.WorkersCount()) g_read_pool.Start(MAX_READ_WORKERS);
 181  }
 182  
 183  /** Build randomized DBParams from the fuzz input, shared by all targets. */
 184  DBParams ConsumeDBParams(FuzzedDataProvider& provider, leveldb::Env* testing_env,
 185                           bool obfuscate, DBOptions options = {})
 186  {
 187      return DBParams{
 188          .path = "dbwrapper_fuzz",
 189          .cache_bytes = provider.ConsumeIntegralInRange<size_t>(64 << 10, 1_MiB),
 190          .obfuscate = obfuscate,
 191          .bloom_filter = provider.ConsumeBool(),
 192          .options = options,
 193          .testing_env = testing_env,
 194          .max_file_size = provider.ConsumeBool()
 195              ? DBWRAPPER_MAX_FILE_SIZE
 196              : provider.ConsumeIntegralInRange<size_t>(1_MiB, 4_MiB),
 197      };
 198  }
 199  
 200  template <typename DrainWorkFn, typename RunOneFn>
 201  void TestDbWrapper(FuzzedDataProvider& provider,
 202                        leveldb::Env* testing_env,
 203                        DrainWorkFn drain_work,
 204                        RunOneFn run_one,
 205                        bool allow_force_compact)
 206  {
 207      SeedRandomStateForTest(SeedRand::ZEROS);
 208  
 209      const bool obfuscate{provider.ConsumeBool()};
 210  
 211      const auto make_db{[&](DBOptions options = {}) {
 212          return std::make_unique<CDBWrapper>(ConsumeDBParams(provider, testing_env, obfuscate, options));
 213      }};
 214      std::unique_ptr<CDBWrapper> dbw{make_db()};
 215  
 216      // Oracle: key → value size. Content is reconstructed via MakeValue().
 217      Oracle oracle;
 218  
 219      LIMITED_WHILE(provider.ConsumeBool(), 1'000)
 220      {
 221          CallOneOf(
 222              provider,
 223              // --- Mutations ---
 224              [&] {
 225                  const auto key{ConsumeKey(provider)};
 226                  const auto size{ConsumeValueSize(provider)};
 227                  drain_work();
 228                  dbw->Write(key, MakeValue(key, size), /*fSync=*/provider.ConsumeBool());
 229                  oracle[key] = size;
 230              },
 231              [&] {
 232                  const auto key{ConsumeKey(provider)};
 233                  drain_work();
 234                  dbw->Erase(key, /*fSync=*/provider.ConsumeBool());
 235                  oracle.erase(key);
 236              },
 237              [&] {
 238                  CDBBatch batch{*dbw};
 239                  std::map<uint16_t, uint32_t> batch_writes;
 240                  std::set<uint16_t> batch_erases;
 241                  const auto fill{[&] {
 242                      LIMITED_WHILE(provider.ConsumeBool(), 20)
 243                      {
 244                          const auto key{ConsumeKey(provider)};
 245                          if (provider.ConsumeBool()) {
 246                              const auto size{ConsumeValueSize(provider)};
 247                              batch.Write(key, MakeValue(key, size));
 248                              batch_writes[key] = size;
 249                              batch_erases.erase(key);
 250                          } else {
 251                              batch.Erase(key);
 252                              batch_erases.insert(key);
 253                              batch_writes.erase(key);
 254                          }
 255                      }
 256                  }};
 257                  fill();
 258                  if (provider.ConsumeBool()) {
 259                      assert(batch.ApproximateSize() >= WRITE_BATCH_HEADER);
 260                      batch.Clear();
 261                      assert(batch.ApproximateSize() == WRITE_BATCH_HEADER);
 262                      batch_writes.clear();
 263                      batch_erases.clear();
 264                      fill();
 265                  }
 266                  drain_work();
 267                  dbw->WriteBatch(batch, /*fSync=*/provider.ConsumeBool());
 268                  for (const auto& [k, v] : batch_writes) oracle[k] = v;
 269                  for (const auto& k : batch_erases) oracle.erase(k);
 270              },
 271              [&] {
 272                  drain_work();
 273                  dbw.reset();
 274                  DBOptions options{};
 275                  if (allow_force_compact && provider.ConsumeBool()) {
 276                      options.force_compact = true;
 277                  }
 278                  dbw = make_db(options);
 279                  VerifyIterator(*dbw, oracle, obfuscate);
 280              },
 281              // --- Reads ---
 282              [&] {
 283                  const auto key{ConsumeKey(provider)};
 284                  std::vector<uint8_t> value;
 285                  const bool found{dbw->Read(key, value)};
 286                  if (const auto it{oracle.find(key)}; it != oracle.end()) {
 287                      assert(found && value == MakeValue(key, it->second));
 288                  } else {
 289                      assert(!found);
 290                  }
 291              },
 292              [&] {
 293                  const auto key{ConsumeKey(provider)};
 294                  assert(dbw->Exists(key) == oracle.contains(key));
 295              },
 296              [&] {
 297                  uint16_t key{};
 298                  if (!oracle.empty() && provider.ConsumeBool()) {
 299                      auto it{oracle.begin()};
 300                      std::advance(it, provider.ConsumeIntegralInRange<size_t>(0, oracle.size() - 1));
 301                      key = it->first;
 302                  } else {
 303                      key = ConsumeKey(provider);
 304                  }
 305                  FailUnserialize wrong_type;
 306                  assert(!dbw->Read(key, wrong_type));
 307              },
 308              [&] {
 309                  const auto seek_key{provider.ConsumeBool()
 310                                          ? std::optional<uint16_t>{ConsumeKey(provider)}
 311                                          : std::nullopt};
 312                  VerifyIterator(*dbw, oracle, obfuscate, seek_key);
 313              },
 314              // --- Stats ---
 315              [&] {
 316                  assert(dbw->IsEmpty() == (oracle.empty() && !obfuscate));
 317              },
 318              [&] {
 319                  const auto [k1, k2]{std::minmax({ConsumeKey(provider), ConsumeKey(provider)}, LevelDBBytewiseU16Cmp{})};
 320                  const size_t estimate_size{dbw->EstimateSize(k1, k2)};
 321                  if (k1 == k2) assert(estimate_size == 0);
 322              },
 323              [&] {
 324                  (void)dbw->DynamicMemoryUsage();
 325              },
 326              // --- Compaction control (no-op when run_one is no-op) ---
 327              [&] {
 328                  run_one();
 329              });
 330      }
 331  
 332      VerifyIterator(*dbw, oracle, obfuscate);
 333      drain_work();
 334  }
 335  
 336  } // namespace
 337  
 338  FUZZ_TARGET(dbwrapper, .init = [] { static auto setup{MakeNoLogFileContext<>()}; })
 339  {
 340      FuzzedDataProvider provider{buffer.data(), buffer.size()};
 341  
 342      const auto memenv{std::unique_ptr<leveldb::Env>{leveldb::NewMemEnv(leveldb::Env::Default())}};
 343      DeterministicEnv det_env{memenv.get()};
 344      TestDbWrapper(
 345          provider, &det_env,
 346          [&] { det_env.DrainWork(); },
 347          [&] { return det_env.RunOne(); },
 348          /*allow_force_compact=*/false);
 349  }
 350  
 351  FUZZ_TARGET(dbwrapper_threaded, .init = [] { static auto setup{MakeNoLogFileContext<>()}; })
 352  {
 353      FuzzedDataProvider provider{buffer.data(), buffer.size()};
 354  
 355      const auto memenv{std::unique_ptr<leveldb::Env>{leveldb::NewMemEnv(leveldb::Env::Default())}};
 356      TestDbWrapper(
 357          provider, memenv.get(),
 358          /*drain_work=*/[] {},
 359          /*run_one=*/[] { return false; },
 360          /*allow_force_compact=*/true);
 361  }
 362  
 363  FUZZ_TARGET(dbwrapper_concurrent_reads, .init = [] { static auto setup{MakeNoLogFileContext<>()}; })
 364  {
 365      StartReadPoolIfNeeded();
 366      SeedRandomStateForTest(SeedRand::ZEROS);
 367  
 368      FuzzedDataProvider provider{buffer.data(), buffer.size()};
 369  
 370      const auto memenv{std::unique_ptr<leveldb::Env>{leveldb::NewMemEnv(leveldb::Env::Default())}};
 371      DeterministicEnv det_env{memenv.get()};
 372  
 373      CDBWrapper db{ConsumeDBParams(provider, &det_env, /*obfuscate=*/provider.ConsumeBool())};
 374  
 375      // Seed the DB. Drain work after small batches so we don't deadlock on a
 376      // scheduled compaction.
 377      const size_t num_entries{provider.ConsumeIntegralInRange<size_t>(100, 5'000)};
 378      std::vector<uint16_t> keys;
 379      keys.reserve(num_entries);
 380      Oracle oracle;
 381      constexpr size_t SEED_BATCH_SIZE{400};
 382      for (size_t start{0}; start < num_entries; start += SEED_BATCH_SIZE) {
 383          CDBBatch batch{db};
 384          const size_t end{std::min(start + SEED_BATCH_SIZE, num_entries)};
 385          for (size_t i{start}; i < end; ++i) {
 386              const auto k{ConsumeKey(provider)};
 387              const auto size{ConsumeValueSize(provider)};
 388              batch.Write(k, MakeValue(k, size));
 389              keys.push_back(k);
 390              oracle[k] = size;
 391          }
 392          det_env.DrainWork();
 393          db.WriteBatch(batch, /*fSync=*/true);
 394      }
 395  
 396      while (provider.ConsumeBool() && det_env.RunOne()) {}
 397  
 398      // Build query list from seeded and random keys.
 399      const size_t num_queries{provider.ConsumeIntegralInRange<size_t>(1, 2'000)};
 400      enum class ReadOp { Read, Exists, IteratorSeek };
 401      std::vector<std::tuple<ReadOp, uint16_t>> queries;
 402      queries.reserve(num_queries);
 403      for (size_t i{0}; i < num_queries; ++i) {
 404          const auto op{provider.PickValueInArray({ReadOp::Read, ReadOp::Exists, ReadOp::IteratorSeek})};
 405          const uint16_t key{provider.ConsumeBool()
 406                                 ? keys[provider.ConsumeIntegralInRange<size_t>(0, keys.size() - 1)]
 407                                 : ConsumeKey(provider)};
 408          queries.emplace_back(op, key);
 409      }
 410  
 411  
 412      // Workers + main thread synchronize on the latch so all reads start together.
 413      std::latch start_latch{static_cast<ptrdiff_t>(MAX_READ_WORKERS + 1)};
 414      std::vector<std::function<void()>> tasks(MAX_READ_WORKERS);
 415      FastRandomContext rng{ConsumeUInt256(provider)};
 416      std::ranges::generate(tasks, [&] {
 417          return [&, seed = rng.rand256()] {
 418              FastRandomContext thread_rng{seed};
 419              std::vector<size_t> order(queries.size());
 420              std::iota(order.begin(), order.end(), size_t{0});
 421              std::ranges::shuffle(order, thread_rng);
 422              const size_t queries_to_run{std::min(queries.size(), MAX_READ_QUERIES_PER_WORKER)};
 423              std::vector<uint8_t> v;
 424              std::string key_str;
 425              start_latch.arrive_and_wait();
 426              const std::unique_ptr<CDBIterator> it{db.NewIterator()};
 427              // Every read must agree with the oracle, the source of truth.
 428              for (const auto i : std::span{order}.first(queries_to_run)) {
 429                  const auto& [op, key] = queries[i];
 430                  switch (op) {
 431                  case ReadOp::Read:
 432                      if (const auto oit{oracle.find(key)}; oit != oracle.end()) {
 433                          assert(db.Read(key, v) && v == MakeValue(key, oit->second));
 434                      } else {
 435                          assert(!db.Read(key, v));
 436                      }
 437                      break;
 438                  case ReadOp::Exists:
 439                      assert(db.Exists(key) == oracle.contains(key));
 440                      break;
 441                  case ReadOp::IteratorSeek:
 442                      it->Seek(key);
 443                      // Skip the obfuscation metadata entry (a non-uint16_t key) if we land
 444                      // on it, so the result matches the oracle, which only tracks user keys.
 445                      if (it->Valid() && it->GetKey(key_str) && key_str == OBFUSCATION_KEY) it->Next();
 446                      if (const auto oit{oracle.lower_bound(key)}; oit != oracle.end()) {
 447                          assert(it->Valid());
 448                          uint16_t actual_key;
 449                          assert(it->GetKey(actual_key) && actual_key == oit->first);
 450                          assert(it->GetValue(v) && v == MakeValue(actual_key, oit->second));
 451                      } else {
 452                          assert(!it->Valid());
 453                      }
 454                      break;
 455                  }
 456              }
 457          };
 458      });
 459      auto futures{*Assert(g_read_pool.Submit(std::move(tasks)))};
 460  
 461      // Release the workers and immediately run the queued compaction on this
 462      // thread, so compaction races against the concurrent reads.
 463      start_latch.arrive_and_wait();
 464      det_env.DrainWork();
 465  
 466      for (auto& fut : futures) fut.get();
 467      det_env.DrainWork();
 468  }
 469