sockaddr: use more precise exception

securedht: move crypto ops to Value
proxy client: fix race condition
2026-11-08 20:37:23 +00:00 · 2018-12-17 16:41:00 -05:00 · 2018-12-17 16:41:00 -05:00 · 2018-12-17 16:41:00 -05:00 · 2018-12-11 14:52:07 -05:00 · 2018-12-11 14:48:26 -05:00
38 changed files with 853 additions and 478 deletions
@@ -1,3 +1,4 @@
+dist: xenial
 sudo: required

 services:
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.2)
+set (opendht_VERSION_MINOR 8.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -59,7 +59,7 @@ endif()

 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
    find_package(Restbed REQUIRED)
-    if (!Jsoncpp_FOUND)
+    if (NOT Jsoncpp_FOUND)
        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
    endif()
 endif()
@@ -99,8 +99,8 @@ endif ()
 if (Nettle_INCLUDE_DIRS)
    include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
 endif ()
-if (RESTBED_INCLUDE)
-    include_directories (SYSTEM "${RESTBED_INCLUDE}")
+if (Restbed_INCLUDE_DIR)
+    include_directories (SYSTEM "${Restbed_INCLUDE_DIR}")
 endif ()
 if (Jsoncpp_INCLUDE_DIRS)
    include_directories (SYSTEM "${Jsoncpp_INCLUDE_DIRS}")
@@ -246,16 +246,14 @@ if (OPENDHT_STATIC)
        ${opendht_HEADERS}
    )
    set_target_properties (opendht-static PROPERTIES OUTPUT_NAME "opendht")
-    if (OPENDHT_LTO)
-        target_link_libraries(opendht-static -flto -fuse-linker-plugin)
-    endif ()
    if (OPENDHT_ARGON2)
        target_include_directories(opendht-static SYSTEM PRIVATE argon2)
    else ()
-        target_link_libraries(opendht-static ${argon2_LIBRARIES})
        target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht-static
+        PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
+        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Jsoncpp_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -267,16 +265,13 @@ if (OPENDHT_SHARED)
    set_target_properties (opendht PROPERTIES IMPORT_SUFFIX "_import.lib")
    set_target_properties (opendht PROPERTIES SOVERSION ${opendht_VERSION_MAJOR} VERSION ${opendht_VERSION})
    target_compile_definitions(opendht PRIVATE OPENDHT_BUILD)
-    if (OPENDHT_LTO)
-        target_link_libraries(opendht PRIVATE -flto -fuse-linker-plugin)
-    endif ()
    if (OPENDHT_ARGON2)
        target_include_directories(opendht SYSTEM PRIVATE argon2)
    else ()
        target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
        target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES})
    install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -146,7 +146,7 @@
      <Optimization>Disabled</Optimization>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <SuppressStartupBanner>false</SuppressStartupBanner>
@@ -169,7 +169,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <Optimization>Disabled</Optimization>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
    </ClCompile>
@@ -193,7 +193,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <IntrinsicFunctions>true</IntrinsicFunctions>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <ProgramDataBaseFileName>$(OutDir)\lib\x86\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -223,7 +223,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <IntrinsicFunctions>true</IntrinsicFunctions>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>$(ProjectDir)contrib\build\include;$(ProjectDir)..\include;$(ProjectDir)..\include\opendht;$(ProjectDir)contrib\build\msgpack-c\include;$(ProjectDir)contrib\build\argon2\include;$(ProjectDir)..\..\include;$(ProjectDir)..\..\argon2\include</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;4273;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <ProgramDataBaseFileName>$(OutDir)\lib\x64\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -16,7 +16,9 @@ if ENABLE_TESTS
 SUBDIRS += tests
 endif

+if HAVE_DOXYGEN
 SUBDIRS += doc
+endif

 ACLOCAL_AMFLAGS = -I m4

@@ -1,17 +1,16 @@
-find_path (RESTBED_INCLUDE restbed
-           HINTS
-           "/usr/include"
-           "/usr/local/include"
-           "/opt/local/include"
-)
-
-if (RESTBED_INCLUDE)
-    message(STATUS "${green}Found Restbed: ${RESTBED_INCLUDE}")
-else()
-    message(FATAL_ERROR "${red}Failed to locate Restbed.}")
-endif()
-
-if (RESTBED_INCLUDE)
-    set(RESTBED_FOUND TRUE)
-    set(Restbed_LIBRARIES restbed)
+if(NOT Restbed_FOUND)
+    find_path (Restbed_INCLUDE_DIR restbed
+               HINTS
+               "/usr/include"
+               "/usr/local/include"
+               "/opt/local/include")
+    find_library(Restbed_LIBRARY restbed
+                 HINTS ${Restbed_ROOT_DIR} PATH_SUFFIXES lib)
+    include(FindPackageHandleStandardArgs)
+    find_package_handle_standard_args(Restbed DEFAULT_MSG Restbed_LIBRARY Restbed_INCLUDE_DIR)
+    if (Restbed_INCLUDE_DIR)
+        set(Restbed_FOUND TRUE)
+        set(Restbed_LIBRARIES ${Restbed_LIBRARY})
+        set(Restbed_INCLUDE_DIRS ${Restbed_INCLUDE_DIR})
+    endif()
 endif()
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 2)
+m4_define([opendht_minor_version], 8)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -130,7 +130,7 @@ PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
 PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])

 AC_ARG_WITH([jsoncpp], AS_HELP_STRING([--without-jsoncpp], [Build without JsonCpp support]))
-AS_IF([test "x$with_jsoncpp" != "xno"], 
+AS_IF([test "x$with_jsoncpp" != "xno"],
      [PKG_CHECK_MODULES([JsonCpp], [jsoncpp >= 1.7.2], [have_jsoncpp=yes], [have_jsoncpp=no])],
      [have_jsoncpp=no])
 AS_IF([test "x$have_jsoncpp" = "xyes"], [
@@ -193,7 +193,9 @@ AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY,
    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"],
    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"])

-AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
+AM_COND_IF([HAVE_DOXYGEN], [
+  AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
+])

 dnl Configure setup.py if we build the python module
 AM_COND_IF([USE_CYTHON], [
@@ -51,6 +51,7 @@ struct OPENDHT_PUBLIC NodeStats {
             incoming_nodes {0};
    unsigned table_depth {0};
    unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
+    unsigned long getNetworkSizeEstimation() const { return 8 * std::exp2(table_depth); }
    std::string toString() const;

 #ifdef OPENDHT_JSONCPP
@@ -86,8 +86,16 @@ struct OPENDHT_PUBLIC PublicKey

    PublicKey& operator=(PublicKey&& o) noexcept;

+    /**
+     * Get public key fingerprint
+     */
    InfoHash getId() const;
+
+    /**
+     * Get public key long fingerprint
+     */
    PkId getLongId() const;
+
    bool checkSignature(const Blob& data, const Blob& signature) const;
    Blob encrypt(const Blob&) const;

@@ -133,6 +141,7 @@ struct OPENDHT_PUBLIC PrivateKey
    PrivateKey(const Blob& import, const std::string& password = {});
    ~PrivateKey();
    explicit operator bool() const { return key; }
+
    PublicKey getPublicKey() const;
    Blob serialize(const std::string& password = {}) const;

@@ -164,8 +173,6 @@ private:
    PrivateKey(const PrivateKey&) = delete;
    PrivateKey& operator=(const PrivateKey&) = delete;
    Blob decryptBloc(const uint8_t* src, size_t src_size) const;
-
-    //friend dht::crypto::Identity dht::crypto::generateIdentity(const std::string&, dht::crypto::Identity, unsigned key_length);
 };


@@ -368,6 +375,7 @@ struct OPENDHT_PUBLIC Certificate {

    /** Same as getPublicKey().getId() */
    InfoHash getId() const;
+    /** Same as getPublicKey().getLongId() */
    PkId getLongId() const;

    /** Read certificate Common Name (CN) */
@@ -404,8 +412,20 @@ struct OPENDHT_PUBLIC Certificate {

    std::string print() const;

+    /**
+     * As a CA, revoke a certificate, adding it to
+     * the attached Certificate Revocation List (CRL)
+     */
    void revoke(const PrivateKey&, const Certificate&);
+
+    /**
+     * Get the list of certificates revoked as as CA.
+     */
    std::vector<std::shared_ptr<RevocationList>> getRevocationLists() const;
+
+    /**
+     * Attach existing revocation list.
+     */
    void addRevocationList(RevocationList&&);
    void addRevocationList(std::shared_ptr<RevocationList>);

@@ -508,9 +528,13 @@ public:
    ~secure_vector() { clean(); }

    static secure_vector<T> getRandom(size_t size) {
-        secure_vector<T> ret(size/sizeof(T));
+        secure_vector<T> ret(size);
        crypto::random_device rdev;
+#ifdef _WIN32
+        std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint8_t>::max() };
+#else
        std::uniform_int_distribution<uint8_t> rand_byte;
+#endif
        std::generate_n((uint8_t*)ret.data_.data(), ret.size()*sizeof(T), std::bind(rand_byte, std::ref(rdev)));
        return ret;
    }
@@ -409,7 +409,9 @@ private:

    // Storage
    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
-    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {});
+    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
+    bool storageErase(const InfoHash& id, Value::Id vid);
+    bool storageRefresh(const InfoHash& id, Value::Id vid);
    void expireStore();
    void expireStorage(InfoHash h);
    void expireStore(decltype(store)::iterator);
@@ -464,7 +466,7 @@ private:
     * Low-level method that will perform a search on the DHT for the specified
     * infohash (id), using the specified IP version (IPv4 or IPv6).
     */
-    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, Query q = {});
+    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});

    void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});
@@ -40,7 +40,7 @@ public:
    virtual NodeStatus getStatus() const = 0;

    /**
-     * Get the ID of the node.
+     * Get the ID of the DHT node.
     */
    virtual const InfoHash& getNodeId() const = 0;

@@ -216,7 +216,7 @@ public:
     */
    virtual void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG)
    {
-        DHT_LOG.DEBUG = debug;
+        DHT_LOG.DBG = debug;
        DHT_LOG.WARN = warn;
        DHT_LOG.ERR = error;
    }
@@ -375,7 +375,7 @@ private:
    void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS

-    bool isDestroying_ {false};
+    std::atomic_bool isDestroying_ {false};
 };

 }
@@ -34,6 +34,10 @@
 #include <mutex>
 #include <restbed>

+#ifdef OPENDHT_JSONCPP
+#include <json/json.h>
+#endif
+
 namespace Json {
    class Value;
 }
@@ -73,15 +77,44 @@ public:
        size_t pushListenersCount;
        /** Average requests per second */
        double requestRate;
+        /** Node Info **/
+        NodeInfo nodeInfo;

        std::string toString() const {
            std::ostringstream ss;
            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
            ss << "Requests: " << requestRate << " per second." << std::endl;
+            auto& ni = nodeInfo;
+            auto& ipv4 = ni.ipv4;
+            if (ipv4.table_depth > 1) {
+                ss << "IPv4 Network estimation: " << ipv4.getNetworkSizeEstimation() << std::endl;;
+            }
+            auto& ipv6 = ni.ipv6;
+            if (ipv6.table_depth > 1) {
+                ss << "IPv6 Network estimation: " << ipv6.getNetworkSizeEstimation() << std::endl;;
+            }
            return ss.str();
        }
+
+#ifdef OPENDHT_JSONCPP
+        /**
+         * Build a json object from a NodeStats
+         */
+        Json::Value toJson() const {
+            Json::Value result;
+            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
+            result["putCount"] = static_cast<Json::UInt64>(putCount);
+            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
+            result["requestRate"] = requestRate;
+            result["nodeInfo"] = nodeInfo.toJson();
+            return result;
+        }
+#endif
    };
-    ServerStats getStats() const;
+
+    ServerStats stats() const { return stats_; }
+
+    void updateStats() const;

    std::shared_ptr<DhtRunner> getNode() const { return dht_; }

@@ -100,6 +133,14 @@ private:
     */
    void getNodeInfo(const std::shared_ptr<restbed::Session>& session) const;

