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>

.github/workflows/ccpp.yml

@@ -7,7 +7,7 @@ env:
   UBUNTU_FULL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
   UBUNTU_MESON_DEPS: "ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
   UBUNTU_MINIMAL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libfmt-dev libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-dev"
-  DEBIAN13_DEPS: "git libtool pkg-config wget cmake libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
+  DEBIAN12_DEPS: "git libtool pkg-config wget cmake libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
 
 on: [push, pull_request]
 
@@ -16,7 +16,7 @@ jobs:
     name: Ubuntu/GCC Meson build
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       - name: Install Dependencies
         run: |
@@ -40,7 +40,7 @@ jobs:
       matrix:
         cxx_standard: [17, 20, 23]
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       - name: Install Minimal Dependencies
         run: |
@@ -65,12 +65,12 @@ jobs:
       - name: Run Unit tests
         run: ctest --output-on-failure --test-dir build
 
-  build-debian13:
-    name: Debian 13/GCC build
+  build-debian12:
+    name: Debian 12/GCC build
     runs-on: ubuntu-latest
 
     container:
-      image: debian:13
+      image: debian:12
       volumes:
         - .:/workspace
       options: --workdir /workspace
@@ -78,12 +78,12 @@ jobs:
         DEBIAN_FRONTEND: noninteractive
 
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       - name: Install Dependencies
         run: |
           apt-get update
-          apt-get install -y $DEBIAN13_DEPS
+          apt-get install -y $DEBIAN12_DEPS
 
       - name: Build llhttp dependency
         run: |
@@ -131,7 +131,7 @@ jobs:
       matrix:
         cxx_standard: [17, 20, 23]
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       - name: Install Homebrew Dependencies
         run: brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl cppunit

.github/workflows/clang-analyzer.yml

@@ -6,7 +6,7 @@ jobs:
     name: Clang Static Analysis
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
       - name: deps
         run: |
           sudo apt update && \

.github/workflows/codeql.yml

@@ -27,7 +27,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v7
+      uses: actions/checkout@v6
 
     - name: Initialize CodeQL
       uses: github/codeql-action/init@v4

.github/workflows/python-wheels.yml

@@ -16,14 +16,14 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ubuntu-latest, ubuntu-24.04-arm, macos-15]
+        os: [ubuntu-latest, ubuntu-24.04-arm, macos-latest]
 
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       # Run locally with: pipx run cibuildwheel --platform linux python
       - name: Build wheels
-        uses: pypa/cibuildwheel@v4.1.1
+        uses: pypa/cibuildwheel@v3.4.1
         with:
           package-dir: python
 
@@ -36,7 +36,7 @@ jobs:
     name: Build wheels on Windows
     runs-on: windows-latest
     env:
-      VCPKG_VERSION: "2026.06.01"
+      VCPKG_VERSION: "2025.07.25"
       VCPKG_INSTALLATION_ROOT: "C:\\vcpkg"
       VCPKG_DISABLE_METRICS: "1"
       VCPKG_DEFAULT_TRIPLET: x64-windows-release
@@ -44,11 +44,11 @@ jobs:
       VCPKG_DEFAULT_BINARY_CACHE: ${{ github.workspace }}\vcpkg_binary_cache
 
     steps:
-      - uses: actions/checkout@v7
+      - uses: actions/checkout@v6
 
       - name: Cache vcpkg installed packages
         id: cache-vcpkg
-        uses: actions/cache@v6
+        uses: actions/cache@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}

.github/workflows/release-package.yml

@@ -25,7 +25,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -66,7 +66,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -107,7 +107,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -148,7 +148,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -190,7 +190,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -231,7 +231,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -273,7 +273,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v4
@@ -309,7 +309,7 @@ jobs:
     name: Build Windows MSI Installer
     runs-on: windows-latest
     env:
-      VCPKG_VERSION: "2026.06.01"
+      VCPKG_VERSION: "2025.07.25"
       VCPKG_INSTALLATION_ROOT: "C:\\vcpkg"
       VCPKG_DISABLE_METRICS: "1"
       VCPKG_DEFAULT_TRIPLET: x64-windows
@@ -319,7 +319,7 @@ jobs:
       contents: read
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
         with:
           submodules: recursive
 
@@ -334,7 +334,7 @@ jobs:
 
       - name: Cache vcpkg installed packages
         id: cache-vcpkg
-        uses: actions/cache@v6
+        uses: actions/cache@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -413,7 +413,7 @@ jobs:
     name: Build Windows dhtnode
     runs-on: windows-latest
     env:
-      VCPKG_VERSION: "2026.06.01"
+      VCPKG_VERSION: "2025.07.25"
       VCPKG_INSTALLATION_ROOT: "C:\\vcpkg"
       VCPKG_DISABLE_METRICS: "1"
       VCPKG_DEFAULT_TRIPLET: x64-windows-static-release
@@ -423,7 +423,7 @@ jobs:
       contents: read
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
         with:
           submodules: recursive
 
@@ -432,7 +432,7 @@ jobs:
 
       - name: Cache vcpkg static installed packages
         id: cache-vcpkg-static
-        uses: actions/cache@v6
+        uses: actions/cache@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -499,7 +499,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Download Windows MSI artifact
         uses: actions/download-artifact@v8

.github/workflows/windows-build.yml

@@ -9,7 +9,7 @@ on:
     branches: [ main, master ]
 
 env:
-  VCPKG_VERSION: "2026.06.01"
+  VCPKG_VERSION: "2026.01.16"
   VCPKG_INSTALLATION_ROOT: "C:\\vcpkg"
   VCPKG_DISABLE_METRICS: "1"
   VCPKG_INSTALLED_DIR: ${{ github.workspace }}\vcpkg_installed
@@ -27,7 +27,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up MSVC
         uses: microsoft/setup-msbuild@v3
@@ -41,7 +41,7 @@ jobs:
 
       - name: Cache vcpkg installed packages
         id: cache-vcpkg
-        uses: actions/cache@v6
+        uses: actions/cache@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -75,7 +75,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up MSVC
         uses: microsoft/setup-msbuild@v3
@@ -89,7 +89,7 @@ jobs:
 
       - name: Cache vcpkg static installed packages
         id: cache-vcpkg-static
-        uses: actions/cache@v6
+        uses: actions/cache@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -125,14 +125,14 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up MSVC
         uses: microsoft/setup-msbuild@v3
 
       - name: Restore vcpkg installed packages
         id: cache-vcpkg-restore
-        uses: actions/cache/restore@v6
+        uses: actions/cache/restore@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -202,14 +202,14 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Set up MSVC
         uses: microsoft/setup-msbuild@v3
 
       - name: Restore vcpkg static installed packages
         id: cache-vcpkg-static-restore
-        uses: actions/cache/restore@v6
+        uses: actions/cache/restore@v5
         with:
           path: |
             ${{ env.VCPKG_INSTALLED_DIR }}
@@ -275,7 +275,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v7
+        uses: actions/checkout@v6
 
       - name: Setup MinGW
         uses: msys2/setup-msys2@v2

CMakeLists.txt

@@ -1,4 +1,4 @@
-cmake_minimum_required (VERSION 3.21...4.4)
+cmake_minimum_required (VERSION 3.21...3.31)
 if(POLICY CMP0091)
     cmake_policy(SET CMP0091 NEW)
 endif()
@@ -10,7 +10,7 @@ if(POLICY CMP0167)
 endif()
 
 set (opendht_VERSION_MAJOR 4)
-set (opendht_VERSION_MINOR 3.0)
+set (opendht_VERSION_MINOR 0.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

include/opendht/crypto.h

@@ -204,13 +204,7 @@ struct OPENDHT_PUBLIC PrivateKey
      * @param algo : the public key algorithm to use, GNUTLS_PK_RSA or GNUTLS_PK_RSA_OAEP
      */
     static PrivateKey generate(unsigned key_length = 4096, gnutls_pk_algorithm_t algo = GNUTLS_PK_RSA);
-
-    /**
-     * Generate a new EC key pair
-     * @param curve : the elliptic curve to use. If GNUTLS_ECC_CURVE_INVALID is provided, the default curve will be used
-     * (currently GNUTLS_ECC_CURVE_SECP384R1).
-     */
-    static PrivateKey generateEC(gnutls_ecc_curve_t curve = GNUTLS_ECC_CURVE_INVALID);
+    static PrivateKey generateEC();
 
     gnutls_privkey_t key {};
     gnutls_x509_privkey_t x509_key {};
@@ -273,15 +267,9 @@ public:
     /**
      * Sign this revocation list using provided key and certificate.
      * Validity_period sets the duration until next update (default to no next update).
-     * If `number` is non-zero, it is stored verbatim (big-endian) as the CRL Number
-     * extension. Otherwise the number is initialized randomly on the first signature
-     * and incremented on subsequent ones.
      */
-    void sign(const PrivateKey&, const Certificate&, duration validity_period = {}, uint64_t number = 0);
-    void sign(const Identity& id, duration validity_period = {}, uint64_t number = 0)
-    {
-        sign(*id.first, *id.second, validity_period, number);
-    }
+    void sign(const PrivateKey&, const Certificate&, duration validity_period = {});
+    void sign(const Identity& id) { sign(*id.first, *id.second); }
 
     bool isSignedBy(const Certificate& issuer) const;
 

include/opendht/dht_proxy_server.h

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

include/opendht/network_engine.h

@@ -489,12 +489,8 @@ private:
     static constexpr size_t MTU {1280};
     static constexpr size_t MAX_PACKET_VALUE_SIZE {600};
     static constexpr size_t MAX_MESSAGE_VALUE_SIZE {56 * 1024};
-    static constexpr size_t RX_MAX_EARLY_BYTES {MAX_MESSAGE_VALUE_SIZE + MAX_VALUE_SIZE + 32};
-    static constexpr size_t RX_MAX_EARLY_PARTS {RX_MAX_EARLY_BYTES / MTU + 2};
 
     void process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from);
-    bool isExpectedReply(const ParsedMessage&, const SockAddr& from);
-    bool tryCompleteMessage(PartialMessage&, const SockAddr& from);
 
     bool rateLimit(const SockAddr& addr);
 
@@ -531,8 +527,7 @@ private:
     {
         return packValueHeader(buf, values.begin(), values.end());
     }
