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.<type>.{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.

Author: frstrtr

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.
 
 ---
 

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

src/core/http_session.cpp

@@ -16,6 +16,7 @@
 
 #include "web_server.hpp"
 #include "filesystem.hpp"
+#include "p2p_message_stats.hpp"
 
 #include <filesystem>
 #include <fstream>
@@ -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<obs::P2PMessage>(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<double>(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")