+    /**
+     * Return ServerStats in JSON format
+     * Method: STATS "/"
+     * Result: HTTP 200, body: Node infos in JSON format
+     * @param session
+     */
+    void getStats(const std::shared_ptr<restbed::Session>& session) const;
+
    /**
     * Return Values of an infoHash
     * Method: GET "/{InfoHash: .*}"
@@ -262,6 +303,8 @@ private:

    const std::string pushServer_;

+    mutable ServerStats stats_;
+
 #if OPENDHT_PUSH_NOTIFICATIONS
    struct Listener;
    struct PushListener;
@@ -158,6 +158,15 @@ public:
        },
        getFilterSet<T>());
    }
+    template <class T>
+    std::future<size_t> listen(InfoHash hash, std::function<bool(std::vector<T>&&, bool)> cb)
+    {
+        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+            return cb(unpackVector<T>(vals), expired);
+        },
+        getFilterSet<T>());
+    }
+
    template <typename T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
    {
@@ -174,6 +183,22 @@ public:
        },
        getFilterSet<T>(f), w);
    }
+    template <typename T>
+    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    {
+        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+            for (const auto& v : vals) {
+                try {
+                    if (not cb(Value::unpack<T>(*v), expired))
+                        return false;
+                } catch (const std::exception&) {
+                    continue;
+                }
+            }
+            return true;
+        },
+        getFilterSet<T>(f), w);
+    }

    void cancelListen(InfoHash h, size_t token);
    void cancelListen(InfoHash h, std::shared_future<size_t> token);
@@ -248,8 +273,14 @@ public:

    void dumpTables() const;

+    /**
+     * Get the public key fingerprint if an identity is used with this node, 0 otherwise.
+     */
    InfoHash getId() const;

+    /**
+     * Get the ID of the DHT node.
+     */
    InfoHash getNodeId() const;

    /**
@@ -416,6 +447,7 @@ private:
     */
    void tryBootstrapContinuously();

+    void stopNetwork();
    void startNetwork(const SockAddr sin4, const SockAddr sin6);
    time_point loop_();

@@ -486,6 +518,7 @@ private:
               status6 {NodeStatus::Disconnected};
    StatusCallback statusCb {nullptr};

+    int stop_writefd {-1};
    int s4 {-1}, s6 {-1};
    SockAddr bound4 {};
    SockAddr bound6 {};
@@ -85,7 +85,7 @@ private:
 };

 struct Logger {
-    LogMethod DEBUG = NOLOG;
+    LogMethod DBG = NOLOG;
    LogMethod WARN = NOLOG;
    LogMethod ERR = NOLOG;
    void setFilter(const InfoHash& f) {
@@ -114,7 +114,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log0(DEBUG, format, args);
+        log0(DBG, format, args);
        va_end(args);
 #endif
    }
@@ -122,7 +122,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log1(DEBUG, f, format, args);
+        log1(DBG, f, format, args);
        va_end(args);
 #endif
    }
@@ -130,7 +130,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log2(DEBUG, f1, f2, format, args);
+        log2(DBG, f1, f2, format, args);
        va_end(args);
 #endif
    }
@@ -40,12 +40,16 @@ struct NodeCache {
    ~NodeCache();

 private:
-    class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
+    class NodeMap : private std::map<InfoHash, std::weak_ptr<Node>> {
    public:
        Sp<Node> getNode(const InfoHash& id);
        Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client);
+        std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, size_t count) const;
        void clearBadNodes();
        void setExpired();
+        void cleanup();
+    private:
+        size_t cleanup_counter {0};
    };

    const NodeMap& cache(sa_family_t af) const { return af == AF_INET ? cache_4 : cache_6; }
