build: use msgpack cmake

CMakeLists.txt

@@ -57,7 +57,7 @@ if (NOT MSVC)
     find_package (PkgConfig REQUIRED)
     find_package (GnuTLS 3.3 REQUIRED)
     pkg_search_module (Nettle REQUIRED nettle)
-    find_package (Msgpack 1.2 REQUIRED)
+    find_package (msgpack REQUIRED)
     if (OPENDHT_TOOLS)
         find_package (Readline 6 REQUIRED)
     endif ()
@@ -329,7 +329,7 @@ if (OPENDHT_STATIC)
             PRIVATE  ${argon2_LIBRARIES}
             PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_STATIC_LIBRARIES}
                    ${Jsoncpp_STATIC_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
-                   ${OPENSSL_STATIC_LIBRARIES})
+                   ${OPENSSL_STATIC_LIBRARIES} msgpackc-cxx)
     else ()
         if (OPENDHT_TOOLS)
             function (add_obj_lib name libfile)
@@ -392,14 +392,14 @@ if (OPENDHT_SHARED)
     target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
     if (APPLE)
         target_link_libraries(opendht
-            PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${OPENSSL_LIBRARIES}
+            PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${OPENSSL_LIBRARIES} msgpackc-cxx
             PRIVATE ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
                     ${Jsoncpp_LIBRARIES}
                     ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY} ${argon2_LIBRARIES}
             SYSTEM "-framework CoreFoundation" "-framework Security")
     else ()
         target_link_libraries(opendht
-            PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${OPENSSL_LIBRARIES}
+            PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${OPENSSL_LIBRARIES} msgpackc-cxx
             PRIVATE ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
                     ${Jsoncpp_LIBRARIES}
                     ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY} ${argon2_LIBRARIES})

cmake/FindMsgpack.cmake

@@ -1,64 +0,0 @@
-# - Try to find msgpack
-# Once done this will define
-#  MSGPACK_FOUND - System has msgpack
-#  MSGPACK_INCLUDE_DIRS - The msgpack include directories
-#  MSGPACK_LIBRARIES - The libraries needed to use msgpack
-
-if(NOT MSGPACK_USE_BUNDLED)
-  find_package(PkgConfig)
-  if (PKG_CONFIG_FOUND)
-    pkg_search_module(PC_MSGPACK QUIET
-      msgpackc>=${Msgpack_FIND_VERSION}
-      msgpack>=${Msgpack_FIND_VERSION})
-  endif()
-else()
-  set(PC_MSGPACK_INCLUDEDIR)
-  set(PC_MSGPACK_INCLUDE_DIRS)
-  set(PC_MSGPACK_LIBDIR)
-  set(PC_MSGPACK_LIBRARY_DIRS)
-  set(LIMIT_SEARCH NO_DEFAULT_PATH)
-endif()
-
-set(MSGPACK_DEFINITIONS ${PC_MSGPACK_CFLAGS_OTHER})
-
-find_path(MSGPACK_INCLUDE_DIR msgpack/version_master.h
-  HINTS ${PC_MSGPACK_INCLUDEDIR} ${PC_MSGPACK_INCLUDE_DIRS}
-  ${LIMIT_SEARCH})
-
-if(MSGPACK_INCLUDE_DIR)
-  file(READ ${MSGPACK_INCLUDE_DIR}/msgpack/version_master.h msgpack_version_h)
-  string(REGEX REPLACE ".*MSGPACK_VERSION_MAJOR +([0-9]+).*" "\\1" MSGPACK_VERSION_MAJOR "${msgpack_version_h}")
-  string(REGEX REPLACE ".*MSGPACK_VERSION_MINOR +([0-9]+).*" "\\1" MSGPACK_VERSION_MINOR "${msgpack_version_h}")
-  string(REGEX REPLACE ".*MSGPACK_VERSION_REVISION +([0-9]+).*" "\\1" MSGPACK_VERSION_REVISION "${msgpack_version_h}")
-  set(MSGPACK_VERSION_STRING "${MSGPACK_VERSION_MAJOR}.${MSGPACK_VERSION_MINOR}.${MSGPACK_VERSION_REVISION}")
-else()
-  set(MSGPACK_VERSION_STRING)
-endif()
-
-# If we're asked to use static linkage, add libmsgpack{,c}.a as a preferred library name.
-if(MSGPACK_USE_STATIC)
-  list(APPEND MSGPACK_NAMES
-    "${CMAKE_STATIC_LIBRARY_PREFIX}msgpackc${CMAKE_STATIC_LIBRARY_SUFFIX}"
-    "${CMAKE_STATIC_LIBRARY_PREFIX}msgpack${CMAKE_STATIC_LIBRARY_SUFFIX}")
-endif()
-
-list(APPEND MSGPACK_NAMES msgpackc msgpack)
-
-find_library(MSGPACK_LIBRARY NAMES ${MSGPACK_NAMES}
-  # Check each directory for all names to avoid using headers/libraries from
-  # different places.
-  NAMES_PER_DIR
-  HINTS ${PC_MSGPACK_LIBDIR} ${PC_MSGPACK_LIBRARY_DIRS}
-  ${LIMIT_SEARCH})
-
-mark_as_advanced(MSGPACK_INCLUDE_DIR MSGPACK_LIBRARY)
-
-set(MSGPACK_LIBRARIES ${MSGPACK_LIBRARY})
-set(MSGPACK_INCLUDE_DIRS ${MSGPACK_INCLUDE_DIR})
-
-include(FindPackageHandleStandardArgs)
-# handle the QUIETLY and REQUIRED arguments and set MSGPACK_FOUND to TRUE
-# if all listed variables are TRUE
-find_package_handle_standard_args(Msgpack
-  REQUIRED_VARS MSGPACK_LIBRARY MSGPACK_INCLUDE_DIR
-  VERSION_VAR MSGPACK_VERSION_STRING)

