search: announce immediately if synced and not pending/completed

Factor various ways of sending listen requests.

include/opendht/default_types.h

@@ -114,6 +114,8 @@ public:
     ImMessage() {}
     ImMessage(dht::Value::Id id, std::string&& m, long d = 0)
         : id(id), msg(std::move(m)), date(d) {}
+    ImMessage(dht::Value::Id id, std::string&& dt, std::string&& m, long d = 0)
+        : id(id), msg(std::move(m)), datatype(std::move(dt)), date(d) {}
 
     virtual void unpackValue(const Value& v) override {
         to = v.recipient;
@@ -123,10 +125,11 @@ public:
     dht::InfoHash to;
     dht::Value::Id id {0};
     std::string msg;
+    std::string datatype;
     long date {0};
     ImStatus status {ImStatus::NONE};
 
-    MSGPACK_DEFINE_MAP(id, msg, date, status)
+    MSGPACK_DEFINE_MAP(id, msg, date, status, datatype)
 };
 
 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>

include/opendht/dht_proxy_client.h

@@ -278,6 +278,10 @@ private:
     size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
     bool doCancelListen(const InfoHash& key, size_t token);
 
+    struct ListenState;
+    void sendListen(const std::shared_ptr<restbed::Request>& request, const ValueCallback&, const Value::Filter& filter, const Sp<ListenState>& state);
+    void sendSubscribe(const std::shared_ptr<restbed::Request>& request, const Sp<proxy::ListenToken>&, const Sp<ListenState>& state);
+
     void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
 
     /**

src/dht.cpp

@@ -516,7 +516,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             });
         }
         auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        r->second.req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
             [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
             { /* on done */
                 if (auto sr = ws.lock()) {
@@ -545,6 +545,11 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 }
             }
         );
+        // Here the request may have failed and the CachedListenStatus removed
+        r = n.listenStatus.find(query);
+        if (r != n.listenStatus.end()) {
+            r->second.req = new_req;
+        }
     }
 }
 
@@ -1535,10 +1540,7 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                     if (ack == n.acked.end() or not ack->second.first) {
                         out << ' ';
                     } else {
-                        if (ack->second.first->completed())
-                            out << 'a';
-                        else if (ack->second.first->pending())
-                            out << 'f';
+                        out << ack->second.first->getStateChar();
                     }
                 }
                 out << "] ";

src/dht_proxy_client.cpp

@@ -33,12 +33,13 @@
 
 namespace dht {
 
+struct DhtProxyClient::ListenState {
+    std::atomic_bool ok {true};
+    std::atomic_bool cancel {false};
+};
+
 struct DhtProxyClient::Listener
 {
-    struct State {
-        std::atomic_bool ok {true};
-        std::atomic_bool cancel {false};
-    };
     ValueCache cache;
     Sp<Scheduler::Job> cacheExpirationJob {};
     ValueCallback cb;
@@ -46,7 +47,7 @@ struct DhtProxyClient::Listener
     Sp<restbed::Request> req;
     std::thread thread;
     unsigned callbackId;
-    Sp<State> state;
+    Sp<ListenState> state;
     Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
@@ -322,15 +323,16 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         std::lock_guard<std::mutex> lock(searchLock_);
         auto id = val->id;
         auto search = searches_.emplace(key, ProxySearch{}).first;
-        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
             if (s != searches_.end()) {
                 auto p = s->second.puts.find(id);
                 if (p != s->second.puts.end()) {
                     doPut(key, p->second.value, {}, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
             }
         })});
@@ -652,14 +654,102 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
     return canceled;
 }
 
+
+void
+DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const ValueCallback& cb, const Value::Filter& filter, const Sp<ListenState>& state) {
+    auto settings = std::make_shared<restbed::Settings>();
+    std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+    settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
+    req->set_method("LISTEN");
+    restbed::Http::async(req,
+        [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                  const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                while (restbed::Http::is_open(req) and not state->cancel) {
+                    restbed::Http::fetch("\n", reply);
+                    if (state->cancel)
+                        break;
+                    std::string body;
+                    reply->get_body(body);
+                    reply->set_body(""); // Reset the body for the next fetch
+
+                    Json::Value json;
+                    std::string err;
+                    Json::CharReaderBuilder rbuilder;
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        auto value = std::make_shared<Value>(json);
+                        if ((not filter or filter(*value)) and cb)  {
+                            std::lock_guard<std::mutex> lock(lockCallbacks);
+                            callbacks_.emplace_back([cb, value, state]() {
+                                if (not state->cancel and not cb({value}, false))
+                                    state->cancel = true;
+                            });
+                            loopSignal_();
+                        }
+                    }
+                }
+            } catch (std::runtime_error&) {
+                if (not state->cancel) {
+                    DHT_LOG.w("Listen closed by the proxy server");
+                    state->ok = false;
+                }
+            }
+        } else {
+            state->ok = false;
+        }
+    }, settings).get();
+    auto& s = *state;
+    if (not s.ok and not s.cancel)
+        opFailed();
+}
+
+void
+DhtProxyClient::sendSubscribe(const std::shared_ptr<restbed::Request>& req, const Sp<proxy::ListenToken>& token, const Sp<ListenState>& state) {
+#if OPENDHT_PUSH_NOTIFICATIONS
+    req->set_method("SUBSCRIBE");
+    fillBodyToGetToken(req);
+    auto settings = std::make_shared<restbed::Settings>();
+    restbed::Http::async(req, [state, token](const std::shared_ptr<restbed::Request>&,
+                                             const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                restbed::Http::fetch("\n", reply);
+                std::string body;
+                reply->get_body(body);
+
+                std::string err;
+                Json::Value json;
+                Json::CharReaderBuilder rbuilder;
+                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                    if (not json.isMember("token")) {
+                        state->ok = false;
+                        return;
+                    }
+                    *token = unpackId(json, "token");
+                }
+            } catch (std::runtime_error&) {
+                // NOTE: Http::close() can occurs here. Ignore this.
+            }
+        } else {
+            state->ok = false;
+        }
+    }, settings).get();
+    auto& s = *state;
+    if (not s.ok and not s.cancel)
+        opFailed();
+#endif
+}
+
 size_t
 DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
 {
     scheduler.syncTime();
     restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
-    auto req = std::make_shared<restbed::Request>(uri);
-    req->set_method(deviceKey_.empty() ? "LISTEN" : "SUBSCRIBE");
-
     std::lock_guard<std::mutex> lock(searchLock_);
     auto search = searches_.find(key);
     if (search == searches_.end()) {
@@ -668,6 +758,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
     }
     DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
 
+    auto req = std::make_shared<restbed::Request>(uri);
     auto token = ++listenerToken_;
     auto l = search->second.listeners.find(token);
     if (l == search->second.listeners.end()) {
@@ -687,9 +778,12 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                 scheduler.edit(l->second.cacheExpirationJob, next);
             }), req, std::move(f)
         }).first;
+    } else {
+        if (l->second.state)
+            l->second.state->cancel = true;
     }
 
-    auto state = std::make_shared<Listener::State>();
+    auto state = std::make_shared<ListenState>();
     l->second.state = state;
     l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
         if (state->cancel)
@@ -715,82 +809,34 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         loopSignal_();
         return true;
     };
+
     auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
     auto vcb = l->second.cb;
     l->second.pushNotifToken = pushNotifToken;
