proxy_server: detach pending push requests before destruction

~DhtProxyServer destroyed requests_ whose ~http::Request notifies DONE; the handler registered by sendPushNotification then erased from the map being destroyed, corrupting the heap (double free under ASan) whenever a push request was still pending, e.g. an unreachable push server. Detach the state-change callbacks and release the requests before the members are destroyed.
storage: re-storing identical content is a refresh, not an edition
2026-11-08 12:28:42 +00:00 · 2026-09-04 14:43:53 -04:00 · 2026-09-04 14:43:53 -04:00 · 2026-09-04 14:43:53 -04:00 · 2026-09-04 14:43:53 -04:00 · 2026-09-04 14:43:53 -04:00
20 changed files with 3678 additions and 91 deletions
@@ -870,6 +870,10 @@ if (BUILD_TESTING)
        tests/test_parsedmessage.cpp
        tests/test_networkengine.h
        tests/test_networkengine.cpp
+        tests/test_opcache.h
+        tests/test_opcache.cpp
+        tests/test_valuecache.h
+        tests/test_valuecache.cpp
    )
    if (OPENDHT_PROXY_CLIENT)
        list (APPEND test_FILES
@@ -76,8 +76,14 @@ public:

    asio::io_context& io_context() const;

+    // Monotonic clock for elapsed-time statistics (request rate, uptime).
+    // Must stay on steady_clock so NTP/wall-clock corrections don't skew durations.
    using clock = std::chrono::steady_clock;
    using time_point = clock::time_point;
+    // Wall clock for persisted subscribe/permanent-put lease expirations,
+    // so they map directly to epoch time and survive restarts.
+    using sys_clock = std::chrono::system_clock;
+    using sys_time_point = sys_clock::time_point;

    struct PushStats
    {
@@ -401,12 +407,12 @@ private:
    std::shared_ptr<ConnectionListener> connListener_;
    struct PermanentPut
    {
-        time_point expiration;
+        sys_time_point expiration;
        std::string pushToken;
        std::string clientId;
        std::shared_ptr<PushSessionContext> sessionCtx;
-        std::unique_ptr<asio::steady_timer> expireTimer;
-        std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
        Sp<Value> value;
        PushType type;
        std::string topic;
@@ -419,7 +425,7 @@ private:
            p.pack("value");
            p.pack(value);
            p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(sys_clock::to_time_t(expiration));
            if (not clientId.empty()) {
                p.pack("cid");
                p.pack(clientId);
@@ -460,12 +466,12 @@ private:
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
    struct Listener
    {
-        time_point expiration;
+        sys_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;
+        std::unique_ptr<asio::system_timer> expireTimer;
+        std::unique_ptr<asio::system_timer> expireNotifyTimer;
        PushType type;
        std::string topic;

@@ -476,7 +482,7 @@ private:
            p.pack("cid");
            p.pack(clientId);
            p.pack("exp");
-            p.pack(to_time_t(expiration));
+            p.pack(sys_clock::to_time_t(expiration));
            if (sessionCtx) {
                std::lock_guard l(sessionCtx->lock);
                p.pack("sid");
@@ -106,9 +106,16 @@ public:

    void timeout(const std::chrono::seconds& timeout, HandlerCb cb = {});

+    /**
+     * Close the underlying socket.
+     * Safe to call from any thread: the actual close is dispatched to
+     * the thread running the io_context.
+     */
    void close();

 private:
+    void doClose();
+
    template<typename T>
    T wrapCallback(T cb) const
    {
@@ -8,7 +8,7 @@ namespace dht {
 namespace proxy {

 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
-constexpr const std::chrono::hours OP_MARGIN {2};   // two hours
+constexpr const std::chrono::hours OP_MARGIN {8};   // eight hours

 using ListenToken = uint64_t;

@@ -25,8 +25,10 @@ struct DhtProxyClient::OperationState

 struct DhtProxyClient::Listener
 {
-    Listener(OpValueCache&& c)
-        : cache(std::move(c))
+    // The proxy server is a single logical source that re-delivers values
+    // (push refresh, listen restart): don't count duplicates.
+    Listener(ValueCallback&& cb)
+        : cache(std::move(cb), false)
    {}

    OpValueCache cache;
@@ -34,26 +36,26 @@ struct DhtProxyClient::Listener
    Sp<OperationState> opstate;
    std::shared_ptr<http::Request> request;
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
-    std::unique_ptr<asio::steady_timer> refreshSubscriberTimer;
+    std::unique_ptr<asio::system_timer> refreshSubscriberTimer;
 #endif
 };

 struct PermanentPut
 {
-    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::steady_timer>&& j, const Sp<std::atomic_bool>& o)
+    PermanentPut(const Sp<Value>& v, std::unique_ptr<asio::system_timer>&& j, const Sp<std::atomic_bool>& o)
        : value(v)
        , refreshPutTimer(std::move(j))
        , ok(o)
    {}

    Sp<Value> value;
-    std::unique_ptr<asio::steady_timer> refreshPutTimer;
+    std::unique_ptr<asio::system_timer> refreshPutTimer;
    Sp<std::atomic_bool> ok;
 };

 struct DhtProxyClient::ProxySearch
 {
-    SearchCache ops {};
+    SearchCache ops {false};
    std::unique_ptr<asio::steady_timer> opExpirationTimer;
    std::map<size_t, Listener> listeners {};
    std::map<Value::Id, PermanentPut> puts {};
@@ -424,7 +426,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
        auto& search = searches_[key];
        if (val->id) {
            auto id = val->id;
-            auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+            auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                        proxy::OP_TIMEOUT - proxy::OP_MARGIN);
            refreshPutTimer->async_wait(
                std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
@@ -540,7 +542,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
                            auto it = search.pendingPuts.find(val);
                            if (it != search.pendingPuts.end()) {
                                auto sok = std::make_shared<std::atomic_bool>(ok);
-                                auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
+                                auto refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
                                                                                            proxy::OP_TIMEOUT
                                                                                                - proxy::OP_MARGIN);
                                refreshPutTimer->async_wait(
@@ -891,8 +893,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
                 * (if the proxy crash for any reason)
                 */
                if (!l->second.refreshSubscriberTimer)
-                    l->second.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
-                l->second.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+                    l->second.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
+                l->second.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                             - proxy::OP_MARGIN);
                l->second.refreshSubscriberTimer->async_wait(
                    std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
@@ -1192,9 +1194,9 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
        for (auto& put : search.second.puts) {
            doPut(key, put.second.value, [ok = put.second.ok](bool result) { *ok = result; }, time_point::max(), true);
            if (!put.second.refreshPutTimer) {
-                put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
+                put.second.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_);
            }
-            put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT
+            put.second.refreshPutTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT
                                                   - proxy::OP_MARGIN);
            put.second.refreshPutTimer->async_wait(
                std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, put.first));
@@ -1275,10 +1277,10 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                auto vid = std::stoull(vidIt->second);
                auto& put = search.puts.at(vid);
                if (!put.refreshPutTimer)
-                    put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_,
-                                                                               std::chrono::steady_clock::now());
+                    put.refreshPutTimer = std::make_unique<asio::system_timer>(httpContext_,
+                                                                               std::chrono::system_clock::now());
                else
-                    put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
+                    put.refreshPutTimer->expires_at(std::chrono::system_clock::now());
                put.refreshPutTimer->async_wait(
                    std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
                ret = PushNotificationResult::PutRefresh;
@@ -1290,12 +1292,12 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                                               : PushNotificationResult::ListenRefresh;
            }
        } else {
-            auto key = InfoHash(notification.at("key"));
            system_clock::time_point sendTime = system_clock::time_point::min();
            try {
                sendTime = system_clock::time_point(std::chrono::milliseconds(std::stoull(notification.at("t"))));
            } catch (...) {
            }
+            auto key = InfoHash(notification.at("key"));
            auto& search = searches_.at(key);
            auto expired = notification.find("exp");
            for (auto& list : search.listeners) {
@@ -1308,13 +1310,25 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
                if (expired == notification.end()) {
                    auto cb = list.second.cb;
                    auto oldValues = list.second.cache.getValues();
+                    auto receivedIds = std::make_shared<std::set<Value::Id>>();
                    get(
                        key,
-                        [cb, sendTime](const std::vector<Sp<Value>>& vals) { return cb(vals, false, sendTime); },
-                        [cb, oldValues, sendTime](bool /*ok*/) {
+                        [cb, sendTime, receivedIds](const std::vector<Sp<Value>>& vals) {
+                            for (const auto& v : vals)
+                                receivedIds->insert(v->id);
+                            return cb(vals, false, sendTime);
+                        },
+                        [cb, oldValues, sendTime, receivedIds](bool ok) {
+                            if (!ok)
+                                return;
                            // Decrement old values refcount to expire values not
                            // present in the new list
-                            cb(oldValues, true, sendTime);
+                            std::vector<Sp<Value>> toExpire;
+                            for (const auto& v : oldValues)
+                                if (receivedIds->find(v->id) == receivedIds->end())
+                                    toExpire.emplace_back(v);
+                            if (!toExpire.empty())
+                                cb(toExpire, true, sendTime);
                        });
                    ret = PushNotificationResult::Values;
                } else {
@@ -1378,9 +1392,9 @@ DhtProxyClient::resubscribe([[maybe_unused]] const InfoHash& key,
    header.method(restinio::http_method_subscribe());
    header.request_target("/key/" + key.toString());
    if (!listener.refreshSubscriberTimer) {
-        listener.refreshSubscriberTimer = std::make_unique<asio::steady_timer>(httpContext_);
+        listener.refreshSubscriberTimer = std::make_unique<asio::system_timer>(httpContext_);
    }
-    listener.refreshSubscriberTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+    listener.refreshSubscriberTimer->expires_at(std::chrono::system_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
    listener.refreshSubscriberTimer->async_wait(
        std::bind(&DhtProxyClient::handleResubscribe, this, std::placeholders::_1, key, token, opstate));
    auto vcb = listener.cb;
@@ -210,7 +210,7 @@ DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
        clientId = cid->as<std::string>();
    }
    if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = sys_clock::from_time_t(exp->as<time_t>());
    }
    if (auto token = findMapValue(o, "token"sv)) {
        pushToken = token->as<std::string>();
@@ -240,7 +240,7 @@ DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
        clientId = cid->as<std::string>();
    }
    if (auto exp = findMapValue(o, "exp"sv)) {
-        expiration = from_time_t(exp->as<time_t>());
+        expiration = sys_clock::from_time_t(exp->as<time_t>());
    }
    if (auto sid = findMapValue(o, "sid"sv)) {
        if (not sessionCtx)
@@ -414,7 +414,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                    logger_->debug("[proxy:server] Loading {} 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.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                       pput.second.expiration);
                        pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut,
                                                                      this,
@@ -437,7 +437,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                }
                                return json;
                            };
-                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            pput.second.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                                 pput.second.expiration
                                                                                                     - proxy::OP_MARGIN);
                            pput.second.expireNotifyTimer->async_wait(
@@ -450,7 +450,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                          pput.second.topic));
                        }
 #endif
-                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, time_point::max(), true);
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, dht::time_point::max(), true);
                    }
                }
            } else {
@@ -487,7 +487,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                                                                 expired);
                                               });
                            // expire notify
-                            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                              listener.expiration
                                                                                                  - proxy::OP_MARGIN);
                            auto jsonProvider = [infoHash = listeners.first.toString(),
@@ -509,7 +509,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
                                          listener.type,
                                          listener.topic));
                            // cancel push listen
-                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(),
+                            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                        listener.expiration);
                            listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
                                                                       this,
@@ -571,6 +571,18 @@ DhtProxyServer::~DhtProxyServer()
    ioContext_->stop();
    if (serverThread_.joinable())
        serverThread_.join();
