Merge pull request #2785 from nkumar04/BC-24107

Fix a bug in build db checkpoint metadata from persistence

Author: nkumar04

bftengine/include/bftengine/DbCheckpointManager.hpp

@@ -180,6 +180,7 @@ class DbCheckpointManager {
   void updateDbCheckpointMetadata();
   void updateLastCmdInfo(const SeqNum&, const std::optional<Timestamp>&);
   void removeDbCheckpointFuture(CheckpointId);
+  void builMetadataFromFileSystem();
   void updateMetrics();
   InternalBftClient* client_{nullptr};
   std::atomic<bool> stopped_ = false;

bftengine/src/bftengine/DbCheckpointManager.cpp

@@ -119,42 +119,7 @@ void DbCheckpointManager::init() {
   // if any dbcheckpoints are present in the file system.
   if (isMetadataErased_) {
     // update metadata for dbcheckpoints from file system
-    const auto& checkpointDir = dbClient_->getCheckpointPath();
-    _fs::path path(checkpointDir);
-    std::vector<_fs::directory_entry> allDBCheckpoints;
-
-    try {
-      if (_fs::exists(path)) {
-        uint64_t lastCreatedDbCheckpoint = 0;
-        for (const auto& entry : _fs::directory_iterator(checkpointDir)) {
-          const auto filenameStr = entry.path().filename().string();
-          if (_fs::is_directory(entry)) {
-            allDBCheckpoints.push_back(entry);
-            const CheckpointId& checkPointId = std::stoi(filenameStr);
-            const auto blockId = checkPointId;
-
-            _fs::file_time_type ftime = _fs::last_write_time(entry.path());
-            std::time_t cftime = decltype(ftime)::clock::to_time_t(ftime);
-            const std::chrono::seconds creationTime = std::chrono::seconds(cftime);
-            {
-              std::scoped_lock lock(lockLastDbCheckpointDesc_);
-              lastCreatedCheckpointMetadata_.emplace(DbCheckpointMetadata::DbCheckPointDescriptor{
-                  checkPointId, creationTime, blockId, lastCheckpointSeqNum_});
-              dbCheckptMetadata_.dbCheckPoints_.insert({checkPointId, lastCreatedCheckpointMetadata_.value()});
-            }
-
-            // extract the last created BlockId from all created dbCheckpoints.
-            lastCreatedDbCheckpoint = std::max(lastCreatedDbCheckpoint, checkPointId);
-            numOfDbCheckpointsCreated_++;
-          }
-        }
-        lastDbCheckpointBlockId_.Get().Set(lastCreatedDbCheckpoint);
-        metrics_.UpdateAggregator();
-      }
-      updateMetrics();
-    } catch (std::exception& e) {
-      LOG_WARN(getLogger(), "Failed to update checkpoints metadata from checkpoint directory" << e.what());
-    }
+    builMetadataFromFileSystem();
   } else {
     // check if there is chkpt data in persistence
     loadCheckpointDataFromPersistence();
@@ -473,4 +438,48 @@ std::map<uint64_t, uint64_t> DbCheckpointManager::getDbSize() {
   }
   return dbSizeMap;
 }
+void DbCheckpointManager::builMetadataFromFileSystem() {
+  // update metadata for dbcheckpoints from file system
+  const auto& checkpointDir = dbClient_->getCheckpointPath();
+  _fs::path path(checkpointDir);
+
+  try {
+    if (_fs::exists(path)) {
+      for (const auto& entry : _fs::directory_iterator(checkpointDir)) {
+        const auto filenameStr = entry.path().filename().string();
+        // directory name is last block id in the db checkpoint
+        // and must be a valid numeric string
+        if (filenameStr.find_first_not_of("0123456789") != string::npos) {
+          LOG_WARN(getLogger(), "Invalid file or directory:" << filenameStr << " found in checkpoint folder");
+          continue;
+        }
+        if (_fs::is_directory(entry)) {
+          const CheckpointId& checkPointId = std::stoi(filenameStr);
+          const auto blockId = checkPointId;
+          _fs::file_time_type ftime = _fs::last_write_time(entry.path());
+          std::time_t cftime = decltype(ftime)::clock::to_time_t(ftime);
+          const std::chrono::seconds creationTime = std::chrono::seconds(cftime);
+          {
+            std::scoped_lock lock(lockLastDbCheckpointDesc_);
+            const auto checkPointDescriptor = DbCheckpointMetadata::DbCheckPointDescriptor{
+                checkPointId, creationTime, blockId, lastCheckpointSeqNum_};
+            dbCheckptMetadata_.dbCheckPoints_.insert({checkPointId, checkPointDescriptor});
+          }
+        }
+      }
+
+      if (auto it = dbCheckptMetadata_.dbCheckPoints_.rbegin(); it != dbCheckptMetadata_.dbCheckPoints_.rend()) {
+        lastCreatedCheckpointMetadata_ = it->second;
+        dbCheckptMetadata_.lastCmdTimestamp_ = it->second.creationTimeSinceEpoch_;
+        lastCheckpointCreationTime_ = dbCheckptMetadata_.lastCmdTimestamp_;
+        lastDbCheckpointBlockId_.Get().Set(it->second.lastBlockId_);
+        metrics_.UpdateAggregator();
+        updateMetrics();
+      }
+      updateDbCheckpointMetadata();
+    }
+  } catch (std::exception& e) {
+    LOG_WARN(getLogger(), "Failed to update checkpoints metadata from checkpoint directory" << e.what());
+  }
+}
 }  // namespace bftEngine::impl

tests/apollo/test_skvbc_dbsnapshot.py

@@ -728,15 +728,22 @@ async def test_verify_dbcheckpoints_after_scale_with_restart(self, bft_network):
         # It will fetch all dbcheckpoints created before and after scale with restart.
         client = bft_network.random_client()
         op = operator.Operator(bft_network.config, client, bft_network.builddir)
+        latest_checkpoint_id = 0
+        snapshot_rep = await op.state_snapshot_req()
+        resp = cmf_msgs.ReconfigurationResponse.deserialize(snapshot_rep)[0]
+        snapshot_id = resp.response.data.snapshot_id
         rep = await op.get_dbcheckpoint_info_request(bft=False)
         data = cmf_msgs.ReconfigurationResponse.deserialize(rep)[0]
-        self.assertTrue(data.success)
+        self.assertTrue(data.success)        
         for r in client.get_rsi_replies().values():
             res = cmf_msgs.ReconfigurationResponse.deserialize(r)
             self.assertEqual(len(res[0].response.db_checkpoint_info), 2)
             dbcheckpoint_info_list = res[0].response.db_checkpoint_info
             self.assertTrue(any(dbcheckpoint_info.block_id ==
                                 300 for dbcheckpoint_info in dbcheckpoint_info_list))
+            for dbcheckpoint_info in dbcheckpoint_info_list:
+                latest_checkpoint_id = max(latest_checkpoint_id, dbcheckpoint_info.block_id)
+            assert(latest_checkpoint_id == snapshot_id)
 
 
     @with_trio
@@ -948,6 +955,7 @@ async def test_restore_from_snapshot_of_other(self, bft_network, tracker):
         await bft_network.wait_for_consensus_path(path_type=ConsensusPathType.OPTIMISTIC_FAST,
                                                   run_ops=lambda: skvbc.send_n_kvs_sequentially(DB_CHECKPOINT_WIN_SIZE),
                                                   threshold=5)
+                                                  
 
     async def validate_stop_on_wedge_point(self, bft_network, skvbc, fullWedge=False):
         with log.start_action(action_type="validate_stop_on_stable_checkpoint") as action: