proxy_client: give subscribe requests a deadline

A subscribe is a short request: the server answers immediately and the
device then relies on push notifications. Unlike a listen, nothing keeps
it under observation. The keepalive is only armed once the server has
answered, and no request in this client ever sets a timeout, so a
subscribe stalled in name resolution, connection or TLS handshake never
completes and never fails either.

The listener then stays registered locally while the server never learns
about it: the device stops being notified, without anything reporting an
error. This is most likely right after a wake-up, when the network stack
is still settling.

Arm a deadline on subscribe requests and fail the operation when it
expires, so restartListeners() sends the subscription again. Streaming
listens are left untouched: they are long lived by design.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

include/opendht/default_types.h

@@ -159,13 +159,7 @@ public:
     {}
 
     static Value::Filter getFilter() { return EncryptedValue::getFilter(); }
-    virtual void unpackValue(const dht::Value& v) override
-    {
-        BaseClass::unpackValue(v);
-        id = v.id;
-    }
 
-    dht::Value::Id id {0}; /// Value ID of the trust request, used for deserialization only.
     std::string service;
     std::string conversationId;
     Blob payload;

include/opendht/dht_proxy_server.h

@@ -146,26 +146,6 @@ public:
 
     std::shared_ptr<DhtRunner> getNode() const { return dht_; }
 
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-    /**
-     * Decide whether a batch of values notified to a push listener must be
-     * delivered as a high (urgent) priority push notification.
-     *
-     * Only real-time call signaling — values whose pushType is "audioCall" or
-     * "videoCall", published at high priority (priority == 0) — warrants an
-     * urgent push. Every other notification (presence/connection-request
-     * refreshes carrying an empty pushType, value expirations, etc.) is sent at
-     * normal priority so it does not saturate the mobile platform's adaptive
-     * high-priority quota; once that quota is tripped, the platform silently
-     * drops genuine call pushes.
-     *
-     * @param values the values being notified to the listener.
-     * @param expired whether this notification is for expired (removed) values.
-     * @return true if the notification must be sent at high priority.
-     */
-    static bool isHighPriorityPush(const std::vector<std::shared_ptr<Value>>& values, bool expired) noexcept;
-#endif // OPENDHT_PUSH_NOTIFICATIONS
-
 private:
     class ConnectionListener;
     struct RestRouterTraitsTls;

include/opendht/http.h

@@ -51,11 +51,11 @@ using ConnectHandlerCb = std::function<void(const asio::error_code& ec, const as
 using ssl_socket_t = restinio::impl::tls_socket_t;
 using socket_t = asio::ip::tcp::socket;
 
-struct OPENDHT_PUBLIC Url
+class OPENDHT_PUBLIC Url
 {
+public:
     Url() = default;
-    Url(std::string url);
-
+    Url(std::string_view url);
     std::string url;
     std::string protocol {"http"};
     std::string_view user;
@@ -149,7 +149,7 @@ public:
     using ResolverCb
         = std::function<void(const asio::error_code& ec, const std::vector<asio::ip::tcp::endpoint>& endpoints)>;
 
-    Resolver(asio::io_context& ctx, std::string url, std::shared_ptr<log::Logger> logger = {});
+    Resolver(asio::io_context& ctx, const std::string& url, std::shared_ptr<log::Logger> logger = {});
     Resolver(asio::io_context& ctx,
              std::string_view host,
              std::string_view service,
@@ -161,6 +161,10 @@ public:
              std::vector<asio::ip::tcp::endpoint> endpoints,
              const bool ssl = false,
              std::shared_ptr<log::Logger> logger = {});
+    Resolver(asio::io_context& ctx,
+             const std::string& url,
+             std::vector<asio::ip::tcp::endpoint> endpoints,
+             std::shared_ptr<log::Logger> logger = {});
 
     ~Resolver();
 
@@ -212,19 +216,19 @@ public:
 
     // resolves implicitly
     Request(asio::io_context& ctx,
-            std::string url,
+            const std::string& url,
             const Json::Value& json,
             OnJsonCb jsoncb,
             std::shared_ptr<log::Logger> logger = {});
-    Request(asio::io_context& ctx, std::string url, OnJsonCb jsoncb, std::shared_ptr<log::Logger> logger = {});
+    Request(asio::io_context& ctx, const std::string& url, OnJsonCb jsoncb, std::shared_ptr<log::Logger> logger = {});
 
-    Request(asio::io_context& ctx, std::string url, std::shared_ptr<log::Logger> logger = {});
+    Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<log::Logger> logger = {});
     Request(asio::io_context& ctx,
             std::string_view host,
             std::string_view service,
             const bool ssl = false,
             std::shared_ptr<log::Logger> logger = {});
-    Request(asio::io_context& ctx, std::string url, OnDoneCb onDone, std::shared_ptr<log::Logger> logger = {});
+    Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<log::Logger> logger = {});
 
     // user defined resolver
     Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, sa_family_t family = AF_UNSPEC);