+    {
+        // ~Request notifies DONE and its handler erases from requests_:
+        // detach pending push requests so nothing calls back into this object.
+        decltype(requests_) pending;
+        {
+            std::lock_guard l(requestLock_);
+            pending = std::move(requests_);
+        }
+        for (auto& r : pending)
+            r.second->add_on_state_change_callback({});
+        pending.clear();
+    }
    if (logger_)
        logger_->debug("[proxy:server] http server closed");
 }
@@ -907,6 +919,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
        Json::Value r;
        auto* char_data = reinterpret_cast<const char*>(request->body().data());
        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
+
        if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) {
            auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
            response.set_body(RESP_MSG_JSON_INCORRECT);
@@ -950,15 +963,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
        }
        auto& listener = *listIt;

-        // Expiration
-        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+        // Expiration (wall-clock: persisted as epoch time, survives restarts)
+        auto timeout = sys_clock::now() + proxy::OP_TIMEOUT;
        listener.expiration = timeout;
        listener.type = type;
        listener.topic = topic;
        if (listener.expireNotifyTimer)
            listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
        else
-            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
+            listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(), timeout - proxy::OP_MARGIN);
        auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() {
            Json::Value json;
            json["timeout"] = h;
@@ -975,7 +988,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
                                                         listener.type,
                                                         listener.topic));
        if (!listener.expireTimer)
-            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+            listener.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
        else
            listener.expireTimer->expires_at(timeout);
        listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
@@ -1405,7 +1418,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                    topic = pVal["topic"].asString();
                }
                std::lock_guard lock(lockSearchPuts_);
-                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+                auto timeout = sys_clock::now() + proxy::OP_TIMEOUT;
                auto& sPuts = puts_[infoHash];
                if (value->id == Value::INVALID_ID) {
                    for (auto& pp : sPuts.puts) {
@@ -1439,7 +1452,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                pput.expiration = timeout;
                if (not pput.expireTimer) {
                    // cancel permanent put
-                    pput.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
+                    pput.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                    if (not pushToken.empty()) {
                        pput.pushToken = pushToken;
@@ -1478,7 +1491,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                        return json;
                    };
                    if (!pput.expireNotifyTimer)
-                        pput.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(),
+                        pput.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
                                                                                      timeout - proxy::OP_MARGIN);
                    else
                        pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -1507,7 +1520,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                        response.done();
                    }
                },
