proxy server: only send call pushes at high priority

Replace the priority-only heuristic in handlePushListen
(highPriority = !expired && minPriority == 0) with isHighPriorityPush,
so the proxy promotes a notification to high (urgent) push priority
only for real-time call signaling.

Previously every value published at the default priority 0 — including
the frequent connection-request refreshes that carry an empty
pushType — triggered a high-priority push. On mobile this floods the
platform's adaptive high-priority quota; once tripped, the platform
silently drops genuine call pushes, so incoming calls fail to ring
while the device is in the background.

Restricting high priority to audioCall/videoCall keeps calls waking the
device while the surrounding presence/sync noise is delivered at normal
priority. This is a strict subset of the previous high-priority set, so
no notification that was normal before becomes high.

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

doc/dhtnode.1

@@ -1,102 +1,193 @@
-.TH DHTNODE 1 2019-06-08
+.TH DHTNODE 1 2026-04-18
 .SH NAME
 .B dhtnode
 - a simple OpenDHT command line node runner.
 .SH SYNOPSIS
 .B dhtnode
 [\fB\-h\fR]
+[\fB\-V\fR]
 [\fB\-v\fR [\fB\-l\fR \fIlogfile\fR] [\fB\-L\fR]]
-[\fB\-i\fR [\fB\-\-save\-identity\fR \fIfile\fR]]
+[\fB\-i\fR [\fB\-I\fR \fIfile\fR]]
+[\fB\-c\fR \fIfile\fR]
+[\fB\-k\fR \fIfile\fR]
+[\fB\-m\fR \fIpassword\fR]
 [\fB\-d\fR] [\fB\-s\fR]
+[\fB\-D\fR] [\fB\-U\fR] [\fB\-P\fR]
 [\fB\-n\fR \fInetwork_id\fR]
 [\fB\-p\fR \fIlocal_port\fR]
 [\fB\-b\fR \fIbootstrap_host\fR[:\fIport\fR]]
-[\fB\-\-certificate\fR \fIfile\fR]
-[\fB\-\-privkey\fR \fIfile\fR]
-[\fB\-\-privkey\-password\fR \fIpassword\fR]
-[\fB\-\-proxyserver\fR \fIport\fR]
-[\fB\-\-proxyclient\fR \fIserver\fR]
+[\fB\-f\fR \fIpath\fR]
+[\fB\-S\fR \fIport\fR]
+[\fB\-e\fR \fIport\fR]
+[\fB\-a\fR \fIaddress\fR]
+[\fB\-w\fR \fIfile\fR]
+[\fB\-K\fR \fIfile\fR]
+[\fB\-M\fR \fIpassword\fR]
+[\fB\-C\fR \fIserver\fR]
+[\fB\-y\fR \fIendpoint\fR]
+[\fB\-z\fR \fIkey\fR]
+[\fB\-u\fR \fIbundleid\fR]
 .SH DESCRIPTION
-Runs an OpenDHT node, with a CLI (default) or as a daemon (with \fB'-d'\fP or \fB'-s'\fP).
+Runs an OpenDHT node, with an interactive CLI (default) or as a daemon (with \fB'-d'\fP or \fB'-s'\fP).
 Commands available in the interactive shell are:
 .EX
     h, help    Print this help message.
-    q, quit    Quit the program.
+    x, quit    Quit the program.
     log        Start/stop printing DHT logs.
 
     Node information:
     ll         Print basic information and stats about the current node.
-    ls         Print basic information about current searches.
-    ld         Print basic information about currenty stored values on this node.
-    lr         Print the full current routing table of this node
+    ls [key]   Print basic information about current search(es).
+    ld [key]   Print basic information about currenty stored values on this
+               node (or for a specific key).
+    lr         Print the full current routing table of this node.
 
     Operations on the DHT:
