opendht/tests/dhtrunnertester.cpp

// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT

#include "dhtrunnertester.h"

#include <opendht/thread_pool.h>

#include <chrono>
#include <mutex>
#include <condition_variable>
using namespace std::chrono_literals;
using namespace std::literals;

namespace test {
CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);

void
DhtRunnerTester::setUp()
{
    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 = -1;
    config.dht_config.node_config.max_store_keys = -1;

    node1.run(0, config);
    node2.run(0, config);
    auto bound = node1.getBound();
    if (bound.isUnspecified())
        bound.setLoopback();
    node2.bootstrap(bound);
}

void
DhtRunnerTester::tearDown()
{
    unsigned done {0};
    std::condition_variable cv;
    std::mutex cv_m;
    auto shutdown = [&] {
        std::lock_guard lk(cv_m);
        done++;
        cv.notify_all();
    };
    node1.shutdown(shutdown);
    node2.shutdown(shutdown);
    std::unique_lock lk(cv_m);
    CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return done == 2u; }));
    node1.join();
    node2.join();
}

void
DhtRunnerTester::testConstructors()
{
    CPPUNIT_ASSERT(node1.getBoundPort());
    CPPUNIT_ASSERT_EQUAL(node1.getBoundPort(), node1.getBound().getPort());
    CPPUNIT_ASSERT(node2.getBoundPort());
    CPPUNIT_ASSERT_EQUAL(node2.getBoundPort(), node2.getBound().getPort());

    dht::DhtRunner::Config config {};
    dht::DhtRunner::Context context {};
    dht::DhtRunner testNode;
    testNode.run(config, std::move(context));
    CPPUNIT_ASSERT(testNode.getBoundPort());
}

void
DhtRunnerTester::testGetPut()
{
    auto key = dht::InfoHash::get("123");
    dht::Value val {"hey"};
    auto val_data = val.data;
    std::promise<bool> p;
    node2.put(key, std::move(val), [&](bool ok) { p.set_value(ok); });
    CPPUNIT_ASSERT(p.get_future().get());
    auto vals = node1.get(key).get();
    CPPUNIT_ASSERT(not vals.empty());
    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()
{
    auto key = dht::InfoHash::get("123");
    auto val = std::make_shared<dht::Value>("hey");
    val->id = 42;
    auto val_data = val->data;
    std::promise<bool> p1;
    std::promise<bool> p2;
    node2.put(key, val, [&](bool ok) { p1.set_value(ok); });
    node2.put(key, val, [&](bool ok) { p2.set_value(ok); });
    auto p1ret = p1.get_future().get();
    auto p2ret = p2.get_future().get();
    CPPUNIT_ASSERT(p1ret);
    CPPUNIT_ASSERT(p2ret);
    auto vals = node1.get(key).get();
    CPPUNIT_ASSERT(not vals.empty());
    CPPUNIT_ASSERT(vals.size() == 1);
    CPPUNIT_ASSERT(vals.front()->data == val_data);
}

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

void
DhtRunnerTester::testListen()
{
    std::mutex mutex;
    std::condition_variable cv;
    std::atomic_uint valueCounta(0);
    std::atomic_uint valueCountb(0);
    std::atomic_uint valueCountc(0);
    std::atomic_uint valueCountd(0);

    unsigned putCount(0);
    unsigned putOkCount1(0);
    unsigned putOkCount2(0);
    unsigned putOkCount3(0);

    auto a = dht::InfoHash::get("234");
    auto b = dht::InfoHash::get("2345");
    auto c = dht::InfoHash::get("23456");
    auto d = dht::InfoHash::get("234567");
    constexpr unsigned N = 256;
    constexpr unsigned SZ = 56 * 1024;

    auto ftokena = node1.listen(a, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
        if (expired)
            valueCounta -= values.size();
        else
            valueCounta += values.size();
        return true;
    });

    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&) {
        /*if (expired)
            valueCountb -= values.size();
        else
            valueCountb += values.size();*/
        valueCountb++;
        return false;
    });

    auto ftokenc = node1.listen(c, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
        if (expired)
            valueCountc -= values.size();
        else
            valueCountc += values.size();
        return true;
    });

    auto ftokend = node1.listen(d, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
        if (expired)
            valueCountd -= values.size();
        else
            valueCountd += values.size();
        return true;
    });

    std::vector<uint8_t> mtu;
    mtu.reserve(SZ);
    for (size_t i = 0; i < SZ; i++)
        mtu.emplace_back((i % 2) ? 'T' : 'M');

    for (unsigned i = 0; i < N; i++) {
        node2.put(a, dht::Value("v1"), [&](bool ok) {
            std::lock_guard lock(mutex);
            putCount++;
            if (ok)
                putOkCount1++;
            cv.notify_all();
        });
        node2.put(b, dht::Value("v2"), [&](bool ok) {
            std::lock_guard lock(mutex);
            putCount++;
            if (ok)
                putOkCount2++;
            cv.notify_all();
        });
        auto bigVal = std::make_shared<dht::Value>();
        bigVal->data = mtu;
        node2.put(c, std::move(bigVal), [&](bool ok) {
            std::lock_guard lock(mutex);
            putCount++;
            if (ok)
                putOkCount3++;
            cv.notify_all();
        });
    }

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return putCount == N * 3u; }));
        CPPUNIT_ASSERT_EQUAL(N, putOkCount1);
        CPPUNIT_ASSERT_EQUAL(N, putOkCount2);
        CPPUNIT_ASSERT_EQUAL(N, putOkCount3);
    }

    CPPUNIT_ASSERT(ftokena.valid());
    CPPUNIT_ASSERT(ftokenb.valid());
    CPPUNIT_ASSERT(ftokenc.valid());
    CPPUNIT_ASSERT(ftokend.valid());

    auto tokena = ftokena.get();
    auto tokenc = ftokenc.get();
    auto tokend = ftokend.get();
    // tokenb might be 0 since the callback returns false.

    CPPUNIT_ASSERT(tokena);
    CPPUNIT_ASSERT(tokenc);
    CPPUNIT_ASSERT(tokend);
    CPPUNIT_ASSERT_EQUAL(N, valueCounta.load());
    CPPUNIT_ASSERT_EQUAL(1u, valueCountb.load());
    CPPUNIT_ASSERT_EQUAL(N, valueCountc.load());
    CPPUNIT_ASSERT_EQUAL(0u, valueCountd.load());

    node1.cancelListen(a, tokena);
    node1.cancelListen(b, std::move(ftokenb));
    node1.cancelListen(c, tokenc);
    node1.cancelListen(d, tokend);
}