@@ -340,6 +344,8 @@ private:
 
     dht::crypto::Identity client_identity_;
     std::shared_ptr<dht::crypto::Certificate> server_ca_;
+    std::string service_;
+    std::string host_;
 
     unsigned int id_;
     static std::atomic_uint ids_;

include/opendht/proxy.h

@@ -10,6 +10,14 @@ namespace proxy {
 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
 constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
 
+/**
+ * Deadline for a subscribe request to complete. Unlike a listen, a subscribe is a short
+ * request: the server answers immediately and the device then relies on push notifications.
+ * Nothing else observes it, so without a deadline a subscribe stalled in name resolution,
+ * connection or TLS handshake would never complete nor fail.
+ */
+constexpr const std::chrono::seconds SUBSCRIBE_TIMEOUT {20};
+
 using ListenToken = uint64_t;
 
 } // namespace proxy

src/base64.cpp

@@ -36,19 +36,4 @@ base64_decode(std::string_view str)
     return buffer;
 }
 
-std::vector<unsigned char>
-base64_default_or_url_decode(std::string_view str)
-{
-    std::vector<unsigned char> buffer(simdutf::maximal_binary_length_from_base64(str.data(), str.size()));
-    simdutf::result r = simdutf::base64_to_binary(str.data(),
-                                                  str.size(),
-                                                  (char*) buffer.data(),
-                                                  simdutf::base64_options::base64_default_or_url);
-    if (r.error) {
-        throw std::invalid_argument(concat("Invalid base64 input: "sv, simdutf::error_to_string(r.error)));
-    } else {
-        buffer.resize(r.count);
-    }
-    return buffer;
-}
 } // namespace dht

src/base64.h

@@ -21,12 +21,5 @@ std::string base64_encode(const std::vector<unsigned char>& str);
  * @return a base64-decoded buffer
  */
 std::vector<unsigned char> base64_decode(std::string_view str);
-/**
- * Decode a buffer in base64 or url style base64 (including '-' '_' ).
- *
- * @param str the input buffer
- * @return a base64-decoded buffer
- */
-std::vector<unsigned char> base64_default_or_url_decode(std::string_view str);
 
-} // namespace dht
+} // namespace dht

src/dht_proxy_client.cpp

@@ -1075,6 +1075,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
         request->set_body(body);
 #endif
         auto rxBuf = std::make_shared<LineSplit>();
+        // Completion flag shared with the subscribe deadline armed below. The request
+        // callbacks replace each other rather than accumulating, so the flag is raised from
+        // the single on_done callback instead of a dedicated one.
+        std::shared_ptr<std::atomic_bool> subscribeDone;
+        if (method != ListenMethod::LISTEN)
+            subscribeDone = std::make_shared<std::atomic_bool>(false);
         request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) {
             try {
                 auto& b = *rxBuf;
@@ -1113,7 +1119,9 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                 opstate->ok.store(false);
             }
         });
