[fix] SIGSEGV when VersionEdit in MANIFEST is corrupted

Summary:
This was reported by our customers in task #4295529.

Cause:
* MANIFEST file contains a VersionEdit, which contains file entries whose 'smallest' and 'largest' internal keys are empty. String with zero characters. Root cause of corruption was not investigated. We should report corruption when this happens. However, we currently SIGSEGV.

Here's what happens:
* VersionEdit encodes zero-strings happily and stores them in smallest and largest InternalKeys. InternalKey::Encode() does assert when `rep_.empty()`, but we don't assert in production environemnts. Also, we should never assert as a result of DB corruption.
* As part of our ConsistencyCheck, we call GetLiveFilesMetaData()
* GetLiveFilesMetadata() calls `file->largest.user_key().ToString()`
* user_key() function does: 1. assert(size > 8) (ooops, no assert), 2. returns `Slice(internal_key.data(), internal_key.size() - 8)`
* since `internal_key.size()` is unsigned int, this call translates to `Slice(whatever, 1298471928561892576182756)`. Bazinga.

Fix:
* VersionEdit checks if InternalKey is valid in `VersionEdit::GetInternalKey()`. If it's invalid, returns corruption.

Lessons learned:
* Always keep in mind that even if you `assert()`, production code will continue execution even if assert fails.
* Never `assert` based on DB corruption. Assert only if the code should guarantee that assert can't fail.

Test Plan: dumped offending manifest. Before: assert. Now: corruption

Reviewers: dhruba, haobo, sdong

Reviewed By: dhruba

CC: leveldb

Differential Revision: https://reviews.facebook.net/D18507
This commit is contained in:
Igor Canadi 2014-05-07 16:52:12 -07:00
parent 313b2e5da1
commit 768d424dd9
2 changed files with 6 additions and 1 deletions

View File

@ -146,6 +146,11 @@ class InternalKey {
AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t)); AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t));
} }
bool Valid() const {
ParsedInternalKey parsed;
return ParseInternalKey(Slice(rep_), &parsed);
}
void DecodeFrom(const Slice& s) { rep_.assign(s.data(), s.size()); } void DecodeFrom(const Slice& s) { rep_.assign(s.data(), s.size()); }
Slice Encode() const { Slice Encode() const {
assert(!rep_.empty()); assert(!rep_.empty());

View File

@ -123,7 +123,7 @@ static bool GetInternalKey(Slice* input, InternalKey* dst) {
Slice str; Slice str;
if (GetLengthPrefixedSlice(input, &str)) { if (GetLengthPrefixedSlice(input, &str)) {
dst->DecodeFrom(str); dst->DecodeFrom(str);
return true; return dst->Valid();
} else { } else {
return false; return false;
} }