http: bound name resolution and reuse the endpoints it found

Request::send waits for Resolver before doing anything else, and the
timeout a caller sets through Request::timeout is only armed later, in
connect. A system resolver that never answers therefore holds the request
for good: async_resolve has no deadline of its own, no timer covers the
wait, and the callbacks queued in Resolver are never invoked. The proxy
client reaches that path on every connectivity change, since
connectivityChanged calls getProxyInfos, so a device coming back from
sleep can sit there with no status, no listener and no error.

getProxyInfos also builds a new Resolver each time and drops the previous
one, discarding endpoints that were still perfectly usable. Resolution is
then redone from scratch at the exact moment the network stack has just
come up and is at its slowest.

Give Resolver a deadline of its own, armed as soon as a query starts, and
report timed_out to whoever is waiting. Let it also start from endpoints
that were already resolved: those answer callbacks immediately so a
connection begins without waiting, while the query still runs and
replaces them as soon as it returns. An address that went stale in the
meantime now costs one connection attempt, which Request retries against
the pending query rather than reporting as a failure.

A failed query no longer clears endpoints that are known to work, so a
transient DNS outage leaves the client with something to connect to.

Co-authored-by: Copilot App <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"
-  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"
+  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"
 
 on: [push, pull_request]
 
@@ -65,12 +65,12 @@ jobs:
       - name: Run Unit tests
         run: ctest --output-on-failure --test-dir build
 
-  build-debian12:
-    name: Debian 12/GCC build
+  build-debian13:
+    name: Debian 13/GCC build
     runs-on: ubuntu-latest
 
     container:
-      image: debian:12
+      image: debian:13
       volumes:
         - .:/workspace
       options: --workdir /workspace
@@ -83,7 +83,7 @@ jobs:
       - name: Install Dependencies
         run: |
           apt-get update
-          apt-get install -y $DEBIAN12_DEPS
+          apt-get install -y $DEBIAN13_DEPS
 
       - name: Build llhttp dependency
         run: |

CMakeLists.txt

@@ -10,7 +10,7 @@ if(POLICY CMP0167)
 endif()
 
 set (opendht_VERSION_MAJOR 4)
-set (opendht_VERSION_MINOR 2.0)
+set (opendht_VERSION_MINOR 3.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

include/opendht/crypto.h

@@ -204,7 +204,13 @@ 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);
-    static PrivateKey generateEC();
+
+    /**
+     * 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);
 
     gnutls_privkey_t key {};
     gnutls_x509_privkey_t x509_key {};
@@ -272,7 +278,8 @@ public:
      * 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) {
+    void sign(const Identity& id, duration validity_period = {}, uint64_t number = 0)
+    {
         sign(*id.first, *id.second, validity_period, number);
     }
 

include/opendht/http.h

@@ -162,11 +162,34 @@ public:
              const bool ssl = false,
              std::shared_ptr<log::Logger> logger = {});
 
+    /**
+     * Start from previously resolved endpoints while querying the url anyway.
+     * Callbacks are answered from the cached endpoints without waiting, and the
+     * query replaces them as soon as it returns, so an address that went stale
+     * costs a connection attempt rather than a resolution delay.
+     */
+    Resolver(asio::io_context& ctx,
+             std::string url,
+             std::vector<asio::ip::tcp::endpoint> endpoints,
+             std::shared_ptr<log::Logger> logger = {});
+
     ~Resolver();
 
     inline const Url& get_url() const { return url_; }
 
-    void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC);
+    /** Endpoints known so far, either cached or resolved. May be empty. */
+    std::vector<asio::ip::tcp::endpoint> get_endpoints() const;
+
+    /** Whether a query is still running, so a different answer may still come. */
+    bool is_resolving() const;
+
+    /**
+     * Register a callback for the resolved endpoints.
+     * When freshOnly is set, a cached answer is not enough: the callback waits
+     * for the pending query, so a caller that failed to reach a cached address
+     * can retry against whatever the system resolver returns.
+     */
+    void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC, bool freshOnly = false);
 
     std::shared_ptr<log::Logger> getLogger() const { return logger_; }
 