-    l->second.thread = std::thread([this,req,filter,vcb,pushNotifToken,state]()
-        {
-            auto settings = std::make_shared<restbed::Settings>();
-            if (deviceKey_.empty()) {
-                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
-            }
-#if OPENDHT_PUSH_NOTIFICATIONS
-            else
-                fillBodyToGetToken(req);
-#endif
+    l->second.req = req;
 
-            restbed::Http::async(req,
-                [this, filter, vcb, pushNotifToken, state](const std::shared_ptr<restbed::Request>& req,
-                                                           const std::shared_ptr<restbed::Response>& reply) {
-                auto code = reply->get_status_code();
-                if (code == 200) {
-                    try {
-                        std::string err;
-                        Json::Value json;
-                        Json::CharReaderBuilder rbuilder;
-                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                        if (!deviceKey_.empty()) {
-                            restbed::Http::fetch("\n", reply);
-                            if (state->cancel)
-                                return;
-                            std::string body;
-                            reply->get_body(body);
-
-                            auto* char_data = reinterpret_cast<const char*>(&body[0]);
-                            if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
-                                *pushNotifToken = unpackId(json, "token");
-                            } else {
-                                state->ok = false;
-                            }
-                        } else {
-                            while (restbed::Http::is_open(req) and not state->cancel) {
-                                restbed::Http::fetch("\n", reply);
-                                if (state->cancel)
-                                    break;
-                                std::string body;
-                                reply->get_body(body);
-                                reply->set_body(""); // Reset the body for the next fetch
-
-                                auto* char_data = reinterpret_cast<const char*>(&body[0]);
-                                if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
-                                    auto value = std::make_shared<Value>(json);
-                                    if ((not filter or filter(*value)) and vcb)  {
-                                        std::lock_guard<std::mutex> lock(lockCallbacks);
-                                        callbacks_.emplace_back([vcb, value, state]() {
-                                            if (not state->cancel and not vcb({value}, false))
-                                                state->cancel = true;
-                                        });
-                                        loopSignal_();
-                                    }
-                                } else {
-                                    state->ok = false;
-                                }
-                            }
-                        }
-                    } catch (std::runtime_error&) {
-                        // NOTE: Http::close() can occurs here. Ignore this.
-                    }
-                } else {
-                    state->ok = false;
+    if (not deviceKey_.empty()) {
+        // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
+        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
+            if (state->cancel)
+                return;
+            std::lock_guard<std::mutex> lock(searchLock_);
+            auto s = searches_.find(key);
+            if (s != searches_.end()) {
+                auto l = s->second.listeners.find(token);
+                if (l != s->second.listeners.end()) {
+                    resubscribe(key, l->second);
                 }
-            }, settings).get();
-            auto& s = *state;
-            if (not s.ok and not s.cancel) {
-                opFailed();
             }
-        }
-    );
+        });
+        l->second.thread = std::thread([this, req,pushNotifToken, state](){
+            sendSubscribe(req,pushNotifToken,state);
+        });
+    } else {
+        l->second.thread = std::thread([this, req, vcb, filter, state]{
+            sendListen(req,vcb,filter,state);
+        });
+    }
     return token;
 }
 
@@ -880,75 +926,28 @@ DhtProxyClient::restartListeners()
     DHT_LOG.d("Restarting listeners");
     std::lock_guard<std::mutex> lock(searchLock_);
     for (auto& search: searches_) {
-    for (auto& l: search.second.listeners) {
-        auto& listener = l.second;
-        auto state = listener.state;
-        if (listener.thread.joinable()) {
-            state->cancel = true;
-            if (listener.req)
-                restbed::Http::close(listener.req);
-            listener.thread.join();
-        }
-        // Redo listen
-        state->cancel = false;
-        state->ok = true;
-        auto filter = listener.filter;
-        auto cb = listener.cb;
-        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
-        auto req = std::make_shared<restbed::Request>(uri);
-        req->set_method("LISTEN");
-        listener.req = req;
-        listener.thread = std::thread([this, filter, cb, req, state]()
-            {
-                auto settings = std::make_shared<restbed::Settings>();
-                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
-
-                auto ok = std::make_shared<std::atomic_bool>(true);
-                restbed::Http::async(req,
-                    [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
-                                           const std::shared_ptr<restbed::Response>& reply) {
-                    auto code = reply->get_status_code();
-
-                    if (code == 200) {
-                        try {
-                            while (restbed::Http::is_open(req) and not state->cancel) {
-                                restbed::Http::fetch("\n", reply);
-                                if (state->cancel)
-                                    break;
-                                std::string body;
-                                reply->get_body(body);
-                                reply->set_body(""); // Reset the body for the next fetch
-
-                                Json::Value json;
-                                std::string err;
-                                Json::CharReaderBuilder rbuilder;
-                                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
-                                    auto value = std::make_shared<Value>(json);
-                                    if ((not filter or filter(*value)) and cb)  {
-                                        std::lock_guard<std::mutex> lock(lockCallbacks);
-                                        callbacks_.emplace_back([cb, value, state]() {
-                                            if (not state->cancel and not cb({value}, false))
-                                                state->cancel = true;
-                                        });
-                                        loopSignal_();
-                                    }
-                                }
-                            }
-                        } catch (std::runtime_error&) {
-                            // NOTE: Http::close() can occurs here. Ignore this.
-                        }
-                    } else {
-                        state->ok = false;
-                    }
-                }, settings).get();
-                auto& s = *state;
-                if (not s.ok and not s.cancel)
-                    opFailed();
+        for (auto& l: search.second.listeners) {
+            auto& listener = l.second;
+            auto state = listener.state;
+            if (listener.thread.joinable()) {
+                state->cancel = true;
+                if (listener.req)
+                    restbed::Http::close(listener.req);
+                listener.thread.join();
             }
-        );
-    }
+            // Redo listen
+            state->cancel = false;
+            state->ok = true;
+            auto filter = listener.filter;
+            auto cb = listener.cb;
+            restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
+            auto req = std::make_shared<restbed::Request>(uri);
+            req->set_method("LISTEN");
+            listener.req = req;
+            listener.thread = std::thread([this, req, cb, filter, state]() {
+                sendListen(req, cb, filter, state);
+            });
+        }
     }
 }
 
