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.

CMakeLists.txt

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

include/opendht/dht_proxy_server.h

@@ -76,8 +76,14 @@ public:
 
     asio::io_context& io_context() const;
 
-    using clock = std::chrono::system_clock;
+    // 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,7 +407,7 @@ 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;
@@ -419,7 +425,7 @@ private:
             p.pack("value");
             p.pack(value);
             p.pack("exp");
-            p.pack(std::chrono::system_clock::to_time_t(expiration));
+            p.pack(sys_clock::to_time_t(expiration));
             if (not clientId.empty()) {
                 p.pack("cid");
                 p.pack(clientId);
@@ -460,7 +466,7 @@ 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;
@@ -476,7 +482,7 @@ private:
             p.pack("cid");
             p.pack(clientId);
             p.pack("exp");
-            p.pack(std::chrono::system_clock::to_time_t(expiration));
+            p.pack(sys_clock::to_time_t(expiration));
             if (sessionCtx) {
                 std::lock_guard l(sessionCtx->lock);
                 p.pack("sid");

include/opendht/http.h

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

src/dht_proxy_client.cpp

@@ -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;
@@ -53,7 +55,7 @@ struct PermanentPut
 
 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 {};
@@ -1210,7 +1212,8 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
         // Connectivity changed, refresh all subscribe
         for (auto& search : searches_)
             for (auto& listener : search.second.listeners)
-                resubscribe(search.first, listener.first, listener.second);
+                if (!listener.second.opstate->ok)
+                    resubscribe(search.first, listener.first, listener.second);
         return;
     }
     if (logger_)
@@ -1307,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 {

src/dht_proxy_server.cpp

@@ -126,8 +126,8 @@ DhtProxyServer::ServerStats::toJson() const
     result["putCount"] = static_cast<Json::UInt64>(putCount);
     result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
     result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
-    result["serverStartTime"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(serverStartTime));
-    result["lastUpdated"] = static_cast<Json::LargestInt>(std::chrono::system_clock::to_time_t(lastUpdated));
+    result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
+    result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
     result["androidPush"] = androidPush.toJson();
     result["iosPush"] = iosPush.toJson();
     result["unifiedPush"] = unifiedPush.toJson();
@@ -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 = std::chrono::system_clock::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 = std::chrono::system_clock::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)
@@ -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");
 }
@@ -951,8 +963,8 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
         }
         auto& listener = *listIt;
 
-        // Expiration
-        auto timeout = 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;
@@ -1406,7 +1418,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
                     topic = pVal["topic"].asString();
                 }
                 std::lock_guard lock(lockSearchPuts_);
-                auto timeout = 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) {

src/http.cpp

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

src/op_cache.cpp

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

src/op_cache.h

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

src/storage.h

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

src/value_cache.h

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

tests/test_dhtproxy.h

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

tests/test_dhtproxy_client.cpp

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

tests/test_dhtrunner.cpp

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

tests/test_dhtrunner.h

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

tests/test_opcache.cpp

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

tests/test_opcache.h

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

tests/test_valuecache.cpp

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

tests/test_valuecache.h

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