build/cmake: set SIMDUTF_CXX_STANDARD to match CMAKE_CXX_STANDARD

CMakeLists.txt

@@ -10,7 +10,7 @@ if(POLICY CMP0167)
 endif()
 
 set (opendht_VERSION_MAJOR 3)
-set (opendht_VERSION_MINOR 7.3)
+set (opendht_VERSION_MINOR 7.6)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -130,6 +130,7 @@ if (NOT MSVC)
                 message(STATUS "simdutf not found with pkgconfig")
                 if (OPENDHT_DOWNLOAD_DEPS)
                     message(STATUS "Fallback mode specified, downloading simdutf")
+                    set(SIMDUTF_CXX_STANDARD ${CMAKE_CXX_STANDARD} CACHE INTERNAL "C++ standard for simdutf downloader")
                     set(SIMDUTF_TESTS OFF CACHE INTERNAL "Disable tests for simdutf downloader")
                     set(SIMDUTF_TOOLS OFF CACHE INTERNAL "Disable tools for simdutf downloader")
                     FetchContent_Declare(simdutf
@@ -870,6 +871,12 @@ if (BUILD_TESTING)
         tests/test_networkengine.h
         tests/test_networkengine.cpp
     )
+    if (OPENDHT_PROXY_CLIENT)
+        list (APPEND test_FILES
+            tests/test_dhtproxy_client.h
+            tests/test_dhtproxy_client.cpp
+        )
+    endif()
     if (OPENDHT_TESTS_NETWORK)
         if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
             list (APPEND test_FILES

include/opendht/crypto.h

@@ -214,7 +214,7 @@ private:
     PrivateKey& operator=(const PrivateKey&) = delete;
     Blob decryptBloc(const uint8_t* src, size_t src_size) const;
 
-    mutable std::mutex publicKeyMutex_ {};
+    mutable std::mutex publicKeyMutex_;
     mutable std::shared_ptr<PublicKey> publicKey_ {};
 };
 
@@ -685,7 +685,7 @@ private:
 
     std::set<std::shared_ptr<RevocationList>, crlNumberCmp> revocation_lists;
 
-    mutable std::mutex publicKeyMutex_ {};
+    mutable std::mutex publicKeyMutex_;
     mutable std::shared_ptr<PublicKey> publicKey_ {};
 };
 

include/opendht/dht.h

@@ -411,6 +411,7 @@ private:
     std::vector<std::pair<std::string, std::string>> bootstrap_nodes {};
     std::chrono::steady_clock::duration bootstrap_period {BOOTSTRAP_PERIOD};
     Sp<Scheduler::Job> bootstrapJob {};
+    bool bootstrap_pending {false};
 
     std::map<InfoHash, Storage> store;
     std::map<SockAddr, StorageBucket, SockAddr::ipCmp> store_quota;
@@ -627,7 +628,7 @@ private:
     /* when we receive a listen request */
     net::RequestAnswer onListen(
         Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version = 0);
-    void onListenDone(const Sp<Node>& status, net::RequestAnswer& a, Sp<Search>& sr);
+    void onListenDone(const Sp<Node>& node, net::RequestAnswer& a, Sp<Search>& sr);
     /* when we receive an announce request */
     net::RequestAnswer onAnnounce(Sp<Node> node,
                                   const InfoHash& hash,

include/opendht/dhtrunner.h

@@ -579,16 +579,16 @@ private:
      */
     void resetDht();
 
-    mutable std::mutex dht_mtx {};
+    mutable std::mutex dht_mtx;
     std::thread dht_thread {};
     std::condition_variable cv {};
-    std::mutex sock_mtx {};
+    std::mutex sock_mtx;
     net::PacketList rcv {};
     decltype(rcv) rcv_free {};
 
     std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
     std::queue<std::function<void(SecureDht&)>> pending_ops {};
-    std::mutex storage_mtx {};
+    std::mutex storage_mtx;
 
     std::atomic<State> running {State::Idle};
     std::atomic_size_t ongoing_ops {0};

include/opendht/logger.h

@@ -196,7 +196,7 @@ private:
     const LogMethod logger_ = {};
     const std::string tag_ {};
     const std::string prefix_ {};
-    std::mutex children_mutex_ {};
+    std::mutex children_mutex_;
     std::vector<std::weak_ptr<Logger>> children_ {};
     bool enable_ {true};
 };

include/opendht/thread_pool.h

@@ -53,7 +53,7 @@ public:
     void detach();
 
 private:
-    std::mutex lock_ {};
+    std::mutex lock_;
     std::condition_variable cv_ {};
     std::queue<std::function<void()>> tasks_ {};
     std::vector<std::unique_ptr<std::thread>> threads_;