@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {

-#ifndef WIN32_NATIVE
+#ifndef _MSC_VER
 #ifdef _WIN32

 /**
@@ -71,7 +71,7 @@ public:
        return secureType(std::move(tmp_type));
    }

-    void registerType(const ValueType& type) {
+    void registerType(const ValueType& type) override {
        if (dht_)
            dht_->registerType(secureType(type));
    }
@@ -89,14 +89,14 @@ public:
     * If the signature can't be checked, or if the data can't be decrypted, it is not returned.
     * Public, non-signed & non-encrypted data is retransmitted as-is.
     */
-    void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}, Where&& w = {});
-    void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}, Where&& w = {}) {
+    void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}, Where&& w = {}) override;
+    void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}, Where&& w = {}) override {
        get(id, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) {
+    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) {
+    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) override {
        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

@@ -148,99 +148,100 @@ public:
    /**
     * SecureDht to Dht proxy
     */
-    void shutdown(ShutdownCallback cb) {
+    void shutdown(ShutdownCallback cb) override {
        dht_->shutdown(cb);
    }
-    void dumpTables() const {
+    void dumpTables() const override {
        dht_->dumpTables();
    }
-    inline const InfoHash& getNodeId() const { return dht_->getNodeId(); }
-    std::pair<size_t, size_t> getStoreSize() const {
+    inline const InfoHash& getNodeId() const override { return dht_->getNodeId(); }
+
+    std::pair<size_t, size_t> getStoreSize() const override {
        return dht_->getStoreSize();
    }
-    std::string getStorageLog() const {
+    std::string getStorageLog() const override {
        return dht_->getStorageLog();
    }
-    std::string getStorageLog(const InfoHash& h) const {
+    std::string getStorageLog(const InfoHash& h) const override {
        return dht_->getStorageLog(h);
    }
-    void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) {
+    void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
        dht_->setStorageLimit(limit);
    }
-    std::vector<NodeExport> exportNodes() {
+    std::vector<NodeExport> exportNodes() override {
        return dht_->exportNodes();
    }
-    std::vector<ValuesExport> exportValues() const {
+    std::vector<ValuesExport> exportValues() const override {
        return dht_->exportValues();
    }
-    void importValues(const std::vector<ValuesExport>& v) {
+    void importValues(const std::vector<ValuesExport>& v) override {
        dht_->importValues(v);
    }
-    NodeStats getNodesStats(sa_family_t af) const {
+    NodeStats getNodesStats(sa_family_t af) const override {
        return dht_->getNodesStats(af);
    }
-    std::vector<unsigned> getNodeMessageStats(bool in = false) {
+    std::vector<unsigned> getNodeMessageStats(bool in = false) override {
        return dht_->getNodeMessageStats(in);
    }
-    std::string getRoutingTablesLog(sa_family_t af) const {
+    std::string getRoutingTablesLog(sa_family_t af) const override {
        return dht_->getRoutingTablesLog(af);
    }
-    std::string getSearchesLog(sa_family_t af) const {
+    std::string getSearchesLog(sa_family_t af) const override {
        return dht_->getSearchesLog(af);
    }
-    std::string getSearchLog(const InfoHash& h, sa_family_t af = AF_UNSPEC) const {
+    std::string getSearchLog(const InfoHash& h, sa_family_t af = AF_UNSPEC) const override {
        return dht_->getSearchLog(h, af);
    }
-    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) {
+    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) override {
        return dht_->getPublicAddress(family);
    }
-    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr& sa) {
+    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr& sa) override {
        return dht_->periodic(buf, buflen, sa);
    }
-    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
+    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) override {
        return dht_->periodic(buf, buflen, from, fromlen);
    }
-    NodeStatus getStatus(sa_family_t af) const {
+    NodeStatus getStatus(sa_family_t af) const override {
        return dht_->getStatus(af);
    }
-    NodeStatus getStatus() const {
+    NodeStatus getStatus() const override {
        return dht_->getStatus();
    }
-    bool isRunning(sa_family_t af = 0) const {
+    bool isRunning(sa_family_t af = 0) const override {
        return dht_->isRunning(af);
    }
-    const ValueType& getType(ValueType::Id type_id) const {
+    const ValueType& getType(ValueType::Id type_id) const override {
        return dht_->getType(type_id);
    }
-    void insertNode(const InfoHash& id, const SockAddr& sa) {
+    void insertNode(const InfoHash& id, const SockAddr& sa) override {
        dht_->insertNode(id, sa);
    }
-    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) {
+    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) override {
        dht_->insertNode(id, sa, salen);
    }
-    void insertNode(const NodeExport& n) {
+    void insertNode(const NodeExport& n) override {
        dht_->insertNode(n);
    }
-    void pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb={}) {
+    void pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb={}) override {
        dht_->pingNode(sa, salen, std::move(cb));
    }
-    void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {}) {
+    void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {}) override {
        dht_->query(key, cb, done_cb, std::move(q));
    }
-    void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) {
+    void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
        dht_->query(key, cb, done_cb, std::move(q));
    }
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const {
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const override {
        return dht_->getLocal(key, f);
    }
-    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const {
+    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override {
        return dht_->getLocalById(key, vid);
    }
    void put(const InfoHash& key,
            Sp<Value> v,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, v, cb, created, permanent);
    }
@@ -248,7 +249,7 @@ public:
            const Sp<Value>& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, v, cb, created, permanent);
    }
@@ -257,7 +258,7 @@ public:
            Value&& v,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, std::move(v), cb, created, permanent);
    }
@@ -265,32 +266,32 @@ public:
            Value&& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, std::move(v), cb, created, permanent);
    }
-    std::vector<Sp<Value>> getPut(const InfoHash& h) {
+    std::vector<Sp<Value>> getPut(const InfoHash& h) override {
        return dht_->getPut(h);
    }
-    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) {
+    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) override {
        return dht_->getPut(h, vid);
    }
-    bool cancelPut(const InfoHash& h, const Value::Id& vid) {
+    bool cancelPut(const InfoHash& h, const Value::Id& vid) override {
        return dht_->cancelPut(h, vid);
    }

-    size_t listen(const InfoHash& key, ValueCallback, Value::Filter={}, Where={});
-    size_t listen(const InfoHash& key, GetCallback cb, Value::Filter = {}, Where w = {});
-    size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
+    size_t listen(const InfoHash& key, ValueCallback, Value::Filter={}, Where={}) override;
+    size_t listen(const InfoHash& key, GetCallback cb, Value::Filter = {}, Where w = {}) override;
+    size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) override {
        return listen(key, bindGetCb(cb), f, w);
    }
-    bool cancelListen(const InfoHash& h, size_t token) {
+    bool cancelListen(const InfoHash& h, size_t token) override {
        return dht_->cancelListen(h, token);
    }
-    void connectivityChanged(sa_family_t af) {
+    void connectivityChanged(sa_family_t af) override {
        dht_->connectivityChanged(af);
    }
-    void connectivityChanged() {
+    void connectivityChanged() override {
        dht_->connectivityChanged();
    }

@@ -298,7 +299,7 @@ public:
        forward_all_ = forward;
    }

-    void setPushNotificationToken(const std::string& token = "") {
+    void setPushNotificationToken(const std::string& token = "") override {
        dht_->setPushNotificationToken(token);
    }

@@ -306,22 +307,20 @@ public:
     * Call linked callback with push_notification
     * @param notification to process
     */
-    void pushNotificationReceived(const std::map<std::string, std::string>& notification) {
+    void pushNotificationReceived(const std::map<std::string, std::string>& notification) override {
        dht_->pushNotificationReceived(notification);
    }

-    void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG)
+    void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG) override
    {
-        DHT_LOG.DEBUG = debug;
-        DHT_LOG.WARN = warn;
-        DHT_LOG.ERR = error;
+        DhtInterface::setLoggers(error, warn, debug);
        dht_->setLoggers(error, warn, debug);
    }

    /**
     * Only print logs related to the given InfoHash (if given), or disable filter (if zeroes).
     */
-    void setLogFilter(const InfoHash& f) {
+    void setLogFilter(const InfoHash& f) override {
        DHT_LOG.setFilter(f);
        dht_->setLogFilter(f);
    }
@@ -67,7 +67,7 @@ public:
     */
    SockAddr(const sockaddr* sa, socklen_t length) {
        if (length > sizeof(sockaddr_storage))
-            throw std::runtime_error("Socket address length is too large");
+            throw std::invalid_argument("Socket address length is too large");
        set(sa, length);
    }
    SockAddr(const sockaddr* sa) {
@@ -78,7 +78,7 @@ public:
            else if(sa->sa_family == AF_INET6)
                len = sizeof(sockaddr_in6);
            else
-                throw std::runtime_error("Unknown address family");
+                throw std::invalid_argument("Unknown address family");
        }
        set(sa, len);
    }
@@ -122,6 +122,8 @@ private:
    std::map<ValueType::Id, ValueType> types {};
 };

+struct CryptoValueCache;
+
 /**
 * A "value" is data potentially stored on the Dht, with some metadata.
 *
@@ -575,7 +577,7 @@ struct OPENDHT_PUBLIC Value
    std::string user_type {};

    /**
-     * Sequence number to avoid replay attacks
+     * Sequence number
     */
    uint16_t seq {0};

@@ -588,6 +590,16 @@ struct OPENDHT_PUBLIC Value
     * Hold encrypted version of the data.
     */
    Blob cypher {};
+
+    bool checkSignature();
+    Sp<Value> decrypt(const crypto::PrivateKey& key);
+
+private:
+    /* Cache for crypto ops */
+    bool signatureChecked {false};
+    bool signatureValid {false};
+    bool decrypted {false};
+    Sp<Value> decryptedValue {};
 };

 using ValuesExport = std::pair<InfoHash, Blob>;
@@ -1,6 +1,6 @@
 lib_LTLIBRARIES = libopendht.la

-libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@
+libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
 libopendht_la_LIBADD   = @Argon2_LIBS@ @JsonCpp_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
 libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
 libopendht_la_SOURCES  = \
@@ -65,8 +65,8 @@ libopendht_la_SOURCES += dht_proxy_client.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_client.h ../include/opendht/dht_interface.h
 endif

-if PROXY_CLIENT_OR_SERVER
 libopendht_la_SOURCES += base64.h base64.cpp
+if PROXY_CLIENT_OR_SERVER
 nobase_include_HEADERS += ../include/opendht/proxy.h
 endif

@@ -62,8 +62,7 @@ NodeStats::toString() const
    ss << "Known nodes: " << good_nodes << " good, " << dubious_nodes << " dubious, " << incoming_nodes << " incoming." << std::endl;
    if (table_depth > 1) {
        ss << "Routing table depth: " << table_depth << std::endl;
-        unsigned long tot_nodes = 8 * std::exp2(table_depth);
-        ss << "Network size estimation: " << tot_nodes << " nodes" << std::endl;
+        ss << "Network size estimation: " << getNetworkSizeEstimation() << " nodes" << std::endl;
    }
    return ss.str();
 }
@@ -81,8 +80,7 @@ NodeStats::toJson() const
    val["incoming"] = static_cast<Json::LargestUInt>(incoming_nodes);
    if (table_depth > 1) {
        val["table_depth"] = static_cast<Json::LargestUInt>(table_depth);
-        unsigned long tot_nodes = 8 * std::exp2(table_depth);
-        val["network_size_estimation"] = static_cast<Json::LargestUInt>(tot_nodes);
+        val["network_size_estimation"] = static_cast<Json::LargestUInt>(getNetworkSizeEstimation());
    }
    return val;
 }
@@ -261,11 +261,11 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
    auto select_q = std::make_shared<Query>(Select {}.field(Value::Field::Id), query ? query->where : Where {});
    auto onSelectDone = [this,ws,query](const net::Request& status,
                                        net::RequestAnswer&& answer) mutable {
-        // retreive search
+        // Retrieve search
        auto sr = ws.lock();
        if (not sr) return;
        const auto& id = sr->id;
-        // retreive search node
+        // Retrieve search node
        auto sn = sr->getNode(status.node);
        if (not sn) return;
        // backward compatibility
@@ -319,19 +319,20 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)

    std::weak_ptr<Search> ws = sr;
    auto cb = sr->callbacks.begin();
+    static const auto ANY_QUERY = std::make_shared<Query>(Select {}, Where {}, true);
    do { /* for all requests to send */
        SearchNode* n = nullptr;
-        auto query = not sr->callbacks.empty() ? cb->second.query : std::make_shared<Query>(Select {}, Where {}, true);
-        const time_point up = not sr->callbacks.empty() and update
-                                ? sr->getLastGetTime(query)
+        auto& query = sr->callbacks.empty() ? ANY_QUERY : cb->second.query;
+        const time_point up = (not sr->callbacks.empty() and update)
+                                ? sr->getLastGetTime(*query)
                                : time_point::min();

        if (pn and pn->canGet(now, up, query)) {
            n = pn;
        } else {
            for (auto& sn : sr->nodes) {
-                if (sn->canGet(now, up, query)) {
-                    n = sn.get();
+                if (sn.canGet(now, up, query)) {
+                    n = &sn;
                    break;
                }
            }
@@ -380,104 +381,136 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
    if (sr->announce.empty())
        return;
    unsigned i = 0;
-    auto probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
    std::weak_ptr<Search> ws = sr;
+
+    auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
+    { /* when put done */
+        if (auto sr = ws.lock()) {
+            onAnnounceDone(req.node, answer, sr);
+            searchStep(sr);
+        }
+    };
+
+    auto onExpired = [this,ws](const net::Request&, bool over)
+    { /* when put expired */
+        if (over)
+            if (auto sr = ws.lock())
+                scheduler.edit(sr->nextSearchStep, scheduler.time());
+    };
+
+    auto onSelectDone =
+    [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
+    { /* on probing done */
+        auto sr = ws.lock();
+        if (not sr) return;
+        const auto& now = scheduler.time();
+        sr->insertNode(req.node, scheduler.time(), answer.ntoken);
+        auto sn = sr->getNode(req.node);
+        if (not sn) return;
+
+        if (not sn->isSynced(now)) {
+            /* Search is now unsynced. Let's call searchStep to sync again. */
+            scheduler.edit(sr->nextSearchStep, now);
+            return;
+        }
+        for (auto& a : sr->announce) {
+            if (sn->getAnnounceTime(a.value->id) > now)
+                continue;
+            bool hasValue {false};
+            uint16_t seq_no = 0;
+            try {
+                const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
+                        [&a](const Sp<FieldValueIndex>& i){
+                            return i->index.at(Value::Field::Id).getInt() == a.value->id;
+                        });
+                if (f != answer.fields.cend() and *f) {
+                    hasValue = true;
+                    seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
+                }
+            } catch (std::out_of_range&) { }
+
+            auto next_refresh_time = now + getType(a.value->type).expiration;
+            /* only put the value if the node doesn't already have it */
+            if (not hasValue or seq_no < a.value->seq) {
+                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
+                                                sr->id,
+                                                a.value,
+                                                a.permanent ? time_point::max() : a.created,
+                                                sn->token,
+                                                onDone,
+                                                onExpired), next_refresh_time);
+            } else if (hasValue and a.permanent) {
+                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
+                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
+                                                sr->id,
+                                                a.value->id,
+                                                sn->token,
+                                                onDone,
+                                                onExpired), next_refresh_time);
+            } else {
+                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
+                        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);
+
+                /* 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);
+                    }
+                });
+            }
+        }
+    };
+
+    Sp<Query> probe_query {};
    const auto& now = scheduler.time();
    for (auto& n : sr->nodes) {
-        if (not n->isSynced(now))
-            continue;
-        if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
-            [&now,&n](const Announce& a) {
-                return n->getAnnounceTime(a.value->id) <= now;
-            }) == sr->announce.cend())
+        if (not n.isSynced(now))
            continue;

-        auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
-        { /* when put done */
-            if (auto sr = ws.lock()) {
-                onAnnounceDone(req.node, answer, sr);
-                searchStep(sr);
+        const auto& gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.cend();
+        if (gs != n.getStatus.cend() and gs->second and gs->second->pending()) {
+            continue;
+        }
+
+        bool sendQuery = false;
+        for (auto& a : sr->announce) {
+            if (n.getAnnounceTime(a.value->id) <= now) {
+                if (a.permanent) {
+                    sendQuery = true;
+                } else {
+                    DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                            sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id);
+                    n.acked[a.value->id] = {
+                        network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired),
+                        now + getType(a.value->type).expiration
+                    };
+                }
            }
-        };
-        auto onExpired = [this,ws](const net::Request&, bool over)
-        { /* when put expired */
-            if (over)
-                if (auto sr = ws.lock())
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-        };
-        auto onSelectDone =
-            [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
-            { /* on probing done */
-                auto sr = ws.lock();
-                if (not sr) return;
-                const auto& now = scheduler.time();
-                sr->insertNode(req.node, scheduler.time(), answer.ntoken);
-                auto sn = sr->getNode(req.node);
-                if (not sn) return;
+        }

-                if (not sn->isSynced(now)) {
-                    /* Search is now unsynced. Let's call searchStep to sync again. */
-                    scheduler.edit(sr->nextSearchStep, now);
-                    return;
-                }
-                for (auto& a : sr->announce) {
-                    if (sn->getAnnounceTime(a.value->id) > now)
-                        continue;
-                    bool hasValue {false};
-                    uint16_t seq_no = 0;
-                    try {
-                        for (const auto& field : answer.fields) {
-                            if (field->index.at(Value::Field::Id).getInt() == a.value->id) {
-                                hasValue = true;
-                                seq_no = static_cast<uint16_t>(field->index.at(Value::Field::SeqNum).getInt());
-                                break;
-                            }
-                        }
-                    } catch (std::out_of_range&) { }
-
-                    auto next_refresh_time = now + getType(a.value->type).expiration;
-                    /* only put the value if the node doesn't already have it */
-                    if (not hasValue or seq_no < a.value->seq) {
-                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
-                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                        sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
-                                                        sr->id,
-                                                        a.value,
-                                                        a.permanent ? time_point::max() : a.created,
-                                                        sn->token,
-                                                        onDone,
-                                                        onExpired), next_refresh_time);
-                    } else if (hasValue and a.permanent) {
-                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
-                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                        sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
-                                                        sr->id,
-                                                        a.value->id,
-                                                        sn->token,
-                                                        onDone,
-                                                        onExpired), next_refresh_time);
-                    } else {
-                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
-                                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);
-
-                        /* step to clear announces */
-                        scheduler.edit(sr->nextSearchStep, now);
-                    }
-                }
-            };
-        DHT_LOG.w(sr->id, n->node->id, "[search %s] [node %s] sending %s",
-                sr->id.toString().c_str(), n->node->toString().c_str(), probe_query->toString().c_str());
-        n->probe_query = probe_query;
-        n->getStatus[probe_query] = network_engine.sendGetValues(n->node,
-                sr->id,
-                *probe_query,
-                -1,
-                onSelectDone,
-                std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
-        if (not n->candidate and ++i == TARGET_NODES)
+        if (sendQuery) {
+            if (not probe_query)
+                probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
+            DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
+                    sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
+            n.probe_query = probe_query;
+            n.getStatus[probe_query] = network_engine.sendGetValues(n.node,
+                    sr->id,
+                    *probe_query,
+                    -1,
+                    onSelectDone,
+                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
+        }
+        if (not n.candidate and ++i == TARGET_NODES)
            break;
    }
 }
@@ -588,8 +621,8 @@ Dht::searchStep(Sp<Search> sr)
            // clear corresponding queries
            for (const auto& get : completed_gets)
                for (auto& sn : sr->nodes) {
-                    sn->getStatus.erase(get.query);
-                    sn->pagination_queries.erase(get.query);
+                    sn.getStatus.erase(get.query);
+                    sn.pagination_queries.erase(get.query);
                }

            /* clearing callbacks for announced values */
@@ -602,16 +635,16 @@ Dht::searchStep(Sp<Search> sr)
        // true if this node is part of the target nodes cluter.
        /*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0;

-        DHT_LOG_DEBUG("[search %s IPv%c] synced%s",
+        DHT_LOG_DBG("[search %s IPv%c] synced%s",
                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/

        if (not sr->listeners.empty()) {
            unsigned i = 0;
            for (auto& n : sr->nodes) {
-                if (not n->isSynced(now))
+                if (not n.isSynced(now))
                    continue;
-                searchSynchedNodeListen(sr, *n);
-                if (not n->candidate and ++i == LISTEN_NODES)
+                searchSynchedNodeListen(sr, n);
+                if (not n.candidate and ++i == LISTEN_NODES)
                    break;
            }
        }
@@ -632,7 +665,7 @@ Dht::searchStep(Sp<Search> sr)
                i++;
        }
        while (sent and sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES);
-        /*DHT_LOG_DEBUG("[search %s IPv%c] step: sent %u requests (total %u).",
+        /*DHT_LOG_DBG("[search %s IPv%c] step: sent %u requests (total %u).",
            sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', i, sr->currentlySolicitedNodeCount());*/
    }

@@ -677,7 +710,7 @@ unsigned Dht::refill(Dht::Search& sr) {

 /* Start a search. */
 Sp<Dht::Search>
-Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, Query q)
+Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, const Sp<Query>& q)
 {
    if (!isRunning(af)) {
        DHT_LOG.e(id, "[search %s IPv%c] unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
@@ -719,26 +752,14 @@ 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));
        DHT_LOG.w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
        if (search_id == 0)
            search_id++;
    }

-    if (gcb or qcb) {
-        auto now = scheduler.time();
-        sr->callbacks.insert(std::make_pair<time_point, Get>(
-            std::move(now),
-            Get { scheduler.time(), f, std::make_shared<Query>(q),
-                qcb ? qcb : QueryCallback {}, gcb ? gcb : GetCallback {}, dcb
-            }
-        ));
-    }
-
+    sr->get(f, q, qcb, gcb, dcb, scheduler);
    refill(*sr);
-    if (sr->nextSearchStep)
-        scheduler.edit(sr->nextSearchStep, scheduler.time());
-    else
-        sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr));

    return sr;
 }