-    b ip:port             Ping potential node at given IP address/port.
-    g [key]               Get values at [key].
-    l [key]               Listen for value changes at [key].
-    p [key] [str]         Put string value at [key].
-    s [key] [str]         Put string value at [key], signed with our generated
-                          private key.
-    e [key] [dest] [str]  Put string value at [key], encrypted for [dest] with
-                          its public key (if found).
+    b <ip:port>            Ping potential node at given IP address/port.
+    g <key>                Get values at <key>.
+    l <key>                Listen for value changes at <key>.
+    cl <key> <token>       Cancel listen for <token> and <key>.
+    p <key> <str>          Put string value at <key>.
+    pp <key> <str>         Put string value at <key> (persistent version).
+    cpp <key> <id>         Cancel persistent put operation for <key> and
+                           value <id>.
+    s <key> <str>          Put string value at <key>, signed with our
+                           generated private key.
+    e <key> <dest> <str>   Put string value at <key>, encrypted for <dest>
+                           with its public key (if found).
+    cc                     Trigger connectivity changed signal.
+
+    Operations with the proxy (if compiled with proxy server support):
+    pst [port] <pushServer>  Start the proxy interface on port.
+    psx [port]               Start the proxy SSL interface on port.
+    psp [port]               Stop the proxy interface on port.
+
+    Operations with the proxy (if compiled with proxy client support):
+    stt [server_address] <device_key>  Start the proxy client.
+    stp                                Stop the proxy client.
+    rs  [token]                        Resubscribe to opendht.
+    rp  [token]                        Inject a push notification.
 .EE
 .SH OPTIONS
 .TP
-\fB\-h\fP
-Prints some help.
+\fB\-h\fP, \fB\-\-help\fP
+Prints help and exits.
 .TP
-\fB\-v\fP
-Enable the verbose mode (log to stdout/stderr by default)
+\fB\-V\fP, \fB\-\-version\fP
+Prints version information and exits.
 .TP
-\fB\-l\fP \fIlog_file\fP
-Write log to file instead of stdout/stderr
+\fB\-v\fP, \fB\-\-verbose\fP
+Enable verbose mode (log to stdout/stderr by default).
 .TP
-\fB\-L\fP
-Write log to syslog instead of stdout/stderr
+\fB\-l\fP \fIlogfile\fP, \fB\-\-logfile\fP \fIlogfile\fP
+Write log to file instead of stdout/stderr.
 .TP
-\fB\-i\fP
-Generate cryptographic identity for the node.
+\fB\-L\fP, \fB\-\-syslog\fP
+Write log to syslog instead of stdout/stderr (enables verbose mode).
 .TP
-\fB\-\-save\-identity\fP \fIfile\fP
+\fB\-i\fP, \fB\-\-identity\fP
+Generate a cryptographic identity (RSA key pair and self-signed certificate)
+for the node. Required to perform cryptographic operations (encrypt/sign values)
+if no private key file is provided.
+.TP
+\fB\-I\fP \fIfile\fP, \fB\-\-save\-identity\fP \fIfile\fP
 Save generated identity (certificate and private key) to given file prefix.
+Files are saved as \fIfile\fP and \fIfile\fP_ca.
 .TP
-\fB\-\-certificate\fP \fIfile\fP
-Load identity certificate from given file.
+\fB\-c\fP \fIfile\fP, \fB\-\-certificate\fP \fIfile\fP
+Load identity certificate from given file. If specified, the certificate is
+published on the DHT.
 .TP
-\fB\-\-privkey\fP \fIfile\fP
-Load identity private key from given file.
+\fB\-k\fP \fIfile\fP, \fB\-\-privkey\fP \fIfile\fP
+Load identity private key from given file. Required to perform cryptographic
+operations if no identity is generated.
 .TP
-\fB\-\-privkey\-password\fP \fIpassword\fP
+\fB\-m\fP \fIpassword\fP, \fB\-\-privkey\-password\fP \fIpassword\fP
 Password to use for private key encryption or decryption (optional).
 .TP
-\fB\-d\fP
-Run the program in daemon mode (will fork in the background).
+\fB\-d\fP, \fB\-\-daemonize\fP
+Run the program in daemon mode (will fork in the background). When verbose mode
+is enabled, logs to syslog by default.
 .TP
-\fB\-s\fP
+\fB\-s\fP, \fB\-\-service\fP
 Run the program in service mode (non\-forking daemon).
 .TP
-\fB\-D\fP
+\fB\-D\fP, \fB\-\-peer\-discovery\fP
 Enables multicast automatic local peer discovery.
 .TP
-\fB\-f\fP \fIfile\fP
-Specify a file path to persist/load the node state.
+\fB\-U\fP, \fB\-\-no\-rate\-limit\fP
+Disable rate limiting (allows unlimited requests and storage).
 .TP
-\fB\-n\fP \fInetwork_id\fP
-Specify the network id. This let you connect to distinct networks and prevents
-the merge of two different networks (available since OpenDHT v0.6.1).
+\fB\-P\fP, \fB\-\-public\-stable\fP
+Enable public stable mode.
 .TP