@@ -81,7 +81,7 @@ public:
 private:
     std::reference_wrapper<ThreadPool> threadPool_;
     const unsigned maxConcurrent_ {1};
-    std::mutex lock_ {};
+    std::mutex lock_;
     unsigned current_ {0};
     std::queue<std::function<void()>> tasks_ {};
 
@@ -114,7 +114,7 @@ public:
 private:
     struct SharedState
     {
-        std::mutex mtx {};
+        std::mutex mtx;
         std::condition_variable cv {};
         unsigned pendingTasks {0};
         unsigned ongoingTasks {0};

meson.build

@@ -5,7 +5,7 @@ project(
     'cython',
     license: 'MIT',
     license_files: 'LICENSE',
-    version: '3.7.3',
+    version: '3.7.6',
     meson_version: '>=1.1.0',
     default_options: ['cpp_std=c++20', 'warning_level=3'],
 )
@@ -281,7 +281,7 @@ if get_option('tests').enabled()
         'tests/tests_runner.cpp',
         dependencies: [opendht_dep, cppunit, jsoncpp, fmt, openssl, msgpack, gnutls],
     )
-    test('Crypto', test_crypto)
+    test('Crypto', test_crypto, timeout: 60)
 
     test_dhtrunner = executable(
         'test_dhtrunner',
@@ -289,7 +289,7 @@ if get_option('tests').enabled()
         'tests/tests_runner.cpp',
         dependencies: [opendht_dep, cppunit, jsoncpp, fmt, openssl, msgpack],
     )
