diff --git a/docs/DASHBOARD_INTEGRATION.md b/docs/DASHBOARD_INTEGRATION.md index 6c97843a0..2f606fdb2 100644 --- a/docs/DASHBOARD_INTEGRATION.md +++ b/docs/DASHBOARD_INTEGRATION.md @@ -104,6 +104,64 @@ custom dashboard, or scraping pool metrics. | `/ban_stats` | P2P ban statistics | | `/api/coin_peers` | Parent-coin daemon peer info | | `/api/node_topology` | Sharechain topology graph | +| `/p2p_stats` | Per-message-type p2p wire counters + tx-pool and embedded-timestamp gauges | + +### `/p2p_stats` + +Read-only observability for the pool p2p protocol. Before it existed, none of +the pool message types emitted anything at any verbosity, so "is this message +type actually moving on the wire?" had no observable answer and a log-grep +returning zero was easy to misread as "not implemented". + +```json +{ + "messages": { "shares": { "in": 412, "out": 388 }, "remember_tx": { "in": 9, "out": 14 }, "...": {} }, + "totals": { "in": 1204, "out": 1189 }, + "trace_enabled": false, + "txpool": { + "known_txs_size": 12043, + "known_txs_order_size": 12043, + "last_have_tx_advert_size": 500, + "last_losing_tx_advert_size": 0, + "have_tx_adverts_sent": 37, + "updated_at": 1785049468 + }, + "sharechain_timestamps": { + "tip_embedded_timestamp": 1785024952, + "tip_lag_seconds": 24516, + "clip_upper_bound": 39, + "delta_samples": 100, + "delta_saturated": 97, + "saturation_fraction": 0.97, + "updated_at": 1785049468 + } +} +``` + +* `messages` — one `in`/`out` pair per canonical p2pool message type, counted at + the two shared choke points every pool message passes through, so the numbers + are identical in meaning across every coin. `unknown` buckets anything that + did not match a known command. `verack`/`pong` belong to the coin-daemon p2p + layer, not the pool protocol, and read 0 on a healthy pool node — that zero is + an answer, not a gap. +* `txpool` — settles whether a peer dashboard showing `TXPOOL=0` for this node + means the pool is genuinely empty (`known_txs_size` 0) or the advert is being + suppressed (`known_txs_size` > 0 with no adverts sent). + `known_txs_order_size` is `null` on lanes with no recency deque (DASH), which + is deliberately distinct from `0`. +* `sharechain_timestamps` — `tip_lag_seconds` is wall-clock now minus the chain + tip's **embedded** `share_data.timestamp`; `saturation_fraction` is the share + of the last 100 embedded deltas pinned exactly to the clip upper bound + (`2*SHARE_PERIOD - 1`; 39 s on DASH). Upstream p2pool clips embedded + timestamps to `[prev+1, prev+2*SHARE_PERIOD-1]`, so once real cadence outruns + that bound the embedded clock saturates, the difficulty retarget goes blind to + hashrate, and the floor decays. **These two fields are the honest + early-warning signal; `pool_hash_rate` and `min_difficulty` are not** — under + saturation they are computed from that same saturated history and report floor + decay rather than the pool. + +`trace_enabled` reports the per-message debug log, which is off by default and +is not switched on by any code path in-tree; the counters are the mechanism. --- diff --git a/src/core/CMakeLists.txt b/src/core/CMakeLists.txt index 4690da32d..8a9468779 100644 --- a/src/core/CMakeLists.txt +++ b/src/core/CMakeLists.txt @@ -28,6 +28,7 @@ set(source opscript.hpp random.hpp random.cpp addr_store.hpp addr_store.cpp + p2p_message_stats.hpp web_server.hpp web_server.cpp http_session.cpp stratum_server.hpp stratum_server.cpp diff --git a/src/core/http_session.cpp b/src/core/http_session.cpp index 5cc739ce1..658daff8e 100644 --- a/src/core/http_session.cpp +++ b/src/core/http_session.cpp @@ -16,6 +16,7 @@ #include "web_server.hpp" #include "filesystem.hpp" +#include "p2p_message_stats.hpp" #include #include @@ -27,6 +28,65 @@ namespace core { +// ── /p2p_stats — read-only p2p wire observability ────────────────────── +// +// Serialises core::obs::p2p_stats() (src/core/p2p_message_stats.hpp). Pure +// reader: it loads relaxed atomics and formats them. It calls NOTHING on the +// node, takes no lock, and can never block the IO or compute thread — which is +// also why it lives here rather than behind a MiningInterface accessor. +// +// Answers, per message type and direction, "did this actually move on the +// wire?" — a question that previously had NO observable answer at any +// verbosity, so a log-grep returning zero was misread as "not implemented". +static nlohmann::json build_p2p_stats_json() +{ + const auto& s = obs::p2p_stats(); + + nlohmann::json messages = nlohmann::json::object(); + for (std::size_t i = 0; i < obs::P2P_MESSAGE_COUNT; ++i) { + const auto m = static_cast(i); + messages[std::string(obs::p2p_message_name(m))] = { + {"in", s.get_in(m)}, + {"out", s.get_out(m)} + }; + } + + const auto samples = s.ts_delta_samples.load(std::memory_order_relaxed); + const auto saturated = s.ts_delta_saturated.load(std::memory_order_relaxed); + const auto order_size = s.known_txs_order_size.load(std::memory_order_relaxed); + + nlohmann::json out; + out["messages"] = std::move(messages); + out["totals"] = {{"in", s.total_in()}, {"out", s.total_out()}}; + out["trace_enabled"] = s.trace_enabled.load(std::memory_order_relaxed); + + // Tx-pool visibility. known_txs_order_size is null on lanes with no + // recency deque (DASH), NOT 0 — an absent sidecar is not an empty one. + out["txpool"] = { + {"known_txs_size", s.known_txs_size.load(std::memory_order_relaxed)}, + {"known_txs_order_size", order_size < 0 ? nlohmann::json(nullptr) + : nlohmann::json(order_size)}, + {"last_have_tx_advert_size", s.last_have_tx_advert_size.load(std::memory_order_relaxed)}, + {"last_losing_tx_advert_size", s.last_losing_tx_advert_size.load(std::memory_order_relaxed)}, + {"have_tx_adverts_sent", s.have_tx_adverts_sent.load(std::memory_order_relaxed)}, + {"updated_at", s.known_txs_updated_at.load(std::memory_order_relaxed)} + }; + + // Sharechain embedded-timestamp health. tip_lag_seconds and + // saturation_fraction are the honest early-warning pair; pool_hash_rate / + // min_difficulty are NOT (under saturation they measure floor history). + out["sharechain_timestamps"] = { + {"tip_embedded_timestamp", s.tip_embedded_timestamp.load(std::memory_order_relaxed)}, + {"tip_lag_seconds", s.tip_lag_seconds.load(std::memory_order_relaxed)}, + {"clip_upper_bound", s.ts_clip_upper_bound.load(std::memory_order_relaxed)}, + {"delta_samples", samples}, + {"delta_saturated", saturated}, + {"saturation_fraction", samples ? static_cast(saturated) / samples : 0.0}, + {"updated_at", s.sharechain_updated_at.load(std::memory_order_relaxed)} + }; + return out; +} + // ── Security helpers ─────────────────────────────────────────────────── // URL-decode percent-encoded strings (%20 → space, etc.) @@ -380,6 +440,8 @@ void HttpSession::process_request() rest_result = mining_interface_->rest_v36_status(); else if (target == "/tracker_debug") rest_result = mining_interface_->rest_tracker_debug(); + else if (target == "/p2p_stats") + rest_result = build_p2p_stats_json(); // Merged mining endpoints else if (target == "/merged_stats") diff --git a/src/core/p2p_message_stats.hpp b/src/core/p2p_message_stats.hpp new file mode 100644 index 000000000..5ddda3292 --- /dev/null +++ b/src/core/p2p_message_stats.hpp @@ -0,0 +1,264 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +#pragma once + +// Process-wide p2p wire observability — OBSERVE ONLY. +// +// WHY THIS EXISTS +// --------------- +// Before this file, not one of the 13 canonical p2pool pool-protocol message +// types emitted anything at any verbosity: no log line, no counter, no JSON +// field. "Is `remember_tx` actually moving on the wire?" could only be answered +// by reading the source and hoping. A live investigation grepped the logs for a +// message name, got zero hits, and concluded the feature was unimplemented — +// when in fact it was fully wired and running. A log-grep of zero is not +// evidence when nothing ever logs. +// +// So: one cheap relaxed atomic per (message type x direction), incremented at +// the two choke points every pool message must pass through +// * inbound — pool::NodeBridge::handle() (src/pool/node.hpp) +// * outbound — pool::Peer::write() (src/pool/peer.hpp) +// plus a handful of gauges the coin lane publishes (known-tx pool size, last +// have_tx advert size, sharechain embedded-timestamp health), read back through +// the read-only /p2p_stats endpoint. +// +// Instrumenting the two shared choke points is deliberate: it means NO per-coin +// protocol file has to be touched, and every coin lane (btc/ltc/dgb/bch/dash/ +// doge/nmc) gets the counters from the same code. +// +// CONSENSUS SAFETY: nothing here is read by share validation, minting, payout, +// coinbase construction or peer behaviour. Every member is a counter or a +// display gauge; nothing in this header can change a byte that goes on the wire. +// +// COST: one relaxed fetch_add per message, plus a length-first string_view +// compare against at most 15 short literals. Messages arrive at single-digit +// per-second rates per peer; this is not measurable. + +#include +#include +#include +#include +#include +#include + +namespace core::obs +{ + +// The canonical p2pool pool-protocol message set. `verack` and `pong` are NOT +// part of the p2pool pool protocol (they belong to the coin-daemon p2p layer); +// they are carried here so the endpoint answers "is this type on the wire?" for +// the full operator-facing list rather than silently omitting two names. They +// read 0 on a healthy pool node — that zero is the answer, not a gap. +enum class P2PMessage : std::size_t +{ + version = 0, + verack, + ping, + pong, + addrme, + addrs, + getaddrs, + shares, + sharereq, + sharereply, + have_tx, + losing_tx, + remember_tx, + forget_tx, + bestblock, + unknown, // catch-all bucket; MUST stay last + COUNT +}; + +inline constexpr std::size_t P2P_MESSAGE_COUNT = static_cast(P2PMessage::COUNT); + +inline constexpr std::array P2P_MESSAGE_NAMES = { + "version", "verack", "ping", "pong", "addrme", "addrs", "getaddrs", + "shares", "sharereq", "sharereply", + "have_tx", "losing_tx", "remember_tx", "forget_tx", "bestblock", + "unknown" +}; + +inline constexpr std::string_view p2p_message_name(P2PMessage m) +{ + auto i = static_cast(m); + return i < P2P_MESSAGE_COUNT ? P2P_MESSAGE_NAMES[i] : P2P_MESSAGE_NAMES[P2P_MESSAGE_COUNT - 1]; +} + +// Wire command fields are fixed-width and NUL-padded; RawMessage::m_command +// still carries that padding until MessageHandler::parse() strips it, and the +// inbound counter runs BEFORE parse. Trim here so "ping\0\0\0\0" matches "ping". +inline constexpr std::string_view trim_command(std::string_view cmd) +{ + const auto z = cmd.find('\0'); + return z == std::string_view::npos ? cmd : cmd.substr(0, z); +} + +inline constexpr P2PMessage p2p_message_from_command(std::string_view cmd) +{ + const auto trimmed = trim_command(cmd); + // string_view::operator== compares size first, so this is 15 cheap + // length checks and at most one memcmp. + for (std::size_t i = 0; i + 1 < P2P_MESSAGE_COUNT; ++i) + if (trimmed == P2P_MESSAGE_NAMES[i]) + return static_cast(i); + return P2PMessage::unknown; +} + +// ── sharechain embedded-timestamp health (pure, KAT-able) ──────────────────── +// +// Upstream p2pool clips every share's EMBEDDED timestamp into +// [prev.timestamp + 1, prev.timestamp + 2*SHARE_PERIOD - 1] +// (p2pool data.py:239-242). When the real share cadence is faster than that +// upper bound the embedded clock SATURATES: consecutive embedded deltas all sit +// exactly on the bound, the embedded chain clock falls further and further +// behind wall-clock, and the difficulty retarget — which reads embedded +// timestamps, not wall-clock — goes blind to actual hashrate. +// +// Under saturation the aggregate gauges LIE: pool_hash_rate and min_difficulty +// are computed from that same saturated embedded history, so they report the +// decaying floor rather than the pool. The two honest early-warning signals are +// (a) tip lag against wall-clock and (b) the fraction of recent embedded deltas +// pinned to the clip bound — which is what this computes. +struct TimestampSaturation +{ + std::uint32_t samples{0}; // number of consecutive-share deltas examined + std::uint32_t saturated{0}; // deltas exactly equal to the clip upper bound + double fraction{0.0}; // saturated / samples (0 when samples == 0) +}; + +/// @param ts_newest_first embedded share timestamps walked tip-first along +/// prev_hash (index 0 = chain tip, index i+1 = parent of index i). +/// @param clip_upper_bound 2*SHARE_PERIOD - 1 for the coin (39 s on DASH). +/// N timestamps yield N-1 deltas. A delta is "saturated" when it equals the +/// bound exactly. Out-of-order pairs (child older than parent — only reachable +/// on a malformed/forked walk) count as a sample but never as saturated, so a +/// corrupt walk can never manufacture a false all-clear OR a false alarm. +inline TimestampSaturation compute_timestamp_saturation( + const std::vector& ts_newest_first, + std::uint32_t clip_upper_bound) +{ + TimestampSaturation out; + if (ts_newest_first.size() < 2 || clip_upper_bound == 0) + return out; + + for (std::size_t i = 0; i + 1 < ts_newest_first.size(); ++i) + { + const std::uint32_t child = ts_newest_first[i]; + const std::uint32_t parent = ts_newest_first[i + 1]; + ++out.samples; + if (child >= parent && (child - parent) == clip_upper_bound) + ++out.saturated; + } + if (out.samples > 0) + out.fraction = static_cast(out.saturated) / static_cast(out.samples); + return out; +} + +/// Wall-clock now MINUS the tip's EMBEDDED timestamp, in seconds. Positive = +/// the embedded chain clock trails real time (6.81 HOURS was measured live on +/// DASH under saturation). Returns 0 when no tip timestamp is known. +inline std::int64_t compute_tip_lag_seconds(std::int64_t now_unix, std::uint32_t tip_embedded_timestamp) +{ + if (tip_embedded_timestamp == 0) + return 0; + return now_unix - static_cast(tip_embedded_timestamp); +} + +// ── the counters themselves ────────────────────────────────────────────────── +struct P2PMessageStats +{ + // DELIVERABLE 1 — per-message-type, per-direction wire counters. + std::array, P2P_MESSAGE_COUNT> in{}; + std::array, P2P_MESSAGE_COUNT> out{}; + + // DELIVERABLE 2 — tx-pool visibility. Settles "is TXPOOL=0 on a peer + // dashboard our pool being genuinely empty, or the advert being suppressed?" + std::atomic known_txs_size{0}; + // -1 = this coin lane has no m_known_txs_order recency deque (DASH does + // not; btc/ltc/dgb do). Rendered as JSON null so an absent sidecar is never + // confused with an empty one. + std::atomic known_txs_order_size{-1}; + std::atomic last_have_tx_advert_size{0}; + std::atomic last_losing_tx_advert_size{0}; + std::atomic have_tx_adverts_sent{0}; + std::atomic known_txs_updated_at{0}; // unix seconds, 0 = never + + // DELIVERABLE 3 — sharechain embedded-timestamp diagnostics. + std::atomic tip_embedded_timestamp{0}; + std::atomic tip_lag_seconds{0}; + std::atomic ts_delta_samples{0}; + std::atomic ts_delta_saturated{0}; + std::atomic ts_clip_upper_bound{0}; // 2*SHARE_PERIOD - 1 + std::atomic sharechain_updated_at{0}; // unix seconds, 0 = never + + // Debug-gated per-message trace. OFF by default and never enabled by any + // code path in-tree — an operator flips it deliberately. The counters, not + // logs, are the hot-path mechanism. + std::atomic trace_enabled{false}; + + void count_in(std::string_view command) noexcept + { + in[static_cast(p2p_message_from_command(command))] + .fetch_add(1, std::memory_order_relaxed); + } + + void count_out(std::string_view command) noexcept + { + out[static_cast(p2p_message_from_command(command))] + .fetch_add(1, std::memory_order_relaxed); + } + + std::uint64_t get_in(P2PMessage m) const noexcept + { + return in[static_cast(m)].load(std::memory_order_relaxed); + } + + std::uint64_t get_out(P2PMessage m) const noexcept + { + return out[static_cast(m)].load(std::memory_order_relaxed); + } + + std::uint64_t total_in() const noexcept + { + std::uint64_t t = 0; + for (const auto& c : in) t += c.load(std::memory_order_relaxed); + return t; + } + + std::uint64_t total_out() const noexcept + { + std::uint64_t t = 0; + for (const auto& c : out) t += c.load(std::memory_order_relaxed); + return t; + } + + /// Test-only helper; never called by the node. + void reset() noexcept + { + for (auto& c : in) c.store(0, std::memory_order_relaxed); + for (auto& c : out) c.store(0, std::memory_order_relaxed); + known_txs_size.store(0, std::memory_order_relaxed); + known_txs_order_size.store(-1, std::memory_order_relaxed); + last_have_tx_advert_size.store(0, std::memory_order_relaxed); + last_losing_tx_advert_size.store(0, std::memory_order_relaxed); + have_tx_adverts_sent.store(0, std::memory_order_relaxed); + known_txs_updated_at.store(0, std::memory_order_relaxed); + tip_embedded_timestamp.store(0, std::memory_order_relaxed); + tip_lag_seconds.store(0, std::memory_order_relaxed); + ts_delta_samples.store(0, std::memory_order_relaxed); + ts_delta_saturated.store(0, std::memory_order_relaxed); + ts_clip_upper_bound.store(0, std::memory_order_relaxed); + sharechain_updated_at.store(0, std::memory_order_relaxed); + } +}; + +/// Process-wide instance. c2pool runs ONE pool node per process (the per-coin +/// binary invariant), so a process-global is exactly per-node scope and needs no +/// plumbing through the node -> web-server seam. +inline P2PMessageStats& p2p_stats() +{ + static P2PMessageStats stats; + return stats; +} + +} // namespace core::obs diff --git a/src/core/test/CMakeLists.txt b/src/core/test/CMakeLists.txt index 46e2fd1c4..af1d18dd2 100644 --- a/src/core/test/CMakeLists.txt +++ b/src/core/test/CMakeLists.txt @@ -36,7 +36,16 @@ if (BUILD_TESTING AND (GTest_FOUND OR GTEST_FOUND)) # per-coin protocol_legacy.cpp / protocol_actual.cpp addrme+addrs handlers. # Same reasoning as above: FOLDED into the EXISTING allowlisted core_test # target, never a new add_executable. - add_executable(core_test pack_test.cpp events_test.cpp chain_test.cpp packet_test.cpp block_broadcast_test.cpp broadcast_convergence_matrix_test.cpp core_merkle_branches_test.cpp version_gate_test.cpp filesystem_test.cpp web_server_submitblock_test.cpp known_txs_eviction_test.cpp tx_advertiser_test.cpp socket_write_queue_test.cpp known_txs_backing_test.cpp random_choice_guard_test.cpp) + # p2p_message_stats_test.cpp: the p2p wire observability layer + # (core/p2p_message_stats.hpp) -- command -> message-type mapping including + # the NUL-padded wire form, the per-direction counters, and the embedded- + # timestamp saturation detector that backs the DASH deploy criterion. + # Header-only over core, and BOTH instrumentation points + # (pool::NodeBridge::handle inbound, pool::Peer::write outbound) are shared + # code every coin lane routes through, so one KAT covers all coins. Same + # reasoning as above: FOLDED into the EXISTING allowlisted core_test target, + # never a new add_executable (the #769 "Not Run" trap). + add_executable(core_test pack_test.cpp events_test.cpp chain_test.cpp packet_test.cpp block_broadcast_test.cpp broadcast_convergence_matrix_test.cpp core_merkle_branches_test.cpp version_gate_test.cpp filesystem_test.cpp web_server_submitblock_test.cpp known_txs_eviction_test.cpp tx_advertiser_test.cpp socket_write_queue_test.cpp known_txs_backing_test.cpp random_choice_guard_test.cpp p2p_message_stats_test.cpp) target_compile_definitions(core_test PRIVATE C2POOL_SRC_ROOT="${CMAKE_SOURCE_DIR}/src") target_link_libraries(core_test PRIVATE GTest::gtest_main core c2pool_merged_mining GTest::gtest) target_link_libraries(core_test PRIVATE c2pool_payout c2pool_hashrate ltc_coin) # OBJECT-lib SCC direct-naming (#22/#39) diff --git a/src/core/test/p2p_message_stats_test.cpp b/src/core/test/p2p_message_stats_test.cpp new file mode 100644 index 000000000..cb74493d9 --- /dev/null +++ b/src/core/test/p2p_message_stats_test.cpp @@ -0,0 +1,227 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// KAT for core/p2p_message_stats.hpp — the p2p wire observability layer. +// +// Folded into the EXISTING allowlisted core_test target (never a new +// add_executable: a standalone target is not in build.yml's --target list, so +// CI never builds it and CTest reports the cases "Not Run" — the #769 trap that +// has now bitten this repo three times). +// +// Covers the two pieces with real logic in them: +// 1. command-string -> message-type mapping, including the NUL-padded wire +// form that reaches the inbound counter BEFORE MessageHandler::parse() +// strips the padding, plus the unknown-command bucket; +// 2. compute_timestamp_saturation() — the embedded-timestamp clip detector +// behind the DASH deploy criterion, and compute_tip_lag_seconds(). +// +// One KAT covers every coin: both instrumentation points (pool::NodeBridge:: +// handle for inbound, pool::Peer::write for outbound) are shared code that all +// coin lanes route through. + +#include + +#include + +#include +#include +#include + +using namespace core::obs; + +// ── 1. command -> message type ────────────────────────────────────────────── + +TEST(P2PMessageStats, MapsEveryCanonicalCommand) +{ + // The full operator-facing list. Each name must map to its own slot. + const std::vector> expected = { + {"version", P2PMessage::version}, + {"verack", P2PMessage::verack}, + {"ping", P2PMessage::ping}, + {"pong", P2PMessage::pong}, + {"addrme", P2PMessage::addrme}, + {"addrs", P2PMessage::addrs}, + {"getaddrs", P2PMessage::getaddrs}, + {"shares", P2PMessage::shares}, + {"sharereq", P2PMessage::sharereq}, + {"sharereply", P2PMessage::sharereply}, + {"have_tx", P2PMessage::have_tx}, + {"losing_tx", P2PMessage::losing_tx}, + {"remember_tx", P2PMessage::remember_tx}, + {"forget_tx", P2PMessage::forget_tx}, + {"bestblock", P2PMessage::bestblock}, + }; + ASSERT_EQ(expected.size(), P2P_MESSAGE_COUNT - 1) // -1 for the unknown bucket + << "message list drifted from the enum"; + + for (const auto& [cmd, type] : expected) { + EXPECT_EQ(p2p_message_from_command(cmd), type) << "command: " << cmd; + EXPECT_EQ(p2p_message_name(type), cmd); + } +} + +TEST(P2PMessageStats, TrimsNulPaddedWireCommands) +{ + // Wire command fields are fixed-width and NUL-padded. The inbound counter + // runs BEFORE MessageHandler::parse() strips the padding, so the untrimmed + // form must map to the same slot — otherwise every inbound message would + // land in the unknown bucket and the counters would be useless. + const std::string padded("ping\0\0\0\0\0\0\0\0", 12); + EXPECT_EQ(p2p_message_from_command(padded), P2PMessage::ping); + EXPECT_EQ(trim_command(padded), "ping"); + + const std::string padded_share("remember_tx\0", 12); + EXPECT_EQ(p2p_message_from_command(padded_share), P2PMessage::remember_tx); +} + +TEST(P2PMessageStats, UnknownCommandsLandInTheUnknownBucket) +{ + EXPECT_EQ(p2p_message_from_command(""), P2PMessage::unknown); + EXPECT_EQ(p2p_message_from_command("not_a_message"), P2PMessage::unknown); + EXPECT_EQ(p2p_message_from_command("ver"), P2PMessage::unknown); // prefix, not a match + EXPECT_EQ(p2p_message_from_command("versionx"), P2PMessage::unknown); // superstring + EXPECT_EQ(p2p_message_from_command("PING"), P2PMessage::unknown); // case sensitive +} + +// ── 2. counters ───────────────────────────────────────────────────────────── + +TEST(P2PMessageStats, CountsInAndOutIndependently) +{ + P2PMessageStats stats; // local instance; the process-global stays untouched + + stats.count_in("shares"); + stats.count_in("shares"); + stats.count_in(std::string("have_tx\0\0", 9)); // NUL-padded wire form + stats.count_out("sharereq"); + stats.count_out("bestblock"); + stats.count_in("garbage"); + + EXPECT_EQ(stats.get_in(P2PMessage::shares), 2u); + EXPECT_EQ(stats.get_in(P2PMessage::have_tx), 1u); + EXPECT_EQ(stats.get_out(P2PMessage::shares), 0u) + << "inbound must not leak into the outbound counter"; + EXPECT_EQ(stats.get_out(P2PMessage::sharereq), 1u); + EXPECT_EQ(stats.get_out(P2PMessage::bestblock), 1u); + EXPECT_EQ(stats.get_in(P2PMessage::unknown), 1u); + + EXPECT_EQ(stats.total_in(), 4u); + EXPECT_EQ(stats.total_out(), 2u); + + stats.reset(); + EXPECT_EQ(stats.total_in(), 0u); + EXPECT_EQ(stats.total_out(), 0u); + EXPECT_EQ(stats.known_txs_order_size.load(), -1) + << "reset must restore the 'lane has no order deque' sentinel, not 0"; +} + +TEST(P2PMessageStats, DefaultsAreZeroAndTraceIsOff) +{ + P2PMessageStats stats; + EXPECT_EQ(stats.total_in(), 0u); + EXPECT_EQ(stats.total_out(), 0u); + EXPECT_EQ(stats.known_txs_size.load(), 0u); + EXPECT_EQ(stats.known_txs_order_size.load(), -1); + EXPECT_FALSE(stats.trace_enabled.load()) + << "per-message tracing must be OFF by default (hot path)"; +} + +// ── 3. embedded-timestamp saturation ──────────────────────────────────────── +// +// DASH: SHARE_PERIOD = 20 s, so the p2pool clip upper bound is 2*20-1 = 39 s. +static constexpr std::uint32_t DASH_CLIP = 39; + +TEST(TimestampSaturation, FullySaturatedChainReportsOne) +{ + // Every consecutive embedded delta pinned to the clip bound: the failure + // mode measured live (embedded clock 6.81 h behind wall-clock). + std::vector ts; // newest first + std::uint32_t t = 1'700'000'000; + for (int i = 0; i < 101; ++i) { ts.push_back(t); t -= DASH_CLIP; } + + const auto sat = compute_timestamp_saturation(ts, DASH_CLIP); + EXPECT_EQ(sat.samples, 100u); + EXPECT_EQ(sat.saturated, 100u); + EXPECT_DOUBLE_EQ(sat.fraction, 1.0); +} + +TEST(TimestampSaturation, HealthyCadenceReportsZero) +{ + // Deltas comfortably under the bound — retarget still sees real cadence. + std::vector ts; + std::uint32_t t = 1'700'000'000; + for (int i = 0; i < 101; ++i) { ts.push_back(t); t -= 20; } + + const auto sat = compute_timestamp_saturation(ts, DASH_CLIP); + EXPECT_EQ(sat.samples, 100u); + EXPECT_EQ(sat.saturated, 0u); + EXPECT_DOUBLE_EQ(sat.fraction, 0.0); +} + +TEST(TimestampSaturation, MixedChainReportsTheFraction) +{ + // 4 saturated deltas out of 10. + const std::vector deltas = + {DASH_CLIP, 20, DASH_CLIP, 5, DASH_CLIP, 12, 20, DASH_CLIP, 7, 1}; + std::vector ts; + std::uint32_t t = 1'700'000'000; + ts.push_back(t); + for (auto d : deltas) { t -= d; ts.push_back(t); } + + const auto sat = compute_timestamp_saturation(ts, DASH_CLIP); + EXPECT_EQ(sat.samples, 10u); + EXPECT_EQ(sat.saturated, 4u); + EXPECT_DOUBLE_EQ(sat.fraction, 0.4); +} + +TEST(TimestampSaturation, OffByOneDeltasAreNotSaturated) +{ + // 38 and 40 are NOT the bound. The detector must be exact — a fuzzy match + // would fire on a healthy chain and this field is a deploy criterion. + const std::vector ts = {1'000'078, 1'000'040, 1'000'000}; + const auto sat = compute_timestamp_saturation(ts, DASH_CLIP); + EXPECT_EQ(sat.samples, 2u); + EXPECT_EQ(sat.saturated, 0u); +} + +TEST(TimestampSaturation, DegenerateInputsAreSafe) +{ + EXPECT_EQ(compute_timestamp_saturation({}, DASH_CLIP).samples, 0u); + EXPECT_EQ(compute_timestamp_saturation({12345}, DASH_CLIP).samples, 0u); + EXPECT_DOUBLE_EQ(compute_timestamp_saturation({}, DASH_CLIP).fraction, 0.0); + // clip 0 is not a real coin configuration; must not divide or fire. + EXPECT_EQ(compute_timestamp_saturation({100, 100, 100}, 0).samples, 0u); +} + +TEST(TimestampSaturation, OutOfOrderPairsCountAsSampleButNeverSaturated) +{ + // Child older than parent is only reachable on a malformed/forked walk. + // It must not manufacture a false all-clear OR a false alarm — including + // when the reversed magnitude happens to equal the clip bound exactly. + const std::vector ts = {1'000'000, 1'000'039, 1'000'078}; + const auto sat = compute_timestamp_saturation(ts, DASH_CLIP); + EXPECT_EQ(sat.samples, 2u); + EXPECT_EQ(sat.saturated, 0u); + + // A well-ordered saturated pair adjacent to a reversed one still counts + // exactly once: {t, t-39, t-39+39=t} -> pair0 saturated, pair1 reversed. + const std::vector mixed = {1'000'039, 1'000'000, 1'000'039}; + const auto sat_mixed = compute_timestamp_saturation(mixed, DASH_CLIP); + EXPECT_EQ(sat_mixed.samples, 2u); + EXPECT_EQ(sat_mixed.saturated, 1u); +} + +// ── 4. tip lag ────────────────────────────────────────────────────────────── + +TEST(TipLag, MeasuresWallClockMinusEmbeddedTimestamp) +{ + // The live DASH observation: embedded tip 6.81 hours behind wall-clock. + const std::int64_t now = 1'700'024'516; + const std::uint32_t tip = 1'700'000'000; + EXPECT_EQ(compute_tip_lag_seconds(now, tip), 24'516); + + // No tip timestamp known yet -> 0, never a bogus "now" sized lag. + EXPECT_EQ(compute_tip_lag_seconds(now, 0), 0); + + // A tip stamped slightly ahead of local wall-clock is legal (clock skew); + // report it honestly as negative rather than clamping. + EXPECT_EQ(compute_tip_lag_seconds(1'700'000'000, 1'700'000'005), -5); +} diff --git a/src/impl/btc/node.cpp b/src/impl/btc/node.cpp index 649189a85..20d4dc55d 100644 --- a/src/impl/btc/node.cpp +++ b/src/impl/btc/node.cpp @@ -1505,6 +1505,17 @@ void NodeImpl::prune_shares(const uint256& /*best_share*/) // Evict oldest-first down to the cap, keeping the most-recently-learned txs. if (m_known_txs.size() > m_max_known_txs) core::evict_known_txs_to_cap(m_known_txs, m_known_txs_order, m_max_known_txs); + + // /p2p_stats gauges (observe-only): the SEND-side truth behind a peer + // dashboard reporting TXPOOL=0 for us. Published here because prune_shares + // is the periodic pass that already touches both containers. Relaxed + // stores, no lock, no consensus/mint/payout state read or written. + core::obs::p2p_stats().known_txs_size.store( + m_known_txs.size(), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_order_size.store( + static_cast(m_known_txs_order.size()), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_updated_at.store( + static_cast(core::timestamp()), std::memory_order_relaxed); if (m_raw_share_cache.size() > m_max_raw_shares) m_raw_share_cache.clear(); } diff --git a/src/impl/dash/node.hpp b/src/impl/dash/node.hpp index 1257a08ac..452d97d1f 100644 --- a/src/impl/dash/node.hpp +++ b/src/impl/dash/node.hpp @@ -33,6 +33,7 @@ #include #include #include // shared downloader helpers (#754) +#include #include #include #include @@ -187,8 +188,83 @@ class NodeImpl : public pool::BaseNode(m_tracker.chain.get_heads().size()); s.fork_count = s.head_count; s.best_share = m_best_share_hash; - std::lock_guard lock(m_snapshot_mutex); - m_snapshot = s; + publish_timestamp_diagnostics(); + { + std::lock_guard lock(m_snapshot_mutex); + m_snapshot = s; + } + } + + // ── /p2p_stats: embedded-timestamp health of the sharechain tip ───────── + // + // OBSERVE-ONLY. Publishes the two fields the aggregate gauges cannot give + // us honestly: + // (a) tip_lag_seconds — wall-clock now MINUS the tip's EMBEDDED + // share timestamp; + // (b) ts_saturation_fraction — over the last 100 shares, the share of + // embedded deltas pinned to the clip upper bound (2*SHARE_PERIOD-1, + // = 39 s on DASH). + // + // Upstream p2pool clips every embedded share timestamp to + // [prev+1, prev+2*SHARE_PERIOD-1] (data.py:239-242). Once real cadence + // outruns that bound the embedded clock saturates, the retarget (which + // reads embedded timestamps) goes blind to hashrate, and the floor decays. + // pool_hash_rate / min_difficulty are computed FROM that saturated history, + // so they measure floor decay, not the pool — which is exactly why these + // two raw fields exist as the deploy criterion instead. + // + // THREADING: reads m_tracker.chain, so it MUST run under the tracker lock. + // Called only from publish_snapshot(), whose three call sites are the two + // run_think()/clean-cycle compute phases (exclusive m_tracker_mutex held) + // and the single-threaded LevelDB load before the compute thread starts. + // + // COST: bounded at 101 chain lookups per think cycle, no allocation beyond + // one 101-entry vector. Nothing here feeds validation, minting or payout. + void publish_timestamp_diagnostics() + { + static constexpr std::size_t SAMPLE_SHARES = 100; + + std::vector ts; + ts.reserve(SAMPLE_SHARES + 1); + + uint256 cursor = m_best_share_hash; + while (ts.size() <= SAMPLE_SHARES && !cursor.IsNull() + && m_tracker.chain.contains(cursor)) + { + std::uint32_t stamp = 0; + uint256 prev; + m_tracker.chain.get_share(cursor).invoke([&](auto* obj) { + stamp = obj->m_timestamp; + prev = obj->m_prev_hash; + }); + ts.push_back(stamp); + cursor = prev; + } + + const std::uint32_t clip = + SharechainConfig::share_period() * 2 - 1; // 39 s on DASH + const auto sat = core::obs::compute_timestamp_saturation(ts, clip); + const std::uint32_t tip_ts = ts.empty() ? 0u : ts.front(); + + auto& stats = core::obs::p2p_stats(); + stats.tip_embedded_timestamp.store(tip_ts, std::memory_order_relaxed); + stats.tip_lag_seconds.store( + core::obs::compute_tip_lag_seconds( + static_cast(core::timestamp()), tip_ts), + std::memory_order_relaxed); + stats.ts_delta_samples.store(sat.samples, std::memory_order_relaxed); + stats.ts_delta_saturated.store(sat.saturated, std::memory_order_relaxed); + stats.ts_clip_upper_bound.store(clip, std::memory_order_relaxed); + stats.sharechain_updated_at.store( + static_cast(core::timestamp()), std::memory_order_relaxed); + + // DELIBERATELY NOT published here: m_known_txs.size(). This body runs + // on the COMPUTE thread, and in the DASH lane every m_known_txs mutator + // is IO-thread-confined WITHOUT taking the tracker lock (see the + // invariant note on advertise_known_txs) — so reading the map from here + // would be an unsynchronised cross-thread read of a container being + // mutated, the exact #828 discipline this file exists to keep. The + // 10 s advert sweep publishes that gauge from the IO thread instead. } mutable std::mutex m_snapshot_mutex; TrackerSnapshot m_snapshot; @@ -946,6 +1022,16 @@ class NodeImpl : public pool::BaseNode(core::timestamp()), std::memory_order_relaxed); + // One clock reading for the whole sweep: run_tx_advert uses it both to // apply the per-peer min-emit interval (never two writes in flight on // one socket) and to stamp the peer after a send. IO-thread-local, so @@ -959,9 +1045,22 @@ class NodeImpl : public pool::BaseNodem_tx_advert, current, [&p](const std::vector& hashes) - { p->write(dash::message_have_tx::make_raw(hashes)); }, + { + // /p2p_stats: size of the LAST have_tx advert actually put + // on the wire, plus how many we have emitted. A pool with + // known_txs_size > 0 but have_tx_adverts_sent == 0 is a + // suppressed advert; both zero is a genuinely empty pool. + auto& stats = core::obs::p2p_stats(); + stats.last_have_tx_advert_size.store(hashes.size(), std::memory_order_relaxed); + stats.have_tx_adverts_sent.fetch_add(1, std::memory_order_relaxed); + p->write(dash::message_have_tx::make_raw(hashes)); + }, [&p](const std::vector& hashes) - { p->write(dash::message_losing_tx::make_raw(hashes)); }, + { + core::obs::p2p_stats().last_losing_tx_advert_size.store( + hashes.size(), std::memory_order_relaxed); + p->write(dash::message_losing_tx::make_raw(hashes)); + }, core::TX_ADVERT_MAX_HASHES_PER_MESSAGE, now); }; diff --git a/src/impl/dgb/node.cpp b/src/impl/dgb/node.cpp index db43446e8..4efe0576b 100644 --- a/src/impl/dgb/node.cpp +++ b/src/impl/dgb/node.cpp @@ -1478,6 +1478,17 @@ void NodeImpl::prune_shares(const uint256& /*best_share*/) // Evict oldest-first down to the cap, keeping the most-recently-learned txs. if (m_known_txs.size() > m_max_known_txs) core::evict_known_txs_to_cap(m_known_txs, m_known_txs_order, m_max_known_txs); + + // /p2p_stats gauges (observe-only): the SEND-side truth behind a peer + // dashboard reporting TXPOOL=0 for us. Published here because prune_shares + // is the periodic pass that already touches both containers. Relaxed + // stores, no lock, no consensus/mint/payout state read or written. + core::obs::p2p_stats().known_txs_size.store( + m_known_txs.size(), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_order_size.store( + static_cast(m_known_txs_order.size()), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_updated_at.store( + static_cast(core::timestamp()), std::memory_order_relaxed); if (m_raw_share_cache.size() > m_max_raw_shares) m_raw_share_cache.clear(); } diff --git a/src/impl/ltc/node.cpp b/src/impl/ltc/node.cpp index a569bbe21..35ba612a0 100644 --- a/src/impl/ltc/node.cpp +++ b/src/impl/ltc/node.cpp @@ -1645,6 +1645,17 @@ void NodeImpl::prune_shares(const uint256& /*best_share*/) // Evict oldest-first down to the cap, keeping the most-recently-learned txs. if (m_known_txs.size() > m_max_known_txs) core::evict_known_txs_to_cap(m_known_txs, m_known_txs_order, m_max_known_txs); + + // /p2p_stats gauges (observe-only): the SEND-side truth behind a peer + // dashboard reporting TXPOOL=0 for us. Published here because prune_shares + // is the periodic pass that already touches both containers. Relaxed + // stores, no lock, no consensus/mint/payout state read or written. + core::obs::p2p_stats().known_txs_size.store( + m_known_txs.size(), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_order_size.store( + static_cast(m_known_txs_order.size()), std::memory_order_relaxed); + core::obs::p2p_stats().known_txs_updated_at.store( + static_cast(core::timestamp()), std::memory_order_relaxed); if (m_raw_share_cache.size() > m_max_raw_shares) m_raw_share_cache.clear(); } diff --git a/src/pool/node.hpp b/src/pool/node.hpp index 57dbb1d57..a2c4f5d86 100644 --- a/src/pool/node.hpp +++ b/src/pool/node.hpp @@ -14,6 +14,7 @@ #include #include #include +#include namespace pool { @@ -208,6 +209,26 @@ class NodeBridge : public virtual Base, public Legacy, public Actual void handle(std::unique_ptr rmsg, const NetService& service) override { + // RECEIVE choke point for the pool protocol: every inbound pool + // message in every coin lane is dispatched from here, whether it ends + // up in the Legacy (p2pool) or Actual (c2pool) protocol handler. The + // inbound counter therefore lives here and NOT in the per-coin + // protocol_legacy.cpp / protocol_actual.cpp files. + // + // Counted BEFORE the connection guard on purpose: a message that + // arrives for an already-dropped peer still crossed the wire, and + // hiding it would reproduce exactly the blind spot this counter set + // exists to close. Observe-only — relaxed fetch_add, message untouched. + if (rmsg) + { + auto& stats = core::obs::p2p_stats(); + stats.count_in(rmsg->m_command); + if (stats.trace_enabled.load(std::memory_order_relaxed)) + LOG_DEBUG_POOL << "[p2p-msg] in " + << std::string(core::obs::trim_command(rmsg->m_command)) + << " <- " << service.to_string(); + } + // Guard: peer may have been removed by a prior error/timeout while // an async_read callback was still in-flight for the same socket. if (!Base::m_connections.contains(service)) diff --git a/src/pool/peer.hpp b/src/pool/peer.hpp index db2a12b61..2d320f9fe 100644 --- a/src/pool/peer.hpp +++ b/src/pool/peer.hpp @@ -2,6 +2,8 @@ #pragma once #include +#include +#include #include #include namespace pool @@ -42,8 +44,27 @@ class Peer : public Data m_socket->close(); } + /// SEND choke point for the pool protocol. EVERY outbound pool message in + /// every coin lane goes through here (verified: the only other + /// Socket::write callers are the coin-daemon p2p connections, a different + /// protocol), which is why the outbound counter lives here and not in the + /// per-coin protocol files. + /// + /// Observe-only: the counter is a relaxed fetch_add and the message is + /// forwarded byte-unchanged. void write(std::unique_ptr rmsg) { + if (rmsg) + { + auto& stats = core::obs::p2p_stats(); + stats.count_out(rmsg->m_command); + // Off by default; nothing in-tree enables it. Counters are the + // hot-path mechanism, this is the opt-in magnifying glass. + if (stats.trace_enabled.load(std::memory_order_relaxed)) + LOG_DEBUG_POOL << "[p2p-msg] out " + << std::string(core::obs::trim_command(rmsg->m_command)) + << " -> " << m_socket->get_addr().to_string(); + } m_socket->write(std::move(rmsg)); }