-\fB\-p\fP \fIlocal_port\fP
-Use UDP port \fIlocal_port\fP for the program to bind to.
+\fB\-f\fP \fIpath\fP, \fB\-\-persist\fP \fIpath\fP
+Specify a file path to persist/load the node state (routing table and values).
 .TP
-\fB\-b\fP \fIbootstrap_host\fP[:\fIport\fP]
-The program needs to be given a node to connect to the network. You use this
-option to provide the ip address of that node.
+\fB\-n\fP \fInetwork_id\fP, \fB\-\-net\fP \fInetwork_id\fP
+Specify the network id. This lets you connect to distinct networks and prevents
+the merge of two different networks. Defaults to 0 (the main public network).
 .TP
-\fB\-\-proxyserver\fP \fIlocal_port\fP
-Run a proxy server bound to this DHT node on HTTP port \fIlocal_port\fP
+\fB\-p\fP \fIlocal_port\fP, \fB\-\-port\fP \fIlocal_port\fP
+Use UDP port \fIlocal_port\fP for the program to bind to. If not set, will use
+any available port. The default OpenDHT port for public nodes is 4222.
 .TP
-\fB\-\-proxyclient\fP \fIserver\fP
-Run this DHT node in proxy client mode, and connect to \fIserver\fP
+\fB\-b\fP \fIbootstrap_host\fP[:\fIport\fP], \fB\-\-bootstrap\fP \fIbootstrap_host\fP[:\fIport\fP]
+Specify a bootstrap node address to connect to the network.
+.TP
+\fB\-S\fP \fIlocal_port\fP, \fB\-\-proxyserver\fP \fIlocal_port\fP
+Run a REST API proxy server bound to this DHT node on HTTP port \fIlocal_port\fP.
+.TP
+\fB\-e\fP \fIlocal_port\fP, \fB\-\-proxyserverssl\fP \fIlocal_port\fP
+Run a REST API proxy server with SSL bound to this DHT node on HTTPS port \fIlocal_port\fP.
+.TP
+\fB\-a\fP \fIaddress\fP, \fB\-\-proxy\-addr\fP \fIaddress\fP
+Proxy server bind address.
+.TP
+\fB\-w\fP \fIfile\fP, \fB\-\-proxy\-certificate\fP \fIfile\fP
+Load proxy SSL certificate from given file.
+.TP
+\fB\-K\fP \fIfile\fP, \fB\-\-proxy\-privkey\fP \fIfile\fP
+Load proxy SSL private key from given file.
+.TP
+\fB\-M\fP \fIpassword\fP, \fB\-\-proxy\-privkey\-password\fP \fIpassword\fP
+Password for the proxy private key encryption or decryption (optional).
+.TP
+\fB\-C\fP \fIserver\fP, \fB\-\-proxyclient\fP \fIserver\fP
+Run this DHT node in proxy client mode, and connect to \fIserver\fP.
+.TP
+\fB\-y\fP \fIendpoint\fP, \fB\-\-pushserver\fP \fIendpoint\fP
+Push notification server endpoint.
+.TP
+\fB\-z\fP \fIkey\fP, \fB\-\-devicekey\fP \fIkey\fP
+Device key for push notifications.
+.TP
+\fB\-u\fP \fIbundleid\fP, \fB\-\-bundleid\fP \fIbundleid\fP
+Bundle ID for push notifications.
+.SH EXAMPLES
+Run a node using bootstrap.jami.net as a bootstrap:
+.PP
+.EX
+dhtnode -b bootstrap.jami.net:4222
+.EE
+.PP
+Run a node as a non-forking daemon on port 4222:
+.PP
+.EX
+dhtnode -b bootstrap.jami.net -p 4222 -s
+.EE
+.PP
+Run a proxy server on port 8000:
+.PP
+.EX
+dhtnode -b bootstrap.jami.net:4222 --proxyserver 8000
+.EE
+.PP
+Use a proxy client to connect to a proxy server:
+.PP
+.EX
+dhtnode --proxyclient dhtproxy.jami.net:80
+.EE
 .SH AUTHORS
 .TP
 Program written by

include/opendht/default_types.h

@@ -159,7 +159,13 @@ 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,6 +146,26 @@ 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/dhtrunner.h

@@ -299,13 +299,11 @@ public:
         putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
     }
 
-    [[deprecated("Use the shared_ptr version instead")]]
     void putEncrypted(InfoHash hash,
                       const std::shared_ptr<crypto::PublicKey>& to,
                       Value&& value,
                       DoneCallback cb = {},
                       bool permanent = false);
