paticcaa · ivanallen · Sep 4, 2025 · Sep 4, 2025
diff --git a/src/common/test/test_rocksdb_txn_store.cc b/src/common/test/test_rocksdb_txn_store.cc
@@ -16,6 +16,7 @@
 #include <fmt/std.h>
 
 #include "common/rocksdb_txn_store.h"
+#include "common/txn_manager.h"
 
 // NOLINTBEGIN(readability-magic-numbers)
 
@@ -110,6 +111,20 @@ TEST_F(TestRocksdbTxnStore, Hset) {
     ASSERT_EQ(get_value(expected_key), "test_value");
 }
 
+TEST_F(TestRocksdbTxnStore, TxnGuard) {
+    auto txn = create_transaction();
+    std::string expected_key = "test_key\1test_field";
+    PAIN_TXN(txn.get()) {
+        auto status = txn->hset("test_key", "test_field", "test_value");
+        EXPECT_TRUE(status.ok()) << status.error_str();
+        EXPECT_FALSE(key_exists(expected_key));
+        return status;
+    };
+
+    ASSERT_TRUE(key_exists(expected_key));
+    ASSERT_EQ(get_value(expected_key), "test_value");
+}
+
 // 测试 hset 多个字段
 TEST_F(TestRocksdbTxnStore, HsetMultipleFields) {
     auto txn = create_transaction();

diff --git a/src/common/txn_manager.cc b/src/common/txn_manager.cc
@@ -0,0 +1,7 @@
+#include "common/txn_manager.h"
+
+namespace pain::common {
+
+thread_local TxnStore* TxnManager::g_txn_store = nullptr;
+
+} // namespace pain::common
diff --git a/src/common/txn_manager.h b/src/common/txn_manager.h
@@ -0,0 +1,69 @@
+#pragma once
+
+#include <pain/base/types.h>
+#include "common/txn_store.h"
+
+namespace pain::common {
+
+class TxnManager {
+public:
+    static const TxnManager& instance() {
+        static TxnManager s_instance;
+        return s_instance;
+    }
+
+    bool in_txn() const {
+        return g_txn_store != nullptr;
+    }
+
+    TxnStore* get_txn_store() const {
+        return g_txn_store;
+    }
+
+    void begin(TxnStore* txn_store) const {
+        g_txn_store = txn_store;
+    }
+
+    Status commit() const {
+        auto status = g_txn_store->commit();
+        g_txn_store = nullptr;
+        return status;
+    }
+
+    Status rollback() const {
+        auto status = g_txn_store->rollback();
+        g_txn_store = nullptr;
+        return status;
+    }
+
+private:
+    static thread_local TxnStore* g_txn_store;
+};
+
+class TxnGuard {
+public:
+    TxnGuard(TxnStore* txn_store) {
+        TxnManager::instance().begin(txn_store);
+    }
+
+    ~TxnGuard() {
+        if (_status.ok()) {
+            TxnManager::instance().commit();
+        } else {
+            TxnManager::instance().rollback();
+        }
+    }
+
+    void operator+(std::function<Status(TxnStore*)> cb) {
+        auto txn_store = TxnManager::instance().get_txn_store();
+        _status = cb(txn_store);
+    }
+
+private:
+    Status _status;
+};
+
+} // namespace pain::common
+
+#define PAIN_TXN(txn_store)                                                                                            \
+    pain::common::TxnGuard(txn_store) + [&]([[maybe_unused]] pain::common::TxnStore * txn) mutable -> pain::Status
diff --git a/src/deva/deva.cc b/src/deva/deva.cc
@@ -1,6 +1,7 @@
 #include "deva/deva.h"
 #include <pain/base/plog.h>
 #include <pain/base/uuid.h>
+#include "common/txn_manager.h"
 #include "deva/macro.h"
 
 #define DEVA_METHOD(name)                                                                                              \
@@ -149,12 +150,44 @@ DEVA_METHOD(SealAndNewChunk) {
     return Status::OK();
 }
 
+Status Deva::set_applied_index(int64_t index) {
+    if (index <= _applied_index) {
+        PLOG_WARN(("desc", "index is applied already")("index", index));
+        return Status::OK();
+    }
+    auto in_txn = common::TxnManager::instance().in_txn();
+    auto this_txn = _store->begin_txn();
+    auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get();
+    if (txn == nullptr) {
+        return Status(EIO, "Failed to begin transaction");
+    }
+    auto status = txn->hset(_meta_key, _applied_index_key, std::to_string(index));
+    return status;
+    if (in_txn) {
+        return Status::OK();
+    }
+
+    status = txn->commit();
+    return status;
+}
+
 Status Deva::save_snapshot(std::string_view path, std::vector<std::string>* files) {
     return _store->check_point(path.data(), files);
 }
 
 Status Deva::load_snapshot(std::string_view path) {
-    return _store->recover(path.data());
+    auto status = _store->recover(path.data());
+    if (!status.ok()) {
+        return status;
+    }
+    // get applied index
+    std::string applied_index_str;
+    status = _store->hget(_meta_key, _applied_index_key, &applied_index_str);
+    if (!status.ok()) {
+        return status;
+    }
+    _applied_index = std::stoll(applied_index_str);
+    return Status::OK();
 }
 
 } // namespace pain::deva
