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

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

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/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