http: bound name resolution and reuse the endpoints it found

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

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

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

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

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

CMakeLists.txt

@@ -502,11 +502,7 @@ list (APPEND opendht_HEADERS
 )
 
 if (OPENDHT_PEER_DISCOVERY)
-    list (APPEND opendht_SOURCES
-        src/peer_discovery/mdns.cpp
-        src/peer_discovery/dns_sd.cpp
-        src/peer_discovery/mdns_transport.cpp
-        src/peer_discovery/peer_discovery.cpp)
+    list (APPEND opendht_SOURCES src/peer_discovery.cpp)
     list (APPEND opendht_HEADERS include/opendht/peer_discovery.h)
     add_definitions(-DOPENDHT_PEER_DISCOVERY)
 endif()
@@ -593,7 +589,6 @@ if (MSVC)
             simdutf::simdutf
             wsock32
             ws2_32
-            iphlpapi
     )
     if (OPENDHT_TOOLS)
         target_link_libraries(opendht PUBLIC unofficial::readline-win32::readline)
@@ -677,7 +672,7 @@ else()
     # System-specific linker pipelines
     if (WIN32 AND MINGW) # MSYS2
         message(STATUS "Build pipeline: Windows/MSYS2-MinGW")
-        target_link_libraries(opendht PUBLIC wsock32 ws2_32 iphlpapi)
+        target_link_libraries(opendht PUBLIC wsock32 ws2_32)
         if(OPENDHT_PROXY_OPENSSL)
             target_link_libraries(opendht PUBLIC Crypt32)
         endif()
@@ -893,8 +888,6 @@ if (BUILD_TESTING)
         endif()
         if (OPENDHT_PEER_DISCOVERY)
             list (APPEND test_FILES
-                tests/test_mdns.h
-                tests/test_mdns.cpp
                 tests/test_peerdiscovery.h
                 tests/test_peerdiscovery.cpp
             )

include/opendht/crypto.h

@@ -204,7 +204,13 @@ struct OPENDHT_PUBLIC PrivateKey
      * @param algo : the public key algorithm to use, GNUTLS_PK_RSA or GNUTLS_PK_RSA_OAEP
      */
     static PrivateKey generate(unsigned key_length = 4096, gnutls_pk_algorithm_t algo = GNUTLS_PK_RSA);
-    static PrivateKey generateEC();
+
+    /**
+     * Generate a new EC key pair
+     * @param curve : the elliptic curve to use. If GNUTLS_ECC_CURVE_INVALID is provided, the default curve will be used
+     * (currently GNUTLS_ECC_CURVE_SECP384R1).
+     */
+    static PrivateKey generateEC(gnutls_ecc_curve_t curve = GNUTLS_ECC_CURVE_INVALID);
 
     gnutls_privkey_t key {};
     gnutls_x509_privkey_t x509_key {};
@@ -272,7 +278,8 @@ public:
      * and incremented on subsequent ones.
      */
     void sign(const PrivateKey&, const Certificate&, duration validity_period = {}, uint64_t number = 0);
