Allow to get the latest sequence number when creating the backup

src/storage/storage.cc

@@ -345,7 +345,7 @@ Status Storage::Open(DBOpenMode mode) {
   return Status::OK();
 }
 
-Status Storage::CreateBackup() {
+Status Storage::CreateBackup(uint64_t *sequence_number) {
   LOG(INFO) << "[storage] Start to create new backup";
   std::lock_guard<std::mutex> lg(config_->backup_mu);
   std::string task_backup_dir = config_->GetBackupDir();
@@ -363,7 +363,7 @@ Status Storage::CreateBackup() {
   }
 
   std::unique_ptr<rocksdb::Checkpoint> checkpoint_guard(checkpoint);
-  s = checkpoint->CreateCheckpoint(tmpdir, config_->rocks_db.write_buffer_size * MiB);
+  s = checkpoint->CreateCheckpoint(tmpdir, config_->rocks_db.write_buffer_size * MiB, sequence_number);
   if (!s.ok()) {
     LOG(WARNING) << "Failed to create checkpoint (snapshot) for backup. Error: " << s.ToString();
     return {Status::DBBackupErr, s.ToString()};

src/storage/storage.h

@@ -123,7 +123,10 @@ class Storage {
   Status SetOptionForAllColumnFamilies(const std::string &key, const std::string &value);
   Status SetDBOption(const std::string &key, const std::string &value);
   Status CreateColumnFamilies(const rocksdb::Options &options);
-  Status CreateBackup();
+  // The sequence_number will be pointed to the value of the sequence number in range of DB,
+  // but can't promise it's the latest sequence number. So you must check it by yourself before
+  // using it.
+  Status CreateBackup(uint64_t *sequence_number = nullptr);
   void DestroyBackup();
   Status RestoreFromBackup();
   Status RestoreFromCheckpoint();

tests/cppunit/storage_test.cc

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <config/config.h>
+#include <gtest/gtest.h>
+#include <status.h>
+#include <storage/storage.h>
+
+#include <filesystem>
+
+TEST(Storage, CreateBackup) {
+  std::error_code ec;
+
+  Config config;
+  config.db_dir = "test_backup_dir";
+  config.slot_id_encoded = false;
+
+  std::filesystem::remove_all(config.db_dir, ec);
+  ASSERT_TRUE(!ec);
+
+  auto storage = std::make_unique<engine::Storage>(&config);
+  auto s = storage->Open();
+  ASSERT_TRUE(s.IsOK());
+
+  constexpr int cnt = 10;
+  for (int i = 0; i < cnt; i++) {
+    rocksdb::WriteBatch batch;
+    batch.Put("k", "v");
+    ASSERT_TRUE(storage->Write(rocksdb::WriteOptions(), &batch).ok());
+  }
+  uint64_t sequence_number = 0;
+  s = storage->CreateBackup(&sequence_number);
+  ASSERT_TRUE(s.IsOK());
+  ASSERT_EQ(cnt, sequence_number);
+  // check if backup success without caring about the sequence number
+  s = storage->CreateBackup();
+  ASSERT_TRUE(s.IsOK());
+
+  std::filesystem::remove_all(config.db_dir, ec);
+  ASSERT_TRUE(!ec);
+}