-    test('DhtRunner', test_dhtrunner)
+    test('DhtRunner', test_dhtrunner, timeout: 60)
 
     test_threadpool = executable(
         'test_threadpool',

src/dht.cpp

@@ -35,7 +35,7 @@ Dht::updateStatus(sa_family_t af)
     auto& d = dht(af);
     auto old = d.status;
     d.status = d.getStatus(scheduler.time());
-    if (d.status == NodeStatus::Disconnected && bootstrapJob && !bootstrap_nodes.empty())
+    if (d.status == NodeStatus::Disconnected && bootstrap_pending && bootstrapJob && !bootstrap_nodes.empty())
         d.status = NodeStatus::Connecting;
     if (d.status != old) {
         auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
@@ -2271,6 +2271,7 @@ Dht::bootstrap()
         return;
     if (logger_)
         logger_->debug("[{}] Bootstraping", myid.to_view());
+    bootstrap_pending = false;
     for (const auto& boootstrap : bootstrap_nodes) {
         try {
             auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
@@ -2297,6 +2298,7 @@ void
 Dht::startBootstrap()
 {
     stopBootstrap();
+    bootstrap_pending = !bootstrap_nodes.empty();
     bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
 }
 
@@ -2304,6 +2306,7 @@ void
 Dht::stopBootstrap()
 {
     scheduler.cancel(bootstrapJob);
+    bootstrap_pending = false;
     bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
@@ -2532,6 +2535,16 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e)
                     continue;
                 n->token.clear();
                 n->last_get_reply = time_point::min();
+                for (auto& listen : n->listenStatus) {
+                    if (listen.second.req and listen.second.req->pending())
+                        n->node->cancelRequest(listen.second.req);
+                    listen.second.req = {};
+                    if (listen.second.refresh) {
+                        listen.second.refresh->cancel();
+                        listen.second.refresh = {};
+                    }
+                    listen.second.cache.onSynced(false);
+                }
                 searchSendGetValues(sr);
                 scheduler.edit(sr->nextSearchStep, scheduler.time());
                 break;
@@ -2694,7 +2707,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
 }
 
 void
-Dht::onListenDone(const Sp<Node>& /* node */, net::RequestAnswer& /* answer */, Sp<Search>& sr)
+Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {
     // if (logger_)
     //     logger_->d(sr->id, node->id, "[search %s] [node %s] Got listen confirmation",
@@ -2702,6 +2715,7 @@ Dht::onListenDone(const Sp<Node>& /* node */, net::RequestAnswer& /* answer */,
 
     if (not sr->done) {
         const auto& now = scheduler.time();
+        sr->insertNode(node, now, answer.ntoken);
         searchSendGetValues(sr);
         scheduler.edit(sr->nextSearchStep, now);
     }

src/dht_proxy_client.cpp

@@ -1376,7 +1376,7 @@ DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
 
     restinio::http_request_header_t header;
     header.method(restinio::http_method_subscribe());
-    header.request_target("/" + key.toString());
+    header.request_target("/key/" + key.toString());
     if (!listener.refreshSubscriberTimer) {
         listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
     }
@@ -1422,17 +1422,20 @@ void
 DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token)
 {
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
-    std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
-    if (deviceKey_ != token) {
+    {
+        std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
+        if (deviceKey_ == token)
+            return;
         deviceKey_ = token;
-        if (statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected) {
-            if (logger_)
-                logger_->debug("[proxy:client] [push] token changed, resubscribing");
-            for (auto& search : searches_) {
-                for (auto& listener : search.second.listeners) {
-                    resubscribe(search.first, listener.first, listener.second);
-                }
-            }
+        if (!(statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected))
+            return;
+    }
+    if (logger_)
+        logger_->debug("[proxy:client] [push] token changed, resubscribing");
+    std::lock_guard<std::mutex> lock(searchLock_);
+    for (auto& search : searches_) {
+        for (auto& listener : search.second.listeners) {
+            resubscribe(search.first, listener.first, listener.second);
         }
     }
 #endif

src/network_engine.cpp

@@ -294,8 +294,10 @@ void
 NetworkEngine::clear()
 {
     for (auto& request : requests) {
-        request.second->cancel();
-        request.second->node->setExpired();
+        if (request.second->node)
+            request.second->node->setExpired();
+        else
+            request.second->setExpired();
     }
     requests.clear();
     partial_messages.clear();

src/network_utils.cpp

@@ -9,7 +9,7 @@
 #include <string>
 #include <cstring>
 #define close(x)                  closesocket(x)
-#define write(s, b, f)            send(s, b, (int) strlen(b), 0)
+#define write(s, b, f)            send(s, b, (int) (f), 0)
 #define _poll(fds, nfds, timeout) WSAPoll(fds, nfds, timeout)
 #else
 #include <fcntl.h>

tests/test_dhtproxy_client.cpp

@@ -0,0 +1,93 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+
+#include "test_dhtproxy_client.h"
+
+#include <any>
+#include <mutex>
+
+#if defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wkeyword-macro"
+#elif defined(__GNUC__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wpedantic"
+#endif
+#define private public
+#include "opendht/http.h"
+#include "opendht/dht_proxy_client.h"
+#undef private
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#elif defined(__GNUC__)
+#pragma GCC diagnostic pop
+#endif
+
+#include "../src/op_cache.cpp"
+#include "../src/dht_proxy_client.cpp"
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyClientTester);
+
+using namespace dht;
+
+void
+DhtProxyClientTester::testResubscribeUsesKeyRoute()
+{
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    DhtProxyClient client({}, {}, [] {}, "http://127.0.0.1:8080", "OpenDHT-Test", "client-id", "push-token");
+
+    DhtProxyClient::Listener listener {OpValueCache([](const std::vector<Sp<Value>>&, bool) { return true; })};
+    listener.opstate = std::make_shared<DhtProxyClient::OperationState>();
+    listener.cb = [](const std::vector<Sp<Value>>&, bool, system_clock::time_point) {
+        return true;
+    };
+
+    auto key = InfoHash::get("proxy-client-resubscribe");
+    client.resubscribe(key, 1, listener);
+
+    CPPUNIT_ASSERT(listener.request);
+    CPPUNIT_ASSERT_EQUAL(std::string("/key/") + key.toString(), std::string(listener.request->header_.request_target()));
+    CPPUNIT_ASSERT(listener.request->body_.find("\"refresh\"") != std::string::npos);
+    CPPUNIT_ASSERT(listener.request->body_.find("true") != std::string::npos);
+#else
+    CPPUNIT_ASSERT(true);
+#endif
+}
+
+void
+DhtProxyClientTester::testSetPushNotificationTokenResubscribesWithNewToken()
+{
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    DhtProxyClient client({}, {}, [] {}, "http://127.0.0.1:8080", "OpenDHT-Test", "client-id", "old-token");
+
+    client.statusIpv4_ = NodeStatus::Connected;
+
+    auto key = InfoHash::get("proxy-client-token-rotation");
+    auto& search = client.searches_[key];
+    auto [it, inserted] = search.listeners.emplace(std::piecewise_construct,
+                                                   std::forward_as_tuple(1),
+                                                   std::forward_as_tuple(OpValueCache(
+                                                       [](const std::vector<Sp<Value>>&, bool) { return true; })));
+    CPPUNIT_ASSERT(inserted);
+
+    auto& listener = it->second;
+    listener.opstate = std::make_shared<DhtProxyClient::OperationState>();
+    listener.cb = [](const std::vector<Sp<Value>>&, bool, system_clock::time_point) {
+        return true;
+    };
+
+    client.setPushNotificationToken("new-token");
+
+    CPPUNIT_ASSERT(listener.request);
+    CPPUNIT_ASSERT_EQUAL(std::string("/key/") + key.toString(), std::string(listener.request->header_.request_target()));
+    CPPUNIT_ASSERT(listener.request->body_.find("new-token") != std::string::npos);
+    CPPUNIT_ASSERT(listener.request->body_.find("old-token") == std::string::npos);
+    CPPUNIT_ASSERT(listener.request->body_.find("\"refresh\"") != std::string::npos);
+    CPPUNIT_ASSERT(listener.request->body_.find("true") != std::string::npos);
+#else
+    CPPUNIT_ASSERT(true);
+#endif
+}
+
+} // namespace test

tests/test_dhtproxy_client.h

@@ -0,0 +1,22 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+#pragma once
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class DhtProxyClientTester : public CppUnit::TestFixture
+{
+    CPPUNIT_TEST_SUITE(DhtProxyClientTester);
+    CPPUNIT_TEST(testResubscribeUsesKeyRoute);
+    CPPUNIT_TEST(testSetPushNotificationTokenResubscribesWithNewToken);
+    CPPUNIT_TEST_SUITE_END();
+
+public:
+    void testResubscribeUsesKeyRoute();
+    void testSetPushNotificationTokenResubscribesWithNewToken();
+};
+
+} // namespace test

tests/test_dhtrunner.cpp

@@ -6,6 +6,7 @@
 #include <opendht/thread_pool.h>
 
 #include <chrono>
+#include <future>
 #include <mutex>
 #include <condition_variable>
 using namespace std::chrono_literals;
@@ -14,6 +15,14 @@ using namespace std::literals;
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
 
+template<typename T>
+T
+getFutureValue(std::future<T> future, std::chrono::steady_clock::duration timeout = 30s)
+{
+    CPPUNIT_ASSERT(std::future_status::ready == future.wait_for(timeout));
+    return future.get();
+}
+
 void
 DhtRunnerTester::setUp()
 {
@@ -72,9 +81,10 @@ DhtRunnerTester::testGetPut()
     dht::Value val {"hey"};
     auto val_data = val.data;
     std::promise<bool> p;
+    auto future = p.get_future();
     node2.put(key, std::move(val), [&](bool ok) { p.set_value(ok); });
-    CPPUNIT_ASSERT(p.get_future().get());
-    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(getFutureValue(std::move(future)));
+    auto vals = getFutureValue(node1.get(key));
     CPPUNIT_ASSERT(not vals.empty());
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
@@ -88,13 +98,15 @@ DhtRunnerTester::testPutDuplicate()
     auto val_data = val->data;
     std::promise<bool> p1;
     std::promise<bool> p2;
+    auto p1future = p1.get_future();
+    auto p2future = p2.get_future();
     node2.put(key, val, [&](bool ok) { p1.set_value(ok); });
     node2.put(key, val, [&](bool ok) { p2.set_value(ok); });
-    auto p1ret = p1.get_future().get();
-    auto p2ret = p2.get_future().get();
+    auto p1ret = getFutureValue(std::move(p1future));
+    auto p2ret = getFutureValue(std::move(p2future));
     CPPUNIT_ASSERT(p1ret);
     CPPUNIT_ASSERT(p2ret);
-    auto vals = node1.get(key).get();
+    auto vals = getFutureValue(node1.get(key));
     CPPUNIT_ASSERT(not vals.empty());
     CPPUNIT_ASSERT(vals.size() == 1);
     CPPUNIT_ASSERT(vals.front()->data == val_data);
@@ -112,13 +124,15 @@ DhtRunnerTester::testPutOverride()
     auto val_data = val2->data;
     std::promise<bool> p1;
     std::promise<bool> p2;
+    auto p1future = p1.get_future();
+    auto p2future = p2.get_future();
     node2.put(key, val, [&](bool ok) { p1.set_value(ok); });
     node2.put(key, val2, [&](bool ok) { p2.set_value(ok); });
-    auto p1ret = p1.get_future().get();
-    auto p2ret = p2.get_future().get();
+    auto p1ret = getFutureValue(std::move(p1future));
+    auto p2ret = getFutureValue(std::move(p2future));
     CPPUNIT_ASSERT(!p1ret);
     CPPUNIT_ASSERT(p2ret);
-    auto vals = node1.get(key).get();
+    auto vals = getFutureValue(node1.get(key));
     CPPUNIT_ASSERT(not vals.empty());
     CPPUNIT_ASSERT(vals.size() == 1);
     CPPUNIT_ASSERT(vals.front()->data == val_data);
@@ -153,8 +167,9 @@ DhtRunnerTester::testImportValuesPreservesRemoteQuota()
     auto value2 = std::make_shared<dht::Value>(std::string(1024, 'b'));
 
     std::promise<bool> p1;
+    auto p1future = p1.get_future();
     node2.put(key1, value1, [&](bool ok) { p1.set_value(ok); });
-    CPPUNIT_ASSERT(p1.get_future().get());
+    CPPUNIT_ASSERT(getFutureValue(std::move(p1future)));
 
     std::this_thread::sleep_for(2s);
     auto exported = node1.exportValues();
@@ -179,13 +194,14 @@ DhtRunnerTester::testImportValuesPreservesRemoteQuota()
     std::this_thread::sleep_for(500ms);
 
     std::promise<bool> p2;
+    auto p2future = p2.get_future();
     node2.put(key2, value2, [&](bool ok) { p2.set_value(ok); });
-    CPPUNIT_ASSERT(p2.get_future().get());
+    CPPUNIT_ASSERT(getFutureValue(std::move(p2future)));
 
     std::this_thread::sleep_for(2s);
 
-    auto vals1 = restored.get(key1).get();
-    auto vals2 = restored.get(key2).get();
+    auto vals1 = getFutureValue(restored.get(key1));
+    auto vals2 = getFutureValue(restored.get(key2));
     CPPUNIT_ASSERT(vals1.size() + vals2.size() <= 1);
 
     restored.shutdown();
@@ -213,7 +229,9 @@ DhtRunnerTester::testImportValuesPreservesStoredExpiration()
     pk.pack_bin(0);
     pk.pack(expiration.time_since_epoch().count());
 
-    node1.importValues({dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}});
+    node1.importValues({
+        dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
+    });
 
     auto exported = node1.exportValues();
     CPPUNIT_ASSERT(!exported.empty());
@@ -264,8 +282,9 @@ DhtRunnerTester::testBootstrapSetsConnectingState()
 
     dht::DhtRunner::Context context;
     context.statusChangedCallback = [&](dht::NodeStatus status4, dht::NodeStatus status6) {
-        std::lock_guard<std::mutex> lock(mutex);
-        sawConnecting = sawConnecting || status4 == dht::NodeStatus::Connecting || status6 == dht::NodeStatus::Connecting;
+        std::lock_guard lock(mutex);
+        sawConnecting = sawConnecting || status4 == dht::NodeStatus::Connecting
+                        || status6 == dht::NodeStatus::Connecting;
         cv.notify_all();
     };
 
@@ -278,10 +297,12 @@ DhtRunnerTester::testBootstrapSetsConnectingState()
     if (bound.isUnspecified())
         bound.setLoopback();
 
-    clientNode.bootstrap("127.0.0.1", std::to_string(bound.getPort()).c_str());
+    clientNode.bootstrap("127.0.0.1", std::to_string(bound.getPort()));
 
-    std::unique_lock<std::mutex> lock(mutex);
-    CPPUNIT_ASSERT(cv.wait_for(lock, 5s, [&] { return sawConnecting; }));
+    {
+        std::unique_lock lock(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lock, 5s, [&] { return sawConnecting; }));
+    }
 
     bootstrapNode.shutdown();
     bootstrapNode.join();
@@ -315,10 +336,13 @@ DhtRunnerTester::testBootstrapThenPutNoRace()
     clientNode.bootstrap(bound);
 
     std::promise<bool> putDone;
-    clientNode.put(dht::InfoHash::get("bootstrap-then-put"), dht::Value {"value"}, [&](bool ok) { putDone.set_value(ok); });
-    CPPUNIT_ASSERT(putDone.get_future().get());
+    auto putFuture = putDone.get_future();
+    clientNode.put(dht::InfoHash::get("bootstrap-then-put"), dht::Value {"value"}, [&](bool ok) {
+        putDone.set_value(ok);
+    });
+    CPPUNIT_ASSERT(getFutureValue(std::move(putFuture)));
 
-    auto vals = bootstrapNode.get(dht::InfoHash::get("bootstrap-then-put")).get();
+    auto vals = getFutureValue(bootstrapNode.get(dht::InfoHash::get("bootstrap-then-put")));
     CPPUNIT_ASSERT(!vals.empty());
 
     bootstrapNode.shutdown();
@@ -327,6 +351,118 @@ DhtRunnerTester::testBootstrapThenPutNoRace()
     clientNode.join();
 }
 
+void
+DhtRunnerTester::testBootstrapMissingNodeThenPutFails()
+{
+    node1.shutdown();
+    node1.join();
+    node2.shutdown();
+    node2.join();
+
+    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;
+
+    auto getUnusedPort = [&]() {
+        dht::DhtRunner tmpNode;
+        tmpNode.run(0, config);
+        auto port = tmpNode.getBoundPort();
+        tmpNode.shutdown();
+        tmpNode.join();
+        return port;
+    };
+
+    auto missingPort = getUnusedPort();
+
+    std::mutex mutex;
+    std::condition_variable cv;
+    bool sawConnecting {false};
+    bool sawDisconnectedAfterConnecting {false};
+    bool putReturnedBeforeDisconnect {false};
+
+    dht::DhtRunner::Context context;
+    context.statusChangedCallback = [&](dht::NodeStatus status4, dht::NodeStatus status6) {
+        std::lock_guard lock(mutex);
+        if (status4 == dht::NodeStatus::Connecting || status6 == dht::NodeStatus::Connecting)
+            sawConnecting = true;
+        if (sawConnecting && status4 == dht::NodeStatus::Disconnected && status6 == dht::NodeStatus::Disconnected)
+            sawDisconnectedAfterConnecting = true;
+        cv.notify_all();
+    };
+
+    dht::DhtRunner clientNode;
+    clientNode.run(0, config, std::move(context));
+    while (clientNode.getBoundPort() == missingPort)
+        missingPort = getUnusedPort();
+
+    std::promise<bool> putDone;
+    auto putFuture = putDone.get_future();
+
+    clientNode.bootstrap("127.0.0.1", std::to_string(missingPort));
+    clientNode.put(dht::InfoHash::get("bootstrap-missing-put"), dht::Value {"value"}, [&](bool ok) {
+        std::lock_guard lock(mutex);
+        putReturnedBeforeDisconnect = !sawDisconnectedAfterConnecting;
+        putDone.set_value(ok);
+        cv.notify_all();
+    });
+
+    {
+        std::unique_lock lock(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lock, 10s, [&] { return sawConnecting; }));
+        CPPUNIT_ASSERT(cv.wait_for(lock, 20s, [&] { return sawDisconnectedAfterConnecting; }));
+    }
+    CPPUNIT_ASSERT(std::future_status::ready == putFuture.wait_for(20s));
+    CPPUNIT_ASSERT(!putFuture.get());
+    CPPUNIT_ASSERT(!putReturnedBeforeDisconnect);
+
+    clientNode.shutdown();
+    clientNode.join();
+}
+
+void
+DhtRunnerTester::testShutdownCompletesWithPendingPut()
+{
+    node1.shutdown();
+    node1.join();
+    node2.shutdown();
+    node2.join();
+
+    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;
+
+    auto getUnusedPort = [&]() {
+        dht::DhtRunner tmpNode;
+        tmpNode.run(0, config);
+        auto port = tmpNode.getBoundPort();
+        tmpNode.shutdown();
+        tmpNode.join();
+        return port;
+    };
+
+    auto missingPort = getUnusedPort();
+
+    dht::DhtRunner clientNode;
+    clientNode.run(0, config);
+    while (clientNode.getBoundPort() == missingPort)
+        missingPort = getUnusedPort();
+
+    clientNode.bootstrap("127.0.0.1", std::to_string(missingPort));
+    clientNode.put(dht::InfoHash::get("shutdown-pending-put"), dht::Value {"value"}, [](bool) {});
+
+    std::promise<void> shutdownDone;
+    auto shutdownFuture = shutdownDone.get_future();
+    clientNode.shutdown([&]() { shutdownDone.set_value(); });
+
+    CPPUNIT_ASSERT(std::future_status::ready == shutdownFuture.wait_for(5s));
+
+    clientNode.join();
+}
+
 void
 DhtRunnerTester::testListen()
 {
@@ -426,9 +562,9 @@ DhtRunnerTester::testListen()
     CPPUNIT_ASSERT(ftokenc.valid());
     CPPUNIT_ASSERT(ftokend.valid());
 
-    auto tokena = ftokena.get();
-    auto tokenc = ftokenc.get();
-    auto tokend = ftokend.get();
+    auto tokena = getFutureValue(std::move(ftokena));
+    auto tokenc = getFutureValue(std::move(ftokenc));
+    auto tokend = getFutureValue(std::move(ftokend));
     // tokenb might be 0 since the callback returns false.
 
     CPPUNIT_ASSERT(tokena);
@@ -468,7 +604,7 @@ DhtRunnerTester::testIdOps()
     };
 
     node2.join();
-    node2.run(42232, config2, std::move(context2));
+    node2.run(0, config2, std::move(context2));
     auto bound = node1.getBound();
     if (bound.isUnspecified())
         bound.setLoopback();
@@ -502,7 +638,7 @@ DhtRunnerTester::testIdOps()
 
     config2.dht_config.id = dht::crypto::generateIdentity();
     node1.join();
-    node1.run(42222, config2, std::move(context1));
+    node1.run(0, config2, std::move(context1));
     bound = node2.getBound();
     if (bound.isUnspecified())
         bound.setLoopback();
@@ -634,7 +770,7 @@ DhtRunnerTester::testListenLotOfBytes()
         CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return valueCount == N; }));
     }
 
