securedht: cache value signature check/decryption results

dht_proxy_client: fix finished behavior

CMakeLists.txt

@@ -1,7 +1,8 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
+cmake_policy (SET CMP0069 NEW)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.2)
+set (opendht_VERSION_MINOR 7.4)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -59,7 +60,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()
@@ -81,13 +82,7 @@ else ()
     add_definitions(-DOPENDHT_LOG=false)
 endif()
 if (OPENDHT_LTO)
-    set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto")
-    if (CMAKE_COMPILER_IS_GNUCC)
-        set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fuse-linker-plugin")
-        set (CMAKE_AR        "gcc-ar")
-        set (CMAKE_NM        "gcc-nm")
-        set (CMAKE_RANLIB    "gcc-ranlib")
-    endif ()
+    set(CMAKE_INTERPROCEDURAL_OPTIMIZATION TRUE)
 endif ()
 
 if (MSGPACK_INCLUDE_DIRS)
@@ -99,8 +94,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 +241,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 +260,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 ()
 

MSVC/opendht.vcxproj

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

cmake/FindRestbed.cmake

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

configure.ac

@@ -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_patch_version], 4)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/callbacks.h

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

include/opendht/crypto.h

@@ -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;
 
@@ -368,6 +377,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 +414,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 +530,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;
     }

include/opendht/dht.h

@@ -409,7 +409,8 @@ 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);
     void expireStore();
     void expireStorage(InfoHash h);
     void expireStore(decltype(store)::iterator);
@@ -464,7 +465,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 = {});

include/opendht/dht_interface.h

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

include/opendht/dht_proxy_server.h

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

include/opendht/dhtrunner.h

@@ -248,8 +248,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 +422,7 @@ private:
      */
     void tryBootstrapContinuously();
 
+    void stopNetwork();
     void startNetwork(const SockAddr sin4, const SockAddr sin6);
     time_point loop_();
 
@@ -486,6 +493,7 @@ private:
                status6 {NodeStatus::Disconnected};
     StatusCallback statusCb {nullptr};
 
+    int stop_writefd {-1};
     int s4 {-1}, s6 {-1};
     SockAddr bound4 {};
     SockAddr bound6 {};

include/opendht/log_enable.h

@@ -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
     }

include/opendht/rng.h

@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {
 
-#ifndef WIN32_NATIVE
+#ifndef _MSC_VER
 #ifdef _WIN32
 
 /**

include/opendht/securedht.h

@@ -310,11 +310,9 @@ public:
         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);
     }
 

include/opendht/value.h

@@ -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.
  *
@@ -588,6 +590,14 @@ struct OPENDHT_PUBLIC Value
      * Hold encrypted version of the data.
      */
     Blob cypher {};
+
+private:
+    friend class SecureDht;
+    /* Cache for crypto ops */
+    bool signatureChecked {false};
+    bool signatureValid {false};
+    bool decrypted {false};
+    Sp<Value> decryptedValue {};
 };
 
 using ValuesExport = std::pair<InfoHash, Blob>;

src/Makefile.am

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

src/callbacks.cpp

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

src/dht.cpp

@@ -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;
@@ -2346,7 +2383,10 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
     }
 
     auto s = store.find(hash);
-    if (s != store.end() and s->second.refresh(now, vid)) {
+    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, hash));
         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",

src/dht_proxy_client.cpp

@@ -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);
                 }
@@ -706,11 +708,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_();
@@ -951,11 +954,11 @@ DhtProxyClient::restartListeners()
     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);
             }

src/dht_proxy_server.cpp

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

src/dhtrunner.cpp

@@ -164,7 +164,7 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    running_network = false;
+    stopNetwork();
     running = false;
     cv.notify_all();
     bootstrap_cv.notify_all();
@@ -413,7 +413,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 +458,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 +469,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 +541,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 +574,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 +599,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 +642,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 +650,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 +957,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();

src/network_engine.cpp

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

src/search.h

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

src/securedht.cpp

@@ -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);
     }
 }
@@ -139,7 +139,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 +147,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 +164,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 +172,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 +187,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 +205,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;
@@ -235,12 +235,17 @@ SecureDht::checkValue(const Sp<Value>& v)
 #endif
             return {};
         }
+        if (v->decrypted) {
+            return v->decryptedValue;
+        }
+        v->decrypted = true;
         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));
+                v->decryptedValue = std::make_shared<Value>(std::move(decrypted_val));
+                return v->decryptedValue;
             }
             // Ignore values belonging to other people
         } catch (const std::exception& e) {
@@ -249,7 +254,12 @@ SecureDht::checkValue(const Sp<Value>& v)
     }
     // Check signed values
     else if (v->isSigned()) {
+        if (v->signatureChecked) {
+            return v->signatureValid ? v : Sp<Value>{};
+        }
+        v->signatureChecked = true;
         if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
+            v->signatureValid = true;
             nodesPubKeys_[v->owner->getId()] = v->owner;
             return v;
         }
@@ -327,14 +337,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 +365,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, {});

src/storage.h

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

src/value_cache.h

@@ -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 {};

tests/cryptotester.cpp

@@ -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() {
 

tests/cryptotester.h

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

tests/dhtproxytester.cpp

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

tools/Makefile.am

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

tools/dhtnode.cpp

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

tools/proxy_node.html

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