proxy client: always resubscribe on connectivity change

Track subscribe/permanent-put expiration with std::chrono::system_clock
(wall clock) instead of steady_clock in both the proxy client and server,
so persisted expirations survive restarts and map directly to epoch time.

- DhtProxyServer::clock is now system_clock; expireTimer/expireNotifyTimer
  use asio::system_timer.
- DhtProxyClient::refreshSubscriberTimer uses asio::system_timer and
  system_clock::now() for subscriber refresh.

include/opendht/dht_proxy_server.h

@@ -76,7 +76,7 @@ public:
 
     asio::io_context& io_context() const;
 
-    using clock = std::chrono::steady_clock;
+    using clock = std::chrono::system_clock;
     using time_point = clock::time_point;
 
     struct PushStats
@@ -405,8 +405,8 @@ private:
         std::string pushToken;
         std::string clientId;
         std::shared_ptr<PushSessionContext> sessionCtx;
-        std::unique_ptr<asio::steady_timer> expireTimer;
-        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
         Sp<Value> value;
         PushType type;
         std::string topic;
@@ -419,7 +419,7 @@ private:
             p.pack("value");
             p.pack(value);
             p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(std::chrono::system_clock::to_time_t(expiration));
             if (not clientId.empty()) {
                 p.pack("cid");
                 p.pack(clientId);
@@ -464,8 +464,8 @@ private:
         std::string clientId;
         std::shared_ptr<PushSessionContext> sessionCtx;
         std::future<size_t> internalToken;
-        std::unique_ptr<asio::steady_timer> expireTimer;
-        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
         PushType type;
         std::string topic;
 
@@ -476,7 +476,7 @@ private:
             p.pack("cid");
             p.pack(clientId);
             p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(std::chrono::system_clock::to_time_t(expiration));
             if (sessionCtx) {
                 std::lock_guard l(sessionCtx->lock);
                 p.pack("sid");

include/opendht/proxy.h

@@ -8,7 +8,7 @@ namespace dht {
 namespace proxy {
 
 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
-constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
+constexpr const std::chrono::hours OP_MARGIN {8};   // eight hours
 
 using ListenToken = uint64_t;
 

src/dht_proxy_client.cpp

@@ -34,20 +34,20 @@ struct DhtProxyClient::Listener
     Sp<OperationState> opstate;
     std::shared_ptr<http::Request> request;
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
-    std::unique_ptr<asio::steady_timer> refreshSubscriberTimer;
+    std::unique_ptr<asio::system_timer> refreshSubscriberTimer;
 #endif
 };
 
 struct PermanentPut
 {
-    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::steady_timer>&& j, const Sp<std::atomic_bool>& o)
+    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::system_timer>&& j, const Sp<std::atomic_bool>& o)
         : value(v)
         , refreshPutTimer(std::move(j))
         , ok(o)
     {}
 
     Sp<Value> value;
-    std::unique_ptr<asio::steady_timer> refreshPutTimer;
+    std::unique_ptr<asio::system_timer> refreshPutTimer;
     Sp<std::atomic_bool> ok;
 };
 
@@ -424,7 +424,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         auto& search = searches_[key];
         if (val->id) {
             auto id = val->id;
-            auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+            auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                         proxy::OP_TIMEOUT - proxy::OP_MARGIN);
             refreshPutTimer->async_wait(
                 std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
@@ -540,7 +540,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
                             auto it = search.pendingPuts.find(val);
                             if (it != search.pendingPuts.end()) {
                                 auto sok = std::make_shared<std::atomic_bool>(ok);
-                                auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+                                auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                                             proxy::OP_TIMEOUT
                                                                                                 - proxy::OP_MARGIN);
                                 refreshPutTimer->async_wait(
@@ -891,8 +891,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
                  * (if the proxy crash for any reason)
                  */
                 if (!l->second.refreshSubscriberTimer)
-                    l->second.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
-                l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+                    l->second.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
+                l->second.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                              - proxy::OP_MARGIN);
                 l->second.refreshSubscriberTimer->async_wait(
                     std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
@@ -1192,9 +1192,9 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         for (auto& put : search.second.puts) {
             doPut(key, put.second.value, [ok = put.second.ok](bool result) { *ok = result; }, time_point::max(), true);
             if (!put.second.refreshPutTimer) {
-                put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
+                put.second.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_);
             }
-            put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+            put.second.refreshPutTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                    - proxy::OP_MARGIN);
             put.second.refreshPutTimer->async_wait(
                 std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, put.first));
@@ -1210,8 +1210,7 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         // Connectivity changed, refresh all subscribe
         for (auto& search : searches_)
             for (auto& listener : search.second.listeners)
-                if (!listener.second.opstate->ok)
-                    resubscribe(search.first, listener.first, listener.second);
+                resubscribe(search.first, listener.first, listener.second);
         return;
     }
     if (logger_)
@@ -1275,10 +1274,10 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                 auto vid = std::stoull(vidIt->second);
                 auto& put = search.puts.at(vid);
                 if (!put.refreshPutTimer)
-                    put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
-                                                                               std::chrono::steady_clock::now());
+                    put.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
+                                                                               std::chrono::system_clock::now());
                 else