-    node3.cancelListen(foo, ftokenfoo.get());
+    node3.cancelListen(foo, getFutureValue(std::move(ftokenfoo)));
 }
 
 void

tests/test_dhtrunner.h

@@ -24,6 +24,8 @@ class DhtRunnerTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testImportValuesPreservesStoredExpiration);
     CPPUNIT_TEST(testBootstrapSetsConnectingState);
     CPPUNIT_TEST(testBootstrapThenPutNoRace);
+    CPPUNIT_TEST(testBootstrapMissingNodeThenPutFails);
+    CPPUNIT_TEST(testShutdownCompletesWithPendingPut);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -66,6 +68,8 @@ public:
     void testImportValuesPreservesStoredExpiration();
     void testBootstrapSetsConnectingState();
     void testBootstrapThenPutNoRace();
+    void testBootstrapMissingNodeThenPutFails();
+    void testShutdownCompletesWithPendingPut();
     /**
      * Test listen method with lot of datas
      */

tests/test_networkengine.cpp

@@ -3,6 +3,38 @@
 
 #include "test_networkengine.h"
 
+#ifdef _MSC_VER
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(NetworkEngineTester);
+
+void
+NetworkEngineTester::setUp()
+{}
+
+void
+NetworkEngineTester::tearDown()
+{}
+
+void
+NetworkEngineTester::testDisabledOnMsvc()
+{
+    CPPUNIT_ASSERT(true);
+}
+
+} // namespace test
+
+#else
+
+#include <any>
+#include <mutex>
+
+#ifdef _WIN32
+#ifdef opendht_EXPORTS
+#undef opendht_EXPORTS
+#endif
+#endif
+
 #include "opendht/node.h"
 #include "opendht/network_engine.h"
 #include "opendht/network_utils.h"
