AccretionDB

// NEW: create-before-delete
auto new_wal = std::make_unique<WAL>(next_id);
new_wal->fsync();
std::swap(wal_, new_wal);
new_wal.reset(); // destructor closes old FD
std::filesystem::remove(old_path); // safe

std::set<std::string> seen_keys;
for (auto& [key, ptr] : lsm_newest_to_oldest) {
  if (seen_keys.count(key)) continue;
  seen_keys.insert(key);
  if (!ptr.is_tombstone()) live.push_back({key, ptr});
}

// OLD
chunk_size += key.size() + 16;

// NEW — matches actual serialization layout
// uint32_t file_id(4) + uint64_t offset(8) + uint32_t length(4) = 20
chunk_size += key.size() + 20;

// OLD: correlated
uint64_t h1 = MurmurHash64A(key, 0x9747b28c);
uint64_t h2 = MurmurHash64A(key, 0x12345678);

// NEW: derived double hashing
uint64_t base = MurmurHash64A(key, 0x9747b28c);
uint64_t h1 = base;
uint64_t h2 = (base >> 33) | (base << 31);

// OLD: destructor races with remove_all
KVStore db(bench_dir);
run_benchmark(db);
std::filesystem::remove_all(bench_dir); // CRASH

// NEW: explicit scope
{
  KVStore db(bench_dir);
  run_benchmark(db);
} // destructor closes all FDs
std::filesystem::remove_all(bench_dir); // safe

for (auto& [key, value] : live_values) {
  db.put(key, value); // increments user_bytes_written
  // subtract internal bytes to keep WA honest
  db.subtract_user_bytes(key.size() + value.size());
}

// OLD: only data section
uint32_t crc = crc32(data_bytes);

// NEW: data + bloom together
uint32_t crc = crc32(data_bytes);
crc = crc32_extend(crc, bloom_bytes);
footer.checksum = crc;

// DEADLOCK: outer lock held, inner re-locks same mutex
bool VersionSet::log_and_apply(VersionEdit* edit) {
  std::lock_guard<std::mutex> lock(mutex_); // outer
  // ... disk IO ...
  {
    std::lock_guard<std::mutex> lock(mutex_); // DEADLOCK
    pending_outputs_.erase(meta.sequence);
  }
}

// FIX: remove the inner lock_guard entirely
bool VersionSet::log_and_apply(VersionEdit* edit) {
  std::lock_guard<std::mutex> lock(mutex_); // outer protects everything
  // ... disk IO ...
  pending_outputs_.erase(meta.sequence); // no nested lock needed
}