dht: move socket to SearchNode

dht: add error handler for sendRefreshValue
network engine: complete request in case of error
2026-11-08 20:37:23 +00:00 · 2020-06-12 02:51:08 -04:00 · 2020-06-12 02:51:08 -04:00 · 2020-06-12 02:51:08 -04:00 · 2020-06-12 02:51:08 -04:00 · 2020-06-11 15:10:48 -04:00
38 changed files with 1042 additions and 605 deletions
@@ -16,13 +16,12 @@ env:

 before_install:
  - |
-    # non llvm builds
-    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]]; then
+    if [[ "$OPENDHT_TEST_JOB" == *"opendht.classic"* ]]; then
+      docker pull aberaud/opendht-deps-bionic;
+    fi
+    if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclient"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
      docker pull aberaud/opendht-deps;
    fi
-
-  - |
-    # classic build
    if [[ "$OPENDHT_TEST_JOB" == *"opendht.llvm"* ]]; then
      docker pull aberaud/opendht-deps-llvm
    fi
@@ -4,7 +4,7 @@ include(CMakeDependentOption)

 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 0.0)
+set (opendht_VERSION_MINOR 1.3)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -37,7 +37,7 @@ option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentati
 # Dependencies
 list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
 find_package (Threads)
-find_package (PkgConfig)
+find_package (PkgConfig REQUIRED)
 find_package (GnuTLS 3.3 REQUIRED)
 pkg_search_module (Nettle nettle)
 find_package (Msgpack 1.2 REQUIRED)
@@ -45,7 +45,6 @@ if (OPENDHT_TOOLS)
    find_package (Readline 6 REQUIRED)
 endif ()
 if (NOT OPENDHT_ARGON2)
-    find_package(PkgConfig)
    pkg_search_module(argon2 libargon2)
    if (argon2_FOUND)
        message("-- Found Argon2: " ${argon2_LIBRARY_DIRS} " (found version \"" ${argon2_VERSION} "\")")
@@ -300,9 +299,9 @@ if (OPENDHT_STATIC)
    endif ()
    target_link_libraries(opendht-static
        PRIVATE  ${argon2_LIBRARIES}
-        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
-               ${Jsoncpp_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
-               ${OPENSSL_LIBRARIES})
+        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_STATIC_LIBRARIES}
+               ${Jsoncpp_STATIC_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
+               ${OPENSSL_STATIC_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -380,7 +379,7 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO

 # Unit tests
 if (OPENDHT_TESTS)
-    FIND_PACKAGE(Cppunit REQUIRED)
+    pkg_search_module(Cppunit REQUIRED cppunit)
    # unit testing
    list (APPEND test_FILES
        tests/infohashtester.h
@@ -412,7 +411,8 @@ if (OPENDHT_TESTS)
        tests/tests_runner.cpp
        ${test_FILES}
    )
-    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${CPPUNIT_INCLUDE_DIRS})
+    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${Cppunit_INCLUDE_DIR})
+    target_link_directories(opendht_unit_tests PRIVATE ${Cppunit_LIBRARY_DIRS})
    if (OPENDHT_SHARED)
        target_link_libraries(opendht_unit_tests opendht)
    else ()
@@ -420,7 +420,7 @@ if (OPENDHT_TESTS)
    endif ()
    target_link_libraries(opendht_unit_tests
       ${CMAKE_THREAD_LIBS_INIT}
-       ${CPPUNIT_LIBRARIES}
+       ${Cppunit_LIBRARIES}
       ${GNUTLS_LIBRARIES}
       ${Jsoncpp_LIBRARIES}
    )
@@ -1,28 +0,0 @@
-INCLUDE(FindPkgConfig)
-PKG_CHECK_MODULES(PC_CPPUNIT "cppunit")
-
-FIND_PATH(CPPUNIT_INCLUDE_DIRS
-    NAMES cppunit/TestCase.h
-    HINTS ${PC_CPPUNIT_INCLUDE_DIR}
-    ${CMAKE_INSTALL_PREFIX}/include
-    PATHS
-    /usr/local/include
-    /usr/include
-)
-
-FIND_LIBRARY(CPPUNIT_LIBRARIES
-    NAMES cppunit
-    HINTS ${PC_CPPUNIT_LIBDIR}
-    ${CMAKE_INSTALL_PREFIX}/lib
-    ${CMAKE_INSTALL_PREFIX}/lib64
-    PATHS
-    ${CPPUNIT_INCLUDE_DIRS}/../lib
-    /usr/local/lib
-    /usr/lib
-)
-
-LIST(APPEND CPPUNIT_LIBRARIES ${CMAKE_DL_LIBS})
-
-INCLUDE(FindPackageHandleStandardArgs)
-FIND_PACKAGE_HANDLE_STANDARD_ARGS(CPPUNIT DEFAULT_MSG CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)
-MARK_AS_ADVANCED(CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 0)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_minor_version], 1)
+m4_define([opendht_patch_version], 3)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -1,33 +1,20 @@
-FROM ubuntu:16.04
+FROM ubuntu:20.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+
 RUN apt-get update && apt-get install -y \
-        build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
-        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libssl-dev \
+        dialog apt-utils \
+    && apt-get clean \
+    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
+
+RUN apt-get update && apt-get install -y \
+        build-essential pkg-config cmake git wget \
+        autotools-dev autoconf \
+        cython3 python3-dev python3-setuptools \
+        libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
+        libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
+        libssl-dev libfmt-dev libhttp-parser-dev libasio-dev libmsgpack-dev \
    && apt-get clean

-RUN echo "*** Installing libfmt ***" \
-    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
-    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
-    && make -j8 && make install \
-    && cd ../ && rm -rf fmt*
-
-# libhttp-parser-dev is 2.1-2 which is too old
-RUN echo "*** Building nodejs/http_parser dependency ***" \
-    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
-    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 install \
-    && cd ../ && rm -rf http-parser*
-
-# libasio-dev (1.10.6-3) is too old
-RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
-    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
-    && make install \
-    && cd ../../ && rm -rf asio*
-
 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -38,19 +25,3 @@ RUN echo "*** Downloading RESTinio ***" \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
    && make -j8 && make install \
    && cd ../../ && rm -rf restinio*
-
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-    && tar -xzf msgpack-2.1.5.tar.gz \
-    && cd msgpack-2.1.5 && mkdir build && cd build \
-    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-    && make -j8 && make install \
-    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
@@ -3,11 +3,13 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>

 RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
 RUN apt-get update && apt-get install -y \
-        apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
+        apt-transport-https build-essential pkg-config git wget libncurses5-dev libreadline-dev nettle-dev \
        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev \
    && apt-get clean

+RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
+
 # libasio-dev (1.10) is too old
 RUN echo "** Building a recent version of asio ***" \
    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
@@ -1,40 +1,21 @@
-FROM ubuntu:16.04
+FROM ubuntu:20.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN apt-get update && apt-get install -y \
+        dialog apt-utils \
+    && apt-get clean \
+    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
+
 RUN apt-get update \
-    && apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev \
+    && apt-get install -y llvm llvm-dev clang make cmake pkg-config git wget libncurses5-dev libreadline-dev \
       nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
       python3-setuptools libcppunit-dev python3-pip \
-       autotools-dev autoconf libssl-dev \
+       autotools-dev autoconf libssl-dev libargon2-dev \
+       libfmt-dev libhttp-parser-dev libasio-dev \
    && apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean

 ENV CC cc
 ENV CXX c++

-RUN echo "*** Installing libfmt ***" \
-    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
-    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
-    && make -j8 && make install \
-    && cd ../ && rm -rf fmt*
-
-# libhttp-parser-dev is 2.1-2 which is too old
-RUN echo "*** Building nodejs/http_parser dependency ***" \
-    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
-    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 install \
-    && cd ../ && rm -rf http-parser*
-
-# libasio-dev (1.10.6-3) is too old
-RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
-    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
-    && make install \
-    && cd ../../ && rm -rf asio*
-
-#patch for https://github.com/Stiffstream/restinio-conan-example/issues/2
-RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
-
 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -45,18 +26,3 @@ RUN echo "*** Downloading RESTinio ***" \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
    && make -j8 && make install \
    && cd ../../ && rm -rf restinio*
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-    && tar -xzf msgpack-2.1.5.tar.gz \
-    && cd msgpack-2.1.5 && mkdir build && cd build \
-    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-    && make -j8 && make install \
-    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps-llvm
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN git clone https://github.com/savoirfairelinux/opendht.git \
+	&& cd opendht && mkdir build && cd build \
+	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht
@@ -1,6 +1,8 @@
-FROM aberaud/opendht-deps
+FROM aberaud/opendht-deps-bionic
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>

+RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
+
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
 	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \
@@ -82,6 +82,8 @@ struct OPENDHT_PUBLIC NodeInfo {
    NodeStats ipv4 {};
    NodeStats ipv6 {};
    size_t ongoing_ops {0};
+    in_port_t bound4 {0};
+    in_port_t bound6 {0};

 #ifdef OPENDHT_JSONCPP
    /**
@@ -371,6 +371,8 @@ private:
    Dht(const Dht&) = delete;
    Dht& operator=(const Dht&) = delete;

+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
+
    InfoHash myid {};

    uint64_t secret {};
@@ -412,7 +414,6 @@ private:
    std::map<size_t, std::tuple<size_t, size_t, size_t>> listeners {};
    size_t listener_token {1};

-    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};

    // timing
    Scheduler scheduler;
@@ -445,7 +446,7 @@ private:
    void reportedAddr(const SockAddr&);

    // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
    bool storageErase(const InfoHash& id, Value::Id vid);
    bool storageRefresh(const InfoHash& id, Value::Id vid);
@@ -600,7 +601,8 @@ private:
            const InfoHash& hash,
            const Blob& token,
            size_t socket_id,
-            const Query& query);
+            const Query& query,
+            int version = 0);
    void onListenDone(const Sp<Node>& status,
            net::RequestAnswer& a,
            Sp<Search>& sr);
@@ -36,6 +36,15 @@
 #include <memory>
 #include <mutex>

+namespace dht {
+enum class PushType {
+    None = 0,
+    Android,
+    iOS
+};
+}
+MSGPACK_ADD_ENUM(dht::PushType)
+
 namespace http {
 class Request;
 struct ListenerSession;
@@ -52,6 +61,13 @@ class DhtRunner;
 using RestRouter = restinio::router::express_router_t<>;
 using RequestStatus = restinio::request_handling_status_t;

+struct ProxyServerConfig {
+    in_port_t port {8000};
+    std::string pushServer {};
+    std::string persistStatePath {};
+    dht::crypto::Identity identity {};
+};
+
 /**
 * Describes the REST API
 */
@@ -66,10 +82,9 @@ public:
     * @note if the server fails to start (if port is already used or reserved),
     * it will fails silently
     */
-    DhtProxyServer(
-       dht::crypto::Identity identity,
-       std::shared_ptr<DhtRunner> dht, in_port_t port = 8000, const std::string& pushServer = "",
-       std::shared_ptr<dht::Logger> logger = {});
+    DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
+        const ProxyServerConfig& config = {},
+        const std::shared_ptr<dht::Logger>& logger = {});

    virtual ~DhtProxyServer();

@@ -83,8 +98,10 @@ public:
    struct ServerStats {
        /** Current number of listen operations */
        size_t listenCount {0};
-        /** Current number of permanent put operations */
+        /** Current number of permanent put operations (hash used) */
        size_t putCount {0};
+        /** Current number of permanent put values */
+        size_t totalPermanentPuts {0};
        /** Current number of push tokens with at least one listen operation */
        size_t pushListenersCount {0};
        /** Average requests per second */
@@ -114,6 +131,7 @@ public:
            Json::Value result;
            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
            result["putCount"] = static_cast<Json::UInt64>(putCount);
+            result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
            result["requestRate"] = requestRate;
            if (nodeInfo)
@@ -282,17 +300,17 @@ private:
     * @param key of the device
     * @param json, the content to send
     */
-    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid, bool highPriority);
+    void sendPushNotification(const std::string& key, Json::Value&& json, PushType type, bool highPriority);

    /**
     * Send push notification with an expire timeout.
     * @param ec
     * @param pushToken
     * @param json
-     * @param isAndroid
+     * @param type
     */
    void handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
