dht: persist values with system time

.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]
 
@@ -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
@@ -83,7 +83,7 @@ jobs:
       - name: Install Dependencies
         run: |
           apt-get update
-          apt-get install -y $DEBIAN13_DEPS
+          apt-get install -y $DEBIAN12_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 3.0)
+set (opendht_VERSION_MINOR 2.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 {};
@@ -278,8 +272,7 @@ 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,34 +162,11 @@ 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_; }
 
-    /** 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);
+    void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC);
 
     std::shared_ptr<log::Logger> getLogger() const { return logger_; }
 
@@ -197,27 +174,17 @@ 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_;
 };
@@ -350,12 +317,6 @@ 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,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 {};

meson.build

@@ -5,7 +5,7 @@ project(
     'cython',
     license: 'MIT',
     license_files: 'LICENSE',
-    version: '4.3.0',
+    version: '4.2.0',
     meson_version: '>=1.1.0',
     default_options: ['cpp_std=c++20', 'warning_level=3'],
 )

src/crypto.cpp

@@ -1294,16 +1294,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);

src/dht_proxy_client.cpp

@@ -652,21 +652,7 @@ DhtProxyClient::getProxyInfos()
     if (logger_)
         logger_->debug("[proxy:client] [status] sending request");
 
-    // 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_);
+    auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
     queryProxyInfo(infoState, resolver, AF_INET);
     queryProxyInfo(infoState, resolver, AF_INET6);
     std::lock_guard l(resolverLock_);

src/dhtrunner.cpp

@@ -554,14 +554,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

@@ -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, const_cast<X509_NAME*>(issuer_name));
+    issuer = X509_find_by_subject(fullchain, issuer_name);
     return issuer;
 }
 
@@ -867,31 +867,12 @@ 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();
@@ -899,29 +880,9 @@ 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()
 {
@@ -944,68 +905,27 @@ filter(const std::vector<asio::ip::tcp::endpoint>& epts, sa_family_t family)
 }
 
 void
-Resolver::add_callback(ResolverCb cb, sa_family_t family, bool freshOnly)
+Resolver::add_callback(ResolverCb cb, sa_family_t family)
 {
     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(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));
+        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));
+                                           });
     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(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();
+                   [cb = std::move(cb),
+                    ec = ec_,
+                    endpoints = family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family)] { cb(ec, endpoints); });
     }
 }
 
@@ -1023,45 +943,30 @@ 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());
                                 }
-                                complete(ec,
-                                         std::vector<asio::ip::tcp::endpoint> {endpoints.begin(), endpoints.end()});
+                                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();
+                                }
                             });
 }
 
@@ -1539,12 +1444,7 @@ 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) {
@@ -1558,28 +1458,17 @@ Request::resolve_and_connect(bool freshOnly)
                 } 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 (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 {
+                            if (ec)
                                 sthis->terminate(asio::error::not_connected);
-                            }
+                            else
+                                sthis->post();
                         }
                     });
                 } else
                     this_.post();
             }
         },
-        family_,
-        freshOnly);
+        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/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/utils.cpp

@@ -245,18 +245,10 @@ from_system_time(system_clock::time_point t, time_point steady_now, system_clock
         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)
+    auto dt = t - system_now;
+    if (dt > system_clock::duration(0) and steady_now > time_point::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)
+    else if (dt < system_clock::duration(0) and steady_now < time_point::min() - dt)
         return time_point::min();
     return steady_now + dt;
 }
@@ -274,13 +266,12 @@ to_system_time(time_point t, time_point steady_now, system_clock::time_point sys
         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)
+    auto dt = t - steady_now;
+    if (dt > 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)
+    else if (dt < duration(0) and system_now <= system_clock::time_point::min() - dt)
         return system_clock::time_point::min();
-    return system_now + dt;
+    return system_now + std::chrono::duration_cast<system_clock::duration>(dt);
 }
 
 system_clock::time_point

tests/test_crypto.cpp

@@ -327,16 +327,6 @@ 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,7 +19,6 @@ 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);
@@ -69,10 +68,6 @@ 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

@@ -303,25 +303,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 +318,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

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

vcpkg.json

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