test(testnet): probe blocked legacy RPC boundary

.testnet/scenarios/rollout/rolling_upgrade.py

@@ -26,6 +26,8 @@ Run: x-testnet --rippled-path @release suite \\
        .testnet/scenarios/rollout/rollout-suite.yml --stop-on-fail
 """
 
+import asyncio
+
 from helpers import CONSENSUS_ENTROPY_FEATURE
 
 VALIDATORS = [0, 1, 2, 3, 4]
@@ -36,6 +38,7 @@ ALL_NODES = VALIDATORS + [UPGRADED_TRACKER, STRAGGLER_TRACKER]
 MESH_MIN = 2
 PROTOCOL = "XRPL/2.2"
 CONSENSUS_ENTROPY_CAPABILITY = "xahau-consensus-entropy"
+GENESIS_ACCOUNT = "rHb9CJAWyB4rj91VRWn96DkukG4bwdtyTh"
 RING_EDGES = {(nid, (nid + 1) % len(ALL_NODES)) for nid in ALL_NODES}
 
 
@@ -136,6 +139,128 @@ async def _restore_ring(ctx, log, *, timeout=90):
     log(f"ring topology restored ({len(RING_EDGES)} directed edges)")
 
 
+def _rpc_summary(result):
+    """Return a compact, log-safe summary of an RPC result."""
+    if result is None:
+        return "no response"
+    if result.get("error"):
+        return f"error={result.get('error')} message={result.get('error_message')}"
+
+    ledger = result.get("ledger") or {}
+    transactions = ledger.get("transactions") or result.get("transactions") or []
+    state = result.get("state") or []
+    tx_types = []
+    for tx in transactions:
+        if isinstance(tx, dict):
+            tx_types.append(str(tx.get("TransactionType", "<unnamed>")))
+        else:
+            tx_types.append("hash")
+    entry_types = sorted(
+        {
+            str(entry.get("LedgerEntryType", "<unnamed>"))
+            for entry in state
+            if isinstance(entry, dict)
+        }
+    )
+    details = ["success"]
+    if ledger:
+        details.append(f"ledger={ledger.get('ledger_index')}")
+    if transactions:
+        details.append(f"txs={len(transactions)} types={tx_types}")
+    if state:
+        details.append(f"state={len(state)} types={entry_types}")
+    if result.get("account_data"):
+        details.append(f"account_seq={result['account_data'].get('Sequence')}")
+    if result.get("marker"):
+        details.append("marker=yes")
+    return "; ".join(details)
+
+
+async def _probe_straggler_rpc(ctx, log, *, ledger_index):
+    """Exercise old-node read APIs after it becomes amendment-blocked.
+
+    Expanded transaction and JSON ledger-data requests intentionally force the
+    old binary to decode objects containing post-upgrade transaction/SLE fields.
+    After every request, independently check that the process still answers RPC.
+    """
+
+    probes = [
+        (
+            "account_info",
+            "account_info",
+            {"account": GENESIS_ACCOUNT, "ledger_index": ledger_index},
+        ),
+        (
+            "ledger header",
+            "ledger",
+            {
+                "ledger_index": ledger_index,
+                "transactions": False,
+                "expand": False,
+            },
+        ),
+        (
+            "ledger transaction hashes",
+            "ledger",
+            {
+                "ledger_index": ledger_index,
+                "transactions": True,
+                "expand": False,
+            },
+        ),
+        (
+            "ledger fully expanded transactions",
+            "ledger",
+            {
+                "ledger_index": ledger_index,
+                "transactions": True,
+                "expand": True,
+            },
+        ),
+        (
+            "ledger_data binary",
+            "ledger_data",
+            {"ledger_index": ledger_index, "binary": True, "limit": 256},
+        ),
+        (
+            "ledger_data JSON",
+            "ledger_data",
+            {"ledger_index": ledger_index, "binary": False, "limit": 256},
+        ),
+    ]
+
+    log(f"probing amendment-blocked n{STRAGGLER_TRACKER} at ledger {ledger_index}")
+    for label, method, params in probes:
+        result = ctx.rpc.request(STRAGGLER_TRACKER, method, params)
+        # Catch a delayed abort after the HTTP response has already been sent.
+        await asyncio.sleep(0.5)
+        alive = bool(ctx.rpc.server_info(STRAGGLER_TRACKER))
+        log(f"n{STRAGGLER_TRACKER} RPC {label}: {_rpc_summary(result)}; alive={alive}")
+        assert alive, f"n{STRAGGLER_TRACKER} crashed after RPC probe: {label}"
+
+
+def _report_straggler_proposals(ctx, log, *, since):
+    """Copy focused old-node proposal handling evidence into scenario output."""
+    proposals = ctx.search_logs(
+        r"Proposal:", since=since, nodes=[STRAGGLER_TRACKER], limit=500
+    )
+    rejected = [
+        entry
+        for entry in proposals.matches
+        if any(
+            token in entry.line.lower()
+            for token in ("drop", "malformed", "disabled", "reject")
+        )
+    ]
+    log(
+        f"n{STRAGGLER_TRACKER} Protocol trace captured {proposals.count} proposal "
+        f"events since vote, including {len(rejected)} drop/reject events"
+    )
+    selected = rejected if rejected else proposals.matches[-20:]
+    for entry in selected[:50]:
+        log(f"n{STRAGGLER_TRACKER} proposal trace: {entry.line}")
+
+
 async def scenario(ctx, log):
     # 1. all-old net healthy AND the ring has formed before we touch it
     await _restore_ring(ctx, log)
@@ -173,6 +298,11 @@ async def scenario(ctx, log):
     )
 
     # 3. vote CE up on the validators (real vote; the seed only pre-satisfied the hold)
+    assert ctx.rpc.log_level(STRAGGLER_TRACKER, "Protocol", "trace"), (
+        f"failed to enable Protocol trace on n{STRAGGLER_TRACKER}"
+    )
+    proposal_trace_start = ctx.mark("straggler-protocol-trace")
+    log(f"enabled Protocol trace on old straggler n{STRAGGLER_TRACKER}")
     ctx.feature(CONSENSUS_ENTROPY_FEATURE, vetoed=False, nodes=VALIDATORS)
     log("voted ConsensusEntropy accept on n0-n4; crossing the flag ledger...")
 
@@ -199,7 +329,14 @@ async def scenario(ctx, log):
     assert not _blocked(ctx, UPGRADED_TRACKER), (
         f"n{UPGRADED_TRACKER} (upgraded tracker) should NOT be amendment-blocked"
     )
-    await ctx.wait_for_ledgers(2, node_id=UPGRADED_TRACKER, timeout=120)
+    probe_ledger = (
+        await ctx.wait_for_ledgers(2, node_id=UPGRADED_TRACKER, timeout=120)
+    ).result
+    await ctx.wait_for_ledger(probe_ledger, node_id=STRAGGLER_TRACKER, timeout=60)
+    await _probe_straggler_rpc(ctx, log, ledger_index=probe_ledger)
+    _report_straggler_proposals(ctx, log, since=proposal_trace_start)
+    snapshot = ctx._network.snapshot("ce-old-straggler-probe", keep_db=False)
+    log(f"captured old-straggler Protocol trace and logs at {snapshot}")
     for nid in (UPGRADED_TRACKER, STRAGGLER_TRACKER):
         assert ctx.rpc.server_info(nid), f"n{nid} RPC down (crashed?)"
     log(