From cd4ea675e3d58a2e271c944956d11c8c55bf12df Mon Sep 17 00:00:00 2001
From: Peter Dillinger <peterd@fb.com>
Date: Fri, 19 Nov 2021 17:30:12 -0800
Subject: [PATCH] Fix backward compatibility with 2.5 through 2.7 (#9189)

Summary:
A bug in https://github.com/facebook/rocksdb/issues/9163 can cause checksum verification to fail if
parsing a properties block fails.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/9189

Test Plan:
check_format_compatible.sh (never quite works locally but
this particular case seems fixed using variants of SHORT_TEST=1).
And added new unit test case.

Reviewed By: ajkr

Differential Revision: D32574626

Pulled By: pdillinger

fbshipit-source-id: 6fa5c8595737b71a3c3d011a52daf6d6c08715d7
---
 db/db_table_properties_test.cc                | 25 +++++++++++++++++++
 .../block_based/block_based_table_builder.cc  |  7 ++++--
 table/meta_blocks.cc                          | 11 ++++----
 3 files changed, 36 insertions(+), 7 deletions(-)

diff --git a/db/db_table_properties_test.cc b/db/db_table_properties_test.cc
index 308b95cef..ef3ab7c48 100644
--- a/db/db_table_properties_test.cc
+++ b/db/db_table_properties_test.cc
@@ -171,6 +171,31 @@ TEST_F(DBTablePropertiesTest, GetPropertiesOfAllTablesTest) {
   SyncPoint::GetInstance()->DisableProcessing();
 }
 
+TEST_F(DBTablePropertiesTest, InvalidIgnored) {
+  // RocksDB versions 2.5 - 2.7 generate some properties that Block considers
+  // invalid in some way. This approximates that.
+
+  // Inject properties block data that Block considers invalid
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlockBasedTableBuilder::WritePropertiesBlock:BlockData",
+      [&](void* block_data) {
+        *reinterpret_cast<Slice*>(block_data) = Slice("X");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Build file
+  for (int i = 0; i < 10; ++i) {
+    ASSERT_OK(db_->Put(WriteOptions(), ToString(i), "val"));
+  }
+  ASSERT_OK(db_->Flush(FlushOptions()));
+
+  SyncPoint::GetInstance()->DisableProcessing();
+
+  // Not crashing is good enough
+  TablePropertiesCollection props;
+  ASSERT_OK(db_->GetPropertiesOfAllTables(&props));
+}
+
 TEST_F(DBTablePropertiesTest, CreateOnDeletionCollectorFactory) {
   ConfigOptions options;
   options.ignore_unsupported_options = false;
diff --git a/table/block_based/block_based_table_builder.cc b/table/block_based/block_based_table_builder.cc
index f7b272154..2624b091a 100644
--- a/table/block_based/block_based_table_builder.cc
+++ b/table/block_based/block_based_table_builder.cc
@@ -1715,8 +1715,11 @@ void BlockBasedTableBuilder::WritePropertiesBlock(
                                          rep_->ioptions.logger,
                                          &property_block_builder);
 
-    WriteRawBlock(property_block_builder.Finish(), kNoCompression,
-                  &properties_block_handle, BlockType::kProperties);
+    Slice block_data = property_block_builder.Finish();
+    TEST_SYNC_POINT_CALLBACK(
+        "BlockBasedTableBuilder::WritePropertiesBlock:BlockData", &block_data);
+    WriteRawBlock(block_data, kNoCompression, &properties_block_handle,
+                  BlockType::kProperties);
   }
   if (ok()) {
 #ifndef NDEBUG
diff --git a/table/meta_blocks.cc b/table/meta_blocks.cc
index 20d6e354d..d544be365 100644
--- a/table/meta_blocks.cc
+++ b/table/meta_blocks.cc
@@ -251,6 +251,9 @@ Status ReadTablePropertiesHelper(
     return s;
   }
 
+  // Unfortunately, Block::size() might not equal block_contents.data.size(),
+  // and Block hides block_contents
+  uint64_t block_size = block_contents.data.size();
   Block properties_block(std::move(block_contents));
   DataBlockIter iter;
   properties_block.NewDataIterator(BytewiseComparator(),
@@ -377,8 +380,7 @@ Status ReadTablePropertiesHelper(
   // (See write_global_seqno comment above)
   if (s.ok() && footer.GetBlockTrailerSize() > 0) {
     s = VerifyBlockChecksum(footer.checksum(), properties_block.data(),
-                            properties_block.size(), file->file_name(),
-                            handle.offset());
+                            block_size, file->file_name(), handle.offset());
     if (s.IsCorruption()) {
       const auto seqno_pos_iter = new_table_properties->properties_offsets.find(
           ExternalSstFilePropertyNames::kGlobalSeqno);
@@ -387,9 +389,8 @@ Status ReadTablePropertiesHelper(
                             block_fetcher.GetBlockSizeWithTrailer());
         uint64_t global_seqno_offset = seqno_pos_iter->second - handle.offset();
         EncodeFixed64(&tmp_buf[static_cast<size_t>(global_seqno_offset)], 0);
-        s = VerifyBlockChecksum(footer.checksum(), tmp_buf.data(),
-                                properties_block.size(), file->file_name(),
-                                handle.offset());
+        s = VerifyBlockChecksum(footer.checksum(), tmp_buf.data(), block_size,
+                                file->file_name(), handle.offset());
       }
     }
   }