-                                      std::function<Json::Value()> json, const bool isAndroid);
+                                      std::function<Json::Value()> json, PushType type);

    /**
     * Remove a push listener between a client and a hash
@@ -309,6 +327,12 @@ private:
    void handlePrintStats(const asio::error_code &ec);
    void updateStats();

+    template <typename Os>
+    void saveState(Os& stream);
+
+    template <typename Is>
+    void loadState(Is& is, size_t size);
+
    using clock = std::chrono::steady_clock;
    using time_point = clock::time_point;

@@ -316,6 +340,9 @@ private:
    std::shared_ptr<DhtRunner> dht_;
    Json::StreamWriterBuilder jsonBuilder_;
    Json::CharReaderBuilder jsonReaderBuilder_;
+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
+
+    std::string persistPath_;

    // http server
    std::thread serverThread_;
@@ -344,6 +371,7 @@ private:
    struct PushSessionContext {
        std::mutex lock;
        std::string sessionId;
+        PushSessionContext(const std::string& id) : sessionId(id) {}
    };
    struct PermanentPut {
        time_point expiration;
@@ -353,9 +381,32 @@ private:
        std::unique_ptr<asio::steady_timer> expireTimer;
        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
        Sp<Value> value;
+        PushType type;
+
+        template <typename Packer>
+        void msgpack_pack(Packer& p) const
+        {
+            p.pack_map(2 + (sessionCtx ? 1 : 0) + (clientId.empty() ? 0 : 1) + (type == PushType::None ? 0 : 2));
+            p.pack("value"); p.pack(value);
+            p.pack("exp"); p.pack(to_time_t(expiration));
+            if (not clientId.empty()) {
+                p.pack("cid"); p.pack(clientId);
+            }
+            if (sessionCtx) {
+                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                p.pack("sid"); p.pack(sessionCtx->sessionId);
+            }
+            if (type != PushType::None) {
+                p.pack("t"); p.pack(type);
+                p.pack("token"); p.pack(pushToken);
+            }
+        }
+
+        void msgpack_unpack(const msgpack::object& o);
    };
    struct SearchPuts {
        std::map<dht::Value::Id, PermanentPut> puts;
+        MSGPACK_DEFINE_ARRAY(puts)
    };
    std::mutex lockSearchPuts_;
    std::map<InfoHash, SearchPuts> puts_;
@@ -367,14 +418,32 @@ private:

 #ifdef OPENDHT_PUSH_NOTIFICATIONS
    struct Listener {
+        time_point expiration;
        std::string clientId;
        std::shared_ptr<PushSessionContext> sessionCtx;
        std::future<size_t> internalToken;
        std::unique_ptr<asio::steady_timer> expireTimer;
        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        PushType type;
+
+        template <typename Packer>
+        void msgpack_pack(Packer& p) const
+        {
+            p.pack_map(sessionCtx ? 4 : 3);
+            p.pack("cid"); p.pack(clientId);
+            p.pack("exp"); p.pack(to_time_t(expiration));
+            if (sessionCtx) {
+                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                p.pack("sid"); p.pack(sessionCtx->sessionId);
+            }
+            p.pack("t"); p.pack(type);
+        }
+
+        void msgpack_unpack(const msgpack::object& o);
    };
    struct PushListener {
        std::map<InfoHash, std::vector<Listener>> listeners;
+        MSGPACK_DEFINE_ARRAY(listeners)
    };
    std::mutex lockPushListeners_;
    std::map<std::string, PushListener> pushListeners_;
@@ -389,7 +389,7 @@ public:
     * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
     * @param cb: Optional callback to receive general state information.
     */
-    void run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context = {});
+    void run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context = {});

    /**
     * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
@@ -237,6 +237,9 @@ public:

    static Hash getRandom();

+    template <typename Rd>
+    static Hash getRandom(Rd&);
+
    template <size_t M>
    OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Hash<M>& h);

@@ -324,6 +327,19 @@ Hash<N>::getRandom()
    return h;
 }

+template <size_t N>
+template <typename Rd>
+Hash<N>
+Hash<N>::getRandom(Rd& rdev)
+{
+    Hash h;
+    std::uniform_int_distribution<uint32_t> rand_int;
+    auto a = reinterpret_cast<uint32_t*>(h.data());
+    auto b = reinterpret_cast<uint32_t*>(h.data() + h.size());
+    std::generate(a, b, std::bind(rand_int, std::ref(rdev)));
+    return h;
+}
+
 struct HexMap : public std::array<std::array<char, 2>, 256> {
    HexMap() {
        for (size_t i=0; i<size(); i++) {
@@ -180,7 +180,8 @@ private:
            const InfoHash&,
            const Blob&,
            Tid,
-            const Query&)> onListen {};
+            const Query&,
+            int)> onListen {};
    /**
     * Called on announce request.
     *
@@ -210,6 +211,7 @@ private:

 public:
    using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
+    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
    using RequestExpiredCb = std::function<void(const Request&, bool)>;

    NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rd, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
@@ -237,9 +239,10 @@ public:
    void clear();

    /**
-     * Sends values (with closest nodes) to a listenner.
+     * Sends values (with closest nodes) to a listener.
     *
-     * @param sa          The address of the listenner.
+     * @deprecated
+     * @param sa          The address of the listener.
     * @param sslen       The length of the sockaddr structure.
     * @param socket_id  The tid to use to write to the request socket.
     * @param hash        The hash key of the value.
@@ -248,13 +251,14 @@ public:
     * @param nodes       The ipv4 closest nodes.
     * @param nodes6      The ipv6 closest nodes.
     * @param values      The values to send.
+     * @param version     If version = 1, a request will be used to answer to the listener
     */
    void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
            std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
-            std::vector<Sp<Value>>&& values, const Query& q);
+            std::vector<Sp<Value>>&& values, const Query& q, int version);

-    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
-    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
+    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
+    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);

    bool isRunning(sa_family_t af) const;
    inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
@@ -358,10 +362,9 @@ public:
                           const InfoHash& hash,
                           const Query& query,
                           const Blob& token,
-                           Sp<Request> previous,
+                           Tid socketId,
                           RequestCb&& on_done,
-                           RequestExpiredCb&& on_expired,
-                           SocketCb&& socket_cb);
+                           RequestExpiredCb&& on_expired);
    /**
     * Send a "announce" request to a given node.
     *
@@ -400,7 +403,26 @@ public:
                                 const Value::Id& vid,
                                 const Blob& token,
                                 RequestCb&& on_done,
+                                 RequestErrorCb&& on_error,
                                 RequestExpiredCb&& on_expired);
+    /**
+     * Send a "update" request to a given node. Used for Listen operations
+     *
+     * @param n           The node.
+     * @param hash        The target hash.
+     * @param values      The values.
+     * @param created     Time id.
+     * @param token       A security token.
+     * @param sid         The socket id.
+     *
+     * @return the request with information concerning its success.
+     */
+    Sp<Request> sendUpdateValues(Sp<Node> n,
+                                 const InfoHash& infohash,
+                                 const std::vector<Sp<Value>>& values,
+                                 time_point created,
+                                 const Blob& token,
+                                 const size_t& sid);

    /**
     * Parses a message and calls appropriate callbacks.
@@ -413,8 +435,8 @@ public:
     */
    void processMessage(const uint8_t *buf, size_t buflen, SockAddr addr);

-    Sp<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
-        auto n = cache.getNode(myid, addr, scheduler.time(), 0);
+    Sp<Node> insertNode(const InfoHash& id, const SockAddr& addr) {
+        auto n = cache.getNode(id, addr, scheduler.time(), 0);
        onNewNode(n, 0);
        return n;
    }
@@ -490,19 +512,20 @@ private:
    void sendRequest(const Sp<Request>& request);

    struct MessageStats {
-        unsigned ping    {0};
-        unsigned find    {0};
-        unsigned get     {0};
-        unsigned put     {0};
-        unsigned listen  {0};
-        unsigned refresh {0};
+        unsigned ping         {0};
+        unsigned find         {0};
+        unsigned get          {0};
+        unsigned put          {0};
+        unsigned listen       {0};
+        unsigned refresh      {0};
+        unsigned updateValue  {0};
    };


    // basic wrapper for socket sendto function
    int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);

-    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
    void maintainRxBuffer(Tid tid);

@@ -21,6 +21,8 @@
 #include <random>
 #include <algorithm>
 #include <functional>
+#include <thread>
+#include <stdexcept>

 namespace dht {
 namespace crypto {
@@ -111,11 +113,18 @@ using random_device = std::random_device;
 template<class T = std::mt19937, std::size_t N = T::state_size>
 auto getSeededRandomEngine () -> typename std::enable_if<!!N, T>::type {
    typename T::result_type random_data[N];
-    random_device source;
-    std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
-    std::seed_seq seeds(std::begin(random_data), std::end(random_data));
-    T seededEngine (seeds);
-    return seededEngine;
+    for (unsigned i=0; i<256; i++) {
+        try {
+            random_device source;
+            std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
+            std::seed_seq seeds(std::begin(random_data), std::end(random_data));
+            T seededEngine (seeds);
+            return seededEngine;
+        } catch (...) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+    }
+    throw std::runtime_error("Can't seed random engine");
 }

 }} // dht::crypto
@@ -173,7 +173,13 @@ unpackMsg(Blob b) {

 msgpack::unpacked unpackMsg(Blob b);

-msgpack::object* findMapValue(const msgpack::object& map, const char* key);
-msgpack::object* findMapValue(const msgpack::object& map, const std::string& key);
+msgpack::object* findMapValue(const msgpack::object& map, const char* key, size_t length);
+
+inline msgpack::object* findMapValue(const msgpack::object& map, const char* key) {
+    return findMapValue(map, key, strlen(key));
+}
+inline msgpack::object* findMapValue(const msgpack::object& map, const std::string& key) {
+    return findMapValue(map, key.c_str(), key.size());
+}

 } // namespace dht
@@ -490,7 +490,29 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                    logger_->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] = {
-                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
+                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
+                    [this, ws, node=sn->node, v=a.value, 
+                     onDone, 
+                     onExpired, 
+                     created = a.permanent ? time_point::max() : a.created,
+                     next_refresh_time
+                    ](const net::Request& req, net::DhtProtocolException&& e){
+                        if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
+                            if (logger_)
+                                logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
+                            if (auto sr = ws.lock()) {
+                                if (auto sn = sr->getNode(node)) {
+                                    sn->acked[v->id] = {
+                                        network_engine.sendAnnounceValue(sn->node, sr->id, v, created, sn->token, onDone, onExpired),
+                                        next_refresh_time
+                                    };
+                                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                    return true;
+                                }
+                            }
+                        }
+                        return false;
+                    }, onExpired),
                    next_refresh_time
                };
            } else {
@@ -569,33 +591,30 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
    std::weak_ptr<Search> ws = sr;
    for (const auto& l : sr->listeners) {
        const auto& query = l.second.query;
-        auto list_token = l.first;
-        if (n.getListenTime(query, listenExp) > scheduler.time())
+        
+        auto r = n.listenStatus.find(query);
+        if (n.getListenTime(r, listenExp) > scheduler.time())
            continue;
        // if (logger_)
        //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
        //        sr->id.toString().c_str(), n.node->toString().c_str());

-        auto r = n.listenStatus.find(query);
        if (r == n.listenStatus.end()) {
            r = n.listenStatus.emplace(std::piecewise_construct,
                std::forward_as_tuple(query),
                std::forward_as_tuple(
-                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.get_cb(values, expired);
+                    l.second.get_cb,
+                    l.second.sync_cb,
+                    n.node->openSocket([this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
+                        /* on new values */
+                        if (auto sr = ws.lock()) {
+                            scheduler.edit(sr->nextSearchStep, scheduler.time());
+                            sr->insertNode(node, scheduler.time(), answer.ntoken);
+                            if (auto sn = sr->getNode(node)) {
+                                sn->onValues(query, std::move(answer), types, scheduler);
+                            }
                        }
-                    }
-                }, [ws,list_token] (ListenSyncStatus status) {
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.sync_cb(status);
-                        }
-                    }
-                })).first;
+                    }))).first;
            r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this,ws,query,node=n.node]{
                if (auto sr = ws.lock()) {
                    if (auto sn = sr->getNode(node)) {
@@ -604,8 +623,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                }
            });
        }