@@ -174,17 +197,27 @@ public:
 
 private:
     void resolve(std::string_view host, std::string_view service);
+    void complete(const asio::error_code& ec, std::vector<asio::ip::tcp::endpoint>&& endpoints);
+
+    /**
+     * A stuck system resolver would otherwise hold every pending request for
+     * good: Request only arms its own timeout once resolution has answered.
+     */
+    static constexpr std::chrono::seconds RESOLVE_TIMEOUT {15};
 
     mutable std::mutex mutex_;
 
     Url url_;
     asio::error_code ec_;
     asio::ip::tcp::resolver resolver_;
+    std::unique_ptr<asio::steady_timer> timer_;
     std::shared_ptr<bool> destroyed_;
     std::vector<asio::ip::tcp::endpoint> endpoints_;
 
     bool completed_ {false};
+    bool resolving_ {false};
     std::queue<ResolverCb> cbs_;
+    std::queue<ResolverCb> freshCbs_;
 
     std::shared_ptr<log::Logger> logger_;
 };
@@ -317,6 +350,12 @@ private:
 
     void connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb = {});
 
+    /**
+     * Resolve then connect. Retried once with freshOnly set when the endpoints
+     * that were handed out came from a cache and turned out to be unreachable.
+     */
+    void resolve_and_connect(bool freshOnly);
+
     void post();
 
     void handle_request(const asio::error_code& ec);

include/opendht/network_engine.h

@@ -489,8 +489,12 @@ 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);
 
@@ -527,7 +531,8 @@ private:
     {
         return packValueHeader(buf, values.begin(), values.end());
     }
