e0d15dc680* feat(p2p): log durable writer quarantines and outbound session failures
A node that durably quarantined a peer writer left no trace of why: the
three p2p quarantine sites returned an error to their session, and the
reconnect loop dropped every outbound session error without logging it,
so a mesh that stopped receiving one writer showed only a quarantined
record count. Outbound sessions are half of all sessions in a full mesh.
Each p2p quarantine site now logs the writer, the failed check and the
local and frozen counters, and a failed outbound session logs its peer
and error, matching the existing inbound failure log.
Signed-off-by: trvon <me@trevon.dev>
* fix(p2p): stop quarantining honest writers for concurrent-session races
Each pair of direct peers normally runs two sessions at once, one
initiated from each side. Both freeze a window of the peer's writer at
handshake, and whichever applies first advances the receiver's durable
prefix under the other. The second session then compared its frozen
window with that live prefix and treated the difference as forged
history: validateHistoryExtension demanded the window start right after
the live commitment ("history extension is not contiguous"), and an
empty window demanded the live counter equal the frozen frontier. Both
durably quarantined an honest writer, after which the node dropped its
documents and refused every later session with it. A 5-node mesh run
logged exactly this (local=30 frontier=32 staged=4: a window frozen at
28 arriving after another session had applied 29-30), and the mesh
never converged.
The receiver now validates a window against durable history instead of
a snapshot. Operations at or below the advanced prefix are replays that
applyDeltas leaves untouched (a different record under a known operation
id is still a fork). The rest must chain from the durable commitment
exactly to the authenticated frontier, and an empty window must match
the durable commitment at its frontier. validateHistoryExtension returns
a typed WriterHistoryViolation for writer faults (fork, forged frontier,
gap, unauthenticated record) and an Error for local or transient
failures, and only a violation quarantines; an Error aborts the session
for a retry on the next cycle.
The security property is unchanged: nothing is applied unless a hash
chain links durable history, itself verified against the writer's
authenticated commitments, to the frontier the writer authenticated in
this handshake. Overlapping replays change no state, and forks, forged
frontiers, gaps and unauthenticated records still quarantine, now also
when they overlap an advanced prefix (new tests cover each).
The pre- and post-exchange history checks get the same treatment. A
live prefix between the proven handshake prefix and the frozen frontier
is consistent, since delta exchange verifies the rest; one beyond the
frontier is compared with the durable commitment at the frontier rather
than rejected as a rollback.
The handshake also clamps the live local-writer window resolver to the
frozen snapshot, so concurrent local writes no longer fail the session
with "local writer-window resolver violated negotiated bounds".
Signed-off-by: trvon <me@trevon.dev>
* feat(p2p): report outbound sessions and durable writer quarantines
`yams p2p status` showed inbound session failures but not outbound
ones, and only a `quarantined` record count that mixes durable writer
quarantine with transient rejections (a record whose causal dependency
from a third writer has not arrived yet). An operator could not tell a
mesh that was retrying from one that had permanently cut off a writer.
Status now reports:
- outbound_sessions / outbound_failures and the last outbound failure
with its peer, counted for both explicit connects and reconnects.
Failed sessions in either direction are aborted and retried.
- quarantined_writers: writers durably quarantined by this node, read
from the replication checkpoint, so it survives restarts. Each
quarantine is also logged with its reason.
The fields travel through MemorySyncStatus, MemorySyncResponse and
protobuf fields 38-42 to the text and JSON status and to `yams daemon
status -d`, following the apply-health fields.
Signed-off-by: trvon <me@trevon.dev>
* fix(memory-sync): verify replayed writer counters against durable history
validateHistoryExtension skipped every staged operation at or below the
writer's durable prefix as a replay without looking at its record hash,
leaving fork detection to applyDeltas. applyDeltas recognizes a fork only
by an operation id it still remembers, and that memory is lost on restart
and bounded by maxTrackedIdentities. A different, internally valid record
for an already-applied counter therefore passed as a harmless replay
instead of being quarantined as the fork it is, whether the window only
replayed old counters or rode the fork along an honest extension.
Before skipping a replayed counter, compare its record hash with the
durable history entry committed at that counter (rebuilding the entry like
computeHistoryCommitmentAt when it is missing). A mismatch is a
WriterHistoryViolation, so the p2p session durably quarantines the writer;
an identical replay is still accepted as a no-op. A history entry that
cannot be read stays a local Error and only aborts the session.
Signed-off-by: trvon <me@trevon.dev>
* fix(memory-sync): log every durable writer quarantine where it happens
Durable writer quarantine was logged only by its two p2p callers. The
duplicate-operation fork path inside applyDeltas and the shared-store scan
path quarantined silently, so the documented guarantee that every durable
quarantine logs its reason and counters did not hold, and an operator saw
a quarantined_writers count with nothing in the log to explain it.
quarantineWriter now takes the reason and is the one place that logs a new
quarantine: writer, reason, source node, and the writer's local history and
version counters. It logs nothing when the writer was already quarantined,
so a repeated request is not reported twice. The p2p callers drop their own
warnings and fold their session context (frontier, staged count, prefix
match) into the reason instead. memory_sync now declares its spdlog
dependency explicitly.
Signed-off-by: trvon <me@trevon.dev>
* fix(cli): show outbound p2p sessions in daemon status
The daemon already reports outbound sessions, failures, and the last
outbound failure, and `yams p2p status` prints them, but `yams daemon
status -d` only gained the quarantined-writers row. An outbound direction
that keeps failing is aborted and retried, never quarantined, so the
detailed status was the place an operator would miss it.
The Memory Sync section now shows an Outbound sessions row (started and
failed counts, warning when any failed) for direct-backend nodes, and a
Last outbound failure row with the peer, the error, and its age.
Signed-off-by: trvon <me@trevon.dev>
---------
Signed-off-by: trvon <me@trevon.dev>
1103 lines
54 KiB
C++
1103 lines
54 KiB
C++
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
// Copyright 2026 YAMS Contributors
|
|
|
|
// pi-lens-ignore: fatal error
|
|
#include <catch2/catch_test_macros.hpp>
|
|
// pi-lens-ignore: fatal error
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include <yams/crypto/hasher.h>
|
|
#include <yams/daemon/p2p/p2p_delta.h>
|
|
#include <yams/memory_sync/writer_auth.h>
|
|
#include <yams/storage/storage_backend.h>
|
|
|
|
#include "../../../src/daemon/p2p/p2p_fuzz.h"
|
|
|
|
#include "common/p2p_test_helpers.h"
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <filesystem>
|
|
#include <functional>
|
|
#include <future>
|
|
#include <limits>
|
|
#include <map>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <span>
|
|
#include <string>
|
|
#include <string_view>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
using namespace std::chrono_literals;
|
|
using yams::daemon::p2p::DeltaExchangeOptions;
|
|
using yams::daemon::p2p::DeltaExchangeStats;
|
|
using yams::daemon::p2p::InMemoryPeerTrustStore;
|
|
using yams::daemon::p2p::P2pClientOptions;
|
|
using yams::daemon::p2p::P2pConnection;
|
|
using yams::daemon::p2p::P2pListener;
|
|
using yams::daemon::p2p::PeerHandshakeConfig;
|
|
using yams::daemon::p2p::TlsIdentity;
|
|
using yams::memory_sync::MemorySyncConfig;
|
|
using yams::memory_sync::MemorySyncService;
|
|
|
|
namespace {
|
|
using namespace yams::p2p_test;
|
|
|
|
yams::Result<void> validateDeltaJson(const nlohmann::json& json,
|
|
const DeltaExchangeOptions& options = {}) {
|
|
const auto encoded = json.dump();
|
|
return yams::daemon::p2p::detail::validateDeltaControlFrame(
|
|
std::span<const std::byte>(reinterpret_cast<const std::byte*>(encoded.data()),
|
|
encoded.size()),
|
|
options);
|
|
}
|
|
|
|
std::string digest(std::span<const std::byte> data) {
|
|
yams::crypto::SHA256Hasher hasher;
|
|
hasher.init();
|
|
hasher.update(data);
|
|
return hasher.finalize();
|
|
}
|
|
|
|
class InMemoryBackend final : public yams::storage::IStorageBackend {
|
|
public:
|
|
yams::Result<void> initialize(const yams::storage::BackendConfig&) override { return {}; }
|
|
|
|
yams::Result<void> store(std::string_view key, std::span<const std::byte> data) override {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
objects_[std::string(key)] = std::vector<std::byte>(data.begin(), data.end());
|
|
return {};
|
|
}
|
|
|
|
yams::Result<std::vector<std::byte>> retrieve(std::string_view key) const override {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
const auto found = objects_.find(std::string(key));
|
|
if (found == objects_.end()) {
|
|
return yams::Error{yams::ErrorCode::NotFound, "test object is absent"};
|
|
}
|
|
return found->second;
|
|
}
|
|
|
|
yams::Result<bool> exists(std::string_view key) const override {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
return objects_.contains(std::string(key));
|
|
}
|
|
|
|
yams::Result<void> remove(std::string_view key) override {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
objects_.erase(std::string(key));
|
|
return {};
|
|
}
|
|
|
|
yams::Result<std::vector<std::string>> list(std::string_view prefix = "") const override {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
std::vector<std::string> result;
|
|
for (const auto& [key, _] : objects_) {
|
|
if (key.starts_with(prefix)) {
|
|
result.push_back(key);
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
yams::Result<yams::StorageStats> getStats() const override { return yams::StorageStats{}; }
|
|
|
|
std::future<yams::Result<void>> storeAsync(std::string_view key,
|
|
std::span<const std::byte> data) override {
|
|
auto ownedKey = std::string(key);
|
|
auto ownedData = std::vector<std::byte>(data.begin(), data.end());
|
|
return std::async(std::launch::deferred,
|
|
[this, key = std::move(ownedKey), data = std::move(ownedData)] {
|
|
return store(key, data);
|
|
});
|
|
}
|
|
|
|
std::future<yams::Result<std::vector<std::byte>>>
|
|
retrieveAsync(std::string_view key) const override {
|
|
auto ownedKey = std::string(key);
|
|
return std::async(std::launch::deferred,
|
|
[this, key = std::move(ownedKey)] { return retrieve(key); });
|
|
}
|
|
|
|
std::string getType() const override { return "test-memory"; }
|
|
bool isRemote() const override { return false; }
|
|
yams::Result<void> flush() override { return {}; }
|
|
|
|
private:
|
|
mutable std::mutex mutex_;
|
|
std::map<std::string, std::vector<std::byte>> objects_;
|
|
};
|
|
|
|
std::unique_ptr<yams::storage::FilesystemBackend> backend(const std::filesystem::path& path) {
|
|
yams::storage::BackendConfig config;
|
|
config.type = "filesystem";
|
|
config.localPath = path;
|
|
auto result = std::make_unique<yams::storage::FilesystemBackend>();
|
|
REQUIRE(result->initialize(config).has_value());
|
|
return result;
|
|
}
|
|
|
|
std::shared_ptr<const yams::memory_sync::WriterAuthenticator>
|
|
writerAuth(std::string nodeId, std::string keyId, const yams::memory_sync::WriterKeyPair& keyPair,
|
|
std::vector<yams::memory_sync::TrustedWriterKey> trusted) {
|
|
yams::memory_sync::WriterAuthConfig config;
|
|
config.required = true;
|
|
config.localWriterId = std::move(nodeId);
|
|
config.localKeyId = std::move(keyId);
|
|
config.localPrivateKeyPem = keyPair.privateKeyPem;
|
|
config.trustedKeys = std::move(trusted);
|
|
auto result =
|
|
yams::memory_sync::WriterAuthenticator::create(std::move(config), "wire-delta-corpus", 1);
|
|
REQUIRE(result.has_value());
|
|
return result.value();
|
|
}
|
|
|
|
TlsIdentity identity(std::string nodeId, const yams::memory_sync::WriterKeyPair& keyPair) {
|
|
auto result = TlsIdentity::fromPrivateKeyPem(std::move(nodeId), keyPair.privateKeyPem);
|
|
REQUIRE(result.has_value());
|
|
return std::move(result.value());
|
|
}
|
|
|
|
PeerHandshakeConfig handshakeConfig(std::string nodeId, MemorySyncService& service,
|
|
bool allowFirstContact) {
|
|
const auto state = service.replicationState();
|
|
return PeerHandshakeConfig{
|
|
.nodeId = std::move(nodeId),
|
|
.corpusId = "wire-delta-corpus",
|
|
.corpusEpoch = 1,
|
|
.localVersion = state.version,
|
|
.localCommitments = state.commitments,
|
|
.localQuarantinedWriters = state.quarantinedWriters,
|
|
.resolveLocalCommitment =
|
|
[&service](std::uint64_t counter) { return service.localHistoryCommitmentAt(counter); },
|
|
.resolveLocalWindow =
|
|
[&service](std::uint64_t peerCounter, std::size_t maxRecords,
|
|
std::size_t maxWireBytes) {
|
|
return service.localHistoryWindowAfter(peerCounter, maxRecords, maxWireBytes);
|
|
},
|
|
.allowFirstContact = allowFirstContact,
|
|
.timeout = 3s};
|
|
}
|
|
|
|
struct ExchangePair {
|
|
yams::Result<DeltaExchangeStats> client;
|
|
yams::Result<DeltaExchangeStats> server;
|
|
};
|
|
|
|
ExchangePair runExchange(MemorySyncService& clientService, MemorySyncService& serverService,
|
|
const yams::memory_sync::WriterKeyPair& clientKey,
|
|
const yams::memory_sync::WriterKeyPair& serverKey,
|
|
InMemoryPeerTrustStore& clientTrust, InMemoryPeerTrustStore& serverTrust,
|
|
bool allowFirstContact,
|
|
DeltaExchangeOptions exchangeOptions = {.maxDeltasPerBatch = 2,
|
|
.maxBatches = 16,
|
|
.timeout = 3s},
|
|
std::function<void(PeerHandshakeConfig&)> mutateServerHandshake = {}) {
|
|
auto clientIdentity = identity("client-node", clientKey);
|
|
auto serverIdentity = identity("server-node", serverKey);
|
|
P2pListener listener(P2pListener::Options{.host = "127.0.0.1",
|
|
.port = 0,
|
|
.identity = std::move(serverIdentity),
|
|
.allowedPeerPins = {},
|
|
.allowUnpinnedPeers = true,
|
|
.handshakeTimeout = 3s,
|
|
.maxConcurrentSessions = 1});
|
|
std::promise<yams::Result<DeltaExchangeStats>> serverDone;
|
|
auto serverFuture = serverDone.get_future();
|
|
const auto batchCapacity = exchangeOptions.maxBatches > exchangeOptions.maxDeltasPerSession /
|
|
exchangeOptions.maxDeltasPerBatch
|
|
? exchangeOptions.maxDeltasPerSession
|
|
: exchangeOptions.maxBatches * exchangeOptions.maxDeltasPerBatch;
|
|
const auto writerAdvance = std::min({exchangeOptions.maxDeltasPerSession, batchCapacity,
|
|
yams::daemon::p2p::kMaxP2pWriterAdvance});
|
|
auto serverHandshake = handshakeConfig("server-node", serverService, allowFirstContact);
|
|
serverHandshake.maxWriterAdvance = writerAdvance;
|
|
serverHandshake.maxWriterWindowBytes = exchangeOptions.maxWireBytesPerSession;
|
|
if (mutateServerHandshake) {
|
|
mutateServerHandshake(serverHandshake);
|
|
}
|
|
listener.setHandler([&](P2pConnection channel) {
|
|
auto handshake =
|
|
yams::daemon::p2p::acceptPeerHandshake(channel, serverHandshake, serverTrust);
|
|
if (!handshake) {
|
|
serverDone.set_value(handshake.error());
|
|
return;
|
|
}
|
|
serverDone.set_value(yams::daemon::p2p::acceptDeltaExchange(
|
|
channel, serverService, handshake.value(), exchangeOptions));
|
|
});
|
|
REQUIRE(listener.start().has_value());
|
|
|
|
auto channel =
|
|
yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1",
|
|
.port = listener.boundPort(),
|
|
.identity = std::move(clientIdentity),
|
|
.expectedPeerPin = {},
|
|
.allowUnpinnedPeer = true,
|
|
.connectTimeout = 3s,
|
|
.handshakeTimeout = 3s});
|
|
REQUIRE(channel.has_value());
|
|
auto clientConfig = handshakeConfig("client-node", clientService, allowFirstContact);
|
|
clientConfig.maxWriterAdvance = writerAdvance;
|
|
clientConfig.maxWriterWindowBytes = exchangeOptions.maxWireBytesPerSession;
|
|
auto clientHandshake =
|
|
yams::daemon::p2p::initiatePeerHandshake(channel.value(), clientConfig, clientTrust);
|
|
REQUIRE(clientHandshake.has_value());
|
|
auto clientResult = yams::daemon::p2p::initiateDeltaExchange(
|
|
channel.value(), clientService, clientHandshake.value(), exchangeOptions);
|
|
REQUIRE(serverFuture.wait_for(5s) == std::future_status::ready);
|
|
auto serverResult = serverFuture.get();
|
|
listener.stop();
|
|
return ExchangePair{.client = std::move(clientResult), .server = std::move(serverResult)};
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("direct P2P delta fuzz seam enforces bounded control messages",
|
|
"[daemon][p2p][delta][fuzz]") {
|
|
CHECK(validateDeltaJson(
|
|
nlohmann::json{{"type", "delta_batch"}, {"count", 1}, {"has_more", false}})
|
|
.has_value());
|
|
CHECK(validateDeltaJson(nlohmann::json{{"type", "replication_mode"}, {"mode", "delta"}})
|
|
.has_value());
|
|
CHECK_FALSE(
|
|
validateDeltaJson(nlohmann::json{{"type", "delta_batch"}, {"count", 0}, {"has_more", true}})
|
|
.has_value());
|
|
CHECK_FALSE(
|
|
validateDeltaJson(nlohmann::json{{"type", "delta_record"}, {"payload_size", 1ULL << 40}})
|
|
.has_value());
|
|
nlohmann::json snapshot{{"type", "snapshot_begin"},
|
|
{"witness", "node-a"},
|
|
{"frontier", {{"counters_", nlohmann::json::object()}}},
|
|
{"commitments", nlohmann::json::array()},
|
|
{"record_count", 0},
|
|
{"payload_bytes", 0},
|
|
{"root_digest", std::string(64, '0')},
|
|
{"witness_key_id", "node-a-v1"},
|
|
{"witness_algorithm", "Ed25519"},
|
|
{"witness_signature", "AA=="}};
|
|
CHECK(validateDeltaJson(snapshot).has_value());
|
|
snapshot["record_count"] = DeltaExchangeOptions{}.maxSnapshotRecords + 1;
|
|
CHECK_FALSE(validateDeltaJson(snapshot).has_value());
|
|
CHECK_FALSE(validateDeltaJson(nlohmann::json{{"type", "unknown"}}).has_value());
|
|
}
|
|
|
|
TEST_CASE("direct P2P cold bootstrap installs an authenticated multiwriter snapshot",
|
|
"[daemon][p2p][delta][cold-bootstrap][network][security]") {
|
|
TempDir clientDir{"cold-client"};
|
|
TempDir serverDir{"cold-server"};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto foreignKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
REQUIRE(foreignKey.has_value());
|
|
const std::vector<yams::memory_sync::TrustedWriterKey> trust = {
|
|
{"client-node", "client-v1", clientKey.value().publicKeyPem, false},
|
|
{"server-node", "server-v1", serverKey.value().publicKeyPem, false},
|
|
{"foreign-node", "foreign-v1", foreignKey.value().publicKeyPem, false},
|
|
};
|
|
|
|
auto foreignBackend = backend(serverDir.path);
|
|
yams::memory_sync::MemorySyncLoop foreign{
|
|
*foreignBackend,
|
|
"foreign-node",
|
|
"wire-delta-corpus",
|
|
1,
|
|
false,
|
|
{},
|
|
{},
|
|
{},
|
|
writerAuth("foreign-node", "foreign-v1", foreignKey.value(), trust)};
|
|
REQUIRE(foreign.publish("user/foreign", bytes("foreign-value")).has_value());
|
|
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path),
|
|
MemorySyncConfig{"server-node",
|
|
60'000,
|
|
"wire-delta-corpus",
|
|
1,
|
|
{},
|
|
false,
|
|
writerAuth("server-node", "server-v1", serverKey.value(), trust)}};
|
|
REQUIRE(serverService.syncFully().has_value());
|
|
REQUIRE(serverService.publish("user/server", bytes("server-value")).has_value());
|
|
REQUIRE(serverService.syncFully().has_value());
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path),
|
|
MemorySyncConfig{"client-node",
|
|
60'000,
|
|
"wire-delta-corpus",
|
|
1,
|
|
{},
|
|
false,
|
|
writerAuth("client-node", "client-v1", clientKey.value(), trust)}};
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
const auto refused = runExchange(clientService, serverService, clientKey.value(),
|
|
serverKey.value(), clientTrust, serverTrust, true);
|
|
CHECK_FALSE((refused.client.has_value() && refused.server.has_value()));
|
|
CHECK(clientService.currentVersion().empty());
|
|
|
|
// A prior TOFU session remains legacy first-contact trust and must never become operator
|
|
// enrollment merely because a later handshake recognizes the same pin.
|
|
const auto exchange = runExchange(clientService, serverService, clientKey.value(),
|
|
serverKey.value(), clientTrust, serverTrust, false);
|
|
CHECK_FALSE((exchange.client.has_value() && exchange.server.has_value()));
|
|
CHECK(clientService.currentVersion().empty());
|
|
|
|
const auto serverPin = identity("server-node", serverKey.value()).spkiPin();
|
|
const auto clientPin = identity("client-node", clientKey.value()).spkiPin();
|
|
REQUIRE(clientTrust.enrollOperatorPeer("server-node", serverPin).has_value());
|
|
REQUIRE(serverTrust.enrollOperatorPeer("client-node", clientPin).has_value());
|
|
const auto enrolled = runExchange(clientService, serverService, clientKey.value(),
|
|
serverKey.value(), clientTrust, serverTrust, false);
|
|
REQUIRE(enrolled.client.has_value());
|
|
REQUIRE(enrolled.server.has_value());
|
|
CHECK(enrolled.client.value().snapshotsReceived == 1);
|
|
CHECK(enrolled.server.value().snapshotsSent == 1);
|
|
REQUIRE(clientService.readCached("user/foreign").has_value());
|
|
REQUIRE(clientService.readCached("user/server").has_value());
|
|
CHECK(text(clientService.readCached("user/foreign").value()) == "foreign-value");
|
|
CHECK(text(clientService.readCached("user/server").value()) == "server-value");
|
|
CHECK(clientService.currentVersion().counters() == serverService.currentVersion().counters());
|
|
|
|
REQUIRE(serverService.publish("user/continuation", bytes("after-snapshot")).has_value());
|
|
const auto continued = runExchange(clientService, serverService, clientKey.value(),
|
|
serverKey.value(), clientTrust, serverTrust, false);
|
|
REQUIRE(continued.client.has_value());
|
|
REQUIRE(continued.server.has_value());
|
|
CHECK(continued.client.value().snapshotsReceived == 0);
|
|
REQUIRE(clientService.readCached("user/continuation").has_value());
|
|
CHECK(text(clientService.readCached("user/continuation").value()) == "after-snapshot");
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta exchange rejects origin not bound to TLS peer",
|
|
"[daemon][p2p][delta][network][security]") {
|
|
TempDir clientDir{"forged-client"};
|
|
TempDir serverDir{"forged-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
auto clientIdentity = identity("client-node", clientKey.value());
|
|
auto serverIdentity = identity("server-node", serverKey.value());
|
|
const std::string serverPin = serverIdentity.spkiPin();
|
|
const std::string clientPin = clientIdentity.spkiPin();
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
std::promise<yams::Result<DeltaExchangeStats>> serverDone;
|
|
auto serverFuture = serverDone.get_future();
|
|
P2pListener listener(P2pListener::Options{.host = "127.0.0.1",
|
|
.port = 0,
|
|
.identity = std::move(serverIdentity),
|
|
.allowedPeerPins = {clientPin},
|
|
.allowUnpinnedPeers = false,
|
|
.handshakeTimeout = 3s,
|
|
.maxConcurrentSessions = 1});
|
|
listener.setHandler([&](P2pConnection channel) {
|
|
auto handshake = yams::daemon::p2p::acceptPeerHandshake(
|
|
channel, handshakeConfig("server-node", serverService, true), serverTrust);
|
|
if (!handshake) {
|
|
serverDone.set_value(handshake.error());
|
|
return;
|
|
}
|
|
serverDone.set_value(yams::daemon::p2p::acceptDeltaExchange(
|
|
channel, serverService, handshake.value(),
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 2, .maxBatches = 4, .timeout = 3s}));
|
|
});
|
|
REQUIRE(listener.start().has_value());
|
|
|
|
auto channel =
|
|
yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1",
|
|
.port = listener.boundPort(),
|
|
.identity = std::move(clientIdentity),
|
|
.expectedPeerPin = serverPin,
|
|
.allowUnpinnedPeer = false,
|
|
.connectTimeout = 3s,
|
|
.handshakeTimeout = 3s});
|
|
REQUIRE(channel.has_value());
|
|
auto handshake = yams::daemon::p2p::initiatePeerHandshake(
|
|
channel.value(), handshakeConfig("client-node", clientService, true), clientTrust);
|
|
REQUIRE(handshake.has_value());
|
|
|
|
const auto payload = bytes("forged");
|
|
yams::memory_sync::MemoryIndexRecord record;
|
|
record.entryHash = digest(payload);
|
|
record.origin = "other-writer";
|
|
REQUIRE(record.vv.increment(record.origin));
|
|
record.corpusId = "wire-delta-corpus";
|
|
record.corpusEpoch = 1;
|
|
record.operationId = yams::memory_sync::makeOperationId(record.origin, 1);
|
|
record.logicalKey = "user/forged";
|
|
record.recordKind = std::string(yams::memory_sync::kMemoryValueRecordKind);
|
|
const nlohmann::json mode{{"type", "replication_mode"}, {"mode", "delta"}};
|
|
const nlohmann::json batch{{"type", "delta_batch"}, {"count", 1}, {"has_more", false}};
|
|
const nlohmann::json wireRecord{{"type", "delta_record"},
|
|
{"logical_key", record.logicalKey},
|
|
{"record", record},
|
|
{"payload_size", payload.size()}};
|
|
for (const auto& json : {mode, batch, wireRecord}) {
|
|
const std::string encoded = json.dump();
|
|
REQUIRE(
|
|
channel.value()
|
|
.writeFrame(std::span<const std::byte>(
|
|
reinterpret_cast<const std::byte*>(encoded.data()), encoded.size()),
|
|
3s)
|
|
.has_value());
|
|
}
|
|
REQUIRE(channel.value()
|
|
.writeFrame(payload, 3s, yams::daemon::p2p::kP2pMaxValuePayloadBytes)
|
|
.has_value());
|
|
|
|
REQUIRE(serverFuture.wait_for(5s) == std::future_status::ready);
|
|
auto serverResult = serverFuture.get();
|
|
REQUIRE_FALSE(serverResult.has_value());
|
|
CHECK(serverResult.error().code == yams::ErrorCode::Unauthorized);
|
|
CHECK(serverService.currentVersion().empty());
|
|
CHECK_FALSE(serverService.readCached("user/forged").has_value());
|
|
listener.stop();
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta receiver rejects aggregate bytes before reading payload",
|
|
"[daemon][p2p][delta][network][security]") {
|
|
TempDir clientDir{"byte-cap-client"};
|
|
TempDir serverDir{"byte-cap-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
auto clientIdentity = identity("client-node", clientKey.value());
|
|
auto serverIdentity = identity("server-node", serverKey.value());
|
|
const std::string serverPin = serverIdentity.spkiPin();
|
|
const std::string clientPin = clientIdentity.spkiPin();
|
|
|
|
const auto payload = bytes(std::string(1024, 'x'));
|
|
yams::memory_sync::MemoryIndexRecord record;
|
|
record.entryHash = digest(payload);
|
|
record.origin = "client-node";
|
|
REQUIRE(record.vv.increment(record.origin));
|
|
record.corpusId = "wire-delta-corpus";
|
|
record.corpusEpoch = 1;
|
|
record.operationId = yams::memory_sync::makeOperationId(record.origin, 1);
|
|
record.logicalKey = "user/oversized-session";
|
|
record.recordKind = std::string(yams::memory_sync::kMemoryValueRecordKind);
|
|
const nlohmann::json mode{{"type", "replication_mode"}, {"mode", "delta"}};
|
|
const nlohmann::json batch{{"type", "delta_batch"}, {"count", 1}, {"has_more", false}};
|
|
const nlohmann::json wireRecord{{"type", "delta_record"},
|
|
{"logical_key", record.logicalKey},
|
|
{"record", record},
|
|
{"payload_size", payload.size()}};
|
|
const auto byteLimit = batch.dump().size() + 4 + wireRecord.dump().size() + 4 + 100;
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
std::promise<yams::Result<DeltaExchangeStats>> serverDone;
|
|
auto serverFuture = serverDone.get_future();
|
|
P2pListener listener(P2pListener::Options{.host = "127.0.0.1",
|
|
.port = 0,
|
|
.identity = std::move(serverIdentity),
|
|
.allowedPeerPins = {clientPin},
|
|
.allowUnpinnedPeers = false,
|
|
.handshakeTimeout = 3s,
|
|
.maxConcurrentSessions = 1});
|
|
listener.setHandler([&](P2pConnection channel) {
|
|
auto handshake = yams::daemon::p2p::acceptPeerHandshake(
|
|
channel, handshakeConfig("server-node", serverService, true), serverTrust);
|
|
if (!handshake) {
|
|
serverDone.set_value(handshake.error());
|
|
return;
|
|
}
|
|
serverDone.set_value(yams::daemon::p2p::acceptDeltaExchange(
|
|
channel, serverService, handshake.value(),
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 2,
|
|
.maxBatches = 2,
|
|
.maxDeltasPerSession = 2,
|
|
.maxWireBytesPerSession = byteLimit,
|
|
.timeout = 3s}));
|
|
});
|
|
REQUIRE(listener.start().has_value());
|
|
|
|
auto channel =
|
|
yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1",
|
|
.port = listener.boundPort(),
|
|
.identity = std::move(clientIdentity),
|
|
.expectedPeerPin = serverPin,
|
|
.allowUnpinnedPeer = false,
|
|
.connectTimeout = 3s,
|
|
.handshakeTimeout = 3s});
|
|
REQUIRE(channel.has_value());
|
|
auto handshake = yams::daemon::p2p::initiatePeerHandshake(
|
|
channel.value(), handshakeConfig("client-node", clientService, true), clientTrust);
|
|
REQUIRE(handshake.has_value());
|
|
for (const auto& json : {mode, batch, wireRecord}) {
|
|
const std::string encoded = json.dump();
|
|
REQUIRE(
|
|
channel.value()
|
|
.writeFrame(std::span<const std::byte>(
|
|
reinterpret_cast<const std::byte*>(encoded.data()), encoded.size()),
|
|
3s)
|
|
.has_value());
|
|
}
|
|
|
|
REQUIRE(serverFuture.wait_for(1s) == std::future_status::ready);
|
|
auto serverResult = serverFuture.get();
|
|
REQUIRE_FALSE(serverResult.has_value());
|
|
CHECK(serverResult.error().code == yams::ErrorCode::ResourceExhausted);
|
|
CHECK(serverService.currentVersion().empty());
|
|
listener.stop();
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta receiver counts actual wire bytes including JSON padding",
|
|
"[daemon][p2p][delta][network][security]") {
|
|
TempDir clientDir{"padded-wire-client"};
|
|
TempDir serverDir{"padded-wire-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
auto clientIdentity = identity("client-node", clientKey.value());
|
|
auto serverIdentity = identity("server-node", serverKey.value());
|
|
const std::string serverPin = serverIdentity.spkiPin();
|
|
const std::string clientPin = clientIdentity.spkiPin();
|
|
|
|
const auto payload = bytes(std::string(64, 'x'));
|
|
yams::memory_sync::MemoryIndexRecord record;
|
|
record.entryHash = digest(payload);
|
|
record.origin = "client-node";
|
|
REQUIRE(record.vv.increment(record.origin));
|
|
record.corpusId = "wire-delta-corpus";
|
|
record.corpusEpoch = 1;
|
|
record.operationId = yams::memory_sync::makeOperationId(record.origin, 1);
|
|
record.logicalKey = "user/padded-wire";
|
|
record.recordKind = std::string(yams::memory_sync::kMemoryValueRecordKind);
|
|
const nlohmann::json mode{{"type", "replication_mode"}, {"mode", "delta"}};
|
|
const nlohmann::json batch{{"type", "delta_batch"}, {"count", 1}, {"has_more", false}};
|
|
const nlohmann::json wireRecord{{"type", "delta_record"},
|
|
{"logical_key", record.logicalKey},
|
|
{"record", record},
|
|
{"payload_size", payload.size()}};
|
|
std::string paddedRecord = wireRecord.dump();
|
|
paddedRecord.insert(paddedRecord.size() - 1, std::string(2048, ' '));
|
|
|
|
// Compact control frames fit this budget; the whitespace-padded record must not.
|
|
const auto byteLimit = batch.dump().size() + 4 + wireRecord.dump().size() + 4 + 256;
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
std::promise<yams::Result<DeltaExchangeStats>> serverDone;
|
|
auto serverFuture = serverDone.get_future();
|
|
P2pListener listener(P2pListener::Options{.host = "127.0.0.1",
|
|
.port = 0,
|
|
.identity = std::move(serverIdentity),
|
|
.allowedPeerPins = {clientPin},
|
|
.allowUnpinnedPeers = false,
|
|
.handshakeTimeout = 3s,
|
|
.maxConcurrentSessions = 1});
|
|
listener.setHandler([&](P2pConnection channel) {
|
|
auto handshake = yams::daemon::p2p::acceptPeerHandshake(
|
|
channel, handshakeConfig("server-node", serverService, true), serverTrust);
|
|
if (!handshake) {
|
|
serverDone.set_value(handshake.error());
|
|
return;
|
|
}
|
|
serverDone.set_value(yams::daemon::p2p::acceptDeltaExchange(
|
|
channel, serverService, handshake.value(),
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 2,
|
|
.maxBatches = 2,
|
|
.maxDeltasPerSession = 2,
|
|
.maxWireBytesPerSession = byteLimit,
|
|
.timeout = 3s}));
|
|
});
|
|
REQUIRE(listener.start().has_value());
|
|
|
|
auto channel =
|
|
yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1",
|
|
.port = listener.boundPort(),
|
|
.identity = std::move(clientIdentity),
|
|
.expectedPeerPin = serverPin,
|
|
.allowUnpinnedPeer = false,
|
|
.connectTimeout = 3s,
|
|
.handshakeTimeout = 3s});
|
|
REQUIRE(channel.has_value());
|
|
auto handshake = yams::daemon::p2p::initiatePeerHandshake(
|
|
channel.value(), handshakeConfig("client-node", clientService, true), clientTrust);
|
|
REQUIRE(handshake.has_value());
|
|
for (const auto& json : {mode, batch}) {
|
|
const std::string encoded = json.dump();
|
|
REQUIRE(
|
|
channel.value()
|
|
.writeFrame(std::span<const std::byte>(
|
|
reinterpret_cast<const std::byte*>(encoded.data()), encoded.size()),
|
|
3s)
|
|
.has_value());
|
|
}
|
|
REQUIRE(channel.value()
|
|
.writeFrame(std::span<const std::byte>(
|
|
reinterpret_cast<const std::byte*>(paddedRecord.data()),
|
|
paddedRecord.size()),
|
|
3s)
|
|
.has_value());
|
|
|
|
REQUIRE(serverFuture.wait_for(1s) == std::future_status::ready);
|
|
auto serverResult = serverFuture.get();
|
|
REQUIRE_FALSE(serverResult.has_value());
|
|
CHECK(serverResult.error().code == yams::ErrorCode::ResourceExhausted);
|
|
CHECK(serverService.currentVersion().empty());
|
|
listener.stop();
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta exchange rejects options above protocol hard bounds",
|
|
"[daemon][p2p][delta][network][limits]") {
|
|
TempDir clientDir{"invalid-limits-client"};
|
|
TempDir serverDir{"invalid-limits-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
auto exchanged = runExchange(
|
|
clientService, serverService, clientKey.value(), serverKey.value(), clientTrust,
|
|
serverTrust, true,
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 1,
|
|
.maxBatches = 1,
|
|
.maxDeltasPerSession = std::numeric_limits<std::size_t>::max(),
|
|
.maxWireBytesPerSession = 1024,
|
|
.timeout = 3s});
|
|
REQUIRE_FALSE(exchanged.client.has_value());
|
|
CHECK(exchanged.client.error().code == yams::ErrorCode::InvalidArgument);
|
|
|
|
auto overTimeout = runExchange(
|
|
clientService, serverService, clientKey.value(), serverKey.value(), clientTrust,
|
|
serverTrust, true,
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 1,
|
|
.maxBatches = 1,
|
|
.maxDeltasPerSession = 1,
|
|
.maxWireBytesPerSession = 1024,
|
|
.timeout = yams::daemon::p2p::kP2pMaxOperationTimeout + 1ms});
|
|
REQUIRE_FALSE(overTimeout.client.has_value());
|
|
CHECK(overTimeout.client.error().code == yams::ErrorCode::InvalidArgument);
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta exchange resumes across aggregate session limits",
|
|
"[daemon][p2p][delta][network][security][continuation]") {
|
|
TempDir clientDir{"bounded-client"};
|
|
TempDir serverDir{"bounded-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
REQUIRE(clientService.publish("user/one", bytes("one")).has_value());
|
|
REQUIRE(clientService.publish("user/two", bytes("two")).has_value());
|
|
REQUIRE(clientService.publish("user/three", bytes("three")).has_value());
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
const DeltaExchangeOptions options{.maxDeltasPerBatch = 2,
|
|
.maxBatches = 2,
|
|
.maxDeltasPerSession = 1,
|
|
.maxWireBytesPerSession = 2048,
|
|
.timeout = 3s};
|
|
|
|
for (std::uint64_t expectedCounter = 1; expectedCounter <= 3; ++expectedCounter) {
|
|
auto exchange =
|
|
runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, expectedCounter == 1, options);
|
|
REQUIRE(exchange.client.has_value());
|
|
REQUIRE(exchange.server.has_value());
|
|
CHECK(exchange.client.value().deltasSent == 1);
|
|
CHECK(exchange.server.value().deltasReceived == 1);
|
|
CHECK(serverService.currentVersion().get("client-node") == expectedCounter);
|
|
}
|
|
|
|
REQUIRE(serverService.readCached("user/one").has_value());
|
|
REQUIRE(serverService.readCached("user/two").has_value());
|
|
REQUIRE(serverService.readCached("user/three").has_value());
|
|
auto converged = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, false, options);
|
|
REQUIRE(converged.client.has_value());
|
|
REQUIRE(converged.server.has_value());
|
|
CHECK(converged.client.value().deltasSent == 0);
|
|
CHECK(converged.server.value().deltasReceived == 0);
|
|
}
|
|
|
|
TEST_CASE("direct P2P catch-up advances beyond 1024 retained operations",
|
|
"[daemon][p2p][delta][network][continuation][large-history][.slow]") {
|
|
MemorySyncService clientService{
|
|
std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
for (int index = 0; index < 1025; ++index) {
|
|
REQUIRE(
|
|
clientService
|
|
.publish("bulk/" + std::to_string(index), bytes("value-" + std::to_string(index)))
|
|
.has_value());
|
|
}
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
const DeltaExchangeOptions options{.maxDeltasPerBatch = 128,
|
|
.maxBatches = 8,
|
|
.maxDeltasPerSession = 1024,
|
|
.maxWireBytesPerSession = 8 * 1024 * 1024,
|
|
.timeout = 10s};
|
|
auto first = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, true, options);
|
|
REQUIRE(first.client.has_value());
|
|
REQUIRE(first.server.has_value());
|
|
CHECK(first.client.value().deltasSent == 1024);
|
|
CHECK(serverService.currentVersion().get("client-node") == 1024);
|
|
CHECK_FALSE(serverService.readCached("bulk/1024").has_value());
|
|
|
|
auto resumed = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, false, options);
|
|
REQUIRE(resumed.client.has_value());
|
|
REQUIRE(resumed.server.has_value());
|
|
CHECK(resumed.client.value().deltasSent == 1);
|
|
CHECK(serverService.currentVersion().get("client-node") == 1025);
|
|
REQUIRE(serverService.readCached("bulk/1024").has_value());
|
|
}
|
|
|
|
TEST_CASE("direct P2P bounded continuation survives receiver restart",
|
|
"[daemon][p2p][delta][network][continuation][restart]") {
|
|
TempDir clientDir{"restart-client"};
|
|
TempDir serverDir{"restart-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto serverService = std::make_unique<MemorySyncService>(
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1});
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
REQUIRE(clientService.publish("user/one", bytes("one")).has_value());
|
|
REQUIRE(clientService.publish("user/two", bytes("two")).has_value());
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
const DeltaExchangeOptions options{.maxDeltasPerBatch = 1,
|
|
.maxBatches = 1,
|
|
.maxDeltasPerSession = 1,
|
|
.maxWireBytesPerSession = 2048,
|
|
.timeout = 3s};
|
|
auto first = runExchange(clientService, *serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, true, options);
|
|
REQUIRE(first.client.has_value());
|
|
REQUIRE(first.server.has_value());
|
|
CHECK(serverService->currentVersion().get("client-node") == 1);
|
|
|
|
serverService.reset();
|
|
serverService = std::make_unique<MemorySyncService>(
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1});
|
|
REQUIRE(serverService->syncFully().has_value());
|
|
auto resumed = runExchange(clientService, *serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, false, options);
|
|
REQUIRE(resumed.client.has_value());
|
|
REQUIRE(resumed.server.has_value());
|
|
CHECK(serverService->currentVersion().get("client-node") == 2);
|
|
REQUIRE(serverService->readCached("user/one").has_value());
|
|
REQUIRE(serverService->readCached("user/two").has_value());
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta stages a false advertised frontier before visibility",
|
|
"[daemon][p2p][delta][network][commitment][security]") {
|
|
TempDir clientDir{"false-frontier-client"};
|
|
TempDir serverDir{"false-frontier-server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
REQUIRE(serverService.publish("user/unverified", bytes("must-stay-staged")).has_value());
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
auto exchanged =
|
|
runExchange(clientService, serverService, clientKey.value(), serverKey.value(), clientTrust,
|
|
serverTrust, true,
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 2, .maxBatches = 16, .timeout = 3s},
|
|
[](PeerHandshakeConfig& handshake) {
|
|
handshake.localCommitments.at("server-node").digest = std::string(64, 'f');
|
|
});
|
|
REQUIRE_FALSE(exchanged.client.has_value());
|
|
REQUIRE_FALSE(exchanged.server.has_value());
|
|
CHECK(clientService.replicationState().quarantinedWriters.contains("server-node"));
|
|
CHECK_FALSE(clientService.readCached("user/unverified").has_value());
|
|
}
|
|
|
|
TEST_CASE("direct P2P delta exchange converges offline corpora and resumes from watermarks",
|
|
"[daemon][p2p][delta][network]") {
|
|
TempDir clientDir{"client"};
|
|
TempDir serverDir{"server"};
|
|
MemorySyncService clientService{
|
|
backend(clientDir.path), MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService serverService{
|
|
backend(serverDir.path), MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
auto clientKey = yams::memory_sync::generateWriterKeyPair();
|
|
auto serverKey = yams::memory_sync::generateWriterKeyPair();
|
|
REQUIRE(clientKey.has_value());
|
|
REQUIRE(serverKey.has_value());
|
|
|
|
for (int index = 0; index < 5; ++index) {
|
|
REQUIRE(clientService
|
|
.publish("user/client-" + std::to_string(index),
|
|
bytes("client-value-" + std::to_string(index)))
|
|
.has_value());
|
|
}
|
|
for (int index = 0; index < 3; ++index) {
|
|
REQUIRE(serverService
|
|
.publish("user/server-" + std::to_string(index),
|
|
bytes("server-value-" + std::to_string(index)))
|
|
.has_value());
|
|
}
|
|
|
|
InMemoryPeerTrustStore clientTrust;
|
|
InMemoryPeerTrustStore serverTrust;
|
|
auto first = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, true);
|
|
REQUIRE(first.client.has_value());
|
|
REQUIRE(first.server.has_value());
|
|
CHECK(first.client.value().deltasSent == 5);
|
|
CHECK(first.client.value().deltasReceived == 3);
|
|
CHECK(first.server.value().deltasReceived == 5);
|
|
CHECK(first.server.value().deltasSent == 3);
|
|
CHECK(first.client.value().batchesSent == 3);
|
|
CHECK(first.server.value().batchesSent == 2);
|
|
|
|
for (int index = 0; index < 5; ++index) {
|
|
auto value = serverService.readCached("user/client-" + std::to_string(index));
|
|
REQUIRE(value.has_value());
|
|
CHECK(text(value.value()) == "client-value-" + std::to_string(index));
|
|
}
|
|
for (int index = 0; index < 3; ++index) {
|
|
auto value = clientService.readCached("user/server-" + std::to_string(index));
|
|
REQUIRE(value.has_value());
|
|
CHECK(text(value.value()) == "server-value-" + std::to_string(index));
|
|
}
|
|
|
|
auto replay = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, false);
|
|
REQUIRE(replay.client.has_value());
|
|
REQUIRE(replay.server.has_value());
|
|
CHECK(replay.client.value().deltasSent == 0);
|
|
CHECK(replay.client.value().deltasReceived == 0);
|
|
CHECK(replay.server.value().deltasSent == 0);
|
|
CHECK(replay.server.value().deltasReceived == 0);
|
|
|
|
REQUIRE(clientService.erase("user/client-0", "client-0").has_value());
|
|
auto tombstone = runExchange(clientService, serverService, clientKey.value(), serverKey.value(),
|
|
clientTrust, serverTrust, false);
|
|
REQUIRE(tombstone.client.has_value());
|
|
REQUIRE(tombstone.server.has_value());
|
|
CHECK(tombstone.client.value().deltasSent == 1);
|
|
CHECK_FALSE(serverService.readCached("user/client-0").has_value());
|
|
}
|
|
|
|
namespace {
|
|
|
|
// Concurrent sessions with one peer (this node's outbound session and the peer's inbound one)
|
|
// each freeze a writer window at handshake and apply it later. These helpers replay the
|
|
// receiver half of such a window through the production receive path without a socket.
|
|
std::vector<std::byte> windowFrames(std::span<const yams::memory_sync::MemoryDelta> deltas) {
|
|
std::vector<std::byte> frames;
|
|
const auto appendJson = [&frames](const nlohmann::json& json) {
|
|
const auto encoded = json.dump();
|
|
yams::daemon::p2p::detail::appendFrame(
|
|
frames, std::span<const std::byte>(reinterpret_cast<const std::byte*>(encoded.data()),
|
|
encoded.size()));
|
|
};
|
|
appendJson({{"type", "delta_batch"}, {"count", deltas.size()}, {"has_more", false}});
|
|
for (const auto& delta : deltas) {
|
|
appendJson({{"type", "delta_record"},
|
|
{"logical_key", delta.logicalKey},
|
|
{"record", delta.record},
|
|
{"payload_size", delta.payload.size()}});
|
|
if (!delta.payload.empty()) {
|
|
yams::daemon::p2p::detail::appendFrame(frames, delta.payload);
|
|
}
|
|
}
|
|
return frames;
|
|
}
|
|
|
|
yams::daemon::p2p::PeerHandshakeResult
|
|
frozenWindow(const std::string& writer, const yams::memory_sync::WriterHistoryCommitment& frontier,
|
|
std::uint64_t verifiedPrefix) {
|
|
yams::daemon::p2p::PeerHandshakeResult handshake;
|
|
handshake.peerNodeId = writer;
|
|
handshake.peerVersion.observe(writer, frontier.counter);
|
|
handshake.peerSeen = handshake.peerVersion.counters();
|
|
if (frontier.counter != 0) {
|
|
handshake.peerCommitments[writer] = frontier;
|
|
}
|
|
handshake.peerPrefixCounter = verifiedPrefix;
|
|
handshake.peerPrefixVerified = true;
|
|
handshake.peerPrefixMatches = true;
|
|
return handshake;
|
|
}
|
|
|
|
yams::Result<DeltaExchangeStats>
|
|
receiveWindow(MemorySyncService& receiver, std::span<const yams::memory_sync::MemoryDelta> deltas,
|
|
const yams::daemon::p2p::PeerHandshakeResult& handshake) {
|
|
const auto frames = windowFrames(deltas);
|
|
yams::daemon::p2p::detail::BufferFrameSource source(frames);
|
|
return yams::daemon::p2p::detail::receiveAllDeltas(
|
|
source, "server-node", receiver, handshake,
|
|
DeltaExchangeOptions{.maxDeltasPerBatch = 16, .maxBatches = 4, .timeout = 3s});
|
|
}
|
|
|
|
yams::memory_sync::WriterHistoryCommitment commitmentAt(MemorySyncService& writer,
|
|
std::uint64_t counter) {
|
|
auto commitment = writer.localHistoryCommitmentAt(counter);
|
|
REQUIRE(commitment.has_value());
|
|
return commitment.value();
|
|
}
|
|
|
|
std::vector<yams::memory_sync::MemoryDelta>
|
|
writerWindow(MemorySyncService& writer, std::uint64_t after, std::uint64_t through) {
|
|
yams::memory_sync::VersionVector peer;
|
|
peer.observe("client-node", after);
|
|
auto batch = writer.exportLocalDeltasAfter(peer, 64, through);
|
|
REQUIRE(batch.has_value());
|
|
REQUIRE(batch.value().deltas.size() == through - after);
|
|
return std::move(batch.value().deltas);
|
|
}
|
|
|
|
void publishSeries(MemorySyncService& writer, std::string_view prefix, int count) {
|
|
for (int index = 1; index <= count; ++index) {
|
|
REQUIRE(writer
|
|
.publish("user/" + std::string(prefix) + std::to_string(index),
|
|
bytes(std::string(prefix) + "-value-" + std::to_string(index)))
|
|
.has_value());
|
|
}
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("direct P2P concurrent sessions that advance a peer writer do not quarantine it",
|
|
"[daemon][p2p][delta][commitment][concurrency]") {
|
|
MemorySyncService writer{std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService receiver{std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
publishSeries(writer, "a", 4);
|
|
const auto shorter = writerWindow(writer, 0, 2);
|
|
const auto longer = writerWindow(writer, 0, 4);
|
|
|
|
// Both sessions froze their windows at receiver counter 0; the shorter one applies first.
|
|
auto first =
|
|
receiveWindow(receiver, shorter, frozenWindow("client-node", commitmentAt(writer, 2), 0));
|
|
REQUIRE(first.has_value());
|
|
REQUIRE(receiver.currentVersion().get("client-node") == 2);
|
|
|
|
SECTION("a longer window overlapping the advanced prefix still applies") {
|
|
auto second = receiveWindow(receiver, longer,
|
|
frozenWindow("client-node", commitmentAt(writer, 4), 0));
|
|
REQUIRE(second.has_value());
|
|
CHECK(receiver.currentVersion().get("client-node") == 4);
|
|
CHECK_FALSE(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
REQUIRE(receiver.readCached("user/a4").has_value());
|
|
}
|
|
|
|
SECTION("an empty window frozen below the advanced prefix verifies against history") {
|
|
auto second =
|
|
receiveWindow(receiver, {}, frozenWindow("client-node", commitmentAt(writer, 1), 1));
|
|
REQUIRE(second.has_value());
|
|
CHECK(receiver.currentVersion().get("client-node") == 2);
|
|
CHECK_FALSE(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
}
|
|
|
|
SECTION("an empty window frozen before the writer existed is consistent with any prefix") {
|
|
auto second = receiveWindow(
|
|
receiver, {},
|
|
frozenWindow("client-node", yams::memory_sync::WriterHistoryCommitment{}, 0));
|
|
REQUIRE(second.has_value());
|
|
CHECK_FALSE(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
}
|
|
}
|
|
|
|
TEST_CASE("direct P2P concurrent windows still quarantine a forked or forged writer",
|
|
"[daemon][p2p][delta][commitment][concurrency][security]") {
|
|
MemorySyncService writer{std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
MemorySyncService receiver{std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"server-node", 60'000, "wire-delta-corpus", 1}};
|
|
publishSeries(writer, "a", 4);
|
|
auto first = receiveWindow(receiver, writerWindow(writer, 0, 2),
|
|
frozenWindow("client-node", commitmentAt(writer, 2), 0));
|
|
REQUIRE(first.has_value());
|
|
|
|
SECTION("a fork that rewrote the advanced prefix") {
|
|
// Same writer identity, different history: its frontier cannot extend the
|
|
// receiver's durable prefix even though the window overlaps it.
|
|
MemorySyncService fork{std::make_unique<InMemoryBackend>(),
|
|
MemorySyncConfig{"client-node", 60'000, "wire-delta-corpus", 1}};
|
|
publishSeries(fork, "forked", 4);
|
|
auto forked = receiveWindow(receiver, writerWindow(fork, 0, 4),
|
|
frozenWindow("client-node", commitmentAt(fork, 4), 0));
|
|
REQUIRE_FALSE(forked.has_value());
|
|
CHECK(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
CHECK_FALSE(receiver.readCached("user/forked4").has_value());
|
|
}
|
|
|
|
SECTION("a forged frontier digest") {
|
|
auto forged = commitmentAt(writer, 4);
|
|
forged.digest = std::string(64, 'f');
|
|
auto rejected = receiveWindow(receiver, writerWindow(writer, 0, 4),
|
|
frozenWindow("client-node", forged, 0));
|
|
REQUIRE_FALSE(rejected.has_value());
|
|
CHECK(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
CHECK_FALSE(receiver.readCached("user/a4").has_value());
|
|
}
|
|
|
|
SECTION("an empty window whose frontier contradicts the advanced prefix") {
|
|
auto forged = commitmentAt(writer, 1);
|
|
forged.digest = std::string(64, 'e');
|
|
auto rejected = receiveWindow(receiver, {}, frozenWindow("client-node", forged, 1));
|
|
REQUIRE_FALSE(rejected.has_value());
|
|
CHECK(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
}
|
|
|
|
SECTION("a window that skips a counter below the frontier it advertised") {
|
|
auto skipped = writerWindow(writer, 0, 4);
|
|
skipped.erase(skipped.begin() + 2); // drop counter 3
|
|
auto rejected = receiveWindow(receiver, skipped,
|
|
frozenWindow("client-node", commitmentAt(writer, 4), 0));
|
|
REQUIRE_FALSE(rejected.has_value());
|
|
CHECK(receiver.replicationState().quarantinedWriters.contains("client-node"));
|
|
}
|
|
}
|