proxy client: always resubscribe on connectivity change

Track subscribe/permanent-put expiration with std::chrono::system_clock
(wall clock) instead of steady_clock in both the proxy client and server,
so persisted expirations survive restarts and map directly to epoch time.

- DhtProxyServer::clock is now system_clock; expireTimer/expireNotifyTimer
  use asio::system_timer.
- DhtProxyClient::refreshSubscriberTimer uses asio::system_timer and
  system_clock::now() for subscriber refresh.

include/opendht/dht_proxy_server.h

@@ -76,7 +76,7 @@ public:
 
     asio::io_context& io_context() const;
 
-    using clock = std::chrono::steady_clock;
+    using clock = std::chrono::system_clock;
     using time_point = clock::time_point;
 
     struct PushStats
@@ -405,8 +405,8 @@ private:
         std::string pushToken;
         std::string clientId;
         std::shared_ptr<PushSessionContext> sessionCtx;
-        std::unique_ptr<asio::steady_timer> expireTimer;
-        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
         Sp<Value> value;
         PushType type;
         std::string topic;
@@ -419,7 +419,7 @@ private:
             p.pack("value");
             p.pack(value);
             p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(std::chrono::system_clock::to_time_t(expiration));
             if (not clientId.empty()) {
                 p.pack("cid");
                 p.pack(clientId);
@@ -464,8 +464,8 @@ private:
         std::string clientId;
         std::shared_ptr<PushSessionContext> sessionCtx;
         std::future<size_t> internalToken;
-        std::unique_ptr<asio::steady_timer> expireTimer;
-        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
         PushType type;
         std::string topic;
 
@@ -476,7 +476,7 @@ private:
             p.pack("cid");
             p.pack(clientId);
             p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(std::chrono::system_clock::to_time_t(expiration));
             if (sessionCtx) {
                 std::lock_guard l(sessionCtx->lock);
                 p.pack("sid");

include/opendht/http.h

@@ -162,34 +162,11 @@ 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_; }
 
-    /** 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);
+    void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC);
 
     std::shared_ptr<log::Logger> getLogger() const { return logger_; }
 
@@ -197,27 +174,17 @@ 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_;
 };
@@ -350,12 +317,6 @@ 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/proxy.h

@@ -8,7 +8,7 @@ namespace dht {
 namespace proxy {
 
 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
-constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
+constexpr const std::chrono::hours OP_MARGIN {8};   // eight hours
 
 using ListenToken = uint64_t;
 

src/dht_proxy_client.cpp

@@ -34,20 +34,20 @@ struct DhtProxyClient::Listener
     Sp<OperationState> opstate;
     std::shared_ptr<http::Request> request;
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
-    std::unique_ptr<asio::steady_timer> refreshSubscriberTimer;
+    std::unique_ptr<asio::system_timer> refreshSubscriberTimer;
 #endif
 };
 
 struct PermanentPut
 {
-    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::steady_timer>&& j, const Sp<std::atomic_bool>& o)
+    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::system_timer>&& j, const Sp<std::atomic_bool>& o)
         : value(v)
         , refreshPutTimer(std::move(j))
         , ok(o)
     {}
 
     Sp<Value> value;
-    std::unique_ptr<asio::steady_timer> refreshPutTimer;
+    std::unique_ptr<asio::system_timer> refreshPutTimer;
     Sp<std::atomic_bool> ok;
 };
 
@@ -424,7 +424,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         auto& search = searches_[key];
         if (val->id) {
             auto id = val->id;
-            auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+            auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                         proxy::OP_TIMEOUT - proxy::OP_MARGIN);
             refreshPutTimer->async_wait(
                 std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
@@ -540,7 +540,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
                             auto it = search.pendingPuts.find(val);
                             if (it != search.pendingPuts.end()) {
                                 auto sok = std::make_shared<std::atomic_bool>(ok);
-                                auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+                                auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                                             proxy::OP_TIMEOUT
                                                                                                 - proxy::OP_MARGIN);
                                 refreshPutTimer->async_wait(
@@ -652,21 +652,7 @@ DhtProxyClient::getProxyInfos()
     if (logger_)
         logger_->debug("[proxy:client] [status] sending request");
 
-    // 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_);
+    auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
     queryProxyInfo(infoState, resolver, AF_INET);
     queryProxyInfo(infoState, resolver, AF_INET6);
     std::lock_guard l(resolverLock_);
@@ -905,8 +891,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
                  * (if the proxy crash for any reason)
                  */
                 if (!l->second.refreshSubscriberTimer)
-                    l->second.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
-                l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+                    l->second.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
+                l->second.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                              - proxy::OP_MARGIN);
                 l->second.refreshSubscriberTimer->async_wait(
                     std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
@@ -1206,9 +1192,9 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         for (auto& put : search.second.puts) {
             doPut(key, put.second.value, [ok = put.second.ok](bool result) { *ok = result; }, time_point::max(), true);
             if (!put.second.refreshPutTimer) {
-                put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
+                put.second.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_);
             }
-            put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+            put.second.refreshPutTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                    - proxy::OP_MARGIN);
             put.second.refreshPutTimer->async_wait(
                 std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, put.first));
