fix db checkpoint async bug (#2998)

The current implementation of the db checkpoint feature has a synchronization bug:
While we take the db checkpoint in the background, we don't align anything with the checkpoint sequence number, i.e. the block number, bft metadata, pending reserved pages, and more.
Once we put more client requests in a different resolution than 150 we start to see a wide set of issues:
For example:
The recovered replica won't start from a stable checkpoint, instead, it starts from the point where the db checkpoint was taken.
In a case where the checkpoint was taken in the middle of another execution phase, we won't have the pending reserved pages to recover correctly.
We trim the block at the point where the db checkpoint was taken, but we don't update the bft metadata accordingly.
Below is an example of part of these issues:
On replica 0, block 302 was created on sequence number 304
0|03-04-2023 10:48:24.125|INFO |skvbctest.replica|post-execution-thread|||304||executeWriteCommand|L:482|ConditionalWrite message handled; writesCounter=302 currBlock=302 | [SQ:1215]
However, on recovery, the recovered replica has the same block that was created on sequence number 305:

5|03-04-2023 10:48:34.293|INFO |skvbctest.replica|post-execution-thread|||305||executeWriteCommand|L:482|ConditionalWrite message handled; writesCounter=1 currBlock=302 | [SQ:3]
In this PR we propose a wide change that fixes the problem.
Till now, the decision of whether to create a db checkpoint was the primary only. If the primary decided that it's time to create the db checkpoint, it sends a bft command whose execution is, creating a DB checkpoint.
(Note that this approach, regardless of the above bugs, is not safe in terms of DOS attacks, a malicious primary can order the replicas to continuously create db checkpoints).
Here we propose a different solution: the decision to create a db checkpoint is based on a deterministic local event (such as how much time has passed since the last created db checkpoint).
This way, once decided a db checkpoint creation callback is registered to the stable sequence number event. Once the replica reaches this stable sequence number, it starts to create the db checkpoint asynchronously, but now the block number is aligned with the sequence number because it was taken right after the sequence number execution.
To make the above feasible, (1) we cannot rely on local timeouts (instead, we consider only the time being received by consensus), (2) the db checkpoint metadata (such as sequence number and timestamp) has to be shared between all replicas (via reserved pages).

Author: Yehonatan Buchnik

bftengine/include/bftengine/DbCheckpointManager.hpp

@@ -34,16 +34,17 @@
 #include "storage/db_interface.h"
 #include "util/filesystem.hpp"
 #include "kv_types.hpp"
+#include "ReservedPagesClient.hpp"
+
 namespace _fs = fs;
 namespace bftEngine::impl {
 using std::chrono::duration_cast;
 using Status = concordUtils::Status;
 using SystemClock = std::chrono::system_clock;
-using Seconds = std::chrono::seconds;
 using InternalBftClient = bftEngine::impl::InternalBFTClient;
 using DbCheckpointId = uint64_t;
 using BlockId = uint64_t;
-class DbCheckpointManager {
+class DbCheckpointManager : public bftEngine::ResPagesClient<DbCheckpointManager, 1> {
  public:
   /***************************************************
    *@Input parameter1: Request sequnce number
@@ -64,7 +65,6 @@ class DbCheckpointManager {
    *@Return: returns a unique db checkpoint id. Else return std::nullopt
    ***************************************************/
   std::optional<CheckpointId> createDbCheckpointAsync(const SeqNum& seqNum,
-                                                      const std::optional<Timestamp>& timestamp,
                                                       const std::optional<DbCheckpointId>& blockId);
 
   /***************************************************
@@ -95,7 +95,7 @@ class DbCheckpointManager {
   //  * the path to the checkpoint
   using PrepareCheckpointCallback = std::function<void(BlockId, const std::string&)>;
 
-  Seconds getLastCheckpointCreationTime() const { return lastCheckpointCreationTime_; }
+  std::chrono::seconds getLastCheckpointCreationTime() const { return lastCheckpointCreationTime_; }
   void initializeDbCheckpointManager(std::shared_ptr<concord::storage::IDBClient> dbClient,
                                      std::shared_ptr<bftEngine::impl::PersistentStorage> p,
                                      std::shared_ptr<concordMetrics::Aggregator> aggregator,
@@ -113,27 +113,19 @@ class DbCheckpointManager {
   void setNextStableSeqNumToCreateSnapshot(const std::optional<SeqNum>& seqNum);
   std::optional<SeqNum> getNextStableSeqNumToCreateSnapshot() const { return nextSeqNumToCreateCheckPt_; }
 
-  static DbCheckpointManager& instance(InternalBftClient* client = nullptr) {
-    static DbCheckpointManager instance(client);
+  static DbCheckpointManager& instance() {
+    static DbCheckpointManager instance;
     return instance;
   }
-  ~DbCheckpointManager() {
+  ~DbCheckpointManager() override {
     stopped_ = true;
     shutdownCond_.notify_all();  // to wake up monitor thread so it can be destructed
     if (monitorThread_.joinable()) monitorThread_.join();
   }
-  void sendInternalCreateDbCheckpointMsg(const SeqNum& seqNum, bool noop);
   BlockId getLastReachableBlock() const;
-  SeqNum getLastStableSeqNum() const;
   void setCheckpointInProcess(bool, concord::kvbc::BlockId) const;
-  void setOnStableSeqNumCb_(std::function<void(SeqNum)> cb) { onStableSeqNumCb_ = cb; }
-  void onStableSeqNum(SeqNum s) {
-    if (onStableSeqNumCb_) onStableSeqNumCb_(s);
-  }
-  void setGetLastStableSeqNumCb(std::function<SeqNum()> cb) { getLastStableSeqNumCb_ = cb; }
   inline void checkAndCreateDbSnapshot(SeqNum seqNum) {
-    if (ReplicaConfig::instance().dbCheckpointFeatureEnabled)
-      createDbCheckpointAsync(seqNum, std::nullopt, std::nullopt);
+    if (ReplicaConfig::instance().dbCheckpointFeatureEnabled) createDbCheckpointAsync(seqNum, std::nullopt);
   }
   void addOnDbCheckpointCreatedCb(std::function<void(SeqNum)> cb) {
     if (cb) onDbCheckpointCreated_.push_back(cb);
@@ -145,26 +137,24 @@ class DbCheckpointManager {
   // only used for apollo test
   std::map<uint64_t, uint64_t> getDbSize();
   void setIsMetadataErased(bool isMetadataErased) { isMetadataErased_ = isMetadataErased; }
+  void saveDbMetadataToReservedPages(SeqNum, std::chrono::seconds timestamp);
+  void loadDbMetadataFromReservedPages();
 
  private:
   logging::Logger getLogger() {
     static logging::Logger logger_(logging::getLogger("concord.bft.db_checkpoint_manager"));
     return logger_;
   }
-  DbCheckpointManager(InternalBftClient* client)
-      : client_(client),
-        metrics_{concordMetrics::Component("rocksdbCheckpoint", std::make_shared<concordMetrics::Aggregator>())},
+  DbCheckpointManager()
+      : metrics_{concordMetrics::Component("rocksdbCheckpoint", std::make_shared<concordMetrics::Aggregator>())},
         maxDbCheckpointCreationTimeMsec_(metrics_.RegisterGauge("maxDbCheckpointCreationTimeInMsecSoFar", 0)),
         lastDbCheckpointSizeInMb_(metrics_.RegisterGauge("lastDbCheckpointSizeInMb", 0)),
         lastDbCheckpointBlockId_(metrics_.RegisterGauge("lastDbCheckpointBlockId", 0)),
         numOfDbCheckpointsCreated_(metrics_.RegisterCounter("numOfDbCheckpointsCreated", 0)) {
     metrics_.Register();
   }
   void init();
-  Status createDbCheckpoint(const DbCheckpointId& checkPointId,
-                            const BlockId& lastBlockId,
-                            const SeqNum& seqNum,
-                            const std::optional<Timestamp>& timestamp);
+  Status createDbCheckpoint(const DbCheckpointId& checkPointId, const BlockId& lastBlockId, const SeqNum& seqNum);
   void removeCheckpoint(const DbCheckpointId& checkPointId);
   void removeAllCheckpoints() const;
   void cleanUp();
@@ -179,11 +169,9 @@ class DbCheckpointManager {
   void checkAndRemove();
   void removeOldestDbCheckpoint();
   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;
   DbCheckpointMetadata dbCheckptMetadata_;
   std::map<CheckpointId, std::future<void>> dbCreateCheckPtFuture_;
@@ -203,9 +191,8 @@ class DbCheckpointManager {
   SeqNum lastCheckpointSeqNum_{0};
   std::optional<DbCheckpointMetadata::DbCheckPointDescriptor> lastCreatedCheckpointMetadata_{std::nullopt};
   std::optional<SeqNum> nextSeqNumToCreateCheckPt_{std::nullopt};
-  std::chrono::seconds lastCheckpointCreationTime_{duration_cast<Seconds>(SystemClock::now().time_since_epoch())};
+  std::chrono::seconds lastCheckpointCreationTime_;
   std::function<void(SeqNum)> onStableSeqNumCb_;
-  std::function<SeqNum()> getLastStableSeqNumCb_;
   std::vector<std::function<void(SeqNum)>> onDbCheckpointCreated_;
   std::string dbCheckPointDirPath_;
   bool isMetadataErased_ = false;
@@ -214,6 +201,12 @@ class DbCheckpointManager {
   concordMetrics::GaugeHandle lastDbCheckpointSizeInMb_;
   concordMetrics::GaugeHandle lastDbCheckpointBlockId_;
   concordMetrics::CounterHandle numOfDbCheckpointsCreated_;
+
+  // The db checkpoint is created as a local decision in the replica once several conditions are met. Hence we need to
+  // share the knowledge about the latest created db checkpoint such that a recovered replica will able to make the same
+  // decision as all other replicas.
+  std::string page_;
+  DbLastCheckpointMetadata shared_metadata_;
 };
 
 }  // namespace bftEngine::impl

bftengine/include/bftengine/DbCheckpointMetadata.hpp

@@ -24,6 +24,20 @@ using CheckpointId = uint64_t;
 using TimeDuration = std::chrono::duration<long>;
 using SeqNum = bftEngine::impl::SeqNum;
 
+// The data about the latest checkpoint is a shared data, hence it is shared via the reserved pages
+struct DbLastCheckpointMetadata : public concord::serialize::SerializableFactory<DbLastCheckpointMetadata> {
+  uint64_t lastCmdSeqNum_{0};
+  TimeDuration lastCmdTimestamp_{0};
+
+  void serializeDataMembers(std::ostream& outStream) const override {
+    serialize(outStream, lastCmdSeqNum_);
+    serialize(outStream, lastCmdTimestamp_);
+  }
+  void deserializeDataMembers(std::istream& inStream) override {
+    deserialize(inStream, lastCmdSeqNum_);
+    deserialize(inStream, lastCmdTimestamp_);
+  }
+};
 struct DbCheckpointMetadata : public concord::serialize::SerializableFactory<DbCheckpointMetadata> {
   struct DbCheckPointDescriptor : public concord::serialize::SerializableFactory<DbCheckPointDescriptor> {
     // unique id for the checkpoint
@@ -54,21 +68,10 @@ struct DbCheckpointMetadata : public concord::serialize::SerializableFactory<DbC
     }
   };
   std::map<CheckpointId, DbCheckPointDescriptor> dbCheckPoints_;
-  uint64_t lastCmdSeqNum_{0};
-  TimeDuration lastCmdTimestamp_{0};
   DbCheckpointMetadata() = default;
-  void serializeDataMembers(std::ostream& outStream) const override {
-    serialize(outStream, lastCmdSeqNum_);
-    serialize(outStream, lastCmdTimestamp_);
-    serialize(outStream, dbCheckPoints_);
-  }
-  void deserializeDataMembers(std::istream& inStream) override {
-    deserialize(inStream, lastCmdSeqNum_);
-    deserialize(inStream, lastCmdTimestamp_);
-    deserialize(inStream, dbCheckPoints_);
-  }
+  void serializeDataMembers(std::ostream& outStream) const override { serialize(outStream, dbCheckPoints_); }
+  void deserializeDataMembers(std::istream& inStream) override { deserialize(inStream, dbCheckPoints_); }
 };
 constexpr size_t DB_CHECKPOINT_METADATA_MAX_SIZE{
-    MAX_ALLOWED_CHECKPOINTS * sizeof(std::pair<CheckpointId, DbCheckpointMetadata::DbCheckPointDescriptor>) +
-    sizeof(DbCheckpointMetadata::lastCmdSeqNum_) + sizeof(DbCheckpointMetadata::lastCmdTimestamp_)};
+    MAX_ALLOWED_CHECKPOINTS * sizeof(std::pair<CheckpointId, DbCheckpointMetadata::DbCheckPointDescriptor>)};
 }  // namespace bftEngine::impl

bftengine/include/bftengine/Replica.hpp

@@ -54,7 +54,7 @@ enum MsgFlag : uint64_t {
   INTERNAL_FLAG = 0x40,
   PUBLISH_ON_CHAIN_OBJECT_FLAG = 0x80,
   CLIENTS_PUB_KEYS_FLAG = 0x100,
-  DB_CHECKPOINT_FLAG = 0x200,
+  NOOP_FLAG = 0x200,
   PRIMARY_ONLY_FLAG = 0x401
 };
 

bftengine/src/bftengine/DbCheckpointManager.cpp

@@ -39,8 +39,7 @@ struct HumanReadable {
 };
 Status DbCheckpointManager::createDbCheckpoint(const CheckpointId& checkPointId,
                                                const BlockId& lastBlockId,
-                                               const SeqNum& seqNum,
-                                               const std::optional<Timestamp>& timestamp) {
+                                               const SeqNum& seqNum) {
   ConcordAssert(dbClient_.get() != nullptr);
   ConcordAssert(ps_.get() != nullptr);
   {
@@ -105,12 +104,10 @@ void DbCheckpointManager::loadCheckpointDataFromPersistence() {
     }
     if (auto it = dbCheckptMetadata_.dbCheckPoints_.rbegin(); it != dbCheckptMetadata_.dbCheckPoints_.rend()) {
       lastCreatedCheckpointMetadata_ = it->second;
-      lastCheckpointSeqNum_ = it->second.lastDbCheckpointSeqNum_;
       lastDbCheckpointBlockId_.Get().Set(it->second.lastBlockId_);
       metrics_.UpdateAggregator();
       LOG_INFO(getLogger(), KVLOG(lastCheckpointSeqNum_));
     }
-    lastCheckpointCreationTime_ = dbCheckptMetadata_.lastCmdTimestamp_;
   }
 }
 
@@ -177,6 +174,27 @@ uint64_t DbCheckpointManager::directorySize(const _fs::path& directory, const bo
   }
   return size;
 }
+void DbCheckpointManager::saveDbMetadataToReservedPages(SeqNum sn, std::chrono::seconds timestamp) {
+  shared_metadata_.lastCmdSeqNum_ = sn;
+  shared_metadata_.lastCmdTimestamp_ = timestamp;
+  std::ostringstream outStream;
+  concord::serialize::Serializable::serialize(outStream, shared_metadata_);
+  auto data = outStream.str();
+  saveReservedPage(0, data.size(), data.data());
+  lastCheckpointCreationTime_ = shared_metadata_.lastCmdTimestamp_;
+}
+
+void DbCheckpointManager::loadDbMetadataFromReservedPages() {
+  if (!loadReservedPage(0, sizeOfReservedPage(), page_.data())) {
+    LOG_WARN(getLogger(), "unable to read db checkpoint shared metadata from reserved pages");
+    return;
+  }
+  std::istringstream inStream;
+  inStream.str(page_);
+  concord::serialize::Serializable::deserialize(inStream, shared_metadata_);
+  lastCheckpointCreationTime_ = shared_metadata_.lastCmdTimestamp_;
+  lastCheckpointSeqNum_ = shared_metadata_.lastCmdSeqNum_;
+}
 
 void DbCheckpointManager::initializeDbCheckpointManager(
     std::shared_ptr<concord::storage::IDBClient> dbClient,
@@ -195,12 +213,9 @@ void DbCheckpointManager::initializeDbCheckpointManager(
   if (getLastBlockIdCb) getLastBlockIdCb_ = getLastBlockIdCb;
   if (checkpointInProcessCb) checkpointInProcessCb_ = checkpointInProcessCb;
   prepareCheckpointCb_ = prepareCheckpointCb;
+  page_.resize(sizeOfReservedPage());
+  loadDbMetadataFromReservedPages();
   if (ReplicaConfig::instance().dbCheckpointFeatureEnabled) {
-    // in case of upgrade, we need to set the lastStableCheckpointSeqNum from persistence
-    // instead of 0. if there is a db checkpoint metatdata present in persistence
-    // then this gets overriden by the lastCheckpointSeqNum loaded from persistence
-    if (getLastStableSeqNumCb_) lastCheckpointSeqNum_ = getLastStableSeqNumCb_();
-
     init();
   } else {
     // db checkpoint is disabled. Cleanup metadata and checkpoints created if any
@@ -232,18 +247,11 @@ DbCheckpointManager::CheckpointState DbCheckpointManager::getCheckpointState(Che
 }
 
 std::optional<CheckpointId> DbCheckpointManager::createDbCheckpointAsync(const SeqNum& seqNum,
-                                                                         const std::optional<Timestamp>& timestamp,
                                                                          const std::optional<DbCheckpointId>& blockId) {
   if (seqNum <= lastCheckpointSeqNum_) {
     LOG_ERROR(getLogger(), "createDb checkpoint failed." << KVLOG(seqNum, lastCheckpointSeqNum_));
     return std::nullopt;
   }
-  // We can have some replicas configured to not create db checkpoint. This is because,
-  // backup functionality can be supported with a minimum of (f+1) replicas also
-  // However, we need to store the last request seqNum and timestamp because, all replicas do consensus
-  // to start the command execution at some seqNum but only replicas with non-zero numOfDbCheckpoints config
-  // actually creates db checkpoint. Hence, primary needs to know what was the last cmd seqNum and timestamp
-  updateLastCmdInfo(seqNum, timestamp);
   if (ReplicaConfig::instance().getmaxNumberOfDbCheckpoints() == 0) {
     LOG_WARN(getLogger(),
              "createDb checkpoint failed. Max allowed db checkpoint is configured to "
@@ -271,8 +279,8 @@ std::optional<CheckpointId> DbCheckpointManager::createDbCheckpointAsync(const S
   if (!blockId.has_value()) {
     DbCheckpointManager::instance().setCheckpointInProcess(true, lastBlockid);
   }
-  auto ret = std::async(std::launch::async, [this, seqNum, timestamp, lastBlockid]() -> void {
-    createDbCheckpoint(lastBlockid, lastBlockid, seqNum, timestamp);
+  auto ret = std::async(std::launch::async, [this, seqNum, lastBlockid]() -> void {
+    createDbCheckpoint(lastBlockid, lastBlockid, seqNum);
   });
   // store the future for the async task
   dbCreateCheckPtFuture_.emplace(std::make_pair(lastBlockid, std::move(ret)));
@@ -335,23 +343,6 @@ void DbCheckpointManager::updateDbCheckpointMetadata() {
   std::vector<uint8_t> v(data.begin(), data.end());
   ps_->setDbCheckpointMetadata(v);
 }
-void DbCheckpointManager::sendInternalCreateDbCheckpointMsg(const SeqNum& seqNum, bool noop) {
-  auto replica_id = bftEngine::ReplicaConfig::instance().getreplicaId();
-  concord::messages::db_checkpoint_msg::CreateDbCheckpoint req;
-  req.sender = replica_id;
-  req.seqNum = seqNum;
-  req.noop = noop;
-  std::vector<uint8_t> data_vec;
-  concord::messages::db_checkpoint_msg::serialize(data_vec, req);
-  std::string sig(SigManager::instance()->getMySigLength(), '\0');
-  SigManager::instance()->sign(reinterpret_cast<char*>(data_vec.data()), data_vec.size(), sig.data());
-  req.signature = std::vector<uint8_t>(sig.begin(), sig.end());
-  data_vec.clear();
-  concord::messages::db_checkpoint_msg::serialize(data_vec, req);
-  std::string strMsg(data_vec.begin(), data_vec.end());
-  std::string cid = "replicaDbCheckpoint_" + std::to_string(seqNum) + "_" + std::to_string(replica_id);
-  if (client_) client_->sendRequest(bftEngine::DB_CHECKPOINT_FLAG, strMsg.size(), strMsg.c_str(), cid);
-}
 
 void DbCheckpointManager::setNextStableSeqNumToCreateSnapshot(const std::optional<SeqNum>& seqNum) {
   if (seqNum == std::nullopt) {
@@ -376,24 +367,6 @@ void DbCheckpointManager::updateMetrics() {
     LOG_DEBUG(GL, "-- RocksDb checkpoint metrics dump--" + metrics_.ToJson());
   }
 }
-void DbCheckpointManager::updateLastCmdInfo(const SeqNum& seqNum, const std::optional<Timestamp>& timestamp) {
-  dbCheckptMetadata_.lastCmdSeqNum_ = seqNum;
-  if (timestamp.has_value()) {
-    dbCheckptMetadata_.lastCmdTimestamp_ = std::chrono::duration_cast<Seconds>(timestamp.value().time_since_epoch);
-  } else {
-    dbCheckptMetadata_.lastCmdTimestamp_ = std::chrono::duration_cast<Seconds>(SystemClock::now().time_since_epoch());
-  }
-  lastCheckpointCreationTime_ = dbCheckptMetadata_.lastCmdTimestamp_;
-  if (maxNumOfCheckpoints_ == 0) {
-    // update lastCmdSeqNum and timestamp
-    std::scoped_lock lock(lock_);
-    updateDbCheckpointMetadata();
-  }
-}
-SeqNum DbCheckpointManager::getLastStableSeqNum() const {
-  if (getLastStableSeqNumCb_) return getLastStableSeqNumCb_();
-  return 0;
-}
 
 void DbCheckpointManager::setCheckpointInProcess(bool flag, concord::kvbc::BlockId blockId) const {
   if (checkpointInProcessCb_) checkpointInProcessCb_(flag, blockId);
@@ -479,8 +452,6 @@ void DbCheckpointManager::builMetadataFromFileSystem() {
 
       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();

bftengine/src/bftengine/Reconfiguration.cpp

@@ -165,6 +165,8 @@ void ReconfigurationHandler::handleWedgeCommands(bool bft_support,
 // so, we create a snapshot of the database, when we stop the replicas
 void ReconfigurationHandler::addCreateDbSnapshotCbOnWedge(bool bft_support) {
   auto wedgePt = bftEngine::ControlStateManager::instance().getCheckpointToStopAt();
+  // We don't save the shared data about the creation time and sequence number of the db checkpoint because this data is
+  // being saved only as part of the automated flow of creating a db checkpoint once in a period.
   if (wedgePt.has_value()) {
     if (bft_support) {
       bftEngine::IControlHandler::instance()->addOnStableCheckpointCallBack(