-        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, r->second.socketId,
            [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
            { /* on done */
                if (auto sr = ws.lock()) {
@@ -625,16 +643,6 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                        if (auto sn = sr->getNode(req.node))
                            sn->listenStatus.erase(query);
                }
-            },
-            [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
-            { /* on new values */
-                if (auto sr = ws.lock()) {
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    sr->insertNode(node, scheduler.time(), answer.ntoken);
-                    if (auto sn = sr->getNode(node)) {
-                        sn->onValues(query, std::move(answer), types, scheduler);
-                    }
-                }
            }
        );
        // Here the request may have failed and the CachedListenStatus removed
@@ -1201,7 +1209,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
                vals.push_back(v.data);
                Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
-                        std::move(vals), l.second.query);
+                        std::move(vals), l.second.query, l.second.version);
            }
        }
    }
@@ -1259,7 +1267,7 @@ Dht::storageErase(const InfoHash& id, Value::Id vid)
 }

 void
-Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
 {
    const auto& now = scheduler.time();
    auto st = store.find(id);
@@ -1275,9 +1283,9 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
        if (not vals.empty()) {
            network_engine.tellListener(node, socket_id, id, WANT4 | WANT6, makeToken(node->getAddr(), false),
                    dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
-                    std::move(vals), query);
+                    std::move(vals), query, version);
        }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
    }
    else
        l->second.refresh(now, std::forward<Query>(query));
@@ -1312,7 +1320,7 @@ Dht::expireStore(decltype(store)::iterator i)
                            id.toString().c_str(),
                            node_listeners.first->toString().c_str());
                    Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids);
+                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
                }
            }
        }
@@ -1743,7 +1751,7 @@ Dht::Dht() : store(), network_engine(logger_, rd, scheduler, {}) {}

 Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l)
    : DhtInterface(l),
-    myid(config.node_id ? config.node_id : InfoHash::getRandom()),
+    myid(config.node_id ? config.node_id : InfoHash::getRandom(rd)),
    store(),
    store_quota(),
    max_store_keys(config.max_store_size ? (int)config.max_store_size : MAX_HASHES),
@@ -1755,7 +1763,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
            std::bind(&Dht::onPing, this, _1),
            std::bind(&Dht::onFindNode, this, _1, _2, _3),
            std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
-            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
            std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
    persistPath(config.persist_path),
@@ -1785,13 +1793,13 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
    secret = std::uniform_int_distribution<uint64_t>{}(rd);
    rotateSecrets();

+    if (not persistPath.empty())
+        loadState(persistPath);
+
    expire();

    if (logger_)
        logger_->d("DHT node initialised with ID %s", myid.toString().c_str());
-
-    if (not persistPath.empty())
-        loadState(persistPath);
 }

 bool
@@ -1982,14 +1990,19 @@ Dht::onDisconnected()
        return;
    if (logger_)
        logger_->d(myid, "Bootstraping");
-    for (const auto& boootstrap : bootstrap_nodes)
+    for (const auto& boootstrap : bootstrap_nodes) {
        try {
-            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
+            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
+            for (auto& ip : ips) {
+                if (ip.getPort() == 0)
+                    ip.setPort(net::DHT_DEFAULT_PORT);
                pingNode(ip);
+            }
        } catch (const std::exception& e) {
            if (logger_)
                logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
        }
+    }
    if (bootstrapJob)
        bootstrapJob->cancel();
    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
@@ -2065,13 +2078,13 @@ Dht::importValues(const std::vector<ValuesExport>& import)
 {
    const auto& now = scheduler.time();

-    for (const auto& node : import) {
-        if (node.second.empty())
+    for (const auto& value : import) {
+        if (value.second.empty())
            continue;

        try {
            msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*)node.second.data(), node.second.size());
+            msgpack::unpack(msg, (const char*)value.second.data(), value.second.size());
            auto valarr = msg.get();
            if (valarr.type != msgpack::type::ARRAY)
                throw msgpack::type_error();
@@ -2086,15 +2099,15 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                    tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                } catch (const std::exception&) {
                    if (logger_)
-                        logger_->e(node.first, "Error reading value at %s", node.first.toString().c_str());
+                        logger_->e(value.first, "Error reading value at %s", value.first.toString().c_str());
                    continue;
                }
                val_time = std::min(val_time, now);
-                storageStore(node.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                storageStore(value.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
            }
        } catch (const std::exception&) {
            if (logger_)
-                logger_->e(node.first, "Error reading values at %s", node.first.toString().c_str());
+                logger_->e(value.first, "Error reading values at %s", value.first.toString().c_str());
            continue;
        }
    }
@@ -2170,7 +2183,6 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
        if (logger_)
            logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
        node->authError();
-        node->cancelRequest(req);
        for (auto& srp : searches(node->getFamily())) {
            auto& sr = srp.second;
            for (auto& n : sr->nodes) {
@@ -2320,7 +2332,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 }

 net::RequestAnswer
-Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
 {
    if (not hash) {
        if (logger_)
@@ -2336,7 +2348,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
        throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
    }
    Query q = query;
-    storageAddListener(hash, node, socket_id, std::move(q));
+    storageAddListener(hash, node, socket_id, std::move(q), version);
    return {};
 }

@@ -2475,7 +2487,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
                            id.toString().c_str(),
                            node_listeners.first->toString().c_str());
                    Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids);
+                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
                }
            }
        }
@@ -2499,12 +2511,24 @@ Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>
 }


+struct DhtState {
+    unsigned v {1};
+    InfoHash id;
+    std::vector<NodeExport> nodes;
+    std::vector<ValuesExport> values;
+
+    MSGPACK_DEFINE_MAP(v, id, nodes, values)
+};
+
 void
 Dht::saveState(const std::string& path) const
 {
+    DhtState state;
+    state.id = myid;
+    state.nodes = exportNodes();
+    state.values = exportValues();
    std::ofstream file(path);
-    msgpack::pack(file, exportNodes());
-    msgpack::pack(file, exportValues());
+    msgpack::pack(file, state);    
 }

 void
@@ -2530,19 +2554,16 @@ Dht::loadState(const std::string& path)
        // Import nodes
        msgpack::object_handle oh;
        if (pac.next(oh)) {
-            {
-                auto imported_nodes = oh.get().as<std::vector<NodeExport>>();
-                if (logger_)
-                    logger_->d("Importing %zu nodes", imported_nodes.size());
-                for (const auto& node : imported_nodes)
-                    insertNode(node);
-            }
-            if (pac.next(oh)) {
-                auto imported_values = oh.get().as<std::vector<ValuesExport>>();
-                if (logger_)
-                    logger_->d("Importing %zu values", imported_values.size());
-                importValues(imported_values);
-            }
+            auto state = oh.get().as<DhtState>();
+            if (logger_)
+                logger_->d("Importing %zu nodes", state.nodes.size());
+            if (state.id)
+                myid = state.id;
+            std::vector<Sp<Node>> tmpNodes;
+            tmpNodes.reserve(state.nodes.size());
+            for (const auto& node : state.nodes)
+                tmpNodes.emplace_back(network_engine.insertNode(node.id, SockAddr(node.ss, node.sslen)));
+            importValues(state.values);
        }
    } catch (const std::exception& e) {
        if (logger_)
@@ -98,7 +98,7 @@ getRandomSessionId(size_t length = 8) {
    static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
    std::string str(length, 0);
    crypto::random_device rdev;
-    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
+    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 2);
    std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
    return str;
 }
@@ -34,6 +34,7 @@
 #include <functional>
 #include <limits>
 #include <iostream>
+#include <fstream>

 using namespace std::placeholders;
 using namespace std::chrono_literals;
@@ -173,26 +174,73 @@ struct DhtProxyServer::RestRouterTraits : public restinio::default_traits_t
    using connection_state_listener_t = ConnectionListener;
 };

-DhtProxyServer::DhtProxyServer(
-    crypto::Identity identity,
-    std::shared_ptr<DhtRunner> dht, in_port_t port, const std::string& pushServer,
-    std::shared_ptr<Logger> logger
+void
+DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
+{
+    if (auto cid = findMapValue(o, "cid")) {
+        clientId = cid->as<std::string>();
+    }
+    if (auto exp = findMapValue(o, "exp")) {
+        expiration = from_time_t(exp->as<time_t>());
+    }
+    if (auto token = findMapValue(o, "token")) {
+        pushToken = token->as<std::string>();
+    }
+    if (auto sid = findMapValue(o, "sid")) {
+        if (not sessionCtx)
+            sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
+        else
+            sessionCtx->sessionId = sid->as<std::string>();
+    }
+    if (auto t = findMapValue(o, "t")) {
+        type = t->as<PushType>();
+    }
+    if (auto val = findMapValue(o, "value")) {
+        value = std::make_shared<dht::Value>(*val);
+    }
+}
+
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+void
+DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
+{
+    if (auto cid = findMapValue(o, "cid")) {
+        clientId = cid->as<std::string>();
+    }
+    if (auto exp = findMapValue(o, "exp")) {
+        expiration = from_time_t(exp->as<time_t>());
+    }
+    if (auto sid = findMapValue(o, "sid")) {
+        if (not sessionCtx)
+            sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
+        else
+            sessionCtx->sessionId = sid->as<std::string>();
+    }
+    if (auto t = findMapValue(o, "t")) {
+        type = t->as<PushType>();
+    }
+}
+#endif
+
+DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
+        const ProxyServerConfig& config,
+        const std::shared_ptr<dht::Logger>& logger
 )
    :   ioContext_(std::make_shared<asio::io_context>()),
-        dht_(dht), logger_(logger),
+        dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
        printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
        connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
