fix(telemetry): correlate WebSocket replies, and stop silent placeholder data

Eight defects found in review of PR 6519. Twenty-four review threads reported
them; ten were duplicates of one another and four were wrong about the code.

The two that corrupt data. Both WebSocket clients reuse the socket after a recv
timeout, and the library queues the late reply, so the NEXT request reads the
previous response. In rpc_load_generator that misattributes latency, and the
skew is permanent rather than one-off. In tx_submitter it is worse: a submit
that reads an account_info reply freezes that account's sequence number and
every later transaction for it fails. Both now correlate replies by request id
under a single overall deadline, with a counter rather than a wall clock, since
time.time() is not monotonic and collides within a tick.

workload_orchestrator never cleared its fixed report paths, so a run that
produced no report silently adopted the previous run's totals -- breaking the
invariant evaluate_exit_gate documents. Reproduced by planting a stale total
and watching it appear in a later summary.

collect_system_metrics reported placeholders as if they were measurements. The
consensus mean used bc with a || echo 0 fallback that neither warned nor
cleared METRICS_COMPLETE, unlike every sibling path; it now uses awk, already a
hard dependency here, which removes the failure mode instead of reporting it.
Note this moves the mean from truncation to rounding, at most 1 ms on a value
of about 45 s. Unmeasurable TPS now warns and clears the flag too. All four
curl probes gained a timeout, not just the one the review named -- an
unresponsive node could hang any of them.

The orchestrator's help text claimed 18-dashboard coverage; there are 15 on
disk, 15 uids in the contract, and the profile already said 15. The
tx_submitter docstring listed twelve transaction types where ten exist, and
claimed issued-currency payments that build_payment never sends.

Two suggested patches were deliberately not taken. A recursive delete of the report
parent sits in a per-task function and would delete earlier phases' reports
mid-run, and recursively remove a caller-supplied --report-dir.

The stale microsecond axis label on ledger-data-sync is real but belongs to
phase 9, which carries a byte-identical copy of that dashboard, so fixing it
here would leave that PR wrong and guarantee a conflict.

docker/telemetry/workload/collect_system_metrics.sh

@@ -68,6 +68,17 @@ OUTPUT_FILE="$3"
 IFS=',' read -ra RPC_PORTS <<<"$RPC_PORTS_CSV"
 SAMPLE_INTERVAL=5
 
+# Hard ceiling on every RPC probe below. curl applies no overall timeout of its
+# own, so a node that accepts the connection and then stops answering — what a
+# stalled job queue looks like from outside — parks the sampling loop for the
+# rest of the run. 5 s is one sample interval and some thousands of times a
+# healthy server_info, so it bounds a wedged node's cost to one lost sample
+# while never truncating a real reply. A probe that hits the ceiling exits
+# non-zero and is therefore skipped rather than recorded, which is the same
+# rule the latency loop already applies to a refused connection; if every
+# probe hits it, the empty file trips the placeholder warning below.
+CURL_MAX_TIME=5
+
 # Reject anything the sample arithmetic cannot use, instead of silently
 # treating it as 0.
 case "$DURATION" in
@@ -136,7 +147,7 @@ trap cleanup EXIT
 INITIAL_SEQ=0
 INITIAL_TIME=$(date +%s)
 for port in "${RPC_PORTS[@]}"; do