-    using PartialMessageKey = std::pair<SockAddr, Tid>;
-    void maintainRxBuffer(const PartialMessageKey& key);
+    void maintainRxBuffer(Tid tid);
 
     /*************
      *  Answers  *
@@ -578,7 +573,7 @@ private:
 
     // requests handling
     std::map<Tid, Sp<Request>> requests {};
-    std::map<PartialMessageKey, PartialMessage> partial_messages;
+    std::map<Tid, PartialMessage> partial_messages;
 
     MessageStats in_stats {}, out_stats {};
     std::set<SockAddr> blacklist {};

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 {8};   // eight hours
+constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
 
 using ListenToken = uint64_t;
 

include/opendht/thread_pool.h

@@ -10,9 +10,7 @@
 #include <future>
 #include <functional>
 
-#if __cplusplus < 202002L
-#include <ciso646>
-#endif
+#include <ciso646> // fix windows compiler bug
 
 namespace dht {
 

include/opendht/utils.h

@@ -85,14 +85,6 @@ using system_clock = std::chrono::system_clock;
 using time_point = clock::time_point;
 using duration = clock::duration;
 
-OPENDHT_PUBLIC time_point from_system_time(system_clock::time_point t,
-                                           time_point steady_now,
-                                           system_clock::time_point system_now);
-OPENDHT_PUBLIC time_point from_system_time(system_clock::time_point t);
-OPENDHT_PUBLIC system_clock::time_point to_system_time(time_point t,
-                                                       time_point steady_now,
-                                                       system_clock::time_point system_now);
-OPENDHT_PUBLIC system_clock::time_point to_system_time(time_point t);
 OPENDHT_PUBLIC time_point from_time_t(std::time_t t);
 OPENDHT_PUBLIC std::time_t to_time_t(time_point t);
 

meson.build

@@ -5,7 +5,7 @@ project(
     'cython',
     license: 'MIT',
     license_files: 'LICENSE',
-    version: '4.3.0',
+    version: '4.0.0',
     meson_version: '>=1.1.0',
     default_options: ['cpp_std=c++20', 'warning_level=3'],
 )
@@ -15,23 +15,10 @@ nettle = dependency('nettle')
 msgpack = dependency('msgpack-cxx', required: false)
 argon2 = dependency('libargon2')
 fmt = dependency('fmt')
-
-proxy_enabled = get_option('proxy_client').enabled() or get_option('proxy_server').enabled()
-
-openssl = dependency('openssl', required: proxy_enabled)
-jsoncpp = dependency('jsoncpp', required: proxy_enabled)
-llhttp = dependency('llhttp', 'libllhttp', required: proxy_enabled)
-simdutf = dependency('simdutf', required: proxy_enabled)
-
-# restinio is a header-only library that may not ship a pkg-config file
-restinio_dep = dependency('restinio', required: false)
-if not restinio_dep.found()
-    cpp = meson.get_compiler('cpp')
-    if cpp.has_header('restinio/all.hpp', required: proxy_enabled)
-        restinio_dep = declare_dependency()
-    endif
-endif
-
+openssl = dependency('openssl', required: get_option('proxy_client'))
+jsoncpp = dependency('jsoncpp', required: get_option('proxy_client'))
+llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
+simdutf = dependency('simdutf', required: get_option('proxy_client'))
 io_uring = host_machine.system() == 'linux' ? dependency('liburing', required: false) : disabler()
 
 deps = [
@@ -46,10 +33,6 @@ deps = [
     llhttp,
 ]
 
-if restinio_dep.found()
-    deps += restinio_dep
-endif
-
 if host_machine.system() == 'linux' and io_uring.found()
     deps += io_uring
 endif
@@ -96,7 +79,7 @@ if get_option('indexation').enabled()
     opendht_src += 'src/indexation/pht.cpp'
     add_project_arguments('-DOPENDHT_INDEXATION', language: 'cpp')
 endif
-if jsoncpp.found() and simdutf.found()
+if jsoncpp.found()
     opendht_src += ['src/base64.cpp']
     add_project_arguments('-DOPENDHT_JSONCPP', language: 'cpp')
     conf_data.set('jsoncpp_lib', ', jsoncpp')

src/crypto.cpp

@@ -571,12 +571,6 @@ PublicKey::toString() const
     }
     if (err != GNUTLS_E_SUCCESS)
         throw CryptoException(std::string("Could not print public key: ") + gnutls_strerror(err));
-    // GnuTLS reports the PEM size including the terminating null byte, which
-    // does not belong to the text and would end up embedded in whatever the
-    // string is copied into, such as the "owner" field of a serialized value.
-    ret.resize(sz);
-    if (not ret.empty() and ret.back() == '\0')
-        ret.pop_back();
     return ret;
 }
 
@@ -1294,16 +1288,14 @@ PrivateKey::generate(unsigned key_length, gnutls_pk_algorithm_t algo)
 }
 
 PrivateKey
-PrivateKey::generateEC(gnutls_ecc_curve_t curve)
+PrivateKey::generateEC()
 {
     gnutls_x509_privkey_t key;
     if (gnutls_x509_privkey_init(&key) != GNUTLS_E_SUCCESS)
         throw CryptoException("Unable to initialize private key.");
     int err = gnutls_x509_privkey_generate(key,
                                            GNUTLS_PK_EC,
-                                           curve == GNUTLS_ECC_CURVE_INVALID
-                                               ? gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA)
-                                               : GNUTLS_CURVE_TO_BITS(curve),
+                                           gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA),
                                            0);
     if (err != GNUTLS_E_SUCCESS) {
         gnutls_x509_privkey_deinit(key);
@@ -1404,11 +1396,8 @@ setRandomSerial(gnutls_x509_crt_t cert)
 {
     std::random_device rdev;
     std::uniform_int_distribution<int64_t> dist {1};
-    const auto cert_serial = dist(rdev);
-    std::array<uint8_t, sizeof(cert_serial)> packed_serial;
-    for (size_t i = 0; i < packed_serial.size(); i++)
-        packed_serial[i] = cert_serial >> (8 * (packed_serial.size() - 1 - i));
-    gnutls_x509_crt_set_serial(cert, packed_serial.data(), packed_serial.size());
+    int64_t cert_serial = dist(rdev);
+    gnutls_x509_crt_set_serial(cert, &cert_serial, sizeof(cert_serial));
 }
 
 Certificate
@@ -1884,8 +1873,34 @@ RevocationList::getUpdateTime() const
     return std::chrono::system_clock::from_time_t(t);
 }
 
+enum class Endian : uint32_t { LITTLE = 0, BIG = 1 };
+
+template<typename T>
+T
+endian(T w, Endian endian = Endian::BIG)
+{
+    // this gets optimized out into if (endian == host_endian) return w;
+    union {
+        uint64_t quad;
+        uint32_t islittle;
+    } t;
+    t.quad = 1;
+    if (t.islittle ^ (uint32_t) endian)
+        return w;
+    T r = 0;
+
+    // decent compilers will unroll this (gcc)
+    // or even convert straight into single bswap (clang)
+    for (size_t i = 0; i < sizeof(r); i++) {
+        r <<= 8;
+        r |= w & 0xff;
+        w >>= 8;
+    }
+    return r;
+}
+
 void
-RevocationList::sign(const PrivateKey& key, const Certificate& ca, duration validity, uint64_t crlNumber)
+RevocationList::sign(const PrivateKey& key, const Certificate& ca, duration validity)
 {
     if (auto err = gnutls_x509_crl_set_version(crl, 2))
         throw CryptoException(std::string("Unable to set CRL version: ") + gnutls_strerror(err));
@@ -1895,28 +1910,19 @@ RevocationList::sign(const PrivateKey& key, const Certificate& ca, duration vali
         throw CryptoException(std::string("Unable to set CRL update time: ") + gnutls_strerror(err));
     if (auto err = gnutls_x509_crl_set_next_update(crl, std::chrono::system_clock::to_time_t(next_update)))
         throw CryptoException(std::string("Unable to set CRL next update time: ") + gnutls_strerror(err));
-    uint64_t number {crlNumber};
+    uint64_t number {0};
+    size_t number_sz {sizeof(number)};
+    unsigned critical {0};
+    gnutls_x509_crl_get_number(crl, &number, &number_sz, &critical);
     if (number == 0) {
-        // The CRL Number extension holds a DER integer of variable length, so it must be
-        // read as a big-endian byte string rather than straight into an uint64_t.
-        auto current = getNumber();
-        for (uint8_t byte : current)
-            number = (number << 8) | byte;
-        if (number == 0) {
-            // Initialize to a random number.
-            std::random_device rdev;
-            do {
-                std::generate_n((uint8_t*) &number, sizeof(number), std::bind(rand_byte, std::ref(rdev)));
-            } while (number == 0);
-        } else
-            number++;
-    }
-    // Store the number in network byte order.
-    uint8_t packedNumber[sizeof(number)];
-    for (size_t i = 0; i < sizeof(packedNumber); i++)
-        packedNumber[i] = (uint8_t) ((number >> (8 * (sizeof(packedNumber) - 1 - i))) & 0xff);
-    if (auto err = gnutls_x509_crl_set_number(crl, packedNumber, sizeof(packedNumber)))
-        throw CryptoException(std::string("Unable to set CRL number: ") + gnutls_strerror(err));
+        // initialize to a random number
+        number_sz = sizeof(number);
+        std::random_device rdev;
+        std::generate_n((uint8_t*) &number, sizeof(number), std::bind(rand_byte, std::ref(rdev)));
+    } else
+        number = endian(endian(number) + 1);
+    if (auto err = gnutls_x509_crl_set_number(crl, &number, sizeof(number)))
+        throw CryptoException(std::string("Unable to set CRL update time: ") + gnutls_strerror(err));
     if (auto err = gnutls_x509_crl_sign2(crl, ca.cert, key.x509_key, GNUTLS_DIG_SHA512, 0))
         throw CryptoException(std::string("Unable to sign certificate revocation list: ") + gnutls_strerror(err));
     // to be able to actually use the CRL we need to serialize/deserialize it

src/dht.cpp

@@ -2270,16 +2270,8 @@ Dht::onDisconnected()
 void
 Dht::bootstrap()
 {
-    if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected) {
-        // The job that led here has already run, and the scheduler does not
-        // clear the handle to a job it has executed. onDisconnected() reads a
-        // set handle as "a bootstrap is already scheduled" and does nothing,
-        // so leaving it set here means the node never tries to rejoin the
-        // network once it has lost it. Clearing it is what says that no
-        // bootstrap is pending any more.
-        bootstrapJob.reset();
+    if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected)
         return;
-    }
     if (logger_)
         logger_->debug("[{}] Bootstraping", myid.to_view());
     bootstrap_pending = false;
@@ -2363,9 +2355,6 @@ Dht::confirmNodes()
 std::vector<ValuesExport>
 Dht::exportValues() const
 {
-    const auto& now = scheduler.time();
-    const auto system_now = system_clock::now();
-
     std::vector<ValuesExport> e {};
     e.reserve(store.size());
     for (const auto& h : store) {
@@ -2378,7 +2367,7 @@ Dht::exportValues() const
         pk.pack_array(vals.size());
         for (const auto& v : vals) {
             pk.pack_array(4);
-            pk.pack(to_system_time(v.created, now, system_now));
+            pk.pack(v.created.time_since_epoch().count());
             v.data->msgpack_pack(pk);
             if (const auto& store_addr = v.store_bucket ? v.store_bucket->getAddr() : SockAddr {}) {
                 pk.pack_bin(store_addr.getLength());
@@ -2386,7 +2375,7 @@ Dht::exportValues() const
             } else {
                 pk.pack_bin(0);
             }
-            pk.pack(to_system_time(v.expiration, now, system_now));
+            pk.pack(v.expiration.time_since_epoch().count());
         }
         ve.second = {buffer.data(), buffer.data() + buffer.size()};
         e.push_back(std::move(ve));
@@ -2398,7 +2387,6 @@ void
 Dht::importValues(const std::vector<ValuesExport>& import)
 {
     const auto& now = scheduler.time();
-    const auto system_now = system_clock::now();
 
     unsigned imported = 0, ignored = 0;
     size_t imported_size = 0;
@@ -2423,15 +2411,8 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                 Value tmp_val;
                 SockAddr store_addr;
                 try {
-                    const bool uses_system_time = valel.via.array.ptr[0].type == msgpack::type::EXT;
-                    if (uses_system_time) {
-                        val_time = from_system_time(valel.via.array.ptr[0].as<system_clock::time_point>(),
-                                                    now,
-                                                    system_now);
-                    } else {
-                        val_time = time_point {
-                            time_point::duration {valel.via.array.ptr[0].as<time_point::duration::rep>()}};
-                    }
+                    val_time = time_point {
+                        time_point::duration {valel.via.array.ptr[0].as<time_point::duration::rep>()}};
                     tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                     if (valel.via.array.size >= 3) {
                         auto addr_blob = valel.via.array.ptr[2].as<Blob>();
@@ -2441,14 +2422,8 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                         }
                     }
                     if (valel.via.array.size >= 4) {
-                        if (uses_system_time) {
-                            expiration = from_system_time(valel.via.array.ptr[3].as<system_clock::time_point>(),
-                                                          now,
-                                                          system_now);
-                        } else {
-                            expiration = time_point {
-                                time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
-                        }
+                        expiration = time_point {
+                            time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
                     }
                 } catch (const std::exception&) {
                     if (logger_)

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::system_timer> refreshSubscriberTimer;
+    std::unique_ptr<asio::steady_timer> refreshSubscriberTimer;
 #endif
 };
 
 struct PermanentPut
 {
-    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::system_timer>&& j, const Sp<std::atomic_bool>& o)
+    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::steady_timer>&& j, const Sp<std::atomic_bool>& o)
         : value(v)
         , refreshPutTimer(std::move(j))
         , ok(o)
     {}
 
     Sp<Value> value;
-    std::unique_ptr<asio::system_timer> refreshPutTimer;
+    std::unique_ptr<asio::steady_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::system_timer>(httpContext_,
+            auto refreshPutTimer = std::make_unique<asio::steady_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::system_timer>(httpContext_,
+                                auto refreshPutTimer = std::make_unique<asio::steady_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::system_timer>(httpContext_);
-                l->second.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
+                    l->second.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
+                l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_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::system_timer>(httpContext_);
+                put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
             }
-            put.second.refreshPutTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
+            put.second.refreshPutTimer->expires_at(std::chrono::steady_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,7 +1210,8 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         // Connectivity changed, refresh all subscribe
         for (auto& search : searches_)
             for (auto& listener : search.second.listeners)
-                resubscribe(search.first, listener.first, listener.second);
+                if (!listener.second.opstate->ok)
+                    resubscribe(search.first, listener.first, listener.second);
         return;
     }
     if (logger_)
@@ -1274,10 +1275,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::system_timer>(httpContext_,
-                                                                               std::chrono::system_clock::now());
+                    put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+                                                                               std::chrono::steady_clock::now());
                 else
-                    put.refreshPutTimer->expires_at(std::chrono::system_clock::now());
+                    put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
                 put.refreshPutTimer->async_wait(
                     std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
                 ret = PushNotificationResult::PutRefresh;
@@ -1289,12 +1290,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) {
@@ -1377,9 +1378,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::system_timer>(httpContext_);
+        listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
     }
-    listener.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+    listener.refreshSubscriberTimer->expires_at(std::chrono::steady_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>(std::chrono::system_clock::to_time_t(serverStartTime));
-    result["lastUpdated"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(lastUpdated));
+    result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
+    result["lastUpdated"] = static_cast<Json::LargestInt>(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 = std::chrono::system_clock::from_time_t(exp->as<time_t>());
+        expiration = 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 = std::chrono::system_clock::from_time_t(exp->as<time_t>());
+        expiration = 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::system_timer>(io_context(),
+                        pput.second.expireTimer = std::make_unique<asio::steady_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::system_timer>(io_context(),
+                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_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 {}, dht::time_point::max(), true);
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, 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::system_timer>(io_context(),
+                            listener.expireNotifyTimer = std::make_unique<asio::steady_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::system_timer>(io_context(),
+                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
                                                                                         listener.expiration);
                             listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
                                                                        this,
@@ -907,7 +907,6 @@ 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);
@@ -952,14 +951,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         auto& listener = *listIt;
 
         // Expiration
-        auto timeout = clock::now() + proxy::OP_TIMEOUT;
+        auto timeout = std::chrono::steady_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::system_timer>(io_context(), timeout - proxy::OP_MARGIN);
+            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
         auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() {
             Json::Value json;
             json["timeout"] = h;
@@ -976,7 +975,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
                                                          listener.type,
                                                          listener.topic));
         if (!listener.expireTimer)
-            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
+            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
         else
             listener.expireTimer->expires_at(timeout);
         listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
@@ -1135,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,
@@ -1177,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;
 }
@@ -1406,7 +1427,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                     topic = pVal["topic"].asString();
                 }
                 std::lock_guard lock(lockSearchPuts_);
-                auto timeout = clock::now() + proxy::OP_TIMEOUT;
+                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
                     for (auto& pp : sPuts.puts) {
@@ -1440,7 +1461,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::system_timer>(io_context(), timeout);
+                    pput.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                     if (not pushToken.empty()) {
                         pput.pushToken = pushToken;
@@ -1479,7 +1500,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         return json;
                     };
                     if (!pput.expireNotifyTimer)
-                        pput.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
+                        pput.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
                                                                                       timeout - proxy::OP_MARGIN);
                     else
                         pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -1508,7 +1529,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         response.done();
                     }
                 },
-                dht::time_point::max(),
+                time_point::max(),
                 permanent);
             return restinio::request_handling_status_t::accepted;
         } else {

src/dhtrunner.cpp

@@ -99,11 +99,6 @@ DhtRunner::run(const Config& config, Context&& context)
         auto state_path = config.dht_config.node_config.persist_path;
         if (not state_path.empty())
             state_path += "_port.txt";
-        // Ports loaded from the persisted state are a best-effort hint to keep a
-        // stable identity across restarts; remember which ones we set so that we
-        // can fall back to random ports if they are already in use (e.g. a fast
-        // restart while the previous instance still holds them).
-        bool triedSavedPort = false;
         if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
             std::ifstream inConfig(state_path);
             if (inConfig.is_open()) {
@@ -115,7 +110,6 @@ DhtRunner::run(const Config& config, Context&& context)
                                               fmt::ptr(this),
                                               port);
                         local4.setPort(port);
-                        triedSavedPort = true;
                     }
                 }
                 if (inConfig >> port) {
@@ -125,7 +119,6 @@ DhtRunner::run(const Config& config, Context&& context)
                                               fmt::ptr(this),
                                               port);
                         local6.setPort(port);
-                        triedSavedPort = true;
                     }
                 }
             }
@@ -143,20 +136,7 @@ DhtRunner::run(const Config& config, Context&& context)
 
         if (config.proxy_server.empty()) {
             if (not context.sock) {
-                try {
-                    context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-                } catch (const DhtException&) {
-                    if (not triedSavedPort)
-                        throw;
-                    // The saved ports are unavailable (typically still held by a
-                    // previous instance): fall back to OS-assigned ports.
-                    if (context.logger)
-                        context.logger->w("[runner %p] Saved ports unavailable, falling back to random ports",
-                                          fmt::ptr(this));
-                    local4.setPort(0);
-                    local6.setPort(0);
-                    context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-                }
+                context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
             }
             context.sock->setOnReceive([&](net::PacketList&& pkts) {
                 net::PacketList ret;
@@ -554,14 +534,8 @@ DhtRunner::registerInsecureType(const ValueType& type)
 void
 DhtRunner::importValues(const std::vector<ValuesExport>& values)
 {
-    {
-        std::lock_guard lck(dht_mtx);
-        dht_->importValues(values);
-    }
-    // Recompute the runner wakeup after import schedules value expirations.
-    std::lock_guard lck(storage_mtx);
-    pending_ops_prio.emplace([](SecureDht&) {});
-    cv.notify_all();
+    std::lock_guard lck(dht_mtx);
+    dht_->importValues(values);
 }
 
 unsigned

src/http.cpp

@@ -302,13 +302,13 @@ static X509*
 issuer_from_chain(STACK_OF(X509) * fullchain)
 {
     X509 *cert, *issuer;
-    const X509_NAME* issuer_name;
+    X509_NAME* issuer_name;
 
     cert = cert_from_chain(fullchain);
     if ((issuer_name = X509_get_issuer_name(cert)) == nullptr)
         return nullptr;
 
-    issuer = X509_find_by_subject(fullchain, const_cast<X509_NAME*>(issuer_name));
+    issuer = X509_find_by_subject(fullchain, issuer_name);
     return issuer;
 }
 
@@ -917,16 +917,8 @@ Resolver::add_callback(ResolverCb cb, sa_family_t family)
                                                else
                                                    cb(ec, filter(endpoints, family));
                                            });
-    else {
-        // Always dispatch the callback on the io_context thread. Invoking it
-        // synchronously would run the caller's continuation (e.g. the whole
-        // connect chain of Request::send) on the calling thread, mutating asio
-        // objects concurrently with the io thread running the io_context.
-        asio::post(resolver_.get_executor(),
-                   [cb = std::move(cb),
-                    ec = ec_,
-                    endpoints = family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family)] { cb(ec, endpoints); });
-    }
+    else
+        cb(ec_, family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family));
 }
 
 void

src/network_engine.cpp

@@ -40,44 +40,8 @@ struct NetworkEngine::PartialMessage
     time_point start;
     time_point last_part;
     std::unique_ptr<ParsedMessage> msg;
-    std::vector<std::unique_ptr<ParsedMessage>> early_parts;
-    size_t early_bytes {};
-    bool request_quota_consumed {};
 };
 
-bool
-NetworkEngine::isExpectedReply(const ParsedMessage& msg, const SockAddr& from)
-{
-    auto request = requests.find(msg.tid);
-    if (request != requests.end() && request->second->pending() && !request->second->node->id
-        && request->second->node->getAddr() == from)
-        return true;
-
-    auto node = cache.getNode(msg.id, from.getFamily());
-    if (!node || node->getAddr() != from)
-        return false;
-
-    auto pending = node->getRequest(msg.tid);
-    return (pending && pending->pending()) || node->getSocket(msg.tid);
-}
-
-bool
-NetworkEngine::tryCompleteMessage(PartialMessage& partial, const SockAddr& from)
-{
-    try {
-        if (!partial.msg->complete())
-            return false;
-        process(std::move(partial.msg), from);
-    } catch (const std::exception& e) {
-        if (logger_)
-            logger_->warn("Error while processing partial message: {}", e.what());
-    } catch (...) {
-        if (logger_)
-            logger_->warn("Unknown error while processing partial message");
-    }
-    return true;
-}
-
 std::vector<Blob>
 serializeValues(const std::vector<Sp<Value>>& st)
 {
@@ -502,58 +466,35 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
 
     // partial value data
     if (msg->type == MessageType::ValueData) {
-        if (msg->fragment_parts.empty()
-            || std::any_of(msg->fragment_parts.begin(), msg->fragment_parts.end(), [](const auto& fragment) {
-                   return fragment.second.second.empty();
-               })) {
+        auto pmsg_it = partial_messages.find(msg->tid);
+        if (pmsg_it == partial_messages.end()) {
+            if (logIncoming_)
+                if (logger_)
+                    logger_->debug("Unable to find partial message");
             rateLimit(from);
             return;
         }
-        PartialMessageKey key {from, msg->tid};
-        auto pmsg_it = partial_messages.find(key);
-        if (pmsg_it == partial_messages.end()) {
-            if (!rateLimit(from))
-                return;
-            auto inserted = partial_messages.emplace(key, PartialMessage {});
-            pmsg_it = inserted.first;
-            pmsg_it->second.from = from;
-            pmsg_it->second.start = now;
-            pmsg_it->second.last_part = now;
-            pmsg_it->second.request_quota_consumed = true;
-            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
-        }
-        auto& pmsg = pmsg_it->second;
-        if (!pmsg.msg) {
-            if (pmsg.early_parts.size() >= RX_MAX_EARLY_PARTS) {
-                partial_messages.erase(pmsg_it);
-                return;
-            }
-            size_t fragment_bytes {};
-            for (const auto& [index, fragment] : msg->fragment_parts) {
-                (void) index;
-                if (fragment.second.size() > RX_MAX_EARLY_BYTES - fragment_bytes) {
-                    partial_messages.erase(pmsg_it);
-                    return;
-                }
-                fragment_bytes += fragment.second.size();
-            }
-            if (fragment_bytes > RX_MAX_EARLY_BYTES - pmsg.early_bytes) {
-                partial_messages.erase(pmsg_it);
-                return;
-            }
-            pmsg.early_bytes += fragment_bytes;
-            pmsg.early_parts.emplace_back(std::move(msg));
-            pmsg.last_part = now;
-            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
+        if (!pmsg_it->second.from.equals(from)) {
+            if (logger_)
+                logger_->debug("Received partial message data from unexpected IP address");
+            rateLimit(from);
             return;
         }
         // append data block
-        if (pmsg.msg->append(*msg)) {
-            pmsg.last_part = now;
-            if (tryCompleteMessage(pmsg, from)) {
+        if (pmsg_it->second.msg->append(*msg)) {
+            pmsg_it->second.last_part = now;
+            // check data completion
+            if (pmsg_it->second.msg->complete()) {
+                try {
+                    // process the full message
+                    process(std::move(pmsg_it->second.msg), from);
+                } catch (const std::exception& e) {
+                    if (logger_)
+                        logger_->warn("Error while processing partial message: {}", e.what());
+                }
                 partial_messages.erase(pmsg_it);
             } else
-                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
+                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
         }
         return;
     }
@@ -564,32 +505,7 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         return;
     }
 
-    if (msg->value_parts.empty()) {
-        if (msg->type > MessageType::Reply) {
-            /* Rate limit requests. */
-            if (!rateLimit(from)) {
-                if (logger_)
-                    logger_->warn("Dropping request due to rate limiting");
-                return;
-            }
-        }
-        try {
-            process(std::move(msg), from);
-        } catch (const std::exception& e) {
-            if (logger_)
-                logger_->warn("Error while processing message: {}", e.what());
-        }
-        return;
-    }
-
-    PartialMessageKey partial_key {from, msg->tid};
-    if (msg->type == MessageType::Reply && !isExpectedReply(*msg, from))
-        return;
-
-    const auto partial = partial_messages.find(partial_key);
-    const bool request_quota_consumed = partial != partial_messages.end() && !partial->second.msg
-                                        && partial->second.request_quota_consumed;
-    if (msg->type > MessageType::Reply && !request_quota_consumed) {
+    if (msg->type > MessageType::Reply) {
         /* Rate limit requests. */
         if (!rateLimit(from)) {
             if (logger_)
@@ -598,28 +514,28 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         }
     }
 
-    // starting partial message session
-    auto inserted = partial_messages.emplace(partial_key, PartialMessage {});
-    auto& pmsg = inserted.first->second;
-    if (not pmsg.msg) {
-        if (inserted.second) {
+    if (msg->value_parts.empty()) {
+        try {
+            process(std::move(msg), from);
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->warn("Error while processing message: {}", e.what());
+            return;
+        }
+    } else {
+        // starting partial message session
+        auto k = msg->tid;
+        auto& pmsg = partial_messages[k];
+        if (not pmsg.msg) {
             pmsg.from = from;
+            pmsg.msg = std::move(msg);
             pmsg.start = now;
-            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, partial_key));
-        }
-        pmsg.msg = std::move(msg);
-        pmsg.last_part = now;
-        for (const auto& part : pmsg.early_parts)
-            pmsg.msg->append(*part);
-        pmsg.early_parts.clear();
-        pmsg.early_bytes = 0;
-        if (tryCompleteMessage(pmsg, from)) {
-            partial_messages.erase(inserted.first);
-        } else {
-            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, partial_key));
-        }
-    } else if (logger_)
-        logger_->error("Partial message from {} with given TID {} already exists", from.toString(), msg->tid);
+            pmsg.last_part = now;
+            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
+            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
+        } else if (logger_)
+            logger_->error("Partial message with given TID {} already exists", k);
+    }
 }
 
 void
