improve logic in epoch end

Author: krypdkat

LoggingEventProcessor.cpp

@@ -971,20 +971,26 @@ void verifyLoggingEvent()
         // exit all requesters
         // serve slower nodes 30 more minutes before officially switching epoch
         Logger::get()->info("Received END_EPOCH message. Serving 30 minutes and then closing BOB");
-        SLEEP(1000ULL * gTimeToWaitEpochEnd); // 30 minutes
-        gStopFlag.store(true);
-        // the endTick tick is a virtual tick, we need to migrate its data to new keys:
-        uint32_t endTick = lastQuorumTick + 1; // the system just "borrow" this tick index
 
-        db_rename("tick_log_range:" + std::to_string(endTick),
-                      "end_epoch:tick_log_range:"+std::to_string(gCurrentProcessingEpoch));
-        db_rename("log_ranges:" + std::to_string(endTick),
-                      "end_epoch:log_ranges:"+std::to_string(gCurrentProcessingEpoch));
         std::string key = "end_epoch_tick:" + std::to_string(gCurrentProcessingEpoch);
+        uint32_t endTick = lastQuorumTick + 1; // the system just "borrow" this tick index
         db_insert_u32(key, endTick);
+        // copy log range struct
+        db_copy("log_ranges:" + std::to_string(endTick), "end_epoch:log_ranges:"+std::to_string(gCurrentProcessingEpoch));
+        // copy log range meta data per tick
+        db_hcopy("tick_log_range:" + std::to_string(endTick), "end_epoch:tick_log_range:"+std::to_string(gCurrentProcessingEpoch));
         // end epoch tick is a virtual tick for logging, we set it back to lastQuorumTick
         db_update_field("db_status", "latest_event_tick", std::to_string(lastQuorumTick));
         db_insert_u32("verified_history:" + std::to_string(gCurrentProcessingEpoch), lastQuorumTick); // update historical tracker
+        SLEEP(1000ULL * gTimeToWaitEpochEnd); // default: 30 minutes
+        gStopFlag.store(true);
+        // the endTick tick is a virtual tick, we need to migrate its data to new keys:
+        // these operation are needed when it does seamless transition because in new epoch
+        // the init tick will be the same as this end epoch tick
+        db_rename("tick_log_range:" + std::to_string(endTick),
+                  "backup_end_epoch:tick_log_range:"+std::to_string(gCurrentProcessingEpoch));
+        db_rename("log_ranges:" + std::to_string(endTick),
+                  "backup_end_epoch:log_ranges:"+std::to_string(gCurrentProcessingEpoch));
     }
     Logger::get()->info("verifyLoggingEvent stopping gracefully.");
 }

database/db.cpp

@@ -1267,6 +1267,34 @@ bool db_rename(const std::string &key1, const std::string &key2) {
     }
 }
 
+bool db_copy(const std::string &key1, const std::string &key2) {
+    if (!g_redis) return false;
+    try {
+        auto val = g_redis->get(key1);
+        if (!val) return false;
+        g_redis->set(key2, *val);
+        return true;
+    } catch (const sw::redis::Error &e) {
+        Logger::get()->error("Redis error in db_copy: {}\n", e.what());
+        return false;
+    }
+}
+
+bool db_hcopy(const std::string &key1, const std::string &key2) {
+    if (!g_redis) return false;
+    try {
+        std::unordered_map<std::string, std::string> fields;
+        g_redis->hgetall(key1, std::inserter(fields, fields.begin()));
+        if (!fields.empty()) {
+            g_redis->hmset(key2, fields.begin(), fields.end());
+        }
+        return true;
+    } catch (const sw::redis::Error &e) {
+        Logger::get()->error("Redis error in db_hcopy: {}\n", e.what());
+        return false;
+    }
+}
+
 bool db_insert_u32(const std::string key, uint32_t value) {
     if (!g_redis) return false;
     try {

database/db.h

@@ -515,4 +515,7 @@ bool db_move_logs_to_kvrocks_by_range(uint16_t epoch, long long fromLogId, long
 bool db_delete_transaction(std::string hash);
 bool db_delete_logs(uint16_t epoch, long long start, long long end);
 
-bool db_get_endepoch_log_range_info(const uint16_t epoch, long long &start, long long &length, LogRangesPerTxInTick &lr);
+bool db_get_endepoch_log_range_info(const uint16_t epoch, long long &start, long long &length, LogRangesPerTxInTick &lr);
+
+bool db_copy(const std::string &key1, const std::string &key2);
+bool db_hcopy(const std::string &key1, const std::string &key2);