paticcaa · ivanallen · Sep 5, 2025 · Sep 5, 2025
diff --git a/protocols/pain/proto/asura.proto b/protocols/pain/proto/asura.proto
@@ -8,9 +8,6 @@ option cc_generic_services = true;
 service AsuraService {
     rpc RegisterDeva(RegisterDevaRequest) returns (RegisterDevaResponse);
     rpc ListDeva(ListDevaRequest) returns (ListDevaResponse);
-    rpc RegisterManusya(RegisterManusyaRequest)
-        returns (RegisterManusyaResponse);
-    rpc ListManusya(ListManusyaRequest) returns (ListManusyaResponse);
 }
 
 message DevaServer {
@@ -40,32 +37,3 @@ message ListDevaResponse {
     Header header = 1;
     repeated DevaServer deva_servers = 2;
 }
-
-message ManusyaServer {
-    UUID id = 1;
-    string ip = 2;
-    uint32 port = 3;
-    UUID pool_id = 4;
-}
-
-message RegisterManusyaRequest {
-    repeated ManusyaServer manusya_servers = 1;
-}
-
-message RegisterManusyaResponse {
-    message Result {
-        UUID id = 1;
-        int32 code = 2;
-        string message = 3;
-    }
-
-    Header header = 1;
-    repeated Result results = 2;
-}
-
-message ListManusyaRequest {}
-
-message ListManusyaResponse {
-    Header header = 1;
-    repeated ManusyaServer manusya_servers = 2;
-}
diff --git a/protocols/pain/proto/common.proto b/protocols/pain/proto/common.proto
@@ -85,3 +85,26 @@ message DirEntry {
 message DirEntries {
     repeated DirEntry entries = 1;
 }
+
+message ManusyaID {
+    string ip = 1;
+    int32 port = 2;
+    UUID uuid = 3;
+}
+
+message StorageInfo {
+    string cluster_id = 1;
+    int64 ctime = 2;
+}
+
+message ManusyaRegistration {
+    ManusyaID manusya_id = 1;
+    StorageInfo storage_info = 2;
+}
+
+message ManusyaDescriptor {
+    ManusyaID manusya_id = 1;
+    StorageInfo storage_info = 2;
+    bool is_alive = 3;
+    uint64 last_heartbeat_time = 4;
+}
diff --git a/protocols/pain/proto/deva.proto b/protocols/pain/proto/deva.proto
@@ -18,6 +18,10 @@ service DevaService {
     rpc SealChunk(SealChunkRequest) returns (SealChunkResponse);
     rpc SealAndNewChunk(SealAndNewChunkRequest)
         returns (SealAndNewChunkResponse);
+
+    rpc ManusyaHeartbeat(ManusyaHeartbeatRequest)
+        returns (ManusyaHeartbeatResponse);
+    rpc ListManusya(ListManusyaRequest) returns (ListManusyaResponse);
 }
 
 enum OpenFlag {
@@ -114,3 +118,18 @@ message SealChunkRequest {
 message SealChunkResponse {
     Header header = 1;
 }
+
+message ManusyaHeartbeatRequest {
+    ManusyaRegistration manusya_registration = 1;
+}
+
+message ManusyaHeartbeatResponse {
+    Header header = 1;
+}
+
+message ListManusyaRequest {}
+
+message ListManusyaResponse {
+    Header header = 1;
+    repeated ManusyaDescriptor manusya_descriptors = 2;
+}
diff --git a/protocols/pain/proto/deva_store.proto b/protocols/pain/proto/deva_store.proto
@@ -18,6 +18,14 @@ message CreateFileResponse {
     FileInfo file_info = 1;
 }
 
+message GetFileInfoRequest {
+    string path = 1;
+}
+
+message GetFileInfoResponse {
+    FileInfo file_info = 1;
+}
+
 message CreateDirRequest {
     string path = 1;
     UUID dir_id = 2;
@@ -64,3 +72,15 @@ message SealChunkResponse {}
 message SealAndNewChunkRequest {}
 
 message SealAndNewChunkResponse {}
+
+message ManusyaHeartbeatRequest {
+    ManusyaRegistration manusya_registration = 1;
+}
+
+message ManusyaHeartbeatResponse {}
+
+message ListManusyaRequest {}
+
+message ListManusyaResponse {
+    repeated ManusyaDescriptor manusya_descriptors = 1;
+}
diff --git a/src/asura/asura_service_impl.cc b/src/asura/asura_service_impl.cc
@@ -39,34 +39,6 @@ void AsuraServiceImpl::RegisterDeva(::google::protobuf::RpcController* controlle
     }
 }
 
-void AsuraServiceImpl::RegisterManusya(::google::protobuf::RpcController* controller,
-                                       [[maybe_unused]] const pain::proto::asura::RegisterManusyaRequest* request,
-                                       [[maybe_unused]] pain::proto::asura::RegisterManusyaResponse* response,
-                                       ::google::protobuf::Closure* done) { // NOLINT(readability-non-const-parameter)
-    ASURA_SPAN(span, controller);
-    brpc::ClosureGuard done_guard(done);
-    for (const auto& manusya_server : request->manusya_servers()) {
-        auto uuid = UUID(manusya_server.id().high(), manusya_server.id().low());
-        auto result = response->add_results();
-        result->mutable_id()->CopyFrom(manusya_server.id());
-        result->set_code(0);
-        result->set_message("ok");
-
-        if (!_store->hexists(ASURA_MANUSYA, uuid.str())) {
-            result->set_code(EEXIST);
-            result->set_message(
-                fmt::format("{} existed. ip:{}, port:{}", uuid.str(), manusya_server.ip(), manusya_server.port()));
-            continue;
-        }
-
-        auto status = _store->hset(ASURA_MANUSYA, uuid.str(), manusya_server.SerializeAsString());
-        if (!status.ok()) {
-            result->set_code(status.error_code());
-            result->set_message(status.error_cstr());
-        }
-    }
-}
-
 void AsuraServiceImpl::ListDeva(::google::protobuf::RpcController* controller,
                                 [[maybe_unused]] const pain::proto::asura::ListDevaRequest* request,
                                 [[maybe_unused]] pain::proto::asura::ListDevaResponse* response,
@@ -82,19 +54,4 @@ void AsuraServiceImpl::ListDeva(::google::protobuf::RpcController* controller,
     }
 }
 
-void AsuraServiceImpl::ListManusya(::google::protobuf::RpcController* controller,
-                                   [[maybe_unused]] const pain::proto::asura::ListManusyaRequest* request,
-                                   [[maybe_unused]] pain::proto::asura::ListManusyaResponse* response,
-                                   ::google::protobuf::Closure* done) { // NOLINT(readability-non-const-parameter)
-    ASURA_SPAN(span, controller);
-    brpc::ClosureGuard done_guard(done);
-    auto it = _store->hgetall(ASURA_MANUSYA);
-    while (it->valid()) {
-        auto manusya = response->add_manusya_servers();
-        auto value = it->value();
-        manusya->ParseFromArray(value.data(), value.size());
-        it->next();
-    }
-}
-
 } // namespace pain::asura
diff --git a/src/asura/asura_service_impl.h b/src/asura/asura_service_impl.h
@@ -10,9 +10,7 @@ class AsuraServiceImpl : public pain::proto::asura::AsuraService {
 public:
     AsuraServiceImpl(common::StorePtr store) : _store(store) {}
     ASURA_RPC_ENTRY(RegisterDeva);
-    ASURA_RPC_ENTRY(RegisterManusya);
     ASURA_RPC_ENTRY(ListDeva);
-    ASURA_RPC_ENTRY(ListManusya);
 
 private:
     common::StorePtr _store;

diff --git a/src/deva/deva.cc b/src/deva/deva.cc
@@ -60,7 +60,10 @@ DEVA_METHOD(CreateFile) {
     file_info->set_mode(request->mode());
     file_info->set_uid(request->uid());
     file_info->set_gid(request->gid());
-    _file_infos[file_uuid] = *file_info;
+    status = update_file_info(file_uuid, *file_info);
+    if (!status.ok()) {
+        return status;
+    }
     return Status::OK();
 }
 
@@ -85,7 +88,10 @@ DEVA_METHOD(CreateDir) {
     file_info->set_mode(request->mode());
     file_info->set_uid(request->uid());
     file_info->set_gid(request->gid());
-    _file_infos[dir_uuid] = *file_info;
+    status = update_file_info(dir_uuid, *file_info);
+    if (!status.ok()) {
+        return status;
+    }
     return Status::OK();
 }
 
@@ -150,6 +156,76 @@ DEVA_METHOD(SealAndNewChunk) {
     return Status::OK();
 }
 
+DEVA_METHOD(GetFileInfo) {
+    SPAN(span);
+    PLOG_INFO(("desc", "get_file_info")("version", version)("index", index));
+    auto& path = request->path();
+    UUID file_uuid;
+    FileType file_type = FileType::kNone;
+    auto status = _namespace.lookup(path.c_str(), &file_uuid, &file_type);
+    if (!status.ok()) {
+        return status;
+    }
+    if (file_type != FileType::kFile) {
+        return Status(EINVAL, fmt::format("{} is not a file", path.c_str()));
+    }
+    proto::FileInfo file_info;
+    status = get_file_info(file_uuid, &file_info);
+    if (!status.ok()) {
+        return status;
+    }
+    response->mutable_file_info()->Swap(&file_info);
+    return Status::OK();
+}
+
+DEVA_METHOD(ManusyaHeartbeat) {
+    SPAN(span);
+    PLOG_INFO(("desc", "manusya_heartbeat")("version", version)("index", index));
+    auto& manusya_id = request->manusya_registration().manusya_id();
+    // auto& storage_info = request->manusya_registration().storage_info();
+    // TODO: check cluster id
+
+    UUID manusya_uuid(manusya_id.uuid().high(), manusya_id.uuid().low());
+    auto it = _manusya_descriptors.find(manusya_uuid);
+    if (it == _manusya_descriptors.end()) {
+        // new manusya
+        PLOG_INFO(("desc", "new manusya")              //
+                  ("manusya_uuid", manusya_uuid.str()) //
+                  ("ip", manusya_id.ip())              //
+                  ("port", manusya_id.port()));
+        ManusyaDescriptor manusya_descriptor;
+        manusya_descriptor.ip = manusya_id.ip();
+        manusya_descriptor.port = manusya_id.port();
+        manusya_descriptor.uuid = manusya_uuid;
+        manusya_descriptor.is_alive = true;
+        manusya_descriptor.update_heartbeat();
+        _manusya_descriptors[manusya_uuid] = manusya_descriptor;
+    } else {
+        auto& manusya_descriptor = it->second;
+        manusya_descriptor.ip = manusya_id.ip();
+        manusya_descriptor.port = manusya_id.port();
+        manusya_descriptor.update_heartbeat();
+    }
+    return Status::OK();
+}
+
+DEVA_METHOD(ListManusya) {
+    SPAN(span);
+    PLOG_INFO(("desc", "list_manusya")("version", version)("index", index));
+    auto manusya_descriptors = response->mutable_manusya_descriptors();
+    for (auto& [_, manusya_descriptor] : _manusya_descriptors) {
+        auto manusya_descriptor_proto = manusya_descriptors->Add();
+        manusya_descriptor_proto->mutable_manusya_id()->set_ip(manusya_descriptor.ip);
+        manusya_descriptor_proto->mutable_manusya_id()->set_port(manusya_descriptor.port);
+        manusya_descriptor_proto->mutable_manusya_id()->mutable_uuid()->set_high(manusya_descriptor.uuid.high());
+        manusya_descriptor_proto->mutable_manusya_id()->mutable_uuid()->set_low(manusya_descriptor.uuid.low());
+        manusya_descriptor_proto->mutable_storage_info()->set_cluster_id(manusya_descriptor.cluster_id);
+        manusya_descriptor_proto->set_is_alive(manusya_descriptor.is_alive);
+        manusya_descriptor_proto->set_last_heartbeat_time(manusya_descriptor.last_heartbeat_time);
+    }
+    return Status::OK();
+}
+
 Status Deva::set_applied_index(int64_t index) {
     if (index <= _applied_index) {
         PLOG_WARN(("desc", "index is applied already")("index", index));
@@ -171,6 +247,48 @@ Status Deva::set_applied_index(int64_t index) {
     return status;
 }
 
+Status Deva::update_file_info(const UUID& id, const proto::FileInfo& file_info) {
+    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(_file_info_key, id.str(), file_info.SerializeAsString());
+    if (!status.ok()) {
+        return status;
+    }
+    if (in_txn) {
+        return Status::OK();
+    }
+    status = txn->commit();
+    if (!status.ok()) {
+        return status;
+    }
+    return status;
+}
+
+Status Deva::get_file_info(const UUID& id, proto::FileInfo* file_info) {
+    std::string file_info_str;
+    auto status = _store->hget(_file_info_key, id.str(), &file_info_str);
+    if (!status.ok()) {
+        return status;
+    }
+    auto ret = file_info->ParseFromString(file_info_str);
+    if (!ret) {
+        return Status(EIO, "Failed to parse file info");
+    }
+    return Status::OK();
+}
+
+Status Deva::remove_file_info(const UUID& id) {
+    auto status = _store->hdel(_file_info_key, id.str());
+    if (!status.ok()) {
+        return status;
+    }
+    return Status::OK();
+}
+
 Status Deva::save_snapshot(std::string_view path, std::vector<std::string>* files) {
     return _store->check_point(path.data(), files);
 }

diff --git a/src/deva/deva.h b/src/deva/deva.h
@@ -8,6 +8,7 @@
 #include "common/txn_manager.h"
 #include "common/txn_store.h"
 #include "deva/container.h"
+#include "deva/manusya_descriptor.h"
 #include "deva/namespace.h"
 #include "deva/op.h"
 
@@ -58,6 +59,9 @@ class Deva : public Container {
     DEVA_ENTRY(CheckInChunk);
     DEVA_ENTRY(SealChunk);
     DEVA_ENTRY(SealAndNewChunk);
+    DEVA_ENTRY(GetFileInfo);
+    DEVA_ENTRY(ManusyaHeartbeat);
+    DEVA_ENTRY(ListManusya);
 
     Status save_snapshot(std::string_view path, std::vector<std::string>* files) override;
     Status load_snapshot(std::string_view path) override;
@@ -69,15 +73,22 @@ class Deva : public Container {
         return index != 0 && index <= _applied_index;
     }
 
+    Status update_file_info(const UUID& id, const proto::FileInfo& file_info);
+    Status get_file_info(const UUID& id, proto::FileInfo* file_info);
+    Status remove_file_info(const UUID& id);
+
 private:
     std::atomic<int> _use_count;
     common::StorePtr _store;
     Namespace _namespace;
-    std::unordered_map<UUID, proto::FileInfo> _file_infos;
+    const char* _file_info_key = "file_info";
     const char* _meta_key = "meta";
     const char* _applied_index_key = "applied_index";
     int64_t _applied_index = 0;
 
+    // don't need persist
+    std::unordered_map<UUID, ManusyaDescriptor> _manusya_descriptors;
+
     friend void intrusive_ptr_add_ref(Deva* deva) {
         ++deva->_use_count;
     }