@@ -1698,9 +1614,9 @@ NetworkEngine::sendError(const SockAddr& addr, Tid tid, uint16_t code, const std
 }
 
 void
-NetworkEngine::maintainRxBuffer(const PartialMessageKey& key)
+NetworkEngine::maintainRxBuffer(Tid tid)
 {
-    auto msg = partial_messages.find(key);
+    auto msg = partial_messages.find(tid);
     if (msg != partial_messages.end()) {
         const auto& now = scheduler.time();
         if (msg->second.start + RX_MAX_PACKET_TIME < now || msg->second.last_part + RX_TIMEOUT < now) {

src/op_cache.cpp

@@ -224,7 +224,7 @@ SearchCache::expire(const time_point& now, const std::function<void(size_t)>& on
     auto ret = nextExpiration_;
     for (auto it = ops.begin(); it != ops.end();) {
         auto expiration = it->second->getExpiration();
-        if (expiration <= now) {
+        if (expiration < now) {
             auto cache = std::move(it->second);
             it = ops.erase(it);
             onCancel(cache->searchToken);

src/parsed_message.h

@@ -274,21 +274,14 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
 
     if (!parsed.a && !parsed.r && !parsed.e && !parsed.u)
         throw msgpack::type_error();
+    auto& req = parsed.a ? *parsed.a : (parsed.r ? *parsed.r : (parsed.u ? *parsed.u : *parsed.e));
 
     if (parsed.e) {
-        if (parsed.e->type != msgpack::type::ARRAY or parsed.e->via.array.size < 1)
+        if (parsed.e->type != msgpack::type::ARRAY)
             throw msgpack::type_error();
         error_code = parsed.e->via.array.ptr[0].as<uint16_t>();
     }
 
-    /* An error-only message carries no request map. The error array must never
-       be reached through the map view of the msgpack object union, and a
-       request field that is not a map must be rejected rather than
-       reinterpreted. */
-    const msgpack::object* req = parsed.a ? parsed.a : (parsed.r ? parsed.r : parsed.u);
-    if (req and req->type != msgpack::type::MAP)
-        throw msgpack::type_error();
-
     struct ParsedReq
     {
         msgpack::object* values;
@@ -297,9 +290,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         msgpack::object* want;
     } parsedReq {};
 
-    const uint32_t req_size = req ? req->via.map.size : 0;
-    for (unsigned i = 0; i < req_size; i++) {
-        auto& o = req->via.map.ptr[i];
+    for (unsigned i = 0; i < req.via.map.size; i++) {
+        auto& o = req.via.map.ptr[i];
         if (o.key.type != msgpack::type::STR)
             continue;
         auto key = o.key.as<std::string_view>();
@@ -378,8 +370,6 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
     } else if (parsedReq.fields) {
         if (auto rfields = findMapValue(*parsedReq.fields, "f"sv)) {
             auto vfields = rfields->as<std::set<Value::Field>>();
-            if (vfields.empty())
-                throw msgpack::type_error();
             if (auto rvalues = findMapValue(*parsedReq.fields, "v"sv)) {
                 if (rvalues->type != msgpack::type::ARRAY)
                     throw msgpack::type_error();

src/thread_pool.cpp

@@ -6,10 +6,8 @@
 #include <atomic>
 #include <thread>
 #include <iostream>
-#include <cmath> // std::pow
-#if __cplusplus < 202002L
-#include <ciso646>
-#endif
+#include <ciso646> // fix windows compiler bug
+#include <cmath>   // std::pow
 
 namespace dht {
 

src/utils.cpp

@@ -238,67 +238,28 @@ operator==(const SockAddr& a, const SockAddr& b)
     return a.equals(b);
 }
 
-time_point
-from_system_time(system_clock::time_point t, time_point steady_now, system_clock::time_point system_now)
-{
-    if (t == system_clock::time_point::min())
-        return time_point::min();
-    if (t == system_clock::time_point::max())
-        return time_point::max();
-    // Saturate before the cast: the steady clock duration may hold a narrower range than the system one.
-    constexpr auto max_dt = std::chrono::duration_cast<system_clock::duration>(duration::max());
-    constexpr auto min_dt = std::chrono::duration_cast<system_clock::duration>(duration::min());
-    auto system_dt = t - system_now;
-    if (system_dt >= max_dt)
-        return time_point::max();
-    if (system_dt <= min_dt)
-        return time_point::min();
-    auto dt = std::chrono::duration_cast<duration>(system_dt);
-    if (dt > duration(0) and steady_now > time_point::max() - dt)
-        return time_point::max();
-    else if (dt < duration(0) and steady_now < time_point::min() - dt)
-        return time_point::min();
-    return steady_now + dt;
-}
-
-time_point
-from_system_time(system_clock::time_point t)
-{
-    return from_system_time(t, clock::now(), system_clock::now());
-}
-
-system_clock::time_point
-to_system_time(time_point t, time_point steady_now, system_clock::time_point system_now)
-{
-    if (t == time_point::min())
-        return system_clock::time_point::min();
-    if (t == time_point::max())
-        return system_clock::time_point::max();
-    // Cast first: mixing clock durations of different periods would overflow the bound computations below.
-    auto dt = std::chrono::duration_cast<system_clock::duration>(t - steady_now);
-    if (dt > system_clock::duration(0) and system_now >= system_clock::time_point::max() - dt)
-        return system_clock::time_point::max();
-    else if (dt < system_clock::duration(0) and system_now <= system_clock::time_point::min() - dt)
-        return system_clock::time_point::min();
-    return system_now + dt;
-}
-
-system_clock::time_point
-to_system_time(time_point t)
-{
-    return to_system_time(t, clock::now(), system_clock::now());
-}
-
 time_point
 from_time_t(std::time_t t)
 {
-    return from_system_time(system_clock::from_time_t(t));
+    auto dt = system_clock::from_time_t(t) - system_clock::now();
+    auto now = clock::now();
+    if (dt > system_clock::duration(0) and now > time_point::max() - dt)
+        return time_point::max();
+    else if (dt < system_clock::duration(0) and now < time_point::min() - dt)
+        return time_point::min();
+    return now + dt;
 }
 
 std::time_t
 to_time_t(time_point t)
 {
-    return system_clock::to_time_t(to_system_time(t));
+    auto dt = t - clock::now();
+    auto now = system_clock::now();
+    if (dt > duration(0) and now >= system_clock::time_point::max() - dt)
+        return system_clock::to_time_t(system_clock::time_point::max());
+    else if (dt < duration(0) and now <= system_clock::time_point::min() - dt)
+        return system_clock::to_time_t(system_clock::time_point::min());
+    return system_clock::to_time_t(now + std::chrono::duration_cast<system_clock::duration>(dt));
 }
 
 Blob

src/value.cpp

@@ -202,9 +202,6 @@ Value::Value(const Json::Value& json)
     const auto& jprio = json["prio"];
     if (jprio.isIntegral())
         priority = jprio.asUInt();
-    const auto& jpt = json[VALUE_KEY_PUSHTYPE];
-    if (jpt.isString())
-        pushType = jpt.asString();
 }
 
 Json::Value
@@ -231,8 +228,6 @@ Value::toJson() const
     }
     if (priority)
         val["prio"] = priority;
-    if (not pushType.empty())
-        val[VALUE_KEY_PUSHTYPE] = pushType;
     return val;
 }
 

tests/test_crypto.cpp

@@ -59,24 +59,6 @@ CryptoTester::testSignatureEncryption()
     }
 }
 
-void
-CryptoTester::testPublicKeyPemIsPlainText()
-{
-    auto key = dht::crypto::PrivateKey::generate();
-    auto pem = key.getPublicKey().toString();
-
-    // The exported key is PEM text: it must not carry the terminating null
-    // byte GnuTLS counts in the exported size, which would otherwise be
-    // embedded in anything the string is serialized into.
-    CPPUNIT_ASSERT(not pem.empty());
-    CPPUNIT_ASSERT_EQUAL(std::string::npos, pem.find('\0'));
-    CPPUNIT_ASSERT_EQUAL('\n', pem.back());
-
-    // The exported text still reads back as the same key.
-    dht::crypto::PublicKey reloaded(dht::Blob(pem.begin(), pem.end()));
-    CPPUNIT_ASSERT_EQUAL(key.getPublicKey().getId(), reloaded.getId());
-}
-
 void
 CryptoTester::testCertificateRevocation()
 {
@@ -113,159 +95,10 @@ CryptoTester::testCertificateRevocation()
     CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
 }
 
-namespace {
-
-/** Read a CRL Number extension blob (big-endian, variable length) as an integer. */
-uint64_t
-crlNumberValue(const dht::Blob& number)
-{
-    uint64_t value {0};
-    for (uint8_t byte : number)
-        value = (value << 8) | byte;
-    return value;
-}
-
-} // namespace
-
-void
-CryptoTester::testRevocationListNumber()
-{
-    auto ca = dht::crypto::generateIdentity("ca");
-    auto device = dht::crypto::generateIdentity("dev", ca);
-
-    dht::crypto::RevocationList crl;
-    CPPUNIT_ASSERT_EQUAL(uint64_t(0), crlNumberValue(crl.getNumber()));
-
-    crl.revoke(*device.second);
-    crl.sign(ca);
-    auto first = crl.getNumber();
-    // An unnumbered list gets a random number on its first signature.
-    CPPUNIT_ASSERT(not first.empty());
-    CPPUNIT_ASSERT(crlNumberValue(first) != 0);
-
-    // Re-signing keeps the same lineage and moves the number forward.
-    crl.sign(ca);
-    auto second = crl.getNumber();
-    CPPUNIT_ASSERT_EQUAL(crlNumberValue(first) + 1, crlNumberValue(second));
-
-    crl.sign(ca);
-    CPPUNIT_ASSERT_EQUAL(crlNumberValue(first) + 2, crlNumberValue(crl.getNumber()));
-
-    // The number is a DER integer, so it is stored with leading zeros stripped.
-    // Incrementing must still work for numbers shorter than 8 bytes.
-    for (uint64_t start : {uint64_t(1), uint64_t(0x7f), uint64_t(0x80), uint64_t(0xffff)}) {
-        dht::crypto::RevocationList small;
-        small.revoke(*device.second);
-        small.sign(ca, {}, start);
-        CPPUNIT_ASSERT_EQUAL(start, crlNumberValue(small.getNumber()));
-        small.sign(ca);
-        CPPUNIT_ASSERT_EQUAL(start + 1, crlNumberValue(small.getNumber()));
-        small.sign(ca);
-        CPPUNIT_ASSERT_EQUAL(start + 2, crlNumberValue(small.getNumber()));
-    }
-}
-
-void
-CryptoTester::testRevocationListExplicitNumber()
-{
-    auto ca = dht::crypto::generateIdentity("ca");
-    auto device = dht::crypto::generateIdentity("dev", ca);
-
-    // Values chosen to cover leading zero bytes, a full-width number and the largest
-    // value that still encodes as a positive 8-byte DER integer.
-    const std::vector<uint64_t> numbers {1,
-                                         42,
-                                         0xff,
-                                         0x100,
-                                         0x0000'0001'0000'0000ull,
-                                         0x0000'02a1'0000'0001ull,
-                                         0x7fff'ffff'ffff'ffffull};
-    for (auto number : numbers) {
-        dht::crypto::RevocationList crl;
-        crl.revoke(*device.second);
-        crl.sign(ca, {}, number);
-        CPPUNIT_ASSERT_EQUAL(number, crlNumberValue(crl.getNumber()));
-        CPPUNIT_ASSERT(crl.isSignedBy(*ca.second));
-        CPPUNIT_ASSERT(crl.isRevoked(*device.second));
-
-        // The number must survive a pack/unpack round-trip.
-        dht::crypto::RevocationList unpacked(crl.getPacked());
-        CPPUNIT_ASSERT_EQUAL(number, crlNumberValue(unpacked.getNumber()));
-    }
-
-    // An explicit number overrides the increment, in both directions.
-    dht::crypto::RevocationList crl;
-    crl.revoke(*device.second);
-    crl.sign(ca, {}, 1000);
-    CPPUNIT_ASSERT_EQUAL(uint64_t(1000), crlNumberValue(crl.getNumber()));
-    crl.sign(ca, {}, 5);
-    CPPUNIT_ASSERT_EQUAL(uint64_t(5), crlNumberValue(crl.getNumber()));
-
-    // Passing 0 falls back to the default behavior: increment the current number.
-    crl.sign(ca, {}, 0);
-    CPPUNIT_ASSERT_EQUAL(uint64_t(6), crlNumberValue(crl.getNumber()));
-}
-
-void
-CryptoTester::testRevocationListPartitionedNumber()
-{
-    // Two devices holding the same authority issue revocation lists independently.
-    // RFC 5280 5.2.3 requires CRL numbers to increase monotonically for a given CRL
-    // scope and issuer; partitioning the low bits by device keeps each sequence
-    // monotonic while making collisions between devices impossible.
-    constexpr unsigned TAG_BITS = 24;
-    constexpr uint64_t TAG_A = 0x0a0b0c;
-    constexpr uint64_t TAG_B = 0x0a0b0d;
-
-    auto ca = dht::crypto::generateIdentity("ca");
-    auto account = dht::crypto::generateIdentity("acc", ca, 4096, true);
-    auto deviceA = dht::crypto::generateIdentity("devA", account);
-    auto deviceB = dht::crypto::generateIdentity("devB", account);
-
-    uint64_t previous {0};
-    dht::crypto::RevocationList crlA;
-    for (unsigned i = 1; i <= 4; i++) {
-        crlA.revoke(*deviceA.second);
-        crlA.sign(account, {}, (uint64_t(i) << TAG_BITS) | TAG_A);
-        auto number = crlNumberValue(crlA.getNumber());
-        CPPUNIT_ASSERT_EQUAL(TAG_A, number & ((uint64_t(1) << TAG_BITS) - 1));
-        CPPUNIT_ASSERT(number > previous);
-        CPPUNIT_ASSERT_EQUAL(uint64_t(0), number >> 63);
-        previous = number;
-    }
-
-    // Both devices branch from the same counter but never mint the same number.
-    dht::crypto::RevocationList branchA;
-    branchA.revoke(*deviceA.second);
-    branchA.sign(account, {}, (uint64_t(9) << TAG_BITS) | TAG_A);
-    dht::crypto::RevocationList branchB;
-    branchB.revoke(*deviceB.second);
-    branchB.sign(account, {}, (uint64_t(9) << TAG_BITS) | TAG_B);
-    CPPUNIT_ASSERT(crlNumberValue(branchA.getNumber()) != crlNumberValue(branchB.getNumber()));
-
-    // Both are valid and revoke only their own target.
-    CPPUNIT_ASSERT(branchA.isSignedBy(*account.second));
-    CPPUNIT_ASSERT(branchB.isSignedBy(*account.second));
-    CPPUNIT_ASSERT(branchA.isRevoked(*deviceA.second));
-    CPPUNIT_ASSERT(not branchA.isRevoked(*deviceB.second));
-    CPPUNIT_ASSERT(branchB.isRevoked(*deviceB.second));
-    CPPUNIT_ASSERT(not branchB.isRevoked(*deviceA.second));
-
-    // A number minted by an older client is fully random, so bit 63 may be set. Masking
-    // it out and taking the counter part still yields a strictly greater number.
-    constexpr uint64_t legacy = 0xdead'beef'0123'4567ull;
-    dht::crypto::RevocationList legacyCrl;
-    legacyCrl.revoke(*deviceA.second);
-    legacyCrl.sign(account, {}, legacy);
-    auto counter = ((crlNumberValue(legacyCrl.getNumber()) & ~(uint64_t(1) << 63)) >> TAG_BITS) + 1;
-    legacyCrl.sign(account, {}, (counter << TAG_BITS) | TAG_A);
-    CPPUNIT_ASSERT(crlNumberValue(legacyCrl.getNumber()) > (legacy & ~(uint64_t(1) << 63)));
-    CPPUNIT_ASSERT_EQUAL(uint64_t(0), crlNumberValue(legacyCrl.getNumber()) >> 63);
-}
-
 void
 CryptoTester::testCertificateRequest()
-{    // Generate CA
+{
+    // Generate CA
     auto ca = dht::crypto::generateIdentity("Test CA");
 
     // Generate signed request
@@ -316,25 +149,6 @@ CryptoTester::testCertificateSerialNumber()
     };
     auto serial = dht::crypto::Certificate(cert_pem).getSerialNumber();
     CPPUNIT_ASSERT(std::equal(SERIAL.begin(), SERIAL.end(), serial.begin(), serial.end()));
-
-    auto key = dht::crypto::PrivateKey::generateEC();
-    for (unsigned i = 0; i < 32; i++) {
-        auto generated = dht::crypto::Certificate::generate(key, "generated", {});
-        auto generatedSerial = generated.getSerialNumber();
-        CPPUNIT_ASSERT(not generatedSerial.empty());
-        CPPUNIT_ASSERT_EQUAL(uint8_t {0}, uint8_t(generatedSerial.front() & 0x80));
-        CPPUNIT_ASSERT(std::any_of(generatedSerial.begin(), generatedSerial.end(), [](uint8_t byte) { return byte != 0; }));
-    }
-}
-
-void
-CryptoTester::testEcKeyCurve()
-{
-    auto key = dht::crypto::PrivateKey::generateEC(GNUTLS_ECC_CURVE_SECP256R1);
-    unsigned int bits = 0;
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(GNUTLS_PK_EC),
-                         gnutls_x509_privkey_get_pk_algorithm2(key.x509_key, &bits));
-    CPPUNIT_ASSERT_EQUAL(256u, bits);
 }
 
 void

tests/test_crypto.h

@@ -12,14 +12,9 @@ class CryptoTester : public CppUnit::TestFixture
 {
     CPPUNIT_TEST_SUITE(CryptoTester);
     CPPUNIT_TEST(testSignatureEncryption);
-    CPPUNIT_TEST(testPublicKeyPemIsPlainText);
     CPPUNIT_TEST(testCertificateRevocation);
-    CPPUNIT_TEST(testRevocationListNumber);
-    CPPUNIT_TEST(testRevocationListExplicitNumber);
-    CPPUNIT_TEST(testRevocationListPartitionedNumber);
     CPPUNIT_TEST(testCertificateRequest);
     CPPUNIT_TEST(testCertificateSerialNumber);
-    CPPUNIT_TEST(testEcKeyCurve);
     CPPUNIT_TEST(testOcsp);
     CPPUNIT_TEST(testAesEncryption);
     CPPUNIT_TEST(testAesEncryptionWithMultipleKeySizes);
@@ -41,26 +36,10 @@ public:
      * Test data signature, encryption and decryption
      */
     void testSignatureEncryption();
-    /**
-     * Test that an exported public key is text only, with no embedded null byte
-     */
-    void testPublicKeyPemIsPlainText();
     /**
      * Test certificate generation, validation and revocation
      */
     void testCertificateRevocation();
-    /**
-     * Test the default CRL Number behavior: random on first signature, incremented after
-     */
-    void testRevocationListNumber();
-    /**
-     * Test signing a revocation list with a caller-provided CRL Number
-     */
-    void testRevocationListExplicitNumber();
-    /**
-     * Test that issuers sharing an authority can partition the CRL Number space
-     */
-    void testRevocationListPartitionedNumber();
     /**
      * Test certificate requests
      */
@@ -69,10 +48,6 @@ public:
      * Test certificate serial number extraction
      */
     void testCertificateSerialNumber();
-    /**
-     * Test EC private key generation with an explicit curve
-     */
-    void testEcKeyCurve();
     /**
      * Test OCSP
      */

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_dhtrunner.cpp

@@ -9,8 +9,6 @@
 #include <future>
 #include <mutex>
 #include <condition_variable>
-#include <fstream>
-#include <cstdio>
 using namespace std::chrono_literals;
 using namespace std::literals;
 
@@ -210,42 +208,6 @@ DhtRunnerTester::testImportValuesPreservesRemoteQuota()
     restored.join();
 }
 
-void
-DhtRunnerTester::testExportValuesUsesSystemTime()
-{
-    auto key = dht::InfoHash::get("export-system-time");
-    const auto created = dht::clock::now();
-    msgpack::sbuffer importBuffer;
-    msgpack::packer<msgpack::sbuffer> pk(&importBuffer);
-    pk.pack_array(1);
-    pk.pack_array(2);
-    pk.pack(created.time_since_epoch().count());
-    dht::Value {std::string("persisted")}.msgpack_pack(pk);
-    node1.importValues({
-        dht::ValuesExport {key, {importBuffer.data(), importBuffer.data() + importBuffer.size()}}
-    });
-
-    auto exported = node1.exportValues();
-    CPPUNIT_ASSERT_EQUAL(size_t(1), exported.size());
-
-    msgpack::unpacked msg;
-    msgpack::unpack(msg, reinterpret_cast<const char*>(exported.front().second.data()), exported.front().second.size());
-    const auto values = msg.get();
-    CPPUNIT_ASSERT(values.type == msgpack::type::ARRAY);
-    CPPUNIT_ASSERT_EQUAL(uint32_t(1), values.via.array.size);
-    const auto& persistedValue = values.via.array.ptr[0];
-    CPPUNIT_ASSERT_EQUAL(uint32_t(4), persistedValue.via.array.size);
-    CPPUNIT_ASSERT(persistedValue.via.array.ptr[0].type == msgpack::type::EXT);
-    CPPUNIT_ASSERT(persistedValue.via.array.ptr[3].type == msgpack::type::EXT);
-
-    const auto persistedCreated = persistedValue.via.array.ptr[0].as<std::chrono::system_clock::time_point>();
-    const auto persistedExpiration = persistedValue.via.array.ptr[3].as<std::chrono::system_clock::time_point>();
-    const auto delta = std::chrono::system_clock::now() - persistedCreated;
-    CPPUNIT_ASSERT(delta >= std::chrono::system_clock::duration::zero());
-    CPPUNIT_ASSERT(delta < 5s);
-    CPPUNIT_ASSERT(persistedExpiration > persistedCreated);
-}
-
 void
 DhtRunnerTester::testImportValuesPreservesStoredExpiration()
 {
@@ -303,25 +265,14 @@ DhtRunnerTester::testImportValuesPreservesStoredExpiration()
 void
 DhtRunnerTester::testImportPermanentValueExpires()
 {
-    static const dht::ValueType SHORT_TYPE {201, "Short perm", 500ms};
-
-    dht::DhtRunner::Config config;
-    config.dht_config.node_config.max_peer_req_per_sec = -1;
-    config.dht_config.node_config.max_req_per_sec = -1;
-    config.dht_config.node_config.max_store_size = -1;
-    config.dht_config.node_config.max_store_keys = -1;
-    dht::DhtRunner restored;
-    restored.run(0, config);
-    restored.registerInsecureType(SHORT_TYPE);
-
-    // Ensure the runner is waiting on a previously computed scheduler deadline.
-    std::promise<void> loopCompleted;
-    auto loopCompletedFuture = loopCompleted.get_future();
-    restored.getNodeInfo([&loopCompleted](auto) { loopCompleted.set_value(); });
-    getFutureValue(std::move(loopCompletedFuture));
+    // Simulate importing a value that was originally stored as permanent
+    // (expiration = time_point::max()). After import the permanent flag is lost,
+    // so the value must expire within the type's normal expiration window.
+    static const dht::ValueType SHORT_TYPE {201, "Short perm", 2s};
 
     auto key = dht::InfoHash::get("import-permanent-expiration");
     auto now = dht::clock::now();
+
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_array(1);
@@ -329,33 +280,63 @@ DhtRunnerTester::testImportPermanentValueExpires()
     pk.pack(now.time_since_epoch().count());
     dht::Value {SHORT_TYPE.id, std::string("was-permanent")}.msgpack_pack(pk);
     pk.pack_bin(0);
+    // Permanent values are exported with time_point::max() expiration
     pk.pack(dht::clock::time_point::max().time_since_epoch().count());
 
+    dht::DhtRunner restored;
+    dht::DhtRunner::Config config;
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+    config.dht_config.node_config.max_store_size = -1;
+    config.dht_config.node_config.max_store_keys = -1;
+    restored.run(0, config);
+
     restored.importValues({
         dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
     });
 
-    auto containsValue = [&] {
-        for (const auto& [exportedKey, data] : restored.exportValues()) {
-            if (exportedKey != key)
-                continue;
+    // Value should be present right after import
+    auto exportedBefore = restored.exportValues();
+    CPPUNIT_ASSERT(!exportedBefore.empty());
+
+    // Wait longer than the default USER_DATA expiration (10 min) is not
+    // practical, but the type is unregistered so the default 10 min expiration
+    // applies.  Instead, register the short type so the cap uses 2 s.
+    // Re-import into a fresh node that knows the short type.
+    dht::DhtRunner restored2;
+    restored2.run(0, config);
+    restored2.registerInsecureType(SHORT_TYPE);
+    restored2.importValues({
+        dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
+    });
+
+    CPPUNIT_ASSERT(!restored2.exportValues().empty());
+
+    // Wait for the short expiration to pass
+    std::this_thread::sleep_for(3s);
+
+    auto restoredValues = restored2.exportValues();
+    bool found = false;
+    for (const auto& ve : restoredValues) {
+        if (ve.first == key) {
             msgpack::unpacked msg;
-            msgpack::unpack(msg, reinterpret_cast<const char*>(data.data()), data.size());
+            msgpack::unpack(msg, (const char*) ve.second.data(), ve.second.size());
             auto valarr = msg.get();
             CPPUNIT_ASSERT(valarr.type == msgpack::type::ARRAY);
-            return valarr.via.array.size != 0;
+            // The value should have expired
+            CPPUNIT_ASSERT_EQUAL(uint32_t(0), valarr.via.array.size);
+            found = true;
         }
-        return false;
-    };
-
-    CPPUNIT_ASSERT(containsValue());
-    auto timeout = dht::clock::now() + 2s;
-    while (containsValue() and dht::clock::now() < timeout)
-        std::this_thread::sleep_for(10ms);
-    CPPUNIT_ASSERT(!containsValue());
+    }
+    // Either the key was cleaned up entirely or it was found with 0 values
+    CPPUNIT_ASSERT(found || restoredValues.empty()
+                   || std::none_of(restoredValues.begin(), restoredValues.end(),
+                                   [&](const auto& ve) { return ve.first == key; }));
 
     restored.shutdown();
     restored.join();
+    restored2.shutdown();
+    restored2.join();
 }
 
 void
@@ -559,34 +540,6 @@ DhtRunnerTester::testShutdownCompletesWithPendingPut()
     clientNode.join();
 }
 
-void
-DhtRunnerTester::testSavedPortFallback()
-{
-    // A saved port that is already taken by another process (e.g. fast
-    // application restart while the old instance still holds it) must not
-    // prevent the node from starting: it should fall back to a random port.
-    dht::DhtRunner::Config config;
-    config.dht_config.node_config.persist_path = "/tmp/opendht_test_port_fallback";
-    std::string portFile = config.dht_config.node_config.persist_path + "_port.txt";
-
-    // node1 (from setUp) holds a port; save it as our persisted port.
-    auto takenPort = node1.getBoundPort();
-    {
-        std::ofstream out(portFile);
-        out << takenPort << std::endl << takenPort << std::endl;
-    }
-
-    dht::DhtRunner testNode;
-    CPPUNIT_ASSERT_NO_THROW(testNode.run(0, config));
-    CPPUNIT_ASSERT(testNode.getBoundPort());
-    CPPUNIT_ASSERT(testNode.getBoundPort() != takenPort);
-
-    testNode.shutdown();
-    testNode.join();
-    std::remove(portFile.c_str());
-    std::remove((config.dht_config.node_config.persist_path + "_port.txt").c_str());
-}
-
 void
 DhtRunnerTester::testListen()
 {

tests/test_dhtrunner.h

@@ -21,14 +21,12 @@ class DhtRunnerTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST(testIdOps);
     CPPUNIT_TEST(testImportValuesPreservesRemoteQuota);
-    CPPUNIT_TEST(testExportValuesUsesSystemTime);
     CPPUNIT_TEST(testImportValuesPreservesStoredExpiration);
     CPPUNIT_TEST(testImportPermanentValueExpires);
     CPPUNIT_TEST(testBootstrapSetsConnectingState);
     CPPUNIT_TEST(testBootstrapThenPutNoRace);
     CPPUNIT_TEST(testBootstrapMissingNodeThenPutFails);
     CPPUNIT_TEST(testShutdownCompletesWithPendingPut);
-    CPPUNIT_TEST(testSavedPortFallback);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -68,14 +66,12 @@ public:
      */
     void testIdOps();
     void testImportValuesPreservesRemoteQuota();
-    void testExportValuesUsesSystemTime();
     void testImportValuesPreservesStoredExpiration();
     void testImportPermanentValueExpires();
     void testBootstrapSetsConnectingState();
     void testBootstrapThenPutNoRace();
     void testBootstrapMissingNodeThenPutFails();
     void testShutdownCompletesWithPendingPut();
-    void testSavedPortFallback();
     /**
      * Test listen method with lot of datas
      */

tests/test_http.cpp

@@ -8,7 +8,6 @@
 
 #include <iostream>
 #include <string>
-#include <atomic>
 #include <chrono>
 #include <condition_variable>
 
@@ -333,36 +332,6 @@ HttpTester::test_parse_url_just_slash()
     CPPUNIT_ASSERT_EQUAL(std::string("/"), parsed.target);
 }
 
-void
-HttpTester::test_completed_resolver_callback_uses_io_thread()
-{
-    asio::io_context ioContext;
-    const asio::ip::tcp::endpoint endpoint(asio::ip::address_v4::loopback(), 80);
-    dht::http::Resolver resolver(ioContext, {endpoint});
-    const auto callerThread = std::this_thread::get_id();
-    std::thread::id callbackThread;
-    asio::error_code callbackError;
-    std::vector<asio::ip::tcp::endpoint> resolvedEndpoints;
-    std::atomic_bool callbackCalled {false};
-
-    resolver.add_callback([&](const asio::error_code& ec, const auto& endpoints) {
-        callbackError = ec;
-        callbackThread = std::this_thread::get_id();
-        resolvedEndpoints = endpoints;
-        callbackCalled = true;
-    });
-
-    CPPUNIT_ASSERT(!callbackCalled);
-    std::thread ioThread([&] { ioContext.run(); });
-    ioThread.join();
-
-    CPPUNIT_ASSERT(callbackCalled);
-    CPPUNIT_ASSERT(!callbackError);
-    CPPUNIT_ASSERT(callbackThread != callerThread);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), resolvedEndpoints.size());
-    CPPUNIT_ASSERT_EQUAL(endpoint, resolvedEndpoints.front());
-}
-
 void
 HttpTester::test_send_json()
 {

tests/test_http.h

@@ -39,7 +39,6 @@ class HttpTester : public CppUnit::TestFixture
     CPPUNIT_TEST(test_parse_url_complex_path_query_fragment);
     CPPUNIT_TEST(test_parse_url_empty);
     CPPUNIT_TEST(test_parse_url_just_slash);
-    CPPUNIT_TEST(test_completed_resolver_callback_uses_io_thread);
     // send
     CPPUNIT_TEST(test_send_json);
     CPPUNIT_TEST_SUITE_END();
@@ -83,7 +82,6 @@ public:
     void test_parse_url_complex_path_query_fragment();
     void test_parse_url_empty();
     void test_parse_url_just_slash();
-    void test_completed_resolver_callback_uses_io_thread();
     /**
      * Test send(json)
      */

tests/test_networkengine.cpp

@@ -214,12 +214,9 @@ makeEngine(std::unique_ptr<TestDatagramSocket>&& socket,
            std::mt19937_64& rd,
            int& onNewNodeCalls)
 {
-    net::NetworkConfig config {};
-    config.max_req_per_sec = -1;
-    config.max_peer_req_per_sec = -1;
     return net::NetworkEngine(
         myid,
-        config,
+        {},
         std::move(socket),
         {},
         rd,
@@ -237,21 +234,6 @@ makeEngine(std::unique_ptr<TestDatagramSocket>&& socket,
         [](Sp<Node>, const InfoHash&, const Blob&, const Value::Id&) { return net::RequestAnswer {}; });
 }
 
-static Sp<Node>
-addPendingGetRequest(net::NetworkEngine& engine, const InfoHash& id, const SockAddr& addr, Tid tid)
-{
-    auto node = engine.insertNode(id, addr);
-    auto request = std::make_shared<net::Request>(
-        MessageType::GetValues,
-        tid,
-        node,
-        Blob {},
-        [](const net::Request&, ParsedMessage&&) {},
-        [](const net::Request&, bool) {});
-    node->requested(request);
-    return node;
-}
-
 void
 NetworkEngineTester::setUp()
 {}
@@ -261,130 +243,24 @@ NetworkEngineTester::tearDown()
 {}
 
 void
-NetworkEngineTester::testCompletesPartialSessionWithDataBeforeHeader()
+NetworkEngineTester::testIgnoresUnknownPartialData()
 {
     Scheduler scheduler;
     std::mt19937_64 rd(0);
     InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
     int onNewNodeCalls = 0;
     auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
 
-    auto serialized = serializeValue("fragment before header");
+    Blob fragment = {1, 2, 3};
+    auto packet = makeValueDataPacketBlob(99, 0, 0, fragment);
     auto from = makeIPv4("127.0.0.2", 5000);
-    addPendingGetRequest(engine, remoteId, from, 99);
-    auto initialOnNewNodeCalls = onNewNodeCalls;
-    auto fragment = makeValueDataPacketBlob(99, 0, 0, serialized);
-    auto header = makeReplyHeaderPacketBlob(remoteId, 99, {serialized.size()});
 
-    engine.processMessage(fragment.data(), fragment.size(), from);
-    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-    CPPUNIT_ASSERT_EQUAL(initialOnNewNodeCalls, onNewNodeCalls);
-
-    engine.processMessage(header.data(), header.size(), from);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
-}
-
-void
-NetworkEngineTester::testBuffersLargeValueBeforeHeader()
-{
-    Scheduler scheduler;
-    std::mt19937_64 rd(8);
-    InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
-    int onNewNodeCalls = 0;
-    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
-
-    auto serialized = serializeValue(std::string(57 * 1024, 'l'));
-    auto from = makeIPv4("127.0.0.2", 5008);
-    addPendingGetRequest(engine, remoteId, from, 101);
-    auto initialOnNewNodeCalls = onNewNodeCalls;
-    for (size_t offset = 0; offset < serialized.size(); offset += TEST_MTU) {
-        auto end = std::min(offset + TEST_MTU, serialized.size());
-        Blob data(serialized.begin() + offset, serialized.begin() + end);
-        auto fragment = makeValueDataPacketBlob(101, 0, offset, data);
-        engine.processMessage(fragment.data(), fragment.size(), from);
-    }
-    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-
-    auto header = makeReplyHeaderPacketBlob(remoteId, 101, {serialized.size()});
-    engine.processMessage(header.data(), header.size(), from);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
-}
-
-void
-NetworkEngineTester::testIgnoresEmptyPartialData()
-{
-    Scheduler scheduler;
-    std::mt19937_64 rd(7);
-    InfoHash myid = InfoHash::getRandom(rd);
-    int onNewNodeCalls = 0;
-    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
-
-    auto packet = makeValueDataPacketBlob(100, 0, 0, {});
-    auto from = makeIPv4("127.0.0.2", 5007);
     engine.processMessage(packet.data(), packet.size(), from);
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
     CPPUNIT_ASSERT_EQUAL(0, onNewNodeCalls);
 }
 
-void
-NetworkEngineTester::testRejectsUnsolicitedPartialReply()
-{
-    Scheduler scheduler;
-    std::mt19937_64 rd(9);
-    InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
-    int onNewNodeCalls = 0;
-    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
-
-    auto from = makeIPv4("127.0.0.2", 5009);
-    auto header = makeReplyHeaderPacketBlob(remoteId, 102, {TEST_MTU});
-    engine.processMessage(header.data(), header.size(), from);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getNodeCacheSize());
-}
-
-void
-NetworkEngineTester::testDropsMalformedCompletedPartialMessage()
-{
-    Scheduler scheduler;
-    std::mt19937_64 rd(10);
-    InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
-    int onNewNodeCalls = 0;
-    auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
-
-    auto from = makeIPv4("127.0.0.2", 5010);
-    addPendingGetRequest(engine, remoteId, from, 103);
-    Blob malformed {0xc1};
-    auto header = makeReplyHeaderPacketBlob(remoteId, 103, {malformed.size()});
-    auto fragment = makeValueDataPacketBlob(103, 0, 0, malformed);
-
-    engine.processMessage(header.data(), header.size(), from);
-    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-    engine.processMessage(fragment.data(), fragment.size(), from);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-
-    auto earlyFrom = makeIPv4("127.0.0.2", 5011);
-    addPendingGetRequest(engine, remoteId, earlyFrom, 104);
-    auto earlyHeader = makeReplyHeaderPacketBlob(remoteId, 104, {malformed.size()});
-    auto earlyFragment = makeValueDataPacketBlob(104, 0, 0, malformed);
-
-    engine.processMessage(earlyFragment.data(), earlyFragment.size(), earlyFrom);
-    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-    engine.processMessage(earlyHeader.data(), earlyHeader.size(), earlyFrom);
-
-    CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-}
-
 void
 NetworkEngineTester::testCompletesPartialSessionFromSameSource()
 {
@@ -398,8 +274,6 @@ NetworkEngineTester::testCompletesPartialSessionFromSameSource()
     std::string data(TEST_MTU * 2 + 64, 'z');
     auto serialized = serializeValue(data);
     auto from = makeIPv4("127.0.0.2", 5001);
-    addPendingGetRequest(engine, remoteId, from, 7);
-    auto initialOnNewNodeCalls = onNewNodeCalls;
     auto header = makeReplyHeaderPacketBlob(remoteId, 7, {serialized.size()});
 
     engine.processMessage(header.data(), header.size(), from);
@@ -413,40 +287,45 @@ NetworkEngineTester::testCompletesPartialSessionFromSameSource()
     }
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
+    CPPUNIT_ASSERT(onNewNodeCalls > 0);
 }
 
 void
-NetworkEngineTester::testSeparatesPartialSessionsBySource()
+NetworkEngineTester::testKeepsSessionForWrongSourceFragment()
 {
     Scheduler scheduler;
     std::mt19937_64 rd(2);
     InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash firstRemoteId = InfoHash::getRandom(rd);
-    InfoHash secondRemoteId = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
     int onNewNodeCalls = 0;
     auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
 
-    auto serialized = serializeValue("same transaction, different peers");
-    auto firstFrom = makeIPv4("127.0.0.2", 5002);
-    auto secondFrom = makeIPv4("127.0.0.3", 5002);
-    addPendingGetRequest(engine, firstRemoteId, firstFrom, 11);
-    addPendingGetRequest(engine, secondRemoteId, secondFrom, 11);
-    auto initialOnNewNodeCalls = onNewNodeCalls;
-    auto firstHeader = makeReplyHeaderPacketBlob(firstRemoteId, 11, {serialized.size()});
-    auto secondHeader = makeReplyHeaderPacketBlob(secondRemoteId, 11, {serialized.size()});
+    std::string data(TEST_MTU + 32, 'w');
+    auto serialized = serializeValue(data);
+    auto goodFrom = makeIPv4("127.0.0.2", 5002);
+    auto badFrom = makeIPv4("127.0.0.3", 5002);
+    auto header = makeReplyHeaderPacketBlob(remoteId, 11, {serialized.size()});
 
-    engine.processMessage(firstHeader.data(), firstHeader.size(), firstFrom);
-    engine.processMessage(secondHeader.data(), secondHeader.size(), secondFrom);
-    CPPUNIT_ASSERT_EQUAL((size_t) 2, engine.getPartialCount());
-
-    auto fragment = makeValueDataPacketBlob(11, 0, 0, serialized);
-    engine.processMessage(fragment.data(), fragment.size(), firstFrom);
+    engine.processMessage(header.data(), header.size(), goodFrom);
     CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-    engine.processMessage(fragment.data(), fragment.size(), secondFrom);
+
+    auto firstEnd = std::min(TEST_MTU, serialized.size());
+    Blob firstFragment(serialized.begin(), serialized.begin() + firstEnd);
+    auto firstPacket = makeValueDataPacketBlob(11, 0, 0, firstFragment);
+    engine.processMessage(firstPacket.data(), firstPacket.size(), badFrom);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
+    CPPUNIT_ASSERT_EQUAL(0, onNewNodeCalls);
+
+    engine.processMessage(firstPacket.data(), firstPacket.size(), goodFrom);
+    if (firstEnd < serialized.size()) {
+        Blob tail(serialized.begin() + firstEnd, serialized.end());
+        auto tailPacket = makeValueDataPacketBlob(11, 0, firstEnd, tail);
+        engine.processMessage(tailPacket.data(), tailPacket.size(), goodFrom);
+    }
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
+    CPPUNIT_ASSERT(onNewNodeCalls > 0);
 }
 
 void

tests/test_networkengine.h

@@ -13,13 +13,9 @@ class NetworkEngineTester : public CppUnit::TestFixture
 #ifdef _MSC_VER
     CPPUNIT_TEST(testDisabledOnMsvc);
 #else
-    CPPUNIT_TEST(testCompletesPartialSessionWithDataBeforeHeader);
-    CPPUNIT_TEST(testBuffersLargeValueBeforeHeader);
-    CPPUNIT_TEST(testIgnoresEmptyPartialData);
-    CPPUNIT_TEST(testRejectsUnsolicitedPartialReply);
-    CPPUNIT_TEST(testDropsMalformedCompletedPartialMessage);
+    CPPUNIT_TEST(testIgnoresUnknownPartialData);
     CPPUNIT_TEST(testCompletesPartialSessionFromSameSource);
-    CPPUNIT_TEST(testSeparatesPartialSessionsBySource);
+    CPPUNIT_TEST(testKeepsSessionForWrongSourceFragment);
     CPPUNIT_TEST(testListenConfirmationCarriesToken);
     CPPUNIT_TEST(testListenConfirmationUpdatesSearchNodeToken);
     CPPUNIT_TEST(testListenReopensSocketAfterNodeExpiration);
@@ -34,13 +30,9 @@ public:
 #ifdef _MSC_VER
     void testDisabledOnMsvc();
 #else
-    void testCompletesPartialSessionWithDataBeforeHeader();
-    void testBuffersLargeValueBeforeHeader();
-    void testIgnoresEmptyPartialData();
-    void testRejectsUnsolicitedPartialReply();
-    void testDropsMalformedCompletedPartialMessage();
+    void testIgnoresUnknownPartialData();
     void testCompletesPartialSessionFromSameSource();
-    void testSeparatesPartialSessionsBySource();
+    void testKeepsSessionForWrongSourceFragment();
     void testListenConfirmationCarriesToken();
     void testListenConfirmationUpdatesSearchNodeToken();
     void testListenReopensSocketAfterNodeExpiration();

tests/test_parsedmessage.cpp

@@ -328,104 +328,6 @@ ParsedMessageTester::testParseRejectsInvalidValuesField()
     CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
 }
 
-void
-ParsedMessageTester::testParseRejectsEmptyFieldSelection()
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    pk.pack_map(3);
-    pk.pack(KEY_R);
-    pk.pack_map(1);
-    pk.pack(KEY_REQ_FIELDS);
-    pk.pack_map(2);
-    pk.pack("f");
-    pk.pack_array(0);
-    pk.pack("v");
-    pk.pack_array(0);
-    pk.pack(KEY_TID);
-    pk.pack(1u);
-    pk.pack(KEY_Y);
-    pk.pack(KEY_R);
-
-    auto msg = msgpack::unpack(buffer.data(), buffer.size());
-    ParsedMessage parsed;
-    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
-}
-
-void
-ParsedMessageTester::testParseRejectsEmptyErrorArray()
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    pk.pack_map(3);
-    pk.pack(KEY_E);
-    pk.pack_array(0);
-    pk.pack(KEY_TID);
-    pk.pack(1u);
-    pk.pack(KEY_Y);
-    pk.pack(KEY_E);
-
-    auto msg = msgpack::unpack(buffer.data(), buffer.size());
-    ParsedMessage parsed;
-    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
-}
-
-void
-ParsedMessageTester::testParseRejectsNonMapRequest()
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    pk.pack_map(4);
-    pk.pack(KEY_A);
-    pk.pack_array(2);
-    pk.pack(1u);
-    pk.pack(2u);
-    pk.pack(KEY_Q);
-    pk.pack(QUERY_PING);
-    pk.pack(KEY_TID);
-    pk.pack(1u);
-    pk.pack(KEY_Y);
-    pk.pack("q");
-
-    auto msg = msgpack::unpack(buffer.data(), buffer.size());
-    ParsedMessage parsed;
-    CPPUNIT_ASSERT_THROW(parsed.msgpack_unpack(msg.get()), msgpack::type_error);
-}
-
-void
-ParsedMessageTester::testParseErrorOnlyMessageKeepsRequestFieldsEmpty()
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    // An error message with no request map: the error array must not be walked
-    // as if it were a sequence of key/value pairs.
-    pk.pack_map(3);
-    pk.pack(KEY_E);
-    pk.pack_array(2);
-    pk.pack(401u);
-    pk.pack("unauthorized");
-    pk.pack(KEY_TID);
-    pk.pack(9u);
-    pk.pack(KEY_Y);
-    pk.pack(KEY_E);
-
-    auto msg = msgpack::unpack(buffer.data(), buffer.size());
-    ParsedMessage parsed;
-    parsed.msgpack_unpack(msg.get());
-
-    CPPUNIT_ASSERT(parsed.type == MessageType::Error);
-    CPPUNIT_ASSERT_EQUAL((uint16_t) 401, parsed.error_code);
-    CPPUNIT_ASSERT(not parsed.id);
-    CPPUNIT_ASSERT(parsed.token.empty());
-    CPPUNIT_ASSERT(parsed.values.empty());
-    CPPUNIT_ASSERT(parsed.nodes4_raw.empty());
-    CPPUNIT_ASSERT(parsed.nodes6_raw.empty());
-}
-
 void
 ParsedMessageTester::testParseIgnoresIncompleteValueDataEntry()
 {

tests/test_parsedmessage.h

@@ -24,10 +24,6 @@ class ParsedMessageTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testParseRejectsNonMapPacket);
     CPPUNIT_TEST(testParseRejectsInvalidValueDataPayload);
     CPPUNIT_TEST(testParseRejectsInvalidValuesField);
