build/cmake: set SIMDUTF_CXX_STANDARD to match CMAKE_CXX_STANDARD

AGENTS.md

@@ -32,14 +32,21 @@ meson setup build . && cd build && ninja
 ctest --output-on-failure --test-dir build
 
 # Single test executable
-./build/test_dhtrunner
-./build/test_crypto
-./build/test_value
-./build/test_infohash
-./build/test_storage
-./build/test_threadpool
+./build/test_testName
 ```
-Network-dependent tests (`test_http`, `test_dhtproxy`, `test_peerdiscovery`) require their corresponding CMake features enabled at configure time.
+
+## dhtnode
+
+The `dhtnode` CLI tool is a simple DHT node implementation using `DhtRunner`.
+
+---
+
+## Commit
+
+After implementing a feature or fixing a bug, commit your changes in clean, atomic commits with descriptive messages with format:
+`topic: short description`
+
+If any test are added or modified, commit them separately before the fix (if any).
 
 ---
 
@@ -65,7 +72,3 @@ DhtRunner          (public API — thread-safe, manages lifecycle & I/O threads)
 Optional components (`DhtProxyServer`, `DhtProxyClient`, `PeerDiscovery`) are compiled in only when the corresponding CMake options are enabled.
 
 ---
-
-## dhtnode
-
-The `dhtnode` CLI tool is a simple DHT node implementation using `DhtRunner`.

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/callbacks.h

@@ -23,7 +23,7 @@ struct Node;
  */
 enum class NodeStatus {
     Disconnected, // 0 nodes
-    Connecting,   // 1+ nodes
+    Connecting,   // bootstrap pending or 1+ nodes
     Connected     // 1+ good nodes
 };
 

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;
@@ -464,7 +465,8 @@ private:
                       const Sp<Value>& value,
                       time_point created,
                       const SockAddr& sa = {},
-                      bool permanent = false);
+                      bool permanent = false,
+                      time_point expiration = time_point::min());
     bool storageRefresh(const InfoHash& id, Value::Id vid);
     void expireStore();
     void expireStorage(InfoHash h);
@@ -626,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/network_engine.h

@@ -547,7 +547,7 @@ private:
     std::pair<Blob, Blob> bufferNodes(
         sa_family_t af, const InfoHash& id, want_t want, std::vector<Sp<Node>>& nodes, std::vector<Sp<Node>>& nodes6);
     /* answer to a listen request */
-    void sendListenConfirmation(const SockAddr& addr, Tid tid);
+    void sendListenConfirmation(const SockAddr& addr, Tid tid, const Blob& token = {});
     /* answer to put request */
     void sendValueAnnounced(const SockAddr& addr, Tid, Value::Id);
     /* answer in case of error */

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

include/opendht/utils.h

@@ -126,6 +126,8 @@ print_time_relative(TimePoint now, TimePoint d)
     return (d > now) ? concat("in "sv, print_duration(d - now)) : concat(print_duration(now - d), " ago"sv);
 }
 
+OPENDHT_PUBLIC std::string printByteCount(size_t bytes);
+
 template<typename Duration = duration>
 class uniform_duration_distribution : public std::uniform_int_distribution<typename Duration::rep>
 {

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,6 +35,8 @@ 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 && bootstrap_pending && bootstrapJob && !bootstrap_nodes.empty())
+        d.status = NodeStatus::Connecting;
     if (d.status != old) {
         auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
         if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
@@ -654,6 +656,19 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
     for (const auto& l : sr->listeners) {
         const auto& query = l.second.query;
 
+        auto openListenSocket = [this, ws, query, node = n.node] {
+            return node->openSocket([this, ws, query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
+                /* on new values */
+                if (auto sr = ws.lock()) {
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                    sr->insertNode(node, scheduler.time(), answer.ntoken);
+                    if (auto sn = sr->getNode(node)) {
+                        sn->onValues(query, std::move(answer), types, scheduler);
+                    }
+                }
+            });
+        };
+
         auto r = n.listenStatus.find(query);
         if (n.getListenTime(r, listenExp) > scheduler.time())
             continue;
@@ -665,21 +680,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             r = n.listenStatus
                     .emplace(std::piecewise_construct,
                              std::forward_as_tuple(query),
-                             std::forward_as_tuple(l.second.get_cb,
-                                                   l.second.sync_cb,
-                                                   n.node->openSocket([this,
-                                                                       ws,
-                                                                       query](const Sp<Node>& node,
-                                                                              net::RequestAnswer&& answer) mutable {
-                                                       /* on new values */
-                                                       if (auto sr = ws.lock()) {
-                                                           scheduler.edit(sr->nextSearchStep, scheduler.time());
-                                                           sr->insertNode(node, scheduler.time(), answer.ntoken);
-                                                           if (auto sn = sr->getNode(node)) {
-                                                               sn->onValues(query, std::move(answer), types, scheduler);
-                                                           }
-                                                       }
-                                                   })))
+                             std::forward_as_tuple(l.second.get_cb, l.second.sync_cb, openListenSocket()))
                     .first;
             r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this, ws, query, node = n.node] {
                 if (auto sr = ws.lock()) {
@@ -688,6 +689,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                     }
                 }
             });
+        } else if (not n.node->getSocket(r->second.socketId)) {
+            r->second.socketId = openListenSocket();
         }
         auto new_req = network_engine.sendListen(
             n.node,
@@ -1417,11 +1420,17 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
 }
 
 bool
-Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa, bool permanent)
+Dht::storageStore(const InfoHash& id,
+                  const Sp<Value>& value,
+                  time_point created,
+                  const SockAddr& sa,
+                  bool permanent,
+                  time_point expiration)
 {
     const auto& now = scheduler.time();
     created = std::min(created, now);
-    auto expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
+    if (expiration == time_point::min())
+        expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
     if (expiration < now)
         return false;
 
@@ -1438,7 +1447,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
     StorageBucket* store_bucket {nullptr};
     bool is_local = !sa;
     if (sa)
-        store_bucket = &store_quota[sa];
+        store_bucket = &store_quota.try_emplace(sa, sa).first->second;
     else
         store_bucket = local_store_quota.get();
 
@@ -1844,15 +1853,18 @@ std::string
 Dht::getStorageLog() const
 {
     std::ostringstream out;
-    for (const auto& s : store)
-        out << printStorageLog(s);
+    for (const auto& s : store) {
+        const auto& st = s.second;
+        out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
+            << printByteCount(st.totalSize()) << "\n";
+    }
     out << std::endl << std::endl;
     std::multimap<size_t, const SockAddr*> q_map;
     for (const auto& ip : store_quota)
         if (ip.second.size())
             q_map.emplace(ip.second.size(), &ip.first);
     for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
-        out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
+        out << "IP " << ip->second->toString() << " uses " << printByteCount(ip->first) << std::endl;
     out << std::endl;
     out << "Total " << store.size() << " storages, " << total_values << " values (";
     if (total_store_size < 1024)
@@ -1875,11 +1887,8 @@ std::string
 Dht::getStorageLog(const InfoHash& h) const
 {
     auto s = store.find(h);
-    if (s == store.end()) {
-        std::ostringstream out;
-        out << "Storage " << h << " empty" << std::endl;
-        return out.str();
-    }
+    if (s == store.end())
+        return fmt::format("Storage {} empty\n", h);
     return printStorageLog(*s);
 }
 
@@ -1889,8 +1898,14 @@ Dht::printStorageLog(const decltype(store)::value_type& s) const
     std::ostringstream out;
     using namespace std::chrono;
     const auto& st = s.second;
-    out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values ("
-        << st.totalSize() << " bytes)" << std::endl;
+    const auto& now = scheduler.time();
+    out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
+        << printByteCount(st.totalSize()) << std::endl;
+    for (const auto& v : st.getValues()) {
+        out << "   Value " << v.data->id << " size: " << printByteCount(v.data->size())
+            << " created: " << print_time_relative(now, v.created)
+            << " expires: " << print_time_relative(now, v.expiration) << std::endl;
+    }
     if (not st.local_listeners.empty())
         out << "   " << st.local_listeners.size() << " local listeners" << std::endl;
     for (const auto& node_listeners : st.listeners) {
@@ -2223,6 +2238,18 @@ void
 Dht::onConnected()
 {
     stopBootstrap();
+
+    auto restartSearches = [this](std::map<InfoHash, Sp<Search>>& searches) {
+        const auto& now = scheduler.time();
+        for (auto& srp : searches) {
+            auto& sr = *srp.second;
+            if (sr.done)
+                continue;
+            scheduler.edit(sr.nextSearchStep, now);
+        }
+    };
+    restartSearches(dht4.searches);
+    restartSearches(dht6.searches);
     auto callbacks = std::move(onConnectCallbacks_);
     while (not callbacks.empty()) {
         callbacks.front()();
@@ -2244,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);
@@ -2270,6 +2298,7 @@ void
 Dht::startBootstrap()
 {
     stopBootstrap();
+    bootstrap_pending = !bootstrap_nodes.empty();
     bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
 }
 
@@ -2277,6 +2306,7 @@ void
 Dht::stopBootstrap()
 {
     scheduler.cancel(bootstrapJob);
+    bootstrap_pending = false;
     bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
@@ -2333,9 +2363,16 @@ Dht::exportValues() const
         const auto& vals = h.second.getValues();
         pk.pack_array(vals.size());
         for (const auto& v : vals) {
-            pk.pack_array(2);
+            pk.pack_array(4);
             pk.pack(v.created.time_since_epoch().count());
             v.data->msgpack_pack(pk);
+            if (const auto& store_addr = v.store_bucket ? v.store_bucket->getAddr() : SockAddr {}) {
+                pk.pack_bin(store_addr.getLength());
+                pk.pack_bin_body(reinterpret_cast<const char*>(store_addr.get()), store_addr.getLength());
+            } else {
+                pk.pack_bin(0);
+            }
+            pk.pack(v.expiration.time_since_epoch().count());
         }
         ve.second = {buffer.data(), buffer.data() + buffer.size()};
         e.push_back(std::move(ve));
@@ -2348,9 +2385,13 @@ Dht::importValues(const std::vector<ValuesExport>& import)
 {
     const auto& now = scheduler.time();
 
+    unsigned imported = 0, ignored = 0;
+    size_t imported_size = 0;
     for (const auto& value : import) {
-        if (value.second.empty())
+        if (value.second.empty()) {
+            ignored++;
             continue;
+        }
 
         try {
             msgpack::unpacked msg;
@@ -2363,25 +2404,53 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                 if (valel.type != msgpack::type::ARRAY or valel.via.array.size < 2)
                     throw msgpack::type_error();
                 time_point val_time;
+                time_point expiration = time_point::min();
                 Value tmp_val;
+                SockAddr store_addr;
                 try {
                     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>();
+                        if (not addr_blob.empty()) {
+                            store_addr = SockAddr(reinterpret_cast<const sockaddr*>(addr_blob.data()),
+                                                  static_cast<socklen_t>(addr_blob.size()));
+                        }
+                    }
+                    if (valel.via.array.size >= 4) {
+                        expiration = time_point {
+                            time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
+                    }
                 } catch (const std::exception&) {
                     if (logger_)
                         logger_->error("Error reading value at {}", value.first.to_view());
+                    ignored++;
                     continue;
                 }
                 val_time = std::min(val_time, now);
-                storageStore(value.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                auto val_size = tmp_val.size();
+                if (storageStore(value.first,
+                                 std::make_shared<Value>(std::move(tmp_val)),
+                                 val_time,
+                                 store_addr,
+                                 false,
+                                 expiration)) {
+                    imported++;
+                    imported_size += val_size;
+                } else {
+                    ignored++;
+                }
             }
         } catch (const std::exception&) {
+            ignored++;
             if (logger_)
                 logger_->error("Error reading values at {}", value.first.to_view());
             continue;
         }
     }
+    if (logger_)
+        logger_->debug("Imported {} values, {}, ignored {}", imported, dht::printByteCount(imported_size), ignored);
 }
 
 std::vector<NodeExport>
@@ -2466,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;
@@ -2622,11 +2701,13 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
     }
     Query q = query;
     storageAddListener(hash, node, socket_id, std::move(q), version);
-    return {};
+    net::RequestAnswer answer {};
+    answer.ntoken = makeToken(node->getAddr(), false);
+    return answer;
 }
 
 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",
@@ -2634,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();
@@ -700,7 +702,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 RequestAnswer answer
                     = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
                 auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
-                sendListenConfirmation(from, msg->tid);
+                sendListenConfirmation(from, msg->tid, answer.ntoken);
                 break;
             }
             case MessageType::UpdateValue: {
@@ -1279,17 +1281,21 @@ NetworkEngine::sendListen(const Sp<Node>& n,
 }
 
 void
-NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid)
+NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid, const Blob& token)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(4 + (config.network ? 1 : 0) + (config.is_client ? 1 : 0));
 
     pk.pack(KEY_R);
-    pk.pack_map(2);
+    pk.pack_map(2 + (not token.empty() ? 1 : 0));
     pk.pack(KEY_REQ_ID);
     pk.pack(myid);
     insertAddr(pk, addr);
+    if (not token.empty()) {
+        pk.pack(KEY_REQ_TOKEN);
+        packToken(pk, token);
+    }
 
     pk.pack(KEY_TID);
     pk.pack(tid);

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>

src/storage.h

@@ -3,8 +3,9 @@
 #pragma once
 
 #include "infohash.h"
-#include "value.h"
 #include "listener.h"
+#include "sockaddr.h"
+#include "value.h"
 
 #include <map>
 #include <utility>
@@ -17,6 +18,11 @@ namespace dht {
 class StorageBucket
 {
 public:
+    StorageBucket() = default;
+    explicit StorageBucket(SockAddr addr)
+        : addr_(std::move(addr))
+    {}
+
     void insert(const InfoHash& id, const Value& value, time_point expiration)
     {
         totalSize_ += value.size();
@@ -56,8 +62,10 @@ public:
     {
         return storedValues_.empty() ? std::pair<InfoHash, Value::Id> {} : storedValues_.begin()->second;
     }
+    const SockAddr& getAddr() const { return addr_; }
 
 private:
+    SockAddr addr_ {};
     std::multimap<time_point, std::pair<InfoHash, Value::Id>> storedValues_;
     size_t totalSize_ {0};
 };

src/utils.cpp

@@ -296,4 +296,19 @@ findMapValue(const msgpack::object& map, const char* key, size_t key_length)
     return nullptr;
 }
 
+std::string
+printByteCount(size_t bytes)
+{
+    static constexpr std::string_view suffixes[] = {"B", "KB", "MB", "GB", "TB"};
+    size_t suffixIndex = 0;
+    double count = static_cast<double>(bytes);
+    while (count >= 1024 && suffixIndex < std::size(suffixes) - 1) {
+        count /= 1024;
+        suffixIndex++;
+    }
+    if (suffixIndex == 0)
+        return fmt::format("{} {}", bytes, suffixes[suffixIndex]);
+    return fmt::format("{:.2f} {} ({} bytes)", count, suffixes[suffixIndex], bytes);
+}
+
 } // namespace dht

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,18 +124,345 @@ 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);
 }
 
+void
+DhtRunnerTester::testImportValuesPreservesRemoteQuota()
+{
+    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 = 1536;
+    config.dht_config.node_config.max_local_store_size = -1;
+    config.dht_config.node_config.max_store_keys = -1;
+
+    node1.run(0, config);
+    node2.run(0, config);
+    auto bound = node1.getBound();
+    if (bound.isUnspecified())
+        bound.setLoopback();
+    node2.bootstrap(bound);
+    std::this_thread::sleep_for(1s);
+
+    auto key1 = dht::InfoHash::get("import-quota-1");
+    auto key2 = dht::InfoHash::get("import-quota-2");
+    auto value1 = std::make_shared<dht::Value>(std::string(1024, 'a'));
+    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(getFutureValue(std::move(p1future)));
+
+    std::this_thread::sleep_for(2s);
+    auto exported = node1.exportValues();
+    CPPUNIT_ASSERT(!exported.empty());
+
+    node1.shutdown();
+    node1.join();
+
+    dht::DhtRunner restored;
+    restored.run(0, config);
+    auto restoredBound = restored.getBound();
+    if (restoredBound.isUnspecified())
+        restoredBound.setLoopback();
+
+    node2.shutdown();
+    node2.join();
+    node2.run(0, config);
+    node2.bootstrap(restoredBound);
+    std::this_thread::sleep_for(1s);
+
+    restored.importValues(exported);
+    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(getFutureValue(std::move(p2future)));
+
+    std::this_thread::sleep_for(2s);
+
+    auto vals1 = getFutureValue(restored.get(key1));
+    auto vals2 = getFutureValue(restored.get(key2));
+    CPPUNIT_ASSERT(vals1.size() + vals2.size() <= 1);
+
+    restored.shutdown();
+    restored.join();
+}
+
+void
+DhtRunnerTester::testImportValuesPreservesStoredExpiration()
+{
+    static const dht::ValueType SHORT_LIVED_TYPE {200, "Short lived", 2s};
+
+    node1.registerInsecureType(SHORT_LIVED_TYPE);
+
+    auto key = dht::InfoHash::get("import-expiration");
+    auto now = dht::clock::now();
+    auto created = now;
+    auto expiration = now + SHORT_LIVED_TYPE.expiration;
+
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_array(1);
+    pk.pack_array(4);
+    pk.pack(created.time_since_epoch().count());
+    dht::Value {SHORT_LIVED_TYPE.id, std::string("short-lived")}.msgpack_pack(pk);
+    pk.pack_bin(0);
+    pk.pack(expiration.time_since_epoch().count());
+
+    node1.importValues({
+        dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
+    });
+
+    auto exported = node1.exportValues();
+    CPPUNIT_ASSERT(!exported.empty());
+
+    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(exported);
+    CPPUNIT_ASSERT(!restored.exportValues().empty());
+
+    std::this_thread::sleep_for(3s);
+
+    auto restoredValues = restored.exportValues();
+    CPPUNIT_ASSERT_EQUAL(size_t(1), restoredValues.size());
+
+    msgpack::unpacked msg;
+    msgpack::unpack(msg, (const char*) restoredValues.front().second.data(), restoredValues.front().second.size());
+    auto valarr = msg.get();
+    CPPUNIT_ASSERT(valarr.type == msgpack::type::ARRAY);
+    CPPUNIT_ASSERT_EQUAL(uint32_t(0), valarr.via.array.size);
+
+    restored.shutdown();
+    restored.join();
+}
+
+void
+DhtRunnerTester::testBootstrapSetsConnectingState()
+{
+    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;
+
+    std::mutex mutex;
+    std::condition_variable cv;
+    bool sawConnecting {false};
+
+    dht::DhtRunner::Context context;
+    context.statusChangedCallback = [&](dht::NodeStatus status4, dht::NodeStatus status6) {
+        std::lock_guard lock(mutex);
+        sawConnecting = sawConnecting || status4 == dht::NodeStatus::Connecting
+                        || status6 == dht::NodeStatus::Connecting;
+        cv.notify_all();
+    };
+
+    dht::DhtRunner bootstrapNode;
+    dht::DhtRunner clientNode;
+    bootstrapNode.run(0, config);
+    clientNode.run(config, std::move(context));
+
+    auto bound = bootstrapNode.getBound();
+    if (bound.isUnspecified())
+        bound.setLoopback();
+
+    clientNode.bootstrap("127.0.0.1", std::to_string(bound.getPort()));
+
+    {
+        std::unique_lock lock(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lock, 5s, [&] { return sawConnecting; }));
+    }
+
+    bootstrapNode.shutdown();
+    bootstrapNode.join();
+    clientNode.shutdown();
+    clientNode.join();
+}
+
+void
+DhtRunnerTester::testBootstrapThenPutNoRace()
+{
+    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;
+
+    dht::DhtRunner bootstrapNode;
+    dht::DhtRunner clientNode;
+    bootstrapNode.run(0, config);
+    clientNode.run(0, config);
+
+    auto bound = bootstrapNode.getBound();
+    if (bound.isUnspecified())
+        bound.setLoopback();
+
+    clientNode.bootstrap(bound);
+
+    std::promise<bool> putDone;
+    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 = getFutureValue(bootstrapNode.get(dht::InfoHash::get("bootstrap-then-put")));
+    CPPUNIT_ASSERT(!vals.empty());
+
+    bootstrapNode.shutdown();
+    bootstrapNode.join();
+    clientNode.shutdown();
+    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()
 {
@@ -223,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);
@@ -265,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();
@@ -299,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();
@@ -431,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

@@ -20,6 +20,12 @@ class DhtRunnerTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST(testIdOps);
+    CPPUNIT_TEST(testImportValuesPreservesRemoteQuota);
+    CPPUNIT_TEST(testImportValuesPreservesStoredExpiration);
+    CPPUNIT_TEST(testBootstrapSetsConnectingState);
+    CPPUNIT_TEST(testBootstrapThenPutNoRace);
+    CPPUNIT_TEST(testBootstrapMissingNodeThenPutFails);
+    CPPUNIT_TEST(testShutdownCompletesWithPendingPut);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -58,6 +64,12 @@ public:
      * Test methods requiring a node identity
      */
     void testIdOps();
+    void testImportValuesPreservesRemoteQuota();
+    void testImportValuesPreservesStoredExpiration();
+    void testBootstrapSetsConnectingState();
+    void testBootstrapThenPutNoRace();
+    void testBootstrapMissingNodeThenPutFails();
+    void testShutdownCompletesWithPendingPut();
     /**
      * Test listen method with lot of datas
      */

tests/test_networkengine.cpp

@@ -3,15 +3,67 @@
 
 #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"
 #include "opendht/utils.h"
 #include "opendht/value.h"
 
+#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/dht.h"
+#undef private
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#elif defined(__GNUC__)
+#pragma GCC diagnostic pop
+#endif
+
 #include "../src/parsed_message.h"
+#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"
@@ -116,6 +168,35 @@ makeValueDataPacketBlob(Tid tid, unsigned index, size_t offset, const Blob& data
     return {buffer.data(), buffer.data() + buffer.size()};
 }
 
+static Blob
+makeListenPacketBlob(const InfoHash& id, const InfoHash& hash, Tid tid, Tid socketId, const Blob& token)
+{
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+
+    pk.pack_map(4);
+    pk.pack(KEY_A);
+    pk.pack_map(5);
+    pk.pack(KEY_REQ_ID);
+    pk.pack(id);
+    pk.pack(KEY_VERSION);
+    pk.pack(1);
+    pk.pack(KEY_REQ_H);
+    pk.pack(hash);
+    pk.pack(KEY_REQ_TOKEN);
+    packToken(pk, token);
+    pk.pack(KEY_REQ_SID);
+    pk.pack(socketId);
+    pk.pack(KEY_Q);
+    pk.pack(QUERY_LISTEN);
+    pk.pack(KEY_TID);
+    pk.pack(tid);
+    pk.pack(KEY_Y);
+    pk.pack(KEY_Q);
+
+    return {buffer.data(), buffer.data() + buffer.size()};
+}
+
 static SockAddr
 makeIPv4(const char* ip, in_port_t port)
 {
@@ -247,4 +328,222 @@ NetworkEngineTester::testKeepsSessionForWrongSourceFragment()
     CPPUNIT_ASSERT(onNewNodeCalls > 0);
 }
 
-} // namespace test
+void
+NetworkEngineTester::testListenConfirmationCarriesToken()
+{
+    Scheduler scheduler;
+    std::mt19937_64 rd(3);
+    InfoHash myid = InfoHash::getRandom(rd);
+    InfoHash remoteId = InfoHash::getRandom(rd);
+    InfoHash hash = InfoHash::getRandom(rd);
+    Blob requestToken {0x01, 0x02, 0x03, 0x04};
+    Blob responseToken {0xaa, 0xbb, 0xcc, 0xdd, 0xee};
+    int onNewNodeCalls = 0;
+    net::NetworkConfig config {};
+    config.max_req_per_sec = -1;
+    config.max_peer_req_per_sec = -1;
+
+    auto socket = std::make_unique<TestDatagramSocket>();
+    auto* socketPtr = socket.get();
+    auto engine = net::NetworkEngine(
+        myid,
+        config,
+        std::move(socket),
+        {},
+        rd,
+        scheduler,
+        [](Sp<net::Request>, net::DhtProtocolException) {},
+        [&onNewNodeCalls](const Sp<Node>&, int) { ++onNewNodeCalls; },
+        [](const InfoHash&, const SockAddr&) {},
+        [](Sp<Node>) { return net::RequestAnswer {}; },
+        [](Sp<Node>, const InfoHash&, want_t) { return net::RequestAnswer {}; },
+        [](Sp<Node>, const InfoHash&, want_t, const Query&) { return net::RequestAnswer {}; },
+        [responseToken](Sp<Node>, const InfoHash&, const Blob&, Tid, const Query&, int) {
+            net::RequestAnswer answer {};
+            answer.ntoken = responseToken;
+            return answer;
+        },
+        [](Sp<Node>, const InfoHash&, const Blob&, const std::vector<Sp<Value>>&, const time_point&) {
+            return net::RequestAnswer {};
+        },
+        [](Sp<Node>, const InfoHash&, const Blob&, const Value::Id&) { return net::RequestAnswer {}; });
+
+    auto from = makeIPv4("127.0.0.2", 5003);
+    auto packet = makeListenPacketBlob(remoteId, hash, 17, 23, requestToken);
+
+    engine.processMessage(packet.data(), packet.size(), from);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, socketPtr->sends.size());
+
+    const auto& sent = socketPtr->sends.front();
+    CPPUNIT_ASSERT(sent.dest == from);
+
+    auto replyObject = msgpack::unpack((const char*) sent.data.data(), sent.data.size());
+    ParsedMessage reply;
+    reply.msgpack_unpack(replyObject.get());
+
+    CPPUNIT_ASSERT(reply.type == MessageType::Reply);
+    CPPUNIT_ASSERT_EQUAL(myid, reply.id);
+    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()
+{
+    Config config {};
+    config.max_req_per_sec = -1;
+    config.max_peer_req_per_sec = -1;
+
+    auto rd = std::make_unique<std::mt19937_64>(4);
+    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", 5004),
+                                       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;
+    sr->listeners.emplace(1, Dht::Search::SearchListener {query, valueCallback, {}});
+
+    auto searchNode = std::make_unique<Dht::SearchNode>(node);
+    auto staleSocketId = node->openSocket([](const Sp<Node>&, net::RequestAnswer&&) {});
+    auto status = searchNode->listenStatus
+                      .emplace(std::piecewise_construct,
+                               std::forward_as_tuple(query),
+                               std::forward_as_tuple(valueCallback, syncCallback, staleSocketId))
+                      .first;
+    status->second.req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
+    status->second.req->reply_time = time_point::min();
+
+    node->setExpired();
+    CPPUNIT_ASSERT(!node->getSocket(staleSocketId));
+
+    auto* searchNodePtr = sr->nodes.emplace_back(std::move(searchNode)).get();
+    searchNodePtr->token = Blob {0x42};
+    searchNodePtr->last_get_reply = localDht.scheduler.time();
+
+    localDht.searchSynchedNodeListen(sr, *searchNodePtr);
+
+    auto listenIt = searchNodePtr->listenStatus.find(query);
+    CPPUNIT_ASSERT(listenIt != searchNodePtr->listenStatus.end());
+    CPPUNIT_ASSERT(listenIt->second.req);
+    CPPUNIT_ASSERT(listenIt->second.socketId);
+    CPPUNIT_ASSERT(listenIt->second.socketId != staleSocketId);
+    CPPUNIT_ASSERT(node->getSocket(listenIt->second.socketId));
+}
+
+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,18 +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

tools/dhtchat.cpp

@@ -12,10 +12,10 @@ extern "C" {
 }
 #include <ctime>
 
-using namespace dht;
+namespace dht {
+namespace tools {
 
 static std::mt19937_64 rd {dht::crypto::getSeededRandomEngine<std::mt19937_64>()};
-static std::uniform_int_distribution<dht::Value::Id> rand_id;
 
 const std::string
 printTime(const std::time_t& now)
@@ -34,12 +34,18 @@ print_usage()
     std::cout << "Report bugs to: https://opendht.net" << std::endl;
 }
 
+} // namespace tools
+} // namespace dht
+
 int
 main(int argc, char** argv)
 {
-    auto params = parseArgs(argc, argv);
+    using namespace dht;
+    using IdDist = std::uniform_int_distribution<dht::Value::Id>;
+
+    auto params = tools::parseArgs(argc, argv);
     if (params.help) {
-        print_usage();
+        tools::print_usage();
         return 0;
     }
 #ifdef _MSC_VER
@@ -58,14 +64,14 @@ main(int argc, char** argv)
         if (not params.bootstrap.empty())
             dht.bootstrap(params.bootstrap);
 
-        print_node_info(dht.getNodeInfo());
+        tools::print_node_info(dht.getNodeInfo());
         std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;
 
         bool connected {false};
         InfoHash room;
         std::future<size_t> token;
 
-        const InfoHash myid = dht.getId();
+        const auto mypk = dht.getPublicKey();
 
 #ifndef _MSC_VER
         // using the GNU History API
@@ -74,7 +80,7 @@ main(int argc, char** argv)
 
         while (true) {
             // using the GNU Readline API
-            std::string line = readLine(connected ? PROMPT : "> ");
+            std::string line = tools::readLine(connected ? tools::PROMPT : "> ");
             if (!line.empty() && line[0] == '\0')
                 break;
             if (line.empty())
@@ -95,12 +101,12 @@ main(int argc, char** argv)
                     }
 
                     token = dht.listen<dht::ImMessage>(room, [&](dht::ImMessage&& msg) {
-                        if (msg.from != myid)
-                            std::cout << msg.from.toString() << " at " << printTime(msg.date) << " (took "
+                        if (msg.owner && msg.owner->getLongId() != mypk->getLongId())
+                            std::cout << msg.from.toString() << " at " << tools::printTime(msg.date) << " (took "
                                       << print_duration(std::chrono::system_clock::now()
                                                         - std::chrono::system_clock::from_time_t(msg.date))
-                                      << ") " << (msg.to == myid ? "ENCRYPTED " : "") << ": " << msg.id << " - "
-                                      << msg.msg << std::endl;
+                                      << ") " << (msg.to == mypk->getId() ? "ENCRYPTED " : "") << ": " << msg.id
+                                      << " - " << msg.msg << std::endl;
                         return true;
                     });
                     connected = true;
@@ -118,14 +124,14 @@ main(int argc, char** argv)
                     std::getline(iss, line);
                     dht.putEncrypted(room,
                                      InfoHash(idstr),
-                                     dht::ImMessage(rand_id(rd), std::move(line), now),
+                                     dht::ImMessage(IdDist()(tools::rd), std::move(line), now),
                                      [](bool ok) {
                                          // dht.cancelPut(room, id);
                                          if (not ok)
                                              std::cout << "Message publishing failed !" << std::endl;
                                      });
                 } else {
-                    dht.putSigned(room, dht::ImMessage(rand_id(rd), std::move(line), now), [](bool ok) {
+                    dht.putSigned(room, dht::ImMessage(IdDist()(tools::rd), std::move(line), now), [](bool ok) {
                         // dht.cancelPut(room, id);
                         if (not ok)
                             std::cout << "Message publishing failed !" << std::endl;

tools/dhtnode.cpp

@@ -16,7 +16,8 @@ extern "C" {
 #include <set>
 #include <thread> // std::this_thread::sleep_for
 
-using namespace dht;
+namespace dht {
+namespace tools {
 
 void
 print_info()
@@ -548,9 +549,14 @@ cmd_loop(std::shared_ptr<DhtRunner>& node,
     std::cout << std::endl << "Stopping node..." << std::endl;
 }
 
+} // namespace tools
+} // namespace dht
+
 int
 main(int argc, char** argv)
 {
+    using namespace dht;
+    using namespace dht::tools;
 #ifdef _WIN32
     SetConsoleOutputCP(CP_UTF8);
     SetConsoleCP(CP_UTF8);

tools/dhtscanner.cpp

@@ -15,7 +15,8 @@ extern "C" {
 #include <condition_variable>
 #include <mutex>
 
-using namespace dht;
+namespace dht {
+namespace tools {
 
 void
 print_usage()
@@ -82,9 +83,13 @@ step(DhtRunner& dht, std::atomic_uint& done, std::shared_ptr<NodeSet> all_nodes,
         });
 }
 
+} // namespace tools
+} // namespace dht
+
 int
 main(int argc, char** argv)
 {
+    using namespace dht::tools;
 #ifdef _MSC_VER
     if (auto err = gnutls_global_init()) {
         fmt::print(stderr, "Failed to initialize GnuTLS: {}\n", gnutls_strerror(err));
@@ -97,7 +102,7 @@ main(int argc, char** argv)
         return EXIT_SUCCESS;
     }
 
-    DhtRunner dht;
+    dht::DhtRunner dht;
     try {
         auto [config, context] = getDhtConfig(params);
         config.dht_config.node_config.client_mode = true;

tools/perftest.cpp

@@ -14,6 +14,14 @@ extern "C" {
 #include <mutex>
 #include <atomic>
 
+namespace dht {
+namespace tests {
+
+constexpr unsigned PINGPONG_MAX = 2048;
+
+using clock = std::chrono::high_resolution_clock;
+using duration = clock::duration;
+
 void
 print_usage()
 {
@@ -21,14 +29,6 @@ print_usage()
     std::cout << "perftest, a simple OpenDHT basic performance tester." << std::endl;
     std::cout << "Report bugs to: https://opendht.net" << std::endl;
 }
-constexpr unsigned PINGPONG_MAX = 2048;
-using namespace dht;
-
-namespace tests {
-
-using clock = std::chrono::high_resolution_clock;
-using duration = clock::duration;
-
 duration
 benchPingPong(unsigned netSize, unsigned n_parallel)
 {
@@ -115,10 +115,13 @@ benchPingPong(unsigned netSize, unsigned n_parallel)
 }
 
 } // namespace tests
+} // namespace dht
 
 int
 main(int argc, char** argv)
 {
+    using namespace dht::tools;
+    using namespace dht::tests;
 #ifdef _MSC_VER
     if (auto err = gnutls_global_init()) {
         std::cerr << "Failed to initialize GnuTLS: " << gnutls_strerror(err) << std::endl;
@@ -140,10 +143,10 @@ main(int argc, char** argv)
         results.reserve(8);
         duration total {0};
         for (unsigned i = 2; i < 32; i *= 2) {
-            auto dt = tests::benchPingPong(i - 2, nparallel);
+            auto dt = benchPingPong(i - 2, nparallel);
             std::cout << "Network size: " << i << std::endl;
-            std::cout << max << " ping-pong done, took " << print_duration(dt) << std::endl;
-            std::cout << print_duration(dt / max) << " per rt, " << max / std::chrono::duration<double>(dt).count()
+            std::cout << max << " ping-pong done, took " << dht::print_duration(dt) << std::endl;
+            std::cout << dht::print_duration(dt / max) << " per rt, " << max / std::chrono::duration<double>(dt).count()
                       << " ping per s" << std::endl
                       << std::endl;
             total += dt;
@@ -155,14 +158,14 @@ main(int argc, char** argv)
 
         std::cout << "Total for " << nparallel << std::endl;
         auto totNum = max * results.size();
-        std::cout << totNum << " ping-pong done, took " << print_duration(total) << std::endl;
-        std::cout << print_duration(total / totNum) << " per rt, "
+        std::cout << totNum << " ping-pong done, took " << dht::print_duration(total) << std::endl;
+        std::cout << dht::print_duration(total / totNum) << " per rt, "
                   << totNum / std::chrono::duration<double>(total).count() << " ping per s" << std::endl
                   << std::endl;
     }
 
-    std::cout << std::endl << "Grand total: " << print_duration(totalTime) << " for " << totalOps << std::endl;
-    std::cout << print_duration(totalTime / totalOps) << " per rt, "
+    std::cout << std::endl << "Grand total: " << dht::print_duration(totalTime) << " for " << totalOps << std::endl;
+    std::cout << dht::print_duration(totalTime / totalOps) << " per rt, "
               << totalOps / std::chrono::duration<double>(totalTime).count() << " ping per s" << std::endl
               << std::endl;
 

tools/tools_common.h

@@ -54,6 +54,9 @@
 #include <sstream>
 #include <fstream>
 
+namespace dht {
+namespace tools {
+
 /*
  * The mapString shall have the following format:
  *
@@ -119,21 +122,6 @@ loadFile(const std::string& path)
     return buffer;
 }
 
-std::string
-printByteCount(size_t bytes)
-{
-    static constexpr std::string_view suffixes[] = {"B", "KB", "MB", "GB", "TB"};
-    size_t suffixIndex = 0;
-    double count = static_cast<double>(bytes);
-    while (count >= 1024 && suffixIndex < std::size(suffixes) - 1) {
-        count /= 1024;
-        suffixIndex++;
-    }
-    if (suffixIndex == 0)
-        return fmt::format("{} {}", bytes, suffixes[suffixIndex]);
-    return fmt::format("{:.2f} {} ({} bytes)", count, suffixes[suffixIndex], bytes);
-}
-
 struct dht_params
 {
     bool help {false}; // print help and exit
@@ -504,3 +492,6 @@ daemonize()
     close(STDERR_FILENO);
 #endif
 }
+
+} // namespace tools
+} // namespace dht

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",