@@ -967,7 +966,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
             if (vidIt != notification.end()) {
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
-                auto put = search.puts.at(vid);
+                auto& put = search.puts.at(vid);
                 scheduler.edit(put.refreshJob, scheduler.time());
                 loopSignal_();
             } else {
@@ -1002,6 +1001,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     if (deviceKey_.empty()) return;
+    scheduler.syncTime();
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
     restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
@@ -1016,37 +1016,9 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     state->ok = true;
     listener.req = req;
     listener.pushNotifToken = pushNotifToken;
+    scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     listener.thread = std::thread([this, req, pushNotifToken, state]() {
-        fillBodyToGetToken(req);
-        auto settings = std::make_shared<restbed::Settings>();
-        restbed::Http::async(req,
-            [pushNotifToken, state](const std::shared_ptr<restbed::Request>&,
-                                       const std::shared_ptr<restbed::Response>& reply) {
-            auto code = reply->get_status_code();
-            if (code == 200) {
-                try {
-                    restbed::Http::fetch("\n", reply);
-                    std::string body;
-                    reply->get_body(body);
-
-                    std::string err;
-                    Json::Value json;
-                    Json::CharReaderBuilder rbuilder;
-                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
-                        if (!json.isMember("token")) return;
-                        *pushNotifToken = unpackId(json, "token");
-                    }
-                } catch (std::runtime_error&) {
-                    // NOTE: Http::close() can occurs here. Ignore this.
-                }
-            } else {
-                state->ok = false;
-            }
-        }, settings).get();
-        auto& s = *state;
-        if (not s.ok and not s.cancel)
-            opFailed();
+        sendSubscribe(req, pushNotifToken, state);
     });
 #endif
 }

src/dht_proxy_server.cpp

@@ -588,8 +588,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                         if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
                             // Build the Value from json
                             auto value = std::make_shared<Value>(root);
-                            auto permanent = root.isMember("permanent");
-                            std::cout << "Got put " << infoHash << " " << value << " " << (permanent ? "permanent" : "") << std::endl;
+                            bool permanent = root.isMember("permanent");
+                            std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl;
 
                             if (permanent) {
                                 std::string pushToken, clientId, platform;
@@ -606,28 +606,30 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 auto vid = value->id;
                                 auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
+                                auto& pput = r.first->second;
                                 if (r.second) {
-                                    r.first->second.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
-                                        std::cout << "Permanent put expired: " << infoHash << std::endl;
+                                    pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
+                                        std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
                                         cancelPut(infoHash, vid);
                                     });
 #if OPENDHT_PUSH_NOTIFICATIONS
-                                    if (not pushToken.empty())
-                                        r.first->second.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+                                    if (not pushToken.empty()) {
+                                        pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                             [this, infoHash, vid, pushToken, clientId, isAndroid]
                                         {
-                                            std::cout << "Permanent put refresh: " << infoHash << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
                                             json["vid"] = std::to_string(vid);
                                             sendPushNotification(pushToken, json, isAndroid);
                                         });
+                                    }
 #endif
                                 } else {
-                                    scheduler_.edit(r.first->second.expireJob, timeout);
-                                    if (r.first->second.expireNotifyJob)
-                                        scheduler_.edit(r.first->second.expireNotifyJob, timeout - proxy::OP_MARGIN);
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -652,7 +654,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                 } else {
                     s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
                 }