-    seq=$(curl -sf "http://localhost:$port" \
+    seq=$(curl -sf --max-time "$CURL_MAX_TIME" "http://localhost:$port" \
         -d '{"method":"server_info"}' 2>/dev/null |
         jq -r '.result.info.validated_ledger.seq // 0' 2>/dev/null || echo 0)
     if [ "$seq" -gt "$INITIAL_SEQ" ]; then
@@ -185,7 +196,7 @@ for sample in $(seq 1 "$SAMPLES"); do
     # and recording that as ~0 ms would pull the reported p99 down.
     for port in "${RPC_PORTS[@]}"; do
         start_ns=$(now_ns)
-        if curl -sf "http://localhost:$port" \
+        if curl -sf --max-time "$CURL_MAX_TIME" "http://localhost:$port" \
             -d '{"method":"server_info"}' >/dev/null 2>&1; then
             end_ns=$(now_ns)
             latency_ms=$(((end_ns - start_ns) / 1000000))
@@ -195,7 +206,7 @@ for sample in $(seq 1 "$SAMPLES"); do
 
     # Record current validated ledger seq.
     for port in "${RPC_PORTS[@]}"; do
-        seq=$(curl -sf "http://localhost:$port" \
+        seq=$(curl -sf --max-time "$CURL_MAX_TIME" "http://localhost:$port" \
             -d '{"method":"server_info"}' 2>/dev/null |
             jq -r '.result.info.validated_ledger.seq // 0' 2>/dev/null || echo 0)
         echo "$seq" >>"$LEDGER_FILE"
@@ -269,7 +280,7 @@ fi
 # TPS calculation from ledger sequence advancement.
 FINAL_SEQ=0
 for port in "${RPC_PORTS[@]}"; do
-    seq=$(curl -sf "http://localhost:$port" \
+    seq=$(curl -sf --max-time "$CURL_MAX_TIME" "http://localhost:$port" \
         -d '{"method":"server_info"}' 2>/dev/null |
         jq -r '.result.info.validated_ledger.seq // 0' 2>/dev/null || echo 0)
     if [ "$seq" -gt "$FINAL_SEQ" ]; then
@@ -291,7 +302,12 @@ if [ "$ELAPSED" -gt 0 ] && [ "$LEDGER_ADVANCE" -gt 0 ]; then
     # but a strict parser rejects the whole file. awk's %.2f always pads.
     TPS=$(awk -v a="$LEDGER_ADVANCE" -v b="$ELAPSED" 'BEGIN { printf "%.2f", a / b }')
 else
+    # No ledger advance means the cluster produced nothing to divide, so 0 here
+    # is an absence of data, not a measured rate. Flagged like every other empty
+    # source so the run reports itself inconclusive instead of a real 0 TPS.
+    warn "Ledger seq did not advance (advance=$LEDGER_ADVANCE over ${ELAPSED}s); tps is a 0 placeholder"
     TPS="0"
+    METRICS_COMPLETE=false
 fi
 
 # Mean inter-ledger interval in ms: DURATION / (distinct ledgers - 1) * 1000.
@@ -304,7 +320,16 @@ if [ -s "$LEDGER_FILE" ]; then
     UNIQUE_LEDGERS=$(sort -u "$LEDGER_FILE" | wc -l)
     # The > 1 test also keeps the divisor below at 1 or more.
     if [ "$UNIQUE_LEDGERS" -gt 1 ]; then
-        CONSENSUS_MEAN=$(echo "scale=0; $DURATION * 1000 / ($UNIQUE_LEDGERS - 1)" | bc 2>/dev/null || echo "0")
+        # awk rather than bc, for the same reason as the TPS calculation above.
+        # bc is an optional package, and the `|| echo "0"` this replaces turned
+        # a missing or failing bc into a 0 that read exactly like a measured
+        # value: no warn(), and METRICS_COMPLETE left true, so the run reported
+        # a complete measurement and exited 0. awk is already required by the
+        # sampling loop and the CPU average, so computing this with awk removes
+        # the failure path rather than reporting it. The divisor is >= 1 by the
+        # test above.
+        CONSENSUS_MEAN=$(awk -v d="$DURATION" -v u="$UNIQUE_LEDGERS" \
+            'BEGIN { printf "%.0f", d * 1000 / (u - 1) }')
     else
         warn "Ledger seq never advanced ($UNIQUE_LEDGERS distinct); consensus_round_mean_ms is a 0 placeholder"
         CONSENSUS_MEAN="0"

docker/telemetry/workload/rpc_load_generator.py

@@ -30,6 +30,7 @@ Usage:
 
 import argparse
 import asyncio
+import itertools
 import json
 import logging
 import math
@@ -90,10 +91,17 @@ DRAIN_TIMEOUT_S = RECV_TIMEOUT_S + 2.0
 
 # Requests allowed to own one connection's recv() at a time. websockets
 # rejects a second concurrent recv() on the same socket, so this must stay 1
-# unless request/response correlation by id is added. Concurrency comes from
-# spreading requests round-robin over the endpoints instead.
+# until a single reader task per connection demultiplexes replies to their
+# waiters; correlating by id, which send_rpc now does, is necessary for that
+# but not sufficient on its own. Concurrency comes from spreading requests
+# round-robin over the endpoints instead.
 MAX_INFLIGHT_PER_CONNECTION = 1
 
+# Source of the ``id`` sent with every request. xrpld echoes a request's id at
+# the top level of the reply, which is what lets send_rpc tell its own reply
+# from one that an earlier, timed-out request left in the receive buffer.
+_request_ids = itertools.count(1)
+
 # Fraction of dispatched requests that must reach the server for a run to
 # count as a measurement. The gate is one connection deep, so the ceiling is
 # len(connections) / round-trip requests per second; asking for more than
@@ -364,6 +372,59 @@ def choose_command(weights: dict[str, int]) -> str:
 # ---------------------------------------------------------------------------
 
 
+async def _recv_matching_reply(
+    conn: Connection, request_id: int, command: str
+) -> dict[str, Any]:
+    """Read replies on ``conn`` until the one for ``request_id`` arrives.
+
+    Cancelling a ``recv()`` does not discard the reply it was waiting for: the
+    library queues incoming messages independently of any reader, so a request
+    that hit RECV_TIMEOUT_S leaves its reply in the buffer. The next request on
+    that connection used to read it and time its own round trip against
+    somebody else's reply -- a near-zero latency, and a ``status`` belonging to
+    a different command. The skew was permanent, because every later request on
+    the connection stayed one reply behind for the rest of the run, so a single
+    timeout quietly invalidated that connection's whole latency distribution.
+    Discarding replies by id puts the stream back in step.
+
+    The caller holds ``conn.gate``, so no other request can consume a reply
+    while this loop runs. The deadline covers the whole exchange rather than
+    each message, so a run of buffered replies cannot extend the wait without
+    bound.
+
+    Args:
+        conn:       Connection to read from, with its gate already held.
+        request_id: The ``id`` sent with this request.
+        command:    RPC command name, for logging.
+
+    Returns:
+        The parsed reply whose ``id`` matches, or that carries no ``id``.
+
+    Raises:
+        asyncio.TimeoutError: If no matching reply arrived within
+                              RECV_TIMEOUT_S.
+    """
+    deadline = time.monotonic() + RECV_TIMEOUT_S
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise asyncio.TimeoutError(f"{command} (id {request_id}) got no reply")
+        raw = await asyncio.wait_for(conn.ws.recv(), timeout=remaining)
+        reply = json.loads(raw)
+        # A reply with no id counts as this request's: a few xrpld error paths
+        # answer before the id is parsed, and treating those as stale would
+        # turn a reported error into a timeout.
+        reply_id = reply.get("id")
+        if reply_id is None or reply_id == request_id:
+            return reply
+        logger.debug(
+            "Discarded a reply for id %s while awaiting id %s (%s)",
+            reply_id,
+            request_id,
+            command,
+        )
+
+
 async def send_rpc(
     conn: Connection,
     command: str,
@@ -388,6 +449,8 @@ async def send_rpc(
                              to the request for context propagation testing.
     """
     request = build_rpc_request(command)
+    request_id = next(_request_ids)
+    request["id"] = request_id
 
     # Inject W3C traceparent for context propagation testing.
     # The rippled WebSocket handler extracts this from the JSON body
@@ -404,11 +467,11 @@ async def send_rpc(
         # being counted as one more failed request.
         try:
             await conn.ws.send(json.dumps(request))
-            raw = await asyncio.wait_for(conn.ws.recv(), timeout=RECV_TIMEOUT_S)
+            reply = await _recv_matching_reply(conn, request_id, command)
             latency = time.monotonic() - t0
             # Native WS responses have {"status": "success", "result": {...}}
             # or {"status": "error", "error": "...", "error_message": "..."}.
-            success = json.loads(raw).get("status") == "success"
+            success = reply.get("status") == "success"
         except REQUEST_FAILURES as exc:
             logger.debug("RPC %s failed: %s", command, exc)
             # No reply, so no latency sample -- see LoadStats.record().

docker/telemetry/workload/tx_submitter.py

@@ -8,13 +8,13 @@ consensus.*, and all associated attributes.
 Pre-funds test accounts from the genesis account, then submits a
 configurable mix of transaction types at a target TPS.
 
-Supported transaction types:
-  - Payment (XRP and issued currencies)
+Supported transaction types (the 10 in TX_BUILDERS):
+  - Payment (XRP transfers)
   - OfferCreate / OfferCancel (DEX activity)
   - TrustSet (trust line creation)
-  - NFTokenMint / NFTokenCreateOffer / NFTokenAcceptOffer
-  - EscrowCreate / EscrowFinish
-  - AMMCreate / AMMDeposit / AMMWithdraw (if amendment enabled)
+  - NFTokenMint / NFTokenCreateOffer (NFT activity)
+  - EscrowCreate / EscrowFinish (escrow lifecycle)
+  - AMMCreate / AMMDeposit (AMM pool operations, if amendment enabled)
 
 Usage:
     python3 tx_submitter.py --endpoint ws://localhost:6006 --tps 5 --duration 120
@@ -26,6 +26,7 @@ Usage:
 
 import argparse
 import asyncio
+import itertools
 import json
 import logging
 import random
@@ -111,6 +112,14 @@ SEQ_REFETCH_AFTER_FAILURES = 5
 # ledger, to stay close to sequences other submitters have advanced.
 SEQ_REFRESH_INTERVAL_S = 10.0
 
+# How long one request waits for the reply that belongs to it.
+RECV_TIMEOUT_S = 30.0
+
+# Source of the ``id`` sent with every request. xrpld echoes a request's id at
+# the top level of the reply, which is what lets ws_request tell its own reply
+# from one that an earlier, timed-out request left in the receive buffer.
+_request_ids = itertools.count(1)
+
 
 def consumes_sequence(engine_result: str | None) -> bool:
     """Report whether an engine result tied up the submitted sequence number.
@@ -243,14 +252,49 @@ async def ws_request(
         The inner ``result`` dict from the response.
 
     Raises:
-        RuntimeError: If the request fails or times out.
+        TimeoutError: If no reply carrying this request's ``id`` arrived within
+                      RECV_TIMEOUT_S.
     """
+    request_id = next(_request_ids)
     request: dict[str, Any] = {"command": command}
     if params:
         request.update(params)
+    # Set after the merge so a caller's parameter can never shadow the id.
+    request["id"] = request_id
     await ws.send(json.dumps(request))
-    raw = await asyncio.wait_for(ws.recv(), timeout=30.0)
-    resp = json.loads(raw)
+
+    # Read until the reply carrying this request's id turns up. Cancelling a
+    # recv() does not discard the reply it was waiting for: the library queues
+    # incoming messages independently of any reader, so a request that timed
+    # out leaves its reply in the buffer and the NEXT request used to read it
+    # instead of its own. That mis-attribution is permanent, because every
+    # later request on the connection stays one reply behind: a submit would
+    # read an account_info reply, see no engine result, and stop advancing the
+    # account's sequence, failing every remaining transaction for a reason
+    # nothing logged. Discarding by id puts the stream back in step.
+    #
+    # The deadline is for the whole exchange rather than per message, so a run
+    # of buffered replies cannot extend the wait without bound. A reply with no
+    # id is accepted as this request's: a few xrpld error paths answer before
+    # the id is parsed, and treating those as stale would hide a real error.
+    deadline = time.monotonic() + RECV_TIMEOUT_S
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError(f"{command} (id {request_id}) got no reply")
+        raw = await asyncio.wait_for(ws.recv(), timeout=remaining)
+        resp = json.loads(raw)
+        reply_id = resp.get("id")
+        if reply_id is None or reply_id == request_id:
+            break
+        _log_first_failure(
+            "stale-reply",
+            "Discarded a reply for id %s while awaiting id %s (%s); an earlier "
+            "request timed out and left it buffered",
+            reply_id,
+            request_id,
+            command,
+        )
 
     # WS command format: {"status": "success", "result": {...}, "type": "response"}
     # On error: {"status": "error", "error": "...", "error_message": "..."}

docker/telemetry/workload/workload_orchestrator.py

@@ -356,6 +356,15 @@ def _launch_phase_tasks(
     Each generator is given the phase duration plus SUBPROCESS_GRACE_SEC, so a
     wedged one is killed instead of stalling the phase.
 
+    Any report left at a path this phase is about to use is deleted first.
+    report_dir defaults to a fixed location and the filenames are derived from
+    the phase index and name, so consecutive runs of a profile reuse the same
+    paths. Without the delete, a generator that produced no report this run
+    left the previous run's file in place, and _collect_task_result read it as
+    this run's result — which also breaks the assumption evaluate_exit_gate
+    documents, that a crashed generator leaves its totals at 0. Afterwards a
+    present file always belongs to this run and a missing one is unambiguous.
+
     Args:
         phase:      Phase dict from the profile.
         endpoints:  List of WebSocket endpoint URLs.
@@ -374,6 +383,7 @@ def _launch_phase_tasks(
     rpc_cfg = phase.get("rpc")
     if rpc_cfg:
         rpc_out = report_dir / f"{prefix}-rpc.json"
+        rpc_out.unlink(missing_ok=True)
         cmd = _build_rpc_cmd(endpoints, rpc_cfg, duration, rpc_out)
         task = asyncio.create_task(run_subprocess(cmd, f"RPC [{name}]", timeout))
         tasks.append(("rpc", rpc_out, task))
@@ -381,6 +391,7 @@ def _launch_phase_tasks(
     tx_cfg = phase.get("tx")
     if tx_cfg:
         tx_out = report_dir / f"{prefix}-tx.json"
+        tx_out.unlink(missing_ok=True)
         cmd = _build_tx_cmd(endpoints[0], tx_cfg, duration, tx_out)
         task = asyncio.create_task(run_subprocess(cmd, f"TX [{name}]", timeout))
         tasks.append(("tx", tx_out, task))
@@ -583,7 +594,7 @@ def parse_args() -> argparse.Namespace:
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog="""
 Profiles:
-  full-validation  Full 18-dashboard coverage (~5 min load + 1 min propagation)
+  full-validation  Full 15-dashboard coverage (~5 min load + 1 min propagation)
   quick-smoke      Fast CI smoke test (~30s load + 30s propagation)
   stress           Heavy sustained load for benchmarking (~3.5 min + 1 min)