-                    put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
+                    put.refreshPutTimer->expires_at(std::chrono::system_clock::now());
                 put.refreshPutTimer->async_wait(
                     std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
                 ret = PushNotificationResult::PutRefresh;
@@ -1290,12 +1289,12 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                                                : PushNotificationResult::ListenRefresh;
             }
         } else {
-            auto key = InfoHash(notification.at("key"));
             system_clock::time_point sendTime = system_clock::time_point::min();
             try {
                 sendTime = system_clock::time_point(std::chrono::milliseconds(std::stoull(notification.at("t"))));
             } catch (...) {
             }
+            auto key = InfoHash(notification.at("key"));
             auto& search = searches_.at(key);
             auto expired = notification.find("exp");
             for (auto& list : search.listeners) {
@@ -1378,9 +1377,9 @@ DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
     header.method(restinio::http_method_subscribe());
     header.request_target("/key/" + key.toString());
     if (!listener.refreshSubscriberTimer) {
-        listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
+        listener.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
     }
-    listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+    listener.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     listener.refreshSubscriberTimer->async_wait(
         std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
     auto vcb = listener.cb;

src/dht_proxy_server.cpp

@@ -126,8 +126,8 @@ DhtProxyServer::ServerStats::toJson() const
     result["putCount"] = static_cast<Json::UInt64>(putCount);
     result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
     result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
-    result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
-    result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
+    result["serverStartTime"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(serverStartTime));
+    result["lastUpdated"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(lastUpdated));
     result["androidPush"] = androidPush.toJson();
     result["iosPush"] = iosPush.toJson();
     result["unifiedPush"] = unifiedPush.toJson();
@@ -210,7 +210,7 @@ DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
         clientId = cid->as<std::string>();
     }
     if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = std::chrono::system_clock::from_time_t(exp->as<time_t>());
     }
     if (auto token = findMapValue(o, "token"sv)) {
         pushToken = token->as<std::string>();
@@ -240,7 +240,7 @@ DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
         clientId = cid->as<std::string>();
     }
     if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = std::chrono::system_clock::from_time_t(exp->as<time_t>());
     }
     if (auto sid = findMapValue(o, "sid"sv)) {
         if (not sessionCtx)
@@ -414,7 +414,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                     logger_->debug("[proxy:server] Loading {} persistent puts", puts_.size());
                 for (auto& put : puts_) {
                     for (auto& pput : put.second.puts) {
-                        pput.second.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
+                        pput.second.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                        pput.second.expiration);
                         pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut,
                                                                       this,
@@ -437,7 +437,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                 }
                                 return json;
                             };
-                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            pput.second.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                                  pput.second.expiration
                                                                                                      - proxy::OP_MARGIN);
                             pput.second.expireNotifyTimer->async_wait(
@@ -450,7 +450,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                           pput.second.topic));
                         }
 #endif
-                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, time_point::max(), true);
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, dht::time_point::max(), true);
                     }
                 }
             } else {
@@ -487,7 +487,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                                                                  expired);
                                                });
                             // expire notify
-                            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                               listener.expiration
                                                                                                   - proxy::OP_MARGIN);
                             auto jsonProvider = [infoHash = listeners.first.toString(),
@@ -509,7 +509,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                           listener.type,
                                           listener.topic));
                             // cancel push listen
-                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                         listener.expiration);
                             listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
                                                                        this,
@@ -907,6 +907,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         Json::Value r;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
         auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
+
         if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) {
             auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
             response.set_body(RESP_MSG_JSON_INCORRECT);
@@ -951,14 +952,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         auto& listener = *listIt;
 
         // Expiration
-        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+        auto timeout = clock::now() + proxy::OP_TIMEOUT;
         listener.expiration = timeout;
         listener.type = type;
         listener.topic = topic;
         if (listener.expireNotifyTimer)
             listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
         else
-            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
+            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(), timeout - proxy::OP_MARGIN);
         auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() {
             Json::Value json;
             json["timeout"] = h;
@@ -975,7 +976,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
                                                          listener.type,
                                                          listener.topic));
         if (!listener.expireTimer)
-            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
         else
             listener.expireTimer->expires_at(timeout);
         listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
@@ -1405,7 +1406,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                     topic = pVal["topic"].asString();
                 }
                 std::lock_guard lock(lockSearchPuts_);
-                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+                auto timeout = clock::now() + proxy::OP_TIMEOUT;
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
                     for (auto& pp : sPuts.puts) {
@@ -1439,7 +1440,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                 pput.expiration = timeout;
                 if (not pput.expireTimer) {
                     // cancel permanent put
-                    pput.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+                    pput.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                     if (not pushToken.empty()) {
                         pput.pushToken = pushToken;
@@ -1478,7 +1479,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         return json;
                     };
                     if (!pput.expireNotifyTimer)
-                        pput.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                        pput.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                       timeout - proxy::OP_MARGIN);
                     else
                         pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -1507,7 +1508,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         response.done();
                     }
                 },
-                time_point::max(),
+                dht::time_point::max(),
                 permanent);
             return restinio::request_handling_status_t::accepted;
         } else {