-        pushServer_(pushServer)
+        pushServer_(config.pushServer)
 {
    if (not dht_)
        throw std::invalid_argument("A DHT instance must be provided");

    if (logger_)
-        logger_->d("[proxy:server] [init] running on %i", port);
-    if (not pushServer.empty()){
+        logger_->d("[proxy:server] [init] running on %i", config.port);
+    if (not pushServer_.empty()){
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
        if (logger_)
-            logger_->d("[proxy:server] [init] using push server %s", pushServer.c_str());
+            logger_->d("[proxy:server] [init] using push server %s", pushServer_.c_str());
 #else
        if (logger_)
            logger_->e("[proxy:server] [init] opendht built without push notification support");
@@ -202,9 +250,9 @@ DhtProxyServer::DhtProxyServer(
    jsonBuilder_["commentStyle"] = "None";
    jsonBuilder_["indentation"] = "";

-    if (!pushServer.empty()){
+    if (!pushServer_.empty()){
        // no host delim, assume port only
-        if (pushServer.find(":") == std::string::npos)
+        if (pushServer_.find(":") == std::string::npos)
            pushServer_ =  "localhost:" + pushServer_;
        // define http request destination for push notifications
        pushHostPort_ = splitPort(pushServer_);
@@ -212,7 +260,7 @@ DhtProxyServer::DhtProxyServer(
            logger_->d("Using push server for notifications: %s:%s", pushHostPort_.first.c_str(),
                                                                     pushHostPort_.second.c_str());
    }
-    if (identity.first and identity.second) {
+    if (config.identity.first and config.identity.second) {
        asio::error_code ec;
        // define tls context
        asio::ssl::context tls_context { asio::ssl::context::sslv23 };
@@ -226,13 +274,13 @@ DhtProxyServer::DhtProxyServer(
        SSL_CTX_set_options(tls_context.native_handle(), SSL_OP_NO_RENEGOTIATION); // CVE-2009-3555
 #endif
        // node private key
-        auto key = identity.first->serialize();
+        auto key = config.identity.first->serialize();
        tls_context.use_private_key(asio::const_buffer{key.data(), key.size()},
                                    asio::ssl::context::file_format::pem, ec);
        if (ec)
            throw std::runtime_error("Error setting node's private key: " + ec.message());
        // certificate chain
-        auto certchain = identity.second->toString(true/*chain*/);
+        auto certchain = config.identity.second->toString(true/*chain*/);
        tls_context.use_certificate_chain(asio::const_buffer{certchain.data(), certchain.size()}, ec);
        if (ec)
            throw std::runtime_error("Error setting certificate chain: " + ec.message());
@@ -241,7 +289,7 @@ DhtProxyServer::DhtProxyServer(
        // build http server
        auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraitsTls>();
        addServerSettings(settings);
-        settings.port(port);
+        settings.port(config.port);
        settings.tls_context(std::move(tls_context));
        httpsServer_ = std::make_unique<restinio::http_server_t<RestRouterTraitsTls>>(
            ioContext_,
@@ -258,7 +306,7 @@ DhtProxyServer::DhtProxyServer(
    else {
        auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraits>();
        addServerSettings(settings);
-        settings.port(port);
+        settings.port(config.port);
        httpServer_ = std::make_unique<restinio::http_server_t<RestRouterTraits>>(
            ioContext_,
            std::forward<restinio::run_on_this_thread_settings_t<RestRouterTraits>>(std::move(settings))
@@ -274,8 +322,158 @@ DhtProxyServer::DhtProxyServer(
    dht->forwardAllMessages(true);
    updateStats();
    printStatsTimer_->async_wait(std::bind(&DhtProxyServer::handlePrintStats, this, std::placeholders::_1));
+
+    if (not persistPath_.empty()) {
+        try {
+            std::ifstream stateFile(persistPath_, std::ios::binary | std::ios::ate);
+            if (stateFile) {
+                std::streamsize size = stateFile.tellg();
+                stateFile.seekg(0, std::ios::beg);
+                if (logger_)
+                    logger_->d("Loading proxy state from %.*s (%td bytes)", (int)persistPath_.size(), persistPath_.c_str(), size);
+                loadState(stateFile, size);
+            }
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->e("Error loading state from file: %s", e.what());
+        }
+    }
 }

+template <typename Os>
+void
+DhtProxyServer::saveState(Os& stream) {
+    msgpack::packer<Os> pk(&stream);
+    pk.pack_map(2);
+    {
+        std::lock_guard<std::mutex> lock(lockSearchPuts_);
+        pk.pack("puts");
+        pk.pack(puts_);
+    }
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    {
+        std::lock_guard<std::mutex> lock(lockListener_);
+        pk.pack("pushListeners");
+        pk.pack(pushListeners_);
+    }
+#endif
+}
+
+template <typename Is>
+void
+DhtProxyServer::loadState(Is& is, size_t size) {
+    msgpack::unpacker pac;
+    pac.reserve_buffer(size);
+    if (is.read(pac.buffer(), size)) {
+        pac.buffer_consumed(size);
+
+        msgpack::object_handle oh;
+        while (pac.next(oh)) {
+            if (oh.get().type != msgpack::type::MAP)
+                continue;
+            if (auto puts = findMapValue(oh.get(), "puts")) {
+                std::lock_guard<std::mutex> lock(lockSearchPuts_);
+                puts_ = puts->as<decltype(puts_)>();
+                if (logger_)
+                    logger_->d("Loading %zu persistent puts", puts_.size());
+                for (auto& put : puts_) {
+                    for (auto& pput : put.second.puts) {
+                        pput.second.expireTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration);
+                        pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this,
+                                                std::placeholders::_1, put.first, pput.first));
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+                        if (not pput.second.pushToken.empty()) {
+                            auto jsonProvider = [infoHash=put.first.toString(), clientId=pput.second.clientId, vid = pput.first, sessionCtx = pput.second.sessionCtx](){
+                                Json::Value json;
+                                json["timeout"] = infoHash;
+                                json["to"] = clientId;
+                                json["vid"] = std::to_string(vid);
+                                if (sessionCtx) {
+                                    std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                    json["s"] = sessionCtx->sessionId;
+                                }
+                                return json;
+                            };
+                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration - proxy::OP_MARGIN);
+                            pput.second.expireNotifyTimer->async_wait(std::bind(
+                                &DhtProxyServer::handleNotifyPushListenExpire, this,
+                                std::placeholders::_1, pput.second.pushToken, std::move(jsonProvider), pput.second.type));
+                        }
+#endif
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple{}, time_point::max(), true);
+                    }
+                }
+            } else {
+                if (logger_)
+                    logger_->d("No persistent puts in state");
+            }
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+            if (auto listeners = findMapValue(oh.get(), "pushListeners")) {
+                std::lock_guard<std::mutex> lock(lockListener_);
+                pushListeners_ = listeners->as<decltype(pushListeners_)>();
+                if (logger_)
+                    logger_->d("Loading %zu push listeners", pushListeners_.size());
+                for (auto& pushListener : pushListeners_) {
+                    for (auto& listeners : pushListener.second.listeners) {
+                        for (auto& listener : listeners.second) {
+                            listener.internalToken = dht_->listen(listeners.first,
+                                [this, infoHash=listeners.first, pushToken=pushListener.first, type=listener.type, clientId=listener.clientId, sessionCtx = listener.sessionCtx]
+                                (const std::vector<std::shared_ptr<Value>>& values, bool expired) {
+                                    // Build message content
+                                    Json::Value json;
+                                    json["key"] = infoHash.toString();
+                                    json["to"] = clientId;
+                                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
+                                    {
+                                        std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                        json["s"] = sessionCtx->sessionId;
+                                    }
+                                    if (expired and values.size() < 2){
+                                        std::stringstream ss;
+                                        for(size_t i = 0; i < values.size(); ++i){
+                                            if(i != 0) ss << ",";
+                                            ss << values[i]->id;
+                                        }
+                                        json["exp"] = ss.str();
+                                    }
+                                    auto maxPrio = 1000u;
+                                    for (const auto& v : values)
+                                        maxPrio = std::min(maxPrio, v->priority);
+                                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0);
+                                    return true;
+                                }
+                            );
+                            // expire notify
+                            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), listener.expiration - proxy::OP_MARGIN);
+                            auto jsonProvider = [infoHash = listeners.first.toString(), clientId = listener.clientId, sessionCtx = listener.sessionCtx](){
+                                Json::Value json;
+                                json["timeout"] = infoHash;
+                                json["to"] = clientId;
+                                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                json["s"] = sessionCtx->sessionId;
+                                return json;
+                            };
+                            listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
+                                                                std::placeholders::_1, pushListener.first, std::move(jsonProvider), listener.type));
+                            // cancel push listen
+                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), listener.expiration);
+                            listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this,
+                                                            std::placeholders::_1, pushListener.first, listeners.first, listener.clientId));
+                        }
+                    }
+                }
+            } else {
+                if (logger_)
+                    logger_->d("No push listeners in state");
+            }
+#endif
+        }
+        if (logger_)
+            logger_->d("loading ended");
+    }
+}
+
+
 asio::io_context&
 DhtProxyServer::io_context() const
 {
@@ -284,6 +482,12 @@ DhtProxyServer::io_context() const

 DhtProxyServer::~DhtProxyServer()
 {
+    if (not persistPath_.empty()) {
+        if (logger_)
+            logger_->d("Saving proxy state to %.*s", (int)persistPath_.size(), persistPath_.c_str());
+        std::ofstream stateFile(persistPath_, std::ios::binary);
+        saveState(stateFile);
+    }
    if (dht_) {
        std::lock_guard<std::mutex> lock(lockListener_);
        for (auto& l : listeners_) {
@@ -359,6 +563,10 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
    stats.pushListenersCount = pushListeners_.size();
 #endif
+    stats.totalPermanentPuts = 0;
+    std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
+        stats.totalPermanentPuts += put.second.puts.size();
+    });
    stats.putCount = puts_.size();
    stats.listenCount = listeners_.size();
    stats.nodeInfo = std::move(info);
@@ -585,12 +793,11 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
                          restinio::router::route_params_t params)
 {
    requestNum_++;
-
-    InfoHash infoHash(params["hash"].to_string());
-    if (!infoHash)
-        infoHash = InfoHash::get(params["hash"].to_string());
-
    try {
+        InfoHash infoHash(params["hash"].to_string());
+        if (!infoHash)
+            infoHash = InfoHash::get(params["hash"].to_string());
+
        std::string err;
        Json::Value r;
        auto* char_data = reinterpret_cast<const char*>(request->body().data());
@@ -607,117 +814,115 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
            response.set_body(RESP_MSG_NO_TOKEN);
            return response.done();
        }
-        auto isAndroid = root["platform"].asString() == "android";
+        auto type = root["platform"].asString() == "android" ? PushType::Android : PushType::iOS;
        auto clientId = root["client_id"].asString();
        auto sessionId = root["session_id"].asString();

        if (logger_)
            logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
-        // ================ Search for existing listener ===================
-        // start the timer
-        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
-        std::lock_guard<std::mutex> lock(lockPushListeners_);

        // Insert new or return existing push listeners of a token
+        std::lock_guard<std::mutex> lock(lockPushListeners_);
        auto& pushListener = pushListeners_[pushToken];
        auto& pushListeners = pushListener.listeners[infoHash];

-        for (auto &listener: pushListeners) {
-            // Found -> Resubscribe
-            if (listener.clientId == clientId) {
-                if (logger_)
-                    logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
-                // Reset timers
-                listener.expireTimer->expires_at(timeout);
-                listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
-                {
-                    std::lock_guard<std::mutex> l(listener.sessionCtx->lock);
-                    listener.sessionCtx->sessionId = sessionId;
-                }
-                // Send response header
-                auto response = std::make_shared<ResponseByPartsBuilder>(
-                    initHttpResponse(request->create_response<ResponseByParts>()));
-                response->flush();
-                if (!root["refresh"].asBool()) {
-                    // No Refresh
-                    dht_->get(infoHash, [this, response](const Sp<Value>& value){
-                        auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
-                        response->append_chunk(output);
-                        response->flush();
-                        return true;
-                    },
-                    [response] (bool){
-                        response->done();
-                    });
-                } else {
-                    // Refresh
-                    response->append_chunk("{}\n");
-                    response->done();
-                }
-                return restinio::request_handling_status_t::accepted;
-            }
+        auto listIt = std::find_if(pushListeners.begin(), pushListeners.end(), [&](const Listener& l) {
+            return l.clientId == clientId;
+        });
+        bool newListener = listIt == pushListeners.end();
+        if (newListener) {
+            pushListeners.emplace_back(Listener{});
+            listIt = std::prev(pushListeners.end());
+            listIt->clientId = clientId;
+            listIt->sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+        } else {
+            std::lock_guard<std::mutex> l(listIt->sessionCtx->lock);
+            listIt->sessionCtx->sessionId = sessionId;
        }
-        // =========== No existing listener for an infoHash ============
-        // Add new listener to list of listeners
-        pushListeners.emplace_back(Listener{});
-        auto &listener = pushListeners.back();
-        listener.clientId = clientId;
-        listener.sessionCtx = std::make_shared<PushSessionContext>();
-        listener.sessionCtx->sessionId = sessionId;
+        auto& listener = *listIt;

-        // Add listen on dht
-        listener.internalToken = dht_->listen(infoHash,
-            [this, infoHash, pushToken, isAndroid, clientId, sessionCtx = listener.sessionCtx]
-            (const std::vector<std::shared_ptr<Value>>& values, bool expired){
-                // Build message content
-                Json::Value json;
-                json["key"] = infoHash.toString();
-                json["to"] = clientId;
-                json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
-                {
-                    std::lock_guard<std::mutex> l(sessionCtx->lock);
-                    json["s"] = sessionCtx->sessionId;
-                }
-                if (expired and values.size() < 2){
-                    std::stringstream ss;
-                    for(size_t i = 0; i < values.size(); ++i){
-                        if(i != 0) ss << ",";
-                        ss << values[i]->id;
-                    }
-                    json["exp"] = ss.str();
-                }
-                auto maxPrio = 1000u;
-                for (const auto& v : values)
-                    maxPrio = std::min(maxPrio, v->priority);
-                sendPushNotification(pushToken, std::move(json), isAndroid, !expired and maxPrio == 0);
-                return true;
-            }
-        );
-        // expire notify
-        if (!listener.expireNotifyTimer)
-            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
-        else
+        // Expiration
+        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+        listener.expiration = timeout;
+        listener.type = type;
+        if (listener.expireNotifyTimer)
            listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
-        auto jsonProvider = [infoHash, clientId, sessionCtx = listener.sessionCtx](){
+        else
+            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
+        auto jsonProvider = [h=infoHash.toString(), clientId, sessionCtx = listener.sessionCtx](){
            Json::Value json;
-            json["timeout"] = infoHash.toString();
+            json["timeout"] = h;
            json["to"] = clientId;
            std::lock_guard<std::mutex> l(sessionCtx->lock);
            json["s"] = sessionCtx->sessionId;
            return json;
        };
        listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
-                                               std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
-        // cancel push listen
+                                               std::placeholders::_1, pushToken, std::move(jsonProvider), listener.type));
        if (!listener.expireTimer)
            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
        else
            listener.expireTimer->expires_at(timeout);
        listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this,
-                                         std::placeholders::_1, pushToken, infoHash, clientId));
-        auto response = initHttpResponse(request->create_response());
-        response.set_body("{}\n");
-        return response.done();
+                                        std::placeholders::_1, pushToken, infoHash, clientId));
+
+        // Send response
+        if (not newListener) {
+            if (logger_)
+                logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
+            // Send response header
+            auto response = std::make_shared<ResponseByPartsBuilder>(initHttpResponse(request->create_response<ResponseByParts>()));
+            response->flush();
+            if (!root["refresh"].asBool()) {
+                // No Refresh
+                dht_->get(infoHash, [this, response](const Sp<Value>& value){
+                    auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
+                    response->append_chunk(output);
+                    response->flush();
+                    return true;
+                },
+                [response] (bool){
+                    response->done();
+                });
+            } else {
+                // Refresh
+                response->append_chunk("{}\n");
+                return response->done();
+            }
+        } else {
+            // =========== No existing listener for an infoHash ============
+            // Add listen on dht
+            listener.internalToken = dht_->listen(infoHash,
+                [this, infoHash, pushToken, type, clientId, sessionCtx = listener.sessionCtx]
+                (const std::vector<std::shared_ptr<Value>>& values, bool expired){
+                    // Build message content
+                    Json::Value json;
+                    json["key"] = infoHash.toString();
+                    json["to"] = clientId;
+                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
+                    {
+                        std::lock_guard<std::mutex> l(sessionCtx->lock);
+                        json["s"] = sessionCtx->sessionId;
+                    }
+                    if (expired and values.size() < 2){
+                        std::stringstream ss;
+                        for(size_t i = 0; i < values.size(); ++i){
+                            if(i != 0) ss << ",";
+                            ss << values[i]->id;
+                        }
+                        json["exp"] = ss.str();
+                    }
+                    auto maxPrio = 1000u;
+                    for (const auto& v : values)
+                        maxPrio = std::min(maxPrio, v->priority);
+                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0);
+                    return true;
+                }
+            );
+            auto response = initHttpResponse(request->create_response());
+            response.set_body("{}\n");
+            return response.done();
+        }
    }
    catch (...) {
        return serverError(*request);
@@ -766,7 +971,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request,

 void
 DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
-                                             std::function<Json::Value()> jsonProvider, const bool isAndroid)
+                                             std::function<Json::Value()> jsonProvider, PushType type)
 {
    if (ec == asio::error::operation_aborted)
        return;
@@ -776,7 +981,7 @@ DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const s
    }
    if (logger_)
        logger_->d("[proxy:server] [subscribe] sending put refresh to %s token", pushToken.c_str());