@@ -1224,8 +1210,7 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         // Connectivity changed, refresh all subscribe
         for (auto& search : searches_)
             for (auto& listener : search.second.listeners)
-                if (!listener.second.opstate->ok)
-                    resubscribe(search.first, listener.first, listener.second);
+                resubscribe(search.first, listener.first, listener.second);
         return;
     }
     if (logger_)
@@ -1289,10 +1274,10 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                 auto vid = std::stoull(vidIt->second);
                 auto& put = search.puts.at(vid);
                 if (!put.refreshPutTimer)
-                    put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
-                                                                               std::chrono::steady_clock::now());
+                    put.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
+                                                                               std::chrono::system_clock::now());
                 else
-                    put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
+                    put.refreshPutTimer->expires_at(std::chrono::system_clock::now());
                 put.refreshPutTimer->async_wait(
                     std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
                 ret = PushNotificationResult::PutRefresh;
@@ -1304,12 +1289,12 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                                                : PushNotificationResult::ListenRefresh;
             }
         } else {
-            auto key = InfoHash(notification.at("key"));
             system_clock::time_point sendTime = system_clock::time_point::min();
             try {
                 sendTime = system_clock::time_point(std::chrono::milliseconds(std::stoull(notification.at("t"))));
             } catch (...) {
             }
+            auto key = InfoHash(notification.at("key"));
             auto& search = searches_.at(key);
             auto expired = notification.find("exp");
             for (auto& list : search.listeners) {
@@ -1392,9 +1377,9 @@ DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
     header.method(restinio::http_method_subscribe());
     header.request_target("/key/" + key.toString());
     if (!listener.refreshSubscriberTimer) {
-        listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
+        listener.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
     }
-    listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+    listener.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     listener.refreshSubscriberTimer->async_wait(
         std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
     auto vcb = listener.cb;

src/dht_proxy_server.cpp

@@ -126,8 +126,8 @@ DhtProxyServer::ServerStats::toJson() const
     result["putCount"] = static_cast<Json::UInt64>(putCount);
     result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
     result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
-    result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
-    result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
+    result["serverStartTime"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(serverStartTime));
+    result["lastUpdated"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(lastUpdated));
     result["androidPush"] = androidPush.toJson();
     result["iosPush"] = iosPush.toJson();
     result["unifiedPush"] = unifiedPush.toJson();
@@ -210,7 +210,7 @@ DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
         clientId = cid->as<std::string>();
     }
     if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = std::chrono::system_clock::from_time_t(exp->as<time_t>());
     }
     if (auto token = findMapValue(o, "token"sv)) {
         pushToken = token->as<std::string>();
@@ -240,7 +240,7 @@ DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
         clientId = cid->as<std::string>();
     }
     if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = std::chrono::system_clock::from_time_t(exp->as<time_t>());
     }
     if (auto sid = findMapValue(o, "sid"sv)) {
         if (not sessionCtx)
@@ -414,7 +414,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                     logger_->debug("[proxy:server] Loading {} persistent puts", puts_.size());
                 for (auto& put : puts_) {
                     for (auto& pput : put.second.puts) {
-                        pput.second.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
+                        pput.second.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                        pput.second.expiration);
                         pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut,
                                                                       this,
@@ -437,7 +437,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                 }
                                 return json;
                             };