-    void maintainRxBuffer(Tid tid);
+    using PartialMessageKey = std::pair<SockAddr, Tid>;
+    void maintainRxBuffer(const PartialMessageKey& key);
 
     /*************
      *  Answers  *
@@ -573,7 +578,7 @@ private:
 
     // requests handling
     std::map<Tid, Sp<Request>> requests {};
-    std::map<Tid, PartialMessage> partial_messages;
+    std::map<PartialMessageKey, PartialMessage> partial_messages;
 
     MessageStats in_stats {}, out_stats {};
     std::set<SockAddr> blacklist {};

include/opendht/utils.h

@@ -85,6 +85,14 @@ 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.2.0',
+    version: '4.3.0',
     meson_version: '>=1.1.0',
     default_options: ['cpp_std=c++20', 'warning_level=3'],
 )

src/crypto.cpp

@@ -1294,14 +1294,16 @@ PrivateKey::generate(unsigned key_length, gnutls_pk_algorithm_t algo)
 }
 
 PrivateKey
-PrivateKey::generateEC()
+PrivateKey::generateEC(gnutls_ecc_curve_t curve)
 {
     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,
-                                           gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA),
+                                           curve == GNUTLS_ECC_CURVE_INVALID
+                                               ? gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA)
+                                               : GNUTLS_CURVE_TO_BITS(curve),
                                            0);
     if (err != GNUTLS_E_SUCCESS) {
         gnutls_x509_privkey_deinit(key);

src/dht.cpp

@@ -2363,6 +2363,9 @@ 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) {
@@ -2375,7 +2378,7 @@ Dht::exportValues() const
         pk.pack_array(vals.size());
         for (const auto& v : vals) {
             pk.pack_array(4);
-            pk.pack(v.created.time_since_epoch().count());
+            pk.pack(to_system_time(v.created, now, system_now));
             v.data->msgpack_pack(pk);
             if (const auto& store_addr = v.store_bucket ? v.store_bucket->getAddr() : SockAddr {}) {
                 pk.pack_bin(store_addr.getLength());
@@ -2383,7 +2386,7 @@ Dht::exportValues() const
             } else {
                 pk.pack_bin(0);
             }
-            pk.pack(v.expiration.time_since_epoch().count());
+            pk.pack(to_system_time(v.expiration, now, system_now));
         }
         ve.second = {buffer.data(), buffer.data() + buffer.size()};
         e.push_back(std::move(ve));
@@ -2395,6 +2398,7 @@ 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;
@@ -2419,8 +2423,15 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                 Value tmp_val;
                 SockAddr store_addr;
                 try {
-                    val_time = time_point {
-                        time_point::duration {valel.via.array.ptr[0].as<time_point::duration::rep>()}};
+                    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>()}};
+                    }
                     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>();
@@ -2430,8 +2441,14 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                         }
                     }
                     if (valel.via.array.size >= 4) {
-                        expiration = time_point {
-                            time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
+                        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>()}};
+                        }
                     }
                 } catch (const std::exception&) {
                     if (logger_)

src/dht_proxy_client.cpp

@@ -652,7 +652,21 @@ DhtProxyClient::getProxyInfos()
     if (logger_)
         logger_->debug("[proxy:client] [status] sending request");
 
-    auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
+    // Reuse what the previous resolver found: the query still runs, but a
+    // connection can start at once instead of waiting for the system resolver,
+    // which is the slowest step when the device wakes up from sleep.
+    std::vector<asio::ip::tcp::endpoint> knownEndpoints;
+    {
+        std::lock_guard l(resolverLock_);
+        if (resolver_)
+            knownEndpoints = resolver_->get_endpoints();
+    }
+    auto resolver = knownEndpoints.empty()
+                        ? std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_)
+                        : std::make_shared<http::Resolver>(httpContext_,
+                                                           proxyUrl_,
+                                                           std::move(knownEndpoints),
+                                                           logger_);
     queryProxyInfo(infoState, resolver, AF_INET);
     queryProxyInfo(infoState, resolver, AF_INET6);
     std::lock_guard l(resolverLock_);

src/dhtrunner.cpp

@@ -554,8 +554,14 @@ DhtRunner::registerInsecureType(const ValueType& type)
 void
 DhtRunner::importValues(const std::vector<ValuesExport>& values)
 {
-    std::lock_guard lck(dht_mtx);
-    dht_->importValues(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();
 }
 
 unsigned

src/http.cpp

@@ -308,7 +308,7 @@ issuer_from_chain(STACK_OF(X509) * fullchain)
     if ((issuer_name = X509_get_issuer_name(cert)) == nullptr)
         return nullptr;
 
-    issuer = X509_find_by_subject(fullchain, issuer_name);
+    issuer = X509_find_by_subject(fullchain, const_cast<X509_NAME*>(issuer_name));
     return issuer;
 }
 
@@ -867,12 +867,31 @@ Resolver::Resolver(asio::io_context& ctx,
     completed_ = true;
 }
 
+Resolver::Resolver(asio::io_context& ctx,
+                   std::string url,
+                   std::vector<asio::ip::tcp::endpoint> endpoints,
+                   std::shared_ptr<dht::Logger> logger)
+    : url_(std::move(url))
+    , resolver_(ctx)
+    , destroyed_(std::make_shared<bool>(false))
+    , endpoints_(std::move(endpoints))
+    , logger_(logger)
+{
+    // Answer from the cached endpoints right away so a connection can start
+    // without waiting for the system resolver, and query anyway so a stale
+    // address is replaced as soon as the answer comes back.
+    completed_ = not endpoints_.empty();
+    resolve(url_.host, url_.service.empty() ? url_.protocol : url_.service);
+}
+
 Resolver::~Resolver()
 {
     decltype(cbs_) cbs;
+    decltype(freshCbs_) freshCbs;
     {
         std::lock_guard lock(mutex_);
         cbs = std::move(cbs_);
+        freshCbs = std::move(freshCbs_);
     }
     while (not cbs.empty()) {
         auto cb = cbs.front();
@@ -880,9 +899,29 @@ Resolver::~Resolver()
             cb(asio::error::operation_aborted, {});
         cbs.pop();
     }
+    while (not freshCbs.empty()) {
+        auto cb = freshCbs.front();
+        if (cb)
+            cb(asio::error::operation_aborted, {});
+        freshCbs.pop();
+    }
     *destroyed_ = true;
 }
 
+std::vector<asio::ip::tcp::endpoint>
+Resolver::get_endpoints() const
+{
+    std::lock_guard lock(mutex_);
+    return endpoints_;
+}
+
+bool
+Resolver::is_resolving() const
+{
+    std::lock_guard lock(mutex_);
+    return resolving_;
+}
+
 void
 Resolver::cancel()
 {
@@ -905,27 +944,68 @@ filter(const std::vector<asio::ip::tcp::endpoint>& epts, sa_family_t family)
 }
 
 void
-Resolver::add_callback(ResolverCb cb, sa_family_t family)
+Resolver::add_callback(ResolverCb cb, sa_family_t family, bool freshOnly)
 {
     std::lock_guard lock(mutex_);
+    auto wrapped = family == AF_UNSPEC
+                       ? std::move(cb)
+                       : ResolverCb {[cb, family](const asio::error_code& ec,
+                                                  const std::vector<asio::ip::tcp::endpoint>& endpoints) {
+                             if (ec)
+                                 cb(ec, endpoints);
+                             else
+                                 cb(ec, filter(endpoints, family));
+                         }};
     if (!completed_)
-        cbs_.emplace(family == AF_UNSPEC ? std::move(cb)
-                                         : [cb, family](const asio::error_code& ec,
-                                                        const std::vector<asio::ip::tcp::endpoint>& endpoints) {
-                                               if (ec)
-                                                   cb(ec, endpoints);
-                                               else
-                                                   cb(ec, filter(endpoints, family));
-                                           });
+        cbs_.emplace(std::move(wrapped));
+    else if (freshOnly and resolving_)
+        // The cached endpoints already failed this caller, so hold the callback
+        // until the query that is still running has something else to offer.
+        freshCbs_.emplace(std::move(wrapped));
     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); });
+                   [cb = std::move(wrapped), ec = ec_, endpoints = endpoints_] { cb(ec, endpoints); });
+    }
+}
+
+void
+Resolver::complete(const asio::error_code& ec, std::vector<asio::ip::tcp::endpoint>&& endpoints)
+{
+    decltype(cbs_) cbs;
+    decltype(freshCbs_) freshCbs;
+    asio::error_code reportedEc;
+    std::vector<asio::ip::tcp::endpoint> reportedEndpoints;
+    {
+        std::lock_guard lock(mutex_);
+        resolving_ = false;
+        if (not ec) {
+            ec_ = ec;
+            endpoints_ = std::move(endpoints);
+        } else if (endpoints_.empty()) {
+            // Nothing cached to fall back on, so the failure has to be reported.
+            ec_ = ec;
+        }
+        completed_ = true;
+        cbs = std::move(cbs_);
+        freshCbs = std::move(freshCbs_);
+        reportedEc = ec_;
+        reportedEndpoints = endpoints_;
+    }
+    while (not cbs.empty()) {
+        auto cb = cbs.front();
+        if (cb)
+            cb(reportedEc, reportedEndpoints);
+        cbs.pop();
+    }
+    while (not freshCbs.empty()) {
+        auto cb = freshCbs.front();
+        if (cb)
+            cb(reportedEc, reportedEndpoints);
+        freshCbs.pop();
     }
 }
 
@@ -943,30 +1023,45 @@ Resolver::resolve(std::string_view host, std::string_view service)
     } else if (service == "https"sv) {
         service = "443"sv;
     }
+
+    {
+        std::lock_guard lock(mutex_);
+        resolving_ = true;
+    }
+
+    if (not timer_)
+        timer_ = std::make_unique<asio::steady_timer>(resolver_.get_executor());
+    timer_->expires_after(RESOLVE_TIMEOUT);
+    timer_->async_wait([this, destroyed = destroyed_](const asio::error_code& ec) {
+        if (ec == asio::error::operation_aborted or *destroyed)
+            return;
+        {
+            std::lock_guard lock(mutex_);
+            if (not resolving_)
+                return;
+            // Claim the completion under the lock so the query answering at the
+            // same moment cannot report a second time.
+            resolving_ = false;
+        }
+        if (logger_)
+            logger_->error("[http:client] [resolver] timed out resolving {:s}", url_.host);
+        resolver_.cancel();
+        complete(asio::error::timed_out, {});
+    });
+
     resolver_.async_resolve(host,
                             service,
                             [this, destroyed = destroyed_](const asio::error_code& ec,
                                                            asio::ip::tcp::resolver::results_type endpoints) {
                                 if (ec == asio::error::operation_aborted or *destroyed)
                                     return;
+                                if (timer_)
+                                    timer_->cancel();
                                 if (logger_) {
                                     logger_->debug("[http:client] [resolver] got result: {:s}", ec.message());
                                 }
-                                decltype(cbs_) cbs;
-                                {
-                                    std::lock_guard lock(mutex_);
-                                    ec_ = ec;
-                                    endpoints_ = std::vector<asio::ip::tcp::endpoint> {endpoints.begin(),
-                                                                                       endpoints.end()};
-                                    completed_ = true;
-                                    cbs = std::move(cbs_);
-                                }
-                                while (not cbs.empty()) {
-                                    auto cb = cbs.front();
-                                    if (cb)
-                                        cb(ec, endpoints_);
-                                    cbs.pop();
-                                }
+                                complete(ec,
+                                         std::vector<asio::ip::tcp::endpoint> {endpoints.begin(), endpoints.end()});
                             });
 }
 
@@ -1444,7 +1539,12 @@ void
 Request::send()
 {
     notify_state_change(State::CREATED);
+    resolve_and_connect(false);
+}
 
+void
+Request::resolve_and_connect(bool freshOnly)
+{
     std::weak_ptr<Request> wthis = shared_from_this();
     resolver_->add_callback(
         [wthis](const asio::error_code& ec, std::vector<asio::ip::tcp::endpoint> endpoints) {
@@ -1458,17 +1558,28 @@ Request::send()
                 } else if (!this_.conn_ or !this_.conn_->is_open()) {
                     this_.connect(std::move(endpoints), [wthis](const asio::error_code& ec) {
                         if (auto sthis = wthis.lock()) {
-                            if (ec)
-                                sthis->terminate(asio::error::not_connected);
-                            else
+                            if (not ec) {
                                 sthis->post();
+                            } else if (sthis->resolver_->is_resolving()) {
+                                // These endpoints came from a cache that no longer
+                                // points anywhere. The query is still running, so
+                                // wait for it rather than fail on a stale address.
+                                if (sthis->logger_)
+                                    sthis->logger_->debug(
+                                        "[http:request:{:d}] cached endpoints unreachable, awaiting resolution",
+                                        sthis->id_);
+                                sthis->resolve_and_connect(true);
+                            } else {
+                                sthis->terminate(asio::error::not_connected);
+                            }
                         }
                     });
                 } else
                     this_.post();
             }
         },
-        family_);
+        family_,
+        freshOnly);
 }
 
 void

src/network_engine.cpp

@@ -40,8 +40,44 @@ 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)
 {
@@ -466,35 +502,58 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
 
     // partial value data
     if (msg->type == MessageType::ValueData) {
-        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");
+        if (msg->fragment_parts.empty()
+            || std::any_of(msg->fragment_parts.begin(), msg->fragment_parts.end(), [](const auto& fragment) {
+                   return fragment.second.second.empty();
+               })) {
             rateLimit(from);
             return;
         }
-        if (!pmsg_it->second.from.equals(from)) {
-            if (logger_)
-                logger_->debug("Received partial message data from unexpected IP address");
-            rateLimit(from);
+        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));
             return;
         }
         // append data block
-        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());
-                }
+        if (pmsg.msg->append(*msg)) {
+            pmsg.last_part = now;
+            if (tryCompleteMessage(pmsg, from)) {
                 partial_messages.erase(pmsg_it);
             } else
-                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
+                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, key));
         }
         return;
     }
@@ -505,7 +564,32 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         return;
     }
 
-    if (msg->type > MessageType::Reply) {
+    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) {
         /* Rate limit requests. */
         if (!rateLimit(from)) {
             if (logger_)
@@ -514,28 +598,28 @@ NetworkEngine::processMessage(const uint8_t* buf, size_t buflen, SockAddr f)
         }
     }
 
