Fix rollback when rollback to tail_lsn.

previously we dont properly handle this case,
which will error out in https://github.com/eBay/HomeStore/blob/2080ad42e31a9ddf0a41d3a112db4d1c020c45ee/src/lib/logstore/log_store.cpp#L315
as the to_lsn + 1 is invalid in this case.

Signed-off-by: Xiaoxi Chen <[email protected]>

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/lib/logstore/log_store.cpp

@@ -283,13 +283,24 @@ void HomeLogStore::flush(logstore_seq_num_t upto_lsn) {
 }
 
 bool HomeLogStore::rollback(logstore_seq_num_t to_lsn) {
+    //Fast path
+    if (to_lsn == m_tail_lsn.load()) {
+	return true;
+    }
+
+    if (to_lsn > m_tail_lsn.load()) {
+        HS_LOG_ASSERT(false, "Attempted to rollback to {} which is larger than tail_lsn {}", to_lsn, m_tail_lsn.load());
+        return false;
+    }
+
     // Validate if the lsn to which it is rolledback to is not truncated.
-    auto ret = m_records.status(to_lsn + 1);
+    auto ret = m_records.status(to_lsn);
     if (ret.is_out_of_range) {
         HS_LOG_ASSERT(false, "Attempted to rollback to {} which is already truncated", to_lsn);
         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);