mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 04:21:51 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
65bbbead34c55dcfdf2fc5706247722123d006bc
opendht/tests/test_dhtrunner.cpp
T

1851 lines
60 KiB
C++
Raw Blame History

// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT
#include "test_dhtrunner.h"
#include <opendht/thread_pool.h>
#include <chrono>
#include <future>
#include <mutex>
#include <condition_variable>
#include <fstream>
#include <filesystem>
#include <cstdio>
using namespace std::chrono_literals;
using namespace std::literals;
namespace test {
CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerPushTester);
CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTransportTester);
namespace {
#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
constexpr auto stoppedPushResult = dht::PushNotificationResult::IgnoredStopped;
#else
constexpr auto stoppedPushResult = dht::PushNotificationResult::IgnoredDisabled;
#endif
}
void
DhtRunnerPushTester::testPushBeforeRun()
{
dht::DhtRunner node;
auto result = node.pushNotificationReceived({});
CPPUNIT_ASSERT(result.wait_for(0s) == std::future_status::ready);
CPPUNIT_ASSERT(result.get() == stoppedPushResult);
}
void
DhtRunnerPushTester::testPushAfterJoin()
{
dht::DhtRunner node;
node.run(0);
node.join();
auto result = node.pushNotificationReceived({});
CPPUNIT_ASSERT(result.wait_for(0s) == std::future_status::ready);
CPPUNIT_ASSERT(result.get() == stoppedPushResult);
}
void
DhtRunnerPushTester::testPushBurstAfterJoin()
{
dht::DhtRunner node;
node.run(0);
node.join();
for (unsigned i = 0; i < 32; ++i) {
auto result = node.pushNotificationReceived({});
CPPUNIT_ASSERT(result.wait_for(0s) == std::future_status::ready);
CPPUNIT_ASSERT(result.get() == stoppedPushResult);
}
}
void
DhtRunnerPushTester::testPushWhileRunning()
{
dht::DhtRunner node;
node.run(0);
auto result = node.pushNotificationReceived({});
CPPUNIT_ASSERT(result.wait_for(5s) == std::future_status::ready);
CPPUNIT_ASSERT(result.get() != dht::PushNotificationResult::IgnoredStopped);
}
template<typename T>
T
getFutureValue(std::future<T> future, std::chrono::steady_clock::duration timeout = 30s)
{
CPPUNIT_ASSERT(std::future_status::ready == future.wait_for(timeout));
return future.get();
}
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;
auto future = p.get_future();
node2.put(key, std::move(val), [&](bool ok) { p.set_value(ok); });
CPPUNIT_ASSERT(getFutureValue(std::move(future)));
auto vals = getFutureValue(node1.get(key));
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;
auto p1future = p1.get_future();
auto p2future = p2.get_future();
node2.put(key, val, [&](bool ok) { p1.set_value(ok); });
node2.put(key, val, [&](bool ok) { p2.set_value(ok); });
auto p1ret = getFutureValue(std::move(p1future));
auto p2ret = getFutureValue(std::move(p2future));
CPPUNIT_ASSERT(p1ret);
CPPUNIT_ASSERT(p2ret);
auto vals = getFutureValue(node1.get(key));
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;
auto p1future = p1.get_future();
auto p2future = p2.get_future();
node2.put(key, val, [&](bool ok) { p1.set_value(ok); });
node2.put(key, val2, [&](bool ok) { p2.set_value(ok); });
auto p1ret = getFutureValue(std::move(p1future));
auto p2ret = getFutureValue(std::move(p2future));
CPPUNIT_ASSERT(!p1ret);
CPPUNIT_ASSERT(p2ret);
auto vals = getFutureValue(node1.get(key));
CPPUNIT_ASSERT(not vals.empty());
CPPUNIT_ASSERT(vals.size() == 1);
CPPUNIT_ASSERT(vals.front()->data == val_data);
}
void
DhtRunnerTester::testImportValuesPreservesRemoteQuota()
{
node1.shutdown();
node1.join();
node2.shutdown();
node2.join();
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 = 1536;
config.dht_config.node_config.max_local_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);
std::this_thread::sleep_for(1s);
auto key1 = dht::InfoHash::get("import-quota-1");
auto key2 = dht::InfoHash::get("import-quota-2");
auto value1 = std::make_shared<dht::Value>(std::string(1024, 'a'));
auto value2 = std::make_shared<dht::Value>(std::string(1024, 'b'));
std::promise<bool> p1;
auto p1future = p1.get_future();
node2.put(key1, value1, [&](bool ok) { p1.set_value(ok); });
CPPUNIT_ASSERT(getFutureValue(std::move(p1future)));
std::this_thread::sleep_for(2s);
auto exported = node1.exportValues();
CPPUNIT_ASSERT(!exported.empty());
node1.shutdown();
node1.join();
dht::DhtRunner restored;
restored.run(0, config);
auto restoredBound = restored.getBound();
if (restoredBound.isUnspecified())
restoredBound.setLoopback();
node2.shutdown();
node2.join();
node2.run(0, config);
node2.bootstrap(restoredBound);
std::this_thread::sleep_for(1s);
restored.importValues(exported);
std::this_thread::sleep_for(500ms);
std::promise<bool> p2;
auto p2future = p2.get_future();
node2.put(key2, value2, [&](bool ok) { p2.set_value(ok); });
CPPUNIT_ASSERT(getFutureValue(std::move(p2future)));
std::this_thread::sleep_for(2s);
auto vals1 = getFutureValue(restored.get(key1));
auto vals2 = getFutureValue(restored.get(key2));
CPPUNIT_ASSERT(vals1.size() + vals2.size() <= 1);
restored.shutdown();
restored.join();
}
void
DhtRunnerTester::testExportValuesUsesSystemTime()
{
auto key = dht::InfoHash::get("export-system-time");
const auto created = dht::clock::now();
msgpack::sbuffer importBuffer;
msgpack::packer<msgpack::sbuffer> pk(&importBuffer);
pk.pack_array(1);
pk.pack_array(2);
pk.pack(created.time_since_epoch().count());
dht::Value {std::string("persisted")}.msgpack_pack(pk);
node1.importValues({
dht::ValuesExport {key, {importBuffer.data(), importBuffer.data() + importBuffer.size()}}
});
auto exported = node1.exportValues();
CPPUNIT_ASSERT_EQUAL(size_t(1), exported.size());
msgpack::unpacked msg;
msgpack::unpack(msg, reinterpret_cast<const char*>(exported.front().second.data()), exported.front().second.size());
const auto values = msg.get();
CPPUNIT_ASSERT(values.type == msgpack::type::ARRAY);
CPPUNIT_ASSERT_EQUAL(uint32_t(1), values.via.array.size);
const auto& persistedValue = values.via.array.ptr[0];
CPPUNIT_ASSERT_EQUAL(uint32_t(4), persistedValue.via.array.size);
CPPUNIT_ASSERT(persistedValue.via.array.ptr[0].type == msgpack::type::EXT);
CPPUNIT_ASSERT(persistedValue.via.array.ptr[3].type == msgpack::type::EXT);
const auto persistedCreated = persistedValue.via.array.ptr[0].as<std::chrono::system_clock::time_point>();
const auto persistedExpiration = persistedValue.via.array.ptr[3].as<std::chrono::system_clock::time_point>();
const auto delta = std::chrono::system_clock::now() - persistedCreated;
CPPUNIT_ASSERT(delta >= std::chrono::system_clock::duration::zero());
CPPUNIT_ASSERT(delta < 5s);
CPPUNIT_ASSERT(persistedExpiration > persistedCreated);
}
void
DhtRunnerTester::testImportValuesPreservesStoredExpiration()
{
static const dht::ValueType SHORT_LIVED_TYPE {200, "Short lived", 2s};
node1.registerInsecureType(SHORT_LIVED_TYPE);
auto key = dht::InfoHash::get("import-expiration");
auto now = dht::clock::now();
auto created = now;
auto expiration = now + SHORT_LIVED_TYPE.expiration;
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_array(1);
pk.pack_array(4);
pk.pack(created.time_since_epoch().count());
dht::Value {SHORT_LIVED_TYPE.id, std::string("short-lived")}.msgpack_pack(pk);
pk.pack_bin(0);
pk.pack(expiration.time_since_epoch().count());
node1.importValues({
dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
});
auto exported = node1.exportValues();
CPPUNIT_ASSERT(!exported.empty());
dht::DhtRunner restored;
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;
restored.run(0, config);
restored.importValues(exported);
CPPUNIT_ASSERT(!restored.exportValues().empty());
std::this_thread::sleep_for(3s);
auto restoredValues = restored.exportValues();
CPPUNIT_ASSERT_EQUAL(size_t(1), restoredValues.size());
msgpack::unpacked msg;
msgpack::unpack(msg, (const char*) restoredValues.front().second.data(), restoredValues.front().second.size());
auto valarr = msg.get();
CPPUNIT_ASSERT(valarr.type == msgpack::type::ARRAY);
CPPUNIT_ASSERT_EQUAL(uint32_t(0), valarr.via.array.size);
restored.shutdown();
restored.join();
}
void
DhtRunnerTester::testImportPermanentValueExpires()
{
static const dht::ValueType SHORT_TYPE {201, "Short perm", 500ms};
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;
dht::DhtRunner restored;
restored.run(0, config);
restored.registerInsecureType(SHORT_TYPE);
// Ensure the runner is waiting on a previously computed scheduler deadline.
std::promise<void> loopCompleted;
auto loopCompletedFuture = loopCompleted.get_future();
restored.getNodeInfo([&loopCompleted](auto) { loopCompleted.set_value(); });
getFutureValue(std::move(loopCompletedFuture));
auto key = dht::InfoHash::get("import-permanent-expiration");
auto now = dht::clock::now();
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_array(1);
pk.pack_array(4);
pk.pack(now.time_since_epoch().count());
dht::Value {SHORT_TYPE.id, std::string("was-permanent")}.msgpack_pack(pk);
pk.pack_bin(0);
pk.pack(dht::clock::time_point::max().time_since_epoch().count());
restored.importValues({
dht::ValuesExport {key, {buffer.data(), buffer.data() + buffer.size()}}
});
auto containsValue = [&] {
for (const auto& [exportedKey, data] : restored.exportValues()) {
if (exportedKey != key)
continue;
msgpack::unpacked msg;
msgpack::unpack(msg, reinterpret_cast<const char*>(data.data()), data.size());
auto valarr = msg.get();
CPPUNIT_ASSERT(valarr.type == msgpack::type::ARRAY);
return valarr.via.array.size != 0;
}
return false;
};
CPPUNIT_ASSERT(containsValue());
auto timeout = dht::clock::now() + 2s;
while (containsValue() and dht::clock::now() < timeout)
std::this_thread::sleep_for(10ms);
CPPUNIT_ASSERT(!containsValue());
restored.shutdown();
restored.join();
}
void
DhtRunnerTester::testBootstrapSetsConnectingState()
{
node1.shutdown();
node1.join();
node2.shutdown();
node2.join();
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;
std::mutex mutex;
std::condition_variable cv;
bool sawConnecting {false};
dht::DhtRunner::Context context;
context.statusChangedCallback = [&](dht::NodeStatus status4, dht::NodeStatus status6) {
std::lock_guard lock(mutex);
sawConnecting = sawConnecting || status4 == dht::NodeStatus::Connecting
|| status6 == dht::NodeStatus::Connecting;
cv.notify_all();
};
dht::DhtRunner bootstrapNode;
dht::DhtRunner clientNode;
bootstrapNode.run(0, config);
clientNode.run(config, std::move(context));
auto bound = bootstrapNode.getBound();
if (bound.isUnspecified())
bound.setLoopback();
clientNode.bootstrap("127.0.0.1", std::to_string(bound.getPort()));
{
std::unique_lock lock(mutex);
CPPUNIT_ASSERT(cv.wait_for(lock, 5s, [&] { return sawConnecting; }));
}
bootstrapNode.shutdown();
bootstrapNode.join();
clientNode.shutdown();
clientNode.join();
}
void
DhtRunnerTester::testBootstrapThenPutNoRace()
{
node1.shutdown();
node1.join();
node2.shutdown();
node2.join();
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;
dht::DhtRunner bootstrapNode;
dht::DhtRunner clientNode;
bootstrapNode.run(0, config);
clientNode.run(0, config);
auto bound = bootstrapNode.getBound();
if (bound.isUnspecified())
bound.setLoopback();
clientNode.bootstrap(bound);
std::promise<bool> putDone;
auto putFuture = putDone.get_future();
clientNode.put(dht::InfoHash::get("bootstrap-then-put"), dht::Value {"value"}, [&](bool ok) {
putDone.set_value(ok);
});
CPPUNIT_ASSERT(getFutureValue(std::move(putFuture)));
auto vals = getFutureValue(bootstrapNode.get(dht::InfoHash::get("bootstrap-then-put")));
CPPUNIT_ASSERT(!vals.empty());
bootstrapNode.shutdown();
bootstrapNode.join();
clientNode.shutdown();
clientNode.join();
}
void
DhtRunnerTester::testBootstrapMissingNodeThenPutFails()
{
node1.shutdown();
node1.join();
node2.shutdown();
node2.join();
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;
auto getUnusedPort = [&]() {
dht::DhtRunner tmpNode;
tmpNode.run(0, config);
auto port = tmpNode.getBoundPort();
tmpNode.shutdown();
tmpNode.join();
return port;
};
auto missingPort = getUnusedPort();
std::mutex mutex;
std::condition_variable cv;
bool sawConnecting {false};
bool sawDisconnectedAfterConnecting {false};
bool putReturnedBeforeDisconnect {false};
dht::DhtRunner::Context context;
context.statusChangedCallback = [&](dht::NodeStatus status4, dht::NodeStatus status6) {
std::lock_guard lock(mutex);
if (status4 == dht::NodeStatus::Connecting || status6 == dht::NodeStatus::Connecting)
sawConnecting = true;
if (sawConnecting && status4 == dht::NodeStatus::Disconnected && status6 == dht::NodeStatus::Disconnected)
sawDisconnectedAfterConnecting = true;
cv.notify_all();
};
dht::DhtRunner clientNode;
clientNode.run(0, config, std::move(context));
while (clientNode.getBoundPort() == missingPort)
missingPort = getUnusedPort();
std::promise<bool> putDone;
auto putFuture = putDone.get_future();
clientNode.bootstrap("127.0.0.1", std::to_string(missingPort));
clientNode.put(dht::InfoHash::get("bootstrap-missing-put"), dht::Value {"value"}, [&](bool ok) {
std::lock_guard lock(mutex);
putReturnedBeforeDisconnect = !sawDisconnectedAfterConnecting;
putDone.set_value(ok);
cv.notify_all();
});
{
std::unique_lock lock(mutex);
CPPUNIT_ASSERT(cv.wait_for(lock, 10s, [&] { return sawConnecting; }));
CPPUNIT_ASSERT(cv.wait_for(lock, 20s, [&] { return sawDisconnectedAfterConnecting; }));
}
CPPUNIT_ASSERT(std::future_status::ready == putFuture.wait_for(20s));
CPPUNIT_ASSERT(!putFuture.get());
CPPUNIT_ASSERT(!putReturnedBeforeDisconnect);
clientNode.shutdown();
clientNode.join();
}
void
DhtRunnerTester::testShutdownCompletesWithPendingPut()
{
node1.shutdown();
node1.join();
node2.shutdown();
node2.join();
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;
auto getUnusedPort = [&]() {
dht::DhtRunner tmpNode;
tmpNode.run(0, config);
auto port = tmpNode.getBoundPort();
tmpNode.shutdown();
tmpNode.join();
return port;
};
auto missingPort = getUnusedPort();
dht::DhtRunner clientNode;
clientNode.run(0, config);
while (clientNode.getBoundPort() == missingPort)
missingPort = getUnusedPort();
clientNode.bootstrap("127.0.0.1", std::to_string(missingPort));
clientNode.put(dht::InfoHash::get("shutdown-pending-put"), dht::Value {"value"}, [](bool) {});
std::promise<void> shutdownDone;
auto shutdownFuture = shutdownDone.get_future();
clientNode.shutdown([&]() { shutdownDone.set_value(); });
CPPUNIT_ASSERT(std::future_status::ready == shutdownFuture.wait_for(5s));
clientNode.join();
}
void
DhtRunnerTester::testSavedPortFallback()
{
// A saved port that is already taken by another process (e.g. fast
// application restart while the old instance still holds it) must not
// prevent the node from starting: it should fall back to a random port.
dht::DhtRunner::Config config;
config.dht_config.node_config.persist_path = "/tmp/opendht_test_port_fallback";
std::string portFile = config.dht_config.node_config.persist_path + "_port.txt";
// node1 (from setUp) holds a port; save it as our persisted port.
auto takenPort = node1.getBoundPort();
{
std::ofstream out(portFile);
out << takenPort << std::endl << takenPort << std::endl;
}
dht::DhtRunner testNode;
CPPUNIT_ASSERT_NO_THROW(testNode.run(0, config));
CPPUNIT_ASSERT(testNode.getBoundPort());
CPPUNIT_ASSERT(testNode.getBoundPort() != takenPort);
testNode.shutdown();
testNode.join();
std::remove(portFile.c_str());
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()
{
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 = getFutureValue(std::move(ftokena));
auto tokenc = getFutureValue(std::move(ftokenc));
auto tokend = getFutureValue(std::move(ftokend));
// 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(0, 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.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 == 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(0, 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.getPublicKey()->getLongId(), std::make_shared<dht::Value>("yo"), [&](bool ok) {
CPPUNIT_ASSERT(ok);
std::lock_guard lk(mutex);
valueCount++;
cv.notify_all();
});
node1.putEncrypted(key, node2.getPublicKey(), 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();
});
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));
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; }));
lk.unlock();
node2.cancelListen(key2, listenToken.get());
}
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, getFutureValue(std::move(ftokenfoo)));
}
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);
}
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 {
/* UDP socket recording sent packets */
class RecordingSocket : public dht::net::DatagramSocket
{
public:
RecordingSocket()
: inner_(std::make_unique<dht::net::UdpSocket>(0))
{
inner_->setOnReceive([this](dht::net::PacketList&& packets) {
onReceived(std::move(packets));
return dht::net::PacketList {};
});
}
~RecordingSocket() { inner_.reset(); }
int sendTo(const dht::SockAddr& dest, const uint8_t* data, size_t size, bool replied) override
{
{
std::lock_guard lk(sentLock_);
sent_.emplace_back(data, data + size);
}
return inner_->sendTo(dest, data, size, replied);
}
bool hasIPv4() const override { return inner_->hasIPv4(); }
bool hasIPv6() const override { return inner_->hasIPv6(); }
const dht::SockAddr& getBoundRef(sa_family_t family = AF_UNSPEC) const override
{
return inner_->getBoundRef(family);
}
void stop() override { inner_->stop(); }
std::vector<dht::Blob> sent() const
{
std::lock_guard lk(sentLock_);
return sent_;
}
private:
std::unique_ptr<dht::net::UdpSocket> inner_;
mutable std::mutex sentLock_;
std::vector<dht::Blob> sent_;
};
dht::DhtRunner::Config
transportTestConfig(dht::TransportConfig transport = {})
{
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.transport = std::move(transport);
return config;
}
RecordingSocket*
runRecorded(dht::DhtRunner& node, const dht::DhtRunner::Config& config)
{
auto socket = std::make_unique<RecordingSocket>();
auto ptr = socket.get();
dht::DhtRunner::Context context;
context.sock = std::move(socket);
node.run(config, std::move(context));
return ptr;
}
dht::SockAddr
localBound(const dht::DhtRunner& node)
{
auto bound = node.getBound();
if (bound.isUnspecified())
bound.setLoopback();
return bound;
}
bool
isEnvelope(const dht::Blob& packet)
{
return dht::transport::isEnvelope(packet.data(), packet.size());
}
void
putGet(dht::DhtRunner& from, dht::DhtRunner& to, const std::string& keyName)
{
auto key = dht::InfoHash::get(keyName);
dht::Value val {keyName};
auto data = val.data;
std::promise<bool> p;
auto future = p.get_future();
from.put(key, std::move(val), [&](bool ok) { p.set_value(ok); });
CPPUNIT_ASSERT(getFutureValue(std::move(future)));
auto vals = getFutureValue(to.get(key));
CPPUNIT_ASSERT(not vals.empty());
CPPUNIT_ASSERT(vals.front()->data == data);
}
dht::TransportConfig
privateTransport(const dht::transport::Authority& authority, std::chrono::system_clock::duration validity = 1h)
{
dht::TransportConfig transport;
transport.key = dht::transport::KeyPair::generate();
transport.authorities = {authority.pub};
transport.credential = authority.issue(transport.key.pub, std::chrono::system_clock::now() + validity);
return transport;
}
} // namespace
void
DhtRunnerTransportTester::testEncryptedByDefault()
{
dht::DhtRunner a, b;
auto socketA = runRecorded(a, transportTestConfig());
auto socketB = runRecorded(b, transportTestConfig());
b.bootstrap(localBound(a));
putGet(b, a, "encrypted");
// a learns b's key from its first request: everything a sends is encrypted
auto sentA = socketA->sent();
CPPUNIT_ASSERT(not sentA.empty());
CPPUNIT_ASSERT(std::all_of(sentA.begin(), sentA.end(), isEnvelope));
// b sends plaintext only until it learns a's key
auto sentB = socketB->sent();
auto firstEnvelope = std::find_if(sentB.begin(), sentB.end(), isEnvelope);
CPPUNIT_ASSERT(firstEnvelope != sentB.end());
CPPUNIT_ASSERT(std::all_of(firstEnvelope, sentB.end(), isEnvelope));
}
void
DhtRunnerTransportTester::testLegacyInterop()
{
dht::TransportConfig legacyTransport;
legacyTransport.enabled = false;
dht::DhtRunner node, legacy;
auto socket = runRecorded(node, transportTestConfig());
auto legacyConfig = transportTestConfig(legacyTransport);
legacy.run(0, legacyConfig);
legacy.bootstrap(localBound(node));
putGet(legacy, node, "legacy to new");
putGet(node, legacy, "new to legacy");
auto sent = socket->sent();
CPPUNIT_ASSERT(not sent.empty());
CPPUNIT_ASSERT(std::none_of(sent.begin(), sent.end(), isEnvelope));
}
void
DhtRunnerTransportTester::testNodeIdFromTransportKey()
{
std::mt19937_64 rd(42);
auto configuredId = dht::InfoHash::getRandom(rd);
dht::TransportConfig transport;
transport.key = dht::transport::KeyPair::generate();
auto config = transportTestConfig(transport);
config.dht_config.node_config.node_id = configuredId;
dht::DhtRunner node;
node.run(0, config);
CPPUNIT_ASSERT_EQUAL(dht::transport::nodeId(transport.key.pub), node.getNodeId());
transport.enabled = false;
config = transportTestConfig(transport);
config.dht_config.node_config.node_id = configuredId;
dht::DhtRunner legacy;
legacy.run(0, config);
CPPUNIT_ASSERT_EQUAL(configuredId, legacy.getNodeId());
}
void
DhtRunnerTransportTester::testPersistedTransportKey()
{
const std::string path = "/tmp/opendht_test_transport_state";
const auto previous = path + ".previous";
std::remove(path.c_str());
std::remove(previous.c_str());
auto runNode = [&](const dht::TransportConfig& transport) {
auto config = transportTestConfig(transport);
config.dht_config.node_config.persist_path = path;
dht::DhtRunner node;
node.run(0, config);
auto id = node.getNodeId();
std::promise<void> stopped;
node.shutdown([&] { stopped.set_value(); });
stopped.get_future().wait();
node.join();
return id;
};
// A generated key is persisted, keeping the node ID
auto generatedId = runNode({});
CPPUNIT_ASSERT_EQUAL(generatedId, runNode({}));
auto perms = std::filesystem::status(path).permissions();
CPPUNIT_ASSERT((perms & (std::filesystem::perms::group_all | std::filesystem::perms::others_all))
== std::filesystem::perms::none);
// A configured key takes precedence and isn't persisted
dht::TransportConfig configured;
configured.key = dht::transport::KeyPair::generate();
std::filesystem::create_hard_link(path, previous);
auto previousSize = std::filesystem::file_size(previous);
CPPUNIT_ASSERT_EQUAL(dht::transport::nodeId(configured.key.pub), runNode(configured));
CPPUNIT_ASSERT(not std::filesystem::equivalent(path, previous));
CPPUNIT_ASSERT_EQUAL(previousSize, std::filesystem::file_size(previous));
auto newId = runNode({});
CPPUNIT_ASSERT(newId != generatedId);
CPPUNIT_ASSERT(newId != dht::transport::nodeId(configured.key.pub));
CPPUNIT_ASSERT_EQUAL(newId, runNode({}));
#ifndef _WIN32
std::filesystem::remove(path);
std::filesystem::create_symlink(previous, path);
CPPUNIT_ASSERT_EQUAL(generatedId, runNode({}));
CPPUNIT_ASSERT(not std::filesystem::is_symlink(path));
CPPUNIT_ASSERT_EQUAL(previousSize, std::filesystem::file_size(previous));
#endif
std::remove(path.c_str());
std::remove(previous.c_str());
}
void
DhtRunnerTransportTester::testPrivateNetwork()
{
auto authority = dht::transport::Authority::generate();
dht::DhtRunner member1, member2, outsider;
auto socket1 = runRecorded(member1, transportTestConfig(privateTransport(authority)));
auto config2 = transportTestConfig(privateTransport(authority));
member2.run(0, config2);
auto outsiderConfig = transportTestConfig(privateTransport(dht::transport::Authority::generate()));
outsider.run(0, outsiderConfig);
member2.bootstrap(localBound(member1));
putGet(member2, member1, "private");
outsider.bootstrap(localBound(member1));
std::this_thread::sleep_for(2s);
CPPUNIT_ASSERT_EQUAL(0u, outsider.getNodesStats(AF_INET).getKnownNodes());
CPPUNIT_ASSERT_EQUAL(1u, member1.getNodesStats(AF_INET).getKnownNodes());
auto sent = socket1->sent();
CPPUNIT_ASSERT(not sent.empty());
CPPUNIT_ASSERT(std::all_of(sent.begin(), sent.end(), isEnvelope));
}
void
DhtRunnerTransportTester::testInvalidPrivateConfig()
{
auto authority = dht::transport::Authority::generate();
auto otherKey = privateTransport(authority);
otherKey.key = dht::transport::KeyPair::generate();
dht::DhtRunner node1;
auto config = transportTestConfig(otherKey);
CPPUNIT_ASSERT_THROW(node1.run(0, config), dht::DhtException);
dht::DhtRunner node2;
config = transportTestConfig(privateTransport(authority, -1s));
CPPUNIT_ASSERT_THROW(node2.run(0, config), dht::DhtException);
auto disabled = privateTransport(authority);
disabled.enabled = false;
dht::DhtRunner node3;
config = transportTestConfig(disabled);
CPPUNIT_ASSERT_THROW(node3.run(0, config), dht::DhtException);
}
} // namespace test
Reference in new issue View Git Blame Copy Permalink