tests/dhtrunner: test empty value

Make Hash default constructor, hex constructor, data(), const
operator[] and fromString constexpr. Change fromString to be
non-throwing: it now uses a validity flag and fills the hash with zeros
on parse error instead of throwing. Add a constexpr toHexArray helper
that returns a std::array<char, N*2> for compile-time hex conversion.
Update tests to include <string_view>, use std::string_view, add
constexpr InfoHash usage and a static_assert for the hex array.

CMakeLists.txt

@@ -865,6 +865,10 @@ if (BUILD_TESTING)
         tests/storagetester.cpp
         tests/threadpooltester.h
         tests/threadpooltester.cpp
+        tests/opcachetester.h
+        tests/opcachetester.cpp
+        tests/valuecachetester.h
+        tests/valuecachetester.cpp
     )
     if (OPENDHT_TESTS_NETWORK)
         if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)

include/opendht/infohash.h

@@ -54,8 +54,8 @@ public:
     typedef typename T::iterator iterator;
     typedef typename T::const_iterator const_iterator;
 
-    Hash() noexcept { data_.fill(0); }
-    Hash(const uint8_t* h, size_t data_len)
+    constexpr Hash() noexcept { data_.fill(0); }
+    constexpr Hash(const uint8_t* h, size_t data_len)
     {
         if (data_len < N)
             data_.fill(0);
@@ -67,7 +67,8 @@ public:
      * hex must be at least 2.HASH_LEN characters long.
      * If too long, only the first 2.HASH_LEN characters are read.
      */
-    explicit Hash(std::string_view hex)
+    explicit constexpr Hash(std::string_view hex)
+        : data_()
     {
         if (hex.size() < 2 * N)
             data_.fill(0);
@@ -78,7 +79,7 @@ public:
     Hash(const msgpack::object& o) { msgpack_unpack(o); }
 
     static constexpr size_t size() noexcept { return N; }
-    const uint8_t* data() const { return data_.data(); }
+    constexpr const uint8_t* data() const { return data_.data(); }
     uint8_t* data() { return data_.data(); }
     iterator begin() { return data_.begin(); }
     const_iterator cbegin() const { return data_.cbegin(); }
@@ -120,7 +121,7 @@ public:
     }
 
     uint8_t& operator[](size_t index) { return data_[index]; }
-    const uint8_t& operator[](size_t index) const { return data_[index]; }
+    constexpr const uint8_t& operator[](size_t index) const { return data_[index]; }
 
     /**
      * Find the lowest 1 bit in an id.
@@ -258,7 +259,7 @@ public:
 
 private:
     T data_;
-    void fromString(const char*);
+    constexpr void fromString(const char*);
 };
 
 #define HASH_LEN 20u
@@ -286,25 +287,29 @@ operator>>(std::istream& s, Hash<N>& h)
 }
 
 template<size_t N>
-void
+constexpr void
 Hash<N>::fromString(const char* in)
 {
-    auto hex2bin = [](char c) -> uint8_t {
+    constexpr auto hex2bin = [](char c, bool& valid) -> uint8_t {
         if (c >= 'a' and c <= 'f')
             return 10 + c - 'a';
         else if (c >= 'A' and c <= 'F')
             return 10 + c - 'A';
         else if (c >= '0' and c <= '9')
             return c - '0';
-        else
-            throw std::domain_error("not an hex character");
+        else {
+            valid = false;
+            return 0;
+        }
     };
-    try {
-        for (size_t i = 0; i < N; i++)
-            data_[i] = (hex2bin(in[2 * i]) << 4) | hex2bin(in[2 * i + 1]);
-    } catch (const std::domain_error&) {
-        data_.fill(0);
+    bool valid = true;
+    for (size_t i = 0; i < N && valid; i++) {
+        uint8_t high = hex2bin(in[2 * i], valid);
+        uint8_t low = hex2bin(in[2 * i + 1], valid);
+        data_[i] = (high << 4) | low;
     }
+    if (!valid)
+        data_.fill(0);
 }
 
 template<size_t N>
@@ -369,6 +374,19 @@ toHex(const std::vector<uint8_t>& data)
     return toHex(data.data(), data.size());
 }
 
+template<size_t N>
+constexpr std::array<char, N * 2>
+toHexArray(const Hash<N>& h)
+{
+    std::array<char, N * 2> ret {};
+    for (size_t i = 0; i < N; i++) {
+        const auto& m = hex_map[h[i]];
+        ret[i * 2] = m[0];
+        ret[i * 2 + 1] = m[1];
+    }
+    return ret;
+}
+
 template<size_t N>
 const char*
 Hash<N>::to_c_str() const

src/dht.cpp

@@ -2651,10 +2651,9 @@ Dht::onAnnounce(Sp<Node> n,
                                          net::DhtProtocolException::PUT_WRONG_TOKEN};
     }
     {
-        // We store a value only if we think we're part of the
-        // SEARCH_NODES nodes around the target id.
+        // We store a value only if we think we're part of the SEARCH_NODES nodes around the target id.
         auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
-        if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
+        if (closest_nodes.size() > TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
             if (logger_)
                 logger_->warn("[node {}] Announce too far from the target. Dropping value.", node.toString());
             return {};
@@ -2680,8 +2679,7 @@ Dht::onAnnounce(Sp<Node> n,
                                    node.toString(),
                                    v->id);
             } else {
-                const auto& type = getType(lv->type);
-                if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
+                if (getType(lv->type).editPolicy(hash, lv, vc, node.id, node.getAddr())) {
                     if (logger_)
                         logger_->debug("[store {}] Editing {}", hash.toString(), vc->toString());
                     storageStore(hash, vc, created, node.getAddr());
@@ -2694,11 +2692,9 @@ Dht::onAnnounce(Sp<Node> n,
             }
         } else {
             // Allow the value to be edited by the storage policy
-            const auto& type = getType(vc->type);
-            if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
+            if (getType(vc->type).storePolicy(hash, vc, node.id, node.getAddr())) {
                 // if (logger_)
-                //     logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(),
-                //     std::to_string(vc->id).c_str());
+                //     logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(), vc->id);
                 storageStore(hash, vc, created, node.getAddr());
             } else {
                 if (logger_)

src/op_cache.cpp

@@ -14,14 +14,15 @@ OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals, const system_clo
         auto viop = values.emplace(v->id, v);
         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++;
+            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);
+                }
+            }
         }
         viop.first->second.updated = t;
     }

src/value_cache.h

@@ -165,6 +165,7 @@ private:
     CallbackQueue addValues(const std::vector<Sp<Value>>& new_values, const TypeStore& types, const time_point& now)
     {
         std::vector<Sp<Value>> nvals;
+        std::vector<Sp<Value>> expired_vals;
         for (const auto& value : new_values) {
             auto v = values.find(value->id);
             if (v == values.end()) {
@@ -175,9 +176,18 @@ private:
                 // refreshed value
                 v->second.created = now;
                 v->second.expiration = now + types.getType(v->second.data->type).expiration;
+                if (*v->second.data != *value) {
+                    expired_vals.emplace_back(v->second.data);
+                    v->second.data = value;
+                    nvals.emplace_back(value);
+                }
             }
         }
         CallbackQueue ret;
+        if (callback and not expired_vals.empty())
+            ret.emplace_back([cb = callback, expired_vals = std::move(expired_vals)]{
+                cb(expired_vals, true);
+            });
         if (callback and not nvals.empty())
             ret.emplace_back([cb = callback, nvals = std::move(nvals)] { cb(nvals, false); });
         return ret;

tests/dhtproxytester.cpp

@@ -9,6 +9,9 @@
 
 #include <chrono>
 #include <condition_variable>
+#include <asio.hpp>
+#include <sys/socket.h>
+#include <sys/time.h>
 
 using namespace std::chrono_literals;
 
@@ -203,6 +206,378 @@ DhtProxyTester::testResubscribeGetValues()
     CPPUNIT_ASSERT(firstVal_data == values.front()->data);
 }
 
+void
+DhtProxyTester::testPushNotification()
+{
+#ifndef OPENDHT_PUSH_NOTIFICATIONS
+    fmt::print(stderr, "Push notifications are not supported in this build, skipping test\n");
+    return;
+#else
+    static constexpr dht::ValueType::Id PUSH_EDITABLE_TYPE_ID = 4242;
+    const dht::ValueType pushEditableType {PUSH_EDITABLE_TYPE_ID,
+                                           "push-editable",
+                                           std::chrono::seconds(1),
+                                           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; }};
+    nodePeer.registerType(pushEditableType);
+    nodeProxy->registerType(pushEditableType);
+
+    clientConfig.push_token = "atlas";
+    nodeClient.run(0, clientConfig);
+    nodeClient.registerType(pushEditableType);
+
+    bool gotValues = false;
+    bool gotExpiredValues = false;
+    std::condition_variable cv;
+    std::mutex cv_m;
+    std::unique_lock<std::mutex> lk(cv_m);
+    auto key = dht::InfoHash::get("GLaDOs");
+    auto valueAuthor = dht::crypto::generateIdentity("PushNotificationAuthor");
+
+    std::vector<std::shared_ptr<dht::Value>> values;
+    std::vector<std::shared_ptr<dht::Value>> expiredValues;
+    auto ftoken = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        if (expired) {
+            expiredValues.insert(expiredValues.end(), v.begin(), v.end());
+            gotExpiredValues = true;
+        } else {
+            values.insert(values.end(), v.begin(), v.end());
+            gotValues = true;
+        }
+        cv.notify_all();
+        return true;
+    });
+
+    // Wait for listen to be sent to proxy
+    cv.wait_for(lk, std::chrono::seconds(1));
+
+    // Peer puts a value
+    dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
+    firstVal.type = PUSH_EDITABLE_TYPE_ID;
+    firstVal.id = 1337;
+    firstVal.seq = 1;
+    firstVal.sign(*valueAuthor.first);
+    auto firstVal_data = firstVal.data;
+    nodePeer.put(key, std::move(firstVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        gotValues = true;
+        cv.notify_all();
+    });
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return gotValues; }));
+    gotValues = false;
+
+    // Simulate push notification
+    std::map<std::string, std::string> push_data;
+    push_data["key"] = key.toString();
+    push_data["t"] = std::to_string(
+        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch())
+            .count());
+    CPPUNIT_ASSERT(nodeClient.pushNotificationReceived(push_data).get() == dht::PushNotificationResult::Values);
+
+    // Wait for client to receive the value
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return gotValues; }));
+    gotValues = false;
+
+    CPPUNIT_ASSERT_EQUAL((size_t) 1u, values.size());
+    CPPUNIT_ASSERT(firstVal_data == values.front()->data);
+
+    // Peer edits the existing value (same id, higher seq)
+    dht::Value secondVal {"You're a monster"};
+    secondVal.type = PUSH_EDITABLE_TYPE_ID;
+    secondVal.id = 1337;
+    secondVal.seq = 2;
+    secondVal.sign(*valueAuthor.first);
+    auto secondVal_data = secondVal.data;
+    nodePeer.put(key, std::move(secondVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        gotValues = true;
+        cv.notify_all();
+    });
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return gotValues; }));
+    gotValues = false;
+
+    // Simulate push notification again
+    push_data["t"] = std::to_string(
+        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch())
+            .count());
+    CPPUNIT_ASSERT(nodeClient.pushNotificationReceived(push_data).get() == dht::PushNotificationResult::Values);
+
+    // Wait for client to receive the edited value
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return gotValues; }));
+    gotValues = false;
+
+    // OpValueCache should update the existing id with the new sequence/data
+    CPPUNIT_ASSERT_EQUAL((size_t) 2u, values.size());
+    CPPUNIT_ASSERT_EQUAL((dht::Value::Id) 1337, values.back()->id);
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 2, values.back()->seq);
+    CPPUNIT_ASSERT(secondVal_data == values.back()->data);
+
+    // Simulate expiration push notification for the edited value
+    push_data["exp"] = std::to_string((dht::Value::Id) 1337);
+    push_data["t"] = std::to_string(
+        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch())
+            .count());
+    CPPUNIT_ASSERT(nodeClient.pushNotificationReceived(push_data).get() == dht::PushNotificationResult::ValuesExpired);
+
+    cv.wait_for(lk, 500ms, [&] { return gotExpiredValues; });
+
+    auto token = ftoken.get();
+    CPPUNIT_ASSERT(token);
+    nodeClient.cancelListen(key, token);
+#endif
+}
+
+void
+DhtProxyTester::testProxyServerClientFullChain()
+{
+#ifndef OPENDHT_PUSH_NOTIFICATIONS
+    fmt::print(stderr, "Push notifications are not supported in this build, skipping test\n");
+    return;
+#else
+    static constexpr dht::ValueType::Id CHAIN_EDITABLE_TYPE_ID = 4343;
+    const dht::ValueType chainEditableType {CHAIN_EDITABLE_TYPE_ID,
+                                            "chain-editable",
+                                            std::chrono::seconds(30),
+                                            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; }};
+
+    uint16_t pushPort = 1024 + (std::rand() % (65535 - 1024));
+    uint16_t proxyPort = 1024 + (std::rand() % (65535 - 1024));
+
+    std::atomic_bool relayStop {false};
+    std::atomic_size_t relayedPushCount {0};
+    std::thread pushRelay([&] {
+        try {
+            asio::io_context io;
+            asio::ip::tcp::acceptor acceptor(io, asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), pushPort));
+            acceptor.non_blocking(true);
+
+            while (not relayStop.load()) {
+                asio::ip::tcp::socket socket(io);
+                std::error_code ec;
+                acceptor.accept(socket, ec);
+                if (ec == asio::error::would_block || ec == asio::error::try_again) {
+                    std::this_thread::sleep_for(5ms);
+                    continue;
+                }
+                if (ec)
+                    continue;
+
+                socket.non_blocking(true);
+
+                std::string raw;
+                raw.reserve(4096);
+                auto deadline = std::chrono::steady_clock::now() + 2s;
+                while (raw.find("\r\n\r\n") == std::string::npos && std::chrono::steady_clock::now() < deadline
+                       && not relayStop.load()) {
+                    std::array<char, 2048> chunk {};
+                    size_t n = socket.read_some(asio::buffer(chunk), ec);
+                    if (!ec) {
+                        raw.append(chunk.data(), n);
+                    } else if (ec == asio::error::would_block || ec == asio::error::try_again) {
+                        std::this_thread::sleep_for(2ms);
+                    } else {
+                        break;
+                    }
+                }
+                if (raw.find("\r\n\r\n") == std::string::npos)
+                    continue;
+
+                auto sep = raw.find("\r\n\r\n");
+                std::string head = raw.substr(0, sep + 4);
+                auto clPos = head.find("Content-Length:");
+                size_t contentLength = 0;
+                if (clPos != std::string::npos) {
+                    clPos += std::string("Content-Length:").size();
+                    while (clPos < head.size() && (head[clPos] == ' ' || head[clPos] == '\t'))
+                        ++clPos;
+                    size_t clEnd = head.find("\r\n", clPos);
+                    contentLength = std::stoul(head.substr(clPos, clEnd - clPos));
+                }
+
+                std::string body = raw.substr(sep + 4);
+                while (body.size() < contentLength && std::chrono::steady_clock::now() < deadline
+                       && not relayStop.load()) {
+                    std::array<char, 2048> chunk {};
+                    size_t n = socket.read_some(asio::buffer(chunk), ec);
+                    if (!ec) {
+                        body.append(chunk.data(), n);
+                    } else if (ec == asio::error::would_block || ec == asio::error::try_again) {
+                        std::this_thread::sleep_for(2ms);
+                    } else {
+                        break;
+                    }
+                }
+                if (body.size() > contentLength)
+                    body.resize(contentLength);
+
+                Json::Value payload;
+                std::string err;
+                auto reader = std::unique_ptr<Json::CharReader>(Json::CharReaderBuilder().newCharReader());
+                if (reader->parse(body.data(), body.data() + body.size(), &payload, &err)
+                    && payload.isMember("notifications") && payload["notifications"].isArray()
+                    && !payload["notifications"].empty() && payload["notifications"][0].isMember("data")) {
+                    const auto& data = payload["notifications"][0]["data"];
+                    std::map<std::string, std::string> notif;
+                    for (const auto& name : data.getMemberNames()) {
+                        const auto& value = data[name];
+                        if (value.isString())
+                            notif[name] = value.asString();
+                        else if (value.isBool())
+                            notif[name] = value.asBool() ? "true" : "false";
+                        else if (value.isUInt64())
+                            notif[name] = std::to_string(value.asUInt64());
+                        else if (value.isInt64())
+                            notif[name] = std::to_string(value.asInt64());
+                    }
+                    nodeClient.pushNotificationReceived(notif);
+                    relayedPushCount++;
+                }
+
+                static const std::string okResp = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK";
+                asio::write(socket, asio::buffer(okResp), ec);
+            }
+        } catch (...) {
+        }
+    });
+    struct RelayGuard
+    {
+        std::atomic_bool& stop;
+        std::thread& th;
+        ~RelayGuard()
+        {
+            stop = true;
+            if (th.joinable())
+                th.join();
+        }
+    } relayGuard {relayStop, pushRelay};
+
+    serverProxy.reset();
+    dht::ProxyServerConfig serverConfig;
+    serverConfig.port = proxyPort;
+    serverConfig.pushServer = "127.0.0.1:" + std::to_string(pushPort);
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, serverConfig);
+
+    clientConfig.proxy_server = "http://127.0.0.1:" + std::to_string(proxyPort);
+    clientConfig.push_token = "relay-token";
+    clientConfig.push_platform = "android";
+
+    nodePeer.registerType(chainEditableType);
+    nodeProxy->registerType(chainEditableType);
+
+    nodeClient.run(0, clientConfig);
+    nodeClient.registerType(chainEditableType);
+
+    bool done = false;
+    std::condition_variable cv;
+    std::mutex cv_m;
+    std::unique_lock<std::mutex> lk(cv_m);
+
+    auto key = dht::InfoHash::get("proxy-full-chain");
+    auto valueAuthor = dht::crypto::generateIdentity("ProxyChainAuthor");
+
+    std::vector<std::shared_ptr<dht::Value>> values;
+    auto token = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& incoming, bool expired) {
+        if (expired)
+            return true;
+        std::lock_guard<std::mutex> lk(cv_m);
+        values.insert(values.end(), incoming.begin(), incoming.end());
+        done = true;
+        cv.notify_all();
+        return true;
+    });
+
+    dht::Value firstVal {"first-version"};
+    firstVal.type = CHAIN_EDITABLE_TYPE_ID;
+    firstVal.id = 424242;
+    firstVal.seq = 1;
+    firstVal.sign(*valueAuthor.first);
+    auto firstData = firstVal.data;
+    nodePeer.put(key, std::move(firstVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        cv.notify_all();
+    });
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; }));
+    done = false;
+    CPPUNIT_ASSERT_EQUAL((size_t) 1u, values.size());
+    CPPUNIT_ASSERT_EQUAL((dht::Value::Id) 424242, values.back()->id);
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 1, values.back()->seq);
+    CPPUNIT_ASSERT(firstData == values.back()->data);
+
+    dht::Value secondVal {"second-version"};
+    secondVal.type = CHAIN_EDITABLE_TYPE_ID;
+    secondVal.id = 424242;
+    secondVal.seq = 2;
+    secondVal.sign(*valueAuthor.first);
+    auto secondData = secondVal.data;
+    nodePeer.put(key, std::move(secondVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        cv.notify_all();
+    });
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; }));
+    done = false;
+    CPPUNIT_ASSERT_EQUAL((size_t) 2u, values.size());
+    CPPUNIT_ASSERT_EQUAL((dht::Value::Id) 424242, values.back()->id);
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 2, values.back()->seq);
+    CPPUNIT_ASSERT(secondData == values.back()->data);
+    CPPUNIT_ASSERT(relayedPushCount.load() >= 2);
+
+    // Older sequence should not update current cached value
+    dht::Value staleVal {"stale-version"};
+    staleVal.type = CHAIN_EDITABLE_TYPE_ID;
+    staleVal.id = 424242;
+    staleVal.seq = 1;
+    staleVal.sign(*valueAuthor.first);
+    nodePeer.put(key, std::move(staleVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        cv.notify_all();
+    });
+    CPPUNIT_ASSERT(not cv.wait_for(lk, 1s, [&] { return done; }));
+    CPPUNIT_ASSERT_EQUAL((size_t) 2u, values.size());
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 2, values.back()->seq);
+    CPPUNIT_ASSERT(secondData == values.back()->data);
+
+    // New id should be delivered as an additional value
+    dht::Value thirdVal {"third-value"};
+    thirdVal.type = CHAIN_EDITABLE_TYPE_ID;
+    thirdVal.id = 424243;
+    thirdVal.seq = 1;
+    thirdVal.sign(*valueAuthor.first);
+    auto thirdData = thirdVal.data;
+    nodePeer.put(key, std::move(thirdVal), [&](bool ok) {
+        std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
+        cv.notify_all();
+    });
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; }));
+    done = false;
+    CPPUNIT_ASSERT_EQUAL((size_t) 3u, values.size());
+    CPPUNIT_ASSERT_EQUAL((dht::Value::Id) 424243, values.back()->id);
+    CPPUNIT_ASSERT_EQUAL((uint16_t) 1, values.back()->seq);
+    CPPUNIT_ASSERT(thirdData == values.back()->data);
+    CPPUNIT_ASSERT(relayedPushCount.load() >= 3);
+
+    nodeClient.cancelListen(key, std::move(token));
+#endif
+}
+
 void
 DhtProxyTester::testPutGet40KChars()
 {

tests/dhtproxytester.h

@@ -18,6 +18,8 @@ class DhtProxyTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testResubscribeGetValues);
+    CPPUNIT_TEST(testPushNotification);
+    CPPUNIT_TEST(testProxyServerClientFullChain);
     CPPUNIT_TEST(testPutGet40KChars);
     CPPUNIT_TEST(testFuzzy);
     CPPUNIT_TEST(testShutdownStop);
@@ -46,6 +48,14 @@ public:
      * 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
      */

tests/dhtrunnertester.cpp

@@ -79,6 +79,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()
 {

tests/dhtrunnertester.h

@@ -19,6 +19,7 @@ class DhtRunnerTester : public CppUnit::TestFixture
     CPPUNIT_TEST(testPutOverride);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
+    CPPUNIT_TEST(testStoreEmptyValue);
     CPPUNIT_TEST(testIdOps);
     CPPUNIT_TEST_SUITE_END();
 
@@ -62,6 +63,10 @@ public:
      * Test listen method with lot of datas
      */
     void testListenLotOfBytes();
+    /**
+     * Test storing an empty value
+     */
+    void testStoreEmptyValue();
     /**
      * Test multithread
      */

tests/infohashtester.cpp

@@ -6,6 +6,7 @@
 // std
 #include <iostream>
 #include <string>
+#include <string_view>
 
 // opendht
 #include "opendht/infohash.h"
@@ -126,10 +127,21 @@ InfoHashTester::testXorCmp()
 void
 InfoHashTester::testHex()
 {
-    const std::string TEST_HASH_STR("01b20304d5060708090a010203e05060708090ae");
+    using namespace std::literals;
+    static constexpr auto H = "0123456789abcdef0123456789abcdef01234567"sv;
+    const std::string TEST_HASH_STR(H);
     dht::InfoHash TEST_HASH(TEST_HASH_STR);
+    static constexpr dht::InfoHash TEST_HASH_CONST(H);
     CPPUNIT_ASSERT_EQUAL(TEST_HASH_STR, TEST_HASH.toString());
+    CPPUNIT_ASSERT_EQUAL(H, TEST_HASH.to_view());
     CPPUNIT_ASSERT_EQUAL(TEST_HASH_STR, dht::toHex(TEST_HASH.data(), TEST_HASH.size()));
+    CPPUNIT_ASSERT_EQUAL(TEST_HASH_CONST.to_view(), H);
+
+    static constexpr auto hexArray = dht::toHexArray(TEST_HASH_CONST);
+    static constexpr auto hexArrayView = std::string_view(hexArray.data(), hexArray.size());
+    static_assert(hexArrayView == H, "Hex array does not match");
+
+    CPPUNIT_ASSERT_EQUAL(std::string_view(TEST_HASH_STR), hexArrayView);
 }
 
 void

tests/opcachetester.cpp

@@ -0,0 +1,287 @@
+/*
+ *  Copyright (C) 2025 Savoir-faire Linux Inc.
+ *
+ *  Author: GitHub Copilot
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "opcachetester.h"
+
+// Hack to test internal class
+#include "../src/op_cache.cpp"
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(OpCacheTester);
+
+void
+OpCacheTester::setUp() {
+}
+
+void
+OpCacheTester::tearDown() {
+}
+
+void
+OpCacheTester::testUpdateRefCount() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 1;
+    v1->seq = 1;
+    std::string data1 = "data1";
+    v1->data = {data1.begin(), data1.end()};
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 1;
+    v2->seq = 2;
+    std::string data2 = "data2";
+    v2->data = {data2.begin(), data2.end()};
+
+    // Add v1
+    cache.onValuesAdded({v1});
+    // Should have 1 value
+    CPPUNIT_ASSERT_EQUAL((size_t)1, cache.size());
+    auto stored = cache.get(1);
+    CPPUNIT_ASSERT(stored);
+    CPPUNIT_ASSERT_EQUAL(v1->seq, stored->seq);
+
+    // Add v2 (update)
+    cache.onValuesAdded({v2});
+    // Should still have 1 value (updated)
+    CPPUNIT_ASSERT_EQUAL((size_t)1, cache.size());
+    stored = cache.get(1);
+    CPPUNIT_ASSERT(stored);
+    CPPUNIT_ASSERT_EQUAL(v2->seq, stored->seq);
+
+    // Now expire v1
+    // Note: onValuesExpired uses ID to find value.
+    cache.onValuesExpired({v1});
+
+    // If refCount was incremented on update, we should still have the value (refCount=1).
+    // If not, it will be removed (refCount=0).
+    
+    // Check if value exists
+    stored = cache.get(1);
+    
+    // This assertion will fail if the bug exists
+    CPPUNIT_ASSERT_MESSAGE("Value should still exist after expiring one source", stored != nullptr);
+    
+    if (stored) {
+        CPPUNIT_ASSERT_EQUAL(v2->seq, stored->seq);
+    }
+
+    // Expire v2
+    cache.onValuesExpired({v2});
+    stored = cache.get(1);
+    CPPUNIT_ASSERT_MESSAGE("Value should be removed after expiring all sources", stored == nullptr);
+}
+
+void
+OpCacheTester::testBasicAddExpire() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 10;
+    v1->data = {'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() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 20;
+    v1->data = {'B'};
+
+    // Add same value twice (simulating two sources)
+    cache.onValuesAdded({v1});
+    cache.onValuesAdded({v1});
+
+    CPPUNIT_ASSERT_EQUAL((size_t)1, cache.size());
+
+    // Expire once
+    cache.onValuesExpired({v1});
+    CPPUNIT_ASSERT_EQUAL((size_t)1, cache.size());
+    CPPUNIT_ASSERT(cache.get(20));
+
+    // Expire twice
+    cache.onValuesExpired({v1});
+    CPPUNIT_ASSERT_EQUAL((size_t)0, cache.size());
+}
+
+void
+OpCacheTester::testUpdateSequence() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 30;
+    v1->seq = 10;
+    v1->data = {'C'};
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 30;
+    v2->seq = 20;
+    v2->data = {'D'};
+
+    auto v3 = std::make_shared<dht::Value>();
+    v3->id = 30;
+    v3->seq = 5;
+    v3->data = {'E'};
+
+    // Add seq 10
+    cache.onValuesAdded({v1});
+    auto stored = cache.get(30);
+    CPPUNIT_ASSERT_EQUAL((uint16_t)10, stored->seq);
+
+    // Add seq 20 (update)
+    cache.onValuesAdded({v2});
+    stored = cache.get(30);
+    CPPUNIT_ASSERT_EQUAL((uint16_t)20, stored->seq);
+
+    // Add seq 5 (older, should not update content but increment refCount)
+    cache.onValuesAdded({v3});
+    stored = cache.get(30);
+    CPPUNIT_ASSERT_EQUAL((uint16_t)20, stored->seq);
+
+    // We added 3 times. Should need 3 expires to remove.
+    cache.onValuesExpired({v1}); // 2 left
+    CPPUNIT_ASSERT(cache.get(30));
+    cache.onValuesExpired({v2}); // 1 left
+    CPPUNIT_ASSERT(cache.get(30));
+    cache.onValuesExpired({v3}); // 0 left
+    CPPUNIT_ASSERT(!cache.get(30));
+}
+
+void
+OpCacheTester::testCallbacks() {
+    int addCount = 0;
+    int expireCount = 0;
+    dht::OpValueCache cache([&](const std::vector<std::shared_ptr<dht::Value>>& vals, bool expired){
+        if (expired) expireCount += vals.size();
+        else addCount += vals.size();
+        return true;
+    });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 40;
+
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL(1, addCount);
+    CPPUNIT_ASSERT_EQUAL(0, expireCount);
+
+    // Adding same value again should NOT trigger callback (no new value added)
+    // Wait, looking at code:
+    // if (viop.second) { newValues.emplace_back(v); } ...
+    // return newValues.empty() ? true : callback(newValues, false);
+    // So if it's a duplicate or update, does it callback?
+    // If update: newValues.emplace_back(v) IS called.
+    // If duplicate (refCount++ only): newValues is NOT added.
+    
+    cache.onValuesAdded({v1});
+    CPPUNIT_ASSERT_EQUAL(1, addCount); // Should still be 1
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 40;
+    v2->seq = 100; // Update
+    v2->data = {'M', 'O', 'D'}; // Change data to ensure inequality
+    
+    cache.onValuesAdded({v2});
+    CPPUNIT_ASSERT_EQUAL(2, addCount); // Should be 2 now (update triggers callback)
+
+    cache.onValuesExpired({v1});
+    CPPUNIT_ASSERT_EQUAL(0, expireCount); // RefCount dec, not removed yet
+
+    cache.onValuesExpired({v1});
+    CPPUNIT_ASSERT_EQUAL(0, expireCount); // RefCount dec, not removed yet
+
+    cache.onValuesExpired({v1});
+    CPPUNIT_ASSERT_EQUAL(1, expireCount); // Removed now
+}
+
+void
+OpCacheTester::testFilters() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    auto v1 = std::make_shared<dht::Value>();
+    v1->id = 50;
+    v1->type = 1;
+
+    auto v2 = std::make_shared<dht::Value>();
+    v2->id = 51;
+    v2->type = 2;
+
+    cache.onValuesAdded({v1, v2});
+
+    auto f1 = dht::Value::TypeFilter(1);
+    auto res1 = cache.get(f1);
+    CPPUNIT_ASSERT_EQUAL((size_t)1, res1.size());
+    CPPUNIT_ASSERT_EQUAL(v1->id, res1[0]->id);
+
+    auto f2 = dht::Value::TypeFilter(2);
+    auto res2 = cache.get(f2);
+    CPPUNIT_ASSERT_EQUAL((size_t)1, res2.size());
+    CPPUNIT_ASSERT_EQUAL(v2->id, res2[0]->id);
+    
+    auto all = cache.getValues();
+    CPPUNIT_ASSERT_EQUAL((size_t)2, all.size());
+}
+
+void
+OpCacheTester::testSyncStatus() {
+    dht::OpValueCache cache([](const std::vector<std::shared_ptr<dht::Value>>&, bool){ return true; });
+
+    // Initial state
+    CPPUNIT_ASSERT(!cache.isSynced());
+
+    // Add a node
+    cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+    // 1 node, 0 synced -> not synced
+    CPPUNIT_ASSERT(!cache.isSynced());
+
+    // Sync the node
+    cache.onNodeChanged(dht::ListenSyncStatus::SYNCED);
+    // 1 node, 1 synced -> synced
+    CPPUNIT_ASSERT(cache.isSynced());
+
+    // Add another node
+    cache.onNodeChanged(dht::ListenSyncStatus::ADDED);
+    // 2 nodes, 1 synced -> not synced
+    CPPUNIT_ASSERT(!cache.isSynced());
+
+    // Sync second node
+    cache.onNodeChanged(dht::ListenSyncStatus::SYNCED);
+    // 2 nodes, 2 synced -> synced
+    CPPUNIT_ASSERT(cache.isSynced());
+
+    // Unsync one
+    cache.onNodeChanged(dht::ListenSyncStatus::UNSYNCED);
+    CPPUNIT_ASSERT(!cache.isSynced());
+
+    // Remove one (the unsynced one)
+    cache.onNodeChanged(dht::ListenSyncStatus::REMOVED);
+    // 1 node, 1 synced -> synced
+    CPPUNIT_ASSERT(cache.isSynced());
+}
+
+}  // namespace test

tests/opcachetester.h

@@ -0,0 +1,51 @@
+/*
+ *  Copyright (C) 2025 Savoir-faire Linux Inc.
+ *
+ *  Author: GitHub Copilot
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class OpCacheTester : public CppUnit::TestFixture {
+    CPPUNIT_TEST_SUITE(OpCacheTester);
+    CPPUNIT_TEST(testUpdateRefCount);
+    CPPUNIT_TEST(testBasicAddExpire);
+    CPPUNIT_TEST(testMultipleSources);
+    CPPUNIT_TEST(testUpdateSequence);
+    CPPUNIT_TEST(testCallbacks);
+    CPPUNIT_TEST(testFilters);
+    CPPUNIT_TEST(testSyncStatus);
+    CPPUNIT_TEST_SUITE_END();
+
+public:
+    void setUp();
+    void tearDown();
+
+    void testUpdateRefCount();
+    void testBasicAddExpire();
+    void testMultipleSources();
+    void testUpdateSequence();
+    void testCallbacks();
+    void testFilters();
+    void testSyncStatus();
+};
+
+}  // namespace test

tests/valuecachetester.cpp

@@ -0,0 +1,254 @@
+/*
+ *  Copyright (C) 2025 Savoir-faire Linux Inc.
+ *
+ *  Author: GitHub Copilot
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "valuecachetester.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
+    cache.onValues({v2}, {}, {}, types, now);
+    CPPUNIT_ASSERT_EQUAL(2, addCount);
+    CPPUNIT_ASSERT_EQUAL_MESSAGE("Old value should be expired on update", 1, 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());
+}
+
+}  // namespace test

tests/valuecachetester.h

@@ -0,0 +1,53 @@
+/*
+ *  Copyright (C) 2025 Savoir-faire Linux Inc.
+ *
+ *  Author: GitHub Copilot
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#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_SUITE_END();
+
+public:
+    void setUp();
+    void tearDown();
+
+    void testUpdate();
+    void testUpdateExpiration();
+    void testExpiration();
+    void testRefresh();
+    void testExplicitExpiration();
+    void testClear();
+    void testSyncStatus();
+    void testMaxValues();
+};
+
+}  // namespace test