-                time_point::max(),
+                dht::time_point::max(),
                permanent);
            return restinio::request_handling_status_t::accepted;
        } else {
@@ -233,11 +233,32 @@ Connection::Connection(asio::io_context& ctx,

 Connection::~Connection()
 {
-    close();
+    // No asynchronous operation can be pending here: their handlers keep
+    // the connection alive (see wrapCallback), so closing synchronously
+    // is safe regardless of the calling thread.
+    doClose();
 }

 void
 Connection::close()
+{
+    // asio sockets are not thread-safe. Closing from a thread other than the
+    // one running the io_context races with in-flight operations on the same
+    // socket (e.g. async_connect closing and reopening it to try the next
+    // endpoint), corrupting the reactor's descriptor state.
+    // Always perform the close on the io_context thread.
+    if (ctx_.get_executor().running_in_this_thread()) {
+        doClose();
+        return;
+    }
+    asio::post(ctx_, [w = weak_from_this()] {
+        if (auto sthis = w.lock())
+            sthis->doClose();
+    });
+}
+
+void
+Connection::doClose()
 {
    std::lock_guard lock(mutex_);
    asio::error_code ec;
@@ -631,12 +652,13 @@ Connection::async_handshake(HandlerCb cb)
 {
    std::lock_guard lock(mutex_);
    if (ssl_socket_) {
-        std::weak_ptr<Connection> wthis = shared_from_this();
+        // Keep the connection alive while the handshake is in flight, like
+        // every other asynchronous operation (see wrapCallback).
        ssl_socket_
-            ->async_handshake(asio::ssl::stream<asio::ip::tcp::socket>::client, [wthis, cb](const asio::error_code& ec) {
+            ->async_handshake(asio::ssl::stream<asio::ip::tcp::socket>::client, [sthis = shared_from_this(), cb](const asio::error_code& ec) {
                if (ec == asio::error::operation_aborted)
                    return;
-                if (auto sthis = wthis.lock()) {
+                {
                    auto& this_ = *sthis;
                    auto verify_ec = SSL_get_verify_result(this_.ssl_socket_->asio_ssl_stream().native_handle());
                    if (this_.logger_) {
@@ -12,18 +12,20 @@ OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals, const system_clo
    std::vector<Sp<Value>> newValues;
    for (const auto& v : vals) {
        auto viop = values.emplace(v->id, v);
+        auto& stored = viop.first->second;
        if (viop.second) {
            newValues.emplace_back(v);
-        } else if (*viop.first->second.data != *v) {
-            // Special case for edition
-            if (v->seq > viop.first->second.data->seq) {
-                viop.first->second.data = v;
-                newValues.emplace_back(v);
-            }
-        } else {
-            viop.first->second.refCount++;
+        } else if (*stored.data != *v) {
+            if (v->seq <= stored.data->seq)
+                continue; // older version from a lagging source
+            // Edited: only the reporting source is known to hold this version
+            stored.data = v;
+            stored.refCount = 1;
+            newValues.emplace_back(v);
+        } else if (countSources) {
+            stored.refCount++;
        }
-        viop.first->second.updated = t;
+        stored.updated = t;
    }
    return newValues.empty() ? true : callback(newValues, false);
 }
@@ -37,6 +39,8 @@ OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals, const system_c
        if (vit != values.end()) {
            if (vit->second.updated > t)
                continue;
+            if (v->seq < vit->second.data->seq)
+                continue; // source held an older version, not counted

            vit->second.updated = t;
            vit->second.refCount--;
@@ -174,7 +178,7 @@ SearchCache::listen(const ValueCallback& get_cb, const Sp<Query>& q, Value::Filt
    auto op = getOp(q);
    if (op == ops.end()) {
        // New query
-        op = ops.emplace(q, std::make_unique<OpCache>()).first;
+        op = ops.emplace(q, std::make_unique<OpCache>(countSources_)).first;
        auto& cache = *op->second;
        cache.searchToken = onListen(
            q,
@@ -240,15 +244,17 @@ SearchCache::expire(const time_point& now, const std::function<void(size_t)>& on
 bool
 SearchCache::get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const
 {
+    if (not gcb)
+        return false;
    auto op = getOp(q);
-    if (op != ops.end()) {
-        auto vals = op->second->get(f);
-        if ((not vals.empty() and not gcb(vals)) or op->second->isSynced()) {
-            dcb(true, {});
-            return true;
-        }
-    }
-    return false;
+    if (op == ops.end() or not op->second->isSynced())
+        return false;
+    auto vals = op->second->get(f);
+    if (not vals.empty())
+        gcb(vals); // a consumer stopping early is not a failure
+    if (dcb)
+        dcb(true, {});
+    return true;
 }

 std::vector<Sp<Value>>
@@ -21,19 +21,29 @@ struct OpCacheValueStorage
 class OpValueCache
 {
 public:
-    OpValueCache(ValueCallback&& cb) noexcept
+    /**
+     * @param countSources  When true, a value is kept until every source
+     *                      that reported it has expired it (per-node caches,
+     *                      IPv4/IPv6 searches, local storage). When false,
+     *                      re-deliveries of the same value are not counted
+     *                      and a single expiration removes it (single
+     *                      logical source, e.g. a proxy server).
+     */
+    OpValueCache(ValueCallback&& cb, bool countSources = true) noexcept
        : callback(std::forward<ValueCallback>(cb))
+        , countSources(countSources)
    {}
    explicit OpValueCache(OpValueCache&& o) noexcept
        : values(std::move(o.values))
        , callback(std::move(o.callback))
+        , countSources(o.countSources)
    {
        o.callback = {};
    }

-    static ValueCallback cacheCallback(ValueCallback&& cb, std::function<void()>&& onCancel)
+    static ValueCallback cacheCallback(ValueCallback&& cb, std::function<void()>&& onCancel, bool countSources = true)
    {
-        return [cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb)),
+        return [cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb), countSources),
                onCancel = std::move(onCancel)](const std::vector<Sp<Value>>& vals, bool expired) {
            auto ret = cache->onValue(vals, expired);
            if (not ret)
@@ -89,19 +99,22 @@ private:
    size_t syncedNodes {0};
    std::map<Value::Id, OpCacheValueStorage> values {};
    ValueCallback callback;
+    bool countSources {true};
 };

 class OpCache
 {
 public:
-    OpCache()
-        : cache([this](const std::vector<Sp<Value>>& vals, bool expired) {
-            if (expired)
-                onValuesExpired(vals);
-            else
-                onValuesAdded(vals);
-            return true;
-        })
+    explicit OpCache(bool countSources = true)
+        : cache(
+              [this](const std::vector<Sp<Value>>& vals, bool expired) {
+                  if (expired)
+                      onValuesExpired(vals);
+                  else
+                      onValuesAdded(vals);
+                  return true;
+              },
+              countSources)
    {}

    bool onValue(const std::vector<Sp<Value>>& vals, bool expired) { return cache.onValue(vals, expired); }
@@ -156,7 +169,9 @@ private:
 class SearchCache
 {
 public:
-    SearchCache() {}
+    explicit SearchCache(bool countSources = true)
+        : countSources_(countSources)
+    {}
    SearchCache(SearchCache&&) = default;

    using OnListen = std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)>;
@@ -192,6 +207,7 @@ private:

    size_t nextToken_ {1};
    time_point nextExpiration_ {time_point::max()};
+    bool countSources_ {true};
 };

 } // namespace dht
@@ -232,6 +232,8 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
        if (it->data != value) {
            size_t size_old = it->data->size();
            ssize_t size_diff = size_new - (ssize_t) size_old;
+            // Re-storing identical content is a refresh, not an edition
+            size_t edited = (*it->data != *value) ? 1 : 0;
            // DHT_LOG.DEBUG("Updating %s -> %s", id.toString().c_str(), value->toString().c_str());
            //  clear quota for previous value
            if (it->store_bucket)
@@ -243,7 +245,7 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
                sb->insert(id, *value, expiration);
            it->data = value;
            total_size += size_diff;
-            return std::make_pair(&(*it), StoreDiff {size_diff, 0, 0, 1});
+            return std::make_pair(&(*it), StoreDiff {size_diff, 0, 0, edited});
        }
    } else {
        // DHT_LOG.DEBUG("Storing %s -> %s", id.toString().c_str(), value->toString().c_str());
@@ -71,11 +71,21 @@ public:
    {
        std::vector<Sp<Value>> expired_values;
        for (auto it = values.begin(); it != values.end();) {
-            if (it->second.expiration <= now) {
+            auto effectiveExpiration = it->second.expiration;
+            if (status == ListenSyncStatus::SYNCED)
+                // While the listen is synced, the remote node pushes
+                // explicit expirations and periodic refreshes.  Add a
+                // grace period proportional to the value's type
+                // expiration so that a single lost UDP refresh
+                // notification does not cause a premature local drop.
+                // A grace of one type-expiration covers exactly one
+                // missed re-announcement cycle.
+                effectiveExpiration += it->second.typeExpiration;
+            if (effectiveExpiration <= now) {
                expired_values.emplace_back(std::move(it->second.data));
                it = values.erase(it);
            } else {
-                next = std::min(next, it->second.expiration);
+                next = std::min(next, effectiveExpiration);
                ++it;
            }
        }
@@ -148,12 +158,14 @@ private:
        Sp<Value> data {};
        time_point created {};
        time_point expiration {};
+        duration typeExpiration {}; /* value type's expiration period, used as grace when synced */

        CacheValueStorage() {}
-        CacheValueStorage(const Sp<Value>& v, time_point t, time_point e)
+        CacheValueStorage(const Sp<Value>& v, time_point t, time_point e, duration te)
            : data(v)
            , created(t)
            , expiration(e)
+            , typeExpiration(te)
        {}
    };

@@ -166,15 +178,22 @@ private:
    {
        std::vector<Sp<Value>> nvals;
        for (const auto& value : new_values) {
+            auto te = types.getType(value->type).expiration;
            auto v = values.find(value->id);
            if (v == values.end()) {
                // new value
                nvals.emplace_back(value);
-                values.emplace(value->id, CacheValueStorage(value, now, now + types.getType(value->type).expiration));
+                values.emplace(value->id, CacheValueStorage(value, now, now + te, te));
            } else {
                // refreshed value
                v->second.created = now;
-                v->second.expiration = now + types.getType(v->second.data->type).expiration;
+                v->second.expiration = now + te;
+                v->second.typeExpiration = te;
+                if (value->seq > v->second.data->seq) {
+                    // edited value (same rule as OpValueCache): emit as add
+                    v->second.data = value;
+                    nvals.emplace_back(value);
+                }
            }
        }
        CallbackQueue ret;
@@ -201,8 +220,10 @@ private:
        auto v = values.find(vid);
        if (v == values.end())
            return;
+        auto te = types.getType(v->second.data->type).expiration;
        v->second.created = now;
-        v->second.expiration = now + types.getType(v->second.data->type).expiration;
+        v->second.expiration = now + te;
+        v->second.typeExpiration = te;
    }
 };

@@ -8,7 +8,6 @@

 #include <opendht/dhtrunner.h>
 #include <opendht/dht_proxy_server.h>
-#include <opendht/log.h>

 namespace test {

@@ -17,10 +16,21 @@ class DhtProxyTester : public CppUnit::TestFixture
    CPPUNIT_TEST_SUITE(DhtProxyTester);
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenValueEdit);
+    CPPUNIT_TEST(testPushListenValueEdit);
    CPPUNIT_TEST(testResubscribeGetValues);
+    CPPUNIT_TEST(testPushNotification);
+    CPPUNIT_TEST(testProxyServerClientFullChain);
    CPPUNIT_TEST(testPutGet40KChars);
    CPPUNIT_TEST(testFuzzy);
    CPPUNIT_TEST(testShutdownStop);
+    CPPUNIT_TEST(testGetAfterListen);
+    CPPUNIT_TEST(testListenDuplicatePut);
+    CPPUNIT_TEST(testPushMultiValueEditExpire);
+    CPPUNIT_TEST(testPushRefreshNoSpuriousExpire);
+    CPPUNIT_TEST(testProxyListenEditChain);
+    CPPUNIT_TEST(testPushRapidEditsNoPhantom);
+    CPPUNIT_TEST(testPushRefreshAfterEditKeepsLatest);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -41,11 +51,28 @@ public:
     * Test listen
     */
    void testListen();
+    /**
+     * Test that editing a value only triggers an add callback via proxy, no expire
+     */
+    void testListenValueEdit();
+    /**
+     * Test that editing a value only triggers an add callback via push,
+     * no phantom expire, and that real expiration is received.
+     */
+    void testPushListenValueEdit();
    /**
     * When a proxy redo a subscribe on the proxy
     * it should retrieve existant values
     */
    void testResubscribeGetValues();
+    /**
+     * Test push notification mechanism and OpValueCache
+     */
+    void testPushNotification();
+    /**
+     * Test end-to-end chain through proxy server and proxy client
+     */
+    void testProxyServerClientFullChain();
    /**
     * Test MTU put/get on dht
     */
@@ -54,6 +81,13 @@ public:
    void testFuzzy();

    void testShutdownStop();
+    void testGetAfterListen();
+    void testListenDuplicatePut();
+    void testPushMultiValueEditExpire();
+    void testPushRefreshNoSpuriousExpire();
+    void testProxyListenEditChain();
+    void testPushRapidEditsNoPhantom();
+    void testPushRefreshAfterEditKeepsLatest();

 private:
    dht::DhtRunner::Config clientConfig {};
@@ -37,7 +37,7 @@ DhtProxyClientTester::testResubscribeUsesKeyRoute()
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
    DhtProxyClient client({}, {}, [] {}, "http://127.0.0.1:8080", "OpenDHT-Test", "client-id", "push-token");

-    DhtProxyClient::Listener listener {OpValueCache([](const std::vector<Sp<Value>>&, bool) { return true; })};
+    DhtProxyClient::Listener listener {ValueCallback([](const std::vector<Sp<Value>>&, bool) { return true; })};
    listener.opstate = std::make_shared<DhtProxyClient::OperationState>();
    listener.cb = [](const std::vector<Sp<Value>>&, bool, system_clock::time_point) {
        return true;
@@ -67,7 +67,7 @@ DhtProxyClientTester::testSetPushNotificationTokenResubscribesWithNewToken()
    auto& search = client.searches_[key];
    auto [it, inserted] = search.listeners.emplace(std::piecewise_construct,
                                                   std::forward_as_tuple(1),
-                                                   std::forward_as_tuple(OpValueCache(
+                                                   std::forward_as_tuple(ValueCallback(
                                                       [](const std::vector<Sp<Value>>&, bool) { return true; })));
    CPPUNIT_ASSERT(inserted);

@@ -91,6 +91,42 @@ DhtRunnerTester::testGetPut()
    CPPUNIT_ASSERT(vals.front()->data == val_data);
 }

+void
+DhtRunnerTester::testStoreEmptyValue()
+{
+    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;
+    config.dht_config.node_config.max_store_size = 100; // Small size to force expireStore()
+    config.dht_config.node_config.max_store_keys = -1;
+
+    dht::DhtRunner testNode;
+    testNode.run(0, config);
+    testNode.bootstrap(node1.getBound());
+
+    auto key = testNode.getId();
+    auto val = std::make_shared<dht::Value>();
+    val->data.clear(); // Ensure size is 0
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, val->size());
+
+    std::promise<bool> p;
+    testNode.put(key, val, [&](bool ok) { p.set_value(ok); });
+    p.get_future().get(); // Wait for put to finish
+
+    // Put a large value to exceed max_store_size and trigger expireStore()
+    auto key2 = dht::InfoHash::get("large_value_test");
+    auto val2 = std::make_shared<dht::Value>();
+    val2->data = std::vector<uint8_t>(200, 0); // 200 bytes
+    std::promise<bool> p2;
+    testNode.put(key2, val2, [&](bool ok) { p2.set_value(ok); });
+    p2.get_future().get(); // Wait for put to finish
+
+    // Wait a bit to ensure the value is processed
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+    testNode.join();
+}
+
 void
 DhtRunnerTester::testPutDuplicate()
 {
@@ -587,6 +623,115 @@ DhtRunnerTester::testSavedPortFallback()
    std::remove((config.dht_config.node_config.persist_path + "_port.txt").c_str());
 }

+void
+DhtRunnerTester::testListenValueEdit()
+{
+    // Register an editable type on both nodes so the DHT allows edits.
+    static constexpr dht::ValueType::Id EDITABLE_TYPE_ID = 9999;
+    const dht::ValueType editableType {EDITABLE_TYPE_ID,
+                                       "editable-test",
+                                       std::chrono::seconds(2),
+                                       dht::ValueType::DEFAULT_STORE_POLICY,
+                                       [](dht::InfoHash,
+                                          const std::shared_ptr<dht::Value>&,
+                                          std::shared_ptr<dht::Value>&,
+                                          const dht::InfoHash&,
+                                          const dht::SockAddr&) { return true; }};
+    node1.registerType(editableType);
+    node2.registerType(editableType);
+
+    auto key = dht::InfoHash::get("listenValueEdit");
+    auto identity = dht::crypto::generateIdentity("ListenEditTester");
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::vector<std::pair<dht::Value::Id, bool>> events; // (id, expired?)
+
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        std::lock_guard lk(mtx);
+        for (const auto& v : vals) {
+            events.emplace_back(v->id, expired);
+        }
+        cv.notify_all();
+        return true;
+    });
+
+    // Put initial value (signed, editable type)
+    auto val1 = std::make_shared<dht::Value>("version1");
+    val1->type = EDITABLE_TYPE_ID;
+    val1->id = 42;
+    val1->seq = 1;
+    val1->sign(*identity.first);
+    {
+        std::promise<bool> p;
+        node2.put(key, val1, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for the first value to be received
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const auto& e) { return !e.second; });
+        }));
+    }
+
+    // Edit the value (same id, higher seq)
+    auto val2 = std::make_shared<dht::Value>("version2");
+    val2->type = EDITABLE_TYPE_ID;
+    val2->id = 42;
+    val2->seq = 2;
+    val2->sign(*identity.first);
+    {
+        std::promise<bool> p;
+        node2.put(key, val2, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for the edited value to be received
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            int adds = 0;
+            for (const auto& e : events)
+                if (!e.second)
+                    adds++;
+            return adds >= 2;
+        }));
+    }
+
+    // Give some extra time for any stray callbacks
+    std::this_thread::sleep_for(500ms);
+
+    // Verify: we should have exactly 2 add events, no expire events.
+    // On value edition the listener should see only the new version
+    // (as an add), not an expiration of the old version.
+    {
+        std::lock_guard lk(mtx);
+        int addCount = 0;
+        int expireCount = 0;
+        for (const auto& e : events) {
+            if (e.second)
+                expireCount++;
+            else
+                addCount++;
+        }
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Should receive 2 add callbacks (original + edit)", 2, addCount);
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Should receive no expire callbacks on edit", 0, expireCount);
+    }
+
+    // Now wait for the edited value to actually expire.
+    // Type expiration is 2s. The cache may add a grace period.
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT_MESSAGE("Edited value should eventually expire", cv.wait_for(lk, 60s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const auto& e) { return e.second; });
+        }));
+    }
+
+    node1.cancelListen(key, ftoken.get());
+}
+
 void
 DhtRunnerTester::testListen()
 {
@@ -799,7 +944,7 @@ DhtRunnerTester::testIdOps()
        valueCountEdit++;
        cv.notify_all();
    });
-    node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool /*expired*/) {
+    auto listenToken = node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool /*expired*/) {
        for (const auto& v : values) {
            if (v->seq == 0)
                CPPUNIT_ASSERT_EQUAL("v1"s, dht::unpackMsg<std::string>(v->data));
@@ -843,6 +988,8 @@ DhtRunnerTester::testIdOps()
    });
    std::unique_lock lk(mutex);
    CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCountEdit == 4u; }));
+    lk.unlock();
+    node2.cancelListen(key2, listenToken.get());
 }

 void
@@ -933,4 +1080,451 @@ DhtRunnerTester::testMultithread()
    CPPUNIT_ASSERT_EQUAL(2 * N, putOkCount);
 }