-    sendPushNotification(pushToken, jsonProvider(), isAndroid, false);
+    sendPushNotification(pushToken, jsonProvider(), type, false);
 }

 void
@@ -818,7 +1023,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
 }

 void
-DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid, bool highPriority)
+DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, PushType type, bool highPriority)
 {
    if (pushServer_.empty())
        return;
@@ -840,7 +1045,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
        Json::Value tokens(Json::arrayValue);
        tokens[0] = token;
        notification["tokens"] = std::move(tokens);
-        notification["platform"] = isAndroid ? 2 : 1;
+        notification["platform"] = type == PushType::Android ? 2 : 1;
        notification["data"] = std::move(json);
        notification["priority"] = highPriority ? "high" : "normal";
        notification["time_to_live"] = 600;
@@ -889,6 +1094,7 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoH
    }
    if (logger_)
        logger_->d("[proxy:server] [put %s] cancel permament put %i", key.toString().c_str(), vid);
+    std::lock_guard<std::mutex> lock(lockSearchPuts_);
    auto sPuts = puts_.find(key);
    if (sPuts == puts_.end())
        return;
@@ -898,6 +1104,8 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoH
        return;
    if (dht_)
        dht_->cancelPut(key, vid);
+    if (put->second.expireTimer)
+        put->second.expireTimer->cancel();
    if (put->second.expireNotifyTimer)
        put->second.expireNotifyTimer->cancel();
    sPutsMap.erase(put);
@@ -941,13 +1149,13 @@ DhtProxyServer::put(restinio::request_handle_t request,
                    platform = pVal["platform"].asString();
                    sessionId = pVal["session_id"].asString();
                }
-                std::unique_lock<std::mutex> lock(lockSearchPuts_);
+                std::lock_guard<std::mutex> lock(lockSearchPuts_);
                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
                auto& sPuts = puts_[infoHash];
                if (value->id == Value::INVALID_ID) {
                    for (auto& pp : sPuts.puts) {
-                        if (pp.second.pushToken == pushToken 
-                            and pp.second.clientId == clientId 
+                        if (pp.second.pushToken == pushToken
+                            and pp.second.clientId == clientId
                            and pp.second.value->contentEquals(*value))
                        {
                            pp.second.expireTimer->expires_at(timeout);
@@ -955,32 +1163,36 @@ DhtProxyServer::put(restinio::request_handle_t request,
                                                        std::placeholders::_1, infoHash, pp.second.value->id));
                            if (not sessionId.empty()) {
                                if (not pp.second.sessionCtx)
-                                    pp.second.sessionCtx = std::make_shared<PushSessionContext>();
-                                std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
-                                pp.second.sessionCtx->sessionId = sessionId;
+                                    pp.second.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+                                else {
+                                    std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
+                                    pp.second.sessionCtx->sessionId = sessionId;
+                                }
                            }
                            auto response = initHttpResponse(request->create_response());
                            response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n");
                            return response.done();
                        }
                    }
+                    value->id = std::uniform_int_distribution<Value::Id>{1}(rd);
                }

                auto vid = value->id;
                auto& pput = sPuts.puts[vid];
                pput.value = value;
+                pput.expiration = timeout;
                if (not pput.expireTimer) {
                    auto &ctx = io_context();
                    // cancel permanent put
                    pput.expireTimer = std::make_unique<asio::steady_timer>(ctx, timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                    if (not pushToken.empty()){
+                        bool isAndroid = platform == "android";
                        pput.pushToken = pushToken;
                        pput.clientId = clientId;
-                        pput.sessionCtx = std::make_shared<PushSessionContext>();
-                        pput.sessionCtx->sessionId = sessionId;
+                        pput.type = isAndroid ? PushType::Android : PushType::iOS;
+                        pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
                        // notify push listen expire
-                        bool isAndroid = platform == "android";
                        auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx](){
                            Json::Value json;
                            json["timeout"] = infoHash.toString();
@@ -997,15 +1209,17 @@ DhtProxyServer::put(restinio::request_handle_t request,
                            pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
                        pput.expireNotifyTimer->async_wait(std::bind(
                            &DhtProxyServer::handleNotifyPushListenExpire, this,
-                            std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
+                            std::placeholders::_1, pushToken, std::move(jsonProvider), pput.type));
                    }
 #endif
                } else {
                    if (not sessionId.empty()) {
                        if (not pput.sessionCtx)
-                            pput.sessionCtx = std::make_shared<PushSessionContext>();
-                        std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
-                        pput.sessionCtx->sessionId = sessionId;
+                            pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+                        else {
+                            std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
+                            pput.sessionCtx->sessionId = sessionId;
+                        }
                    }
                    pput.expireTimer->expires_at(timeout);
                    if (pput.expireNotifyTimer)
@@ -32,6 +32,8 @@
 #include "dht_proxy_client.h"
 #endif

+#include <fstream>
+
 namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
@@ -90,16 +92,46 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Conf
 {
    auto res4 = SockAddr::resolve(ip4, service);
    auto res6 = SockAddr::resolve(ip6, service);
-    run(res4.empty() ? SockAddr() : res4.front(),
-        res6.empty() ? SockAddr() : res6.front(), config, std::move(context));
+    if (res4.empty())
+        res4.emplace_back();
+    if (res6.empty())
+        res6.emplace_back();
+    run(res4.front(), res6.front(), config, std::move(context));
 }

 void
-DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context)
+DhtRunner::run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context)
 {
    if (running == State::Idle) {
+        auto state_path = config.dht_config.node_config.persist_path;
+        if (not state_path.empty()) {
+            state_path += "_port.txt";
+            std::ifstream inConfig(state_path);
+            if (inConfig.is_open()) {
+                in_port_t port;
+                if (inConfig >> port) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                    if (local4.getPort() == 0)
+                        local4.setPort(port);
+                }
+                if (inConfig >> port) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                    if (local6.getPort() == 0)
+                        local6.setPort(port);
+                }
+            }
+        }
+
        if (not context.sock)
            context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
+
+        if (not state_path.empty()) {
+            std::ofstream outConfig(state_path);
+            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
+            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
+        }
        run(config, std::move(context));
    }
 }
