proxy client: fix leak in sendListen

All encrypted values put on the DHT currently have the default priority
of 0 because the priority information is ignored by the Value::encrypt
function.

.github/workflows/ccpp.yml

@@ -87,7 +87,7 @@ jobs:
 
   build-macos:
     name: macOS/Clang build
-    runs-on: macos-12
+    runs-on: macOS-15
     steps:
     - uses: actions/checkout@v4
     - name: Dependencies
@@ -96,7 +96,7 @@ jobs:
 
     - name: expected-lite
       run: |
-        sudo mkdir /usr/local/include/nonstd && \
+        sudo mkdir -p /usr/local/include/nonstd && \
         sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
           -O /usr/local/include/nonstd/expected.hpp
 

CMakeLists.txt

@@ -97,7 +97,7 @@ if (NOT MSVC)
 
         # llhttp
         find_path(LLHTTP_INCLUDE_DIR llhttp.h)
-        find_library(LLHTTP_LIBRARY libllhttp)
+        find_library(LLHTTP_LIBRARY NAMES llhttp libllhttp)
         if (LLHTTP_INCLUDE_DIR AND LLHTTP_LIBRARY)
             message(STATUS "Found llhttp ${LLHTTP_INCLUDE_DIR} ${LLHTTP_LIBRARY}")
             add_library(llhttp_static STATIC IMPORTED)

c/opendht.cpp

@@ -344,6 +344,7 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
         config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
         config.push_topic = conf->push_topic ? std::string(conf->push_topic) : std::string{};
         config.push_platform = conf->push_platform ? std::string(conf->push_platform) : std::string{};
+        config.proxy_user_agent = conf->proxy_user_agent ? std::string(conf->proxy_user_agent) : std::string{};
         config.peer_discovery = conf->peer_discovery;
         config.peer_publish = conf->peer_publish;
 

c/opendht_c.h