-    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) {
+    // starting partial message session
+    auto inserted = partial_messages.emplace(partial_key, PartialMessage {});
+    auto& pmsg = inserted.first->second;
+    if (not pmsg.msg) {
+        if (inserted.second) {
             pmsg.from = from;
-            pmsg.msg = std::move(msg);
             pmsg.start = now;
-            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);
-    }
+            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);
 }
 
 void
@@ -1614,9 +1698,9 @@ NetworkEngine::sendError(const SockAddr& addr, Tid tid, uint16_t code, const std
 }
 
 void
-NetworkEngine::maintainRxBuffer(Tid tid)
+NetworkEngine::maintainRxBuffer(const PartialMessageKey& key)
 {
-    auto msg = partial_messages.find(tid);
+    auto msg = partial_messages.find(key);
     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/parsed_message.h

@@ -274,14 +274,21 @@ 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)
+        if (parsed.e->type != msgpack::type::ARRAY or parsed.e->via.array.size < 1)
             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;
@@ -290,8 +297,9 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         msgpack::object* want;
     } parsedReq {};
 
-    for (unsigned i = 0; i < req.via.map.size; i++) {
-        auto& o = req.via.map.ptr[i];
+    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];
         if (o.key.type != msgpack::type::STR)
             continue;
         auto key = o.key.as<std::string_view>();