-            } catch (...) {
+            } catch (const std::exception& e) {
+                std::cout << "Error performing put: " << e.what() << std::endl;
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
             }
         }

src/dhtrunner.cpp

@@ -342,7 +342,9 @@ std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getPublicAddress(af);
+    if (auto dht = activeDht())
+        return dht->getPublicAddress(af);
+    return {};
 }
 std::vector<std::string>
 DhtRunner::getPublicAddressStr(sa_family_t af)
@@ -599,20 +601,24 @@ DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Wher
 void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
-    std::lock_guard<std::mutex> lck(storage_mtx);
-    pending_ops.emplace([=](SecureDht& dht) {
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
 #if OPENDHT_PROXY_CLIENT
-        auto it = listeners_.find(token);
-        if (it == listeners_.end()) return;
-        if (it->second.tokenClassicDht)
-            dht_->cancelListen(h, it->second.tokenClassicDht);
-        if (it->second.tokenProxyDht and dht_via_proxy_)
-            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-        listeners_.erase(it);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(token);
+            if (it == listeners_.end()) return;
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+        });
 #else
-        dht.cancelListen(h, token);
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, token);
+        });
 #endif // OPENDHT_PROXY_CLIENT
-    });
+    }
     cv.notify_all();
 }
 
@@ -621,20 +627,21 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            auto token = ftoken.get();
 #if OPENDHT_PROXY_CLIENT
-            auto it = listeners_.find(token);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(ftoken.get());
             if (it == listeners_.end()) return;
             if (it->second.tokenClassicDht)
                 dht_->cancelListen(h, it->second.tokenClassicDht);
             if (it->second.tokenProxyDht and dht_via_proxy_)
                 dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
             listeners_.erase(it);
-#else
-            dht.cancelListen(h, token);
-#endif // OPENDHT_PROXY_CLIENT
         });
+#else
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, ftoken.get());
+        });
+#endif // OPENDHT_PROXY_CLIENT
     }
     cv.notify_all();
 }
@@ -971,7 +978,7 @@ DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& da
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
+        pending_ops_prio.emplace([=](SecureDht&) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });

src/network_engine.cpp

@@ -292,24 +292,20 @@ NetworkEngine::requestStep(Sp<Request> sreq)
     auto err = send((char*)req.msg.data(), req.msg.size(),
             (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
             node.getAddr());
-    std::weak_ptr<Request> wreq = sreq;
     if (err == ENETUNREACH  ||
         err == EHOSTUNREACH ||
         err == EAFNOSUPPORT ||
         err == EPIPE)
     {
-        scheduler.add(now, [this,wreq]() mutable {
-            if (auto req = wreq.lock()) {
-                req->node->setExpired();
-                if (not req->node->id)
-                    requests.erase(req->tid);
-            }
-        });
+        node.setExpired();
+        if (not node.id)
+            requests.erase(req.tid);
     } else {
         if (err != EAGAIN) {
             ++req.attempt_count;
         }
         req.last_try = now;
+        std::weak_ptr<Request> wreq = sreq;
         scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] {
             if (auto req = wreq.lock())
                 requestStep(req);

src/request.h

@@ -57,6 +57,15 @@ struct Request {
     bool pending() const { return state_ == State::PENDING; }
     bool over() const { return not pending(); }
     State getState() const { return state_; }
+    char getStateChar() const {
+        switch (state_) {
+            case State::PENDING:   return 'f';
+            case State::CANCELLED: return 'c';
+            case State::EXPIRED:   return 'e';
+            case State::COMPLETED: return 'a';
+            default:               return '?';
+        }
+    }
 
     Request(State state = State::PENDING) : state_(state) {}
     Request(MessageType type, Tid tid,

src/search.h

@@ -331,7 +331,7 @@ struct Dht::SearchNode {
         return ((gs != getStatus.cend() and gs->second and gs->second->pending())
                 or ack == acked.cend() or not ack->second.first or ack->second.first->pending())
                 ? time_point::max()
-                : ack->second.second - REANNOUNCE_MARGIN;
+                : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
     }
 
     /**

tools/dhtnode.cpp

@@ -199,6 +199,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             continue;
         } else if (op == "cc") {
             dht->connectivityChanged();
+            continue;
         }
 #if OPENDHT_PROXY_SERVER
         else if (op == "pst") {