@@ -148,6 +148,7 @@ struct OPENDHT_PUBLIC dht_runner_config {
     const char* push_token;
     const char* push_topic;
     const char* push_platform;
+    const char* proxy_user_agent;
     bool peer_discovery;
     bool peer_publish;
     dht_certificate* server_ca;

include/opendht/dht_proxy_client.h

@@ -53,19 +53,21 @@ public:
     DhtProxyClient();
 
     explicit DhtProxyClient(
-        std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
-        std::function<void()> loopSignal, const std::string& serverHost,
-        const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
+        std::shared_ptr<crypto::Certificate> serverCA,
+        crypto::Identity clientIdentity,
+        std::function<void()> loopSignal,
+        const std::string& serverHost,
+        const std::string& userAgent = "",
+        const std::string& pushClientId = "",
+        const std::string& pushToken = "",
+        const std::string& pushTopic = "",
+        const std::string& pushPlatform = "",
+        std::shared_ptr<Logger> logger = {}
+    );
 
     void setHeaderFields(http::Request& request);
 
-    virtual void setPushNotificationToken(const std::string& token) override {
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-        deviceKey_ = token;
-#else
-        (void) token;
-#endif
-    }
+    virtual void setPushNotificationToken(const std::string& token) override;
 
     virtual void setPushNotificationTopic(const std::string& topic) override {
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
@@ -351,7 +353,7 @@ private:
     std::string proxyUrl_;
     dht::crypto::Identity clientIdentity_;
     std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
-    //std::pair<std::string, std::string> serverHostService_;
+    std::string userAgent_ {"OpenDHT"};
     std::string pushClientId_;
     std::string pushSessionId_;
 

include/opendht/dht_proxy_server.h

@@ -98,6 +98,32 @@ public:
 
     asio::io_context& io_context() const;
 
+    using clock = std::chrono::steady_clock;
+    using time_point = clock::time_point;
+
+    struct PushStats {
+        uint64_t highPriorityCount {0};
+        uint64_t normalPriorityCount {0};
+
+        void increment(bool highPriority) {
+            if (highPriority)
+                highPriorityCount++;
+            else
+                normalPriorityCount++;
+        }
+
+        Json::Value toJson() const {
+            Json::Value val;
+            val["highPriorityCount"] = static_cast<Json::UInt64>(highPriorityCount);
+            val["normalPriorityCount"] = static_cast<Json::UInt64>(normalPriorityCount);
+            return val;
+        }
+
+        std::string toString() const {
+            return fmt::format("{} high priority, {} normal priority", highPriorityCount, normalPriorityCount);
+        }
+    };
+
     struct ServerStats {
         /** Current number of listen operations */
         size_t listenCount {0};
@@ -107,6 +133,17 @@ public:
         size_t totalPermanentPuts {0};
         /** Current number of push tokens with at least one listen operation */
         size_t pushListenersCount {0};
+
+        /** Time at which the server was started */
+        time_point serverStartTime;
+        /** Last time at which the stats were updated */
+        time_point lastUpdated;
+        /** Total number of push notification requests that the server attempted to
+          * send since being started, broken down by type and priority level */
+        PushStats androidPush;
+        PushStats iosPush;
+        PushStats unifiedPush;
+
         /** Average requests per second */
         double requestRate {0};
         /** Node Info **/
@@ -115,6 +152,10 @@ public:
         std::string toString() const {
             std::ostringstream ss;
             ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
+            ss << "Push requests in the last " << print_duration(lastUpdated - serverStartTime) << ": "
+                                               << "[Android: " << androidPush.toString() << "], "
+                                               << "[iOS: " << iosPush.toString() << "], "
+                                               << "[Unified: " << unifiedPush.toString() << "]" << std::endl;
             ss << "Requests: " << requestRate << " per second." << std::endl;
             if (nodeInfo) {
                 auto& ipv4 = nodeInfo->ipv4;
@@ -136,6 +177,11 @@ public:
             result["putCount"] = static_cast<Json::UInt64>(putCount);
             result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
             result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
+            result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
+            result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
+            result["androidPush"] = androidPush.toJson();
+            result["iosPush"] = iosPush.toJson();
+            result["unifiedPush"] = unifiedPush.toJson();
             result["requestRate"] = requestRate;
             if (nodeInfo)
                 result["nodeInfo"] = nodeInfo->toJson();
@@ -366,9 +412,6 @@ private:
     template <typename Is>
     void loadState(Is& is, size_t size);
 
-    using clock = std::chrono::steady_clock;
-    using time_point = clock::time_point;
-
     std::shared_ptr<asio::io_context> ioContext_;
     std::shared_ptr<DhtRunner> dht_;
     Json::StreamWriterBuilder jsonBuilder_;
@@ -393,6 +436,11 @@ private:
     std::shared_ptr<ServerStats> stats_;
     std::shared_ptr<NodeInfo> nodeInfo_ {};
     std::unique_ptr<asio::steady_timer> printStatsTimer_;
+    const time_point serverStartTime_;
+    mutable std::mutex pushStatsMutex_;
+    PushStats androidPush_;
+    PushStats iosPush_;
+    PushStats unifiedPush_;
 
     // Thread-safe access to listeners map.
     std::mutex lockListener_;

include/opendht/dhtrunner.h

@@ -64,6 +64,7 @@ public:
         std::string push_token {};
         std::string push_topic {};
         std::string push_platform {};
+        std::string proxy_user_agent {};
         bool peer_discovery {false};
         bool peer_publish {false};
         std::shared_ptr<dht::crypto::Certificate> server_ca;

meson.build

@@ -16,6 +16,7 @@ llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
 io_uring = dependency('liburing', required: false)
 
 deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp, io_uring]
+conf_data = configuration_data()
 
 add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language : 'cpp')
 if io_uring.found()
@@ -45,8 +46,6 @@ opendht_src = [
     'src/thread_pool.cpp',
 ]
 
-conf_data = configuration_data()
-
 if get_option('indexation').enabled()
     opendht_src += 'src/indexation/pht.cpp'
     add_project_arguments('-DOPENDHT_INDEXATION', language : 'cpp')

python/opendht.pyx

@@ -500,12 +500,13 @@ cdef class DhtConfig(object):
     def setRateLimit(self, ssize_t max_req_per_sec, ssize_t max_peer_req_per_sec):
         self._config.dht_config.node_config.max_req_per_sec = max_req_per_sec
         self._config.dht_config.node_config.max_peer_req_per_sec = max_peer_req_per_sec
-    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", Certificate server_ca = Certificate()):
+    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", str user_agent="", Certificate server_ca = Certificate()):
         self._config.proxy_server = proxy_server.encode()
         self._config.push_node_id = push_node_id.encode()
         self._config.push_token = push_token.encode()
         self._config.push_topic = push_topic.encode()
         self._config.push_platform = push_platform.encode()
+        self._config.proxy_user_agent = user_agent.encode()
         self._config.server_ca = server_ca._cert
     def setPeerDiscovery(self, bool peer_discovery, bool peer_publish):
         self._config.peer_discovery = peer_discovery

python/opendht_cpp.pxd

@@ -246,6 +246,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
             string push_token
             string push_topic
             string push_platform
+            string proxy_user_agent
             bool peer_discovery
             bool peer_publish
             shared_ptr[Certificate] server_ca

rust/src/dhtrunner.rs

@@ -49,6 +49,7 @@ impl DhtRunnerConfig {
             push_token: ptr::null(),
             push_topic: ptr::null(),
             push_platform: ptr::null(),
+            proxy_user_agent: ptr::null(),
             peer_discovery: false,
             peer_publish: false,
             server_ca: ptr::null_mut(),
@@ -84,6 +85,10 @@ impl DhtRunnerConfig {
         self.push_platform = CString::new(push_platform).unwrap().into_raw();
     }
 
+    pub fn set_proxy_user_agent(&mut self, proxy_user_agent: &str) {
+        self.proxy_user_agent = CString::new(proxy_user_agent).unwrap().into_raw();
+    }
+
     pub fn set_identity(&mut self, certificate: Box<DhtCertificate>, privatekey: Box<PrivateKey>) {
         self.dht_config.id.privatekey = Box::into_raw(privatekey);
         self.dht_config.id.certificate = Box::into_raw(certificate);

rust/src/ffi.rs

@@ -120,6 +120,7 @@ pub struct DhtRunnerConfig
     pub push_token: *const c_char,
     pub push_topic: *const c_char,
     pub push_platform: *const c_char,
+    pub proxy_user_agent: *const c_char,
     pub peer_discovery: bool,
     pub peer_publish: bool,
     pub server_ca: *mut DhtCertificate,

src/crypto.cpp

@@ -1164,11 +1164,17 @@ saveIdentity(const Identity& id, const std::string& path, const std::string& pri
         auto ca_key_data = id.first->serialize(privkey_password);
         std::ofstream key_file(path + ".pem");
         key_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!key_file)
+            throw CryptoException("Could not write private key file");
     }
     {
         auto ca_key_data = id.second->getPacked();
         std::ofstream crt_file(path + ".crt");
         crt_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!crt_file)
+            throw CryptoException("Could not write certificate file");
     }
 }
 
@@ -1276,12 +1282,14 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
         if (not ca.second->isCA()) {
             throw CryptoException("Signing certificate must be CA");
         }
-        if (err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
         ret.issuer = ca.second;
     } else {
-        if (err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
     }

src/dht.cpp

@@ -93,7 +93,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
                     r.second.done_cb(false, {});
                 sr.second->callbacks.clear();
                 for (const auto& a : sr.second->announce) {
-                    if (a.callback) a.callback(false, {});
+                    for (auto& cb : a.callbacks) cb(false, {});
                 }
                 sr.second->announce.clear();
                 sr.second->listeners.clear();

src/dht_proxy_client.cpp

@@ -106,13 +106,26 @@ getRandomSessionId(size_t length = 8) {
 DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(
-        std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
-        std::function<void()> signal, const std::string& serverHost,
-        const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
+        std::shared_ptr<dht::crypto::Certificate> serverCA,
+        dht::crypto::Identity clientIdentity,
+        std::function<void()> signal,
+        const std::string& serverHost,
+        const std::string& userAgent,
+        const std::string& pushClientId,
+        const std::string& pushToken,
+        const std::string& pushTopic,
+        const std::string& pushPlatform,
+        std::shared_ptr<dht::Logger> logger
+)
     : DhtInterface(logger)
     , proxyUrl_(serverHost)
     , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
-    , pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
+    , userAgent_(userAgent)
+    , pushClientId_(pushClientId)
+    , pushSessionId_(getRandomSessionId())
+    , deviceKey_(pushToken)
+    , notificationTopic_(pushTopic)
+    , platform_(pushPlatform)
     , loopSignal_(signal)
     , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
 {
@@ -484,7 +497,7 @@ DhtProxyClient::buildRequest(const std::string& target)
         request->set_certificate_authority(serverCertificate_);
     if (clientIdentity_.first and clientIdentity_.second)
         request->set_identity(clientIdentity_);
-    request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
+    request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
     return request;
 }
 
@@ -1108,10 +1121,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
             std::lock_guard<std::mutex> l(requestLock_);
             requests_[reqid] = request;
         }
-        request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) {
+        request->add_on_status_callback([r=std::weak_ptr(request), seconds = this->listenKeepIdle()] (unsigned status_code) {
             if(status_code == 200) {
                 // increase TCP_KEEPIDLE to save power
-                request->get_connection()->set_keepalive(seconds);
+                if (auto request = r.lock()) {
+                    request->get_connection()->set_keepalive(seconds);
+                }
             }
         });
         request->send();
@@ -1381,4 +1396,24 @@ DhtProxyClient::fillBody(bool resubscribe)
 }
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
+void
+DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) {
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
+    if (deviceKey_ != token) {
+        deviceKey_ = token;
+        if (statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected) {
+            if (logger_)
+                logger_->d("[proxy:client] [push] token changed, resubscribing");
+            for (auto& search : searches_) {
+                for (auto& listener : search.second.listeners) {
+                    resubscribe(search.first, listener.first, listener.second);
+                }
+            }
+        }
+    }
+#endif
+}
+
+
 } // namespace dht