-    void sign(const Identity& id, duration validity_period = {}, uint64_t number = 0) {
+    void sign(const Identity& id, duration validity_period = {}, uint64_t number = 0)
+    {
         sign(*id.first, *id.second, validity_period, number);
     }
 

include/opendht/http.h

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

include/opendht/peer_discovery.h

@@ -3,13 +3,12 @@
 #pragma once
 
 #include "def.h"
+#include "sockaddr.h"
 #include "infohash.h"
 #include "logger.h"
-#include "sockaddr.h"
 #include "utils.h"
 
-#include <functional>
-#include <memory>
+#include <asio/steady_timer.hpp>
 #include <thread>
 
 namespace asio {
@@ -21,26 +20,66 @@ namespace dht {
 class OPENDHT_PUBLIC PeerDiscovery
 {
 public:
-    using PeerDiscoveredCallback = std::function<void(const InfoHash&, SockAddr&&)>;
+    static constexpr in_port_t DEFAULT_PORT = 8888;
+    using ServiceDiscoveredCallback = std::function<void(msgpack::object&&, SockAddr&&)>;
 
-    PeerDiscovery(std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
+    PeerDiscovery(in_port_t port = DEFAULT_PORT,
+                  std::shared_ptr<asio::io_context> ioContext = {},
+                  std::shared_ptr<Logger> logger = {});
     ~PeerDiscovery();
 
-    void startDiscovery(NetId network, PeerDiscoveredCallback callback);
-    void startPublish(const InfoHash& nodeId, NetId network, in_port_t port);
+    /**
+     * startDiscovery - Keep Listening data from the sender until node is joinned
+     * or stop is called
+     */
+    void startDiscovery(std::string_view type, ServiceDiscoveredCallback callback);
 
-    bool stopDiscovery();
-    bool stopPublish();
+    template<typename T>
+    void startDiscovery(std::string_view type, std::function<void(T&&, SockAddr&&)> cb)
+    {
+        startDiscovery(type, [cb](msgpack::object&& ob, SockAddr&& addr) { cb(ob.as<T>(), std::move(addr)); });
+    }
+
+    /**
+     * startPublish - Keeping sending data until node is joinned or stop is called
+     */
+    void startPublish(std::string_view type, const msgpack::sbuffer& pack_buf);
+    void startPublish(sa_family_t domain, std::string_view type, const msgpack::sbuffer& pack_buf);
+
+    template<typename T>
+    void startPublish(std::string_view type, const T& object)
+    {
+        msgpack::sbuffer buf;
+        msgpack::pack(buf, object);
+        startPublish(type, buf);
+    }
+
+    /**
+     * Thread Stopper
+     */
     void stop();
 
+    /**
+     * Remove possible callBack to discovery
+     */
+    bool stopDiscovery(std::string_view type);
+
+    /**
+     * Remove different serivce message to send
+     */
+    bool stopPublish(std::string_view type);
+    bool stopPublish(sa_family_t domain, std::string_view type);
+
     void connectivityChanged();
+
     void stopConnectivityChanged();
 
 private:
-    class Impl;
-    std::shared_ptr<Impl> impl_;
+    class DomainPeerDiscovery;
+    std::unique_ptr<DomainPeerDiscovery> peerDiscovery4_;
+    std::unique_ptr<DomainPeerDiscovery> peerDiscovery6_;
     std::shared_ptr<asio::io_context> ioContext_;
-    std::thread ioRunner_;
+    std::thread ioRunnner_;
 };
 
-} // namespace dht
+} // namespace dht

meson.build

@@ -123,21 +123,13 @@ if get_option('push_notifications').enabled()
     add_project_arguments('-DOPENDHT_PUSH_NOTIFICATIONS', language: 'cpp')
 endif
 if get_option('peer_discovery').enabled()
-    opendht_src += [
-        'src/peer_discovery/mdns.cpp',
-        'src/peer_discovery/dns_sd.cpp',
-        'src/peer_discovery/mdns_transport.cpp',
-        'src/peer_discovery/peer_discovery.cpp',
-    ]
+    opendht_src += 'src/peer_discovery.cpp'
     add_project_arguments('-DOPENDHT_PEER_DISCOVERY', language: 'cpp')
-    if host_machine.system() == 'windows'
-        deps += meson.get_compiler('cpp').find_library('iphlpapi')
-    endif
 endif
 opendht = library(
     'opendht',
     opendht_src,
-    include_directories: [opendht_inc, opendht_interface_inc],
+    include_directories: opendht_inc,
     dependencies: deps,
     cpp_args: ['-DOPENDHT_BUILD', '-Dopendht_EXPORTS'],
     install: true,
@@ -335,14 +327,6 @@ if get_option('tests').enabled()
     endif
 
     if get_option('peer_discovery').enabled()
-        test_mdns = executable(
-            'test_mdns',
-            'tests/test_mdns.cpp',
-            'tests/tests_runner.cpp',
-            dependencies: [opendht_dep, cppunit, jsoncpp, fmt, openssl, msgpack],
-        )
-        test('Mdns', test_mdns)
-
         test_peerdiscovery = executable(
             'test_peerdiscovery',
             'tests/test_peerdiscovery.cpp',

src/crypto.cpp

@@ -1294,14 +1294,16 @@ PrivateKey::generate(unsigned key_length, gnutls_pk_algorithm_t algo)
 }
 
 PrivateKey
-PrivateKey::generateEC()
+PrivateKey::generateEC(gnutls_ecc_curve_t curve)
 {
     gnutls_x509_privkey_t key;
     if (gnutls_x509_privkey_init(&key) != GNUTLS_E_SUCCESS)
         throw CryptoException("Unable to initialize private key.");
     int err = gnutls_x509_privkey_generate(key,
                                            GNUTLS_PK_EC,
-                                           gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA),
+                                           curve == GNUTLS_ECC_CURVE_INVALID
+                                               ? gnutls_sec_param_to_pk_bits(GNUTLS_PK_EC, GNUTLS_SEC_PARAM_ULTRA)
+                                               : GNUTLS_CURVE_TO_BITS(curve),
                                            0);
     if (err != GNUTLS_E_SUCCESS) {
         gnutls_x509_privkey_deinit(key);

src/dht_proxy_client.cpp

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

src/dhtrunner.cpp

@@ -21,6 +21,16 @@ namespace dht {
 
 using namespace std::literals;
 
+constexpr auto PEER_DISCOVERY_DHT_SERVICE = "dht"sv;
+
+struct NodeInsertionPack
+{
+    dht::InfoHash nodeId;
+    in_port_t port;
+    dht::NetId net;
+    MSGPACK_DEFINE(nodeId, port, net)
+};
+
 DhtRunner::DhtRunner()
     : dht_()
 {
@@ -237,7 +247,7 @@ DhtRunner::run(const Config& config, Context&& context)
         if (config.peer_discovery or config.peer_publish) {
 #ifdef OPENDHT_PEER_DISCOVERY
             peerDiscovery_ = context.peerDiscovery ? std::move(context.peerDiscovery)
-                                                   : std::make_shared<PeerDiscovery>(nullptr, context.logger);
+                                                   : std::make_shared<PeerDiscovery>();
 #else
             std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl;
 #endif
@@ -246,22 +256,32 @@ DhtRunner::run(const Config& config, Context&& context)
 #ifdef OPENDHT_PEER_DISCOVERY
         auto netId = config.dht_config.node_config.network;
         if (config.peer_discovery) {
-            peerDiscovery_->startDiscovery(netId, [this](const InfoHash& nodeId, SockAddr&& address) {
-                if (nodeId != dht_->getNodeId())
-                    bootstrap(nodeId, address);
-            });
+            peerDiscovery_->startDiscovery<NodeInsertionPack>(PEER_DISCOVERY_DHT_SERVICE,
+                                                              [this, netId](NodeInsertionPack&& v, SockAddr&& addr) {
+                                                                  addr.setPort(v.port);
+                                                                  if (v.nodeId != dht_->getNodeId() && netId == v.net) {
+                                                                      bootstrap(v.nodeId, addr);
+                                                                  }
+                                                              });
         }
         if (config.peer_publish) {
+            msgpack::sbuffer sbuf_node;
+            NodeInsertionPack adc;
+            adc.net = netId;
+            adc.nodeId = dht_->getNodeId();
             if (auto socket = dht_->getSocket()) {
-                const auto& bound4 = socket->getBoundRef(AF_INET);
-                const auto& bound6 = socket->getBoundRef(AF_INET6);
-                const auto port4 = bound4 ? bound4.getPort() : 0;
-                const auto port6 = bound6 ? bound6.getPort() : 0;
-                if (port4 and port6 and port4 != port6) {
-                    if (logger_)
-                        logger_->error("Unable to publish peer discovery with different IPv4 and IPv6 ports");
-                } else if (const auto port = port4 ? port4 : port6) {
-                    peerDiscovery_->startPublish(dht_->getNodeId(), netId, port);
+                // IPv4
+                if (const auto& bound4 = socket->getBoundRef(AF_INET)) {
+                    adc.port = bound4.getPort();
+                    msgpack::pack(sbuf_node, adc);
+                    peerDiscovery_->startPublish(AF_INET, PEER_DISCOVERY_DHT_SERVICE, sbuf_node);
+                }
+                // IPv6
+                if (const auto& bound6 = socket->getBoundRef(AF_INET6)) {
+                    adc.port = bound6.getPort();
+                    sbuf_node.clear();
+                    msgpack::pack(sbuf_node, adc);
+                    peerDiscovery_->startPublish(AF_INET6, PEER_DISCOVERY_DHT_SERVICE, sbuf_node);
                 }
             }
         }

src/http.cpp

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

src/peer_discovery.cpp

@@ -0,0 +1,474 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+
+#include "peer_discovery.h"
+#include "network_utils.h"
+#include "utils.h"
+
+#ifdef __ANDROID__
+#include "compat/getif_workaround_android.h"
+#endif
+
+#include <asio.hpp>
+
+using namespace std::literals;
+
+namespace dht {
+
+// Organization-local Scope multicast
+constexpr char MULTICAST_ADDRESS_IPV4[] = "239.192.0.1";
+constexpr char MULTICAST_ADDRESS_IPV6[] = "ff08::101";
+
+class PeerDiscovery::DomainPeerDiscovery
+{
+public:
+    DomainPeerDiscovery(asio::ip::udp domain,
+                        in_port_t port,
+                        Sp<asio::io_context> ioContext = {},
+                        Sp<Logger> logger = {});
+    ~DomainPeerDiscovery();
+
+    void startDiscovery(std::string_view type, ServiceDiscoveredCallback callback);
+    void startPublish(std::string_view type, const msgpack::sbuffer& pack_buf);
+
+    void stop();
+
+    bool stopDiscovery(std::string_view type);
+    bool stopPublish(std::string_view type);
+
+    void connectivityChanged();
+
+    void stopConnectivityChanged();
+
+private:
+    Sp<Logger> logger_;
+    // dmtx_ for callbackmap_ and drunning_ (write)
+    std::mutex dmtx_;
+    // mtx_ for messages_ and lrunning (listen)
+    std::mutex mtx_;
+    std::shared_ptr<asio::io_context> ioContext_;
+
+    static constexpr dht::duration PeerDiscovery_PERIOD_MAX {std::chrono::minutes(1)};
+    static constexpr std::chrono::seconds PeerDiscovery_PERIOD {10};
+    asio::steady_timer peerDiscoveryTimer;
+    std::chrono::steady_clock::duration peerDiscovery_period {PeerDiscovery_PERIOD};
+
+    asio::ip::udp::socket sockFd_;
+    asio::ip::udp::endpoint sockAddrSend_;
+
+    std::array<char, 64 * 1024> receiveBuf_;
+    asio::ip::udp::endpoint receiveFrom_;
+
+    msgpack::sbuffer sbuf_;
+    std::map<std::string, msgpack::sbuffer, std::less<>> messages_;
+    std::map<std::string, ServiceDiscoveredCallback, std::less<>> callbackmap_;
+    bool lrunning_ {false};
+    bool drunning_ {false};
+
+    void loopListener();
+    void query(const asio::ip::udp::endpoint& peer);
+    void reloadMessages();
+
+    void stopDiscovery();
+    void stopPublish();
+
+    void publish(const asio::ip::udp::endpoint& peer);
+
+    void reDiscover();
+};
+
+PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain,
+                                                        in_port_t port,
+                                                        Sp<asio::io_context> ioContext,
+                                                        Sp<Logger> logger)
+    : logger_(logger)
+    , ioContext_(ioContext)
+    , peerDiscoveryTimer(*ioContext_)
+    , sockFd_(*ioContext_, domain)
+    , sockAddrSend_(asio::ip::make_address(domain.family() == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6),
+                    port)
+{
+    try {
+        sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address()));
+        sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
+        sockFd_.bind({domain, port});
+    } catch (const std::exception& e) {
+#ifdef __ANDROID__
+        if (domain.family() == AF_INET && strcmp(e.what(), "set_option: No such device") == 0) {
+            try {
+                sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
+                auto mc_interface = workaround::get_interface();
+                sockFd_.set_option(asio::ip::multicast::outbound_interface(mc_interface));
+                sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address().to_v4(), mc_interface));
+                sockFd_.bind({domain, port});
+            } catch (const std::exception& e) {
+                if (logger_)
+                    logger_->error("Unable to start peer discovery using android workaround: {}", e.what());
+            }
+        } else
+#endif
+            if (logger_)
+            logger_->error("Unable to start peer discovery for {}: {}",
+                           domain.family() == AF_INET ? "IPv4" : "IPv6",
+                           e.what());
+    }
+}
+
+PeerDiscovery::DomainPeerDiscovery::~DomainPeerDiscovery()
+{
+    stop();
+    sockFd_.close();
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::startDiscovery(std::string_view type, ServiceDiscoveredCallback callback)
+{
+    std::lock_guard lck(dmtx_);
+    callbackmap_[std::string(type)] = callback;
+    if (not drunning_) {
+        drunning_ = true;
+        asio::post(*ioContext_, [this]() {
+            loopListener();
+            query(sockAddrSend_);
+        });
+    }
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::loopListener()
+{
+    std::lock_guard lck(dmtx_);
+    if (not drunning_)
+        return;
+    sockFd_.async_receive_from(asio::buffer(receiveBuf_),
+                               receiveFrom_,
+                               [this](const asio::error_code& error, size_t bytes) {
+                                   if (error == asio::error::operation_aborted)
+                                       return;
+                                   if (error) {
+                                       if (logger_)
+                                           logger_->error("Error receiving message: {}", error.message());
+                                   }
+                                   try {
+                                       auto rcv = msgpack::unpack(receiveBuf_.data(), bytes);
+                                       msgpack::object obj = rcv.get();
+
+                                       if (obj.type == msgpack::type::STR) {
+                                           if (lrunning_ and obj.as<std::string_view>() == "q"sv)
+                                               publish(receiveFrom_);
+                                       } else if (obj.type == msgpack::type::MAP) {
+                                           for (unsigned i = 0; i < obj.via.map.size; i++) {
+                                               auto& o = obj.via.map.ptr[i];
+                                               if (o.key.type != msgpack::type::STR)
+                                                   continue;
+                                               ServiceDiscoveredCallback cb;
+                                               {
+                                                   std::lock_guard lck(dmtx_);
+                                                   if (drunning_) {
+                                                       auto callback = callbackmap_.find(o.key.as<std::string_view>());
+                                                       if (callback != callbackmap_.end())
+                                                           cb = callback->second;
+                                                   } else
+                                                       return;
+                                               }
+                                               if (cb)
+                                                   cb(std::move(o.val),
+                                                      SockAddr {receiveFrom_.data(), (socklen_t) receiveFrom_.size()});
+                                           }
+                                       } else {
+                                           throw msgpack::type_error {};
+                                       }
+                                   } catch (const std::exception& e) {
+                                       if (logger_)
+                                           logger_->error("Error receiving packet: {}", e.what());
+                                   }
+                                   loopListener();
+                               });
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::query(const asio::ip::udp::endpoint& peer)
+{
+    std::lock_guard lck(dmtx_);
+    if (not drunning_)
+        return;
+
+    msgpack::sbuffer pbuf_request;
+    msgpack::pack(pbuf_request, "q");
+
+    sockFd_.async_send_to(asio::buffer(pbuf_request.data(), pbuf_request.size()),
+                          peer,
+                          [logger = logger_, peer](const asio::error_code& ec, size_t) {
+                              if (ec and (ec != asio::error::operation_aborted) and logger)
+                                  logger->warn("Error sending packet to: {} with err: {}",
+                                               peer.address().to_string(),
+                                               ec.message());
+                          });
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::publish(const asio::ip::udp::endpoint& peer)
+{
+    std::lock_guard lck(mtx_);
+    if (not lrunning_)
+        return;
+
+    sockFd_.async_send_to(asio::buffer((const void*) sbuf_.data(), sbuf_.size()),
+                          peer,
+                          [logger = logger_, peer](const asio::error_code& ec, size_t) {
+                              if (ec and (ec != asio::error::operation_aborted) and logger)
+                                  logger->warn("Error sending packet to: {} with err: {}",
+                                               peer.address().to_string(),
+                                               ec.message());
+                          });
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::startPublish(std::string_view type, const msgpack::sbuffer& pack_buf)
+{
+    msgpack::sbuffer pack_buf_c(pack_buf.size());
+    pack_buf_c.write(pack_buf.data(), pack_buf.size());
+
+    std::lock_guard lck(mtx_);
+    messages_[std::string(type)] = std::move(pack_buf_c);
+    reloadMessages();
+    lrunning_ = true;
+    asio::post(*ioContext_, [this]() { publish(sockAddrSend_); });
+}
+
+bool
+PeerDiscovery::DomainPeerDiscovery::stopDiscovery(std::string_view type)
+{
+    std::lock_guard lck(dmtx_);
+    auto it = callbackmap_.find(type);
+    if (it != callbackmap_.end()) {
+        callbackmap_.erase(it);
+        if (callbackmap_.empty())
+            stopDiscovery();
+        return true;
+    }
+    return false;
+}
+
+bool
+PeerDiscovery::DomainPeerDiscovery::stopPublish(std::string_view type)
+{
+    std::lock_guard lck(mtx_);
+    auto it = messages_.find(type);
+    if (it != messages_.end()) {
+        messages_.erase(it);
+        if (messages_.empty())
+            stopPublish();
+        else
+            reloadMessages();
+        return true;
+    }
+    return false;
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::stopDiscovery()
+{
+    drunning_ = false;
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::stopPublish()
+{
+    lrunning_ = false;
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::stop()
+{
+    {
+        std::lock_guard lck(dmtx_);
+        stopDiscovery();
+    }
+    {
+        std::lock_guard lck(mtx_);
+        stopPublish();
+    }
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::reloadMessages()
+{
+    sbuf_.clear();
+    msgpack::packer<msgpack::sbuffer> pk(&sbuf_);
+    pk.pack_map(messages_.size());
+    for (const auto& m : messages_) {
+        pk.pack(m.first);
+        sbuf_.write(m.second.data(), m.second.size());
+    }
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::reDiscover()
+{
+    asio::error_code ec;
+
+    sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address()), ec);
+    if (ec and logger_)
+        logger_->warn("Unable to multicast on {}: {}", sockAddrSend_.address().to_string(), ec.message());
+    query(sockAddrSend_);
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::connectivityChanged()
+{
+    asio::post(*ioContext_, [this]() {
+        reDiscover();
+        publish(sockAddrSend_);
+    });
+    if (logger_)
+        logger_->debug("PeerDiscovery: connectivity changed");
+
+    if (peerDiscovery_period == PeerDiscovery_PERIOD_MAX) {
+        peerDiscovery_period = PeerDiscovery_PERIOD;
+    } else {
+        peerDiscoveryTimer.expires_after(peerDiscovery_period);
+        peerDiscoveryTimer.async_wait([this](const asio::error_code& ec) {
+            if (ec == asio::error::operation_aborted)
+                return;
+            connectivityChanged();
+        });
+        peerDiscovery_period = std::min(peerDiscovery_period * 2, PeerDiscovery_PERIOD_MAX);
+    }
+}
+
+void
+PeerDiscovery::DomainPeerDiscovery::stopConnectivityChanged()
+{
+    peerDiscoveryTimer.cancel();
+    peerDiscovery_period = PeerDiscovery_PERIOD;
+}
+
+PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
+{
+    if (not ioContext) {
+        ioContext = std::make_shared<asio::io_context>();
+        ioContext_ = ioContext;
+        ioRunnner_ = std::thread([logger, ioContext] {
+            try {
+                if (logger)
+                    logger->d("[peerdiscovery] starting io_context");
+                auto work = asio::make_work_guard(*ioContext);
+                ioContext->run();
+                if (logger)
+                    logger->d("[peerdiscovery] io_context stopped");
+            } catch (const std::exception& ex) {
+                if (logger)
+                    logger->e("[peerdiscovery] run error: %s", ex.what());
+            }
+        });
+    }
+
+    try {
+        peerDiscovery4_.reset(new DomainPeerDiscovery(asio::ip::udp::v4(), port, ioContext, logger));
+    } catch (const std::exception& e) {
+        if (logger)
+            logger->e("[peerdiscovery] Unable to start IPv4: %s", e.what());
+    }
+    try {
+        peerDiscovery6_.reset(new DomainPeerDiscovery(asio::ip::udp::v6(), port, ioContext, logger));
+    } catch (const std::exception& e) {
+        if (logger)
+            logger->e("[peerdiscovery] Unable to start IPv6: %s", e.what());
+    }
+}
+
+PeerDiscovery::~PeerDiscovery()
+{
+    stop();
+    if (ioContext_)
+        ioContext_->stop();
+    if (ioRunnner_.joinable())
+        ioRunnner_.join();
+}
+
+void
+PeerDiscovery::startDiscovery(std::string_view type, ServiceDiscoveredCallback callback)
+{
+    if (peerDiscovery4_)
+        peerDiscovery4_->startDiscovery(type, callback);
+    if (peerDiscovery6_)
+        peerDiscovery6_->startDiscovery(type, callback);
+}
+
+void
+PeerDiscovery::startPublish(std::string_view type, const msgpack::sbuffer& pack_buf)
+{
+    if (peerDiscovery4_)
+        peerDiscovery4_->startPublish(type, pack_buf);
+    if (peerDiscovery6_)
+        peerDiscovery6_->startPublish(type, pack_buf);
+}
+
+void
+PeerDiscovery::startPublish(sa_family_t domain, std::string_view type, const msgpack::sbuffer& pack_buf)
+{
+    if (domain == AF_INET) {
+        if (peerDiscovery4_)
+            peerDiscovery4_->startPublish(type, pack_buf);
+    } else if (domain == AF_INET6) {
+        if (peerDiscovery6_)
+            peerDiscovery6_->startPublish(type, pack_buf);
+    }
+}
+
+void
+PeerDiscovery::stop()
+{
+    if (peerDiscovery4_)
+        peerDiscovery4_->stop();
+    if (peerDiscovery6_)
+        peerDiscovery6_->stop();
+}
+
+bool
+PeerDiscovery::stopDiscovery(std::string_view type)
+{
+    bool stopped4 = peerDiscovery4_ and peerDiscovery4_->stopDiscovery(type);
+    bool stopped6 = peerDiscovery6_ and peerDiscovery6_->stopDiscovery(type);
+    return stopped4 or stopped6;
+}
+
+bool
+PeerDiscovery::stopPublish(std::string_view type)
+{
+    bool stopped4 = peerDiscovery4_ and peerDiscovery4_->stopPublish(type);
+    bool stopped6 = peerDiscovery6_ and peerDiscovery6_->stopPublish(type);
+    return stopped4 or stopped6;
+}
+
+bool
+PeerDiscovery::stopPublish(sa_family_t domain, std::string_view type)
+{
+    if (domain == AF_INET) {
+        return peerDiscovery4_ and peerDiscovery4_->stopPublish(type);
+    } else if (domain == AF_INET6) {
+        return peerDiscovery6_ and peerDiscovery6_->stopPublish(type);
+    }
+    return false;
+}
+
+void
+PeerDiscovery::connectivityChanged()
+{
+    if (peerDiscovery4_)
+        peerDiscovery4_->connectivityChanged();
+    if (peerDiscovery6_)
+        peerDiscovery6_->connectivityChanged();
+}
+
+void
+PeerDiscovery::stopConnectivityChanged()
+{
+    if (peerDiscovery4_)
+        peerDiscovery4_->stopConnectivityChanged();
+    if (peerDiscovery6_)
+        peerDiscovery6_->stopConnectivityChanged();
+}
+
+} /* namespace dht */

src/peer_discovery/dns_sd.cpp

@@ -1,504 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-
-#include "dns_sd.h"
-
-#include <algorithm>
-#include <charconv>
-#include <map>
-
-namespace dht::mdns::dns_sd {
-namespace {
-
-constexpr std::string_view SERVICE_NAME = "_opendht._udp.local.";
-constexpr std::string_view SERVICE_ENUMERATION_NAME = "_services._dns-sd._udp.local.";
-constexpr std::string_view SERVICE_SUFFIX = "._opendht._udp.local.";
-constexpr std::string_view HOST_SUFFIX = ".local.";
-
-std::vector<uint8_t>
-txtString(std::string value)
-{
-    return {value.begin(), value.end()};
-}
-
-std::string
-asciiLower(std::string_view value)
-{
-    std::string lower;
-    lower.reserve(value.size());
-    for (const auto character : value) {
-        if (character >= 'A' and character <= 'Z')
-            lower.push_back(static_cast<char>(character + ('a' - 'A')));
-        else
-            lower.push_back(character);
-    }
-    return lower;
-}
-
-void
-validateService(const Service& service)
-{
-    if (not service.nodeId)
-        throw ParseError("DNS-SD service requires a node ID");
-    if (service.port == 0)
-        throw ParseError("DNS-SD service requires a port");
-    if (service.addresses.empty())
-        throw ParseError("DNS-SD service requires an address");
-}
-
-Message
-makeAdvertisement(const Service& service, uint32_t serviceTtl, uint32_t hostTtl)
-{
-    validateService(service);
-    const auto instance = instanceName(service.nodeId);
-    const auto host = hostName(service.nodeId);
-
-    Message message;
-    message.response = true;
-    message.authoritative = true;
-    message.answers.push_back({serviceName(), Type::PTR, CLASS_IN, false, serviceTtl, PtrData {instance}});
-    message.additionals.push_back({
-        instance, Type::SRV, CLASS_IN, true, hostTtl, SrvData {0, 0, service.port, host}
-    });
-
-    TxtData txt;
-    txt.strings.emplace_back(txtString("txtvers=1"));
-    txt.strings.emplace_back(txtString("id=" + service.nodeId.toString()));
-    txt.strings.emplace_back(txtString("net=" + std::to_string(service.network)));
-    message.additionals.push_back({instance, Type::TXT, CLASS_IN, true, serviceTtl, std::move(txt)});
-
-    for (const auto& address : service.addresses) {
-        std::visit(
-            [&](const auto& data) {
-                using Data = std::decay_t<decltype(data)>;
-                constexpr auto type = std::is_same_v<Data, AData> ? Type::A : Type::AAAA;
-                message.additionals.push_back({host, type, CLASS_IN, true, hostTtl, data});
-            },
-            address);
-    }
-    return message;
-}
-
-std::optional<InfoHash>
-nodeIdFromInstance(const Name& instance)
-{
-    const auto& labels = instance.labels();
-    const auto service = serviceName();
-    const auto& serviceLabels = service.labels();
-    if (labels.size() != serviceLabels.size() + 1
-        or not std::equal(labels.begin() + 1, labels.end(), serviceLabels.begin(), serviceLabels.end()))
-        return {};
-    if (labels.front().size() != 40)
-        return {};
-    const InfoHash nodeId {labels.front()};
-    if (not nodeId)
-        return {};
-    return nodeId;
-}
-
-struct TxtValues
-{
-    InfoHash nodeId;
-    NetId network {};
-};
-
-std::optional<TxtValues>
-parseTxt(const TxtData& txt)
-{
-    std::map<std::string, std::string> values;
-    for (const auto& raw : txt.strings) {
-        const std::string_view string {reinterpret_cast<const char*>(raw.data()), raw.size()};
-        const auto separator = string.find('=');
-        if (separator == std::string_view::npos or separator == 0)
-            continue;
-        const auto keyView = string.substr(0, separator);
-        if (std::any_of(keyView.begin(), keyView.end(), [](char value) {
-                return value < 0x20 or value > 0x7e or value == '=';
-            }))
-            continue;
-        const auto key = asciiLower(keyView);
-        if ((key == "txtvers" or key == "id" or key == "net") and values.count(key))
-            return {};
-        values.emplace(key, std::string(string.substr(separator + 1)));
-    }
-
-    if (values.size() < 3 or values["txtvers"] != "1" or values["id"].size() != 40)
-        return {};
-    TxtValues parsed;
-    parsed.nodeId = InfoHash {values["id"]};
-    if (not parsed.nodeId)
-        return {};
-    const auto& network = values["net"];
-    const auto result = std::from_chars(network.data(), network.data() + network.size(), parsed.network);
-    if (result.ec != std::errc {} or result.ptr != network.data() + network.size())
-        return {};
-    return parsed;
-}
-
-template<typename Callback>
-void
-forEachRecord(const Message& message, Callback&& callback)
-{
-    for (const auto& record : message.answers)
-        callback(record);
-    for (const auto& record : message.authorities)
-        callback(record);
-    for (const auto& record : message.additionals)
-        callback(record);
-}
-
-bool
-matches(const Question& question, const ResourceRecord& record)
-{
-    return question.name == record.name and (question.type == Type::ANY or question.type == record.type)
-           and (question.classCode == CLASS_ANY or question.classCode == record.classCode);
-}
-
-bool
-sameRecord(const ResourceRecord& lhs, const ResourceRecord& rhs)
-{
-    return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode and lhs.data == rhs.data;
-}
-
-bool
-isKnownAnswer(const Message& query, const ResourceRecord& record)
-{
-    return std::any_of(query.answers.begin(), query.answers.end(), [&](const ResourceRecord& known) {
-        return sameRecord(known, record) and known.ttl > record.ttl / 2;
-    });
-}
-
-} // namespace
-
-class Cache::Impl
-{
-public:
-    explicit Impl(Now now)
-        : now_(std::move(now))
-    {}
-
-    struct Entry
-    {
-        ResourceRecord record;
-        Clock::time_point expires;
-    };
-
-    std::vector<Service> update(const Message& message)
-    {
-        expire();
-        std::vector<const ResourceRecord*> records;
-        forEachRecord(message, [&](const ResourceRecord& record) { records.push_back(&record); });
-
-        std::vector<const ResourceRecord*> flushed;
-        for (const auto* record : records) {
-            if (not record->cacheFlush)
-                continue;
-            const auto alreadyFlushed = std::any_of(flushed.begin(), flushed.end(), [&](const ResourceRecord* other) {
-                return other->name == record->name and other->type == record->type
-                       and other->classCode == record->classCode;
-            });
-            if (not alreadyFlushed) {
-                flushed.push_back(record);
-                entries_.erase(std::remove_if(entries_.begin(),
-                                              entries_.end(),
-                                              [&](const Entry& entry) {
-                                                  return entry.record.name == record->name
-                                                         and entry.record.type == record->type
-                                                         and entry.record.classCode == record->classCode;
-                                              }),
-                               entries_.end());
-            }
-        }
-
-        for (const auto* record : records) {
-            if (record->ttl == 0) {
-                entries_.erase(std::remove_if(entries_.begin(),
-                                              entries_.end(),
-                                              [&](const Entry& entry) { return sameRecord(entry.record, *record); }),
-                               entries_.end());
-                continue;
-            }
-            const auto expires = now_() + std::chrono::seconds(record->ttl);
-            const auto existing = std::find_if(entries_.begin(), entries_.end(), [&](const Entry& entry) {
-                return sameRecord(entry.record, *record);
-            });
-            if (existing == entries_.end())
-                entries_.push_back({*record, expires});
-            else {
-                existing->record = *record;
-                existing->expires = expires;
-            }
-        }
-
-        auto current = collect();
-        std::vector<Service> discovered;
-        for (const auto& service : current) {
-            if (std::find_if(emitted_.begin(),
-                             emitted_.end(),
-                             [&](const Service& previous) { return equal(previous, service); })
-                == emitted_.end())
-                discovered.push_back(service);
-        }
-        emitted_ = std::move(current);
-        return discovered;
-    }
-
-    std::vector<Service> services()
-    {
-        expire();
-        auto current = collect();
-        emitted_ = current;
-        return current;
-    }
-
-    void clear()
-    {
-        entries_.clear();
-        emitted_.clear();
-    }
-
-private:
-    static bool equal(const Service& lhs, const Service& rhs)
-    {
-        return lhs.nodeId == rhs.nodeId and lhs.network == rhs.network and lhs.port == rhs.port
-               and lhs.addresses == rhs.addresses;
-    }
-
-    void expire()
-    {
-        const auto now = now_();
-        entries_.erase(std::remove_if(entries_.begin(),
-                                      entries_.end(),
-                                      [&](const Entry& entry) { return entry.expires <= now; }),
-                       entries_.end());
-    }
-
-    std::vector<Service> collect() const
-    {
-        std::vector<Service> result;
-        for (const auto& entry : entries_) {
-            if (entry.record.type != Type::PTR or entry.record.classCode != CLASS_IN
-                or entry.record.name != serviceName())
-                continue;
-            const auto* ptr = std::get_if<PtrData>(&entry.record.data);
-            if (not ptr)
-                continue;
-
-            Message message;
-            message.response = true;
-            message.answers.push_back(entry.record);
-            for (const auto& additional : entries_) {
-                if (&additional != &entry and additional.record.type != Type::PTR)
-                    message.additionals.push_back(additional.record);
-            }
-            const auto service = resolve(message);
-            if (service and std::none_of(result.begin(), result.end(), [&](const Service& existing) {
-                    return equal(existing, *service);
-                }))
-                result.push_back(*service);
-        }
-        return result;
-    }
-
-    Now now_;
-    std::vector<Entry> entries_;
-    std::vector<Service> emitted_;
-};
-
-Cache::Cache(Now now)
-    : impl_(std::make_unique<Impl>(std::move(now)))
-{}
-
-Cache::~Cache() = default;
-Cache::Cache(Cache&&) noexcept = default;
-Cache& Cache::operator=(Cache&&) noexcept = default;
-
-std::vector<Service>
-Cache::update(const Message& message)
-{
-    return impl_->update(message);
-}
-
-std::vector<Service>
-Cache::services()
-{
-    return impl_->services();
-}
-
-void
-Cache::clear()
-{
-    impl_->clear();
-}
-
-Name
-serviceName()
-{
-    return Name::fromString(SERVICE_NAME);
-}
-
-Name
-instanceName(const InfoHash& nodeId)
-{
-    if (not nodeId)
-        throw ParseError("DNS-SD instance requires a node ID");
-    return Name::fromString(nodeId.toString() + std::string(SERVICE_SUFFIX));
-}
-
-Name
-hostName(const InfoHash& nodeId)
-{
-    if (not nodeId)
-        throw ParseError("mDNS host requires a node ID");
-    return Name::fromString(nodeId.toString() + std::string(HOST_SUFFIX));
-}
-
-Message
-announcement(const Service& service)
-{
-    return makeAdvertisement(service, SERVICE_RECORD_TTL, HOST_RECORD_TTL);
-}
-
-Message
-goodbye(const Service& service)
-{
-    return makeAdvertisement(service, 0, 0);
-}
-
-Message
-browseQuery()
-{
-    Message query;
-    query.questions.push_back({serviceName(), Type::PTR, CLASS_IN, false});
-    return query;
-}
-
-std::optional<Message>
-respond(const Message& query, const Service& service)
-{
-    if (query.response)
-        return {};
-
-    const auto advertisement = announcement(service);
-    Message response;
-    response.id = query.id;
-    response.response = true;
-    response.authoritative = true;
-
-    const ResourceRecord serviceType {
-        Name::fromString(SERVICE_ENUMERATION_NAME),
-        Type::PTR,
-        CLASS_IN,
-        false,
-        SERVICE_RECORD_TTL,
-        PtrData {serviceName()},
-    };
-
-    auto addMatching = [&](const ResourceRecord& record) {
-        const auto question = std::find_if(query.questions.begin(), query.questions.end(), [&](const Question& q) {
-            return matches(q, record);
-        });
-        if (question == query.questions.end() or (not question->unicastResponse and isKnownAnswer(query, record)))
-            return;
-        if (std::find_if(response.answers.begin(),
-                         response.answers.end(),
-                         [&](const ResourceRecord& answer) { return sameRecord(answer, record); })
-            == response.answers.end())
-            response.answers.push_back(record);
-    };
-
-    addMatching(serviceType);
-    for (const auto& record : advertisement.answers)
-        addMatching(record);
-    for (const auto& record : advertisement.additionals)
-        addMatching(record);
-    if (response.answers.empty())
-        return {};
-
-    const auto metaOnly = std::all_of(response.answers.begin(), response.answers.end(), [&](const auto& answer) {
-        return sameRecord(answer, serviceType);
-    });
-    if (metaOnly)
-        return response;
-
-    response.additionals = advertisement.additionals;
-    response.additionals.erase(std::remove_if(response.additionals.begin(),
-                                              response.additionals.end(),
-                                              [&](const ResourceRecord& record) {
-                                                  return std::any_of(response.answers.begin(),
-                                                                     response.answers.end(),
-                                                                     [&](const ResourceRecord& answer) {
-                                                                         return sameRecord(answer, record);
-                                                                     });
-                                              }),
-                               response.additionals.end());
-    return response;
-}
-
-std::optional<Service>
-resolve(const Message& message)
-{
-    std::vector<Name> instances;
-    forEachRecord(message, [&](const ResourceRecord& record) {
-        if (record.type == Type::PTR and record.classCode == CLASS_IN and record.name == serviceName()) {
-            if (const auto ptr = std::get_if<PtrData>(&record.data))
-                instances.push_back(ptr->name);
-        }
-    });
-    if (instances.size() != 1)
-        return {};
-
-    const auto nodeId = nodeIdFromInstance(instances.front());
-    if (not nodeId)
-        return {};
-    const auto expectedHost = hostName(*nodeId);
-    const SrvData* srv = nullptr;
-    const TxtData* txt = nullptr;
-    std::vector<AddressData> addresses;
-    auto duplicate = false;
-
-    forEachRecord(message, [&](const ResourceRecord& record) {
-        if (record.classCode != CLASS_IN)
-            return;
-        if (record.name == instances.front()) {
-            if (record.type == Type::SRV) {
-                if (srv)
-                    duplicate = true;
-                else
-                    srv = std::get_if<SrvData>(&record.data);
-            } else if (record.type == Type::TXT) {
-                if (txt)
-                    duplicate = true;
-                else
-                    txt = std::get_if<TxtData>(&record.data);
-            }
-        } else if (record.name == expectedHost) {
-            if (record.type == Type::A) {
-                if (const auto data = std::get_if<AData>(&record.data))
-                    addresses.emplace_back(*data);
-            } else if (record.type == Type::AAAA) {
-                if (const auto data = std::get_if<AaaaData>(&record.data))
-                    addresses.emplace_back(*data);
-            }
-        }
-    });
-
-    if (duplicate or not srv or not txt or srv->priority != 0 or srv->weight != 0 or srv->port == 0
-        or srv->target != expectedHost or addresses.empty())
-        return {};
-    const auto txtValues = parseTxt(*txt);
-    if (not txtValues or txtValues->nodeId != *nodeId)
-        return {};
-
-    Service service;
-    service.nodeId = *nodeId;
-    service.network = txtValues->network;
-    service.port = srv->port;
-    for (auto& address : addresses) {
-        if (std::find(service.addresses.begin(), service.addresses.end(), address) == service.addresses.end())
-            service.addresses.emplace_back(std::move(address));
-    }
-    return service;
-}
-
-} // namespace dht::mdns::dns_sd

src/peer_discovery/dns_sd.h

@@ -1,61 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-#pragma once
-
-#include "mdns.h"
-
-#include "opendht/infohash.h"
-#include "opendht/utils.h"
-
-#include <chrono>
-#include <functional>
-#include <memory>
-#include <optional>
-
-namespace dht::mdns::dns_sd {
-
-constexpr uint32_t HOST_RECORD_TTL = 120;
-constexpr uint32_t SERVICE_RECORD_TTL = 4500;
-
-using AddressData = std::variant<AData, AaaaData>;
-
-struct Service
-{
-    InfoHash nodeId;
-    NetId network {};
-    in_port_t port {};
-    std::vector<AddressData> addresses;
-};
-
-class OPENDHT_PUBLIC Cache
-{
-public:
-    using Clock = std::chrono::steady_clock;
-    using Now = std::function<Clock::time_point()>;
-
-    explicit Cache(Now now = [] { return Clock::now(); });
-    ~Cache();
-
-    Cache(Cache&&) noexcept;
-    Cache& operator=(Cache&&) noexcept;
-
-    std::vector<Service> update(const Message& message);
-    std::vector<Service> services();
-    void clear();
-
-private:
-    class Impl;
-    std::unique_ptr<Impl> impl_;
-};
-
-OPENDHT_PUBLIC Name serviceName();
-OPENDHT_PUBLIC Name instanceName(const InfoHash& nodeId);
-OPENDHT_PUBLIC Name hostName(const InfoHash& nodeId);
-
-OPENDHT_PUBLIC Message announcement(const Service& service);
-OPENDHT_PUBLIC Message goodbye(const Service& service);
-OPENDHT_PUBLIC Message browseQuery();
-OPENDHT_PUBLIC std::optional<Message> respond(const Message& query, const Service& service);
-OPENDHT_PUBLIC std::optional<Service> resolve(const Message& message);
-
-} // namespace dht::mdns::dns_sd

src/peer_discovery/mdns.cpp

@@ -1,563 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-
-#include "mdns.h"
-
-#include <algorithm>
-#include <cctype>
-#include <limits>
-#include <map>
-#include <unordered_map>
-
-namespace dht::mdns {
-namespace {
-
-constexpr uint16_t FLAG_RESPONSE = 0x8000;
-constexpr uint16_t FLAG_AUTHORITATIVE = 0x0400;
-constexpr uint16_t FLAG_TRUNCATED = 0x0200;
-constexpr uint16_t CLASS_TOP_BIT = 0x8000;
-constexpr uint16_t CLASS_MASK = 0x7fff;
-
-char
-asciiLower(char value)
-{
-    const auto byte = static_cast<unsigned char>(value);
-    return static_cast<char>(std::tolower(byte));
-}
-
-void
-validateLabels(const std::vector<std::string>& labels)
-{
-    size_t wireSize = 0;
-    for (const auto& label : labels) {
-        if (label.empty() or label.size() > 63)
-            throw ParseError("invalid DNS label length");
-        wireSize += label.size() + 1;
-        if (wireSize > 255)
-            throw ParseError("DNS name is too long");
-    }
-}
-
-std::string
-suffixKey(const std::vector<std::string>& labels, size_t first)
-{
-    std::string key;
-    for (auto index = first; index < labels.size(); ++index) {
-        key.push_back(static_cast<char>(labels[index].size()));
-        key.append(labels[index]);
-    }
-    key.push_back('\0');
-    return key;
-}
-
-class Writer
-{
-public:
-    void write8(uint8_t value)
-    {
-        ensure(1);
-        packet_.push_back(value);
-    }
-
-    void write16(uint16_t value)
-    {
-        ensure(2);
-        packet_.push_back(static_cast<uint8_t>(value >> 8));
-        packet_.push_back(static_cast<uint8_t>(value));
-    }
-
-    void write32(uint32_t value)
-    {
-        write16(static_cast<uint16_t>(value >> 16));
-        write16(static_cast<uint16_t>(value));
-    }
-
-    template<size_t Size>
-    void write(const std::array<uint8_t, Size>& bytes)
-    {
-        write(bytes.data(), bytes.size());
-    }
-
-    void write(const uint8_t* bytes, size_t size)
-    {
-        ensure(size);
-        packet_.insert(packet_.end(), bytes, bytes + size);
-    }
-
-    void writeName(const Name& name)
-    {
-        const auto& labels = name.labels();
-        for (size_t index = 0; index < labels.size(); ++index) {
-            const auto key = suffixKey(labels, index);
-            if (const auto found = suffixes_.find(key); found != suffixes_.end()) {
-                write16(static_cast<uint16_t>(0xc000 | found->second));
-                return;
-            }
-            if (packet_.size() < 0x4000)
-                suffixes_.emplace(key, static_cast<uint16_t>(packet_.size()));
-            write8(static_cast<uint8_t>(labels[index].size()));
-            write(reinterpret_cast<const uint8_t*>(labels[index].data()), labels[index].size());
-        }
-        write8(0);
-    }
-
-    size_t size() const noexcept { return packet_.size(); }
-
-    void patch16(size_t offset, uint16_t value)
-    {
-        if (offset + 2 > packet_.size())
-            throw ParseError("invalid DNS packet patch offset");
-        packet_[offset] = static_cast<uint8_t>(value >> 8);
-        packet_[offset + 1] = static_cast<uint8_t>(value);
-    }
-
-    Packet take() { return std::move(packet_); }
-
-private:
-    void ensure(size_t additional) const
-    {
-        if (additional > MAX_PACKET_SIZE - packet_.size())
-            throw ParseError("mDNS packet exceeds maximum size");
-    }
-
-    Packet packet_;
-    std::unordered_map<std::string, uint16_t> suffixes_;
-};
-
-class Reader
-{
-public:
-    Reader(const uint8_t* packet, size_t size)
-        : packet_(packet)
-        , size_(size)
-    {}
-
-    uint8_t read8()
-    {
-        require(1);
-        return packet_[position_++];
-    }
-
-    uint16_t read16()
-    {
-        const auto high = read8();
-        const auto low = read8();
-        return static_cast<uint16_t>((high << 8) | low);
-    }
-
-    uint32_t read32() { return (static_cast<uint32_t>(read16()) << 16) | read16(); }
-
-    Name readName()
-    {
-        std::vector<std::string> labels;
-        std::vector<size_t> visited;
-        auto cursor = position_;
-        auto jumped = false;
-        size_t wireSize = 0;
-
-        while (true) {
-            if (cursor >= size_)
-                throw ParseError("truncated DNS name");
-            if (std::find(visited.begin(), visited.end(), cursor) != visited.end())
-                throw ParseError("DNS compression pointer loop");
-            visited.push_back(cursor);
-
-            const auto length = packet_[cursor];
-            if ((length & 0xc0) == 0xc0) {
-                if (cursor + 1 >= size_)
-                    throw ParseError("truncated DNS compression pointer");
-                const auto pointer = static_cast<size_t>(((length & 0x3f) << 8) | packet_[cursor + 1]);
-                if (pointer >= cursor)
-                    throw ParseError("DNS compression pointer does not point backwards");
-                if (not jumped)
-                    position_ = cursor + 2;
-                cursor = pointer;
-                jumped = true;
-                continue;
-            }
-            if (length & 0xc0)
-                throw ParseError("invalid DNS label prefix");
-
-            ++cursor;
-            if (length == 0) {
-                if (not jumped)
-                    position_ = cursor;
-                break;
-            }
-            if (length > 63 or length > size_ - cursor)
-                throw ParseError("invalid or truncated DNS label");
-            wireSize += length + 1;
-            if (wireSize > 255)
-                throw ParseError("DNS name is too long");
-            labels.emplace_back(reinterpret_cast<const char*>(packet_ + cursor), length);
-            cursor += length;
-            if (not jumped)
-                position_ = cursor;
-        }
-        return Name {std::move(labels)};
-    }
-
-    std::vector<uint8_t> readBytes(size_t size)
-    {
-        require(size);
-        std::vector<uint8_t> bytes(packet_ + position_, packet_ + position_ + size);
-        position_ += size;
-        return bytes;
-    }
-
-    size_t position() const noexcept { return position_; }
-    void setPosition(size_t position)
-    {
-        if (position > size_)
-            throw ParseError("invalid DNS packet position");
-        position_ = position;
-    }
-
-private:
-    void require(size_t count) const
-    {
-        if (count > size_ - position_)
-            throw ParseError("truncated DNS packet");
-    }
-
-    const uint8_t* packet_;
-    size_t size_;
-    size_t position_ {};
-};
-
-uint16_t
-checkedCount(size_t count)
-{
-    if (count > std::numeric_limits<uint16_t>::max())
-        throw ParseError("too many DNS records");
-    return static_cast<uint16_t>(count);
-}
-
-void
-encodeQuestion(Writer& writer, const Question& question)
-{
-    writer.writeName(question.name);
-    writer.write16(static_cast<uint16_t>(question.type));
-    writer.write16(static_cast<uint16_t>(question.classCode | (question.unicastResponse ? CLASS_TOP_BIT : 0)));
-}
-
-void
-encodeNsec(Writer& writer, const NsecData& nsec)
-{
-    writer.writeName(nsec.nextDomain);
-    std::map<uint8_t, std::array<uint8_t, 32>> windows;
-    std::map<uint8_t, uint8_t> lengths;
-    for (const auto type : nsec.types) {
-        const auto value = static_cast<uint16_t>(type);
-        const auto window = static_cast<uint8_t>(value >> 8);
-        const auto low = static_cast<uint8_t>(value);
-        const auto byte = static_cast<uint8_t>(low / 8);
-        const auto bit = static_cast<uint8_t>(low % 8);
-        windows[window][byte] |= static_cast<uint8_t>(1u << (7 - bit));
-        lengths[window] = std::max<uint8_t>(lengths[window], byte + 1);
-    }
-    for (const auto& [window, bitmap] : windows) {
-        writer.write8(window);
-        writer.write8(lengths[window]);
-        writer.write(bitmap.data(), lengths[window]);
-    }
-}
-
-void
-encodeRecord(Writer& writer, const ResourceRecord& record)
-{
-    writer.writeName(record.name);
-    writer.write16(static_cast<uint16_t>(record.type));
-    writer.write16(static_cast<uint16_t>(record.classCode | (record.cacheFlush ? CLASS_TOP_BIT : 0)));
-    writer.write32(record.ttl);
-    const auto lengthOffset = writer.size();
-    writer.write16(0);
-    const auto dataOffset = writer.size();
-
-    switch (record.type) {
-    case Type::A:
-        writer.write(std::get<AData>(record.data).address);
-        break;
-    case Type::AAAA:
-        writer.write(std::get<AaaaData>(record.data).address);
-        break;
-    case Type::PTR:
-        writer.writeName(std::get<PtrData>(record.data).name);
-        break;
-    case Type::TXT:
-        for (const auto& string : std::get<TxtData>(record.data).strings) {
-            if (string.size() > 255)
-                throw ParseError("DNS TXT string is too long");
-            writer.write8(static_cast<uint8_t>(string.size()));
-            writer.write(string.data(), string.size());
-        }
-        break;
-    case Type::SRV: {
-        const auto& srv = std::get<SrvData>(record.data);
-        writer.write16(srv.priority);
-        writer.write16(srv.weight);
-        writer.write16(srv.port);
-        writer.writeName(srv.target);
-        break;
-    }
-    case Type::NSEC:
-        encodeNsec(writer, std::get<NsecData>(record.data));
-        break;
-    default: {
-        const auto& bytes = std::get<UnknownData>(record.data).bytes;
-        writer.write(bytes.data(), bytes.size());
-        break;
-    }
-    }
-
-    const auto dataSize = writer.size() - dataOffset;
-    if (dataSize > std::numeric_limits<uint16_t>::max())
-        throw ParseError("DNS record data is too large");
-    writer.patch16(lengthOffset, static_cast<uint16_t>(dataSize));
-}
-
-Question
-decodeQuestion(Reader& reader)
-{
-    Question question;
-    question.name = reader.readName();
-    question.type = static_cast<Type>(reader.read16());
-    const auto classField = reader.read16();
-    question.classCode = classField & CLASS_MASK;
-    question.unicastResponse = classField & CLASS_TOP_BIT;
-    return question;
-}
-
-NsecData
-decodeNsec(Reader& reader, size_t end)
-{
-    NsecData nsec;
-    nsec.nextDomain = reader.readName();
-    while (reader.position() < end) {
-        const auto window = reader.read8();
-        const auto length = reader.read8();
-        if (length == 0 or length > 32 or length > end - reader.position())
-            throw ParseError("invalid DNS NSEC bitmap");
-        const auto bitmap = reader.readBytes(length);
-        for (size_t byte = 0; byte < bitmap.size(); ++byte) {
-            for (uint8_t bit = 0; bit < 8; ++bit) {
-                if (bitmap[byte] & (1u << (7 - bit))) {
-                    const auto value = static_cast<uint16_t>((window << 8) | (byte * 8 + bit));
-                    nsec.types.push_back(static_cast<Type>(value));
-                }
-            }
-        }
-    }
-    return nsec;
-}
-
-ResourceRecord
-decodeRecord(Reader& reader)
-{
-    ResourceRecord record;
-    record.name = reader.readName();
-    record.type = static_cast<Type>(reader.read16());
-    const auto classField = reader.read16();
-    record.classCode = classField & CLASS_MASK;
-    record.cacheFlush = classField & CLASS_TOP_BIT;
-    record.ttl = reader.read32();
-    const auto dataLength = reader.read16();
-    const auto dataStart = reader.position();
-    const auto dataEnd = dataStart + dataLength;
-    if (dataEnd < dataStart)
-        throw ParseError("invalid DNS record length");
-
-    switch (record.type) {
-    case Type::A: {
-        if (dataLength != 4)
-            throw ParseError("invalid DNS A record length");
-        AData data;
-        const auto bytes = reader.readBytes(data.address.size());
-        std::copy(bytes.begin(), bytes.end(), data.address.begin());
-        record.data = data;
-        break;
-    }
-    case Type::AAAA: {
-        if (dataLength != 16)
-            throw ParseError("invalid DNS AAAA record length");
-        AaaaData data;
-        const auto bytes = reader.readBytes(data.address.size());
-        std::copy(bytes.begin(), bytes.end(), data.address.begin());
-        record.data = data;
-        break;
-    }
-    case Type::PTR:
-        record.data = PtrData {reader.readName()};
-        break;
-    case Type::TXT: {
-        TxtData data;
-        while (reader.position() < dataEnd) {
-            const auto length = reader.read8();
-            if (length > dataEnd - reader.position())
-                throw ParseError("truncated DNS TXT string");
-            data.strings.push_back(reader.readBytes(length));
-        }
-        record.data = std::move(data);
-        break;
-    }
-    case Type::SRV: {
-        if (dataLength < 7)
-            throw ParseError("invalid DNS SRV record length");
-        SrvData data;
-        data.priority = reader.read16();
-        data.weight = reader.read16();
-        data.port = reader.read16();
-        data.target = reader.readName();
-        record.data = std::move(data);
-        break;
-    }
-    case Type::NSEC:
-        record.data = decodeNsec(reader, dataEnd);
-        break;
-    default:
-        record.data = UnknownData {reader.readBytes(dataLength)};
-        break;
-    }
-
-    if (reader.position() != dataEnd)
-        throw ParseError("DNS record data length mismatch");
-    return record;
-}
-
-} // namespace
-
-Name::Name(std::vector<std::string> labels)
-    : labels_(std::move(labels))
-{
-    validateLabels(labels_);
-}
-
-Name
-Name::fromString(std::string_view name)
-{
-    if (name.empty() or name == ".")
-        return {};
-    if (name.back() == '.')
-        name.remove_suffix(1);
-
-    std::vector<std::string> labels;
-    size_t start = 0;
-    while (start <= name.size()) {
-        const auto end = name.find('.', start);
-        const auto length = (end == std::string_view::npos ? name.size() : end) - start;
-        if (length == 0)
-            throw ParseError("empty DNS label");
-        labels.emplace_back(name.substr(start, length));
-        if (end == std::string_view::npos)
-            break;
-        start = end + 1;
-    }
-    return Name {std::move(labels)};
-}
-
-std::string
-Name::toString() const
-{
-    if (labels_.empty())
-        return ".";
-    std::string name;
-    for (const auto& label : labels_) {
-        name.append(label);
-        name.push_back('.');
-    }
-    return name;
-}
-
-bool
-operator==(const Name& lhs, const Name& rhs) noexcept
-{
-    if (lhs.labels_.size() != rhs.labels_.size())
-        return false;
-    for (size_t index = 0; index < lhs.labels_.size(); ++index) {
-        const auto& left = lhs.labels_[index];
-        const auto& right = rhs.labels_[index];
-        if (left.size() != right.size())
-            return false;
-        if (not std::equal(left.begin(), left.end(), right.begin(), [](char a, char b) {
-                return asciiLower(a) == asciiLower(b);
-            }))
-            return false;
-    }
-    return true;
-}
-
-Packet
-encode(const Message& message)
-{
-    const auto recordCount = message.questions.size() + message.answers.size() + message.authorities.size()
-                             + message.additionals.size();
-    if (recordCount > MAX_RECORD_COUNT)
-        throw ParseError("too many DNS records");
-
-    Writer writer;
-    writer.write16(message.id);
-    uint16_t flags = 0;
-    if (message.response)
-        flags |= FLAG_RESPONSE;
-    if (message.authoritative)
-        flags |= FLAG_AUTHORITATIVE;
-    if (message.truncated)
-        flags |= FLAG_TRUNCATED;
-    writer.write16(flags);
-    writer.write16(checkedCount(message.questions.size()));
-    writer.write16(checkedCount(message.answers.size()));
-    writer.write16(checkedCount(message.authorities.size()));
-    writer.write16(checkedCount(message.additionals.size()));
-
-    for (const auto& question : message.questions)
-        encodeQuestion(writer, question);
-    for (const auto& record : message.answers)
-        encodeRecord(writer, record);
-    for (const auto& record : message.authorities)
-        encodeRecord(writer, record);
-    for (const auto& record : message.additionals)
-        encodeRecord(writer, record);
-    return writer.take();
-}
-
-Message
-decode(const uint8_t* packet, size_t size)
-{
-    if (packet == nullptr)
-        throw ParseError("null DNS packet");
-    if (size > MAX_PACKET_SIZE)
-        throw ParseError("mDNS packet exceeds maximum size");
-
-    Reader reader(packet, size);
-    Message message;
-    message.id = reader.read16();
-    const auto flags = reader.read16();
-    message.response = flags & FLAG_RESPONSE;
-    message.authoritative = flags & FLAG_AUTHORITATIVE;
-    message.truncated = flags & FLAG_TRUNCATED;
-    const auto questionCount = reader.read16();
-    const auto answerCount = reader.read16();
-    const auto authorityCount = reader.read16();
-    const auto additionalCount = reader.read16();
-    const auto recordCount = static_cast<size_t>(questionCount) + answerCount + authorityCount + additionalCount;
-    if (recordCount > MAX_RECORD_COUNT)
-        throw ParseError("too many DNS records");
-
-    message.questions.reserve(questionCount);
-    message.answers.reserve(answerCount);
-    message.authorities.reserve(authorityCount);
-    message.additionals.reserve(additionalCount);
-    for (size_t index = 0; index < questionCount; ++index)
-        message.questions.push_back(decodeQuestion(reader));
-    for (size_t index = 0; index < answerCount; ++index)
-        message.answers.push_back(decodeRecord(reader));
-    for (size_t index = 0; index < authorityCount; ++index)
-        message.authorities.push_back(decodeRecord(reader));
-    for (size_t index = 0; index < additionalCount; ++index)
-        message.additionals.push_back(decodeRecord(reader));
-    if (reader.position() != size)
-        throw ParseError("trailing bytes in DNS packet");
-    return message;
-}
-
-} // namespace dht::mdns

src/peer_discovery/mdns.h

@@ -1,194 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-#pragma once
-
-#include "opendht/def.h"
-
-#include <array>
-#include <cstddef>
-#include <cstdint>
-#include <stdexcept>
-#include <string>
-#include <string_view>
-#include <variant>
-#include <vector>
-
-namespace dht::mdns {
-
-constexpr size_t MAX_PACKET_SIZE = 9000;
-constexpr size_t MAX_RECORD_COUNT = 1024;
-constexpr uint16_t CLASS_IN = 1;
-constexpr uint16_t CLASS_ANY = 255;
-
-enum class Type : uint16_t {
-    A = 1,
-    PTR = 12,
-    TXT = 16,
-    AAAA = 28,
-    SRV = 33,
-    NSEC = 47,
-    ANY = 255,
-};
-
-class OPENDHT_PUBLIC ParseError : public std::runtime_error
-{
-public:
-    using std::runtime_error::runtime_error;
-};
-
-class OPENDHT_PUBLIC Name
-{
-public:
-    Name() = default;
-    explicit Name(std::vector<std::string> labels);
-
-    static Name fromString(std::string_view name);
-
-    const std::vector<std::string>& labels() const noexcept { return labels_; }
-    std::string toString() const;
-
-    friend bool operator==(const Name& lhs, const Name& rhs) noexcept;
-    friend bool operator!=(const Name& lhs, const Name& rhs) noexcept { return not(lhs == rhs); }
-
-private:
-    std::vector<std::string> labels_;
-};
-
-OPENDHT_PUBLIC bool operator==(const Name& lhs, const Name& rhs) noexcept;
-
-struct Question
-{
-    Name name;
-    Type type {Type::ANY};
-    uint16_t classCode {CLASS_IN};
-    bool unicastResponse {false};
-};
-
-struct AData
-{
-    std::array<uint8_t, 4> address {};
-};
-
-struct AaaaData
-{
-    std::array<uint8_t, 16> address {};
-};
-
-struct PtrData
-{
-    Name name;
-};
-
-struct TxtData
-{
-    std::vector<std::vector<uint8_t>> strings;
-};
-
-struct SrvData
-{
-    uint16_t priority {};
-    uint16_t weight {};
-    uint16_t port {};
-    Name target;
-};
-
-struct NsecData
-{
-    Name nextDomain;
-    std::vector<Type> types;
-};
-
-struct UnknownData
-{
-    std::vector<uint8_t> bytes;
-};
-
-using RData = std::variant<AData, AaaaData, PtrData, TxtData, SrvData, NsecData, UnknownData>;
-
-struct ResourceRecord
-{
-    Name name;
-    Type type {Type::ANY};
-    uint16_t classCode {CLASS_IN};
-    bool cacheFlush {false};
-    uint32_t ttl {};
-    RData data {UnknownData {}};
-};
-
-struct Message
-{
-    uint16_t id {};
-    bool response {false};
-    bool authoritative {false};
-    bool truncated {false};
-    std::vector<Question> questions;
-    std::vector<ResourceRecord> answers;
-    std::vector<ResourceRecord> authorities;
-    std::vector<ResourceRecord> additionals;
-};
-
-using Packet = std::vector<uint8_t>;
-
-OPENDHT_PUBLIC Packet encode(const Message& message);
-OPENDHT_PUBLIC Message decode(const uint8_t* packet, size_t size);
-
-inline bool
-operator==(const Question& lhs, const Question& rhs)
-{
-    return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode
-           and lhs.unicastResponse == rhs.unicastResponse;
-}
-
-inline bool
-operator==(const AData& lhs, const AData& rhs)
-{
-    return lhs.address == rhs.address;
-}
-inline bool
-operator==(const AaaaData& lhs, const AaaaData& rhs)
-{
-    return lhs.address == rhs.address;
-}
-inline bool
-operator==(const PtrData& lhs, const PtrData& rhs)
-{
-    return lhs.name == rhs.name;
-}
-inline bool
-operator==(const TxtData& lhs, const TxtData& rhs)
-{
-    return lhs.strings == rhs.strings;
-}
-inline bool
-operator==(const SrvData& lhs, const SrvData& rhs)
-{
-    return lhs.priority == rhs.priority and lhs.weight == rhs.weight and lhs.port == rhs.port
-           and lhs.target == rhs.target;
-}
-inline bool
-operator==(const NsecData& lhs, const NsecData& rhs)
-{
-    return lhs.nextDomain == rhs.nextDomain and lhs.types == rhs.types;
-}
-inline bool
-operator==(const UnknownData& lhs, const UnknownData& rhs)
-{
-    return lhs.bytes == rhs.bytes;
-}
-
-inline bool
-operator==(const ResourceRecord& lhs, const ResourceRecord& rhs)
-{
-    return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode
-           and lhs.cacheFlush == rhs.cacheFlush and lhs.ttl == rhs.ttl and lhs.data == rhs.data;
-}
-
-inline bool
-operator==(const Message& lhs, const Message& rhs)
-{
-    return lhs.id == rhs.id and lhs.response == rhs.response and lhs.authoritative == rhs.authoritative
-           and lhs.truncated == rhs.truncated and lhs.questions == rhs.questions and lhs.answers == rhs.answers
-           and lhs.authorities == rhs.authorities and lhs.additionals == rhs.additionals;
-}
-
-} // namespace dht::mdns

src/peer_discovery/mdns_transport.cpp

@@ -1,455 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-
-#include "mdns_transport.h"
-
-#include <asio.hpp>
-#include <algorithm>
-#include <array>
-#include <cerrno>
-#include <cstring>
-#include <future>
-#include <mutex>
-
-#ifdef _WIN32
-#include <iphlpapi.h>
-#else
-#include <ifaddrs.h>
-#include <net/if.h>
-#endif
-
-namespace dht::mdns {
-namespace {
-
-constexpr in_port_t MDNS_PORT = 5353;
-constexpr std::string_view MDNS_IPV4 = "224.0.0.251";
-constexpr std::string_view MDNS_IPV6 = "ff02::fb";
-
-std::vector<InterfaceAddress>
-interfaceAddresses([[maybe_unused]] asio::io_context& ioContext)
-{
-    std::vector<InterfaceAddress> result;
-#ifdef _WIN32
-    ULONG size = 15000;
-    std::vector<uint8_t> buffer(size);
-    auto* adapters = reinterpret_cast<IP_ADAPTER_ADDRESSES*>(buffer.data());
-    auto status = GetAdaptersAddresses(AF_UNSPEC,
-                                       GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST | GAA_FLAG_SKIP_DNS_SERVER,
-                                       nullptr,
-                                       adapters,
-                                       &size);
-    if (status == ERROR_BUFFER_OVERFLOW) {
-        buffer.resize(size);
-        adapters = reinterpret_cast<IP_ADAPTER_ADDRESSES*>(buffer.data());
-        status = GetAdaptersAddresses(AF_UNSPEC,
-                                      GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST | GAA_FLAG_SKIP_DNS_SERVER,
-                                      nullptr,
-                                      adapters,
-                                      &size);
-    }
-    if (status == NO_ERROR) {
-        InterfaceId id = 0;
-        for (auto* adapter = adapters; adapter; adapter = adapter->Next) {
-            if (adapter->OperStatus != IfOperStatusUp)
-                continue;
-            ++id;
-            const auto multicast = not(adapter->Flags & IP_ADAPTER_NO_MULTICAST);
-            const auto loopback = adapter->IfType == IF_TYPE_SOFTWARE_LOOPBACK;
-            for (auto* current = adapter->FirstUnicastAddress; current; current = current->Next) {
-                const auto* socketAddress = current->Address.lpSockaddr;
-                if (socketAddress->sa_family == AF_INET) {
-                    const auto* address = reinterpret_cast<const sockaddr_in*>(socketAddress);
-                    asio::ip::address_v4::bytes_type bytes;
-                    std::memcpy(bytes.data(), &address->sin_addr, bytes.size());
-                    result.push_back({id, adapter->IfIndex, asio::ip::address_v4(bytes), multicast, loopback});
-                } else if (socketAddress->sa_family == AF_INET6) {
-                    const auto* address = reinterpret_cast<const sockaddr_in6*>(socketAddress);
-                    asio::ip::address_v6::bytes_type bytes;
-                    std::memcpy(bytes.data(), &address->sin6_addr, bytes.size());
-                    result.push_back({id,
-                                      adapter->Ipv6IfIndex,
-                                      asio::ip::address_v6(bytes, address->sin6_scope_id),
-                                      multicast,
-                                      loopback});
-                }
-            }
-        }
-    }
-#else
-    ifaddrs* addresses = nullptr;
-    if (getifaddrs(&addresses) != 0)
-        return result;
-    for (auto* current = addresses; current; current = current->ifa_next) {
-        if (not current->ifa_addr or not(current->ifa_flags & IFF_UP))
-            continue;
-        const auto family = current->ifa_addr->sa_family;
-        if (family == AF_INET) {
-            const auto* address = reinterpret_cast<const sockaddr_in*>(current->ifa_addr);
-            asio::ip::address_v4::bytes_type bytes;
-            std::memcpy(bytes.data(), &address->sin_addr, bytes.size());
-            const auto index = if_nametoindex(current->ifa_name);
-            result.push_back({index,
-                              index,
-                              asio::ip::address_v4(bytes),
-                              bool(current->ifa_flags & IFF_MULTICAST),
-                              bool(current->ifa_flags & IFF_LOOPBACK)});
-        } else if (family == AF_INET6) {
-            const auto* address = reinterpret_cast<const sockaddr_in6*>(current->ifa_addr);
-            asio::ip::address_v6::bytes_type bytes;
-            std::memcpy(bytes.data(), &address->sin6_addr, bytes.size());
-            const auto index = if_nametoindex(current->ifa_name);
-            result.push_back({index,
-                              index,
-                              asio::ip::address_v6(bytes, address->sin6_scope_id),
-                              bool(current->ifa_flags & IFF_MULTICAST),
-                              bool(current->ifa_flags & IFF_LOOPBACK)});
-        }
-    }
-    freeifaddrs(addresses);
-#endif
-    return result;
-}
-
-} // namespace
-
-std::vector<NetworkInterface>
-groupInterfaces(const std::vector<InterfaceAddress>& addresses)
-{
-    std::vector<NetworkInterface> result;
-    const auto hasNonLoopback = std::any_of(addresses.begin(), addresses.end(), [](const InterfaceAddress& address) {
-        return address.multicast and not address.loopback and not address.address.is_unspecified()
-               and not address.address.is_multicast();
-    });
-
-    for (const auto& address : addresses) {
-        if (address.id == 0 or not address.multicast or (hasNonLoopback and address.loopback)
-            or address.address.is_unspecified() or address.address.is_multicast())
-            continue;
-
-        auto scope = std::find_if(result.begin(), result.end(), [&](const NetworkInterface& interface) {
-            return interface.id == address.id;
-        });
-        if (scope == result.end()) {
-            NetworkInterface interface;
-            interface.id = address.id;
-            result.push_back(std::move(interface));
-            scope = std::prev(result.end());
-        }
-
-        dns_sd::AddressData record;
-        if (address.address.is_v4()) {
-            scope->ipv4Index = address.index;
-            if (not scope->ipv4Outbound)
-                scope->ipv4Outbound = address.address.to_v4();
-            record = AData {address.address.to_v4().to_bytes()};
-        } else {
-            const auto bytes = address.address.to_v6().to_bytes();
-            const auto mapped = std::all_of(bytes.begin(), bytes.begin() + 10, [](uint8_t byte) { return byte == 0; })
-                                and bytes[10] == 0xff and bytes[11] == 0xff;
-            if (mapped)
-                continue;
-            scope->ipv6Index = address.index;
-            record = AaaaData {bytes};
-        }
-        if (std::find(scope->addresses.begin(), scope->addresses.end(), record) == scope->addresses.end())
-            scope->addresses.emplace_back(std::move(record));
-    }
-    result.erase(std::remove_if(result.begin(),
-                                result.end(),
-                                [](const NetworkInterface& interface) { return interface.addresses.empty(); }),
-                 result.end());
-    return result;
-}
-
-class MdnsTransport::Impl : public std::enable_shared_from_this<MdnsTransport::Impl>
-{
-public:
-    Impl(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger)
-        : ioContext_(std::move(ioContext))
-        , logger_(std::move(logger))
-        , interfaces_(groupInterfaces(interfaceAddresses(*ioContext_)))
-    {}
-
-    struct Socket
-    {
-        explicit Socket(asio::io_context& ioContext)
-            : socket(ioContext)
-        {}
-
-        asio::ip::udp::socket socket;
-        std::array<uint8_t, MAX_PACKET_SIZE> receiveBuffer {};
-        asio::ip::udp::endpoint sender;
-        asio::ip::udp::endpoint multicast;
-        InterfaceId interface {};
-        unsigned ipv6Index {};
-    };
-
-    void start(ReceiveCallback callback)
-    {
-        callback_ = std::move(callback);
-        configure();
-    }
-
-    void send(Packet packet) { send(std::move(packet), {}, {}); }
-
-    void send(Packet packet, std::optional<asio::ip::udp::endpoint> destination, std::optional<InterfaceId> interface)
-    {
-        auto sharedPacket = std::make_shared<Packet>(std::move(packet));
-        asio::post(*ioContext_, [weak = weak_from_this(), sharedPacket, destination, interface] {
-            const auto self = weak.lock();
-            if (not self)
-                return;
-            for (const auto& state : self->sockets_) {
-                if (interface and state->interface != *interface)
-                    continue;
-                const auto target = destination.value_or(state->multicast);
-                if (target.address().is_v4() != state->multicast.address().is_v4())
-                    continue;
-                state->socket.async_send_to(asio::buffer(*sharedPacket),
-                                            target,
-                                            [logger = self->logger_, sharedPacket](const asio::error_code& error,
-                                                                                   size_t) {
-                                                if (error and error != asio::error::operation_aborted and logger)
-                                                    logger->warn("Unable to send mDNS packet: {}", error.message());
-                                            });
-            }
-        });
-    }
-
-    void sendAndWait(Packet packet, InterfaceId interface)
-    {
-        if (ioContext_->get_executor().running_in_this_thread()) {
-            sendNow(packet, interface);
-            return;
-        }
-        auto completed = std::make_shared<std::promise<void>>();
-        auto future = completed->get_future();
-        asio::post(*ioContext_, [weak = weak_from_this(), packet = std::move(packet), interface, completed] {
-            if (const auto self = weak.lock())
-                self->sendNow(packet, interface);
-            completed->set_value();
-        });
-        future.wait_for(std::chrono::milliseconds(250));
-    }
-
-    void refresh()
-    {
-        {
-            std::lock_guard lock(interfacesMutex_);
-            interfaces_ = groupInterfaces(interfaceAddresses(*ioContext_));
-        }
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->configure();
-        });
-    }
-
-    void stop()
-    {
-        if (ioContext_->get_executor().running_in_this_thread()) {
-            stopNow();
-            return;
-        }
-        auto completed = std::make_shared<std::promise<void>>();
-        auto future = completed->get_future();
-        asio::post(*ioContext_, [weak = weak_from_this(), completed] {
-            if (const auto self = weak.lock())
-                self->stopNow();
-            completed->set_value();
-        });
-        future.wait_for(std::chrono::milliseconds(250));
-    }
-
-    std::vector<NetworkInterface> interfaces() const
-    {
-        std::lock_guard lock(interfacesMutex_);
-        return interfaces_;
-    }
-
-private:
-    void sendNow(const Packet& packet, InterfaceId interface)
-    {
-        for (const auto& state : sockets_) {
-            if (state->interface != interface)
-                continue;
-            asio::error_code error;
-            state->socket.send_to(asio::buffer(packet), state->multicast, 0, error);
-            if (error and logger_)
-                logger_->warn("Unable to send final mDNS packet: {}", error.message());
-        }
-    }
-
-    void stopNow()
-    {
-        callback_ = {};
-        stopSockets();
-    }
-
-    void configure()
-    {
-        stopSockets();
-        std::vector<NetworkInterface> currentInterfaces;
-        {
-            std::lock_guard lock(interfacesMutex_);
-            currentInterfaces = interfaces_;
-        }
-        for (const auto& interface : currentInterfaces) {
-            if (interface.ipv4Outbound)
-                configure(interface, asio::ip::udp::v4(), asio::ip::make_address(MDNS_IPV4));
-            if (interface.ipv6Index)
-                configure(interface, asio::ip::udp::v6(), asio::ip::make_address(MDNS_IPV6));
-        }
-    }
-
-    void configure(const NetworkInterface& interface, asio::ip::udp protocol, const asio::ip::address& multicast)
-    {
-        auto state = std::make_shared<Socket>(*ioContext_);
-        asio::error_code error;
-        state->socket.open(protocol, error);
-        if (error)
-            return logSocketError(protocol, error);
-        state->socket.set_option(asio::ip::udp::socket::reuse_address(true), error);
-#if defined(SO_REUSEPORT) && !defined(_WIN32)
-        if (not error) {
-            const int enabled = 1;
-            if (::setsockopt(state->socket.native_handle(), SOL_SOCKET, SO_REUSEPORT, &enabled, sizeof(enabled)) != 0)
-                error.assign(errno, asio::error::get_system_category());
-        }
-#endif
-        if (error)
-            return logSocketError(protocol, error);
-        if (protocol == asio::ip::udp::v6())
-            state->socket.set_option(asio::ip::v6_only(true), error);
-        state->socket.bind({protocol, MDNS_PORT}, error);
-        if (error)
-            return logSocketError(protocol, error);
-        state->socket.set_option(asio::ip::multicast::hops(255), error);
-        if (error)
-            return logSocketError(protocol, error);
-        state->socket.set_option(asio::ip::multicast::enable_loopback(true), error);
-        if (protocol == asio::ip::udp::v4()) {
-            state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v4(), *interface.ipv4Outbound), error);
-            if (not error)
-                state->socket.set_option(asio::ip::multicast::outbound_interface(*interface.ipv4Outbound), error);
-        } else {
-            state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v6(), interface.ipv6Index), error);
-            if (not error)
-                state->socket.set_option(asio::ip::multicast::outbound_interface(interface.ipv6Index), error);
-        }
-        if (error)
-            return logSocketError(protocol, error);
-
-        if (protocol == asio::ip::udp::v6())
-            state->multicast = {asio::ip::address_v6(multicast.to_v6().to_bytes(), interface.ipv6Index), MDNS_PORT};
-        else
-            state->multicast = {multicast, MDNS_PORT};
-        state->interface = interface.id;
-        state->ipv6Index = interface.ipv6Index;
-        sockets_.push_back(state);
-        receive(state);
-    }
-
-    void receive(const std::shared_ptr<Socket>& state)
-    {
-        state->socket.async_receive_from(asio::buffer(state->receiveBuffer),
-                                         state->sender,
-                                         [weak = weak_from_this(), state](const asio::error_code& error, size_t size) {
-                                             const auto self = weak.lock();
-                                             if (not self)
-                                                 return;
-                                             if (not error and self->callback_) {
-                                                 Packet packet(state->receiveBuffer.begin(),
-                                                               state->receiveBuffer.begin() + size);
-                                                 self->callback_(packet,
-                                                                 {state->sender, state->interface, state->ipv6Index});
-                                             } else if (error != asio::error::operation_aborted and self->logger_) {
-                                                 self->logger_->warn("Unable to receive mDNS packet: {}",
-                                                                     error.message());
-                                             }
-                                             if (state->socket.is_open())
-                                                 self->receive(state);
-                                         });
-    }
-
-    void stopSockets()
-    {
-        for (const auto& state : sockets_) {
-            asio::error_code error;
-            state->socket.cancel(error);
-            state->socket.close(error);
-        }
-        sockets_.clear();
-    }
-
-    void logSocketError(asio::ip::udp protocol, const asio::error_code& error)
-    {
-        if (logger_)
-            logger_->warn("Unable to start {} mDNS transport: {}",
-                          protocol == asio::ip::udp::v4() ? "IPv4" : "IPv6",
-                          error.message());
-    }
-
-    std::shared_ptr<asio::io_context> ioContext_;
-    std::shared_ptr<Logger> logger_;
-    mutable std::mutex interfacesMutex_;
-    std::vector<NetworkInterface> interfaces_;
-    std::vector<std::shared_ptr<Socket>> sockets_;
-    ReceiveCallback callback_;
-};
-
-MdnsTransport::MdnsTransport(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger)
-    : impl_(std::make_shared<Impl>(std::move(ioContext), std::move(logger)))
-{}
-
-MdnsTransport::~MdnsTransport() = default;
-
-void
-MdnsTransport::start(ReceiveCallback callback)
-{
-    impl_->start(std::move(callback));
-}
-
-void
-MdnsTransport::send(Packet packet)
-{
-    impl_->send(std::move(packet));
-}
-
-void
-MdnsTransport::send(Packet packet, InterfaceId interface)
-{
-    impl_->send(std::move(packet), {}, interface);
-}
-
-void
-MdnsTransport::send(Packet packet, const asio::ip::udp::endpoint& destination, InterfaceId interface)
-{
-    impl_->send(std::move(packet), destination, interface);
-}
-
-void
-MdnsTransport::sendAndWait(Packet packet, InterfaceId interface)
-{
-    impl_->sendAndWait(std::move(packet), interface);
-}
-
-void
-MdnsTransport::refresh()
-{
-    impl_->refresh();
-}
-
-void
-MdnsTransport::stop()
-{
-    if (impl_)
-        impl_->stop();
-}
-
-std::vector<NetworkInterface>
-MdnsTransport::interfaces() const
-{
-    return impl_->interfaces();
-}
-
-} // namespace dht::mdns

src/peer_discovery/mdns_transport.h

@@ -1,74 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-#pragma once
-
-#include "dns_sd.h"
-
-#include "opendht/logger.h"
-
-#include <asio/ip/address.hpp>
-#include <asio/ip/udp.hpp>
-#include <cstdint>
-#include <functional>
-#include <memory>
-#include <optional>
-
-namespace asio {
-class io_context;
-}
-
-namespace dht::mdns {
-
-using InterfaceId = uint64_t;
-
-struct InterfaceAddress
-{
-    InterfaceId id {};
-    unsigned index {};
-    asio::ip::address address;
-    bool multicast {true};
-    bool loopback {false};
-};
-
-struct NetworkInterface
-{
-    InterfaceId id {};
-    unsigned ipv4Index {};
-    unsigned ipv6Index {};
-    std::optional<asio::ip::address_v4> ipv4Outbound;
-    std::vector<dns_sd::AddressData> addresses;
-};
-
-OPENDHT_PUBLIC std::vector<NetworkInterface> groupInterfaces(const std::vector<InterfaceAddress>& addresses);
-
-struct ReceiveContext
-{
-    asio::ip::udp::endpoint source;
-    InterfaceId interface {};
-    unsigned ipv6Index {};
-};
-
-class MdnsTransport
-{
-public:
-    using ReceiveCallback = std::function<void(const Packet&, const ReceiveContext&)>;
-
-    MdnsTransport(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger = {});
-    ~MdnsTransport();
-
-    void start(ReceiveCallback callback);
-    void send(Packet packet);
-    void send(Packet packet, InterfaceId interface);
-    void send(Packet packet, const asio::ip::udp::endpoint& destination, InterfaceId interface);
-    void sendAndWait(Packet packet, InterfaceId interface);
-    void refresh();
-    void stop();
-
-    std::vector<NetworkInterface> interfaces() const;
-
-private:
-    class Impl;
-    std::shared_ptr<Impl> impl_;
-};
-
-} // namespace dht::mdns

src/peer_discovery/peer_discovery.cpp

@@ -1,438 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-
-#include "peer_discovery.h"
-
-#include "dns_sd.h"
-#include "mdns_transport.h"
-
-#include <asio.hpp>
-#include <chrono>
-#include <cstring>
-#include <map>
-#include <mutex>
-#include <optional>
-#include <random>
-
-namespace dht {
-namespace {
-
-constexpr auto BROWSE_INTERVAL = std::chrono::seconds(60);
-constexpr auto CONNECTIVITY_RETRY = std::chrono::seconds(10);
-constexpr auto CONNECTIVITY_RETRY_MAX = std::chrono::minutes(1);
-
-SockAddr
-socketAddress(const mdns::dns_sd::AddressData& data, in_port_t port, const mdns::ReceiveContext& context)
-{
-    SockAddr address;
-    std::visit(
-        [&](const auto& record) {
-            using Record = std::decay_t<decltype(record)>;
-            if constexpr (std::is_same_v<Record, mdns::AData>) {
-                address.setFamily(AF_INET);
-                std::memcpy(&address.getIPv4().sin_addr, record.address.data(), record.address.size());
-            } else {
-                address.setFamily(AF_INET6);
-                std::memcpy(&address.getIPv6().sin6_addr, record.address.data(), record.address.size());
-                const asio::ip::address_v6 parsed(record.address);
-                if (parsed.is_link_local())
-                    address.getIPv6().sin6_scope_id = context.ipv6Index;
-            }
-        },
-        data);
-    address.setPort(port);
-    return address;
-}
-
-} // namespace
-
-class PeerDiscovery::Impl : public std::enable_shared_from_this<PeerDiscovery::Impl>
-{
-public:
-    Impl(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger)
-        : ioContext_(std::move(ioContext))
-        , logger_(std::move(logger))
-        , transport_(ioContext_, logger_)
-        , browseTimer_(*ioContext_)
-        , announceTimer_(*ioContext_)
-        , connectivityTimer_(*ioContext_)
-    {}
-
-    void start()
-    {
-        transport_.start([weak = weak_from_this()](const mdns::Packet& packet, const mdns::ReceiveContext& context) {
-            if (const auto self = weak.lock())
-                self->receive(packet, context);
-        });
-        if (logger_)
-            logger_->debug("[peer discovery {:p}] Started.", fmt::ptr(this));
-    }
-
-    void startDiscovery(NetId network, PeerDiscoveredCallback callback)
-    {
-        {
-            std::lock_guard lock(mutex_);
-            discoveryNetwork_ = network;
-            discoveryCallback_ = std::move(callback);
-        }
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->browse();
-        });
-    }
-
-    void startPublish(const InfoHash& nodeId, NetId network, in_port_t port)
-    {
-        mdns::dns_sd::Service service {nodeId, network, port, {}};
-        if (transport_.interfaces().empty())
-            throw mdns::ParseError("DNS-SD service requires a multicast interface");
-        {
-            std::lock_guard lock(mutex_);
-            published_ = std::move(service);
-        }
-        announce();
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            const auto self = weak.lock();
-            if (not self)
-                return;
-            self->announceTimer_.expires_after(std::chrono::seconds(1));
-            self->announceTimer_.async_wait([weak](const asio::error_code& error) {
-                if (not error) {
-                    if (const auto locked = weak.lock())
-                        locked->announce();
-                }
-            });
-        });
-    }
-
-    bool stopDiscovery()
-    {
-        std::lock_guard lock(mutex_);
-        const auto stopped = bool(discoveryCallback_);
-        discoveryCallback_ = {};
-        discoveryNetwork_.reset();
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->browseTimer_.cancel();
-        });
-        return stopped;
-    }
-
-    bool stopPublish()
-    {
-        std::optional<mdns::dns_sd::Service> service;
-        {
-            std::lock_guard lock(mutex_);
-            service = std::exchange(published_, {});
-        }
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->announceTimer_.cancel();
-        });
-        if (service)
-            advertise(*service, true, false);
-        return bool(service);
-    }
-
-    void stop()
-    {
-        stopConnectivityChanged();
-        stopDiscovery();
-        std::optional<mdns::dns_sd::Service> service;
-        {
-            std::lock_guard lock(mutex_);
-            service = std::exchange(published_, {});
-        }
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->announceTimer_.cancel();
-        });
-        if (service)
-            advertise(*service, true, true);
-        transport_.stop();
-    }
-
-    void connectivityChanged()
-    {
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock())
-                self->handleConnectivityChanged();
-        });
-    }
-
-    void stopConnectivityChanged()
-    {
-        asio::post(*ioContext_, [weak = weak_from_this()] {
-            if (const auto self = weak.lock()) {
-                self->connectivityTimer_.cancel();
-                self->connectivityRetry_ = CONNECTIVITY_RETRY;
-            }
-        });
-    }
-
-private:
-    void handleConnectivityChanged()
-    {
-        caches_.clear();
-        transport_.refresh();
-        refreshPublication();
-        browse();
-
-        connectivityTimer_.expires_after(connectivityRetry_);
-        connectivityTimer_.async_wait([weak = weak_from_this()](const asio::error_code& error) {
-            if (not error) {
-                if (const auto self = weak.lock())
-                    self->handleConnectivityChanged();
-            }
-        });
-        connectivityRetry_ = std::min(connectivityRetry_ * 2,
-                                      std::chrono::duration_cast<std::chrono::steady_clock::duration>(
-                                          CONNECTIVITY_RETRY_MAX));
-    }
-
-    void browse()
-    {
-        bool active;
-        {
-            std::lock_guard lock(mutex_);
-            active = bool(discoveryCallback_);
-        }
-        if (not active)
-            return;
-        if (logger_)
-            logger_->debug("[peer discovery {:p}] Browsing.", fmt::ptr(this));
-        transport_.send(mdns::encode(mdns::dns_sd::browseQuery()));
-        browseTimer_.expires_after(BROWSE_INTERVAL);
-        browseTimer_.async_wait([weak = weak_from_this()](const asio::error_code& error) {
-            if (not error) {
-                if (const auto self = weak.lock())
-                    self->browse();
-            }
-        });
-    }
-
-    void announce()
-    {
-        std::optional<mdns::dns_sd::Service> service;
-        {
-            std::lock_guard lock(mutex_);
-            service = published_;
-        }
-        if (logger_)
-            logger_->debug("[peer discovery {:p}] Announcing.", fmt::ptr(this));
-        if (service)
-            advertise(*service, false, false);
-    }
-
-    void refreshPublication()
-    {
-        std::optional<mdns::dns_sd::Service> service;
-        {
-            std::lock_guard lock(mutex_);
-            if (published_)
-                service = published_;
-        }
-        if (service)
-            advertise(*service, false, false);
-    }
-
-    void advertise(const mdns::dns_sd::Service& service, bool goodbye, bool wait)
-    {
-        for (const auto& interface : transport_.interfaces()) {
-            auto scoped = service;
-            scoped.addresses = interface.addresses;
-            auto packet = mdns::encode(goodbye ? mdns::dns_sd::goodbye(scoped) : mdns::dns_sd::announcement(scoped));
-            if (wait)
-                transport_.sendAndWait(std::move(packet), interface.id);
-            else
-                transport_.send(std::move(packet), interface.id);
-        }
-    }
-
-    void receive(const mdns::Packet& packet, const mdns::ReceiveContext& context)
-    {
-        if (logger_)
-            logger_->debug("[peer discovery {:p}] Got packet of size {}.", fmt::ptr(this), packet.size());
-        try {
-            const auto message = mdns::decode(packet.data(), packet.size());
-            if (message.response)
-                discover(message, context);
-            else
-                answer(message, context);
-        } catch (const std::exception& error) {
-            if (logger_)
-                logger_->warn("Ignoring invalid mDNS packet: {}", error.what());
-        }
-    }
-
-    void answer(const mdns::Message& query, const mdns::ReceiveContext& context)
-    {
-        std::optional<mdns::dns_sd::Service> service;
-        {
-            std::lock_guard lock(mutex_);
-            service = published_;
-        }
-        if (service) {
-            const auto interfaces = transport_.interfaces();
-            const auto interface = std::find_if(interfaces.begin(), interfaces.end(), [&](const auto& candidate) {
-                return candidate.id == context.interface;
-            });
-            if (interface == interfaces.end())
-                return;
-            service->addresses = interface->addresses;
-            if (auto response = mdns::dns_sd::respond(query, *service)) {
-                const auto unicast = context.source.port() != 5353
-                                     or std::any_of(query.questions.begin(),
-                                                    query.questions.end(),
-                                                    [](const mdns::Question& question) {
-                                                        return question.unicastResponse;
-                                                    });
-                if (context.source.port() != 5353)
-                    response->questions = query.questions;
-                auto packet = mdns::encode(*response);
-                const auto sharedAnswer = std::any_of(response->answers.begin(),
-                                                      response->answers.end(),
-                                                      [](const mdns::ResourceRecord& record) {
-                                                          return not record.cacheFlush;
-                                                      });
-                if (not sharedAnswer) {
-                    if (unicast)
-                        transport_.send(std::move(packet), context.source, context.interface);
-                    else
-                        transport_.send(std::move(packet), context.interface);
-                    return;
-                }
-
-                static thread_local std::mt19937 random {std::random_device {}()};
-                const auto delay = std::chrono::milliseconds(std::uniform_int_distribution<int> {20, 120}(random));
-                auto timer = std::make_shared<asio::steady_timer>(*ioContext_, delay);
-                timer->async_wait([weak = weak_from_this(), timer, packet = std::move(packet), context, unicast](
-                                      const asio::error_code& error) mutable {
-                    if (error)
-                        return;
-                    const auto self = weak.lock();
-                    if (not self)
-                        return;
-                    if (unicast)
-                        self->transport_.send(std::move(packet), context.source, context.interface);
-                    else
-                        self->transport_.send(std::move(packet), context.interface);
-                });
-            }
-        }
-    }
-
-    void discover(const mdns::Message& message, const mdns::ReceiveContext& context)
-    {
-        PeerDiscoveredCallback callback;
-        std::optional<NetId> network;
-        std::optional<InfoHash> localNode;
-        {
-            std::lock_guard lock(mutex_);
-            callback = discoveryCallback_;
-            network = discoveryNetwork_;
-            if (published_)
-                localNode = published_->nodeId;
-        }
-        if (not callback or not network)
-            return;
-
-        const auto cache = caches_.try_emplace(context.interface).first;
-        for (const auto& service : cache->second.update(message)) {
-            if (service.network != *network or (localNode and service.nodeId == *localNode))
-                continue;
-            for (const auto& address : service.addresses)
-                callback(service.nodeId, socketAddress(address, service.port, context));
-        }
-    }
-
-    std::shared_ptr<asio::io_context> ioContext_;
-    std::shared_ptr<Logger> logger_;
-    mdns::MdnsTransport transport_;
-    asio::steady_timer browseTimer_;
-    asio::steady_timer announceTimer_;
-    asio::steady_timer connectivityTimer_;
-    std::chrono::steady_clock::duration connectivityRetry_ {CONNECTIVITY_RETRY};
-    std::mutex mutex_;
-    std::optional<NetId> discoveryNetwork_;
-    PeerDiscoveredCallback discoveryCallback_;
-    std::optional<mdns::dns_sd::Service> published_;
-    std::map<mdns::InterfaceId, mdns::dns_sd::Cache> caches_;
-};
-
-PeerDiscovery::PeerDiscovery(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger)
-{
-    if (not ioContext) {
-        ioContext = std::make_shared<asio::io_context>();
-        ioContext_ = ioContext;
-    }
-    impl_ = std::make_shared<Impl>(ioContext, logger);
-    impl_->start();
-
-    if (ioContext_) {
-        ioRunner_ = std::thread([logger = std::move(logger), ioContext = std::move(ioContext)] {
-            try {
-                auto work = asio::make_work_guard(*ioContext);
-                ioContext->run();
-            } catch (const std::exception& error) {
-                if (logger)
-                    logger->error("Peer discovery event loop failed: {}", error.what());
-            }
-        });
-    }
-}
-
-PeerDiscovery::~PeerDiscovery()
-{
-    stop();
-    impl_.reset();
-    if (ioContext_)
-        ioContext_->stop();
-    if (ioRunner_.joinable())
-        ioRunner_.join();
-}
-
-void
-PeerDiscovery::startDiscovery(NetId network, PeerDiscoveredCallback callback)
-{
-    impl_->startDiscovery(network, std::move(callback));
-}
-
-void
-PeerDiscovery::startPublish(const InfoHash& nodeId, NetId network, in_port_t port)
-{
-    impl_->startPublish(nodeId, network, port);
-}
-
-bool
-PeerDiscovery::stopDiscovery()
-{
-    return impl_->stopDiscovery();
-}
-
-bool
-PeerDiscovery::stopPublish()
-{
-    return impl_->stopPublish();
-}
-
-void
-PeerDiscovery::stop()
-{
-    if (impl_)
-        impl_->stop();
-}
-
-void
-PeerDiscovery::connectivityChanged()
-{
-    impl_->connectivityChanged();
-}
-
-void
-PeerDiscovery::stopConnectivityChanged()
-{
-    impl_->stopConnectivityChanged();
-}
-
-} // namespace dht

