From 768d424dd97f6c02d09f02c3be5cd4b8c999dfab Mon Sep 17 00:00:00 2001
From: Igor Canadi <icanadi@fb.com>
Date: Wed, 7 May 2014 16:52:12 -0700
Subject: [PATCH] [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
---
 db/dbformat.h      | 5 +++++
 db/version_edit.cc | 2 +-
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/db/dbformat.h b/db/dbformat.h
index 1647661b8..4356dd934 100644
--- a/db/dbformat.h
+++ b/db/dbformat.h
@@ -146,6 +146,11 @@ class InternalKey {
     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()); }
   Slice Encode() const {
     assert(!rep_.empty());
diff --git a/db/version_edit.cc b/db/version_edit.cc
index 24d7f0d9f..2ac35c58c 100644
--- a/db/version_edit.cc
+++ b/db/version_edit.cc
@@ -123,7 +123,7 @@ static bool GetInternalKey(Slice* input, InternalKey* dst) {
   Slice str;
   if (GetLengthPrefixedSlice(input, &str)) {
     dst->DecodeFrom(str);
-    return true;
+    return dst->Valid();
   } else {
     return false;
   }