include/opendht/dht.h

@@ -573,7 +573,8 @@ private:
      *
      * @param sr  The search to execute its operations.
      */
-    void searchStep(Sp<Search>);
+    void searchStep(std::weak_ptr<Search> ws);
+
     void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
 
     void dumpSearch(const Search& sr, std::ostream& out) const;

src/dht.cpp

@@ -288,7 +288,7 @@ Dht::searchNodeGetDone(const net::Request& req,
             if (srn->syncJob)
                 scheduler.edit(srn->syncJob, syncTime);
             else
-                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, sr));
+                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, ws));
         }
         onGetValuesDone(req.node, answer, sr, query);
     }
@@ -490,13 +490,18 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
             } catch (std::out_of_range&) { }
 
             auto next_refresh_time = now + getType(a.value->type).expiration;
+            auto& acked = sn->acked[a.value->id];
+            if (acked.refresh) {
+                acked.refresh->cancel();
+                acked.refresh.reset();
+            }
             /* only put the value if the node doesn't already have it */
             if (not hasValue or seq_no < a.value->seq) {
                 if (logger_)
                     logger_->d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto created = a.permanent ? time_point::max() : a.created;
-                sn->acked[a.value->id] = {
+                acked = {
                     network_engine.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
                     next_refresh_time
                 };
@@ -504,7 +509,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                 if (logger_)
                     logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = {
+                acked = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
                     [this, ws, node=sn->node, v=a.value,
                      onDone,
@@ -536,17 +541,13 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
                 ack_req->reply_time = now;
-                sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
+                acked = {std::move(ack_req), next_refresh_time};
 
                 /* step to clear announces */
                 scheduler.edit(sr->nextSearchStep, now);
             }
             if (a.permanent) {
-                scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, [this,ws] {
-                    if (auto sr = ws.lock()) {
-                        searchStep(sr);
-                    }
-                });
+                acked.refresh = scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, std::bind(&Dht::searchStep, this, ws));
             }
         }
     };
@@ -644,8 +645,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 if (auto sr = ws.lock()) {
                     scheduler.edit(sr->nextSearchStep, scheduler.time());
                     if (auto sn = sr->getNode(req.node)) {
-                        scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, sr));
-                        sn->onListenSynced(query);
+                        auto job = scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, ws));
+                        sn->onListenSynced(query, true, std::move(job));
                     }
                     onListenDone(req.node, answer, sr);
                 }
@@ -671,8 +672,9 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
 /* When a search is in progress, we periodically call search_step to send
    further requests. */
 void
