From c0df0133e456958d13e23bdfe6f3aee2b6274d59 Mon Sep 17 00:00:00 2001 From: frstrtr Date: Sun, 26 Jul 2026 11:30:52 +0400 Subject: [PATCH] core: p2p message counters, tx-pool gauges and embedded-timestamp diagnostics Not one of the canonical p2pool pool-protocol message types emitted anything at any verbosity, and m_known_txs.size() was exposed nowhere -- no log, no dashboard field, no JSON. A live investigation had to guess, and a log-grep returning zero was misread as "not implemented" for code that was fully wired and running. A grep of zero is not evidence when nothing ever logs. Adds core/p2p_message_stats.hpp: a process-wide struct of relaxed atomics, incremented at the two choke points every pool message must pass through -- pool::NodeBridge::handle() inbound and pool::Peer::write() outbound. Both are shared code, so every coin lane is instrumented without touching a single per-coin protocol file, and the numbers mean the same thing across coins. Read back through a new read-only endpoint, GET /p2p_stats: messages..{in,out} per-type wire counters for version, verack, ping, pong, addrme, addrs, getaddrs, shares, sharereq, sharereply, have_tx, losing_tx, remember_tx, forget_tx, bestblock, plus an unknown bucket totals.{in,out} txpool.known_txs_size m_known_txs.size() txpool.known_txs_order_size m_known_txs_order.size(), null where the lane has no recency deque (DASH) txpool.last_have_tx_advert_size size of the last have_tx put on the wire txpool.last_losing_tx_advert_size txpool.have_tx_adverts_sent sharechain_timestamps.tip_lag_seconds now - tip share_data.timestamp sharechain_timestamps.saturation_fraction share of the last 100 embedded deltas pinned to the clip bound sharechain_timestamps.clip_upper_bound 2*SHARE_PERIOD-1 (39 s on DASH) sharechain_timestamps.{delta_samples,delta_saturated,tip_embedded_timestamp} The txpool block settles whether TXPOOL=0 on a peer dashboard means our pool is genuinely empty or the advert is being suppressed. The timestamp block exists because the aggregate gauges proved untrustworthy. Upstream p2pool clips embedded share timestamps to [prev+1, prev+2*SHARE_PERIOD-1] (data.py:239-242); once real cadence outruns that bound the embedded clock saturates, the difficulty retarget goes blind to hashrate and the floor decays. A 6.81 hour embedded lag was measured live. pool_hash_rate and min_difficulty are computed from that same saturated history, so they report floor decay rather than the pool -- tip lag and saturation fraction are the honest early-warning pair. DASH publishes the timestamp diagnostics from publish_snapshot() (already under the exclusive tracker lock, bounded at 101 chain lookups per think cycle) and the tx-pool gauges from the 10 s advert sweep; btc/ltc/dgb publish their tx-pool gauges from prune_shares(). Observability only: counters and read-only fields. No consensus, minting, share bytes or peer behaviour changes. Per-message logging stays off by default behind a flag nothing in-tree enables. Tests folded into the EXISTING allowlisted core_test target (never a new add_executable -- that is the "Not Run" trap): 12 cases over the command mapping including the NUL-padded wire form, the per-direction counters, and the saturation detector. --- docs/DASHBOARD_INTEGRATION.md | 58 +++++ src/core/CMakeLists.txt | 1 + src/core/http_session.cpp | 62 ++++++ src/core/p2p_message_stats.hpp | 264 +++++++++++++++++++++++ src/core/test/CMakeLists.txt | 11 +- src/core/test/p2p_message_stats_test.cpp | 227 +++++++++++++++++++ src/impl/btc/node.cpp | 11 + src/impl/dash/node.hpp | 107 ++++++++- src/impl/dgb/node.cpp | 11 + src/impl/ltc/node.cpp | 11 + src/pool/node.hpp | 21 ++ src/pool/peer.hpp | 21 ++ 12 files changed, 800 insertions(+), 5 deletions(-) create mode 100644 src/core/p2p_message_stats.hpp create mode 100644 src/core/test/p2p_message_stats_test.cpp 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)); }