fix(session): keep RPC routing alive across reconnects

The server kept one session per peer address. When a reconnect registered
a new session before the old one was torn down, the new entry was dropped
and the old session's disconnect then erased the address entirely, so
B -> A requests silently stopped being delivered.

Keep every session, route to the newest, and skip routing cleanup while
another session for the same peer remains. Also surface synchronous send
failures and briefly retry RPCs that no endpoint could accept.

Log: keep RPC routing alive across reconnects
PMS: TASK-396845
Influence: B -> A requests keep being delivered when a peer reconnects.

Author: re2zero

src/lib/common/manager/sessionmanager.cpp

@@ -14,9 +14,12 @@
 #include <QDir>
 #include <QStandardPaths>
 #include <QCoreApplication>
+#include <QTimer>
 #include <QUuid>
 
 namespace {
+constexpr int kRpcRetryCount = 3;
+constexpr int kRpcRetryDelayMs = 300;
 // Transfer services are generated as random UUIDs. Restricting the accepted
 // format keeps a peer from naming a reserved service such as "barrier" and
 // stopping that forward when the transfer is released.
@@ -368,7 +371,28 @@ void SessionManager::sendRpcRequest(const QString &ip, int type, const QString &
     DLOG << "sendRpcRequest " << request;
 #endif
     WLOG << "sendRpcRequest " << request;
-    _session_worker->sendAsyncRequest(ip, request);
+    if (!_session_worker->sendAsyncRequest(ip, request))
+        scheduleRpcRetry(ip, request, kRpcRetryCount);
+}
+
+void SessionManager::scheduleRpcRetry(const QString &ip, const proto::OriginMessage &request,
+                                      int remaining)
+{
+    if (remaining <= 0) {
+        WLOG << "Giving up RPC after retries, target:" << ip.toStdString()
+             << "mask:" << request.mask;
+        return;
+    }
+
+    // No endpoint claimed the target, so nothing was sent. The session may be
+    // reconnecting or its IP mapping may not be ready yet; retry briefly
+    // instead of dropping the request silently.
+    WLOG << "RPC target not connected, retry in" << kRpcRetryDelayMs
+         << "ms, target:" << ip.toStdString() << "mask:" << request.mask;
+    QTimer::singleShot(kRpcRetryDelayMs, this, [this, ip, request, remaining] {
+        if (_session_worker && !_session_worker->sendAsyncRequest(ip, request))
+            scheduleRpcRetry(ip, request, remaining - 1);
+    });
 }
 
 void SessionManager::sendTunnelMessage(const QString &target, int mask, const QByteArray &payload)

src/lib/common/manager/sessionmanager.h

@@ -14,6 +14,8 @@
 
 #include <functional>
 
+namespace proto { struct OriginMessage; }
+
 class SessionWorker;
 class TransferWorker;
 class FileSizeCounter;
@@ -85,6 +87,7 @@ public slots:
     std::shared_ptr<TransferWorker> createTransWorker(const QString &jobid);
     void releaseTransWorker(const QString &jobid);
     void releaseTunnelJob(const QString &jobid);
+    void scheduleRpcRetry(const QString &target, const proto::OriginMessage &request, int remaining);
 
     struct TunnelJob {
         QString serviceId;

src/lib/common/manager/sessionworker.cpp

@@ -395,7 +395,8 @@ bool SessionWorker::sendAsyncRequest(const QString &target, const proto::OriginM
         return true;
     }
 
-    DLOG << "Failed to send async request to target:" << target.toStdString();
+    WLOG << "Failed to send async request to target:" << target.toStdString()
+         << "mask:" << request.mask;
     return false;
 }
 
@@ -586,12 +587,11 @@ bool SessionWorker::doAsyncRequest(T *endpoint, const std::string& target, const
 {
     if (endpoint && endpoint->hasConnected(target)) {
         DLOG << "doAsyncRequest to " << target;
-        endpoint->asyncRequestWithHandler(target, request, [this](int32_t type, const std::string &response) {
+        return endpoint->asyncRequestWithHandler(target, request, [this](int32_t type, const std::string &response) {
             DLOG << "doAsyncRequest callback: " << type << " " << response;
             QString res = QString::fromStdString(response);
             emit onRpcResult(type, res);
         });
-        return true;
     }
     return false;
 }

src/lib/common/session/protoclient.cpp

@@ -15,6 +15,8 @@ void ProtoClient::DisconnectAndStop()
 
 bool ProtoClient::hasConnected(const std::string &ip)
 {
+    // A stale host is tolerated here: onSend returns 0 when the session is
+    // not handshaked, and the caller then retries or falls back.
     return ip == connectedHost();
 }
 

src/lib/common/session/protoendpoint.cpp

@@ -38,9 +38,10 @@ proto::OriginMessage ProtoEndpoint::syncRequest(const std::string &target, const
     return proto::OriginMessage();
 }
 
-void ProtoEndpoint::asyncRequestWithHandler(const std::string &target, const proto::OriginMessage &request, RpcHandler resultHandler)
+bool ProtoEndpoint::asyncRequestWithHandler(const std::string &target, const proto::OriginMessage &request, RpcHandler resultHandler)
 {
-    asyncRequest(target, request);  // 发送异步请求
+    if (!asyncRequest(target, request))  // 发送异步请求
+        return false;
 
     int32_t type = request.mask;  // 获取请求的类型
     // 启动一个新的线程来处理结果
@@ -51,10 +52,11 @@ void ProtoEndpoint::asyncRequestWithHandler(const std::string &target, const pro
 
     // 在启动的新线程中处理结果，主线程可以继续执行其他任务
     handlerThread.detach();
+    return true;
 }
 
 
-void ProtoEndpoint::asyncRequest(const std::string &target, const proto::OriginMessage &msg)
+bool ProtoEndpoint::asyncRequest(const std::string &target, const proto::OriginMessage &msg)
 {
     int32_t type = msg.mask;
     std::future<proto::OriginMessage> futureResponse;
@@ -69,8 +71,24 @@ void ProtoEndpoint::asyncRequest(const std::string &target, const proto::OriginM
         clearActiveTarget();
     }
 
+    // request() fails synchronously when the session cannot send (for
+    // example it is not handshaked or the target lookup failed). Surface that
+    // so the caller can retry instead of silently dropping the RPC.
+    if (futureResponse.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready) {
+        try {
+            const auto early = futureResponse.get();
+            // Preserve a (theoretical) immediate response.
+            std::promise<proto::OriginMessage> ready;
+            ready.set_value(early);
+            futureResponse = ready.get_future();
+        } catch (const std::exception &) {
+            return false;
+        }
+    }
+
     std::lock_guard<std::mutex> locker(_lock);
     _responses_by_type.insert(std::make_pair(type, PendingResponse { std::move(futureResponse), msg.id }));
+    return true;
 }
 
 void ProtoEndpoint::sendMessage(const std::string &target, int32_t mask, const std::string &payload)

src/lib/common/session/protoendpoint.h

@@ -27,8 +27,9 @@ class ProtoEndpoint: public FBE::proto::FinalClient
     void sendDisRequest();
     proto::OriginMessage syncRequest(const std::string &target, const proto::OriginMessage &msg);
 
-    // async call request and with 3s timeout result callback
-    void asyncRequestWithHandler(const std::string &target, const proto::OriginMessage &request, RpcHandler resultHandler);
+    // async call request and with 3s timeout result callback. Returns false
+    // when the request could not be handed to the session at all.
+    bool asyncRequestWithHandler(const std::string &target, const proto::OriginMessage &request, RpcHandler resultHandler);
 
     // Send a one-way frame to a specific peer. It does not create a request
     // future and must not be routed through the multicast fallback.
@@ -50,7 +51,7 @@ class ProtoEndpoint: public FBE::proto::FinalClient
     }
 
 private:
-    void asyncRequest(const std::string &target, const proto::OriginMessage &msg);
+    bool asyncRequest(const std::string &target, const proto::OriginMessage &msg);
     std::string getResponse(int32_t type);
 
 protected:

src/lib/common/session/protoserver.cpp

@@ -146,37 +146,32 @@ class ProtoSession : public NetUtil::Asio::SSLSession, public FBE::proto::FinalC
 
 bool ProtoServer::hasConnected(const std::string &ip)
 {
-    // std::cout << "check hasConnected: " << ip << std::endl;
+    bool found = false;
 
     // Use consistent lock ordering to avoid deadlock: always acquire mapping_lock before session_lock
-    std::shared_lock<std::shared_mutex> mapping_locker(_ipmapping_lock);
-    std::shared_lock<std::shared_mutex> session_locker(_sessionids_lock);
-
-    // print all session ids
-    // for (auto &session : _session_ids) {
-    //     std::cout << "session ip: " << session.first << " id: " << session.second << std::endl;
-    // }
-
-    // First try to find by direct IP match
-    auto direct_it = _session_ids.find(ip);
-    if (direct_it != _session_ids.end()) {
-        std::cout << "hasConnected: " << ip << " true (direct match)" << std::endl;
-        return true;
-    }
+    {
+        std::shared_lock<std::shared_mutex> mapping_locker(_ipmapping_lock);
+        std::shared_lock<std::shared_mutex> session_locker(_sessionids_lock);
 
-    // If not found, try to find via IP mapping (for NAT/Router scenarios)
-    // Check if this IP is a real IP that maps to a remote endpoint IP
-    auto real_to_remote = _real_to_remote_ip.find(ip);
-    if (real_to_remote != _real_to_remote_ip.end()) {
-        auto session_it = _session_ids.find(real_to_remote->second);
-        if (session_it != _session_ids.end()) {
-            std::cout << "hasConnected: " << ip << " true (via IP mapping: " << real_to_remote->second << ")" << std::endl;
-            return true;
+        found = _session_ids.find(ip) != _session_ids.end();
+
+        // If not found, try via IP mapping (for NAT/Router scenarios)
+        if (!found) {
+            auto real_to_remote = _real_to_remote_ip.find(ip);
+            if (real_to_remote != _real_to_remote_ip.end())
+                found = _session_ids.find(real_to_remote->second) != _session_ids.end();
         }
     }
 
-    std::cout << "hasConnected: " << ip << " false" << std::endl;
-    return false;
+    if (!found) {
+        std::cout << "hasConnected: " << ip << " false" << std::endl;
+        return false;
+    }
+
+    // A registered session may still be completing its TLS handshake; onSend
+    // reports that as a send failure, which lets the caller retry.
+    std::cout << "hasConnected: " << ip << " true" << std::endl;
+    return true;
 }
 
 bool ProtoServer::startHeartbeat()
@@ -315,8 +310,9 @@ void ProtoServer::onConnected(std::shared_ptr<NetUtil::Asio::SSLSession>& sessio
 
     {
         std::unique_lock<std::shared_mutex> locker(_sessionids_lock);
-        // std::cout << "Server onConnected insert session ip: " << addr << " id: " << session->id() << std::endl;
-        _session_ids.insert(std::make_pair(addr, session->id()));
+        // Keep every session: a reconnect can register a new session before
+        // the old one is torn down. Routing always picks the newest one.
+        _session_ids.emplace(addr, session->id());
     }
 
     _callbacks->onStateChanged(RPC_CONNECTED, addr);
@@ -329,36 +325,47 @@ void ProtoServer::onDisconnected(std::shared_ptr<NetUtil::Asio::SSLSession>& ses
     auto search_uuid = session->id();
     std::string addr = "";
     std::string real_ip = "";
-    
+    bool replaced = false;
+
     // Use consistent lock ordering: mapping_lock before session_lock to avoid deadlock
     {
         std::unique_lock<std::shared_mutex> mapping_locker(_ipmapping_lock);
         std::unique_lock<std::shared_mutex> session_locker(_sessionids_lock);
-        
+
         // Find and remove session
         auto it = std::find_if(_session_ids.begin(), _session_ids.end(), [search_uuid](const std::pair<std::string, BaseKit::UUID>& pair) {
             return pair.second == search_uuid;
         });
 
         if (it != _session_ids.end()) {
-            //std::cout << "find connected by uuid, ip：" << it->first << std::endl;
             addr = it->first;
             _session_ids.erase(it);
-            
-            // Clean up IP mapping for this remote endpoint
-            auto remote_to_real = _remote_to_real_ip.find(addr);
-            if (remote_to_real != _remote_to_real_ip.end()) {
-                real_ip = remote_to_real->second;
-                _real_to_remote_ip.erase(real_ip);
-                _remote_to_real_ip.erase(addr);
-                std::cout << "Cleaned up IP mapping for disconnected session: " << real_ip << " -> " << addr << std::endl;
+
+            // A newer session for the same peer may already be connected.
+            // Keep its routing state and do not report the peer as gone.
+            replaced = _session_ids.find(addr) != _session_ids.end();
+
+            if (!replaced) {
+                // Clean up IP mapping for this remote endpoint
+                auto remote_to_real = _remote_to_real_ip.find(addr);
+                if (remote_to_real != _remote_to_real_ip.end()) {
+                    real_ip = remote_to_real->second;
+                    _real_to_remote_ip.erase(real_ip);
+                    _remote_to_real_ip.erase(addr);
+                    std::cout << "Cleaned up IP mapping for disconnected session: " << real_ip << " -> " << addr << std::endl;
+                }
             }
         } else {
             std::cout << "did not find connected id:" << search_uuid << std::endl;
             return;
         }
     }
 
+    if (replaced) {
+        std::cout << "Session superseded, keep connection state for: " << addr << std::endl;
+        return;
+    }
+
     _callbacks->onStateChanged(RPC_DISCONNECTED, addr);
 }
 
@@ -372,40 +379,51 @@ size_t ProtoServer::onSend(const void *data, size_t size)
         return size;
     }
 
-    std::shared_ptr<NetUtil::Asio::SSLSession> session = nullptr;
     std::string target_ip = _active_target;
+    _active_target.clear();
+
+    BaseKit::UUID sessionId;
+    bool found = false;
 
-    // First try direct session lookup
+    // Use consistent lock ordering: mapping_lock before session_lock. Prefer
+    // the newest session when a reconnecting peer has both an old and a new
+    // one registered.
     {
+        std::shared_lock<std::shared_mutex> mapping_locker(_ipmapping_lock);
         std::shared_lock<std::shared_mutex> session_locker(_sessionids_lock);
-        auto session_it = _session_ids.find(target_ip);
-        if (session_it != _session_ids.end()) {
-            session = FindSession(session_it->second);
-        }
-    }
 
-    // If not found, try via IP mapping (for NAT/Router scenarios)
-    if (!session) {
-        // Use consistent lock ordering: mapping_lock before session_lock
-        std::shared_lock<std::shared_mutex> mapping_locker(_ipmapping_lock);
-        auto real_to_remote = _real_to_remote_ip.find(target_ip);
-        if (real_to_remote != _real_to_remote_ip.end()) {
-            std::string remote_ip = real_to_remote->second;
-            std::shared_lock<std::shared_mutex> session_locker(_sessionids_lock);
-            auto session_it = _session_ids.find(remote_ip);
-            if (session_it != _session_ids.end()) {
-                session = FindSession(session_it->second);
-                std::cout << "Found session via IP mapping: " << target_ip << " -> " << remote_ip << std::endl;
-            }
+        auto newestFor = [this](const std::string &addr, BaseKit::UUID *id) {
+            auto range = _session_ids.equal_range(addr);
+            if (range.first == range.second)
+                return false;
+            auto newest = range.second;
+            --newest;
+            *id = newest->second;
+            return true;
+        };
+
+        found = newestFor(target_ip, &sessionId);
+
+        // If not found, try via IP mapping (for NAT/Router scenarios)
+        if (!found) {
+            auto real_to_remote = _real_to_remote_ip.find(target_ip);
+            if (real_to_remote != _real_to_remote_ip.end())
+                found = newestFor(real_to_remote->second, &sessionId);
         }
     }
 
-    if (session) {
-        session->SendAsync(data, size);
-    } else {
+    if (!found) {
         std::cout << "No session found for target: " << target_ip << std::endl;
+        return 0;
+    }
+
+    auto session = FindSession(sessionId);
+    if (!session || !session->IsHandshaked()) {
+        std::cout << "Session not ready for target: " << target_ip << std::endl;
+        return 0;
     }
 
-    _active_target = "";
-    return size;
+    // Report failure instead of pretending the frame was delivered; the FBE
+    // request then fails its future instead of hanging until timeout.
+    return session->SendAsync(data, size) ? size : 0;
 }

src/lib/common/session/protoserver.h

@@ -1,4 +1,4 @@
-// SPDX-FileCopyrightText: 2023 UnionTech Software Technology Co., Ltd.
+// SPDX-FileCopyrightText: 2023 - 2026 UnionTech Software Technology Co., Ltd.
 //
 // SPDX-License-Identifier: GPL-3.0-or-later
 
@@ -40,9 +40,10 @@ class ProtoServer : public NetUtil::Asio::SSLServer, public ProtoEndpoint
     void onHeartbeatTimeout(bool canceled);
 
 private:
-    // <ip, sessionid>
+    // <ip, sessionid>. A reconnecting peer can briefly have both the old and
+    // the new session, so keep all of them and route to the newest one.
     std::shared_mutex _sessionids_lock;
-    std::map<std::string, BaseKit::UUID> _session_ids;
+    std::multimap<std::string, BaseKit::UUID> _session_ids;
 
     // IP mapping for NAT/Router scenarios: <real_ip, remote_endpoint_ip>
     std::shared_mutex _ipmapping_lock;