From e9208cfbe477e0b5bceb18b8f476262956a468ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cyrille=20B=C3=A9raud?= Date: Thu, 3 Sep 2026 07:50:51 -0400 Subject: [PATCH] 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> --- include/opendht/http.h | 41 +++++++++- src/dht_proxy_client.cpp | 16 +++- src/http.cpp | 173 ++++++++++++++++++++++++++++++++------- 3 files changed, 197 insertions(+), 33 deletions(-) diff --git a/include/opendht/http.h b/include/opendht/http.h index 9adec063..6060da3b 100644 --- a/include/opendht/http.h +++ b/include/opendht/http.h @@ -162,11 +162,34 @@ public: const bool ssl = false, std::shared_ptr 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 endpoints, + std::shared_ptr 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 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 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&& 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 timer_; std::shared_ptr destroyed_; std::vector endpoints_; bool completed_ {false}; + bool resolving_ {false}; std::queue cbs_; + std::queue freshCbs_; std::shared_ptr logger_; }; @@ -317,6 +350,12 @@ private: void connect(std::vector&& 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); diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 96e4d414..c49702c0 100644 --- a/src/dht_proxy_client.cpp +++ b/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(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 knownEndpoints; + { + std::lock_guard l(resolverLock_); + if (resolver_) + knownEndpoints = resolver_->get_endpoints(); + } + auto resolver = knownEndpoints.empty() + ? std::make_shared(httpContext_, proxyUrl_, logger_) + : std::make_shared(httpContext_, + proxyUrl_, + std::move(knownEndpoints), + logger_); queryProxyInfo(infoState, resolver, AF_INET); queryProxyInfo(infoState, resolver, AF_INET6); std::lock_guard l(resolverLock_); diff --git a/src/http.cpp b/src/http.cpp index e7639f97..aa897f73 100644 --- a/src/http.cpp +++ b/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 endpoints, + std::shared_ptr logger) + : url_(std::move(url)) + , resolver_(ctx) + , destroyed_(std::make_shared(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 +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& 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& 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& 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&& endpoints) +{ + decltype(cbs_) cbs; + decltype(freshCbs_) freshCbs; + asio::error_code reportedEc; + std::vector 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(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 {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 {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 wthis = shared_from_this(); resolver_->add_callback( [wthis](const asio::error_code& ec, std::vector 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