src/utils.cpp

@@ -245,10 +245,18 @@ from_system_time(system_clock::time_point t, time_point steady_now, system_clock
         return time_point::min();
     if (t == system_clock::time_point::max())
         return time_point::max();
-    auto dt = t - system_now;
-    if (dt > system_clock::duration(0) and steady_now > time_point::max() - dt)
+    // Saturate before the cast: the steady clock duration may hold a narrower range than the system one.
+    constexpr auto max_dt = std::chrono::duration_cast<system_clock::duration>(duration::max());
+    constexpr auto min_dt = std::chrono::duration_cast<system_clock::duration>(duration::min());
+    auto system_dt = t - system_now;
+    if (system_dt >= max_dt)
         return time_point::max();
-    else if (dt < system_clock::duration(0) and steady_now < time_point::min() - dt)
+    if (system_dt <= min_dt)
+        return time_point::min();
+    auto dt = std::chrono::duration_cast<duration>(system_dt);
+    if (dt > duration(0) and steady_now > time_point::max() - dt)
+        return time_point::max();
+    else if (dt < duration(0) and steady_now < time_point::min() - dt)
         return time_point::min();
     return steady_now + dt;
 }
@@ -266,12 +274,13 @@ to_system_time(time_point t, time_point steady_now, system_clock::time_point sys
         return system_clock::time_point::min();
     if (t == time_point::max())
         return system_clock::time_point::max();
-    auto dt = t - steady_now;
-    if (dt > duration(0) and system_now >= system_clock::time_point::max() - dt)
+    // Cast first: mixing clock durations of different periods would overflow the bound computations below.
+    auto dt = std::chrono::duration_cast<system_clock::duration>(t - steady_now);
+    if (dt > system_clock::duration(0) and system_now >= system_clock::time_point::max() - dt)
         return system_clock::time_point::max();
-    else if (dt < duration(0) and system_now <= system_clock::time_point::min() - dt)
+    else if (dt < system_clock::duration(0) and system_now <= system_clock::time_point::min() - dt)
         return system_clock::time_point::min();
-    return system_now + std::chrono::duration_cast<system_clock::duration>(dt);
+    return system_now + dt;
 }
 
 system_clock::time_point