@@ -475,6 +507,10 @@ DhtRunner::getNodeInfo() const {
        info.node_id = dht->getNodeId();
        info.ipv4 = dht->getNodesStats(AF_INET);
        info.ipv6 = dht->getNodesStats(AF_INET6);
+        if (auto sock = dht->getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
    }
    info.ongoing_ops = ongoing_ops;
    return info;
@@ -492,6 +528,10 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
        info.node_id = dht.getNodeId();
        info.ipv4 = dht.getNodesStats(AF_INET);
        info.ipv6 = dht.getNodesStats(AF_INET6);
+        if (auto sock = dht.getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
        info.ongoing_ops = ongoing_ops;
        cb(std::move(sinfo));
        opEnded();
@@ -30,8 +30,9 @@ namespace dht {
 struct Listener {
    time_point time;
    Query query;
+    int version;

-    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}
+    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}

    void refresh(time_point t, Query&& q) {
        time = t;
@@ -31,7 +31,8 @@ enum class MessageType {
    Refresh,
    Listen,
    ValueData,
-    ValueUpdate
+    ValueUpdate,
+    UpdateValue
 };

 } /* namespace net */
@@ -48,23 +48,6 @@ static constexpr uint8_t v4prefix[16] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xF
 constexpr unsigned SEND_NODES {8};


-/* Transaction-ids are 4-bytes long, with the first two bytes identifying
- * the kind of request, and the remaining two a sequence number in
- * host order.
- */
-struct TransId final : public std::array<uint8_t, 4> {
-    TransId() { std::fill(begin(), end(), 0); }
-    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
-    TransId(uint32_t id) {
-        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
-    }
-
-    uint32_t toInt() const {
-        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
-    }
-};
-
-
 struct NetworkEngine::PartialMessage {
    SockAddr from;
    time_point start;
@@ -140,11 +123,15 @@ void
 NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
        const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
        std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
-        const Query& query)
+        const Query& query, int version)
 {
    auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
    try {
-        sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        if (version >= 1) {
+            sendUpdateValues(node, hash, values, scheduler.time(), ntoken, socket_id);
+        } else {
+            sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        }
    } catch (const std::overflow_error& e) {
        if (logger_)
            logger_->e("Can't send value: buffer not large enough !");
@@ -152,63 +139,103 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }

 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(config.network?1:0));

-    pk.pack(KEY_U);
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(KEY_REQ_ID); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(KEY_REQ_REFRESHED);
-        pk.pack(values);
-        if (logger_)
-            logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
-    }
+    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));

-    pk.pack(KEY_TID); pk.pack(socket_id);
-    pk.pack(KEY_Y); pk.pack(KEY_R);
+    pk.pack(version >= 1 ? KEY_A : KEY_U);
+        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
+        pk.pack(KEY_REQ_ID); pk.pack(myid);
+        if (version >= 1) {
+            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
+        }
+        if (not token.empty()) {
+            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+        }
+        if (not values.empty()) {
+            pk.pack(KEY_REQ_REFRESHED);
+            pk.pack(values);
+            if (logger_)
+                logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
+        }
+
+    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

+    if (version >= 1) {
+        Tid tid (n->getNewTid());
+
+        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+        pk.pack(KEY_TID); pk.pack(tid);
+
+        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+            Blob(buffer.data(), buffer.data() + buffer.size()),
+            [=](const Request&, ParsedMessage&&) { /* on done */ },
+            [=](const Request&, bool) { /* on expired */ }
+        );
+        sendRequest(req);
+        ++out_stats.updateValue;
+        return;
+    }
+    pk.pack(KEY_TID); pk.pack(socket_id);
+
    // send response
    send(n->getAddr(), buffer.data(), buffer.size());
 }

 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(config.network?1:0));

-    pk.pack(KEY_U);
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(KEY_REQ_ID); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(KEY_REQ_EXPIRED);
-        pk.pack(values);
-        if (logger_)
-            logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
-    }
+    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));

-    pk.pack(KEY_TID); pk.pack(socket_id);
-    pk.pack(KEY_Y); pk.pack(KEY_R);
+    pk.pack(version >= 1 ? KEY_A : KEY_U);
+        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
+        pk.pack(KEY_REQ_ID); pk.pack(myid);
+        if (version >= 1) {
+            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
+        }
+        if (not token.empty()) {
+            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+        }
+        if (not values.empty()) {
+            pk.pack(KEY_REQ_EXPIRED);
+            pk.pack(values);
+            if (logger_)
+                logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
+        }
+
+    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

+    if (version >= 1) {
+        Tid tid (n->getNewTid());
+
+        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+        pk.pack(KEY_TID);     pk.pack(tid);
+
+        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+            Blob(buffer.data(), buffer.data() + buffer.size()),
+            [=](const Request&, ParsedMessage&&) { /* on done */ },
+            [=](const Request&, bool) { /* on expired */ }
+        );
+        sendRequest(req);
+        ++out_stats.updateValue;
+        return;
+    }
+    pk.pack(KEY_TID); pk.pack(socket_id);
+
    // send response
    send(n->getAddr(), buffer.data(), buffer.size());
 }
@@ -533,7 +560,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
            {
                req->last_try = time_point::min();
                req->reply_time = time_point::min();
-                onError(req, DhtProtocolException {msg->error_code});
+                if (not req->setError(DhtProtocolException {msg->error_code}))
+                    onError(req, DhtProtocolException {msg->error_code});
            } else {
                if (logIncoming_)
                    if (logger_)
@@ -545,8 +573,11 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
        case MessageType::Reply:
            if (req) { /* request reply */
                auto& r = *req;
-                if (r.getType() == MessageType::AnnounceValue or r.getType() == MessageType::Listen)
+                if (r.getType() == MessageType::AnnounceValue
+                 or r.getType() == MessageType::Listen
+                 or r.getType() == MessageType::Refresh) {
                    r.node->authSuccess();
+                }
                r.reply_time = scheduler.time();

                deserializeNodes(*msg, from);
@@ -617,18 +648,29 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                if (logIncoming_ and logger_)
                    logger_->d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                ++in_stats.listen;
-                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
                auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                sendListenConfirmation(from, msg->tid);
                break;
            }
+            case MessageType::UpdateValue: {
+                if (logIncoming_ and logger_)
+                    logger_->d(msg->info_hash, node->id, "[node %s] got 'update' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                ++in_stats.updateValue;
+                if (auto rsocket = node->getSocket(msg->socket_id))
+                    rsocket->on_receive(node, std::move(*msg));
+                else if (logger_)
+                    logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                sendListenConfirmation(from, msg->tid);
+                break;
+            }
            default:
                break;
            }
        } catch (const std::overflow_error& e) {
            if (logger_)
                logger_->e("Can't send value: buffer not large enough !");
-        } catch (DhtProtocolException& e) {
+        } catch (const DhtProtocolException& e) {
            sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
        }
    }
@@ -654,7 +696,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf

 Sp<Request>
 NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (node->getNewTid());
+    Tid tid (node->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -663,15 +705,14 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
     pk.pack(KEY_REQ_ID); pk.pack(myid);

    pk.pack(KEY_Q); pk.pack(QUERY_PING);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                              pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&&) {
            if (logger_)
@@ -701,9 +742,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -716,7 +755,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 Sp<Request>
 NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
        RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -732,15 +771,14 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
    }

    pk.pack(KEY_Q); pk.pack(QUERY_FIND);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -762,7 +800,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 Sp<Request>
 NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
        RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -786,15 +824,14 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
      }

    pk.pack(KEY_Q); pk.pack(QUERY_GET);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -896,7 +933,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }

 void
-NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
    msgpack::sbuffer buffer;
    unsigned i=0;
@@ -911,8 +948,7 @@ NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals
                pk.pack(KEY_NETID); pk.pack(config.network);
            }
            pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
+            pk.pack(KEY_TID); pk.pack(tid);
            pk.pack(KEY_V); pk.pack_map(1);
                pk.pack(i); pk.pack_map(2);
                    pk.pack(std::string("o")); pk.pack(start);
@@ -967,9 +1003,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
        }
    }

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                      pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1039,54 +1073,35 @@ NetworkEngine::sendListen(Sp<Node> n,
        const InfoHash& hash,
        const Query& query,
        const Blob& token,
-        Sp<Request> previous,
+        Tid socketId,
        RequestCb&& on_done,
-        RequestExpiredCb&& on_expired,
-        SocketCb&& socket_cb)
+        RequestExpiredCb&& on_expired)
 {
-    Tid socket;
-    TransId tid (n->getNewTid());
-    if (previous and previous->node == n) {
-        socket = previous->getSocket();
-    } else {
-        if (previous)
-            if (logger_)
-                logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str());
-        socket = n->openSocket(std::move(socket_cb));
-    }
-
-    if (not socket) {
-        if (logger_)
-            logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
-        return {};
-    }
-    TransId sid(socket);
-
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));

    auto has_query = not query.where.empty() or not query.select.empty();
-    pk.pack(KEY_A); pk.pack_map(4 + has_query);
+    pk.pack(KEY_A); pk.pack_map(5 + has_query);
      pk.pack(KEY_REQ_ID);    pk.pack(myid);
+      pk.pack(KEY_VERSION);   pk.pack(1);
      pk.pack(KEY_REQ_H);     pk.pack(hash);
      pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
-                              pk.pack_bin_body((const char*)sid.data(), sid.size());
+      pk.pack(KEY_REQ_SID);   pk.pack(socketId);
      if (has_query) {
          pk.pack(KEY_REQ_QUERY); pk.pack(query);
      }

    pk.pack(KEY_Q); pk.pack(QUERY_LISTEN);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done)
@@ -1095,8 +1110,7 @@ NetworkEngine::sendListen(Sp<Node> n,
        [=](const Request& req_status, bool done) { /* on expired */
            if (on_expired)
                on_expired(req_status, done);
-        },
-        socket
+        }
    );
    sendRequest(req);
    ++out_stats.listen;
@@ -1113,9 +1127,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1134,7 +1146,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
        RequestCb&& on_done,
        RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1150,15 +1162,14 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);

    pk.pack(KEY_Q);   pk.pack(QUERY_PUT);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y);   pk.pack(KEY_Q);
    pk.pack(KEY_UA);  pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1184,15 +1195,62 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
    return req;
 }

+Sp<Request>
+NetworkEngine::sendUpdateValues(Sp<Node> n,
+                                const InfoHash& infohash,
+                                const std::vector<Sp<Value>>& values,
+                                time_point created,
+                                const Blob& token,
+                                const size_t& socket_id)
+{
+    Tid tid (n->getNewTid());
+    Tid sid (socket_id);
+
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(5+(config.network?1:0));
+
+    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
+      pk.pack(KEY_REQ_ID);     pk.pack(myid);
+      pk.pack(KEY_VERSION);    pk.pack(1);
+      pk.pack(KEY_REQ_H);      pk.pack(infohash);
+      pk.pack(KEY_REQ_SID);   pk.pack(sid);
+      auto v = packValueHeader(buffer, values);
+      if (created < scheduler.time()) {
+          pk.pack(KEY_REQ_CREATION);
+          pk.pack(to_time_t(created));
+      }
+      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
+
+    pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_Y);   pk.pack(KEY_Q);
+    pk.pack(KEY_UA);  pk.pack(my_v);
+    if (config.network) {
+        pk.pack(KEY_NETID); pk.pack(config.network);
+    }
+
+    auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+        Blob(buffer.data(), buffer.data() + buffer.size()),
+        [=](const Request&, ParsedMessage&&) { /* on done */ },
+        [=](const Request&, bool) { /* on expired */ }
+    );
+    req->parts = std::move(v);
+    sendRequest(req);
+    ++out_stats.updateValue;
+    return req;
+}
+
 Sp<Request>
 NetworkEngine::sendRefreshValue(Sp<Node> n,
                const InfoHash& infohash,
                const Value::Id& vid,
                const Blob& token,
                RequestCb&& on_done,
+                RequestErrorCb&& on_error,
                RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1204,15 +1262,14 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
      pk.pack(KEY_REQ_TOKEN);    pk.pack(token);

    pk.pack(KEY_Q); pk.pack(QUERY_REFRESH);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1226,6 +1283,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                }
            }
        },
