Fix rollback when rollback to tail_lsn.

conanfile.py

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

src/include/homestore/logstore/log_store.hpp

@@ -225,6 +225,7 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
      * @brief Rollback the given instance to the given sequence number
      *
      * @param to_lsn Sequence number back which logs are to be rollbacked
+     * the to_lsn will be the tail_lsn after rollback.
      * @return True on success
      */
     bool rollback(logstore_seq_num_t to_lsn);

src/lib/logstore/log_store.cpp

@@ -283,13 +283,17 @@ void HomeLogStore::flush(logstore_seq_num_t upto_lsn) {
 }
 
 bool HomeLogStore::rollback(logstore_seq_num_t to_lsn) {
-    // Validate if the lsn to which it is rolledback to is not truncated.
-    auto ret = m_records.status(to_lsn + 1);
-    if (ret.is_out_of_range) {
-        HS_LOG_ASSERT(false, "Attempted to rollback to {} which is already truncated", to_lsn);
+    //Fast path
+    if (to_lsn == m_tail_lsn.load()) {
+	return true;
+    }
+
+    if (to_lsn > m_tail_lsn.load() || to_lsn < m_start_lsn.load()) {
+        HS_LOG_ASSERT(false, "Attempted to rollback to {} which is not in the range of [{}, {}]", to_lsn, m_start_lsn.load(), m_tail_lsn.load());
         return false;
     }
 
+    THIS_LOGSTORE_LOG(INFO, "Rolling back to {}, tail {}", to_lsn, m_tail_lsn.load());
     bool do_flush{false};
     do {
         {

src/tests/test_log_dev.cpp

@@ -262,6 +262,9 @@ TEST_F(LogDevTest, Rollback) {
     logstore_seq_num_t cur_lsn = 0;
     kickstart_inserts(log_store, cur_lsn, 500);
 
+    LOGINFO("Step 3.0: Rollback last 0 entries and validate if pre-rollback entries are intact");
+    rollback_validate(log_store, cur_lsn, 0); // Last entry = 500
+
     LOGINFO("Step 3: Rollback last 50 entries and validate if pre-rollback entries are intact");
     rollback_validate(log_store, cur_lsn, 50); // Last entry = 450
 

src/tests/test_log_store_long_run.cpp

@@ -166,11 +166,6 @@ class SampleLogStoreClient {
     }
 
     void rollback_validate(uint32_t num_lsns_to_rollback) {
-
-        if (m_log_store->truncated_upto() == m_log_store->get_contiguous_completed_seq_num(-1)) {
-            // No records to rollback.
-            return;
-        }
         if ((m_cur_lsn - num_lsns_to_rollback - 1) <= m_log_store->get_contiguous_issued_seq_num(-1)) { return; }
         auto const upto_lsn = m_cur_lsn.fetch_sub(num_lsns_to_rollback) - num_lsns_to_rollback - 1;
         m_log_store->rollback(upto_lsn);