diff --git a/src/deva/deva.h b/src/deva/deva.h
@@ -1,11 +1,16 @@
 #pragma once
 
+#include <pain/base/plog.h>
 #include <pain/base/types.h>
 #include <boost/intrusive_ptr.hpp>
 #include "pain/proto/deva_store.pb.h"
 #include "common/store.h"
+#include "common/txn_manager.h"
+#include "common/txn_store.h"
 #include "deva/container.h"
 #include "deva/namespace.h"
+#include "deva/op.h"
+
 #define DEVA_ENTRY(name)                                                                                               \
     Status name([[maybe_unused]] int32_t version,                                                                      \
                 [[maybe_unused]] const pain::proto::deva::store::name##Request* request,                               \
@@ -15,7 +20,25 @@
                    [[maybe_unused]] const pain::proto::deva::store::name##Request* request,                            \
                    [[maybe_unused]] pain::proto::deva::store::name##Response* response,                                \
                    [[maybe_unused]] int64_t index) {                                                                   \
-        return name(version, request, response, index);                                                                \
+        if (!need_apply(OpType::k##name)) {                                                                            \
+            return name(version, request, response, index);                                                            \
+        }                                                                                                              \
+        if (check_index_is_applied(index)) {                                                                           \
+            PLOG_INFO(("desc", "index is applied already")("index", index));                                           \
+            return Status::OK();                                                                                       \
+        }                                                                                                              \
+        auto txn = _store->begin_txn();                                                                                \
+        auto status = Status::OK();                                                                                    \
+        PAIN_TXN(txn.get()) {                                                                                          \
+            status = set_applied_index(index);                                                                         \
+            if (!status.ok()) {                                                                                        \
+                PLOG_ERROR(("desc", "set applied index failed")("index", index)("error", status.error_str()));         \
+                return status;                                                                                         \
+            }                                                                                                          \
+            status = name(version, request, response, index);                                                          \
+            return status;                                                                                             \
+        };                                                                                                             \
+        return status;                                                                                                 \
     }
 
 namespace pain::deva {
@@ -41,12 +64,19 @@ class Deva : public Container {
 
 private:
     Status create(const std::string& path, const UUID& id, FileType type);
+    Status set_applied_index(int64_t applied_index);
+    bool check_index_is_applied(int64_t index) const {
+        return index != 0 && index <= _applied_index;
+    }
 
 private:
     std::atomic<int> _use_count;
     common::StorePtr _store;
     Namespace _namespace;
     std::unordered_map<UUID, proto::FileInfo> _file_infos;
+    const char* _meta_key = "meta";
+    const char* _applied_index_key = "applied_index";
+    int64_t _applied_index = 0;
 
     friend void intrusive_ptr_add_ref(Deva* deva) {
         ++deva->_use_count;

diff --git a/src/deva/namespace.cc b/src/deva/namespace.cc
@@ -1,6 +1,7 @@
 #include "deva/namespace.h"
 #include <pain/base/plog.h>
 #include <pain/base/scope_exit.h>
+#include "common/txn_manager.h"
 #include "common/txn_store.h"
 
 namespace pain::deva {
@@ -14,8 +15,10 @@ Status Namespace::load() {
 }
 
 Status Namespace::create(const UUID& parent, const std::string& name, FileType type, const UUID& inode) {
-    auto txn = _store->begin_txn();
-    if (!txn) {
+    auto in_txn = common::TxnManager::instance().in_txn();
+    auto this_txn = _store->begin_txn();
+    auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get();
+    if (txn == nullptr) {
         return Status(EIO, "Failed to begin transaction");
     }
     auto rollback = make_scope_exit([&txn]() {
@@ -24,6 +27,9 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t
             PLOG_ERROR(("desc", "Failed to rollback")("status", status));
         }
     });
+    if (in_txn) {
+        rollback.release();
+    }
     // find parent
     std::string dentries_str;
     proto::DirEntries dentries;
@@ -75,7 +81,14 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t
     if (!file_info.SerializeToString(&file_info_str)) {
         return Status(EBADMSG, "Failed to serialize file info");
     }
-    txn->hset(_inode_key, inode.str(), file_info_str);
+    status = txn->hset(_inode_key, inode.str(), file_info_str);
+    if (!status.ok()) {
+        return status;
+    }
+
+    if (in_txn) {
+        return Status::OK();
+    }
 
     status = txn->commit();
     if (!status.ok()) {
@@ -86,17 +99,22 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t
 }
 
 Status Namespace::remove(const UUID& parent, const std::string& name) {
-    // find parent
-    auto txn = _store->begin_txn();
-    if (!txn) {
+    auto in_txn = common::TxnManager::instance().in_txn();
+    auto this_txn = _store->begin_txn();
+    auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get();
+    if (txn == nullptr) {
         return Status(EIO, "Failed to begin transaction");
     }
+    // find parent
     auto rollback = make_scope_exit([&txn]() {
         auto status = txn->rollback();
         if (!status.ok()) {
             PLOG_ERROR(("desc", "Failed to rollback")("status", status));
         }
     });
+    if (in_txn) {
+        rollback.release();
+    }
     // get dentries
     std::string dentries_str;
     auto status = txn->hget(_dentry_key, parent.str(), &dentries_str);
@@ -122,10 +140,20 @@ Status Namespace::remove(const UUID& parent, const std::string& name) {
     if (!dentries.SerializeToString(&dentries_str)) {
         return Status(EBADMSG, "Failed to serialize dentries");
     }
-    txn->hset(_dentry_key, parent.str(), dentries_str);
+    status = txn->hset(_dentry_key, parent.str(), dentries_str);
+    if (!status.ok()) {
+        return status;
+    }
 
     // remove file info
-    txn->hdel(_inode_key, file_id.str());
+    status = txn->hdel(_inode_key, file_id.str());
+    if (!status.ok()) {
+        return status;
+    }
+
+    if (in_txn) {
+        return Status::OK();
+    }
 
     status = txn->commit();
     if (!status.ok()) {
@@ -138,18 +166,34 @@ Status Namespace::remove(const UUID& parent, const std::string& name) {
 void Namespace::list(const UUID& parent, std::list<DirEntry>* entries) const {
     entries->clear();
     std::string dentries_str;
-    auto status = _store->hget(_dentry_key, parent.str(), &dentries_str);
+    auto in_txn = common::TxnManager::instance().in_txn();
+    auto this_txn = _store->begin_txn();
+    auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get();
+    if (txn == nullptr) {
+        return;
+    }
+    auto status = txn->hget(_dentry_key, parent.str(), &dentries_str);
     if (!status.ok()) {
+        PLOG_ERROR(("desc", "Failed to get dentries")("status", status));
         return;
     }
     proto::DirEntries dentries;
     if (!dentries.ParseFromString(dentries_str)) {
+        PLOG_ERROR(("desc", "Failed to parse dentries")("status", status));
         return;
     }
     for (const auto& dentry : dentries.entries()) {
         entries->emplace_back(
             UUID(dentry.file_id().high(), dentry.file_id().low()), dentry.name(), static_cast<FileType>(dentry.type()));
     }
+    if (in_txn) {
+        return;
+    }
+
+    status = txn->commit();
+    if (!status.ok()) {
+        PLOG_ERROR(("desc", "Failed to commit")("status", status));
+    }
 }
 
 // parse path such as /a/b/c to ["a", "b", "c"]
@@ -184,9 +228,16 @@ Status Namespace::lookup(const char* path, UUID* inode, FileType* file_type) con
     UUID parent = _root;
     *file_type = FileType::kDirectory;
 
+    auto in_txn = common::TxnManager::instance().in_txn();
+    auto this_txn = _store->begin_txn();
+    auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get();
+    if (txn == nullptr) {
+        return Status(EIO, "Failed to begin transaction");
+    }
+
     for (const auto& component : components) {
         std::string dentries_str;
-        auto status = _store->hget(_dentry_key, parent.str(), &dentries_str);
+        auto status = txn->hget(_dentry_key, parent.str(), &dentries_str);
         if (!status.ok()) {
             return status;
         }
@@ -208,7 +259,15 @@ Status Namespace::lookup(const char* path, UUID* inode, FileType* file_type) con
         *file_type = static_cast<FileType>(entry->type());
     }
     *inode = parent;
-    return Status::OK();
+    if (in_txn) {
+        return Status::OK();
+    }
+
+    status = txn->commit();
+    if (!status.ok()) {
+        PLOG_ERROR(("desc", "Failed to commit")("status", status));
+    }
+    return status;
 }
 
 } // namespace pain::deva