@@ -29,6 +61,9 @@
 #include "../src/search.h"
 
 // Hack to test internal classes
+#include "../src/op_cache.cpp"
+#include "../src/routing_table.cpp"
+#include "../src/dht.cpp"
 #include "../src/node.cpp"
 #include "../src/node_cache.cpp"
 #include "../src/network_engine.cpp"
@@ -352,6 +387,51 @@ NetworkEngineTester::testListenConfirmationCarriesToken()
     CPPUNIT_ASSERT(reply.token == responseToken);
 }
 
+void
+NetworkEngineTester::testListenConfirmationUpdatesSearchNodeToken()
+{
+    Config config {};
+    config.max_req_per_sec = -1;
+    config.max_peer_req_per_sec = -1;
+
+    auto rd = std::make_unique<std::mt19937_64>(6);
+    Dht localDht(std::make_unique<TestDatagramSocket>(), config, {}, std::move(rd));
+
+    auto node = std::make_shared<Node>(InfoHash::getRandom(localDht.rd),
+                                       makeIPv4("127.0.0.2", 5006),
+                                       localDht.rd,
+                                       false);
+    auto sr = std::make_shared<Dht::Search>();
+    auto query = std::make_shared<Query>();
+    ValueCallback valueCallback = [](const std::vector<std::shared_ptr<Value>>&, bool) {
+        return true;
+    };
+
+    sr->id = InfoHash::getRandom(localDht.rd);
+    sr->af = AF_INET;
+    sr->nextSearchStep = localDht.scheduler.add(time_point::max(), [] {});
+    sr->listeners.emplace(1, Dht::Search::SearchListener {query, valueCallback, {}});
+
+    auto searchNode = std::make_unique<Dht::SearchNode>(node);
+    auto socketId = node->openSocket([](const Sp<Node>&, net::RequestAnswer&&) {});
+    searchNode->listenStatus.emplace(std::piecewise_construct,
+                                     std::forward_as_tuple(query),
+                                     std::forward_as_tuple(valueCallback, dht::SyncCallback {}, socketId));
+    auto* searchNodePtr = sr->nodes.emplace_back(std::move(searchNode)).get();
+
+    CPPUNIT_ASSERT(searchNodePtr->token.empty());
+    CPPUNIT_ASSERT(!searchNodePtr->isSynced(localDht.scheduler.time()));
+
+    net::RequestAnswer answer {};
+    answer.ntoken = Blob {0xaa, 0xbb, 0xcc, 0xdd};
+
+    localDht.onListenDone(node, answer, sr);
+
+    CPPUNIT_ASSERT(searchNodePtr->token == answer.ntoken);
+    CPPUNIT_ASSERT(searchNodePtr->last_get_reply == localDht.scheduler.time());
+    CPPUNIT_ASSERT(searchNodePtr->isSynced(localDht.scheduler.time()));
+}
+
 void
 NetworkEngineTester::testListenReopensSocketAfterNodeExpiration()
 {
@@ -404,4 +484,66 @@ NetworkEngineTester::testListenReopensSocketAfterNodeExpiration()
     CPPUNIT_ASSERT(node->getSocket(listenIt->second.socketId));
 }
 
