Possible consistency bug with iterators, snapshots, batched deletes.

We've been using leveldb for a project and have hit what appears to be a consistency bug in leveldb. Without delving into leveldb internals, we've produced a relatively simple program that should be deterministic, but actually fails nondeterministically.

The program below writes entries to a single shard 'A' (for these purposes, our "shards" are single-character prefixes to the keys). It occasionally creates a snapshotted iterator over shard 'A', and processes/deletes all the entries in it, using batched writes.

The bugged behavior we're seeing appears to be that sometimes (rarely) an entry will be deleted from shard 'A', will correctly be missing from the next iteration, but then reappears again. It does not happen to every entry in a batch, just rare isolated entries. This bug occurs a lot more often in a multithreaded program, but still shows up with a single thread.

This description of the bug may be slightly inaccurate or misleading (we're just going from our observations). However, it should be easy to run the C++17 code below for yourself. It's single-threaded, deterministically seeded, and should not exhibit nondeterministic behavior. But the "Mismatch" assertion is thrown at different times each run.

```
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
using namespace std;

string randString(int shard) {
string ret;
ret += char(shard);
for (int i = 0; i < 200; i++) ret += char(rand()%26+'a');
return ret;
}

int main() {
// Use deterministic seed.
srand(0);

// Generate a sequence of operations.
// For shard 'A', we will occasionally snapshot, run over all entries, and delete them
// (while doing miscellaneous gets and puts to other shards).
vector> v;
{
vector curShard;
vector dataShard;
// If you're not regularly seeing a failure you can bump this constant up.
while (v.size() < 5000000) {
if (rand()%100000 == 0) {
// Run through shard 'A', deleting the entries.
// The "get"s and "put"s here are not strictly necessary for the failure to occur,
// but seem to make it more likely.
v.push_back({"write"});
sort(curShard.begin(), curShard.end());
curShard.erase(unique(curShard.begin(), curShard.end()), curShard.end());
v.push_back({"snapshot"});
for (auto const& s : curShard) {
v.push_back({"delete", s});
if (rand()%10 == 0) v.push_back({"get", dataShard[rand()%dataShard.size()]});
if (rand()%500 == 0) {
v.push_back({"put", randString('B'), randString('Z')});
v.push_back({"write"});
}
}
v.push_back({"put", randString('B'), randString('b')});
v.push_back({"write"});
v.push_back({"release"});
curShard.clear();
} else if (rand()%100 == 0) {
string s = randString('Z');
v.push_back({"put", s, randString('a')});
v.push_back({"write"});
dataShard.push_back(s);
} else {
string s = randString('A');
v.push_back({"put", s, ""});
curShard.push_back(s);
}
}
cout << "Generated " << v.size() << " operations." << endl;
}

// Apply operations deterministically to a local leveldb.
std::string dbName = "./randomdb";
std::filesystem::remove_all(dbName);
leveldb::DB* ldb = NULL;
leveldb::Options dbOp;
dbOp.create_if_missing = dbOp.error_if_exists = true;
// A smaller write buffer isn't strictly needed but makes the failure more likely.
dbOp.write_buffer_size = 1 * 1024 * 1024;
leveldb::Status status = leveldb::DB::Open(dbOp, dbName, &ldb);
assert(status.ok());

int snapshots = 0;
leveldb::WriteBatch batch;
for (int vi = 0; vi < v.size(); vi++) {
if (v[vi][0] == "put") {
batch.Put(v[vi][1], v[vi][2]);
} else if (v[vi][0] == "get") {
string value;
status = ldb->Get({}, v[vi][1], &value);
assert(status.ok() || status.IsNotFound());
} else if (v[vi][0] == "write") {
status = ldb->Write({}, &batch);
assert(status.ok());
batch.Clear();
} else if (v[vi][0] == "snapshot") {
snapshots++;
const leveldb::Snapshot* snapshot = ldb->GetSnapshot();
leveldb::Iterator* shardIter = ldb->NewIterator({.snapshot=snapshot});
shardIter->Seek("A");
assert(shardIter->status().ok());

for (vi++; v[vi][0] != "release"; vi++) {
assert(vi < v.size());
if (v[vi][0] == "delete") {
// Within this snapshot, we expect all the entries currently in shard 'A' to
// be seen in order. We delete them as we go.
assert(v[vi][1][0] == 'A');
assert(shardIter->Valid());
leveldb::Slice key = shardIter->key();
string s(key.data(), key.size());
if (s == v[vi][1]) {
batch.Delete(key);
shardIter->Next();
assert(shardIter->status().ok());
} else {
// This assertion seems to blow because occasionally a random shard 'A'
// entry will rise from the dead in a later snapshot.
cout << "Mismatch at: " << vi << endl << v[vi][1] << endl << s << endl;
assert(false);
}
} else if (v[vi][0] == "put") {
batch.Put(v[vi][1], v[vi][2]);
} else if (v[vi][0] == "get") {
string value;
status = ldb->Get({.snapshot=snapshot}, v[vi][1], &value);
assert(status.ok());
} else if (v[vi][0] == "write") {
status = ldb->Write({}, &batch);
assert(status.ok());
batch.Clear();
} else {
assert(false);
}
}

// Shard 'A' is now empty.
assert(!shardIter->Valid() || shardIter->key()[0] != 'A');
delete shardIter;
ldb->ReleaseSnapshot(snapshot);
}
}

delete ldb;
cout << "Verified " << snapshots << " snapshots. No failure." << endl;
}
```