+void
+DhtRunnerTester::testGetAfterListen()
+{
+    // Test that get() after a synced listen delivers values exactly once
+    // from the cache, and that the GetCallback return value is honored.
+    auto key = dht::InfoHash::get("getAfterListen");
+
+    // Put a value first
+    auto val = std::make_shared<dht::Value>("cached_val");
+    {
+        std::promise<bool> p;
+        node2.put(key, val, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Start a listen on node1 and wait for it to sync
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::atomic_int listenAddCount {0};
+
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (!expired)
+            listenAddCount += vals.size();
+        cv.notify_all();
+        return true;
+    });
+
+    // Wait for listen to receive the value
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return listenAddCount.load() >= 1; }));
+    }
+
+    // Now do a get — should be served from the synced cache
+    auto vals = getFutureValue(node1.get(key));
+    CPPUNIT_ASSERT_MESSAGE("get() should return values", not vals.empty());
+    CPPUNIT_ASSERT(val->data == vals.front()->data);
+
+    // The listen callback should not have been called again by the get
+    std::this_thread::sleep_for(200ms);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Listen should not fire extra callbacks from get()", 1, listenAddCount.load());
+
+    node1.cancelListen(key, ftoken.get());
+}
+
+void
+DhtRunnerTester::testListenDuplicatePut()
+{
+    // Test that putting the exact same value multiple times only triggers
+    // one add callback in the listener (no duplicate delivery).
+    auto key = dht::InfoHash::get("listenDupPut");
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::vector<std::pair<dht::Value::Id, bool>> events;
+
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        std::lock_guard lk(mtx);
+        for (const auto& v : vals)
+            events.emplace_back(v->id, expired);
+        cv.notify_all();
+        return true;
+    });
+
+    // Put the same value 3 times
+    auto val = std::make_shared<dht::Value>("duplicate_test");
+    val->id = 123;
+    for (int i = 0; i < 3; i++) {
+        std::promise<bool> p;
+        node2.put(key, val, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for at least 1 add to arrive
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const auto& e) { return !e.second; });
+        }));
+    }
+
+    // Give time for any spurious callbacks
+    std::this_thread::sleep_for(500ms);
+
+    // Should have exactly 1 add, 0 expires
+    {
+        std::lock_guard lk(mtx);
+        int addCount = 0, expireCount = 0;
+        for (const auto& e : events) {
+            if (e.second)
+                expireCount++;
+            else
+                addCount++;
+        }
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Duplicate puts should produce only 1 add callback", 1, addCount);
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("No expire callbacks expected", 0, expireCount);
+    }
+
+    node1.cancelListen(key, ftoken.get());
+}
+
+void
+DhtRunnerTester::testListenMultiSourceExpire()
+{
+    // Test that when multiple nodes store the same value, the listener sees
+    // exactly 1 add (no duplicates). When the value expires naturally, the
+    // listener sees exactly 1 expire.
+    static constexpr dht::ValueType::Id SHORT_TYPE_ID = 8888;
+    const dht::ValueType shortType {SHORT_TYPE_ID,
+                                    "short-lived",
+                                    std::chrono::seconds(2),
+                                    dht::ValueType::DEFAULT_STORE_POLICY,
+                                    [](dht::InfoHash,
+                                       const std::shared_ptr<dht::Value>&,
+                                       std::shared_ptr<dht::Value>&,
+                                       const dht::InfoHash&,
+                                       const dht::SockAddr&) { return true; }};
+    node1.registerType(shortType);
+    node2.registerType(shortType);
+
+    auto key = dht::InfoHash::get("multiSourceExpire");
+    auto identity = dht::crypto::generateIdentity("MultiSourceTester");
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    std::vector<std::pair<dht::Value::Id, bool>> events;
+
+    // node1 listens; node1 and node2 put
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        std::lock_guard lk(mtx);
+        for (const auto& v : vals)
+            events.emplace_back(v->id, expired);
+        cv.notify_all();
+        return true;
+    });
+
+    // Both node1 and node2 put the same value (same id, same seq)
+    auto val = std::make_shared<dht::Value>("multi_source");
+    val->type = SHORT_TYPE_ID;
+    val->id = 55;
+    val->seq = 1;
+    val->sign(*identity.first);
+
+    {
+        std::promise<bool> p1, p2;
+        node1.put(key, val, [&](bool ok) { p1.set_value(ok); });
+        node2.put(key, val, [&](bool ok) { p2.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p1.get_future()));
+        CPPUNIT_ASSERT(getFutureValue(p2.get_future()));
+    }
+
+    // Wait for at least one add
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const auto& e) { return !e.second; });
+        }));
+    }
+
+    // Allow time for duplicate callbacks
+    std::this_thread::sleep_for(1s);
+
+    // Should have exactly 1 add (deduplication in OpValueCache)
+    {
+        std::lock_guard lk(mtx);
+        int addCount = 0;
+        for (const auto& e : events)
+            if (!e.second)
+                addCount++;
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Multiple sources should produce only 1 add callback", 1, addCount);
+    }
+
+    // Wait for natural expiration (type is 2s + grace period)
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT_MESSAGE("Value should eventually expire", cv.wait_for(lk, 60s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const auto& e) { return e.second; });
+        }));
+    }
+
+    // Should have exactly 1 expire (no duplicate expires from multiple sources)
+    {
+        std::lock_guard lk(mtx);
+        int expireCount = 0;
+        for (const auto& e : events)
+            if (e.second)
+                expireCount++;
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Should receive exactly 1 expire callback", 1, expireCount);
+    }
+
+    node1.cancelListen(key, ftoken.get());
+}
+
+void
+DhtRunnerTester::testListenEditChainThenExpire()
+{
+    // Chain of 5 rapid edits (seq 1→5) on the same value, then natural expiration.
+    // Must see 5 adds (one per edit), 0 expires during edits, then exactly 1 expire.
+    static constexpr dht::ValueType::Id CHAIN_TYPE_ID = 7654;
+    const dht::ValueType chainType {CHAIN_TYPE_ID,
+                                    "chain-edit",
+                                    std::chrono::seconds(2),
+                                    dht::ValueType::DEFAULT_STORE_POLICY,
+                                    [](dht::InfoHash,
+                                       const std::shared_ptr<dht::Value>&,
+                                       std::shared_ptr<dht::Value>&,
+                                       const dht::InfoHash&,
+                                       const dht::SockAddr&) { return true; }};
+    node1.registerType(chainType);
+    node2.registerType(chainType);
+
+    auto key = dht::InfoHash::get("editChainExpire");
+    auto identity = dht::crypto::generateIdentity("ChainEditTester");
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    struct Event
+    {
+        dht::Value::Id id;
+        uint16_t seq;
+        bool expired;
+    };
+    std::vector<Event> events;
+
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        std::lock_guard lk(mtx);
+        for (const auto& v : vals)
+            events.push_back({v->id, v->seq, expired});
+        cv.notify_all();
+        return true;
+    });
+
+    // Rapid chain of 5 edits — wait for the first to arrive before continuing
+    constexpr int NUM_EDITS = 5;
+    {
+        auto v = std::make_shared<dht::Value>("version1");
+        v->type = CHAIN_TYPE_ID;
+        v->id = 99;
+        v->seq = 1;
+        v->sign(*identity.first);
+        std::promise<bool> p;
+        node2.put(key, v, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for first add to arrive before sending more edits
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const Event& e) { return !e.expired; });
+        }));
+    }
+
+    // Now send remaining edits rapidly
+    for (int seq = 2; seq <= NUM_EDITS; seq++) {
+        auto v = std::make_shared<dht::Value>("version" + std::to_string(seq));
+        v->type = CHAIN_TYPE_ID;
+        v->id = 99;
+        v->seq = seq;
+        v->sign(*identity.first);
+        std::promise<bool> p;
+        node2.put(key, v, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for edits to arrive (at least 2: initial + one or more updates)
+    // Note: rapid edits may coalesce — the DHT stores only the latest seq,
+    // so the listener may not see every intermediate version.
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&] {
+            int adds = 0;
+            for (const auto& e : events)
+                if (!e.expired)
+                    adds++;
+            return adds >= 2;
+        }));
+    }
+
+    // Wait a bit more for any remaining events
+    std::this_thread::sleep_for(1s);
+
+    // Verify: multiple adds, 0 expires during edits (no phantom expiry)
+    int totalAdds;
+    {
+        std::lock_guard lk(mtx);
+        totalAdds = 0;
+        int expireCount = 0;
+        for (const auto& e : events) {
+            if (e.expired)
+                expireCount++;
+            else
+                totalAdds++;
+        }
+        CPPUNIT_ASSERT_MESSAGE("Should receive at least 2 add callbacks", totalAdds >= 2);
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("No expires during edit chain", 0, expireCount);
+
+        // Verify the latest add has the highest seq
+        uint16_t maxSeq = 0;
+        for (const auto& e : events)
+            if (!e.expired && e.seq > maxSeq)
+                maxSeq = e.seq;
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Latest add should have the final seq", (uint16_t) NUM_EDITS, maxSeq);
+    }
+
+    // Wait for natural expiration
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT_MESSAGE("Value should eventually expire", cv.wait_for(lk, 60s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const Event& e) { return e.expired; });
+        }));
+    }
+
+    // Verify exactly 1 expire with the latest seq
+    {
+        std::lock_guard lk(mtx);
+        int expireCount = 0;
+        uint16_t expiredSeq = 0;
+        for (const auto& e : events) {
+            if (e.expired) {
+                expireCount++;
+                expiredSeq = e.seq;
+            }
+        }
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Should receive exactly 1 expire", 1, expireCount);
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Expired value should be the latest version", (uint16_t) NUM_EDITS, expiredSeq);
+    }
+
+    node1.cancelListen(key, ftoken.get());
+}
+
+void
+DhtRunnerTester::testListenMultiValuePartialExpire()
+{
+    // Two values with different types: one short-lived (2s), one long-lived (60s).
+    // The short one expires while the long one stays. Verify isolation:
+    // no spurious expire for the long-lived value.
+    static constexpr dht::ValueType::Id SHORT_TYPE_ID = 3001;
+    static constexpr dht::ValueType::Id LONG_TYPE_ID = 3002;
+    const dht::ValueType shortType {SHORT_TYPE_ID,
+                                    "short-partial",
+                                    std::chrono::seconds(2),
+                                    dht::ValueType::DEFAULT_STORE_POLICY,
+                                    [](dht::InfoHash,
+                                       const std::shared_ptr<dht::Value>&,
+                                       std::shared_ptr<dht::Value>&,
+                                       const dht::InfoHash&,
+                                       const dht::SockAddr&) { return true; }};
+    const dht::ValueType longType {LONG_TYPE_ID,
+                                   "long-partial",
+                                   std::chrono::seconds(600),
+                                   dht::ValueType::DEFAULT_STORE_POLICY,
+                                   [](dht::InfoHash,
+                                      const std::shared_ptr<dht::Value>&,
+                                      std::shared_ptr<dht::Value>&,
+                                      const dht::InfoHash&,
+                                      const dht::SockAddr&) { return true; }};
+    node1.registerType(shortType);
+    node1.registerType(longType);
+    node2.registerType(shortType);
+    node2.registerType(longType);
+
+    auto key = dht::InfoHash::get("multiValuePartialExpire");
+    auto identity = dht::crypto::generateIdentity("PartialExpireTester");
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    struct Event
+    {
+        dht::Value::Id id;
+        bool expired;
+    };
+    std::vector<Event> events;
+
+    auto ftoken = node1.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        std::lock_guard lk(mtx);
+        for (const auto& v : vals)
+            events.push_back({v->id, expired});
+        cv.notify_all();
+        return true;
+    });
+
+    // Put short-lived value (id=1)
+    auto shortVal = std::make_shared<dht::Value>("short");
+    shortVal->type = SHORT_TYPE_ID;
+    shortVal->id = 1;
+    shortVal->seq = 1;
+    shortVal->sign(*identity.first);
+    {
+        std::promise<bool> p;
+        node2.put(key, shortVal, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Put long-lived value (id=2)
+    auto longVal = std::make_shared<dht::Value>("long");
+    longVal->type = LONG_TYPE_ID;
+    longVal->id = 2;
+    longVal->seq = 1;
+    longVal->sign(*identity.first);
+    {
+        std::promise<bool> p;
+        node2.put(key, longVal, [&](bool ok) { p.set_value(ok); });
+        CPPUNIT_ASSERT(getFutureValue(p.get_future()));
+    }
+
+    // Wait for both adds
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] {
+            int adds = 0;
+            for (const auto& e : events)
+                if (!e.expired)
+                    adds++;
+            return adds >= 2;
+        }));
+    }
+
+    // Wait for short-lived value to expire
+    {
+        std::unique_lock lk(mtx);
+        CPPUNIT_ASSERT_MESSAGE("Short-lived value should expire", cv.wait_for(lk, 60s, [&] {
+            return std::any_of(events.begin(), events.end(), [](const Event& e) { return e.id == 1 && e.expired; });
+        }));
+    }
+
+    // Give extra time for any spurious callbacks
+    std::this_thread::sleep_for(1s);
+
+    // Verify: value 1 expired, value 2 did NOT expire
+    {
+        std::lock_guard lk(mtx);
+        bool longExpired = std::any_of(events.begin(), events.end(), [](const Event& e) {
+            return e.id == 2 && e.expired;
+        });
+        CPPUNIT_ASSERT_MESSAGE("Long-lived value should NOT be expired", !longExpired);
+
+        int shortExpires = 0;
+        for (const auto& e : events)
+            if (e.id == 1 && e.expired)
+                shortExpires++;
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Short-lived value should expire exactly once", 1, shortExpires);
+    }
+
+    node1.cancelListen(key, ftoken.get());
+}
+
 } // namespace test