-} // namespace test
+void
+NetworkEngineTester::testUnauthorizedListenFlushClearsListenState()
+{
+    Config config {};
+    config.max_req_per_sec = -1;
+    config.max_peer_req_per_sec = -1;
+
+    auto rd = std::make_unique<std::mt19937_64>(5);
+    Dht localDht(std::make_unique<TestDatagramSocket>(), config, {}, std::move(rd));
+
+    auto node = std::make_shared<Node>(InfoHash::getRandom(localDht.rd),
+                                       makeIPv4("127.0.0.2", 5005),
+                                       localDht.rd,
+                                       false);
+    auto sr = std::make_shared<Dht::Search>();
+    auto query = std::make_shared<Query>();
+    ValueCallback valueCallback = [](const std::vector<std::shared_ptr<Value>>&, bool) {
+        return true;
+    };
+    dht::SyncCallback syncCallback {};
+
+    sr->id = InfoHash::getRandom(localDht.rd);
+    sr->af = AF_INET;
+
+    auto searchNode = std::make_unique<Dht::SearchNode>(node);
+    auto socketId = node->openSocket([](const Sp<Node>&, net::RequestAnswer&&) {});
+    auto listenIt = searchNode->listenStatus
+                        .emplace(std::piecewise_construct,
+                                 std::forward_as_tuple(query),
+                                 std::forward_as_tuple(valueCallback, syncCallback, socketId))
+                        .first;
+    listenIt->second.req = std::make_shared<net::Request>();
+    listenIt->second.refresh = localDht.scheduler.add(localDht.scheduler.time() + std::chrono::seconds(1), [] {});
+    searchNode->token = Blob {0x10, 0x11, 0x12};
+    searchNode->last_get_reply = localDht.scheduler.time();
+    auto* searchNodePtr = sr->nodes.emplace_back(std::move(searchNode)).get();
+
+    localDht.searches(AF_INET).emplace(sr->id, sr);
+
+    auto failingReq = std::make_shared<net::Request>(
+        MessageType::Listen,
+        node->getNewTid(),
+        node,
+        Blob {},
+        [](const net::Request&, ParsedMessage&&) {},
+        [](const net::Request&, bool) {});
+
+    localDht.onError(failingReq,
+                     net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED,
+                                                net::DhtProtocolException::LISTEN_WRONG_TOKEN});
+
+    CPPUNIT_ASSERT(searchNodePtr->token.empty());
+    CPPUNIT_ASSERT(searchNodePtr->last_get_reply == time_point::min());
+
+    auto listenStatusIt = searchNodePtr->listenStatus.find(query);
+    CPPUNIT_ASSERT(listenStatusIt != searchNodePtr->listenStatus.end());
+    CPPUNIT_ASSERT(!listenStatusIt->second.req);
+    CPPUNIT_ASSERT(!listenStatusIt->second.refresh);
+}
+
+} // namespace test
+
+#endif