src/dht_proxy_server.cpp

@@ -223,6 +223,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
     :   ioContext_(std::make_shared<asio::io_context>()),
         dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
         printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
+        serverStartTime_(clock::now()),
         connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
         pushServer_(config.pushServer),
         bundleId_(config.bundleId)
@@ -542,7 +543,15 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
     stats.requestRate = count / dt.count();
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     stats.pushListenersCount = pushListeners_.size();
+    {
+        std::lock_guard lk(pushStatsMutex_);
+        stats.androidPush = androidPush_;
+        stats.iosPush = iosPush_;
+        stats.unifiedPush = unifiedPush_;
+    }
 #endif
+    stats.serverStartTime = serverStartTime_;
+    stats.lastUpdated = now;
     stats.totalPermanentPuts = 0;
     std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
         stats.totalPermanentPuts += put.second.puts.size();
@@ -1107,10 +1116,14 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             notification["tokens"] = std::move(tokens);
             notification["platform"] = type == PushType::Android ? 2 : 1;
             notification["data"] = std::move(json);
-            notification["priority"] = highPriority ? "high" : "normal";
-            if (type == PushType::Android)
-                notification["time_to_live"] = 3600 * 24; // 24 hours
-            else {
+            auto priority = highPriority ? "high" : "normal";
+            if (type == PushType::Android) {
+                Json::Value androidConfig(Json::objectValue);
+                androidConfig["priority"] = priority;
+                androidConfig["ttl"] = "86400s"; // time to live = 24 hours
+                notification["android"] = std::move(androidConfig);
+            } else {
+                notification["priority"] = priority;
                 const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
                 uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
                 notification["expiration"] = exp;
@@ -1151,6 +1164,21 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             requests_[reqid] = request;
         }
         request->send();
+        // For monitoring purposes
+        std::lock_guard lk(pushStatsMutex_);
+        switch (type) {
+        case PushType::Android:
+            androidPush_.increment(highPriority);
+            break;
+        case PushType::iOS:
+            iosPush_.increment(highPriority);
+            break;
+        case PushType::UnifiedPush:
+            unifiedPush_.increment(highPriority);
+            break;
+        default:
+            break;
+        }
     }
     catch (const std::exception &e){
         if (logger_)

src/dhtrunner.cpp

@@ -1120,13 +1120,13 @@ DhtRunner::enableProxy(bool proxify)
                         cv.notify_all();
                     }
                 },