-Dht::searchStep(Sp<Search> sr)
+Dht::searchStep(std::weak_ptr<Search> ws)
 {
+    auto sr = ws.lock();
     if (not sr or sr->expired or sr->done) return;
 
     const auto& now = scheduler.time();
@@ -825,7 +827,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
         sr->expired = false;
         sr->nodes.clear();
         sr->nodes.reserve(SEARCH_NODES+1);
-        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, sr));
+        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
         if (logger_)
             logger_->w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         if (search_id == 0)
@@ -1278,8 +1280,10 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
     if (auto vs = store.first) {
         total_store_size += store.second.size_diff;
         total_values += store.second.values_diff;
+        if (vs->expiration_job)
+            vs->expiration_job->cancel();
         if (not permanent) {
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+            vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
         }
         if (total_store_size > max_store_size) {
             expireStore();
@@ -1621,10 +1625,10 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                 out << "[";
                 for (const auto& a : sr.announce) {
                     auto ack = n.acked.find(a.value->id);
-                    if (ack == n.acked.end() or not ack->second.first) {
+                    if (ack == n.acked.end() or not ack->second.req) {
                         out << ' ';
                     } else {
-                        out << ack->second.first->getStateChar();
+                        out << ack->second.req->getStateChar();
                     }
                 }
                 out << "] ";
@@ -2193,25 +2197,25 @@ Dht::exportNodes() const
     if (b4 != dht4.buckets.end()) {
         for (auto& n : b4->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     const auto b6 = dht6.buckets.findBucket(myid);
     if (b6 != dht6.buckets.end()) {
         for (auto& n : b6->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     for (auto b = dht4.buckets.begin(); b != dht4.buckets.end(); ++b) {
         if (b == b4) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     for (auto b = dht6.buckets.begin(); b != dht6.buckets.end(); ++b) {
         if (b == b6) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     return nodes;
 }
@@ -2484,7 +2488,7 @@ Dht::onAnnounce(Sp<Node> n,
             if (*lv == *vc) {
                 storageRefresh(hash, v->id);
                 if (logger_)
-                    logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node.toString().c_str(), std::to_string(v->id).c_str());
+                    logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
             } else {
                 const auto& type = getType(lv->type);
                 if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2527,7 +2531,7 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
     }
     if (storageRefresh(hash, vid)) {
         if (logger_)
-            logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node->toString().c_str(), std::to_string(vid).c_str());
+            logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
     } else {
         if (logger_)
             logger_->d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2563,8 +2567,15 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
         }
 
         auto expiration = s->second.refresh(now, vid, types);
-        if (expiration != time_point::max())
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        if (expiration.first) {
+            if (expiration.first->expiration_job) {
+                expiration.first->expiration_job->cancel();
+                expiration.first->expiration_job.reset();
+            }
+            if (expiration.second != time_point::max()) {
+                expiration.first->expiration_job = scheduler.add(expiration.second, std::bind(&Dht::expireStorage, this, id));
+            }
+        }
         return true;
     }
     return false;

src/search.h

@@ -49,12 +49,21 @@ struct Dht::Announce {
 };
 
 struct Dht::SearchNode {
+
+    struct AnnounceStatus {
+        Sp<net::Request> req {};
+        Sp<Scheduler::Job> refresh {};
+        time_point refresh_time;
+        AnnounceStatus(){};
+        AnnounceStatus(Sp<net::Request> r, time_point t): req(std::move(r)), refresh_time(t)
+         {}
+    };
     /**
      * Foreach value id, we keep track of a pair (net::Request, time_point) where the
      * request is the request returned by the network engine and the time_point
      * is the next time at which the value must be refreshed.
      */
-    using AnnounceStatus = std::map<Value::Id, std::pair<Sp<net::Request>, time_point>>;
+    using AnnounceStatusMap = std::map<Value::Id, AnnounceStatus>;
     /**
      * Foreach Query, we keep track of the request returned by the network
      * engine when we sent the "get".
@@ -63,6 +72,7 @@ struct Dht::SearchNode {
 
     struct CachedListenStatus {
         ValueCache cache;
+        Sp<Scheduler::Job> refresh {};
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
         Tid socketId {0};
@@ -87,7 +97,7 @@ struct Dht::SearchNode {
 
     SyncStatus getStatus {};    /* get/sync status */
     NodeListenerStatus listenStatus {}; /* listen status */
-    AnnounceStatus acked {};    /* announcement status for a given value id */
+    AnnounceStatusMap acked {};    /* announcement status for a given value id */
 
     Blob token {};                                 /* last token the node sent to us after a get request */
     time_point last_get_reply {time_point::min()}; /* last time received valid token */
@@ -239,9 +249,12 @@ struct Dht::SearchNode {
         }
     }
 
-    void onListenSynced(const Sp<Query>& q, bool synced = true) {
+    void onListenSynced(const Sp<Query>& q, bool synced = true, Sp<Scheduler::Job> refreshJob = {}) {
         auto l = listenStatus.find(q);
         if (l != listenStatus.end()) {
+            if (l->second.refresh)
+                l->second.refresh->cancel();
+            l->second.refresh = std::move(refreshJob);
             l->second.cache.onSynced(synced);
         }
     }
@@ -292,16 +305,18 @@ struct Dht::SearchNode {
 
     bool isAnnounced(Value::Id vid) const {
         auto ack = acked.find(vid);
-        if (ack == acked.end() or not ack->second.first)
+        if (ack == acked.end() or not ack->second.req)
             return false;
-        return ack->second.first->completed();
+        return ack->second.req->completed();
     }
     void cancelAnnounce() {
         for (const auto& status : acked) {
-            const auto& req = status.second.first;
+            const auto& req = status.second.req;
             if (req and req->pending()) {
                 node->cancelRequest(req);
             }
+            if (status.second.refresh)
+                status.second.refresh->cancel();
         }
         acked.clear();
     }
@@ -328,14 +343,21 @@ struct Dht::SearchNode {
         return listen_status->second.req->reply_time + listen_expire > now;
     }
     void cancelListen() {
-        for (const auto& status : listenStatus)
+        for (const auto& status : listenStatus) {
             node->cancelRequest(status.second.req);
-        listenStatus.clear();
+            if (status.second.refresh)
+                status.second.refresh->cancel();
+            if (status.second.cacheExpirationJob)
+                status.second.cacheExpirationJob->cancel();
+        }
+        listenStatus.clear(); 
     }
     void cancelListen(const Sp<Query>& query) {
         auto it = listenStatus.find(query);
         if (it != listenStatus.end()) {
             node->cancelRequest(it->second.req);
+            if (it->second.refresh)
+                it->second.refresh->cancel();
             listenStatus.erase(it);
         }
     }
@@ -345,13 +367,13 @@ struct Dht::SearchNode {
      */
     time_point getAnnounceTime(Value::Id vid) const {
         const auto& ack = acked.find(vid);
-        if (ack == acked.cend() or not ack->second.first) {
+        if (ack == acked.cend() or not ack->second.req) {
             return time_point::min();
         }
-        if (ack->second.first->completed()) {
-            return ack->second.second - REANNOUNCE_MARGIN;
+        if (ack->second.req->completed()) {
+            return ack->second.refresh_time - REANNOUNCE_MARGIN;
         }
-        return ack->second.first->pending() ? time_point::max() : time_point::min();
+        return ack->second.req->pending() ? time_point::max() : time_point::min();
     }
 
     /**
@@ -541,17 +563,15 @@ struct Dht::Search {
         if (not opExpirationJob)
             opExpirationJob = scheduler.add(time_point::max(), [this,&scheduler]{
                 auto nextExpire = cache.expire(scheduler.time(), [&](size_t t){
-                    Sp<Query> query;
                     const auto& ll = listeners.find(t);
                     if (ll != listeners.cend()) {
-                        query = ll->second.query;
+                        auto query = ll->second.query;
                         listeners.erase(ll);
-                    }
-                    for (auto& sn : nodes) {
-                        if (listeners.empty())
-                            sn->cancelListen();
-                        else if (query)
-                            sn->cancelListen(query);
+                        if (listeners.empty()) {
+                            for (auto& sn : nodes) sn->cancelListen();
+                        } else if (query) {
+                            for (auto& sn : nodes) sn->cancelListen(query);
+                        }
                     }
                 });
                 scheduler.edit(opExpirationJob, nextExpire);
@@ -588,8 +608,10 @@ struct Dht::Search {
         for (auto& n : nodes) {
             auto ackIt = n->acked.find(vid);
             if (ackIt != n->acked.end()) {
-                if (ackIt->second.first)
-                    ackIt->second.first->cancel();
+                if (ackIt->second.req)
+                    ackIt->second.req->cancel();
+                if (ackIt->second.refresh)
+                    ackIt->second.refresh->cancel();
                 n->acked.erase(ackIt);
             }
         }
@@ -606,7 +628,7 @@ struct Dht::Search {
             announce.emplace_back(Announce {permanent, value, created, callback});
             for (auto& n : nodes) {
                 n->probe_query.reset();
-                n->acked[value->id].first.reset();
+                n->acked[value->id].req.reset();
             }
         } else {
             a_sr->permanent = permanent;
@@ -614,7 +636,7 @@ struct Dht::Search {
             if (a_sr->value != value) {
                 a_sr->value = value;
                 for (auto& n : nodes) {
-                    n->acked[value->id].first.reset();
+                    n->acked[value->id].req.reset();
                     n->probe_query.reset();
                 }
             }
@@ -738,8 +760,8 @@ struct Dht::Search {
             for (auto& n : nodes) {
                 auto ackIt = n->acked.find(it->value->id);
                 if (ackIt != n->acked.end()) {
-                    if (ackIt->second.first)
-                        ackIt->second.first->cancel();
+                    if (ackIt->second.req)
+                        ackIt->second.req->cancel();
                     n->acked.erase(ackIt);
                 }
             }

src/storage.h

@@ -58,6 +58,7 @@ struct ValueStorage {
     Sp<Value> data {};
     time_point created {};
     time_point expiration {};
+    Sp<Scheduler::Job> expiration_job {};
     StorageBucket* store_bucket {nullptr};
 
     ValueStorage() {}
@@ -142,14 +143,15 @@ struct Storage {
      * @param vid  The value id
      * @return time of the next expiration, time_point::max() if no expiration
      */
-    time_point refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
+    std::pair<ValueStorage*, time_point>
+    refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
         for (auto& vs : values)
             if (vs.data->id == vid) {
                 vs.created = now;
                 vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
-                return vs.expiration;
+                return {&vs, vs.expiration};
             }
-        return time_point::max();
+        return {nullptr, time_point::max()};
     }
 
     size_t listen(ValueCallback& cb, Value::Filter& f, const Sp<Query>& q);
@@ -197,9 +199,9 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
     if (it != values.end()) {
         /* Already there, only need to refresh */
         it->created = created;
-        size_t size_old = it->data->size();
-        ssize_t size_diff = size_new - (ssize_t)size_old;
         if (it->data != value) {
+            size_t size_old = it->data->size();
+            ssize_t size_diff = size_new - (ssize_t)size_old;
             //DHT_LOG.DEBUG("Updating %s -> %s", id.toString().c_str(), value->toString().c_str());
             // clear quota for previous value
             if (it->store_bucket)
@@ -213,7 +215,6 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
             total_size += size_diff;
             return std::make_pair(&(*it), StoreDiff{size_diff, 0, 0});
         }
-        return std::make_pair(nullptr, StoreDiff{});
     } else {
         //DHT_LOG.DEBUG("Storing %s -> %s", id.toString().c_str(), value->toString().c_str());
         if (values.size() < MAX_VALUES) {
@@ -224,8 +225,8 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
                 sb->insert(id, *value, expiration);
             return std::make_pair(&values.back(), StoreDiff{size_new, 1, 0});
         }
-        return std::make_pair(nullptr, StoreDiff{});
     }
+    return std::make_pair(nullptr, StoreDiff{});
 }
 
 Storage::StoreDiff
@@ -239,6 +240,8 @@ Storage::remove(const InfoHash& id, Value::Id vid)
     ssize_t size = it->data->size();
     if (it->store_bucket)
         it->store_bucket->erase(id, *it->data, it->expiration);
+    if (it->expiration_job)
+        it->expiration_job->cancel();
     total_size -= size;
     values.erase(it);
     return {-size, -1, 0};
@@ -287,6 +290,8 @@ Storage::expire(const InfoHash& id, time_point now)
         size_diff -= v.data->size();
         if (v.store_bucket)
             v.store_bucket->erase(id, *v.data, v.expiration);
+        if (v.expiration_job)
+            v.expiration_job->cancel();
         ret.emplace_back(std::move(v.data));
     });
     total_size += size_diff;