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