@@ -370,6 +378,8 @@ 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/utils.cpp

@@ -238,28 +238,67 @@ 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)
 {
-    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;
+    return from_system_time(system_clock::from_time_t(t));
 }
 
 std::time_t
 to_time_t(time_point 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));
+    return system_clock::to_time_t(to_system_time(t));
 }
 
 Blob

tests/test_crypto.cpp

@@ -327,6 +327,16 @@ CryptoTester::testCertificateSerialNumber()
     }
 }
 
+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
 CryptoTester::testOcsp()
 {

tests/test_crypto.h

@@ -19,6 +19,7 @@ class CryptoTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testRevocationListPartitionedNumber);
     CPPUNIT_TEST(testCertificateRequest);
     CPPUNIT_TEST(testCertificateSerialNumber);
+    CPPUNIT_TEST(testEcKeyCurve);
     CPPUNIT_TEST(testOcsp);
     CPPUNIT_TEST(testAesEncryption);
     CPPUNIT_TEST(testAesEncryptionWithMultipleKeySizes);
@@ -68,6 +69,10 @@ public:
      * Test certificate serial number extraction
      */
     void testCertificateSerialNumber();
+    /**
+     * Test EC private key generation with an explicit curve
+     */
+    void testEcKeyCurve();
     /**
      * Test OCSP
      */