-                config_.proxy_server, config_.push_node_id, logger_);
-        if (not config_.push_token.empty())
-            dht_via_proxy->setPushNotificationToken(config_.push_token);
-        if (not config_.push_topic.empty())
-            dht_via_proxy->setPushNotificationTopic(config_.push_topic);
-        if (not config_.push_platform.empty())
-            dht_via_proxy->setPushNotificationPlatform(config_.push_platform);
+                config_.proxy_server,
+                config_.proxy_user_agent,
+                config_.push_node_id,
+                config_.push_token,
+                config_.push_topic,
+                config_.push_platform,
+                logger_);
         dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
         // and use it
         use_proxy = proxify;

src/search.h

@@ -45,7 +45,7 @@ struct Dht::Announce {
     bool permanent;
     Sp<Value> value;
     time_point created;
-    DoneCallback callback;
+    std::vector<DoneCallback> callbacks;
 };
 
 struct Dht::SearchNode {
@@ -442,8 +442,9 @@ struct Dht::Search {
             g.second.done_cb = {};
         }
         for (auto& a : announce) {
-            a.callback(false, {});
-            a.callback = {};
+            for (auto& cb : a.callbacks)
+                cb(false, {});
+            a.callbacks.clear();
         }
     }
 
@@ -627,7 +628,9 @@ struct Dht::Search {
             return a.value->id == value->id;
         });
         if (a_sr == announce.end()) {
-            announce.emplace_back(Announce {permanent, value, created, callback});
+            auto& a = announce.emplace_back(Announce {permanent, value, created, {}} );
+            if (callback)
+                a.callbacks.emplace_back(std::move(callback));
             for (auto& n : nodes) {
                 n->probe_query.reset();
                 n->acked[value->id].req.reset();
@@ -635,24 +638,26 @@ struct Dht::Search {
         } else {
             a_sr->permanent = permanent;
             a_sr->created = created;
-            if (a_sr->value != value) {
+            if (a_sr->value != value && *a_sr->value != *value) {
+                // Value is updated, previous ops are failed
                 a_sr->value = value;
                 for (auto& n : nodes) {
                     n->acked[value->id].req.reset();
                     n->probe_query.reset();
                 }
-            }
-            if (isAnnounced(value->id)) {
-                if (a_sr->callback)
-                    a_sr->callback(true, {});
-                a_sr->callback = {};
+                for (auto& cb: a_sr->callbacks)
+                    cb(false, {});
+                a_sr->callbacks.clear();
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
+            } else if (isAnnounced(value->id)) {
+                // Same value, already announced
                 if (callback)
                     callback(true, {});
-                return;
             } else {
-                if (a_sr->callback)
-                    a_sr->callback(false, {});
-                a_sr->callback = callback;
+                // Same value, not announced yet
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
             }
         }
     }
@@ -722,8 +727,8 @@ struct Dht::Search {
             std::vector<DoneCallback> a_cbs;
             a_cbs.reserve(announce.size());
             for (auto ait = announce.begin() ; ait != announce.end(); ) {
-                if (ait->callback)
-                    a_cbs.emplace_back(std::move(ait->callback));
+                a_cbs.insert(a_cbs.end(), std::make_move_iterator(ait->callbacks.begin()), std::make_move_iterator(ait->callbacks.end()));
+                ait->callbacks.clear();
                 if (not ait->permanent)
                     ait = announce.erase(ait);
                 else
@@ -747,9 +752,11 @@ struct Dht::Search {
                 if (vid != Value::INVALID_ID and (!a.value || a.value->id != vid))
                     return true;
                 if (isAnnounced(a.value->id)) {
-                    if (a.callback) {
-                        a.callback(true, getNodes());
-                        a.callback = nullptr;
+                    if (!a.callbacks.empty()) {
+                        const auto& nodes = getNodes();
+                        for (auto& cb : a.callbacks)
+                            cb(true, nodes);
+                        a.callbacks.clear();
                     }
                     if (not a.permanent)
                         return false;

src/value.cpp

@@ -271,6 +271,7 @@ Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
     setRecipient(to.getId());
     sign(from);
     Value nv {id};
+    nv.priority = priority;
     nv.setCypher(to.encrypt(getToEncrypt()));
     return nv;
 }

tests/dhtrunnertester.cpp

@@ -90,6 +90,59 @@ DhtRunnerTester::testGetPut() {
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
 
+void
+DhtRunnerTester::testPutDuplicate() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("hey");
+    val->id = 42;
+    auto val_data = val->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
+void
+DhtRunnerTester::testPutOverride() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("meh");
+    val->id = 42;
+    auto val2 = std::make_shared<dht::Value>("hey");
+    val2->id = 42;
+    CPPUNIT_ASSERT_EQUAL(val->id, val2->id);
+    auto val_data = val->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val2, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(!p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
 void
 DhtRunnerTester::testListen() {
     std::mutex mutex;

tests/dhtrunnertester.h

@@ -31,6 +31,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(DhtRunnerTester);
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
+    CPPUNIT_TEST(testPutDuplicate);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST(testIdOps);
@@ -55,6 +56,14 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test get and put methods
      */
     void testGetPut();
+    /**
+     * Test get and multiple put
+     */
+    void testPutDuplicate();
+    /**
+     * Test get and multiple put with changing value
+     */
+    void testPutOverride();
     /**
      * Test listen method
      */