tests/test_crypto.cpp

@@ -327,6 +327,16 @@ CryptoTester::testCertificateSerialNumber()
     }
 }
 
+void
+CryptoTester::testEcKeyCurve()
+{
+    auto key = dht::crypto::PrivateKey::generateEC(GNUTLS_ECC_CURVE_SECP256R1);
+    unsigned int bits = 0;
+    CPPUNIT_ASSERT_EQUAL(static_cast<int>(GNUTLS_PK_EC),
+                         gnutls_x509_privkey_get_pk_algorithm2(key.x509_key, &bits));
+    CPPUNIT_ASSERT_EQUAL(256u, bits);
+}
+
 void
 CryptoTester::testOcsp()
 {

tests/test_crypto.h

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

tests/test_mdns.cpp

@@ -1,280 +0,0 @@
-// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
-// SPDX-License-Identifier: MIT
-
-#include "test_mdns.h"
-
-#include "../src/peer_discovery/dns_sd.h"
-#include "../src/peer_discovery/mdns.h"
-#include "../src/peer_discovery/mdns_transport.h"
-
-#include <array>
-#include <string>
-
-namespace test {
-
-using namespace dht::mdns;
-
-CPPUNIT_TEST_SUITE_REGISTRATION(MdnsTester);
-
-void
-MdnsTester::testName()
-{
-    const auto name = Name::fromString("Node._opendht._udp.local.");
-    CPPUNIT_ASSERT_EQUAL(std::string("Node._opendht._udp.local."), name.toString());
-    CPPUNIT_ASSERT(name == Name::fromString("node._OpenDHT._UDP.LOCAL"));
-    CPPUNIT_ASSERT_THROW(Name::fromString("bad..local."), ParseError);
-    CPPUNIT_ASSERT_THROW(Name::fromString(std::string(64, 'a') + ".local."), ParseError);
-}
-
-void
-MdnsTester::testRecordRoundTrip()
-{
-    const auto service = Name::fromString("_opendht._udp.local.");
-    const auto instance = Name::fromString("0123456789abcdef0123456789abcdef01234567._opendht._udp.local.");
-    const auto host = Name::fromString("0123456789abcdef0123456789abcdef01234567.local.");
-
-    Message message;
-    message.response = true;
-    message.authoritative = true;
-    message.answers.push_back({service, Type::PTR, CLASS_IN, false, 4500, PtrData {instance}});
-    message.additionals.push_back({
-        instance, Type::SRV, CLASS_IN, true, 120, SrvData {0, 0, 4222, host}
-    });
-    message.additionals.push_back(
-        {instance,
-         Type::TXT,
-         CLASS_IN,
-         true,
-         4500,
-         TxtData {{{'t', 'x', 't', 'v', 'e', 'r', 's', '=', '1'}, {'n', 'e', 't', '=', '0'}}}});
-    message.additionals.push_back({host, Type::A, CLASS_IN, true, 120, AData {{192, 0, 2, 1}}});
-    message.additionals.push_back(
-        {host, Type::AAAA, CLASS_IN, true, 120, AaaaData {{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1}}});
-    message.additionals.push_back({
-        host, Type::NSEC, CLASS_IN, true, 120, NsecData {host, {Type::A, Type::AAAA}}
-    });
-
-    const auto packet = encode(message);
-    const auto decoded = decode(packet.data(), packet.size());
-
-    CPPUNIT_ASSERT(decoded.response);
-    CPPUNIT_ASSERT(decoded.authoritative);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), decoded.answers.size());
-    CPPUNIT_ASSERT_EQUAL(size_t(5), decoded.additionals.size());
-    CPPUNIT_ASSERT(decoded == message);
-}
-
-void
-MdnsTester::testCompression()
-{
-    Message message;
-    message.questions.push_back({Name::fromString("_opendht._udp.local."), Type::PTR, CLASS_IN, false});
-    message.questions.push_back({Name::fromString("node._opendht._udp.local."), Type::SRV, CLASS_IN, false});
-
-    const auto packet = encode(message);
-    CPPUNIT_ASSERT(std::find(packet.begin(), packet.end(), uint8_t {0xc0}) != packet.end());
-    CPPUNIT_ASSERT(decode(packet.data(), packet.size()) == message);
-}
-
-void
-MdnsTester::testMalformedPackets()
-{
-    const std::array<uint8_t, 3> truncated {
-        {0, 0, 0}
-    };
-    CPPUNIT_ASSERT_THROW(decode(truncated.data(), truncated.size()), ParseError);
-
-    const std::array<uint8_t, 18> compressionLoop {
-        {
-         0, 0,
-         0, 0,
-         0, 1,
-         0, 0,
-         0, 0,
-         0, 0,
-         0xc0, 0x0c,
-         0, static_cast<uint8_t>(Type::PTR),
-         0, CLASS_IN,
-         }
-    };
-    CPPUNIT_ASSERT_THROW(decode(compressionLoop.data(), compressionLoop.size()), ParseError);
-}
-
-void
-MdnsTester::testPacketSizeLimit()
-{
-    std::array<uint8_t, MAX_PACKET_SIZE + 1> oversized {};
-    CPPUNIT_ASSERT_THROW(decode(oversized.data(), oversized.size()), ParseError);
-}
-
-void
-MdnsTester::testDnsSdAdvertisement()
-{
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.network = 42;
-    service.port = 4222;
-    service.addresses.emplace_back(AData {
-        {192, 0, 2, 1}
-    });
-    service.addresses.emplace_back(AaaaData {
-        {0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1}
-    });
-
-    const auto message = dns_sd::announcement(service);
-    CPPUNIT_ASSERT(message.response);
-    CPPUNIT_ASSERT(message.authoritative);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), message.answers.size());
-    CPPUNIT_ASSERT_EQUAL(size_t(4), message.additionals.size());
-    CPPUNIT_ASSERT_EQUAL(std::string("_opendht._udp.local."), message.answers.front().name.toString());
-
-    const auto resolved = dns_sd::resolve(message);
-    CPPUNIT_ASSERT(resolved);
-    CPPUNIT_ASSERT_EQUAL(service.nodeId, resolved->nodeId);
-    CPPUNIT_ASSERT_EQUAL(service.network, resolved->network);
-    CPPUNIT_ASSERT_EQUAL(service.port, resolved->port);
-    CPPUNIT_ASSERT_EQUAL(service.addresses.size(), resolved->addresses.size());
-}
-
-void
-MdnsTester::testDnsSdGoodbye()
-{
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.port = 4222;
-    service.addresses.emplace_back(AData {
-        {192, 0, 2, 1}
-    });
-
-    const auto message = dns_sd::goodbye(service);
-    CPPUNIT_ASSERT(std::all_of(message.answers.begin(), message.answers.end(), [](const auto& record) {
-        return record.ttl == 0;
-    }));
-    CPPUNIT_ASSERT(std::all_of(message.additionals.begin(), message.additionals.end(), [](const auto& record) {
-        return record.ttl == 0;
-    }));
-}
-
-void
-MdnsTester::testDnsSdValidation()
-{
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.network = 42;
-    service.port = 4222;
-    service.addresses.emplace_back(AData {
-        {192, 0, 2, 1}
-    });
-
-    auto message = dns_sd::announcement(service);
-    auto& txt = std::get<TxtData>(message.additionals[1].data);
-    txt.strings.push_back({'n', 'e', 't', '=', '4', '3'});
-    CPPUNIT_ASSERT(not dns_sd::resolve(message));
-
-    message = dns_sd::announcement(service);
-    auto& srv = std::get<SrvData>(message.additionals[0].data);
-    srv.target = Name::fromString("other.local.");
-    CPPUNIT_ASSERT(not dns_sd::resolve(message));
-
-    CPPUNIT_ASSERT_THROW(dns_sd::announcement({}), ParseError);
-}
-
-void
-MdnsTester::testDnsSdQueryResponse()
-{
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.network = 42;
-    service.port = 4222;
-    service.addresses.emplace_back(AData {
-        {192, 0, 2, 1}
-    });
-
-    const auto query = dns_sd::browseQuery();
-    CPPUNIT_ASSERT(not query.response);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), query.questions.size());
-    CPPUNIT_ASSERT(query.questions.front().name == dns_sd::serviceName());
-    CPPUNIT_ASSERT(query.questions.front().type == Type::PTR);
-
-    const auto response = dns_sd::respond(query, service);
-    CPPUNIT_ASSERT(response);
-    CPPUNIT_ASSERT(dns_sd::resolve(*response));
-
-    auto knownAnswerQuery = query;
-    knownAnswerQuery.answers = response->answers;
-    CPPUNIT_ASSERT(not dns_sd::respond(knownAnswerQuery, service));
-
-    knownAnswerQuery.answers.front().ttl = dns_sd::SERVICE_RECORD_TTL / 2;
-    CPPUNIT_ASSERT(dns_sd::respond(knownAnswerQuery, service));
-
-    Message metaQuery;
-    const auto metaName = Name::fromString("_services._dns-sd._udp.local.");
-    metaQuery.questions.push_back({metaName, Type::PTR, CLASS_IN, false});
-    const auto metaResponse = dns_sd::respond(metaQuery, service);
-    CPPUNIT_ASSERT(metaResponse);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), metaResponse->answers.size());
-    CPPUNIT_ASSERT(metaResponse->answers.front().name == metaName);
-    CPPUNIT_ASSERT(std::get<PtrData>(metaResponse->answers.front().data).name == dns_sd::serviceName());
-    CPPUNIT_ASSERT(metaResponse->additionals.empty());
-
-    metaQuery.answers = metaResponse->answers;
-    CPPUNIT_ASSERT(not dns_sd::respond(metaQuery, service));
-}
-
-void
-MdnsTester::testDnsSdCache()
-{
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.network = 42;
-    service.port = 4222;
-    service.addresses.emplace_back(AData {
-        {192, 0, 2, 1}
-    });
-
-    auto now = dns_sd::Cache::Clock::now();
-    dns_sd::Cache cache([&] { return now; });
-    auto ptr = dns_sd::announcement(service);
-    auto details = ptr;
-    ptr.additionals.clear();
-    details.answers.clear();
-
-    CPPUNIT_ASSERT(cache.update(ptr).empty());
-    const auto discovered = cache.update(details);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), discovered.size());
-    CPPUNIT_ASSERT_EQUAL(service.nodeId, discovered.front().nodeId);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), cache.services().size());
-
-    now += std::chrono::seconds(dns_sd::HOST_RECORD_TTL + 1);
-    CPPUNIT_ASSERT(cache.services().empty());
-}
-
-void
-MdnsTester::testInterfaceScoping()
-{
-    const std::vector<InterfaceAddress> addresses {
-        {1, 1, asio::ip::make_address("192.0.2.1"), true,  false},
-        {1, 1, asio::ip::make_address("fe80::1"),   true,  false},
-        {2, 2, asio::ip::make_address("fe80::2"),   true,  false},
-        {3, 3, asio::ip::make_address("fe80::3"),   false, false},
-        {4, 4, asio::ip::make_address("::1"),       true,  true },
-    };
-
-    const auto interfaces = groupInterfaces(addresses);
-    CPPUNIT_ASSERT_EQUAL(size_t(2), interfaces.size());
-    CPPUNIT_ASSERT_EQUAL(InterfaceId {1}, interfaces[0].id);
-    CPPUNIT_ASSERT_EQUAL(size_t(2), interfaces[0].addresses.size());
-    CPPUNIT_ASSERT_EQUAL(InterfaceId {2}, interfaces[1].id);
-    CPPUNIT_ASSERT_EQUAL(size_t(1), interfaces[1].addresses.size());
-
-    dns_sd::Service service;
-    service.nodeId = dht::InfoHash("0123456789abcdef0123456789abcdef01234567");
-    service.port = 4222;
-    service.addresses = interfaces[0].addresses;
-    const auto announcement = dns_sd::announcement(service);
-    const auto resolved = dns_sd::resolve(announcement);
-    CPPUNIT_ASSERT(resolved);
-    CPPUNIT_ASSERT(resolved->addresses == interfaces[0].addresses);
-}
-
-} // namespace test