@@ -751,15 +772,6 @@ Dht::announce(const InfoHash& id,
        time_point created,
        bool permanent)
 {
-    const auto& now = scheduler.time();
-    if (!value) {
-        if (callback)
-            callback(false, {});
-        return;
-    }
-    created = std::min(now, created);
-    storageStore(id, value, created);
-
    auto& srs = searches(af);
    auto srp = srs.find(id);
    auto sr = srp == srs.end() ? search(id, af) : srp->second;
@@ -776,8 +788,8 @@ Dht::announce(const InfoHash& id,
    if (a_sr == sr->announce.end()) {
        sr->announce.emplace_back(Announce {permanent, value, created, callback});
        for (auto& n : sr->nodes) {
-            n->probe_query.reset();
-            n->acked[value->id].first.reset();
+            n.probe_query.reset();
+            n.acked[value->id].first.reset();
        }
    } else {
        a_sr->permanent = permanent;
@@ -785,8 +797,8 @@ Dht::announce(const InfoHash& id,
        if (a_sr->value != value) {
            a_sr->value = value;
            for (auto& n : sr->nodes) {
-                n->acked[value->id].first.reset();
-                n->probe_query.reset();
+                n.acked[value->id].first.reset();
+                n.probe_query.reset();
            }
        }
        if (sr->isAnnounced(value->id)) {
@@ -911,13 +923,20 @@ struct GetStatus : public OpStatus {
 void
 Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
 {
-    scheduler.syncTime();
-
+    if (not val) {
+        if (callback)
+            callback(false, {});
+        return;
+    }
    if (val->id == Value::INVALID_ID) {
        crypto::random_device rdev;
        std::uniform_int_distribution<Value::Id> rand_id {};
        val->id = rand_id(rdev);
    }
+    scheduler.syncTime();
+    const auto& now = scheduler.time();
+    created = std::min(now, created);
+    storageStore(id, val, created, {}, permanent);

    DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());

@@ -980,10 +999,10 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 {
    scheduler.syncTime();

-    Query q {{}, where};
+    auto q = std::make_shared<Query>(Select {}, std::move(where));
    auto op = std::make_shared<GetStatus<Value>>();
+    auto f = filter.chain(q->where.getFilter());

-    auto f = filter.chain(q.where.getFilter());
    auto add_values = [op,f](const std::vector<Sp<Value>>& values) {
        std::vector<Sp<Value>> newvals {};
        for (const auto& v : values) {
@@ -1049,16 +1068,17 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
    /* Try to answer this search locally. */
    qcb(local_fields);

+    auto sq = std::make_shared<Query>(std::move(q));
    Dht::search(id, AF_INET, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv4");
        op->status4 = {true, ok};
        doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, q);
+    }, f, sq);
    Dht::search(id, AF_INET6, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv6");
        op->status6 = {true, ok};
        doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, q);
+    }, f, sq);
 }

 std::vector<Sp<Value>>
@@ -1121,6 +1141,8 @@ bool
 Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 {
    bool canceled {false};
+    if (storageErase(id, vid))
+        canceled = true;
    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>> srs) {
        auto srp = srs.find(id);
        if (srp == srs.end())
@@ -1189,11 +1211,11 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
 }

 bool
-Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa)
+Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa, bool permanent)
 {
    const auto& now = scheduler.time();
    created = std::min(created, now);
-    auto expiration = created + getType(value->type).expiration;
+    auto expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
    if (expiration < now)
        return false;

@@ -1215,7 +1237,9 @@ 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;
-        scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        if (not permanent) {
+            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        }
        if (total_store_size > max_store_size) {
            expireStore();
        }
@@ -1225,6 +1249,18 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
    return std::get<0>(store);
 }

+bool
+Dht::storageErase(const InfoHash& id, Value::Id vid)
+{
+    auto st = store.find(id);
+    if (st == store.end())
+        return false;
+    auto ret = st->second.remove(id, vid);
+    total_store_size += ret.size_diff;
+    total_values += ret.values_diff;
+    return ret.values_diff;
+}
+
 void
 Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
 {
@@ -1354,9 +1390,9 @@ Dht::connectivityChanged(sa_family_t af)
    network_engine.connectivityChanged(af);
    for (auto& sp : searches(af))
        for (auto& sn : sp.second->nodes) {
-            for (auto& ls : sn->listenStatus)
-                sn->node->cancelRequest(ls.second.req);
-            sn->listenStatus.clear();
+            for (auto& ls : sn.listenStatus)
+                sn.node->cancelRequest(ls.second.req);
+            sn.listenStatus.clear();
        }
    reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
        return addr.second.getFamily() == af;
@@ -1503,42 +1539,42 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
    auto last_get = sr.getLastGetTime();
    for (const auto& n : sr.nodes) {
        i++;
-        out << std::setfill (' ') << std::setw(3) << InfoHash::commonBits(sr.id, n->node->id) << ' ' << n->node->id;
-        out << ' ' << (findNode(n->node->id, sr.af) ? '*' : ' ');
+        out << std::setfill (' ') << std::setw(3) << InfoHash::commonBits(sr.id, n.node->id) << ' ' << n.node->id;
+        out << ' ' << (findNode(n.node->id, sr.af) ? '*' : ' ');
        out << " [";
-        if (auto pendingCount = n->node->getPendingMessageCount())
+        if (auto pendingCount = n.node->getPendingMessageCount())
            out << pendingCount;
        else
            out << ' ';
-        out << (n->node->isExpired() ? 'x' : ' ') << "]";
+        out << (n.node->isExpired() ? 'x' : ' ') << "]";

        // Get status
        {
-            char g_i = n->pending(n->getStatus) ? (n->candidate ? 'c' : 'f') : ' ';
-            char s_i = n->isSynced(now) ? (n->last_get_reply > last_get ? 'u' : 's') : '-';
+            char g_i = n.pending(n.getStatus) ? (n.candidate ? 'c' : 'f') : ' ';
+            char s_i = n.isSynced(now) ? (n.last_get_reply > last_get ? 'u' : 's') : '-';
            out << " [" << s_i << g_i << "] ";
        }

        // Listen status
        if (not sr.listeners.empty()) {
-            if (n->listenStatus.empty())
+            if (n.listenStatus.empty())
                out << "    ";
            else
                out << "["
-                    << (n->isListening(now) ? 'l' : (n->pending(n->listenStatus) ? 'f' : ' ')) << "] ";
+                    << (n.isListening(now) ? 'l' : (n.pending(n.listenStatus) ? 'f' : ' ')) << "] ";
        }

        // Announce status
        if (not sr.announce.empty()) {
-            if (n->acked.empty()) {
+            if (n.acked.empty()) {
                out << "   ";
                for (size_t a=0; a < sr.announce.size(); a++)
                    out << ' ';
            } else {
                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) {
+                    auto ack = n.acked.find(a.value->id);
+                    if (ack == n.acked.end() or not ack->second.first) {
                        out << ' ';
                    } else {
                        out << ack->second.first->getStateChar();
@@ -1547,7 +1583,7 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                out << "] ";
            }
        }
-        out << n->node->getAddrStr() << std::endl;
+        out << n.node->getAddrStr() << std::endl;
    }
 }

@@ -1583,11 +1619,6 @@ Dht::getStorageLog() const
    std::stringstream out;
    for (const auto& s : store)
        out << printStorageLog(s);
-    out << "Total " << store.size() << " storages, " << total_values << " values (";
-    if (total_store_size < 1024)
-        out << total_store_size << " bytes)";
-    else
-        out << (total_store_size/1024) << " KB)";
    out << std::endl << std::endl;
    std::multimap<size_t, const SockAddr*> q_map;
    for (const auto& ip : store_quota)
@@ -1596,6 +1627,12 @@ Dht::getStorageLog() const
    for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
        out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
    out << std::endl;
+    out << "Total " << store.size() << " storages, " << total_values << " values (";
+    if (total_store_size < 1024)
+        out << total_store_size << " bytes)";
+    else
+        out << (total_store_size/1024) << " / " << (max_store_size/1024) << " KB)";
+    out << std::endl;
    return out.str();
 }

@@ -1739,7 +1776,7 @@ Dht::Dht(const int& s, const int& s6, Config config)
 bool
 Dht::neighbourhoodMaintenance(RoutingTable& list)
 {
-    //DHT_LOG_DEBUG("neighbourhoodMaintenance");
+    //DHT_LOG_DBG("neighbourhoodMaintenance");
    auto b = list.findBucket(myid);
    if (b == list.end())
        return false;
@@ -2094,9 +2131,9 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
        for (auto& srp : searches(node->getFamily())) {
            auto& sr = srp.second;
            for (auto& n : sr->nodes) {
-                if (n->node != node) continue;
-                n->token.clear();
-                n->last_get_reply = time_point::min();
+                if (n.node != node) continue;
+                n.token.clear();
+                n.last_get_reply = time_point::min();
                searchSendGetValues(sr);
                scheduler.edit(sr->nextSearchStep, scheduler.time());
                break;
@@ -2307,7 +2344,8 @@ Dht::onAnnounce(Sp<Node> n,
        Sp<Value> vc = v;
        if (lv) {
            if (*lv == *vc) {
-                DHT_LOG.d(hash, node.id, "[store %s] nothing to do for %s", hash.toString().c_str(), lv->toString().c_str());
+                storageRefresh(hash, v->id);
+                DHT_LOG.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());
            } else {
                const auto& type = getType(lv->type);
                if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2339,14 +2377,11 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
 {
    using namespace net;

-    const auto& now = scheduler.time();
    if (not tokenMatch(token, node->getAddr())) {
        DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
        throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
    }
-
-    auto s = store.find(hash);
-    if (s != store.end() and s->second.refresh(now, vid)) {
+    if (storageRefresh(hash, vid)) {
        DHT_LOG.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());
    } else {
        DHT_LOG.d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2356,6 +2391,20 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
    return {};
 }

+bool
+Dht::storageRefresh(const InfoHash& id, Value::Id vid)
+{
+    const auto& now = scheduler.time();
+    auto s = store.find(id);
+    if (s != store.end()) {
+        auto expiration = s->second.refresh(now, vid, types);
+        if (expiration != time_point::max())
+            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        return true;
+    }
+    return false;
+}
+
 void
 Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {
@@ -254,7 +254,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
    o.thread = std::thread([=](){
        // Try to contact the proxy and set the status to connected when done.
        // will change the connectivity status
-        auto ok = std::make_shared<std::atomic_bool>(true);
+        struct GetState{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; };
+        auto state = std::make_shared<GetState>();
        restbed::Http::async(req,
            [=](const std::shared_ptr<restbed::Request>& req,
                const std::shared_ptr<restbed::Response>& reply) {
@@ -262,9 +263,9 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va

            if (code == 200) {
                try {
-                    while (restbed::Http::is_open(req) and not *finished) {
+                    while (restbed::Http::is_open(req) and not *finished and not state->stop) {
                        restbed::Http::fetch("\n", reply);
-                        if (*finished)
+                        if (*finished or state->stop)
                            break;
                        std::string body;
                        reply->get_body(body);
@@ -279,29 +280,30 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                            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, finished]() {
-                                    if (not *finished and not cb({value}))
-                                        *finished = true;
+                                callbacks_.emplace_back([cb, value, state]() {
+                                    if (not state->stop and not cb({value}))
+                                        state->stop = true;
                                });
                                loopSignal_();
                            }
                        } else {
-                            *ok = false;
+                            state->ok = false;
                        }
                    }
                } catch (std::runtime_error& e) { }
            } else {
-                *ok = false;
+                state->ok = false;
            }
        }).wait();
        if (donecb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
-                donecb(*ok, {});
+                donecb(state->ok, {});
+                state->stop = true;
            });
            loopSignal_();
        }
-        if (!ok) {
+        if (!state->ok) {
            // Connection failed, update connectivity
            opFailed();
        }
@@ -342,7 +344,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
                if (p != s->second.puts.end()) {
                    doPut(key, p->second.value,
                    [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                        *ok = !result;
+                        *ok = result;
                    }, time_point::max(), true);
                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                }
@@ -576,6 +578,8 @@ DhtProxyClient::getProxyInfos()
 void
 DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, sa_family_t family)
 {
+    if (isDestroying_)
+        return;
    std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
    auto oldStatus = std::max(statusIpv4_, statusIpv6_);
    auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
@@ -706,11 +710,12 @@ DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const V
                    Json::CharReaderBuilder rbuilder;
                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        auto expired = json.get("expired", Json::Value(false)).asBool();
                        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))
+                            callbacks_.emplace_back([cb, value, state, expired]() {
+                                if (not state->cancel and not cb({value}, expired))
                                    state->cancel = true;
                            });
                            loopSignal_();
@@ -948,14 +953,16 @@ DhtProxyClient::getConnectivityStatus()
 void
 DhtProxyClient::restartListeners()
 {
+    if (isDestroying_) return;
+    std::lock_guard<std::mutex> lock(searchLock_);
    DHT_LOG.d("Refresh permanent puts");
    for (auto& search : searches_) {
        for (auto& put : search.second.puts) {
-            if (!put.second.ok) {
+            if (!*put.second.ok) {
                auto ok = put.second.ok;
                doPut(search.first, put.second.value,
                [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                    *ok = !result;
+                    *ok = result;
                }, time_point::max(), true);
                scheduler.edit(put.second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
            }
@@ -971,7 +978,6 @@ DhtProxyClient::restartListeners()
        return;
    }
    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;
@@ -70,6 +70,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
        auto resource = std::make_shared<restbed::Resource>();
        resource->set_path("/");
        resource->set_method_handler("GET", std::bind(&DhtProxyServer::getNodeInfo, this, _1));
+        resource->set_method_handler("STATS", std::bind(&DhtProxyServer::getStats, this, _1));
        service_->publish(resource);
        resource = std::make_shared<restbed::Resource>();
        resource->set_path("/{hash: .*}");
@@ -137,7 +138,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
    printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this] {
        if (stopListeners) return;
        if (service_->is_up())
-            std::cout << getStats().toString() << std::endl;
+            updateStats();
        // Refresh stats cache
        auto newInfo = dht_->getNodeInfo();
        {
@@ -178,31 +179,30 @@ DhtProxyServer::stop()
        server_thread.join();
 }

-DhtProxyServer::ServerStats
-DhtProxyServer::getStats() const
+void
+DhtProxyServer::updateStats() const
 {
-    ServerStats ret {};
    auto now = clock::now();
    auto last = lastStatsReset_.exchange(now);
    auto count = requestNum_.exchange(0);
    auto dt = std::chrono::duration<double>(now - last);
-    ret.requestRate = count / dt.count();
+    stats_.requestRate = count / dt.count();
 #if OPENDHT_PUSH_NOTIFICATIONS
-    ret.pushListenersCount = pushListeners_.size();
+    stats_.pushListenersCount = pushListeners_.size();
 #endif
-    ret.putCount = puts_.size();
-    ret.listenCount = currentListeners_.size();
-    return ret;
+    stats_.putCount = puts_.size();
+    stats_.listenCount = currentListeners_.size();
+    stats_.nodeInfo = nodeInfo_;
 }

 void
-DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) const
+DhtProxyServer::getNodeInfo(const Sp<restbed::Session>& session) const
 {
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
    session->fetch(content_length,
-        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
+        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
        {
            try {
                if (dht_) {
@@ -235,14 +235,44 @@ DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) co
 }

 void
-DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
+DhtProxyServer::getStats(const Sp<restbed::Session>& session) const
+{
+    requestNum_++;
+    const auto request = session->get_request();
+    int content_length = std::stoi(request->get_header("Content-Length", "0"));
+    session->fetch(content_length,
+        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
+        {
+            try {
+                if (dht_) {
+#ifdef OPENDHT_JSONCPP
+                    Json::StreamWriterBuilder wbuilder;
+                    wbuilder["commentStyle"] = "None";
+                    wbuilder["indentation"] = "";
+                    auto output = Json::writeString(wbuilder, stats_.toJson()) + "\n";
+                    s->close(restbed::OK, output);
+#else
+                    s->close(restbed::NotFound, "{\"err\":\"JSON not enabled on this instance\"}");
+#endif
+                }
+                else
+                    s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
+            } catch (...) {
+                s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
+            }
+        }
+    );
+}
+
+void
+DhtProxyServer::get(const Sp<restbed::Session>& session) const
 {
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
    auto hash = request->get_path_parameter("hash");
    session->fetch(content_length,
-        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
+        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
        {
            try {
                if (dht_) {
@@ -250,23 +280,21 @@ DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
                    if (!infoHash) {
                        infoHash = InfoHash::get(hash);
                    }
-                    s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
-                        auto cacheSession = std::weak_ptr<restbed::Session>(s);
-
-                        dht_->get(infoHash, [cacheSession](std::shared_ptr<Value> value) {
-                            auto s = cacheSession.lock();
-                            if (!s) return false;
-                            // Send values as soon as we get them
-                            Json::StreamWriterBuilder wbuilder;
-                            wbuilder["commentStyle"] = "None";
-                            wbuilder["indentation"] = "";
-                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
-                            s->yield(output, [](const std::shared_ptr<restbed::Session> /*session*/){ });
-                            return true;
-                        }, [s](bool /*ok* */) {
-                            // Communication is finished
+                    s->yield(restbed::OK, "", [=](const Sp<restbed::Session>&) {});
+                    dht_->get(infoHash, [s](const Sp<Value>& value) {
+                        if (s->is_closed()) return false;
+                        // Send values as soon as we get them
+                        Json::StreamWriterBuilder wbuilder;
+                        wbuilder["commentStyle"] = "None";
+                        wbuilder["indentation"] = "";
+                        auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
+                        s->yield(output, [](const Sp<restbed::Session>& /*session*/){ });
+                        return true;
+                    }, [s](bool /*ok* */) {
+                        // Communication is finished
+                        if (not s->is_closed()) {
                            s->close();
-                        });
+                        }
                    });
                } else {
                    s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
@@ -279,7 +307,7 @@ DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
 }

 void
-DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
+DhtProxyServer::listen(const Sp<restbed::Session>& session)
 {
    requestNum_++;
    const auto request = session->get_request();
@@ -289,7 +317,7 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
    if (!infoHash)
        infoHash = InfoHash::get(hash);
    session->fetch(content_length,
-        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
+        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
        {
            try {
                if (dht_) {
@@ -307,7 +335,7 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
                    // cache the session to avoid an incrementation of the shared_ptr's counter
                    // else, the session->close() will not close the socket.
                    auto cacheSession = std::weak_ptr<restbed::Session>(s);
-                    listener.token = dht_->listen(infoHash, [cacheSession](std::shared_ptr<Value> value) {
+                    listener.token = dht_->listen(infoHash, [cacheSession](const std::vector<Sp<Value>>& values, bool expired) {
                        auto s = cacheSession.lock();
                        if (!s) return false;
                        // Send values as soon as we get them
@@ -315,8 +343,13 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
                            Json::StreamWriterBuilder wbuilder;
                            wbuilder["commentStyle"] = "None";
                            wbuilder["indentation"] = "";
-                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
-                            s->yield(output, [](const std::shared_ptr<restbed::Session>){ });
+                            for (const auto& value : values) {
+                                auto val = value->toJson();
+                                if (expired)
+                                    val["expired"] = true;
+                                auto output = Json::writeString(wbuilder, val) + "\n";
+                                s->yield(output, [](const Sp<restbed::Session>&){ });
+                            }
                        }
                        return !s->is_closed();
                    });
@@ -452,7 +485,6 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
 void
 DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
 {
-
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
@@ -37,6 +37,7 @@
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #define close(x) closesocket(x)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
 #endif

 namespace dht {
@@ -164,7 +165,7 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    running_network = false;
+    stopNetwork();
    running = false;
    cv.notify_all();
    bootstrap_cv.notify_all();
@@ -413,7 +414,7 @@ DhtRunner::loop_()
        while (not received.empty()) {
            auto& pck = received.front();
            auto delay = clock::now() - pck.received;
-            if (delay > std::chrono::seconds(1))
+            if (delay > std::chrono::milliseconds(500))
                std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
            else
                wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
@@ -458,8 +459,10 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    if (is_ipv6)
        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
    int rc = bind(sock, addr.get(), addr.getLength());
-    if(rc < 0)
+    if(rc < 0) {
+        close(sock);
        throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
+    }
    sockaddr_storage ss;
    socklen_t ss_len = sizeof(ss);
    getsockname(sock, (sockaddr*)&ss, &ss_len);
@@ -467,12 +470,64 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    return sock;
 }

+#ifdef _WIN32
+inline void udpPipe(int fds[2])
+{
+    int lst = socket(AF_INET, SOCK_DGRAM, 0);
+    if (lst < 0)
+        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
+    sockaddr_in inaddr;
+    sockaddr addr;
+    memset(&inaddr, 0, sizeof(inaddr));
+    memset(&addr, 0, sizeof(addr));
+    inaddr.sin_family = AF_INET;
+    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    inaddr.sin_port = 0;
+    int yes=1;
+    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
+    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
+    if (rc < 0) {
+        close(lst);
+        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
+    }
+    socklen_t len = sizeof(inaddr);
+    getsockname(lst, &addr, &len);
+    fds[0] = lst;
+    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
+    connect(fds[1], &addr, len);
+}
+#endif
+
+void
+DhtRunner::stopNetwork()
+{
+    running_network = false;
+    if (stop_writefd != -1) {
+        if (write(stop_writefd, "\0", 1) == -1) {
+            perror("write");
+        }
+    }
+}
+
 void
 DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 {
-    running_network = false;
+    stopNetwork();
    if (rcv_thread.joinable())
        rcv_thread.join();
+
+    int stopfds[2];
+#ifndef _WIN32
+    auto status = pipe(stopfds);
+    if (status == -1) {
+        throw DhtException("Can't open pipe");
+    }
+#else
+    udpPipe(stopfds);
+#endif
+    int stop_readfd = stopfds[0];
+    stop_writefd = stopfds[1];
+
    s4 = -1;
    s6 = -1;

@@ -487,19 +542,19 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #endif

    running_network = true;
-    rcv_thread = std::thread([this]() {
+    rcv_thread = std::thread([this, stop_readfd]() {
        try {
            while (running_network) {
-                struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000};
                fd_set readfds;

                FD_ZERO(&readfds);
+                FD_SET(stop_readfd, &readfds);
                if(s4 >= 0)
                    FD_SET(s4, &readfds);
                if(s6 >= 0)
                    FD_SET(s6, &readfds);

-                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv);
+                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, nullptr);
                if(rc < 0) {
                    if(errno != EINTR) {
                        perror("select");
@@ -520,13 +575,13 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                    else if(s6 >= 0 && FD_ISSET(s6, &readfds))
                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                    else
-                        break;
+                        continue;
                    if (rc > 0) {
                        {
                            std::lock_guard<std::mutex> lck(sock_mtx);
                            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
                                std::cerr << "Dropping packet: queue is full!" << std::endl;
-                                continue;
+                                rcv.pop();
                            }
                            rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
                        }
@@ -545,6 +600,11 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
        s6 = -1;
        bound4 = {};
        bound6 = {};
+        if (stop_readfd != -1)
+            close(stop_readfd);
+        if (stop_writefd != -1)
+            close(stop_writefd);
+        stop_writefd = -1;
    });
 }

@@ -583,6 +643,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
        pending_ops.emplace([=](SecureDht& dht) mutable {
+#if OPENDHT_PROXY_CLIENT
            auto tokenbGlobal = listener_token_++;
            Listener listener {};
            listener.hash = hash;
@@ -590,21 +651,20 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
            listener.w = std::move(w);
            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
                if (not vcb(vals, expired)) {
-#if OPENDHT_PROXY_CLIENT
                    cancelListen(hash, tokenbGlobal);
-#endif
                    return false;
                }
                return true;
            };
-#if OPENDHT_PROXY_CLIENT
            if (use_proxy)
                listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
            else
-#endif
                listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
            listeners_.emplace(tokenbGlobal, std::move(listener));
            ret_token->set_value(tokenbGlobal);
+#else
+            ret_token->set_value(dht.listen(hash, vcb, f, w));
+#endif
        });
    }
    cv.notify_all();
@@ -898,8 +958,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-    listeners_.clear();
 #if OPENDHT_PROXY_CLIENT
+    listeners_.clear();
    dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
    dht_.reset();
@@ -185,7 +185,7 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }

 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -215,7 +215,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&
 }

 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -419,7 +419,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
        msg->msgpack_unpack(msg_res.get());
    } catch (const std::exception& e) {
        DHT_LOG.w("Can't parse message of size %lu: %s", buflen, e.what());
-        DHT_LOG.DEBUG.logPrintable(buf, buflen);
+        //DHT_LOG.DBG.logPrintable(buf, buflen);
        return;
    }

@@ -551,7 +551,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
        }
        case MessageType::Reply:
            if (req) { /* request reply */
-                if (msg->type == MessageType::AnnounceValue or msg->type == MessageType::Listen)
+                if (req->getType() == MessageType::AnnounceValue or req->getType() == MessageType::Listen)
                    req->node->authSuccess();
                req->reply_time = scheduler.time();

@@ -977,7 +977,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
            pk.pack(std::string("v")); pk.pack_array(st.size()*fields.size());
            for (const auto& v : st)
                v->msgpack_pack_fields(fields, pk);
-            //DHT_LOG_DEBUG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields",
+            //DHT_LOG_DBG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields",
            //        nodes.size(), nodes6.size(), st.size(), fields.size());
        }
    }
@@ -20,6 +20,9 @@

 namespace dht {

+constexpr size_t CLEANUP_MAX_NODES {1024};
+constexpr size_t CLEANUP_FREQ {1024};
+
 NodeCache::~NodeCache()
 {
    cache_4.setExpired();
@@ -41,29 +44,33 @@ NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, boo
 std::vector<Sp<Node>>
 NodeCache::getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) const
 {
-    const auto& c = cache(sa_f);
+    return cache(sa_f).getCachedNodes(id, count);
+}

+std::vector<Sp<Node>>
+NodeCache::NodeMap::getCachedNodes(const InfoHash& id, size_t count) const
+{
    std::vector<Sp<Node>> nodes;
-    nodes.reserve(std::min(c.size(), count));
-    NodeMap::const_iterator it;
-    auto dec_it = [&c](NodeMap::const_iterator& it) {
+    nodes.reserve(std::min(size(), count));
+    const_iterator it;
+    auto dec_it = [this](const_iterator& it) {
        auto ret = it;
-        it = (it == c.cbegin()) ? c.cend() : std::prev(it);
+        it = (it == cbegin()) ? cend() : std::prev(it);
        return ret;
    };

-    auto it_p = c.lower_bound(id),
+    auto it_p = lower_bound(id),
        it_n = it_p;
-    if (not c.empty())
+    if (not empty())
        dec_it(it_p); /* Create 2 separate iterator if we could */

-    while (nodes.size() < count and (it_n != c.cend() or it_p != c.cend())) {
+    while (nodes.size() < count and (it_n != cend() or it_p != cend())) {
        /* If one of the iterator is at the end, then take the other one
           If they are both in middle of somewhere comapre both and take
           the closest to the id. */
-        if (it_p == c.cend())       it = it_n++;
-        else if (it_n == c.cend())  it = dec_it(it_p);
-        else                        it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;
+        if (it_p == cend())      it = it_n++;
+        else if (it_n == cend()) it = dec_it(it_p);
+        else                     it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;

        if (auto n = it->second.lock())
            if ( not n->isExpired() and not n->isClient() )
@@ -104,6 +111,10 @@ NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point
    if (not node) {
        node = std::make_shared<Node>(id, addr, client);
        it.first->second = node;
+        if (cleanup_counter++ == CLEANUP_FREQ) {
+            cleanup();
+            cleanup_counter = 0;
+        }
    } else if (confirm or node->isOld(now)) {
        node->update(addr);
    }
@@ -120,6 +131,7 @@ NodeCache::NodeMap::clearBadNodes() {
            erase(it++);
        }
    }
+    cleanup_counter = 0;
 }

 void
@@ -128,6 +140,21 @@ NodeCache::NodeMap::setExpired() {
        if (auto n = wn.second.lock())
            n->setExpired();
    clear();
+    cleanup_counter = 0;
+}
+
+void
+NodeCache::NodeMap::cleanup()
+{
+    auto it = lower_bound(InfoHash::getRandom());
+    for (size_t n = 0, maxNodes = std::min(size(), CLEANUP_MAX_NODES); n != maxNodes; n++) {
+        if (it == end())
+            it = begin();
+        if (it->second.expired())
+            erase(it++);
+        else
+            ++it;
+    }
 }

 }
@@ -91,9 +91,9 @@ struct Dht::SearchNode {

    SearchNode() : node() {}
    SearchNode(const SearchNode&) = delete;
-    SearchNode(SearchNode&&) = delete;
+    SearchNode(SearchNode&&) = default;
    SearchNode& operator=(const SearchNode&) = delete;
-    SearchNode& operator=(SearchNode&&) = delete;
+    SearchNode& operator=(SearchNode&&) = default;

    SearchNode(const Sp<Node>& node) : node(node) {}
    ~SearchNode() {
@@ -136,7 +136,7 @@ struct Dht::SearchNode {
     *
     * @return true if we can send get, else false.
     */
-    bool canGet(time_point now, time_point update, Sp<Query> q = {}) const {
+    bool canGet(time_point now, time_point update, const Sp<Query>& q) const {
        if (node->isExpired())
            return false;

@@ -147,9 +147,9 @@ struct Dht::SearchNode {
            if (s.second and s.second->pending())
                pending = true;
            if (s.first and q and q->isSatisfiedBy(*s.first) and s.second) {
-                if (s.second->pending() and not pending_sq_status)
+                if (s.second->pending())
                    pending_sq_status = true;
-                if (s.second->completed() and not (update > s.second->reply_time) and not completed_sq_status)
+                else if (s.second->completed() and not (update > s.second->reply_time))
                    completed_sq_status = true;
                if (completed_sq_status and pending_sq_status)
                    break;
@@ -157,7 +157,7 @@ struct Dht::SearchNode {
        }

        return (not pending and now > last_get_reply + Node::NODE_EXPIRE_TIME) or
-                not (hasStartedPagination(q) or completed_sq_status or pending_sq_status);
+                not (completed_sq_status or pending_sq_status or hasStartedPagination(q));
    }

    /**
@@ -320,22 +320,21 @@ struct Dht::SearchNode {
    }

    /**
-     * Assumng the node is synced, should a "put" request be sent to this node now ?
+     * Assuming the node is synced, should a "put" request be sent to this node now ?
     */
    time_point getAnnounceTime(Value::Id vid) const {
        const auto& ack = acked.find(vid);
-        const auto& gs = probe_query ? getStatus.find(probe_query) : getStatus.cend();
-        if ((ack == acked.cend() or not ack->second.first) and (gs == getStatus.cend()
-                                                          or not gs->second or not gs->second->pending()))
+        if (ack == acked.cend() or not ack->second.first) {
            return time_point::min();
-        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.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
+        }
+        if (ack->second.first->completed()) {
+            return ack->second.second - REANNOUNCE_MARGIN;
+        }
+        return ack->second.first->pending() ? time_point::max() : time_point::min();
    }

    /**
-     * Assumng the node is synced, should the "listen" request with Query q be
+     * Assuming the node is synced, should the "listen" request with Query q be
     * sent to this node now ?
     */
    time_point getListenTime(const Sp<Query>& q) const {
@@ -369,7 +368,7 @@ struct Dht::Search {

    bool expired {false};              /* no node, or all nodes expired */
    bool done {false};                 /* search is over, cached for later */
-    std::vector<std::unique_ptr<SearchNode>> nodes {};
+    std::vector<SearchNode> nodes {};

    /* pending puts */
    std::vector<Announce> announce {};
@@ -404,17 +403,17 @@ struct Dht::Search {
    bool insertNode(const Sp<Node>& n, time_point now, const Blob& token={});

    SearchNode* getNode(const Sp<Node>& n) {
-        for (auto& sn : nodes)
-            if (sn->node == n)
-                return sn.get();
-        return nullptr;
+        auto srn = std::find_if(nodes.begin(), nodes.end(), [&](SearchNode& sn) {
+            return n == sn.node;
+        });
+        return (srn == nodes.end()) ? nullptr : &(*srn);
    }

    /* number of concurrent sync requests */
    unsigned currentlySolicitedNodeCount() const {
        unsigned count = 0;
        for (const auto& n : nodes)
-            if (not n->isBad() and n->pendingGet())
+            if (not n.isBad() and n.pendingGet())
                count++;
        return count;
    }
@@ -430,7 +429,8 @@ struct Dht::Search {
     *
     * @param query  The query identifying a 'get' request.
     */
-    time_point getLastGetTime(Sp<Query> query = {}) const;
+    time_point getLastGetTime(const Query&) const;
+    time_point getLastGetTime() const;

    /**
     * Is this get operation done ?
@@ -447,12 +447,12 @@ struct Dht::Search {
     */
    void setDone(const Get& get) {
        for (auto& n : nodes) {
-            auto pqs = n->pagination_queries.find(get.query);
-            if (pqs != n->pagination_queries.cend()) {
+            auto pqs = n.pagination_queries.find(get.query);
+            if (pqs != n.pagination_queries.cend()) {
                for (auto& pq : pqs->second)
-                    n->getStatus.erase(pq);
+                    n.getStatus.erase(pq);
            }
-            n->getStatus.erase(get.query);
+            n.getStatus.erase(get.query);
        }
        if (get.done_cb)
            get.done_cb(true, getNodes());
@@ -465,9 +465,9 @@ struct Dht::Search {
     */
    void setDone() {
        for (auto& n : nodes) {
-            n->getStatus.clear();
-            n->listenStatus.clear();
-            n->acked.clear();
+            n.getStatus.clear();
+            n.listenStatus.clear();
+            n.acked.clear();
        }
        done = true;
    }
@@ -475,6 +475,17 @@ struct Dht::Search {
    bool isAnnounced(Value::Id id) const;
    bool isListening(time_point now) const;

+    void get(Value::Filter f, const Sp<Query>& q, const QueryCallback& qcb, const GetCallback& gcb, const DoneCallback& dcb, Scheduler& scheduler) {
+        if (gcb or qcb) {
+            const auto& now = scheduler.time();
+            callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
+            auto values = cache.get(f);
+            if (not values.empty())
+                gcb(values);
+            scheduler.edit(nextSearchStep, now);
+        }
+    }
+
    size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
        //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){
@@ -499,9 +510,9 @@ struct Dht::Search {
                    }
                    for (auto& sn : nodes) {
                        if (listeners.empty())
-                            sn->cancelListen();
+                            sn.cancelListen();
                        else if (query)
-                            sn->cancelListen(query);
+                            sn.cancelListen(query);
                    }
                });
                scheduler.edit(opExpirationJob, nextExpire);
@@ -513,14 +524,14 @@ struct Dht::Search {
     * @return The number of non-good search nodes.
     */
    unsigned getNumberOfBadNodes() const {
-        return std::count_if(nodes.begin(), nodes.end(), [](const std::unique_ptr<SearchNode>& sn) {
-            return sn->isBad();
+        return std::count_if(nodes.begin(), nodes.end(), [](const SearchNode& sn) {
+            return sn.isBad();
        });
    }
    unsigned getNumberOfConsecutiveBadNodes() const {
        unsigned count = 0;
-        std::find_if(nodes.begin(), nodes.end(), [&count](const std::unique_ptr<SearchNode>& sn) {
-            if (not sn->isBad())
+        std::find_if(nodes.begin(), nodes.end(), [&count](const SearchNode& sn) {
+            if (not sn.isBad())
                return true;
            ++count;
            return false;
@@ -539,7 +550,7 @@ struct Dht::Search {
     */
    bool removeExpiredNode(const time_point& now) {
        for (auto e = nodes.cend(); e != nodes.cbegin();) {
-            const Node& n = *(*(--e))->node;
+            const Node& n = *(--e)->node;
            if (n.isRemovable(now)) {
                //std::cout << "Removing expired node " << n.id << " from IPv" << (af==AF_INET?'4':'6') << " search " << id << std::endl;
                nodes.erase(e);
@@ -612,7 +623,7 @@ struct Dht::Search {
        // remove acked for cleared annouces
        for (auto it = announced; it != announce.end(); ++it) {
            for (auto& n : nodes)
-                n->acked.erase(it->value->id);
+                n.acked.erase(it->value->id);
        }
        announce.erase(announced, announce.end());
    }
@@ -645,13 +656,13 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
    auto n = nodes.end();
    while (n != nodes.begin()) {
        --n;
-        if ((*n)->node == snode) {
+        if (n->node == snode) {
            found = true;
            break;
        }

        /* Node not found. We could insert it after this one. */
-        if (id.xorCmp(nid, (*n)->node->id) > 0) {
+        if (id.xorCmp(nid, n->node->id) > 0) {
            ++n;
            break;
        }
@@ -675,7 +686,7 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
            full = nodes.size() - bad >=  SEARCH_NODES;
            while (std::distance(nodes.cbegin(), t) - bad >  SEARCH_NODES) {
                --t;
-                if ((*t)->isBad())
+                if (t->isBad())
                    bad--;
            }
        }
@@ -691,7 +702,7 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
        if (nodes.empty()) {
            step_time = time_point::min();
        }
-        n = nodes.insert(n, std::unique_ptr<SearchNode>(new SearchNode(snode)));
+        n = nodes.insert(n, SearchNode(snode));
        node.setTime(now);
        new_search_node = true;
        if (node.isExpired()) {
@@ -703,16 +714,16 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
        }

        while (nodes.size() - bad >  SEARCH_NODES) {
-            if (not expired and nodes.back()->isBad())
+            if (not expired and nodes.back().isBad())
                bad--;
            nodes.pop_back();
        }
    }
    if (not token.empty()) {
-        (*n)->candidate = false;
-        (*n)->last_get_reply = now;
+        n->candidate = false;
+        n->last_get_reply = now;
        if (token.size() <= 64)
-            (*n)->token = token;
+            n->token = token;
        expired = false;
    }
    if (new_search_node)
@@ -726,7 +737,7 @@ Dht::Search::getNodes() const
    std::vector<Sp<Node>> ret {};
    ret.reserve(nodes.size());
    for (const auto& sn : nodes)
-        ret.emplace_back(sn->node);
+        ret.emplace_back(sn.node);
    return ret;
 }

@@ -735,9 +746,9 @@ Dht::Search::isSynced(time_point now) const
 {
    unsigned i = 0;
    for (const auto& n : nodes) {
-        if (n->isBad())
+        if (n.isBad())
            continue;
-        if (not n->isSynced(now))
+        if (not n.isSynced(now))
            return false;
        if (++i == TARGET_NODES)
            break;
@@ -746,11 +757,20 @@ Dht::Search::isSynced(time_point now) const
 }

 time_point
-Dht::Search::getLastGetTime(Sp<Query> q) const
+Dht::Search::getLastGetTime(const Query& q) const
 {
    time_point last = time_point::min();
    for (const auto& g : callbacks)
-        last = std::max(last, (not q or q->isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
+        last = std::max(last, (q.isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
+    return last;
+}
+
+time_point
+Dht::Search::getLastGetTime() const
+{
+    time_point last = time_point::min();
+    for (const auto& g : callbacks)
+        last = std::max(last, g.second.start);
    return last;
 }

@@ -759,9 +779,9 @@ Dht::Search::isDone(const Get& get) const
 {
    unsigned i = 0;
    for (const auto& sn : nodes) {
-        if (sn->isBad())
+        if (sn.isBad())
            continue;
-        if (not sn->isDone(get))
+        if (not sn.isDone(get))
            return false;
        if (++i == TARGET_NODES)
            break;
@@ -776,9 +796,9 @@ Dht::Search::isAnnounced(Value::Id id) const
        return false;
    unsigned i = 0;
    for (const auto& n : nodes) {
-        if (n->isBad())
+        if (n.isBad())
            continue;
-        if (not n->isAnnounced(id))
+        if (not n.isAnnounced(id))
            return false;
        if (++i == TARGET_NODES)
            return true;
@@ -793,14 +813,14 @@ Dht::Search::isListening(time_point now) const
        return false;
    unsigned i = 0;
    for (const auto& n : nodes) {
-        if (n->isBad())
+        if (n.isBad())
            continue;
        SearchNode::NodeListenerStatus::const_iterator ls {};
-        for (ls = n->listenStatus.begin(); ls != n->listenStatus.end() ; ++ls) {
-            if (n->isListening(now, ls))
+        for (ls = n.listenStatus.begin(); ls != n.listenStatus.end() ; ++ls) {
+            if (n.isListening(now, ls))
                break;
        }
-        if (ls == n->listenStatus.end())
+        if (ls == n.listenStatus.end())
            return false;
        if (++i == LISTEN_NODES)
            break;
@@ -56,7 +56,7 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
            1
        }, [this](bool ok) {
            if (ok)
-                DHT_LOG.DEBUG("SecureDht: public key announced successfully");
+                DHT_LOG.DBG("SecureDht: public key announced successfully");
        }, {}, true);
    }
 }
@@ -68,27 +68,19 @@ ValueType
 SecureDht::secureType(ValueType&& type)
 {
    type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
-        if (v->isSigned()) {
-            if (!v->owner or !v->owner->checkSignature(v->getToSign(), v->signature)) {
-                DHT_LOG.WARN("Signature verification failed");
-                return false;
-            }
-            else
-                DHT_LOG.WARN("Signature verification succeeded");
-        }
+        if (v->isSigned())
+            return v->checkSignature();
        return type.storePolicy(id, v, nid, a);
    };
    type.editPolicy = [this,type](InfoHash id, const Sp<Value>& o, Sp<Value>& n, const InfoHash& nid, const SockAddr& a) {
-        if (!o->isSigned())
+        if (not o->isSigned())
            return type.editPolicy(id, o, n, nid, a);
-        if (o->owner != n->owner) {
+        if (o->owner != n->owner or not n->isSigned()) {
            DHT_LOG.WARN("Edition forbidden: owner changed.");
            return false;
        }
-        if (!o->owner or !o->owner->checkSignature(n->getToSign(), n->signature)) {
-            DHT_LOG.WARN("Edition forbidden: signature verification failed.");
+        if (not n->checkSignature())
            return false;
-        }
        if (o->seq == n->seq) {
            // If the data is exactly the same,
            // it can be reannounced, possibly by someone else.
@@ -139,7 +131,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
    }
    InfoHash h = crt->getPublicKey().getId();
    if (node == h) {
-        DHT_LOG.DEBUG("Registering certificate for %s", h.toString().c_str());
+        DHT_LOG.DBG("Registering certificate for %s", h.toString().c_str());
        auto it = nodesCertificates_.find(h);
        if (it == nodesCertificates_.end())
            std::tie(it, std::ignore) = nodesCertificates_.emplace(h, std::move(crt));
@@ -147,7 +139,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
            it->second = std::move(crt);
        return it->second;
    } else {
-        DHT_LOG.DEBUG("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
+        DHT_LOG.WARN("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
        return nullptr;
    }
 }
@@ -164,7 +156,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
 {
    Sp<crypto::Certificate> b = getCertificate(node);
    if (b && *b) {
-        DHT_LOG.DEBUG("Using certificate from cache for %s", node.toString().c_str());
+        DHT_LOG.DBG("Using certificate from cache for %s", node.toString().c_str());
        if (cb)
            cb(b);
        return;
@@ -172,7 +164,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
    if (localQueryMethod_) {
        auto res = localQueryMethod_(node);
        if (not res.empty()) {
-            DHT_LOG.DEBUG("Registering certificate from local store for %s", node.toString().c_str());
+            DHT_LOG.DBG("Registering certificate from local store for %s", node.toString().c_str());
            nodesCertificates_.emplace(node, res.front());
            if (cb)
                cb(res.front());
@@ -187,7 +179,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
        for (const auto& v : vals) {
            if (auto cert = registerCertificate(node, v->data)) {
                *found = true;
-                DHT_LOG.DEBUG("Found certificate for %s", node.toString().c_str());
+                DHT_LOG.DBG("Found certificate for %s", node.toString().c_str());
                if (cb)
                    cb(cert);
                return false;
@@ -205,7 +197,7 @@ SecureDht::findPublicKey(const InfoHash& node, std::function<void(const Sp<const
 {
    auto pk = getPublicKey(node);
    if (pk && *pk) {
-        DHT_LOG.DEBUG("Found public key from cache for %s", node.toString().c_str());
+        DHT_LOG.DBG("Found public key from cache for %s", node.toString().c_str());
        if (cb)
            cb(pk);
        return;
@@ -236,24 +228,22 @@ SecureDht::checkValue(const Sp<Value>& v)
            return {};
        }
        try {
-            Value decrypted_val (decrypt(*v));
-            if (decrypted_val.recipient == getId()) {
-                if (decrypted_val.owner)
-                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
-                return std::make_shared<Value>(std::move(decrypted_val));
+            if (auto decrypted_val = v->decrypt(*key_)) {
+                if (decrypted_val->owner)
+                    nodesPubKeys_[decrypted_val->owner->getId()] = decrypted_val->owner;
+                return decrypted_val;
            }
-            // Ignore values belonging to other people
        } catch (const std::exception& e) {
            DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
        }
    }
    // Check signed values
    else if (v->isSigned()) {
-        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
-            nodesPubKeys_[v->owner->getId()] = v->owner;
+        if (v->checkSignature()) {
+            if (v->owner)
+                nodesPubKeys_[v->owner->getId()] = v->owner;
            return v;
-        }
-        else
+        } else
            DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
    }
    // Forward normal values
@@ -327,14 +317,13 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
    // Check if we are already announcing a value
    auto p = dht_->getPut(hash, val->id);
    if (p && val->seq <= p->seq) {
-        DHT_LOG.DEBUG("Found previous value being announced.");
        val->seq = p->seq + 1;
    }

    // Check if data already exists on the dht
    get(hash,
        [val,this] (const std::vector<Sp<Value>>& vals) {
-            DHT_LOG.DEBUG("Found online previous value being announced.");
+            DHT_LOG.DBG("Found online previous value being announced.");
            for (const auto& v : vals) {
                if (!v->isSigned())
                    DHT_LOG.ERR("Existing non-signed value seems to exists at this location.");
@@ -356,7 +345,7 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
 void
 SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val, DoneCallback callback, bool permanent)
 {
-    findPublicKey(to, [=](const Sp<const crypto::PublicKey> pk) {
+    findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
        if(!pk || !*pk) {
            if (callback)
                callback(false, {});
@@ -154,15 +154,16 @@ struct Storage {
     *
     * @param now  The reference to now
     * @param vid  The value id
-     * @return true if a value storage was updated, false otherwise
+     * @return time of the next expiration, time_point::max() if no expiration
     */
-    bool refresh(const time_point& now, const Value::Id& vid) {
+    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;
-                return true;
+                vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
+                return vs.expiration;
            }
-        return false;
+        return time_point::max();
    }

    StoreDiff remove(const InfoHash& id, Value::Id);
@@ -233,6 +233,39 @@ Value::toJson() const
    return val;
 }

+bool
+Value::checkSignature()
+{
+    if (!signatureChecked) {
+        signatureChecked = true;
+        if (isSigned()) {
+            signatureValid = owner and owner->checkSignature(getToSign(), signature);
+        }
+    }
+    return signatureValid;
+}
+
+Sp<Value>
+Value::decrypt(const crypto::PrivateKey& key)
+{
+    if (not decrypted) {
+        decrypted = true;
+        if (isEncrypted()) {
+            auto decryptedBlob = key.decrypt(cypher);
+            std::unique_ptr<Value> v {new Value(id)};
+            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            v->msgpack_unpack_body(msg.get());
+            if (v->recipient != key.getPublicKey().getId())
+                throw crypto::DecryptError("Recipient mismatch");
+            // Ignore values belonging to other people
+            if (not v->owner or not v->owner->checkSignature(v->getToSign(), v->signature))
+                throw crypto::DecryptError("Signature mismatch");
+            decryptedValue = std::move(v);
+        }
+    }
+    return decryptedValue;
+}
+
 uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
    uint64_t ret = 0;
@@ -66,7 +66,7 @@ public:
    CallbackQueue expireValues(const time_point& now, time_point& next) {
        std::vector<Sp<Value>> expired_values;
        for (auto it = values.begin(); it != values.end();) {
-            if (it->second.expiration < now) {
+            if (it->second.expiration <= now) {
                expired_values.emplace_back(std::move(it->second.data));
                it = values.erase(it);
            } else {
@@ -74,7 +74,7 @@ public:
                ++it;
            }
        }
-        while (values.size() >= MAX_VALUES) {
+        while (values.size() > MAX_VALUES) {
            // too many values, remove oldest values
            time_point oldest_creation = time_point::max();
            auto oldest_value = values.end();
@@ -127,7 +127,7 @@ private:
    ValueCache& operator=(ValueCache&&) = delete;

    /* The maximum number of values we store in the cache. */
-    static constexpr unsigned MAX_VALUES {1024};
+    static constexpr unsigned MAX_VALUES {4096};

    struct CacheValueStorage {
        Sp<Value> data {};
@@ -46,6 +46,43 @@ CryptoTester::testSignatureEncryption() {
    CPPUNIT_ASSERT(data == decrypted);
 }

+void
+CryptoTester::testCertificateRevocation()
+{
+    auto ca1 = dht::crypto::generateIdentity("ca1");
+    auto account1 = dht::crypto::generateIdentity("acc1", ca1, 4096, true);
+    auto device11 = dht::crypto::generateIdentity("dev11", account1);
+    auto device12 = dht::crypto::generateIdentity("dev12", account1);
+
+
+    dht::crypto::TrustList list;
+    list.add(*ca1.second);
+    auto v = list.verify(*account1.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
+
+    list.add(*account1.second);
+    v = list.verify(*device11.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
+    v = list.verify(*device12.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
+
+    auto ca2 = dht::crypto::generateIdentity("ca2");
+    auto account2 = dht::crypto::generateIdentity("acc2", ca2, 4096, true);
+    auto device2 = dht::crypto::generateIdentity("dev2", account2);
+
+    v = list.verify(*device2.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
+    
+    account1.second->revoke(*account1.first, *device11.second);
+    dht::crypto::TrustList list2;
+    list2.add(*account1.second);
+
+    v = list2.verify(*device11.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
+    v = list2.verify(*device12.second);
+    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
+}
+
 void
 CryptoTester::tearDown() {

@@ -28,6 +28,7 @@ namespace test {
 class CryptoTester : public CppUnit::TestFixture {
    CPPUNIT_TEST_SUITE(CryptoTester);
    CPPUNIT_TEST(testSignatureEncryption);
+    CPPUNIT_TEST(testCertificateRevocation);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -40,9 +41,13 @@ class CryptoTester : public CppUnit::TestFixture {
     */
    void tearDown();
    /**
-     * Test the differents behaviors of constructors
+     * Test data signature, encryption and decryption
     */
    void testSignatureEncryption();
+    /**
+     * Test certificate generation, validation and revocation
+     */
+    void testCertificateRevocation();
 };

 }  // namespace test
@@ -116,10 +116,7 @@ DhtProxyTester::testListen() {
    // All values
    dht::Value secondVal {"You're a monster"};
    auto secondVal_data = secondVal.data;
-    nodePeer.put(key, std::move(secondVal), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
+    nodePeer.put(key, std::move(secondVal));
    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
    // Here values should contains 3 values
    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);
@@ -1,7 +1,7 @@
 bin_PROGRAMS = dhtnode dhtchat dhtscanner
 noinst_HEADERS = tools_common.h

-AM_CPPFLAGS = -I../include
+AM_CPPFLAGS = -I../include @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@

 dhtnode_SOURCES = dhtnode.cpp
 dhtnode_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @Argon2_LDFLAGS@ @GnuTLS_LIBS@
@@ -144,7 +144,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PROXY_SERVER
            for (const auto& proxy : proxies) {
                std::cout << "Stats for proxy on port " << proxy.first << std::endl;
-                std::cout << "  " << proxy.second->getStats().toString() << std::endl;
+                std::cout << "  " << proxy.second->stats().toString() << std::endl;
            }
 #endif
            continue;
@@ -63,11 +63,15 @@ $(function() {
                        if (!element || element.length == 0)
                            return;
                        const o = JSON.parse(element);
-                        const d = window.atob(o.data);
-                        const delay = Math.max(0, start-now);
-                        $('<li class="list-group-item"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
-                        lastAppended = i+1;
-                        start = Math.max(start, now)+25;
+                        if (o.expired) {
+                            $('#value'+o.id).slideUp(100, (e) => $(e.target).remove());
+                        } else {
+                            const d = window.atob(o.data);
+                            const delay = Math.max(0, start-now);
+                            $('<li class="list-group-item" id="value'+o.id+'"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
+                            lastAppended = i+1;
+                            start = Math.max(start, now)+25;
+                        }
                    }
                } else if (this.status !== 0) {
                    group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));
Author	SHA1	Message	Date
Adrien Béraud	b3511d223b	sockaddr: use more precise exception	2018-12-17 16:41:00 -05:00
Adrien Béraud	d1aca77809	securedht: move crypto ops to Value	2018-12-17 16:41:00 -05:00
Adrien Béraud	306d573223	proxy client: fix race condition	2018-12-17 16:41:00 -05:00
Sébastien Blin	20f4355b73	Merge pull request #356 from savoirfairelinux/build securedht: add override	2018-12-11 14:52:07 -05:00
Adrien Béraud	a2d7cd56a8	Merge pull request #358 from atraczyk/master replace _write with send when building for windows	2018-12-11 14:48:26 -05:00
Andreas Traczyk	5cd53f3624	replace _write with send when building for windows	2018-12-11 14:37:14 -05:00
Adrien Béraud	32eea55f10	autotools: don't build doc if doxygen not found	2018-12-11 14:04:35 -05:00
Adrien Béraud	f2f913f93f	securedht: add override for relevant methods	2018-12-11 14:04:35 -05:00
Adrien Béraud	cb9fe9d1fe	Merge pull request #354 from atraczyk/master msvc: replace write with _write when building on windows	2018-12-06 12:23:36 -05:00
Andreas Traczyk	dcab32ef4f	replace write with _write when building with msvc	2018-12-06 12:16:59 -05:00
Adrien Béraud	4948bd7ad3	node cache: remove expired weak pointers from cache Remove expired Node weak pointers regularly, in a way that ensures that: * every expired pointer gets eventually cleaned up * expired pointer ratio remains reasonably low * no cleanup pass blocks the node for too long, * impact on performance is negligible Impact on performance: Every N insertions, at most N nodes are tested by iterating the map -> performance impact in O(1) (constant).	2018-11-29 19:27:32 -05:00
Adrien Béraud	4416b91461	update to version 1.8.0	2018-11-28 18:44:03 -05:00
Adrien Béraud	378ae396f0	Merge pull request #351 from hanyf/master A obvious bug on network engine	2018-11-28 12:48:02 -05:00
Yanfei Han	dc4a790a38	A obviouse bug on network engine	2018-11-29 00:47:00 +08:00
Sébastien Blin	07dd8f5648	Merge pull request #350 from savoirfairelinux/feature/cache Feature/cache	2018-11-28 09:22:06 -05:00
Adrien Béraud	f7433a2289	travis: update to xenial	2018-11-26 16:48:00 -05:00
Adrien Béraud	ce4a65ff00	dhtrunner: add APIs for listen<T> with expiration information	2018-11-26 16:34:19 -05:00
Adrien Béraud	092f5cfef0	network_engine: fix warning	2018-11-26 16:34:19 -05:00
Adrien Béraud	12a45e7558	cmake: remove unnecessary flags for lto	2018-11-26 16:30:18 -05:00
Adrien Béraud	18d359a7d2	dht: refresh value if already exists	2018-11-26 01:57:57 -05:00
Adrien Béraud	c6bb67d1ac	dht: add storageRefresh	2018-11-26 01:57:57 -05:00
Adrien Béraud	f4cd7636ce	securedht: use signature cache in storePolicy, editPolicy	2018-11-26 01:57:57 -05:00
Adrien Béraud	2dd4e640c4	dhtrunner: add APIs for listen<T> with expiration information	2018-11-25 23:01:03 -05:00
Adrien Béraud	9a301e3590	Revert "cmake: use CMAKE_INTERPROCEDURAL_OPTIMIZATION for LTO" This reverts commit `b7b5152348`.	2018-11-24 15:32:29 -05:00
Sébastien Blin	aa16b9a235	Merge pull request #348 from savoirfairelinux/feature/probepermanent dht: don't send probe query for non permanent puts	2018-11-23 10:10:35 -05:00
Adrien Béraud	b2542a57df	securedht: cache value signature check/decryption results	2018-11-22 15:43:55 -05:00
Adrien Béraud	00c1be9cdc	logger: use DBG for DEBUG to avoid name conflicts	2018-11-22 15:43:55 -05:00
Adrien Béraud	e414ca89c7	dht: don't send probe query for non permanent puts	2018-11-22 15:43:55 -05:00
Adrien Béraud	4d26537b93	Merge pull request #346 from AmarOk1412/fix/finished dht_proxy_client: fix finished behavior	2018-11-11 22:19:50 -05:00
Sébastien Blin	00a9333de7	dht_proxy_client: fix finished behavior callback_.emplace_back() returns immediately, so, in some case, finished can be set to true even before we execute one callback. Finished should not be used into callback. This patch adds a finished variable only used by the callbacks to avoid this case. This improve the reactivity of the proxy client.	2018-11-11 15:27:57 -05:00
Adrien Béraud	971f96bf95	dhtrunner: improve getId()/getNodeId() documentation.	2018-10-29 16:23:21 -04:00
Adrien Béraud	f9a12a9dca	crypto: use proper size for getRandom() Provided size should be the number of elements, not the number of bytes. Also, if secure_vector where used with sizeof(T) > 1 (currently not the case), there would be memory write to invalid address.	2018-10-29 16:21:51 -04:00
Adrien Béraud	c05a93ae54	proxy server: cleanup	2018-10-29 12:45:52 -04:00
Adrien Béraud	ff83a6daab	proxy: support value expiration	2018-10-29 12:45:52 -04:00
Adrien Béraud	b7b5152348	cmake: use CMAKE_INTERPROCEDURAL_OPTIMIZATION for LTO	2018-10-27 16:38:37 -04:00
Adrien Béraud	a3768d1e50	cmake: use LINK_FLAGS for LTO	2018-10-27 14:57:51 -04:00
Adrien Béraud	94926cea22	proxy server: avoid early session close	2018-10-27 13:21:26 -04:00
Adrien Béraud	82ca2d1993	cmake: use single target_link_libraries for static library	2018-10-27 12:30:40 -04:00
Adrien Béraud	3e148599c5	proxy server: stop get if session is closed	2018-10-27 12:20:00 -04:00
Adrien Béraud	02d5592fdb	cmake: improve restbed, jsoncpp search & usage	2018-10-27 11:51:42 -04:00
Adrien Béraud	328b306fd0	crypto: add documentation	2018-10-27 11:50:26 -04:00
Adrien Béraud	522a89b6f8	dht: use search cache for get ops	2018-10-27 11:49:55 -04:00
Adrien Béraud	897b451f44	crypto: fix build on Win32	2018-10-26 13:16:02 -04:00
Adrien Béraud	ba3a42697c	Merge pull request #331 from Dale-M/master Some additions to automake infrastructure to get clean default build.	2018-09-20 14:24:34 -04:00
Adrien Béraud	5112a322de	dht: reschedule permanent put refresh after request	2018-09-19 01:54:54 -04:00
Adrien Béraud	85961c6a62	storage: properly refresh expiration	2018-09-19 01:54:54 -04:00
Adrien Béraud	1eb1f3d5ac	storageStore: add permanent mode	2018-09-19 01:54:54 -04:00
Adrien Béraud	726c138d90	dht: move storageStore to put	2018-09-19 01:54:54 -04:00
Adrien Béraud	1b6b1ca072	comments: fix typos	2018-09-19 01:54:54 -04:00
Dale Mellor	77da72c603	Some additions to automake infrastructure to get clean default build. Be sure to pull in paths to headers for json and MsgPack dependencies. Be sure to always build the base64 encoder and decoder, as they are needed.	2018-09-17 09:40:07 +01:00
Adrien Béraud	7fb2da7c54	dhtrunner: avoid polling, use pipe to stop select	2018-09-15 17:22:12 -04:00
Sébastien Blin	6721a272ea	Merge pull request #327 from savoirfairelinux/value_cache value_cache: expire value if expiration is now	2018-08-12 10:42:22 -04:00
Adrien Béraud	246af2f065	value_cache: expire value if expiration is now	2018-08-11 21:27:18 -04:00
Adrien Béraud	57aba453b0	Merge pull request #325 from AmarOk1412/siteProxy proxyserver: add stats endpoint for monitoring	2018-08-10 15:05:07 -04:00
Sébastien Blin	f54fcd768f	proxyserver: add stats endpoint for monitoring	2018-08-10 15:03:09 -04:00
Sébastien Blin	5b81476baf	Merge pull request #326 from savoirfairelinux/tests tests: add unit tests for certificate revocation	2018-08-09 17:11:20 -04:00
Adrien Béraud	5b510cadfb	tests: add unit tests for certificate revocation	2018-08-09 14:51:02 -04:00
Adrien Béraud	c794781644	update to version 1.7.4	2018-07-31 13:45:56 -04:00
Adrien Béraud	2ef4999b0f	Merge pull request #322 from AmarOk1412/incorrectOk dhtproxyclient: fix ok when a permanent PUT operations fails	2018-07-31 10:16:05 -04:00
Sébastien Blin	6069fba18f	dhtproxyclient: fix ok when a permanent PUT operations fails	2018-07-31 09:42:23 -04:00
Adrien Béraud	37ddd8c7f9	update to version 1.7.3	2018-07-25 16:40:47 -04:00
Adrien Béraud	23057a3082	unit tests: fix race condition	2018-07-25 11:54:28 -04:00
Adrien Béraud	dbdd3f8f63	Merge pull request #319 from AmarOk1412/fixLD dhtrunner: don't use listeners structure without proxy client enabled	2018-07-25 11:32:55 -04:00
Sébastien Blin	751e79a250	dhtrunner: don't use listeners structure without proxy client enabled The token used to cancel a listener was wrong. Indeed, tokenBGlobal was used to get the linked listener, not to identify the listener inside dhtrunner.	2018-07-25 11:28:46 -04:00
Adrien Béraud	7a961ffc51	Merge pull request #318 from atraczyk/master build: replace WIN32_NATIVE macro with _MSC_VER	2018-07-23 14:01:17 -04:00
Andreas Traczyk	bfada958b9	build: replace WIN32_NATIVE macro with _MSC_VER	2018-07-23 13:45:52 -04:00
Sébastien Blin	78d91a8e0d	Merge pull request #316 from savoirfairelinux/proxy_server dhtrunner: decrease max packet handling delay to 500ms	2018-07-13 14:37:09 -04:00
Adrien Béraud	cd5dc38a57	dhtrunner: decrease max packet handling delay to 500ms Allows for faster recovery in case of CPU contention	2018-07-13 14:29:59 -04:00
Sébastien Blin	1ff2eeb220	Merge pull request #313 from savoirfairelinux/proxy_server Proxy server	2018-06-28 09:16:13 -04:00
Adrien Béraud	ffbb9e23ef	dht: show usage at the end, print max store size	2018-06-27 11:54:15 -04:00
Adrien Béraud	af0a30c9c5	dhtrunner: drop oldest packet if queue is full	2018-06-26 20:41:04 -04:00
Sébastien Blin	97d5e3875d	Merge pull request #312 from savoirfairelinux/dht_performance Dht performance	2018-06-21 11:49:09 -04:00
Adrien Béraud	2509e8bc5c	Revert "search: use pointer to SearchNode" This reverts commit `3a41f2ce56`.	2018-06-20 14:50:40 -04:00
Adrien Béraud	e349912c07	search: improve canGet performance	2018-06-20 14:50:40 -04:00