@@ -18,7 +18,9 @@ class DhtRunnerTester : public CppUnit::TestFixture
    CPPUNIT_TEST(testPutDuplicate);
    CPPUNIT_TEST(testPutOverride);
    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenValueEdit);
    CPPUNIT_TEST(testListenLotOfBytes);
+    CPPUNIT_TEST(testStoreEmptyValue);
    CPPUNIT_TEST(testIdOps);
    CPPUNIT_TEST(testImportValuesPreservesRemoteQuota);
    CPPUNIT_TEST(testExportValuesUsesSystemTime);
@@ -29,6 +31,11 @@ class DhtRunnerTester : public CppUnit::TestFixture
    CPPUNIT_TEST(testBootstrapMissingNodeThenPutFails);
    CPPUNIT_TEST(testShutdownCompletesWithPendingPut);
    CPPUNIT_TEST(testSavedPortFallback);
+    CPPUNIT_TEST(testGetAfterListen);
+    CPPUNIT_TEST(testListenDuplicatePut);
+    CPPUNIT_TEST(testListenMultiSourceExpire);
+    CPPUNIT_TEST(testListenEditChainThenExpire);
+    CPPUNIT_TEST(testListenMultiValuePartialExpire);
    CPPUNIT_TEST_SUITE_END();

    dht::DhtRunner node1 {};
@@ -76,10 +83,23 @@ public:
    void testBootstrapMissingNodeThenPutFails();
    void testShutdownCompletesWithPendingPut();
    void testSavedPortFallback();
+    void testGetAfterListen();
+    void testListenDuplicatePut();
+    void testListenMultiSourceExpire();
+    void testListenEditChainThenExpire();
+    void testListenMultiValuePartialExpire();
+    /**
+     * Test that editing a value only triggers an add callback, no expire
+     */
+    void testListenValueEdit();
    /**
     * Test listen method with lot of datas
     */
    void testListenLotOfBytes();
+    /**
+     * Test storing an empty value
+     */
+    void testStoreEmptyValue();
    /**
     * Test multithread
     */