tests/test_mdns.h

@@ -1,40 +0,0 @@
-// 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 MdnsTester : public CppUnit::TestFixture
-{
-    CPPUNIT_TEST_SUITE(MdnsTester);
-    CPPUNIT_TEST(testName);
-    CPPUNIT_TEST(testRecordRoundTrip);
-    CPPUNIT_TEST(testCompression);
-    CPPUNIT_TEST(testMalformedPackets);
-    CPPUNIT_TEST(testPacketSizeLimit);
-    CPPUNIT_TEST(testDnsSdAdvertisement);
-    CPPUNIT_TEST(testDnsSdGoodbye);
-    CPPUNIT_TEST(testDnsSdValidation);
-    CPPUNIT_TEST(testDnsSdQueryResponse);
-    CPPUNIT_TEST(testDnsSdCache);
-    CPPUNIT_TEST(testInterfaceScoping);
-    CPPUNIT_TEST_SUITE_END();
-
-public:
-    void testName();
-    void testRecordRoundTrip();
-    void testCompression();
-    void testMalformedPackets();
-    void testPacketSizeLimit();
-    void testDnsSdAdvertisement();
-    void testDnsSdGoodbye();
-    void testDnsSdValidation();
-    void testDnsSdQueryResponse();
-    void testDnsSdCache();
-    void testInterfaceScoping();
-};
-
-} // namespace test