tests/test_dhtrunner.cpp

@@ -210,6 +210,42 @@ 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()
 {
@@ -267,14 +303,25 @@ DhtRunnerTester::testImportValuesPreservesStoredExpiration()
 void
 DhtRunnerTester::testImportPermanentValueExpires()
 {
-    // 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};
+    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));
 
     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);
@@ -282,63 +329,33 @@ 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()}}
     });
 
-    // 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) {
+    auto containsValue = [&] {
+        for (const auto& [exportedKey, data] : restored.exportValues()) {
+            if (exportedKey != key)
+                continue;
             msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*) ve.second.data(), ve.second.size());
+            msgpack::unpack(msg, reinterpret_cast<const char*>(data.data()), data.size());
             auto valarr = msg.get();
             CPPUNIT_ASSERT(valarr.type == msgpack::type::ARRAY);
-            // The value should have expired
-            CPPUNIT_ASSERT_EQUAL(uint32_t(0), valarr.via.array.size);
-            found = true;
+            return valarr.via.array.size != 0;
         }
-    }
-    // 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; }));
+        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());
 
     restored.shutdown();
     restored.join();
-    restored2.shutdown();
-    restored2.join();
 }
 
 void

tests/test_dhtrunner.h

@@ -21,6 +21,7 @@ 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);
@@ -67,6 +68,7 @@ public:
      */
     void testIdOps();
     void testImportValuesPreservesRemoteQuota();