tests/test_networkengine.h

@@ -10,22 +10,34 @@ namespace test {
 class NetworkEngineTester : public CppUnit::TestFixture
 {
     CPPUNIT_TEST_SUITE(NetworkEngineTester);
+#ifdef _MSC_VER
+    CPPUNIT_TEST(testDisabledOnMsvc);
+#else
     CPPUNIT_TEST(testIgnoresUnknownPartialData);
     CPPUNIT_TEST(testCompletesPartialSessionFromSameSource);
     CPPUNIT_TEST(testKeepsSessionForWrongSourceFragment);
     CPPUNIT_TEST(testListenConfirmationCarriesToken);
+    CPPUNIT_TEST(testListenConfirmationUpdatesSearchNodeToken);
     CPPUNIT_TEST(testListenReopensSocketAfterNodeExpiration);
+    CPPUNIT_TEST(testUnauthorizedListenFlushClearsListenState);
+#endif
     CPPUNIT_TEST_SUITE_END();
 
 public:
     void setUp();
     void tearDown();
 
+#ifdef _MSC_VER
+    void testDisabledOnMsvc();
+#else
     void testIgnoresUnknownPartialData();
     void testCompletesPartialSessionFromSameSource();
     void testKeepsSessionForWrongSourceFragment();
     void testListenConfirmationCarriesToken();
+    void testListenConfirmationUpdatesSearchNodeToken();
     void testListenReopensSocketAfterNodeExpiration();
+    void testUnauthorizedListenFlushClearsListenState();
+#endif
 };
 
 } // namespace test

vcpkg.json

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