+        on_error,
        [=](const Request& req_status, bool done) { /* on expired */
            if (on_expired) {
                on_expired(req_status, done);
@@ -1248,9 +1306,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
      pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
      insertAddr(pk, addr);

-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1269,7 +1325,7 @@ NetworkEngine::sendError(const SockAddr& addr,
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4 + (include_id?1:0));
+    pk.pack_map(4 + (include_id?1:0) + (config.network?1:0));

    pk.pack(KEY_E); pk.pack_array(2);
      pk.pack(code);
@@ -1280,9 +1336,7 @@ NetworkEngine::sendError(const SockAddr& addr,
          pk.pack(KEY_REQ_ID); pk.pack(myid);
    }

-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_E);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -116,7 +116,6 @@ Node::cancelRequest(const Sp<net::Request>& req)
 {
    if (req) {
        req->cancel();
-        closeSocket(req->closeSocket());
        requests_.erase(req->getTid());
    }
 }
@@ -54,10 +54,12 @@ static const std::string KEY_REQ_EXPIRED {"exp"};
 static const std::string KEY_REQ_REFRESHED {"re"};
 static const std::string KEY_REQ_FIELDS {"fileds"};
 static const std::string KEY_REQ_WANT {"w"};
+static const std::string KEY_VERSION {"ve"};

 static const std::string QUERY_PING {"ping"};
 static const std::string QUERY_FIND {"find"};
 static const std::string QUERY_GET {"get"};
+static const std::string QUERY_UPDATE {"update"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -113,6 +115,7 @@ struct ParsedMessage {
    uint16_t error_code;
    /* reported address by the distant node */
    std::string ua;
+    int version {0};
    SockAddr addr;
    void msgpack_unpack(const msgpack::object& o);

@@ -224,6 +227,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
        type = MessageType::AnnounceValue;
    else if (parsed.q == QUERY_REFRESH)
        type = MessageType::Refresh;
+    else if (parsed.q == QUERY_UPDATE)
+        type = MessageType::UpdateValue;
    else
        throw msgpack::type_error();

@@ -295,6 +300,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
            parsedReq.fields = &o.val;
        else if (key == KEY_REQ_WANT)
            parsedReq.want = &o.val;
+        else if (key == KEY_VERSION)
+            version = o.val.as<int>();
    }

    if (parsedReq.sa) {
@@ -27,6 +27,7 @@ struct Node;
 namespace net {

 class NetworkEngine;
+class DhtProtocolException;
 struct ParsedMessage;

 /*!
@@ -72,16 +73,19 @@ struct Request {
            Sp<Node> node,
            Blob&& msg,
            std::function<void(const Request&, ParsedMessage&&)> on_done,
-            std::function<void(const Request&, bool)> on_expired,
-            Tid socket = 0) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
+    Request(MessageType type, Tid tid,
+            Sp<Node> node,
+            Blob&& msg,
+            std::function<void(const Request&, ParsedMessage&&)> on_done,
+            std::function<bool(const Request&, DhtProtocolException&&)> on_error,
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }

    Tid getTid() const { return tid; }
    MessageType getType() const { return type; }

-    Tid getSocket() { return socket; }
-    Tid closeSocket() { auto ret = socket; socket = 0; return ret; }
-
    void setExpired() {
        if (pending()) {
            state_ = Request::State::EXPIRED;
@@ -96,6 +100,15 @@ struct Request {
            clear();
        }
    }
+    bool setError(DhtProtocolException&& e) {
+        if (pending()) {
+            state_ = Request::State::EXPIRED;
+            bool handled = on_error and on_error(*this, std::forward<DhtProtocolException>(e));
+            clear();
+            return handled;
+        }
+        return true;
+    }

    void cancel() {
        if (pending()) {
@@ -113,6 +126,7 @@ private:

    void clear() {
        on_done = {};
+        on_error = {};
        on_expired = {};
        msg = {};
        parts = {};
@@ -128,11 +142,11 @@ private:
    time_point last_try {time_point::min()};   /* time of the last attempt to process the request. */

    std::function<void(const Request&, ParsedMessage&&)> on_done {};
+    std::function<bool(const Request&, DhtProtocolException&&)> on_error {};
    std::function<void(const Request&, bool)> on_expired {};

    Blob msg {};                      /* the serialized message. */
    std::vector<Blob> parts;
-    Tid socket;   /* the socket used for further reponses. */
 };

 } /* namespace net  */
@@ -65,10 +65,16 @@ struct Dht::SearchNode {
        ValueCache cache;
        Sp<Scheduler::Job> cacheExpirationJob {};
        Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
-         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
+        Tid socketId {0};
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback scb, Tid sid)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)), socketId(sid) {}
        CachedListenStatus(CachedListenStatus&&) = delete;
        CachedListenStatus(const CachedListenStatus&) = delete;
+        ~CachedListenStatus() {
+            if (socketId and req and req->node) {
+                req->node->closeSocket(socketId);
+            }
+        }
    };
    using NodeListenerStatus = std::map<Sp<Query>, CachedListenStatus>;

@@ -344,13 +350,15 @@ struct Dht::SearchNode {
     * 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, duration listen_expire) const {
-        auto listen_status = listenStatus.find(q);
-        if (listen_status == listenStatus.end() or not listen_status->second.req)
+    time_point getListenTime(const decltype(listenStatus)::const_iterator listen_status, duration listen_expire) const {
+        if (listen_status == listenStatus.cend() or not listen_status->second.req)
            return time_point::min();
        return listen_status->second.req->pending() ? time_point::max() :
            listen_status->second.req->reply_time + listen_expire - REANNOUNCE_MARGIN;
    }
+    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
+        return getListenTime(listenStatus.find(q), listen_expire);
+    }

    /**
     * Is this node expired or candidate
@@ -277,28 +277,16 @@ unpackMsg(Blob b) {
 }

 msgpack::object*
-findMapValue(const msgpack::object& map, const char* key) {
+findMapValue(const msgpack::object& map, const char* key, size_t key_length) {
    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
    for (unsigned i = 0; i < map.via.map.size; i++) {
        auto& o = map.via.map.ptr[i];
        if (o.key.type == msgpack::type::STR
+            && key_length == o.key.via.str.size
            && std::strncmp(o.key.via.str.ptr, key, o.key.via.str.size) == 0)
            return &o.val;
    }
    return nullptr;
 }

-msgpack::object*
-findMapValue(const msgpack::object& map, const std::string& key) {
-    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
-    for (unsigned i = 0; i < map.via.map.size; i++) {
-        auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR
-            && key.size() == o.key.via.str.size
-            && std::strncmp(o.key.via.str.ptr, key.data(), o.key.via.str.size) == 0)
-            return &o.val;
-    }
-    return nullptr;
-}
-
 }
@@ -44,12 +44,12 @@ DhtProxyTester::setUp() {
    nodeProxy->bootstrap(nodePeer.getBound());

    auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
-    auto serverIdentity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity);

-    serverProxy = std::make_unique<dht::DhtProxyServer>(
-            //dht::crypto::Identity{}, // http
-            serverIdentity,            // https
-            nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090");
+    dht::ProxyServerConfig serverConfig;
+    serverConfig.identity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity);
+    serverConfig.port = 8080;
+    serverConfig.pushServer = "127.0.0.1:8090";
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, serverConfig);

    clientConfig.server_ca = serverCAIdentity.second;
    clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester");
@@ -167,4 +167,51 @@ DhtRunnerTester::testListen() {
    node1.cancelListen(d, tokend);
 }

+void
+DhtRunnerTester::testListenLotOfBytes() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::atomic_uint valueCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);
+
+    std::string data(10000, 'a');
+
+    auto foo = dht::InfoHash::get("foo");
+    constexpr unsigned N = 50;
+
+    for (unsigned i=0; i<N; i++) {
+        node2.put(foo, data, [&](bool ok) {
+            std::lock_guard<std::mutex> lock(mutex);
+            putCount++;
+            if (ok) putOkCount++;
+            cv.notify_all();
+        });
+    }
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N; }));
+    }
+
+    dht::DhtRunner node3 {};
+    dht::DhtRunner::Config config;
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+    node3.run(42242, config);
+    node3.bootstrap(node1.getBound());
+
+    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
+        valueCount++;
+        cv.notify_all();
+        return true;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; }));
+    }
+
+    node3.cancelListen(foo, ftokenfoo.get());
+}
+
 }  // namespace test
@@ -32,6 +32,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testConstructors);
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenLotOfBytes);
    CPPUNIT_TEST_SUITE_END();

    dht::DhtRunner node1 {};
@@ -57,6 +58,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     * Test listen method
     */
    void testListen();
+    /**
+     * Test listen method with lot of datas
+     */
+    void testListenLotOfBytes();
 };

 }  // namespace test
@@ -39,9 +39,10 @@ HttpTester::setUp() {
    nodeProxy->run(0, /*identity*/{}, /*threaded*/true);
    nodeProxy->bootstrap(nodePeer->getBound());

-    serverProxy = std::unique_ptr<dht::DhtProxyServer>(
-        new dht::DhtProxyServer(
-            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
+    dht::ProxyServerConfig config;
+    config.port = 8080;
+    config.pushServer = "127.0.0.1:8090";
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, config);
 }

 void
@@ -64,10 +64,10 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        dht.run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);

-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
        std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;

        bool connected {false};
@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(*node, params);
+    print_node_info(node->getNodeInfo());
    std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;

 #ifndef WIN32_NATIVE