+    void testExportValuesUsesSystemTime();
     void testImportValuesPreservesStoredExpiration();
     void testImportPermanentValueExpires();
     void testBootstrapSetsConnectingState();

tests/test_networkengine.cpp

@@ -214,9 +214,12 @@ 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,
@@ -234,6 +237,21 @@ 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()
 {}
@@ -243,24 +261,130 @@ NetworkEngineTester::tearDown()
 {}
 
 void
-NetworkEngineTester::testIgnoresUnknownPartialData()
+NetworkEngineTester::testCompletesPartialSessionWithDataBeforeHeader()
 {
     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);
 
-    Blob fragment = {1, 2, 3};
-    auto packet = makeValueDataPacketBlob(99, 0, 0, fragment);
+    auto serialized = serializeValue("fragment before header");
     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()
 {
@@ -274,6 +398,8 @@ 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);
@@ -287,45 +413,40 @@ NetworkEngineTester::testCompletesPartialSessionFromSameSource()
     }
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > 0);
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
 }
 
 void
-NetworkEngineTester::testKeepsSessionForWrongSourceFragment()
+NetworkEngineTester::testSeparatesPartialSessionsBySource()
 {
     Scheduler scheduler;
     std::mt19937_64 rd(2);
     InfoHash myid = InfoHash::getRandom(rd);
-    InfoHash remoteId = InfoHash::getRandom(rd);
+    InfoHash firstRemoteId = InfoHash::getRandom(rd);
+    InfoHash secondRemoteId = InfoHash::getRandom(rd);
     int onNewNodeCalls = 0;
     auto engine = makeEngine(std::make_unique<TestDatagramSocket>(), scheduler, myid, rd, onNewNodeCalls);
 
-    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()});
+    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()});
 
-    engine.processMessage(header.data(), header.size(), goodFrom);
+    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);
     CPPUNIT_ASSERT_EQUAL((size_t) 1, engine.getPartialCount());
-
-    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);
-    }
+    engine.processMessage(fragment.data(), fragment.size(), secondFrom);
 
     CPPUNIT_ASSERT_EQUAL((size_t) 0, engine.getPartialCount());
-    CPPUNIT_ASSERT(onNewNodeCalls > 0);
+    CPPUNIT_ASSERT(onNewNodeCalls > initialOnNewNodeCalls);
 }
 
 void

tests/test_networkengine.h

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

tests/test_parsedmessage.cpp

@@ -328,6 +328,104 @@ 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,6 +24,10 @@ 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);
@@ -53,6 +57,10 @@ public:
     void testParseRejectsNonMapPacket();
     void testParseRejectsInvalidValueDataPayload();
     void testParseRejectsInvalidValuesField();
+    void testParseRejectsEmptyFieldSelection();
+    void testParseRejectsEmptyErrorArray();
+    void testParseRejectsNonMapRequest();
+    void testParseErrorOnlyMessageKeepsRequestFieldsEmpty();
     void testParseIgnoresIncompleteValueDataEntry();
     void testCompleteEmpty();
     void testCompleteIncomplete();

vcpkg.json

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