void
DhtRunnerTester::testIdOps()
{
    std::mutex mutex;
    std::condition_variable cv;
    unsigned identityCount(0);
    unsigned valueCount(0);
    unsigned valueCountEdit(0);

    dht::DhtRunner::Config config2;
    config2.dht_config.node_config.max_peer_req_per_sec = -1;
    config2.dht_config.node_config.max_req_per_sec = -1;
    config2.dht_config.id = dht::crypto::generateIdentity();

    dht::DhtRunner::Context context2;
    context2.identityAnnouncedCb = [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        identityCount++;
        cv.notify_all();
    };

    node2.join();
    node2.run(42232, config2, std::move(context2));
    auto bound = node1.getBound();
    if (bound.isUnspecified())
        bound.setLoopback();
    node2.bootstrap(bound);

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return identityCount == 1; }));
    }

    node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt) {
        CPPUNIT_ASSERT(crt);
        std::lock_guard lk(mutex);
        valueCount++;
        cv.notify_all();
    });

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCount == 1u; }));
        CPPUNIT_ASSERT_EQUAL(1u, identityCount);
    }

    dht::DhtRunner::Context context1;
    context1.identityAnnouncedCb = [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        identityCount++;
        cv.notify_all();
    };

    config2.dht_config.id = dht::crypto::generateIdentity();
    node1.join();
    node1.run(42222, config2, std::move(context1));
    bound = node2.getBound();
    if (bound.isUnspecified())
        bound.setLoopback();
    node1.bootstrap(bound);

    auto key = dht::InfoHash::get("key");
    node1.putEncrypted(key, node2.getId(), dht::Value("yo"), [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        valueCount++;
        cv.notify_all();
    });

    node1.putEncrypted(key, node2.getPublicKey(), std::make_shared<dht::Value>("yo"), [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        valueCount++;
        cv.notify_all();
    });

    node2.listen<std::string>(key, [&](std::string&& value) {
        CPPUNIT_ASSERT_EQUAL("yo"s, value);
        std::lock_guard lk(mutex);
        valueCount++;
        cv.notify_all();
        return true;
    });

    auto key2 = dht::InfoHash::get("key2");
    auto editValue = std::make_shared<dht::Value>("v1");
    node1.putSigned(key2, editValue, [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        valueCountEdit++;
        cv.notify_all();
    });
    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));
            else if (v->seq == 1)
                CPPUNIT_ASSERT_EQUAL("v2"s, dht::unpackMsg<std::string>(v->data));
            CPPUNIT_ASSERT_EQUAL(v->owner->getLongId(), node1.getPublicKey()->getLongId());
        }
        std::lock_guard lk(mutex);
        valueCountEdit += values.size();
        cv.notify_all();
        return true;
    });

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return identityCount == 2u; }));
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCount == 5u; }));
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCountEdit == 2u; }));
    }

    node2.findCertificate(node1.getPublicKey()->getLongId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt) {
        CPPUNIT_ASSERT(crt);
        std::lock_guard lk(mutex);
        valueCount++;
        cv.notify_all();
    });

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCount == 6u; }));
    }

    // editValue->data = dht::packMsg("v2");
    editValue = std::make_shared<dht::Value>(editValue->id);
    editValue->data = dht::packMsg("v2");
    node1.putSigned(key2, editValue, [&](bool ok) {
        CPPUNIT_ASSERT(ok);
        std::lock_guard lk(mutex);
        valueCountEdit++;
        cv.notify_all();
    });
    std::unique_lock lk(mutex);
    CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&] { return valueCountEdit == 4u; }));
}