tests/test_peerdiscovery.cpp

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: MIT
 
 #include "test_peerdiscovery.h"
+#include "opendht/value.h"
 
 #include <mutex>
 #include <condition_variable>
@@ -10,8 +11,24 @@ namespace test {
 
 using namespace std::literals;
 
-constexpr dht::NetId NETWORK_ID = 10;
-constexpr in_port_t NODE_PORT = 50000;
+constexpr unsigned MULTICAST_PORT = 2222;
+constexpr auto DHT_NODE_NAME {"dht"sv};
+constexpr auto JAMI_NODE_NAME {"jami"sv};
+
+struct DhtNode
+{
+    dht::InfoHash nodeid;
+    in_port_t node_port;
+    dht::NetId nid;
+    MSGPACK_DEFINE(nodeid, node_port, nid)
+};
+struct JamiNode
+{
+    int num;
+    char cha;
+    std::string str;
+    MSGPACK_DEFINE(num, cha, str)
+};
 
 CPPUNIT_TEST_SUITE_REGISTRATION(PeerDiscoveryTester);
 
@@ -22,40 +39,60 @@ PeerDiscoveryTester::setUp()
 void
 PeerDiscoveryTester::testMulticastToTwoNodes()
 {
-    const auto nodeId = dht::InfoHash::get("opendht01");
+    DhtNode dhtNode;
+    dhtNode.nid = 10;
+    dhtNode.node_port = 50000;
+    dhtNode.nodeid = dht::InfoHash::get("opendht01");
+
+    JamiNode jamiNode;
+    jamiNode.num = 100;
+    jamiNode.cha = 'a';
+    jamiNode.str = "jami01";
 
     std::mutex lock;
     std::condition_variable cv;
-    unsigned discovered {0};
-    unsigned wrongNetwork {0};
+    unsigned countDht {0};
+    unsigned countJami {0};
     {
         std::unique_lock l(lock);
-        dht::PeerDiscovery publisher;
-        dht::PeerDiscovery browser;
-        dht::PeerDiscovery otherNetworkBrowser;
+        dht::PeerDiscovery testDht(MULTICAST_PORT);
+        dht::PeerDiscovery testJami(MULTICAST_PORT);
 
-        browser.startDiscovery(NETWORK_ID, [&](const dht::InfoHash& peerId, dht::SockAddr&& address) {
-            CPPUNIT_ASSERT_EQUAL(nodeId, peerId);
-            CPPUNIT_ASSERT_EQUAL(NODE_PORT, address.getPort());
+        testJami.startDiscovery<DhtNode>(DHT_NODE_NAME, [&](DhtNode&& v, dht::SockAddr&&) {
+            CPPUNIT_ASSERT_EQUAL(dhtNode.node_port, v.node_port);
+            CPPUNIT_ASSERT_EQUAL(dhtNode.nodeid, v.nodeid);
+            CPPUNIT_ASSERT_EQUAL(dhtNode.nid, v.nid);
             {
                 std::lock_guard l(lock);
-                discovered++;
+                countDht++;
             }
             cv.notify_all();
         });
-        otherNetworkBrowser.startDiscovery(NETWORK_ID + 1, [&](const dht::InfoHash&, dht::SockAddr&&) {
-            std::lock_guard l(lock);
-            wrongNetwork++;
+
+        testJami.startDiscovery(JAMI_NODE_NAME, [&](msgpack::object&& obj, dht::SockAddr&&) {
+            auto v = obj.as<JamiNode>();
+            CPPUNIT_ASSERT_EQUAL(jamiNode.num, v.num);
+            CPPUNIT_ASSERT_EQUAL(jamiNode.cha, v.cha);
+            CPPUNIT_ASSERT_EQUAL(jamiNode.str, v.str);
+            {
+                std::lock_guard l(lock);
+                countJami++;
+            }
+            cv.notify_all();
         });
 
-        publisher.startPublish(nodeId, NETWORK_ID, NODE_PORT);
-        CPPUNIT_ASSERT(cv.wait_for(l, std::chrono::seconds(5), [&] { return discovered > 0; }));
-        CPPUNIT_ASSERT_EQUAL(0u, wrongNetwork);
+        testDht.startPublish(DHT_NODE_NAME, dhtNode);
+        CPPUNIT_ASSERT(cv.wait_for(l, std::chrono::seconds(5), [&] { return countDht > 0; }));
+
+        testDht.startPublish(JAMI_NODE_NAME, jamiNode);
+        CPPUNIT_ASSERT(cv.wait_for(l, std::chrono::seconds(5), [&] { return countDht > 1 and countJami > 0; }));
+        // we don't verify count values since its a continious multicasting
 
         l.unlock();
-        publisher.stopPublish();
-        browser.stopDiscovery();
-        otherNetworkBrowser.stopDiscovery();
+        testDht.stopPublish(DHT_NODE_NAME);
+        testDht.stopPublish(JAMI_NODE_NAME);
+        testJami.stopDiscovery(DHT_NODE_NAME);
+        testJami.stopDiscovery(JAMI_NODE_NAME);
     }
 }