-    CPPUNIT_TEST(testParseRejectsEmptyFieldSelection);
-    CPPUNIT_TEST(testParseRejectsEmptyErrorArray);
-    CPPUNIT_TEST(testParseRejectsNonMapRequest);
-    CPPUNIT_TEST(testParseErrorOnlyMessageKeepsRequestFieldsEmpty);
     CPPUNIT_TEST(testParseIgnoresIncompleteValueDataEntry);
     CPPUNIT_TEST(testCompleteEmpty);
     CPPUNIT_TEST(testCompleteIncomplete);
@@ -57,10 +53,6 @@ public:
     void testParseRejectsNonMapPacket();
     void testParseRejectsInvalidValueDataPayload();
     void testParseRejectsInvalidValuesField();
-    void testParseRejectsEmptyFieldSelection();
-    void testParseRejectsEmptyErrorArray();
-    void testParseRejectsNonMapRequest();
-    void testParseErrorOnlyMessageKeepsRequestFieldsEmpty();
     void testParseIgnoresIncompleteValueDataEntry();
     void testCompleteEmpty();
     void testCompleteIncomplete();

tests/test_value.cpp

@@ -137,29 +137,4 @@ ValueTester::testPushTypePreservedAfterEncrypt()
     CPPUNIT_ASSERT(encrypted.isEncrypted());
 }
 