-                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            pput.second.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                                  pput.second.expiration
                                                                                                      - proxy::OP_MARGIN);
                             pput.second.expireNotifyTimer->async_wait(
@@ -450,7 +450,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                           pput.second.topic));
                         }
 #endif
-                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, time_point::max(), true);
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, dht::time_point::max(), true);
                     }
                 }
             } else {
@@ -487,7 +487,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                                                                  expired);
                                                });
                             // expire notify
-                            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                               listener.expiration
                                                                                                   - proxy::OP_MARGIN);
                             auto jsonProvider = [infoHash = listeners.first.toString(),
@@ -509,7 +509,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                           listener.type,
                                           listener.topic));
                             // cancel push listen
-                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                         listener.expiration);
                             listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
                                                                        this,
@@ -907,6 +907,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         Json::Value r;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
         auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
+
         if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) {
             auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
             response.set_body(RESP_MSG_JSON_INCORRECT);
@@ -951,14 +952,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         auto& listener = *listIt;
 
         // Expiration
-        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+        auto timeout = clock::now() + proxy::OP_TIMEOUT;
         listener.expiration = timeout;
         listener.type = type;
         listener.topic = topic;
         if (listener.expireNotifyTimer)
             listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
         else
-            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
+            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(), timeout - proxy::OP_MARGIN);
         auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() {
             Json::Value json;
             json["timeout"] = h;
@@ -975,7 +976,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
                                                          listener.type,
                                                          listener.topic));
         if (!listener.expireTimer)
-            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
         else
             listener.expireTimer->expires_at(timeout);
         listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
@@ -1405,7 +1406,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                     topic = pVal["topic"].asString();
                 }
                 std::lock_guard lock(lockSearchPuts_);
-                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+                auto timeout = clock::now() + proxy::OP_TIMEOUT;
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
                     for (auto& pp : sPuts.puts) {
@@ -1439,7 +1440,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                 pput.expiration = timeout;
                 if (not pput.expireTimer) {
                     // cancel permanent put
-                    pput.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+                    pput.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                     if (not pushToken.empty()) {
                         pput.pushToken = pushToken;
@@ -1478,7 +1479,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         return json;
                     };
                     if (!pput.expireNotifyTimer)
-                        pput.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                        pput.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                       timeout - proxy::OP_MARGIN);
                     else
                         pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -1507,7 +1508,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                         response.done();
                     }
                 },
-                time_point::max(),
+                dht::time_point::max(),
                 permanent);
             return restinio::request_handling_status_t::accepted;
         } else {

src/http.cpp

@@ -867,31 +867,12 @@ 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();
@@ -899,29 +880,9 @@ 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()
 {
@@ -944,68 +905,27 @@ filter(const std::vector<asio::ip::tcp::endpoint>& epts, sa_family_t family)
 }
 
 void
-Resolver::add_callback(ResolverCb cb, sa_family_t family, bool freshOnly)
+Resolver::add_callback(ResolverCb cb, sa_family_t family)
 {
     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(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));
+        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));
+                                           });
     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(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();
+                   [cb = std::move(cb),
+                    ec = ec_,
+                    endpoints = family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family)] { cb(ec, endpoints); });
     }
 }
 
@@ -1023,45 +943,30 @@ 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());
                                 }
-                                complete(ec,
-                                         std::vector<asio::ip::tcp::endpoint> {endpoints.begin(), endpoints.end()});
+                                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();
+                                }
                             });
 }
 
@@ -1539,12 +1444,7 @@ 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) {
@@ -1558,28 +1458,17 @@ Request::resolve_and_connect(bool freshOnly)
                 } 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 (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 {
+                            if (ec)
                                 sthis->terminate(asio::error::not_connected);
-                            }
+                            else
+                                sthis->post();
                         }
                     });
                 } else
                     this_.post();
             }
         },
-        family_,
-        freshOnly);
+        family_);
 }
 
 void