void
DhtRunnerTester::testListenLotOfBytes()
{
    std::mutex mutex;
    std::condition_variable cv;
    std::atomic_uint valueCount(0);
    unsigned putCount(0);
    unsigned putOkCount(0);

    std::string data(10000, 'a');

    auto foo = dht::InfoHash::get("foo");
    constexpr unsigned N = 1024;

    for (unsigned i = 0; i < N; i++) {
        node2.put(foo, data, [&](bool ok) {
            std::lock_guard lock(mutex);
            putCount++;
            if (ok)
                putOkCount++;
            cv.notify_all();
        });
    }
    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return putCount == N; }));
    }

    dht::DhtRunner node3 {};
    dht::DhtRunner::Config config;
    config.dht_config.node_config.max_peer_req_per_sec = -1;
    config.dht_config.node_config.max_req_per_sec = -1;
    node3.run(42242, config);
    auto bound = node1.getBound();
    if (bound.isUnspecified())
        bound.setLoopback();
    node3.bootstrap(bound);

    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
        valueCount++;
        cv.notify_all();
        return true;
    });

    {
        std::unique_lock lk(mutex);
        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return valueCount == N; }));
    }

    node3.cancelListen(foo, ftokenfoo.get());
}

void
DhtRunnerTester::testMultithread()
{
    std::mutex mutex;
    std::condition_variable cv;
    unsigned putCount(0);
    unsigned putOkCount(0);

    constexpr unsigned N = 2048;

    for (unsigned i = 0; i < N; i++) {
        dht::ThreadPool::computation().run([&] {
            node2.put(dht::InfoHash::get("123" + std::to_string(i)), "hehe", [&](bool ok) {
                std::lock_guard lock(mutex);
                putCount++;
                if (ok)
                    putOkCount++;
                cv.notify_all();
            });
            node2.get(
                dht::InfoHash::get("123" + std::to_string(N - i - 1)),
                [](const std::shared_ptr<dht::Value>&) { return true; },
                [&](bool ok) {
                    std::lock_guard lock(mutex);
                    putCount++;
                    if (ok)
                        putOkCount++;
                    cv.notify_all();
                });
        });
    }
    std::unique_lock lk(mutex);
    CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&] { return putCount == 2 * N; }));
    CPPUNIT_ASSERT_EQUAL(2 * N, putOkCount);
}

} // namespace test