-void
-ValueTester::testPushTypeJsonRoundTrip()
-{
-#ifdef OPENDHT_JSONCPP
-    dht::Value original {(const uint8_t*) "hello", 5};
-    original.id = 42;
-    original.priority = 1;
-    original.pushType = "videoCall";
-
-    dht::Value restored(original.toJson());
-
-    CPPUNIT_ASSERT_EQUAL(original.id, restored.id);
-    CPPUNIT_ASSERT_EQUAL(original.priority, restored.priority);
-    CPPUNIT_ASSERT_EQUAL(original.pushType, restored.pushType);
-
-    // No pushType: the field must stay absent from the JSON and empty after parse.
-    dht::Value plain {(const uint8_t*) "data", 4};
-    plain.id = 7;
-    auto json = plain.toJson();
-    CPPUNIT_ASSERT(!json.isMember("pt"));
-    dht::Value plainRestored(json);
-    CPPUNIT_ASSERT_EQUAL(std::string(), plainRestored.pushType);
-#endif
-}
-
 } // namespace test

tests/test_value.h

@@ -16,7 +16,6 @@ class ValueTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testPushTypeMsgpackRoundTrip);
     CPPUNIT_TEST(testPushTypeAbsentAfterUnpack);
     CPPUNIT_TEST(testPushTypePreservedAfterEncrypt);
-    CPPUNIT_TEST(testPushTypeJsonRoundTrip);
     CPPUNIT_TEST_SUITE_END();
 
 public:
@@ -39,7 +38,6 @@ public:
     void testPushTypeMsgpackRoundTrip();
     void testPushTypeAbsentAfterUnpack();
     void testPushTypePreservedAfterEncrypt();
-    void testPushTypeJsonRoundTrip();
 };
 
 } // namespace test

vcpkg-configuration.json

@@ -1,7 +1,7 @@
 {
   "default-registry": {
     "kind": "git",
-    "baseline": "f3e10653cc27d62a37a3763cd84b38bca07c6075",
+    "baseline": "66c0373dc7fca549e5803087b9487edfe3aca0a1",
     "repository": "https://github.com/microsoft/vcpkg"
   },
   "registries": [

vcpkg.json

@@ -1,6 +1,6 @@
 {
   "name": "opendht",
-  "version-string": "4.3.0",
+  "version-string": "4.0.0",
   "port-version": 1,
   "description": "OpenDHT: A lightweight C++17 Distributed Hash Table library.",
   "homepage": "https://opendht.net",