-    [[deprecated("Use the shared_ptr version instead")]]
     void putEncrypted(InfoHash hash,
                       const std::shared_ptr<crypto::PublicKey>& to,
                       Value&& value,
@@ -322,6 +320,14 @@ public:
     {
         putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
     }
+    void putEncrypted(InfoHash hash, const PkId& to, Value&& value, DoneCallback cb = {}, bool permanent = false)
+    {
+        putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
+    }
+    void putEncrypted(InfoHash hash, const PkId& to, Value&& value, DoneCallbackSimple cb, bool permanent = false)
+    {
+        putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), bindDoneCb(cb), permanent);
+    }
 
     /**
      * Insert known nodes to the routing table, without necessarly ping them.

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;
 
-class OPENDHT_PUBLIC Url
+struct OPENDHT_PUBLIC Url
 {
-public:
     Url() = default;
-    Url(std::string_view url);
+    Url(std::string 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, const std::string& url, std::shared_ptr<log::Logger> logger = {});
+    Resolver(asio::io_context& ctx, std::string url, std::shared_ptr<log::Logger> logger = {});
     Resolver(asio::io_context& ctx,
              std::string_view host,
              std::string_view service,
@@ -161,10 +161,6 @@ 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();
 
@@ -216,19 +212,19 @@ public:
 
     // resolves implicitly
     Request(asio::io_context& ctx,
-            const std::string& url,
+            std::string url,
             const Json::Value& json,
             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, OnJsonCb jsoncb, 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 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, const std::string& url, OnDoneCb onDone, std::shared_ptr<log::Logger> logger = {});
+    Request(asio::io_context& ctx, 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);
@@ -344,8 +340,6 @@ 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/thread_pool.h

@@ -55,10 +55,12 @@ public:
 private:
     std::mutex lock_;
     std::condition_variable cv_ {};
+    std::condition_variable stopCv_ {};
     std::queue<std::function<void()>> tasks_ {};
     std::vector<std::unique_ptr<std::thread>> threads_;
     unsigned readyThreads_ {0};
     bool running_ {true};
+    bool stopping_ {false};
 
     unsigned minThreads_;
     const unsigned maxThreads_;

src/base64.cpp

@@ -36,4 +36,19 @@ 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,5 +21,12 @@ 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_server.cpp

@@ -1134,6 +1134,25 @@ 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,
@@ -1176,20 +1195,23 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash,
         json["exp"] = json["ids"];
     }
 
-    auto minPriority = 1000u;
-    for (const auto& v : values)
-        minPriority = std::min(minPriority, v->priority);
+    const bool highPriority = isHighPriorityPush(values, expired);
 
-    if (logger_)
-        logger_->debug("[proxy:server] [listen {}] [client {}] [session {}] [expired {}] [priority {}] [values {}]",
+    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 {}]",
                        infoHash,
                        clientId,
                        sessionCtx->sessionId,
                        expired,
                        minPriority,
+                       highPriority,
                        values.size());
+    }
 
-    sendPushNotification(pushToken, std::move(json), type, !expired and minPriority == 0, topic);
+    sendPushNotification(pushToken, std::move(json), type, highPriority, topic);
 
     return true;
 }
@@ -1205,14 +1227,7 @@ DhtProxyServer::sendPushNotification(
     try {
         std::shared_ptr<http::Request> request;
         if (type == PushType::UnifiedPush) {
-            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));
+            request = std::make_shared<http::Request>(io_context(), token, logger_);
         } else {
             request = std::make_shared<http::Request>(io_context(),
                                                       pushHostPort_.first,
@@ -1231,8 +1246,8 @@ DhtProxyServer::sendPushNotification(
             Blob auth;   ///< Auth secret
             auto delim = topicView.find('|');
             if (delim != std::string_view::npos) {
-                pubKey = base64_decode(topicView.substr(0, delim));
-                auth = base64_decode(topicView.substr(delim + 1));
+                pubKey = base64_default_or_url_decode(topicView.substr(0, delim));
+                auth = base64_default_or_url_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_view url_str)
-    : url(url_str)
+Url::Url(std::string url_str)
+    : url(std::move(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, const std::string& url, std::shared_ptr<dht::Logger> logger)
-    : url_(url)
+Resolver::Resolver(asio::io_context& ctx, std::string url, std::shared_ptr<dht::Logger> logger)
+    : url_(std::move(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,
-                 const std::string& url,
+                 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, url, logger))
+    , resolver_(std::make_shared<Resolver>(ctx, std::move(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, const std::string& url, OnJsonCb jsoncb, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, 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, url, logger))
+    , resolver_(std::make_shared<Resolver>(ctx, std::move(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, const std::string& url, OnJsonCb jsoncb,
     });
 }
 
-Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, std::string url, std::shared_ptr<dht::Logger> logger)
     : logger_(logger)
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
+    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
 {
     init_default_headers();
 }
 
-Request::Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
+Request::Request(asio::io_context& ctx, std::string url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
     : logger_(logger)
     , id_(Request::ids_++)
     , ctx_(ctx)
-    , resolver_(std::make_shared<Resolver>(ctx, url, logger))
+    , resolver_(std::make_shared<Resolver>(ctx, std::move(url), logger))
 {
     init_default_headers();
     add_on_done_callback(std::move(onDone));
@@ -1086,7 +1086,6 @@ 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);
 }
 

src/thread_pool.cpp

@@ -78,6 +78,8 @@ ThreadPool::run(std::function<void()>&& cb)
                             break;
                         task = std::move(tasks_.front());
                         tasks_.pop();
+                        if (tasks_.empty() && stopping_)
+                            stopCv_.notify_one();
                     }
 
                     // run task
@@ -135,7 +137,9 @@ ThreadPool::stop(bool wait)
 {
     std::unique_lock l(lock_);
     if (wait) {
-        cv_.wait(l, [&]() { return tasks_.empty(); });
+        stopping_ = true;
+        stopCv_.wait(l, [&]() { return tasks_.empty(); });
+        stopping_ = false;
     }
     running_ = false;
     tasks_ = {};

tests/test_dhtproxy.cpp

@@ -327,4 +327,39 @@ 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,6 +21,9 @@ 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:
@@ -55,6 +58,15 @@ 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;

tests/test_threadpool.cpp

@@ -80,6 +80,46 @@ ThreadPoolTester::testContext()
     CPPUNIT_ASSERT_EQUAL(N, count.load());
 }
 
+void
+ThreadPoolTester::testJoinNonEmptyTaskQueue()
+{
+    dht::ThreadPool pool(2);
+
+    constexpr unsigned N = 128;
+    std::atomic_uint count {0};
+
+    // Schedule enough work so that the task queue is non-empty when calling
+    // pool.join() below.
+    for (unsigned i = 0; i < N; i++)
+        pool.run([&] {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            count++;
+        });
+
+    std::atomic_bool joined {false};
+    std::condition_variable cv;
+    std::thread joinThread([&] {
+        CPPUNIT_ASSERT(count.load() < N);
+        pool.join();
+        joined = true;
+        cv.notify_one();
+    });
+
+    // After calling join(), all pending tasks should complete, and then all
+    // threads in the pool should join.
+    std::mutex mtx;
+    std::unique_lock lock(mtx);
+    bool poolJoined = cv.wait_for(lock, std::chrono::seconds(10), [&] { return joined.load(); });
+    if (!poolJoined) {
+        // Detach joinThread if we timed out, otherwise the test can crash.
+        joinThread.detach();
+    } else {
+        joinThread.join();
+    }
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Some tasks were not completed", N, count.load());
+    CPPUNIT_ASSERT_MESSAGE("Thread pool failed to join before timeout", poolJoined);
+}
+
 void
 ThreadPoolTester::tearDown()
 {}

tests/test_threadpool.h

@@ -14,6 +14,7 @@ class ThreadPoolTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testThreadPool);
     CPPUNIT_TEST(testExecutor);
     CPPUNIT_TEST(testContext);
+    CPPUNIT_TEST(testJoinNonEmptyTaskQueue);
     CPPUNIT_TEST_SUITE_END();
 
 public:
@@ -29,6 +30,7 @@ public:
     void testThreadPool();
     void testExecutor();
     void testContext();
+    void testJoinNonEmptyTaskQueue();
 };
 
 } // namespace test

tools/dhtchat.cpp

@@ -123,7 +123,7 @@ main(int argc, char** argv)
                     iss >> idstr;
                     std::getline(iss, line);
                     dht.putEncrypted(room,
-                                     InfoHash(idstr),
+                                     PkId(idstr),
                                      dht::ImMessage(IdDist()(tools::rd), std::move(line), now),
                                      [](bool ok) {
                                          // dht.cancelPut(room, id);