-        request->add_on_done_callback([this, opstate, reqid](const http::Response& response) {
+        request->add_on_done_callback([this, opstate, reqid, subscribeDone](const http::Response& response) {
+            if (subscribeDone)
+                subscribeDone->store(true);
             if (response.status_code != 200) {
                 if (logger_)
                     logger_->error("[proxy:client] [listen] send request #{} failed with code={}",
@@ -1138,6 +1146,29 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
                 request->get_connection()->set_keepalive(seconds);
             }
         });
+        if (subscribeDone) {
+            // Give the subscribe a deadline. Nothing else would ever release it: the
+            // keepalive above is only armed once the server has answered, and no other
+            // request in this client sets a timeout. A subscribe that never completes
+            // leaves the listener registered locally while the server never learns about
+            // it, so the device stops being notified without anything reporting an error.
+            // Cancelling runs the on_done callback, which clears the request, and failing
+            // the operation lets restartListeners() send the subscription again.
+            std::weak_ptr<http::Request> wreq = request;
+            request->timeout(proxy::SUBSCRIBE_TIMEOUT,
+                             [this, reqid, opstate, subscribeDone, wreq](const asio::error_code& ec) {
+                                 if (ec == asio::error::operation_aborted or subscribeDone->load()
+                                     or isDestroying_)
+                                     return;
+                                 if (logger_)
+                                     logger_->error("[proxy:client] [listen] subscribe request #{} timed out",
+                                                    reqid);
+                                 opstate->ok.store(false);
+                                 if (auto req = wreq.lock())
+                                     req->cancel();
+                                 opFailed();
+                             });
+        }
         request->send();
     } catch (const std::exception& e) {
         if (logger_)

src/dht_proxy_server.cpp

@@ -1134,25 +1134,6 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code& ec,
         pushListeners_.erase(pushListener);
 }
 
-bool
-DhtProxyServer::isHighPriorityPush(const std::vector<std::shared_ptr<Value>>& values, bool expired) noexcept
-{
-    // Expired-value notifications (de-registration) are never time-critical.
-    if (expired)
-        return false;
-    // Promote to high (urgent) push priority only for real-time call signaling.
-    // Background traffic — most notably connection-request refreshes, which
-    // carry an empty pushType — stays at normal priority so it does not trip the
-    // mobile platform's adaptive high-priority quota, which would otherwise
-    // silently drop genuine call pushes once the high-priority noise ratio gets
-    // too high.
-    for (const auto& v : values) {
-        if (v->priority == 0 and (v->pushType == "videoCall" or v->pushType == "audioCall"))
-            return true;
-    }
-    return false;
-}
-
 bool
 DhtProxyServer::handlePushListen(const InfoHash& infoHash,
                                  const std::string& pushToken,
@@ -1195,23 +1176,20 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash,
         json["exp"] = json["ids"];
     }
 
-    const bool highPriority = isHighPriorityPush(values, expired);
+    auto minPriority = 1000u;
+    for (const auto& v : values)
+        minPriority = std::min(minPriority, v->priority);
 
-    if (logger_) {
-        auto minPriority = 1000u;
-        for (const auto& v : values)
-            minPriority = std::min(minPriority, v->priority);
-        logger_->debug("[proxy:server] [listen {}] [client {}] [session {}] [expired {}] [priority {}] [high {}] [values {}]",
+    if (logger_)
+        logger_->debug("[proxy:server] [listen {}] [client {}] [session {}] [expired {}] [priority {}] [values {}]",
                        infoHash,
                        clientId,
                        sessionCtx->sessionId,
                        expired,
                        minPriority,
-                       highPriority,
                        values.size());
-    }
 
-    sendPushNotification(pushToken, std::move(json), type, highPriority, topic);
+    sendPushNotification(pushToken, std::move(json), type, !expired and minPriority == 0, topic);
 
     return true;
 }
@@ -1227,7 +1205,14 @@ DhtProxyServer::sendPushNotification(
     try {
         std::shared_ptr<http::Request> request;
         if (type == PushType::UnifiedPush) {
-            request = std::make_shared<http::Request>(io_context(), token, logger_);
+            http::Url tokenUrl(token);
+            request = std::make_shared<http::Request>(io_context(),
+                                                      concat(tokenUrl.protocol, "://"sv, tokenUrl.host),
+                                                      tokenUrl.service,
+                                                      tokenUrl.protocol.find("https") == 0,
+                                                      logger_);
+            request->set_target(tokenUrl.target);
+            request->set_header_field(restinio::http_field_t::host, std::string(tokenUrl.host));
         } else {
             request = std::make_shared<http::Request>(io_context(),
                                                       pushHostPort_.first,
@@ -1246,8 +1231,8 @@ DhtProxyServer::sendPushNotification(
             Blob auth;   ///< Auth secret
             auto delim = topicView.find('|');
             if (delim != std::string_view::npos) {
-                pubKey = base64_default_or_url_decode(topicView.substr(0, delim));
-                auth = base64_default_or_url_decode(topicView.substr(delim + 1));
+                pubKey = base64_decode(topicView.substr(0, delim));
+                auth = base64_decode(topicView.substr(delim + 1));
             }
 
             request->set_header_field(restinio::http_field_t::ttl, "86400");

src/http.cpp

@@ -42,8 +42,8 @@ constexpr std::string_view ORIGIN_PROTOCOL {"//"sv};
 constexpr const char HTTP_HEADER_DELIM[] = "\r\n\r\n";
 constexpr unsigned MAX_REDIRECTS {5};
 
-Url::Url(std::string url_str)
-    : url(std::move(url_str))
+Url::Url(std::string_view url_str)
+    : url(url_str)
 {
     std::string_view u = url;
     if (u.empty())
@@ -830,8 +830,8 @@ Connection::timeout(const std::chrono::seconds& timeout, HandlerCb cb)
 
 // Resolver
 
-Resolver::Resolver(asio::io_context& ctx, std::string url, std::shared_ptr<dht::Logger> logger)
-    : url_(std::move(url))
+Resolver::Resolver(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger)
+    : url_(url)
     , resolver_(ctx)
     , destroyed_(std::make_shared<bool>(false))
     , logger_(logger)
@@ -967,14 +967,14 @@ Resolver::resolve(std::string_view host, std::string_view service)
 std::atomic_uint Request::ids_ {1};
 
 Request::Request(asio::io_context& ctx,
-                 std::string url,
+                 const std::string& url,
                  const Json::Value& json,
                  OnJsonCb jsoncb,
                  std::shared_ptr<dht::Logger> logger)
     : logger_(std::move(logger))
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
+    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
 {
     init_default_headers();
     set_header_field(restinio::http_field_t::content_type, std::string(HTTP_HEADER_CONTENT_TYPE_JSON));
@@ -997,11 +997,11 @@ Request::Request(asio::io_context& ctx,
     });
 }
 
-Request::Request(asio::io_context& ctx, std::string url, OnJsonCb jsoncb, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, const std::string& url, OnJsonCb jsoncb, std::shared_ptr<dht::Logger> logger)
     : logger_(std::move(logger))
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
+    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
 {
     init_default_headers();
     set_header_field(restinio::http_field_t::accept, std::string(HTTP_HEADER_CONTENT_TYPE_JSON));
@@ -1022,20 +1022,20 @@ Request::Request(asio::io_context& ctx, std::string url, OnJsonCb jsoncb, std::s
     });
 }
 
-Request::Request(asio::io_context& ctx, std::string url, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger)
     : logger_(logger)
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
+    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
 {
     init_default_headers();
 }
 
-Request::Request(asio::io_context& ctx, std::string url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
     : logger_(logger)
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
+    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
 {
     init_default_headers();
     add_on_done_callback(std::move(onDone));
@@ -1086,6 +1086,7 @@ Request::Request(asio::io_context& ctx,
     , family_(family)
     , resolver_(resolver)
 {
+    set_header_field(restinio::http_field_t::host, concat(get_url().host, ":", get_url().service));
     set_target(Url(target).target);
 }
 

tests/test_dhtproxy.cpp

@@ -327,39 +327,4 @@ DhtProxyTester::testShutdownStop()
     CPPUNIT_ASSERT_EQUAL(2 * C, callback_count.load());
 }
 
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-void
-DhtProxyTester::testPushNotificationPriority()
-{
-    auto mkValue = [](unsigned priority, const std::string& pushType) {
-        auto v = std::make_shared<dht::Value>();
-        v->priority = priority;
-        v->pushType = pushType;
-        return v;
-    };
-
-    // Real-time call signaling at high priority => urgent push.
-    CPPUNIT_ASSERT(dht::DhtProxyServer::isHighPriorityPush({mkValue(0, "videoCall")}, false));
-    CPPUNIT_ASSERT(dht::DhtProxyServer::isHighPriorityPush({mkValue(0, "audioCall")}, false));
-
-    // Connection-request refresh (empty pushType) is noise => normal push.
-    CPPUNIT_ASSERT(!dht::DhtProxyServer::isHighPriorityPush({mkValue(0, "")}, false));
-
-    // Any other pushType is not a call => normal push.
-    CPPUNIT_ASSERT(!dht::DhtProxyServer::isHighPriorityPush({mkValue(0, "message")}, false));
-
-    // Expired (de-registration) notifications are never urgent, even for calls.
-    CPPUNIT_ASSERT(!dht::DhtProxyServer::isHighPriorityPush({mkValue(0, "videoCall")}, true));
-
-    // A call value explicitly published at normal priority stays normal.
-    CPPUNIT_ASSERT(!dht::DhtProxyServer::isHighPriorityPush({mkValue(1, "videoCall")}, false));
-
-    // Mixed batch: a single high-priority call value is enough to make it urgent.
-    CPPUNIT_ASSERT(dht::DhtProxyServer::isHighPriorityPush({mkValue(0, ""), mkValue(0, "videoCall")}, false));
-
-    // Empty batch => normal.
-    CPPUNIT_ASSERT(!dht::DhtProxyServer::isHighPriorityPush({}, false));
-}
-#endif // OPENDHT_PUSH_NOTIFICATIONS
-
 } // namespace test

tests/test_dhtproxy.h

@@ -21,9 +21,6 @@ class DhtProxyTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testPutGet40KChars);
     CPPUNIT_TEST(testFuzzy);
     CPPUNIT_TEST(testShutdownStop);
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-    CPPUNIT_TEST(testPushNotificationPriority);
-#endif
     CPPUNIT_TEST_SUITE_END();
 
 public:
@@ -58,15 +55,6 @@ public:
 
     void testShutdownStop();
 
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-    /**
-     * The proxy must only request a high-priority push for real-time call
-     * signaling (audioCall/videoCall at priority 0); everything else stays at
-     * normal priority.
-     */
-    void testPushNotificationPriority();
-#endif
-
 private:
     dht::DhtRunner::Config clientConfig {};
     dht::DhtRunner nodePeer;