@@ -0,0 +1,674 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+
+#include "test_opcache.h"
+
+// Hack to test internal class
+#include "../src/op_cache.cpp"
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(OpCacheTester);
+
+namespace {
+
+using Values = std::vector<std::shared_ptr<dht::Value>>;
+
+struct Event
+{
+    dht::Value::Id id;
+    uint16_t seq;
+    bool expired;
+};
+
+std::shared_ptr<dht::Value>
+makeValue(dht::Value::Id id, uint16_t seq, std::string data)
+{
+    auto v = std::make_shared<dht::Value>();
+    v->id = id;
+    v->seq = seq;
+    v->data = {data.begin(), data.end()};
+    return v;
+}
+
+dht::ValueCallback
+recordEvents(std::vector<Event>& events)
+{
+    return [&events](const Values& vals, bool expired) {
+        for (const auto& v : vals)
+            events.push_back({v->id, v->seq, expired});
+        return true;
+    };
+}
+
+unsigned
+count(const std::vector<Event>& events, bool expired)
+{
+    unsigned n = 0;
+    for (const auto& e : events)
+        if (e.expired == expired)
+            n++;
+    return n;
+}
+
+} // namespace
+
+void
+OpCacheTester::setUp()
+{}
+
+void
+OpCacheTester::tearDown()
+{}
+
+void
+OpCacheTester::testBasicAddExpire()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+
+    auto v1 = makeValue(10, 0, "A");
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+    CPPUNIT_ASSERT(cache.get(10));
+
+    std::vector<dht::Value::Id> ids = {v1->id};
+    cache.onValuesExpired(ids);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT(!cache.get(10));
+}
+
+void
+OpCacheTester::testMultipleSources()
+{
+    // Two sources report the same value: it must survive until both have
+    // expired it, and the user must see exactly one add and one expire.
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v = makeValue(20, 0, "B");
+    cache.onValuesAdded({v}); // source A
+    cache.onValuesAdded({v}); // source B
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+
+    cache.onValuesExpired({v}); // source A goes away
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Value still held by source B", (size_t) 1, cache.size());
+    CPPUNIT_ASSERT_EQUAL(0u, count(events, true));
+
+    cache.onValuesExpired({v}); // source B goes away
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+}
+
+void
+OpCacheTester::testEditResetsRefCount()
+{
+    // A and B hold v1. A reports the edit v2: only A is known to hold v2, so
+    // the refcount restarts at 1. B then confirms v2 (refcount 2).
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(1, 1, "data1");
+    auto v2 = makeValue(1, 2, "data2");
+
+    cache.onValuesAdded({v1}); // A
+    cache.onValuesAdded({v1}); // B
+    cache.onValuesAdded({v2}); // A edits
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+    CPPUNIT_ASSERT_EQUAL(v2->seq, cache.get(1)->seq);
+    CPPUNIT_ASSERT_EQUAL(2u, count(events, false));
+
+    cache.onValuesExpired({v2}); // A expires v2 -> nobody known to hold it
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Edit reset the refcount to the editing source", (size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+
+    // Same scenario, but B confirms v2 before A expires it.
+    events.clear();
+    cache.onValuesAdded({v1}); // A
+    cache.onValuesAdded({v1}); // B
+    cache.onValuesAdded({v2}); // A edits
+    cache.onValuesAdded({v2}); // B confirms
+    cache.onValuesExpired({v2});
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Still held by B", (size_t) 1, cache.size());
+    cache.onValuesExpired({v2});
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(2u, count(events, false));
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+}
+
+void
+OpCacheTester::testEditChainMultiSource()
+{
+    // Rapid edit chain seen from two sources with interleaving: the user
+    // must see each version once as an add and a single final expire.
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(7, 1, "v1");
+    auto v2 = makeValue(7, 2, "v2");
+    auto v3 = makeValue(7, 3, "v3");
+
+    cache.onValuesAdded({v1}); // A
+    cache.onValuesAdded({v1}); // B
+    cache.onValuesAdded({v2}); // A
+    cache.onValuesAdded({v3}); // A (B skipped v2)
+    cache.onValuesAdded({v2}); // B, late and stale: ignored
+    cache.onValuesAdded({v3}); // B
+    CPPUNIT_ASSERT_EQUAL(3u, count(events, false));
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 3, cache.get(7)->seq);
+
+    cache.onValuesExpired({v3}); // A
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+    cache.onValuesExpired({v3}); // B
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 3, events.back().seq);
+}
+
+void
+OpCacheTester::testStaleAddIgnored()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+
+    auto v1 = makeValue(30, 10, "C");
+    auto v2 = makeValue(30, 20, "D");
+    auto v3 = makeValue(30, 5, "E");
+
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 10, cache.get(30)->seq);
+    cache.onValuesAdded({v2});
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 20, cache.get(30)->seq);
+
+    // Older version from a lagging source: neither stored nor counted
+    cache.onValuesAdded({v3});
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 20, cache.get(30)->seq);
+    cache.onValuesExpired({v2});
+    CPPUNIT_ASSERT_MESSAGE("Stale add must not have incremented the refcount", !cache.get(30));
+}
+
+void
+OpCacheTester::testStaleExpireIgnored()
+{
+    // A source that still holds an older version expiring it must not
+    // affect the newer version held by others.
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(31, 1, "old");
+    auto v2 = makeValue(31, 2, "new");
+
+    cache.onValuesAdded({v1}); // A
+    cache.onValuesAdded({v1}); // B
+    cache.onValuesAdded({v2}); // A edits; B never receives v2
+
+    cache.onValuesExpired({v1}); // B expires its stale copy
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+    CPPUNIT_ASSERT_EQUAL(0u, count(events, true));
+
+    cache.onValuesExpired({v2}); // A
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 2, events.back().seq);
+}
+
+void
+OpCacheTester::testCallbacks()
+{
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(40, 0, "");
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+
+    // Duplicate from another source: counted, but not re-delivered
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+
+    auto v2 = makeValue(40, 100, "MOD");
+    cache.onValuesAdded({v2});
+    CPPUNIT_ASSERT_EQUAL(2u, count(events, false));
+
+    // The edit reset the refcount: one expire removes it
+    cache.onValuesExpired({v2});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+OpCacheTester::testFilters()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+
+    auto v1 = makeValue(50, 0, "");
+    v1->type = 1;
+    auto v2 = makeValue(51, 0, "");
+    v2->type = 2;
+    cache.onValuesAdded({v1, v2});
+
+    auto res1 = cache.get(dht::Value::TypeFilter(1));
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, res1.size());
+    CPPUNIT_ASSERT_EQUAL(v1->id, res1[0]->id);
+
+    auto res2 = cache.get(dht::Value::TypeFilter(2));
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, res2.size());
+    CPPUNIT_ASSERT_EQUAL(v2->id, res2[0]->id);
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 2, cache.getValues().size());
+}
+
+void
+OpCacheTester::testSyncStatus()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+
+    CPPUNIT_ASSERT(!cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+    CPPUNIT_ASSERT(!cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::SYNCED);
+    CPPUNIT_ASSERT(cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+    CPPUNIT_ASSERT(!cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::SYNCED);
+    CPPUNIT_ASSERT(cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::UNSYNCED);
+    CPPUNIT_ASSERT(!cache.isSynced());
+    cache.onNodeChanged(dht::ListenSyncStatus::REMOVED);
+    CPPUNIT_ASSERT(cache.isSynced());
+}
+
+void
+OpCacheTester::testGetWhileSynced()
+{
+    dht::SearchCache searchCache;
+    int getCallCount = 0;
+    bool doneCalled = false;
+    bool doneSuccess = false;
+
+    auto v1 = makeValue(100, 0, "X");
+    auto v2 = makeValue(101, 0, "Y");
+    auto query = std::make_shared<dht::Query>();
+
+    auto token = searchCache.listen([](const Values&, bool) { return true; },
+                                    query,
+                                    {},
+                                    [&](const std::shared_ptr<dht::Query>&,
+                                        dht::ValueCallback vcb,
+                                        dht::SyncCallback scb) -> size_t {
+                                        scb(dht::ListenSyncStatus::ADDED);
+                                        vcb({v1, v2}, false);
+                                        scb(dht::ListenSyncStatus::SYNCED);
+                                        return 1;
+                                    });
+    CPPUNIT_ASSERT(token != 0);
+
+    bool result = searchCache.get(
+        {},
+        query,
+        [&](const Values& vals) {
+            getCallCount++;
+            CPPUNIT_ASSERT_EQUAL((size_t) 2, vals.size());
+            return true;
+        },
+        [&](bool success, const std::vector<std::shared_ptr<dht::Node>>&) {
+            doneCalled = true;
+            doneSuccess = success;
+        });
+
+    CPPUNIT_ASSERT_MESSAGE("get() should be served from a synced cache", result);
+    CPPUNIT_ASSERT_EQUAL(1, getCallCount);
+    CPPUNIT_ASSERT(doneCalled);
+    CPPUNIT_ASSERT(doneSuccess);
+}
+
+void
+OpCacheTester::testGetWhileNotSynced()
+{
+    // An unsynced cache must not answer a get: the caller falls back to the
+    // network and values are delivered once, from the network only.
+    dht::SearchCache searchCache;
+    std::vector<dht::Value::Id> deliveries;
+
+    auto v1 = makeValue(200, 0, "A");
+    auto query = std::make_shared<dht::Query>();
+    dht::SyncCallback storedScb;
+
+    auto token = searchCache.listen([](const Values&, bool) { return true; },
+                                    query,
+                                    {},
+                                    [&](const std::shared_ptr<dht::Query>&,
+                                        dht::ValueCallback vcb,
+                                        dht::SyncCallback scb) -> size_t {
+                                        storedScb = scb;
+                                        scb(dht::ListenSyncStatus::ADDED);
+                                        vcb({v1}, false);
+                                        return 1;
+                                    });
+    CPPUNIT_ASSERT(token != 0);
+
+    bool doneCalled = false;
+    auto gcb = [&](const Values& vals) {
+        for (const auto& v : vals)
+            deliveries.emplace_back(v->id);
+        return true;
+    };
+    auto dcb = [&](bool, const std::vector<std::shared_ptr<dht::Node>>&) {
+        doneCalled = true;
+    };
+
+    CPPUNIT_ASSERT(!searchCache.get({}, query, gcb, dcb));
+    CPPUNIT_ASSERT(deliveries.empty());
+    CPPUNIT_ASSERT(!doneCalled);
+
+    // Simulated network reply
+    gcb({v1});
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, deliveries.size());
+
+    storedScb(dht::ListenSyncStatus::SYNCED);
+    deliveries.clear();
+    CPPUNIT_ASSERT(searchCache.get({}, query, gcb, dcb));
+    CPPUNIT_ASSERT(doneCalled);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, deliveries.size());
+}
+
+void
+OpCacheTester::testGetEmptySynced()
+{
+    dht::SearchCache searchCache;
+    int getCallCount = 0;
+    bool doneCalled = false;
+    bool doneSuccess = false;
+    auto query = std::make_shared<dht::Query>();
+
+    auto token = searchCache.listen([](const Values&, bool) { return true; },
+                                    query,
+                                    {},
+                                    [&](const std::shared_ptr<dht::Query>&,
+                                        dht::ValueCallback,
+                                        dht::SyncCallback scb) -> size_t {
+                                        scb(dht::ListenSyncStatus::ADDED);
+                                        scb(dht::ListenSyncStatus::SYNCED);
+                                        return 1;
+                                    });
+    CPPUNIT_ASSERT(token != 0);
+
+    bool result = searchCache.get(
+        {},
+        query,
+        [&](const Values&) {
+            getCallCount++;
+            return true;
+        },
+        [&](bool success, const std::vector<std::shared_ptr<dht::Node>>&) {
+            doneCalled = true;
+            doneSuccess = success;
+        });
+
+    CPPUNIT_ASSERT(result);
+    CPPUNIT_ASSERT_EQUAL(0, getCallCount);
+    CPPUNIT_ASSERT(doneCalled);
+    CPPUNIT_ASSERT(doneSuccess);
+}
+
+void
+OpCacheTester::testGetWithoutCallback()
+{
+    // Dht::query() passes no GetCallback: the cache cannot answer it and
+    // must let the network path run instead of calling an empty function.
+    dht::SearchCache searchCache;
+    auto v1 = makeValue(300, 0, "Q");
+    auto query = std::make_shared<dht::Query>();
+
+    searchCache.listen([](const Values&, bool) { return true; },
+                       query,
+                       {},
+                       [&](const std::shared_ptr<dht::Query>&, dht::ValueCallback vcb, dht::SyncCallback scb) -> size_t {
+                           scb(dht::ListenSyncStatus::ADDED);
+                           vcb({v1}, false);
+                           scb(dht::ListenSyncStatus::SYNCED);
+                           return 1;
+                       });
+
+    bool doneCalled = false;
+    CPPUNIT_ASSERT(!searchCache.get({}, query, {}, [&](bool, const std::vector<std::shared_ptr<dht::Node>>&) {
+        doneCalled = true;
+    }));
+    CPPUNIT_ASSERT(!doneCalled);
+}
+
+void
+OpCacheTester::testGetCallbackReturnsFalse()
+{
+    // A consumer returning false means "I have enough", not a failure.
+    dht::SearchCache searchCache;
+    auto v1 = makeValue(1000, 0, "X");
+    auto query = std::make_shared<dht::Query>();
+
+    searchCache.listen([](const Values&, bool) { return true; },
+                       query,
+                       {},
+                       [&](const std::shared_ptr<dht::Query>&, dht::ValueCallback vcb, dht::SyncCallback scb) -> size_t {
+                           scb(dht::ListenSyncStatus::ADDED);
+                           vcb({v1}, false);
+                           scb(dht::ListenSyncStatus::SYNCED);
+                           return 1;
+                       });
+
+    bool doneCalled = false;
+    bool doneSuccess = false;
+    int gcbCallCount = 0;
+    bool result = searchCache.get(
+        {},
+        query,
+        [&](const Values&) {
+            gcbCallCount++;
+            return false;
+        },
+        [&](bool success, const std::vector<std::shared_ptr<dht::Node>>&) {
+            doneCalled = true;
+            doneSuccess = success;
+        });
+
+    CPPUNIT_ASSERT(result);
+    CPPUNIT_ASSERT_EQUAL(1, gcbCallCount);
+    CPPUNIT_ASSERT(doneCalled);
+    CPPUNIT_ASSERT_MESSAGE("Stopping early is reported as success, like the network path", doneSuccess);
+}
+
+void
+OpCacheTester::testTimestampOrdering()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+
+    auto v1 = makeValue(400, 0, "T");
+    auto t1 = std::chrono::system_clock::now();
+    auto t2 = t1 + std::chrono::seconds(10);
+
+    cache.onValue({v1}, false, t2);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    cache.onValue({v1}, true, t1);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Stale expiration should be ignored", (size_t) 1, cache.size());
+
+    cache.onValue({v1}, true, t2);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+OpCacheTester::testSourceDropKeepsValue()
+{
+    // Two per-node ValueCaches feed one Search-level OpCache. When one node
+    // leaves the search (its ValueCache is destroyed, expiring everything it
+    // held) the value must stay visible as long as the other node holds it.
+    std::vector<Event> events;
+    dht::SearchCache searchCache;
+    auto query = std::make_shared<dht::Query>();
+    dht::ValueCallback toOpCache;
+    dht::SyncCallback toOpCacheSync;
+
+    searchCache.listen(recordEvents(events),
+                       query,
+                       {},
+                       [&](const std::shared_ptr<dht::Query>&, dht::ValueCallback vcb, dht::SyncCallback scb) -> size_t {
+                           toOpCache = std::move(vcb);
+                           toOpCacheSync = std::move(scb);
+                           return 1;
+                       });
+
+    dht::TypeStore types;
+    auto now = dht::clock::now();
+    auto v = makeValue(42, 0, "hello");
+
+    auto makeNodeCache = [&] {
+        return std::make_unique<dht::ValueCache>([&](const Values& vals, bool expired) { toOpCache(vals, expired); },
+                                                 [&](dht::ListenSyncStatus s) { toOpCacheSync(s); });
+    };
+    auto nodeA = makeNodeCache();
+    auto nodeB = makeNodeCache();
+    nodeA->onValues({v}, {}, {}, types, now);
+    nodeA->onSynced(true);
+    nodeB->onValues({v}, {}, {}, types, now);
+    nodeB->onSynced(true);
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, searchCache.get(dht::Value::Filter {}).size());
+
+    nodeA.reset();
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("No expire while another node holds the value", 0u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, searchCache.get(dht::Value::Filter {}).size());
+
+    nodeB->onValues({v}, {}, {}, types, now + std::chrono::seconds(1));
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Refresh from the remaining node is not re-delivered", 1u, count(events, false));
+
+    nodeB.reset();
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Last holder gone: expire once", 1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, searchCache.get(dht::Value::Filter {}).size());
+}
+
+void
+OpCacheTester::testHighChurnRefCountConsistency()
+{
+    dht::OpValueCache cache([](const Values&, bool) { return true; });
+    auto v = makeValue(800, 0, "CHURN");
+
+    for (int i = 0; i < 50; i++) {
+        cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+        cache.onValuesAdded({v});
+        cache.onValuesExpired({v});
+        cache.onNodeChanged(dht::ListenSyncStatus::REMOVED);
+    }
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+
+    // Overlapping churn: add A, add B, expire A, expire B
+    for (int i = 0; i < 20; i++) {
+        cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+        cache.onValuesAdded({v});
+        cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+        cache.onValuesAdded({v});
+
+        cache.onValuesExpired({v});
+        cache.onNodeChanged(dht::ListenSyncStatus::REMOVED);
+        CPPUNIT_ASSERT_EQUAL_MESSAGE("Value kept while one source remains", (size_t) 1, cache.size());
+
+        cache.onValuesExpired({v});
+        cache.onNodeChanged(dht::ListenSyncStatus::REMOVED);
+        CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    }
+
+    // Extra expirations from sources that never added must be no-ops
+    cache.onValuesExpired({v});
+    cache.onValuesExpired({v});
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+OpCacheTester::testValueUpdateSingleSourcePhantom()
+{
+    // Editing a value (same id, higher seq) is delivered as a single add of
+    // the new version, without a phantom expiration of the old one.
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(700, 1, "OLD");
+    auto v2 = makeValue(700, 2, "NEW");
+
+    cache.onValuesAdded({v1});
+    events.clear();
+    cache.onValuesAdded({v2});
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, events.size());
+    CPPUNIT_ASSERT(!events[0].expired);
+    CPPUNIT_ASSERT_EQUAL(v2->id, events[0].id);
+    CPPUNIT_ASSERT_EQUAL(v2->seq, cache.get(700)->seq);
+}
+
+void
+OpCacheTester::testExpireAfterUpdate()
+{
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events));
+
+    auto v1 = makeValue(1200, 1, "V1");
+    auto v2 = makeValue(1200, 5, "V2");
+
+    cache.onValuesAdded({v1});
+    cache.onValuesAdded({v2});
+    CPPUNIT_ASSERT_EQUAL(v2->seq, cache.get(1200)->seq);
+
+    cache.onValuesExpired({v2});
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Expired value should be the updated one", v2->seq, events.back().seq);
+
+    // Once gone, an older version is accepted again
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL(v1->seq, cache.get(1200)->seq);
+}
+
+void
+OpCacheTester::testSingleSourceMode()
+{
+    // Single-source caches (proxy client) get the same values re-delivered
+    // on refresh: duplicates must not be counted, one expire removes.
+    std::vector<Event> events;
+    dht::OpValueCache cache(recordEvents(events), false);
+    auto v = makeValue(1100, 0, "DUP");
+
+    for (int i = 0; i < 10; i++)
+        cache.onValuesAdded({v});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    cache.onValuesExpired({v});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+
+    cache.onValuesExpired({v});
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+}
+
+void
+OpCacheTester::testSingleSourceModeSearchCache()
+{
+    std::vector<Event> events;
+    dht::SearchCache searchCache(false);
+    auto query = std::make_shared<dht::Query>();
+    dht::ValueCallback vcbStored;
+
+    searchCache.listen(recordEvents(events),
+                       query,
+                       {},
+                       [&](const std::shared_ptr<dht::Query>&, dht::ValueCallback vcb, dht::SyncCallback) -> size_t {
+                           vcbStored = std::move(vcb);
+                           return 1;
+                       });
+
+    auto v = makeValue(1300, 0, "S");
+    vcbStored({v}, false);
+    vcbStored({v}, false); // re-delivery after listen restart
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, false));
+
+    vcbStored({v}, true);
+    CPPUNIT_ASSERT_EQUAL(1u, count(events, true));
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, searchCache.get(dht::Value::Filter {}).size());
+}
+
+} // namespace test
@@ -0,0 +1,63 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+#pragma once
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class OpCacheTester : public CppUnit::TestFixture
+{
+    CPPUNIT_TEST_SUITE(OpCacheTester);
+    CPPUNIT_TEST(testBasicAddExpire);
+    CPPUNIT_TEST(testMultipleSources);
+    CPPUNIT_TEST(testEditResetsRefCount);
+    CPPUNIT_TEST(testEditChainMultiSource);
+    CPPUNIT_TEST(testStaleAddIgnored);
+    CPPUNIT_TEST(testStaleExpireIgnored);
+    CPPUNIT_TEST(testCallbacks);
+    CPPUNIT_TEST(testFilters);
+    CPPUNIT_TEST(testSyncStatus);
+    CPPUNIT_TEST(testGetWhileSynced);
+    CPPUNIT_TEST(testGetWhileNotSynced);
+    CPPUNIT_TEST(testGetEmptySynced);
+    CPPUNIT_TEST(testGetWithoutCallback);
+    CPPUNIT_TEST(testGetCallbackReturnsFalse);
+    CPPUNIT_TEST(testTimestampOrdering);
+    CPPUNIT_TEST(testSourceDropKeepsValue);
+    CPPUNIT_TEST(testHighChurnRefCountConsistency);
+    CPPUNIT_TEST(testValueUpdateSingleSourcePhantom);
+    CPPUNIT_TEST(testExpireAfterUpdate);
+    CPPUNIT_TEST(testSingleSourceMode);
+    CPPUNIT_TEST(testSingleSourceModeSearchCache);
+    CPPUNIT_TEST_SUITE_END();
+
+public:
+    void setUp();
+    void tearDown();
+
+    void testBasicAddExpire();
+    void testMultipleSources();
+    void testEditResetsRefCount();
+    void testEditChainMultiSource();
+    void testStaleAddIgnored();
+    void testStaleExpireIgnored();
+    void testCallbacks();
+    void testFilters();
+    void testSyncStatus();
+    void testGetWhileSynced();
+    void testGetWhileNotSynced();
+    void testGetEmptySynced();
+    void testGetWithoutCallback();
+    void testGetCallbackReturnsFalse();
+    void testTimestampOrdering();
+    void testSourceDropKeepsValue();
+    void testHighChurnRefCountConsistency();
+    void testValueUpdateSingleSourcePhantom();
+    void testExpireAfterUpdate();
+    void testSingleSourceMode();
+    void testSingleSourceModeSearchCache();
+};
+
+} // namespace test
@@ -0,0 +1,560 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+
+#include "test_valuecache.h"
+#include "../src/value_cache.h"
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(ValueCacheTester);
+
+void
+ValueCacheTester::setUp()
+{}
+
+void
+ValueCacheTester::tearDown()
+{}
+
+void
+ValueCacheTester::testUpdate()
+{
+    int addCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (!expired)
+            addCount += vals.size();
+    });
+
+    dht::TypeStore types;
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->seq = 1;
+    v1->data = {'A'};
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 1;
+    v2->seq = 2;
+    v2->data = {'B'};
+
+    auto now = std::chrono::steady_clock::now();
+
+    // Add v1
+    cache.onValues({v1}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+
+    // Update v1 -> v2
+    cache.onValues({v2}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Callback should be called for update", 2, addCount);
+}
+
+void
+ValueCacheTester::testUpdateExpiration()
+{
+    int addCount = 0;
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+        else
+            addCount += vals.size();
+    });
+
+    dht::TypeStore types;
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->seq = 1;
+    v1->data = {'A'};
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 1;
+    v2->seq = 2;
+    v2->data = {'B'};
+
+    auto now = std::chrono::steady_clock::now();
+
+    // Add v1
+    cache.onValues({v1}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+    CPPUNIT_ASSERT_EQUAL(0, expireCount);
+
+    // Update v1 -> v2: only the new value is emitted as an add,
+    // no expire callback for the old value.
+    cache.onValues({v2}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(2, addCount);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("No expire callback on edit", 0, expireCount);
+}
+
+void
+ValueCacheTester::testExpiration()
+{
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 10;
+    v1->data = {'A'};
+
+    // Default expiration is 10 minutes
+    auto now = std::chrono::steady_clock::now();
+    cache.onValues({v1}, {}, {}, types, now);
+
+    // Advance time past expiration
+    auto future = now + std::chrono::minutes(11);
+
+    // expireValues returns next expiration time
+    cache.expireValues(future);
+
+    CPPUNIT_ASSERT_EQUAL(1, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testRefresh()
+{
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 20;
+    v1->data = {'B'};
+
+    auto now = std::chrono::steady_clock::now();
+    cache.onValues({v1}, {}, {}, types, now);
+
+    // Refresh at +5 mins
+    auto later = now + std::chrono::minutes(5);
+    cache.onValues({}, {v1->id}, {}, types, later);
+
+    // Should not expire at +11 mins (original expiration was +10, so +10 from start)
+    // But refresh should extend it to +10 from 'later' = +15 from start.
+
+    auto checkTime = now + std::chrono::minutes(11);
+    cache.expireValues(checkTime);
+
+    CPPUNIT_ASSERT_EQUAL(0, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // Should expire at +16 mins
+    auto future = now + std::chrono::minutes(16);
+    cache.expireValues(future);
+    CPPUNIT_ASSERT_EQUAL(1, expireCount);
+}
+
+void
+ValueCacheTester::testExplicitExpiration()
+{
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 30;
+    v1->data = {'C'};
+
+    auto now = std::chrono::steady_clock::now();
+    cache.onValues({v1}, {}, {}, types, now);
+
+    cache.onValues({}, {}, {v1->id}, types, now);
+
+    CPPUNIT_ASSERT_EQUAL(1, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testClear()
+{
+    dht::ValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool) {});
+    dht::TypeStore types;
+    auto now = std::chrono::steady_clock::now();
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 2;
+
+    cache.onValues({v1, v2}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL((size_t) 2, cache.size());
+
+    auto cbs = cache.clear();
+    // Execute callbacks
+    for (auto& cb : cbs)
+        cb();
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testSyncStatus()
+{
+    std::vector<dht::ListenSyncStatus> statuses;
+    {
+        dht::ValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool) {},
+                              [&](dht::ListenSyncStatus status) { statuses.push_back(status); });
+
+        CPPUNIT_ASSERT_EQUAL((size_t) 1, statuses.size());
+        CPPUNIT_ASSERT(statuses.back() == dht::ListenSyncStatus::ADDED);
+
+        cache.onSynced(true);
+        CPPUNIT_ASSERT_EQUAL((size_t) 2, statuses.size());
+        CPPUNIT_ASSERT(statuses.back() == dht::ListenSyncStatus::SYNCED);
+
+        // Calling onSynced(true) again should not trigger the callback
+        cache.onSynced(true);
+        CPPUNIT_ASSERT_EQUAL((size_t) 2, statuses.size());
+
+        cache.onSynced(false);
+        CPPUNIT_ASSERT_EQUAL((size_t) 3, statuses.size());
+        CPPUNIT_ASSERT(statuses.back() == dht::ListenSyncStatus::UNSYNCED);
+
+        // Calling onSynced(false) again should not trigger the callback
+        cache.onSynced(false);
+        CPPUNIT_ASSERT_EQUAL((size_t) 3, statuses.size());
+
+        // Set to synced before destruction to test UNSYNCED on destruction
+        cache.onSynced(true);
+        CPPUNIT_ASSERT_EQUAL((size_t) 4, statuses.size());
+        CPPUNIT_ASSERT(statuses.back() == dht::ListenSyncStatus::SYNCED);
+    }
+
+    // Destructor should have been called
+    CPPUNIT_ASSERT_EQUAL((size_t) 6, statuses.size());
+    CPPUNIT_ASSERT(statuses[4] == dht::ListenSyncStatus::UNSYNCED);
+    CPPUNIT_ASSERT(statuses[5] == dht::ListenSyncStatus::REMOVED);
+}
+
+void
+ValueCacheTester::testMaxValues()
+{
+    dht::ValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool) {});
+    dht::TypeStore types;
+    auto now = std::chrono::steady_clock::now();
+
+    // MAX_VALUES is 4096
+    for (int i = 0; i < 4100; ++i) {
+        auto v = std::make_shared<dht::Value>();
+        v->id = (dht::Value::Id) i + 1;
+        cache.onValues({v}, {}, {}, types, now + std::chrono::milliseconds(i));
+    }
+
+    // Should be capped at 4096
+    CPPUNIT_ASSERT_EQUAL((size_t) 4096, cache.size());
+}
+
+void
+ValueCacheTester::testUpdateTypeExpiration()
+{
+    // Test that when a value is updated with a different type, the expiration
+    // uses the new type's expiration period, not the old one.
+    int addCount = 0;
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+        else
+            addCount += vals.size();
+    });
+
+    // Register a custom type with a short expiration (2 minutes)
+    dht::TypeStore types;
+    dht::ValueType shortType(42, "short", std::chrono::minutes(2));
+    types.registerType(shortType);
+
+    // Register a custom type with a long expiration (30 minutes)
+    dht::ValueType longType(99, "long", std::chrono::minutes(30));
+    types.registerType(longType);
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->type = 42; // short type: 2 min expiration
+    v1->data = {'A'};
+
+    auto now = std::chrono::steady_clock::now();
+
+    // Add v1 with short type
+    cache.onValues({v1}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+
+    // Update with new type (long expiration)
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 1;
+    v2->seq = 1;
+    v2->type = 99; // long type: 30 min expiration
+    v2->data = {'B'};
+
+    cache.onValues({v2}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(2, addCount);
+
+    // At +3 minutes: should NOT be expired (long type has 30 min expiration)
+    // Before the fix, the old type (2 min) would be used, causing expiration here
+    expireCount = 0;
+    auto at3min = now + std::chrono::minutes(3);
+    cache.expireValues(at3min);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Value updated to long-expiry type should NOT expire at 3 min", 0, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // At +31 minutes: should be expired
+    auto at31min = now + std::chrono::minutes(31);
+    cache.expireValues(at31min);
+    CPPUNIT_ASSERT_EQUAL(1, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testSameSeqEditIgnored()
+{
+    // Like OpValueCache, an edit is only recognized by an increasing seq:
+    // a different payload with the same seq is a refresh, not an edit.
+    int addCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (!expired)
+            addCount += vals.size();
+    });
+    dht::TypeStore types;
+    auto now = std::chrono::steady_clock::now();
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->seq = 3;
+    v1->data = {'A'};
+    cache.onValues({v1}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+
+    auto sameSeq = std::make_shared<dht::Value>();
+    sameSeq->id = 1;
+    sameSeq->seq = 3;
+    sameSeq->data = {'B'};
+    cache.onValues({sameSeq}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+
+    auto older = std::make_shared<dht::Value>();
+    older->id = 1;
+    older->seq = 2;
+    older->data = {'C'};
+    cache.onValues({older}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+
+    auto newer = std::make_shared<dht::Value>();
+    newer->id = 1;
+    newer->seq = 4;
+    newer->data = {'D'};
+    cache.onValues({newer}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(2, addCount);
+}
+
+void
+ValueCacheTester::testShortExpirationRefreshLost()
+{
+    // Demonstrates that a value with a short expiration type survives
+    // past its nominal expiration when the listen is synced, thanks to
+    // a proportional grace period (= typeExpiration).
+    //
+    // For a 5s type:  effective expiration when synced = expiration + 5s = 10s
+    //
+    // Scenario:
+    //   t=0 : Value V added → nominal expiration t+5s, effective t+10s
+    //   t=6 : Past nominal but within grace → alive
+    //   t=8 : Refresh arrives → V refreshed → expiration reset to t8+5s, effective t8+10s=t18s
+    //   t=14: Still alive (within refreshed effective range)
+    //   t=20: No more refreshes → past t8+10s=t18s → expired
+
+    int addCount = 0;
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+        else
+            addCount += vals.size();
+    });
+
+    // Register a type with a very short expiration (5 seconds)
+    dht::TypeStore types;
+    dht::ValueType shortType(100, "short_lived", std::chrono::seconds(5));
+    types.registerType(shortType);
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->type = 100;
+    v1->data = {'V'};
+
+    auto t0 = std::chrono::steady_clock::now();
+
+    // t=0: Listen response arrives with V
+    cache.onValues({v1}, {}, {}, types, t0);
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+    CPPUNIT_ASSERT_EQUAL(0, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // Mark cache as synced (listen is active)
+    cache.onSynced(true);
+
+    // t=6: Past nominal 5s expiration, but grace (5s) gives effective 10s
+    auto t6 = t0 + std::chrono::seconds(6);
+    cache.expireValues(t6);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Value should survive past nominal expiration thanks to proportional grace",
+                                 0,
+                                 expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // t=8: Value refresh arrives → expiration reset to t8+5s=t13s, effective t8+10s=t18s
+    auto t8 = t0 + std::chrono::seconds(8);
+    cache.onValues({}, {v1->id}, {}, types, t8);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // t=14: Within refreshed effective range (t18s) → alive
+    auto t14 = t0 + std::chrono::seconds(14);
+    cache.expireValues(t14);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Value should survive past short expiration after refresh", 0, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // t=20: No more refreshes → past effective t18s → must expire
+    auto t20 = t0 + std::chrono::seconds(20);
+    cache.expireValues(t20);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Value must eventually expire even while synced if server stops refreshing",
+                                 1,
+                                 expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testShortExpirationHighChurn()
+{
+    // Simulates high churn: multiple rapid add/expire/refresh cycles with
+    // short-lived values, verifying the cache stays consistent.
+
+    int addCount = 0;
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+        else
+            addCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    dht::ValueType shortType(101, "fast_churn", std::chrono::seconds(3));
+    types.registerType(shortType);
+
+    auto now = std::chrono::steady_clock::now();
+
+    // Simulate 10 churn cycles: add value, advance time, refresh, advance, expire
+    for (int cycle = 0; cycle < 10; cycle++) {
+        auto v = std::make_shared<dht::Value>();
+        v->id = static_cast<dht::Value::Id>(500 + cycle);
+        v->type = 101;
+        v->data = {static_cast<uint8_t>('A' + cycle)};
+
+        auto t = now + std::chrono::seconds(cycle * 10);
+
+        // Add the value
+        cache.onValues({v}, {}, {}, types, t);
+
+        // Refresh halfway through
+        auto tRefresh = t + std::chrono::seconds(2);
+        cache.onValues({}, {v->id}, {}, types, tRefresh);
+
+        // Explicit expire from server
+        auto tExpire = t + std::chrono::seconds(8);
+        cache.onValues({}, {}, {v->id}, types, tExpire);
+    }
+
+    // All 10 values were added and explicitly expired
+    CPPUNIT_ASSERT_EQUAL(10, addCount);
+    CPPUNIT_ASSERT_EQUAL(10, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testGracePeriodWhileSynced()
+{
+    // Verify that when the listen is synced, the proportional grace period
+    // (= typeExpiration) delays time-based expiration but doesn't prevent
+    // it entirely.  For a 2s type: effective expiration = 2s + 2s = 4s.
+
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    dht::ValueType shortType(110, "short", std::chrono::seconds(2));
+    types.registerType(shortType);
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->type = 110;
+    v1->data = {'X'};
+
+    auto t0 = std::chrono::steady_clock::now();
+    cache.onValues({v1}, {}, {}, types, t0);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // Mark as synced
+    cache.onSynced(true);
+
+    // t=3: Past nominal 2s, within effective 4s → grace protects
+    auto t3 = t0 + std::chrono::seconds(3);
+    cache.expireValues(t3);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Synced cache: grace period should prevent early expiry", 0, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 1, cache.size());
+
+    // t=5: Past effective 4s (2s + 2s grace) → must expire
+    auto t5 = t0 + std::chrono::seconds(5);
+    cache.expireValues(t5);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Synced cache: value must eventually expire after grace period", 1, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+void
+ValueCacheTester::testUnsyncExpiresImmediately()
+{
+    // When the cache goes from synced to unsynced, the grace period is
+    // removed and already-past-due values expire on the next check.
+    // Type expiration = 5s. Grace when synced = 5s → effective = 10s.
+
+    int expireCount = 0;
+    dht::ValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired) {
+        if (expired)
+            expireCount += vals.size();
+    });
+
+    dht::TypeStore types;
+    dht::ValueType shortType(120, "short", std::chrono::seconds(5));
+    types.registerType(shortType);
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->type = 120;
+    v1->data = {'U'};
+
+    auto t0 = std::chrono::steady_clock::now();
+    cache.onValues({v1}, {}, {}, types, t0);
+    cache.onSynced(true);
+
+    // t=7: past nominal 5s, within effective 10s while synced → alive
+    auto t7 = t0 + std::chrono::seconds(7);
+    cache.expireValues(t7);
+    CPPUNIT_ASSERT_EQUAL(0, expireCount);
+
+    // Now unsync (node went away / connectivity loss)
+    cache.onSynced(false);
+
+    // Same time point t=7: past nominal 5s, no grace → expires
+    cache.expireValues(t7);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Unsynced cache should immediately expire past-due values", 1, expireCount);
+    CPPUNIT_ASSERT_EQUAL((size_t) 0, cache.size());
+}
+
+} // namespace test
@@ -0,0 +1,49 @@
+// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
+// SPDX-License-Identifier: MIT
+#pragma once
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class ValueCacheTester : public CppUnit::TestFixture
+{
+    CPPUNIT_TEST_SUITE(ValueCacheTester);
+    CPPUNIT_TEST(testUpdate);
+    CPPUNIT_TEST(testUpdateExpiration);
+    CPPUNIT_TEST(testExpiration);
+    CPPUNIT_TEST(testRefresh);
+    CPPUNIT_TEST(testExplicitExpiration);
+    CPPUNIT_TEST(testClear);
+    CPPUNIT_TEST(testSyncStatus);
+    CPPUNIT_TEST(testMaxValues);
+    CPPUNIT_TEST(testUpdateTypeExpiration);
+    CPPUNIT_TEST(testSameSeqEditIgnored);
+    CPPUNIT_TEST(testShortExpirationRefreshLost);
+    CPPUNIT_TEST(testShortExpirationHighChurn);
+    CPPUNIT_TEST(testGracePeriodWhileSynced);
+    CPPUNIT_TEST(testUnsyncExpiresImmediately);
+    CPPUNIT_TEST_SUITE_END();
+
+public:
+    void setUp();
+    void tearDown();
+
+    void testUpdate();
+    void testUpdateExpiration();
+    void testExpiration();
+    void testRefresh();
+    void testExplicitExpiration();
+    void testClear();
+    void testSyncStatus();
+    void testMaxValues();
+    void testUpdateTypeExpiration();
+    void testSameSeqEditIgnored();
+    void testShortExpirationRefreshLost();
+    void testShortExpirationHighChurn();
+    void testGracePeriodWhileSynced();
+    void testUnsyncExpiresImmediately();
+};
+
+} // namespace test
Author	SHA1	Message	Date
Adrien Beraud	0cddee9fa3	proxy_server: detach pending push requests before destruction ~DhtProxyServer destroyed requests_ whose ~http::Request notifies DONE; the handler registered by sendPushNotification then erased from the map being destroyed, corrupting the heap (double free under ASan) whenever a push request was still pending, e.g. an unreachable push server. Detach the state-change callbacks and release the requests before the members are destroyed.	2026-09-04 14:43:53 -04:00
Adrien Beraud	3aec1a39d8	storage: re-storing identical content is a refresh, not an edition Storage::store reported edited_values=1 whenever the stored Sp<Value> differed from the incoming one, even with equal content. A local put() of an already-stored value was therefore announced to local listeners as a new value a second time, unbalancing listen caches that count how many sources hold a value. Only report an edition when the content actually changed; the expiration is still refreshed.	2026-09-04 14:43:53 -04:00
Adrien Béraud	8a8b4d41ca	op_cache, proxy_client: honor GetCallback return and guard push refresh expiry - SearchCache::get: if gcb returns false, call dcb(false, {}) instead of silently proceeding with dcb(true, {}). - DhtProxyClient push refresh: skip expiration of old values when the get() completion reports failure (ok == false), preventing spurious expires on transient errors. In pushNotificationReceived, track the set of value IDs received in the refresh GET and only expire values whose IDs are not in the new set. Previously all old values were expired on refresh, incorrectly expiring edited values.	2026-09-04 14:43:53 -04:00
Adrien Beraud	3627090257	value_cache: detect editions by increasing seq, like OpValueCache ValueCache emitted any content change of a known value as an edition while OpValueCache only accepts a higher seq. A node reporting an older version made ValueCache replace its copy and re-emit versions that were then dropped or, with source counting, double counted downstream. Use the same rule at both levels.	2026-09-04 14:43:53 -04:00
Adrien Béraud	88af795eaa	value_cache, op_cache: fix value expiration after edition ValueCache::addValues no longer emits an expire callback for the old version when a value is edited; only the add callback for the new version is emitted. Add a grace period in expireValues when the cache is synced, using the value type's expiration as an extra margin before emitting expirations. OpValueCache::onValuesAdded: on edit, expire the old value and add the new one without incrementing refCount. SearchCache::get: only serve from cache when synced.	2026-09-04 14:43:53 -04:00
Adrien Béraud	2e509193d4	tests: add listen value edit tests for DhtRunner and proxy Add testListenValueEdit to DhtRunnerTester: verifies that editing a value triggers an add callback (not an expire+add), and that the edited value eventually expires. Add testListenValueEdit and testPushListenValueEdit to DhtProxyTester: same semantics tested over the long-poll and push notification proxy paths. Fix proxy tearDown order to destroy DhtProxyServer before shutting down the proxy DhtRunner, preventing use-after-free on cancel.	2026-09-04 14:43:53 -04:00
Adrien Béraud	527ecd9d19	tests: add OpCacheTester and ValueCacheTester Add unit tests for OpValueCache and ValueCache in isolation. OpCacheTester covers add/edit/expire flows, refcount management, and value synchronization lifecycle. ValueCacheTester covers edition handling and value expiration.	2026-09-04 14:41:11 -04:00
Adrien BeraudandCopilot	3a383ac873	http: close connections on the io_context thread asio sockets are not thread-safe. http::Connection::close() used to close the socket synchronously on the calling thread. Request::cancel() and Request::~Request() (via terminate()) are called from arbitrary threads (e.g. DhtProxyClient::stop(), cancelAllListeners(), cancelListen()), so the close raced with in-flight operations on the io_context thread. In particular, asio::async_connect's range_connect_op closes and re-opens the socket when trying the next endpoint (common on mobile where the IPv6 endpoint fails first). A concurrent close() from another thread corrupts the epoll reactor's per-descriptor state, leading to crashes in epoll_reactor::deregister_descriptor / register_descriptor called from range_connect_op::process. Dispatch the close to the io_context thread when not already on it. Also keep the connection alive during the TLS handshake, like every other asynchronous operation, so that the destructor can never run while an operation is pending. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-09-04 14:13:20 -04:00
Adrien Beraud	3149fe60c4	proxy: increase OP_MARGIN to 8 hours	2026-09-04 11:04:42 -04:00
Adrien Beraud	d5956e0778	proxy: use system_clock for subscriber expiration timeouts Track subscribe/permanent-put expiration with std::chrono::system_clock (wall clock) instead of steady_clock in both the proxy client and server, so persisted expirations survive restarts and map directly to epoch time. - DhtProxyServer::clock is now system_clock; expireTimer/expireNotifyTimer use asio::system_timer. - DhtProxyClient::refreshSubscriberTimer uses asio::system_timer and system_clock::now() for subscriber refresh.	2026-09-04 11:04:42 -04:00