@@ -143,22 +143,23 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            print_help();
            continue;
        } else if (op == "ll") {
-            auto nodeInfo = node->getNodeInfo();
-            print_node_info(*node, params);
-            std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
-            std::cout << "IPv4 stats:" << std::endl;
-            std::cout << nodeInfo.ipv4.toString() << std::endl;
-            std::cout << "IPv6 stats:" << std::endl;
-            std::cout << nodeInfo.ipv6.toString() << std::endl;
+            node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
+                print_node_info(*nodeInfo);
+                std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
+                std::cout << "IPv4 stats:" << std::endl;
+                std::cout << nodeInfo->ipv4.toString() << std::endl;
+                std::cout << "IPv6 stats:" << std::endl;
+                std::cout << nodeInfo->ipv6.toString() << std::endl;
 #ifdef OPENDHT_PROXY_SERVER
-            for (const auto& proxy : proxies) {
-                std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
-                if (auto stats = proxy.second->stats())
-                    std::cout << "  " << stats->toString() << std::endl;
-                else
-                    std::cout << "  (stats not available yet)" << std::endl;                
-            }
+                for (const auto& proxy : proxies) {
+                    std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
+                    if (auto stats = proxy.second->stats())
+                        std::cout << "  " << stats->toString() << std::endl;
+                    else
+                        std::cout << "  (stats not available yet)" << std::endl;                
+                }
 #endif
+            });
            continue;
        } else if (op == "lr") {
            std::cout << "IPv4 routing table:" << std::endl;
@@ -190,14 +191,10 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            continue;
        } else if (op == "b") {
            iss >> idstr;
-            try {
-                auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty())
-                    addr.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-                node->bootstrap(addr.first.c_str(), addr.second.c_str());
-            } catch (const std::exception& e) {
-                std::cerr << e.what() << std::endl;
-            }
+            if (not idstr.empty())
+                node->bootstrap(idstr);
+            else
+                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
            continue;
        } else if (op == "log") {
            iss >> idstr;
@@ -221,14 +218,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
                iss >> idstr;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
            try {
-                unsigned int port = std::stoi(idstr);
-                proxies.emplace(port, std::unique_ptr<DhtProxyServer>(
-                    new DhtProxyServer(
-                        dht::crypto::Identity{}, node, port
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-                        ,pushServer
-#endif
-                )));
+                in_port_t port = std::stoi(idstr);
+                ProxyServerConfig serverConfig;
+                serverConfig.port = port;
+                serverConfig.pushServer = pushServer;
+                proxies.emplace(port, std::make_unique<DhtProxyServer>(node, serverConfig));
            } catch (...) { }
            continue;
        } else if (op == "psx") {
@@ -239,13 +233,12 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif // OPENDHT_PUSH_NOTIFICATIONS
            try {
                if (params.proxy_id.first and params.proxy_id.second){
-                    unsigned int port = std::stoi(idstr);
-                    proxies.emplace(port, std::unique_ptr<DhtProxyServer>(
-                        new DhtProxyServer(params.proxy_id, node, port
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-                                           ,pushServer
-#endif
-                    )));
+                    in_port_t port = std::stoi(idstr);
+                    ProxyServerConfig serverConfig;
+                    serverConfig.identity = params.proxy_id;
+                    serverConfig.port = port;
+                    serverConfig.pushServer = pushServer;
+                    proxies.emplace(port, std::make_unique<DhtProxyServer>(node, serverConfig));
                }
                else {
                    std::cerr << "Missing Identity private key or certificate" << std::endl;
@@ -537,25 +530,32 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        node->run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
-            node->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty()) {
+            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
+            node->bootstrap(params.bootstrap);
        }

 #ifdef OPENDHT_PROXY_SERVER
        std::map<in_port_t, std::unique_ptr<DhtProxyServer>> proxies;
 #endif
-        if (params.proxyserverssl and params.proxy_id.first and params.proxy_id.second){
+        if (params.proxyserver or params.proxyserverssl) {
 #ifdef OPENDHT_PROXY_SERVER
-            proxies.emplace(params.proxyserverssl, std::unique_ptr<DhtProxyServer>(
-                new DhtProxyServer(params.proxy_id,
-                                   node, params.proxyserverssl, params.pushserver, dhtConf.second.logger)));
-        }
-        if (params.proxyserver) {
-            proxies.emplace(params.proxyserver, std::unique_ptr<DhtProxyServer>(
-                new DhtProxyServer(
-                    dht::crypto::Identity{},
-                    node, params.proxyserver, params.pushserver, dhtConf.second.logger)));
+            ProxyServerConfig serverConfig;
+            serverConfig.pushServer = params.pushserver;
+            if (params.proxyserverssl and params.proxy_id.first and params.proxy_id.second){
+                serverConfig.identity = params.proxy_id;
+                serverConfig.port = params.proxyserverssl;
+                if (not params.persist_path.empty())
+                    serverConfig.persistStatePath = params.persist_path + '_' + std::to_string(serverConfig.port);
+                proxies.emplace(params.proxyserverssl, std::make_unique<DhtProxyServer>(node, serverConfig, dhtConf.second.logger));
+            }
+            if (params.proxyserver) {
+                serverConfig.identity = {};
+                serverConfig.port = params.proxyserver;
+                if (not params.persist_path.empty())
+                    serverConfig.persistStatePath = params.persist_path + '_' + std::to_string(serverConfig.port);
+                proxies.emplace(params.proxyserver, std::make_unique<DhtProxyServer>(node, serverConfig, dhtConf.second.logger));
+            }
 #else
            std::cerr << "DHT proxy server requested but OpenDHT built without proxy server support." << std::endl;
            exit(EXIT_FAILURE);
@@ -7,7 +7,7 @@ stats_total = {"users":0, "pushListenersCount":0, "listenCount":0, "totalListene

 for i in range(80,101):
    print("Collecting stats for proxy " + str(i))
-    response = requests.request('STATS', 'http://dhtproxy.jami.net:' + str(i))
+    response = requests.request('STATS', 'http://127.0.0.1:' + str(i))

    if response.status_code == 200:
        result = response.json()
@@ -38,20 +38,17 @@ for i in range(80,101):
        except:
            pass
        try:
-            total = 0
-            for h,v in result["puts"].items():
-                total += int(v)
-            stats['totalPermanentPuts'] = total
-            stats_total['totalPermanentPuts'] += total
+            stats['totalPermanentPuts'] = int(result["totalPermanentPuts"])
+            stats_total['totalPermanentPuts'] += int(result["totalPermanentPuts"])
        except:
            pass
-        
+
        stats['timestamp'] = str(ts)

-        with open("stats_proxy_" + str(i), "a") as stat_file:
-            stat_file.write(str(stats))
-            stat_file.write('\n')
+        #with open("stats_proxy_" + str(i), "a") as stat_file:
+        #    stat_file.write(str(stats))
+        #    stat_file.write('\n')

-with open("stats_proxy_total", "a") as stat_file:
+with open("stats_proxy_total", "w") as stat_file:
    stat_file.write(str(stats_total))
    stat_file.write('\n')
@@ -91,10 +91,10 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        dht.run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);

-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
        std::cout << "Scanning network..." << std::endl;
        auto all_nodes = std::make_shared<NodeSet>();

@@ -124,7 +124,7 @@ struct dht_params {
    bool log {false};
    bool syslog {false};
    std::string logfile {};
-    std::pair<std::string, std::string> bootstrap {};
+    std::string bootstrap {};
    dht::NetId network {0};
    in_port_t port {0};
    in_port_t proxyserver {0};
@@ -191,16 +191,14 @@ getDhtConfig(dht_params& params)
    return {std::move(config), std::move(context)};
 }

-void print_node_info(const dht::DhtRunner& node, const dht_params& params) {
-    std::cout << "OpenDHT node " << node.getNodeId() << " running on ";
-    auto port4 = node.getBoundPort(AF_INET);
-    auto port6 = node.getBoundPort(AF_INET6);
-    if (port4 == port6)
-        std::cout << "port " << port4 << std::endl;
+void print_node_info(const dht::NodeInfo& info) {
+    std::cout << "OpenDHT node " << info.node_id << " running on ";
+    if (info.bound4 == info.bound6)
+        std::cout << "port " << info.bound4 << std::endl;
    else
-        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
-    if (params.id.first)
-        std::cout << "Public key ID " << node.getId() << std::endl;
+        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
+    if (info.id)
+        std::cout << "Public key ID " << info.id << std::endl;
 }

 static const constexpr struct option long_options[] = {
@@ -291,10 +289,7 @@ parseArgs(int argc, char **argv) {
            params.public_stable = true;
            break;
        case 'b':
-            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
-            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
-                params.bootstrap.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-            }
+            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
            break;
        case 'V':
            params.version = true;
Author	SHA1	Message	Date
Adrien Béraud	08c7356481	dht: move socket to SearchNode	2020-06-12 02:51:08 -04:00
Adrien Béraud	e35f551c38	dht: add error handler for sendRefreshValue	2020-06-12 02:51:08 -04:00
Adrien Béraud	bd8c9d05d8	network engine: complete request in case of error	2020-06-12 02:51:08 -04:00
Adrien Béraud	6158751a32	network engine: auth success in case of refresh	2020-06-12 02:51:08 -04:00
Adrien Béraud	4c7d126782	network engine: fix sendError with non-default netId	2020-06-11 15:10:48 -04:00
Sébastien Blin	3e24ea1212	rng.h: fix build	2020-05-25 18:50:35 -04:00
Adrien Béraud	a7ac979da6	update to version 2.1.3	2020-05-25 15:18:51 -04:00
Adrien Béraud	87f48f0a28	getSeededRandomEngine: retry in case of exception	2020-05-25 15:18:51 -04:00
Adrien Béraud	15586a3cb1	travis: update configuration	2020-05-24 14:24:30 -04:00
Adrien Béraud	2195c0ed3e	proxy server: allow build without push notification support	2020-05-24 14:11:13 -04:00
Adrien Béraud	af1e904895	dockerfile: add pkg-config	2020-05-24 11:52:56 -04:00
Adrien Béraud	a4fbf7ced0	docker: update	2020-05-24 02:04:45 -04:00
Adrien Béraud	1b7e0679d5	proxy server: check push token when loading state	2020-05-24 02:04:45 -04:00
Sébastien Blin	9292c1116c	proxy: count permanent put values by hash in stats	2020-05-24 01:25:31 -04:00
Adrien Béraud	a2afb5bb31	network engine: send confirmation for value update	2020-05-24 00:20:27 -04:00
Adrien Béraud	15fbfdebcc	dht: only load id if set in configuration	2020-05-22 12:28:25 -04:00
Adrien Béraud	15eb221cd1	dht: save id, port when saving state	2020-05-22 12:15:10 -04:00
Adrien Béraud	7d3781acc8	dht: use pseudo-rng for ID generation	2020-05-22 12:14:47 -04:00
Sébastien Blin	f3100a6752	proxyclient: avoid to generate tokens with \0	2020-05-20 12:23:59 -04:00
Adrien Béraud	e3cd2a9d02	proxy server: add missing listener type	2020-05-19 19:52:51 -04:00
Adrien Béraud	19ccc3a327	proxy server: restart timers on resubscribe	2020-05-19 19:50:20 -04:00
Adrien Béraud	4587c0925f	utils: always check length in findMapValue	2020-05-19 19:50:20 -04:00
Adrien Béraud	fc96245891	cmake: cleanup	2020-05-19 19:50:20 -04:00
Adrien Béraud	a27815b4a9	proxy server: persist ops on disk	2020-05-19 19:50:20 -04:00
Adrien Béraud	5daaf372b4	proxyserver: generate id before put	2020-05-11 12:31:07 -04:00
Sébastien Blin	cc51d2684c	proxy: only keep last total stat The rest is not needed.	2020-05-05 17:00:07 -04:00
Sébastien Blin	28b5893239	listener: fully migrate to requests This patch migrates tellListenerRefreshed and tellListenerExpired to the new behavior	2020-05-05 10:38:03 -04:00
Adrien Béraud	c754b5cfce	update to version 2.1.0	2020-04-30 14:58:25 -04:00
Adrien Béraud	b0c1249710	network engine: check for logger	2020-04-29 20:23:19 -04:00
Adrien Béraud	c41ee417fb	tools: cleanup	2020-04-29 19:22:24 -04:00
Adrien Béraud	47ba0415ca	tools: use bootstrap split/default port from dht	2020-04-29 17:12:11 -04:00
Adrien Béraud	99857e25cc	dhtnode: use async version of getNodeInfo	2020-04-29 16:13:36 -04:00
Adrien Béraud	7f832ba9d5	NodeInfo: add bound ports	2020-04-29 16:00:02 -04:00
Sébastien Blin	8c4a894f93	misc: replace TransId to Tid	2020-04-28 13:55:29 -04:00
Sébastien Blin	dd3de412fb	dht: move updateValues logic into network_engine	2020-04-28 13:55:29 -04:00
Sébastien Blin	4ad5e6c9ba	misc: clean	2020-04-28 13:55:29 -04:00
Sébastien Blin	b445818521	tests: enable test for listening with lot of values	2020-04-28 13:55:29 -04:00
Sébastien Blin	7c14ea1730	network_engine: fix pack for sendUpdateValues	2020-04-28 13:55:29 -04:00
Sébastien Blin	7bf9f861a5	dht: send update values for listeners	2020-04-28 13:55:29 -04:00
Sébastien Blin	4cf5c87761	network_engin: move to pack(socket)	2020-04-28 13:55:29 -04:00
Sébastien Blin	e1c6fec581	dht: starts to add update and version support	2020-04-28 13:55:29 -04:00
Sébastien Blin	31da5b91fa	[WIP]net: add updateValues to replace Listen operations Actually, packets are not retransmitted for listen operations. This leads to some problems when they are already several values present at one hash. Ideally, packets should be transmitted inside a request to avoid packet loss and get all datas. This patch breaks the compatibility with previous versions	2020-04-28 13:55:29 -04:00
Adrien Béraud	01bcb1a750	network engine: completely remove TransId	2020-04-24 18:46:46 -04:00
Adrien Béraud	35348bf77f	dht: use default port for bootstraping	2020-04-24 17:45:40 -04:00
Adrien Béraud	ae58ea8312	search: expire only after candidate expiration	2020-04-24 16:32:07 -04:00
Adrien Béraud	f30d21e96f	move bootstrap logic from DhtRunner to Dht	2020-04-24 16:32:07 -04:00