proxy server: collapse low-priority push notifications

When a device is unreachable (deep doze, offline), FCM queues
non-collapsible messages individually, with a hard limit of 100
pending messages per device. A Jami device that is a member of a few
active swarms receives a sustained stream of connection-churn and
expiration notifications (observed: 27-44/min, ~12k/day), so a phone
suspended for a few tens of minutes is enough to overflow the queue.
Once full, FCM silently drops messages, including the high-priority
notification announcing an incoming call: the callee never wakes up
and the caller gets a timeout after the ICE window (observed in
production, with the call notification never delivered while pushes
sent after the device woke up arrived in under a second).

Mark Android notifications as collapsible (collapse_key) when they
are neither high-priority (calls, messages, invitations - as set by
the client through Value::priority) nor resubscribe requests. While
collapsed messages are pending, FCM only keeps the last one per key,
so routine sync noise can no longer evict actionable notifications.

A single constant key is used instead of one key per DHT listen key:
FCM only honors 4 distinct collapse keys per device, and the Android
client reacts to any low-priority notification the same way (wake up
and resynchronize against the proxy), so finer granularity would not
change client behavior.

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

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/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_;

python/CMakeLists.txt

@@ -28,18 +28,8 @@ install(CODE [[
         if(NOT _pip_check EQUAL 0)
             message(WARNING "Skipping Python package install: pip not available for ${Python3_EXECUTABLE}.")
         else()
-            set(_pip_install_args
-                install
-                --no-deps
-                --no-build-isolation
-                --break-system-packages
-                --prefix ${CMAKE_INSTALL_PREFIX})
-            if(DEFINED ENV{DESTDIR} AND NOT "$ENV{DESTDIR}" STREQUAL "")
-                list(APPEND _pip_install_args --root "$ENV{DESTDIR}")
-            endif()
-            list(APPEND _pip_install_args .)
             execute_process(
-                COMMAND ${Python3_EXECUTABLE} -m pip ${_pip_install_args}
+                COMMAND ${Python3_EXECUTABLE} -m pip install --no-deps --break-system-packages .
                 WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/python
                 OUTPUT_VARIABLE _pip_output
                 ERROR_VARIABLE _pip_error

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

@@ -1205,14 +1205,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 +1224,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");
@@ -1274,6 +1267,21 @@ DhtProxyServer::sendPushNotification(
                 Json::Value androidConfig(Json::objectValue);
                 androidConfig["priority"] = priority;
                 androidConfig["ttl"] = "86400s"; // time to live = 24 hours
+                // Make routine sync notifications (connection churn, value
+                // expirations) collapsible: FCM keeps only the last pending
+                // message per collapse key when the device is unreachable
+                // (dozing/offline), while non-collapsible messages pile up in
+                // a 100-message queue. When that queue overflows, FCM silently
+                // drops messages - including the high-priority ones that
+                // announce incoming calls. A single constant key is used
+                // because FCM only honors 4 distinct collapse keys per device
+                // and the Android client reacts to any of these notifications
+                // the same way: wake up and resynchronize against the proxy.
+                // High-priority notifications (calls, messages, invites - see
+                // Value::priority set by the client) and resubscribe requests
+                // remain non-collapsible so none of them can be overwritten.
+                if (!highPriority && !isResubscribe)
+                    androidConfig["collapse_key"] = "sync";
                 notification["android"] = std::move(androidConfig);
             } else {
                 notification["priority"] = priority;

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);
 }