skip append dummy logs to log dev

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "6.6.8"
+    version = "6.6.9"
 
     homepage = "/~https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/logstore/log_store.hpp

@@ -232,6 +232,7 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
 
     auto start_lsn() const { return m_start_lsn.load(std::memory_order_acquire); }
     auto tail_lsn() const { return m_tail_lsn.load(std::memory_order_acquire); }
+    auto next_lsn() const { return m_next_lsn.load(std::memory_order_acquire); }
 
     nlohmann::json dump_log_store(const log_dump_req& dump_req = log_dump_req());
 

src/lib/logstore/log_dev.cpp

@@ -530,9 +530,10 @@ uint64_t LogDev::truncate() {
         auto lstore = store.log_store;
         if (lstore == nullptr) { continue; }
         auto const [trunc_lsn, trunc_ld_key, tail_lsn] = lstore->truncate_info();
-        if (trunc_lsn == tail_lsn) {
-            THIS_LOGDEV_LOG(DEBUG, "Store_id={} didn't have any writes since last truncation, skipping ", store_id);
-            m_logdev_meta.remove_all_rollback_records(store_id, m_stopped /* persist_now */);
+        if (trunc_ld_key.idx <= m_last_truncate_idx) {
+            THIS_LOGDEV_LOG(DEBUG,
+                            "Store_id={} didn't need to truncate since last truncation, skipping. last_truncate_idx={}",
+                            store_id, m_last_truncate_idx);
             continue;
         }
         HS_DBG_ASSERT_GE(trunc_ld_key.idx, m_last_truncate_idx, "Trying to truncate logid which is already truncated");

src/lib/logstore/log_store.cpp

@@ -189,17 +189,35 @@ void HomeLogStore::truncate(logstore_seq_num_t upto_lsn, bool in_memory_truncate
 
 #endif
 
+    // In normal write and compact path, upto_lsn is expected to be no larger than m_tail_lsn after the flush.
+    // So upto_lsn > m_tail_lsn is expected to exist only in baseline resync path.
+    // In baseline resync path, we truncate all entries up to upto_lsn, and set m_trunc_ld_key to the ld_key of
+    // m_tail_lsn so that log dev can truncate all log records.
+    // We also need to update m_tail_lsn and m_next_lsn to make sure logstore's idx is always = raft's idx - 1.
     if (upto_lsn > m_tail_lsn) {
         THIS_LOGSTORE_LOG(WARN,
-                          "Truncating issued on lsn={} which is greater than tail_lsn={}, truncating upto tail_lsn",
+                          "Truncating issued on lsn={} which is greater than tail_lsn={}",
                           upto_lsn, m_tail_lsn.load(std::memory_order_relaxed));
         m_trunc_ld_key = m_records.at(m_tail_lsn).m_trunc_key;
-        upto_lsn = m_tail_lsn;
+
+        // update m_tail_lsn if it is less than upto_lsn
+        auto current_tail_lsn = m_tail_lsn.load(std::memory_order_relaxed);
+        while (current_tail_lsn < upto_lsn &&
+               !m_tail_lsn.compare_exchange_weak(current_tail_lsn, upto_lsn, std::memory_order_relaxed)) {}
+
+        // update m_next_lsn if it is less than upto_lsn + 1
+        auto current_next_lsn = m_next_lsn.load(std::memory_order_relaxed);
+        while (current_next_lsn < upto_lsn + 1 &&
+               !m_next_lsn.compare_exchange_weak(current_next_lsn, upto_lsn + 1, std::memory_order_relaxed)) {}
+
+        // insert an empty record to make sure m_records has enough size to truncate
+        logdev_key empty_ld_key;
+        m_records.create_and_complete(upto_lsn, logstore_record(empty_ld_key, empty_ld_key));
     } else {
         m_trunc_ld_key = m_records.at(upto_lsn).m_trunc_key;
-        THIS_LOGSTORE_LOG(TRACE, "Truncating logstore upto lsn={} , m_trunc_ld_key index {} offset {}", upto_lsn,
-                          m_trunc_ld_key.idx, m_trunc_ld_key.dev_offset);
     }
+    THIS_LOGSTORE_LOG(TRACE, "Truncating logstore upto lsn={} , m_trunc_ld_key index {} offset {}", upto_lsn,
+                      m_trunc_ld_key.idx, m_trunc_ld_key.dev_offset);
     m_records.truncate(upto_lsn);
     m_start_lsn.store(upto_lsn + 1);
     if (!in_memory_truncate_only) { m_logdev->truncate(); }

src/lib/replication/log_store/home_raft_log_store.cpp

@@ -361,12 +361,10 @@ bool HomeRaftLogStore::compact(ulong compact_lsn) {
         // release this assert if for some use case, we should tolorant this case;
         // for now, don't expect this case to happen.
         // RELEASE_ASSERT(false, "compact_lsn={} is beyond the current max_lsn={}", compact_lsn, cur_max_lsn);
-        REPL_STORE_LOG(DEBUG, "Adding dummy entries during compact from={} upto={}", cur_max_lsn + 1,
-                       to_store_lsn(compact_lsn));
-        // We need to fill the remaining entries with dummy data.
-        for (auto lsn{cur_max_lsn + 1}; lsn <= to_store_lsn(compact_lsn); ++lsn) {
-            append(m_dummy_log_entry);
-        }
+
+        // if compact_lsn is beyond the current max_lsn, it indicates a hole from cur_max_lsn to compact_lsn.
+        // we directly compact and truncate up to compact_lsn assuming there are dummy logs.
+        REPL_STORE_LOG(DEBUG, "Compact with dummy logs from {} to={}", cur_max_lsn + 1, to_store_lsn(compact_lsn));
     }
     m_log_store->truncate(to_store_lsn(compact_lsn));
     return true;

src/tests/test_log_dev.cpp

@@ -201,10 +201,11 @@ class LogDevTest : public ::testing::Test {
         read_all_verify(log_store);
     }
 
-    void truncate_validate(std::shared_ptr< HomeLogStore > log_store, logstore_seq_num_t* last_lsn = nullptr) {
+    void truncate_validate(std::shared_ptr< HomeLogStore > log_store, logstore_seq_num_t* trunc_lsn = nullptr) {
         auto upto = log_store->get_contiguous_completed_seq_num(-1);
-        if (last_lsn) {
-            ASSERT_EQ(upto, *last_lsn);
+        if (trunc_lsn && *trunc_lsn != upto) {
+            LOGWARN("Truncate issued upto {} but real upto lsn in log store is {}", *trunc_lsn, upto);
+            upto = *trunc_lsn;
         }
 
         LOGINFO("truncate_validate upto {}", upto);
@@ -314,7 +315,6 @@ TEST_F(LogDevTest, ReTruncate) {
     auto logdev_id = logstore_service().create_new_logdev();
     s_max_flush_multiple = logstore_service().get_logdev(logdev_id)->get_flush_size_multiple();
     auto log_store = logstore_service().create_new_log_store(logdev_id, false);
-    auto store_id = log_store->get_store_id();
 
     LOGINFO("Step 2: Issue sequential inserts with q depth of 10");
     logstore_seq_num_t cur_lsn = 0;
@@ -337,6 +337,50 @@ TEST_F(LogDevTest, ReTruncate) {
     read_all_verify(log_store);
 }
 
+TEST_F(LogDevTest, TruncateWithOverlappingLSN) {
+    LOGINFO("Step 1: Create a single logstore to start truncate with overlapping LSN test");
+    auto logdev_id = logstore_service().create_new_logdev();
+    s_max_flush_multiple = logstore_service().get_logdev(logdev_id)->get_flush_size_multiple();
+    auto log_store = logstore_service().create_new_log_store(logdev_id, false);
+
+    LOGINFO("Step 2: Insert 500 entries");
+    logstore_seq_num_t cur_lsn = 0;
+    kickstart_inserts(log_store, cur_lsn, 500);
+
+    LOGINFO("Step 3: Read and verify all entries");
+    read_all_verify(log_store);
+
+    LOGINFO("Step 4: Truncate 100 entries");
+    logstore_seq_num_t trunc_lsn = 99;
+    truncate_validate(log_store, &trunc_lsn);
+    ASSERT_EQ(log_store->start_lsn(), trunc_lsn + 1);
+    ASSERT_EQ(log_store->tail_lsn(), 499);
+    ASSERT_EQ(log_store->next_lsn(), 500);
+    ASSERT_EQ(log_store->truncated_upto(), trunc_lsn);
+
+    LOGINFO("Step 5: Read and verify all entries");
+    read_all_verify(log_store);
+
+    LOGINFO("Step 6: Truncate all with overlapping lsn");
+    trunc_lsn = 1999999;
+    truncate_validate(log_store, &trunc_lsn);
+    ASSERT_EQ(log_store->start_lsn(), trunc_lsn + 1);
+    ASSERT_EQ(log_store->tail_lsn(), trunc_lsn);
+    ASSERT_EQ(log_store->next_lsn(), 2000000);
+    ASSERT_EQ(log_store->truncated_upto(), trunc_lsn);
+
+    LOGINFO("Step 7 Read and verify all entries");
+    read_all_verify(log_store);
+
+    LOGINFO("Step 8: Append 500 entries")
+    cur_lsn = log_store->next_lsn();
+    kickstart_inserts(log_store, cur_lsn, 500);
+    ASSERT_EQ(log_store->next_lsn(), 2000500);
+
+    LOGINFO("Step 9: Read and verify all entries");
+    read_all_verify(log_store);
+}
+
 TEST_F(LogDevTest, CreateRemoveLogDev) {
     auto num_logdev = SISL_OPTIONS["num_logdevs"].as< uint32_t >();
     std::vector< std::shared_ptr< HomeLogStore > > log_stores;