mirror of
https://github.com/XRPLF/rippled.git
synced 2026-07-31 19:10:25 +00:00
817c773162
Two gaps meant the last failure produced no evidence of its cause. The node-log upload was gated on `if: failure()`, but the validation step sets continue-on-error, so the job is not failing at that point and the condition never fired. Every failed run silently skipped the one artifact that records why a node did not reach consensus. It now keys on the validation step's own outcome, and also collects the harness logs. Transaction failures were logged at DEBUG, which CI does not enable, so a run where all 3052 submissions failed on a refused connection reported nothing about it. The first occurrence of each distinct failure kind is now a warning and repeats stay at DEBUG, so one refused connection says so once instead of 3052 times.
882 lines
27 KiB
Python
882 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Transaction Submitter for rippled telemetry validation.
|
|
|
|
Generates diverse transaction types against a rippled cluster to exercise
|
|
the full span and metric surface: tx.process, tx.apply, ledger.build,
|
|
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)
|
|
- OfferCreate / OfferCancel (DEX activity)
|
|
- TrustSet (trust line creation)
|
|
- NFTokenMint / NFTokenCreateOffer / NFTokenAcceptOffer
|
|
- EscrowCreate / EscrowFinish
|
|
- AMMCreate / AMMDeposit / AMMWithdraw (if amendment enabled)
|
|
|
|
Usage:
|
|
python3 tx_submitter.py --endpoint ws://localhost:6006 --tps 5 --duration 120
|
|
|
|
# Custom transaction mix:
|
|
python3 tx_submitter.py --endpoint ws://localhost:6006 \\
|
|
--weights '{"Payment":50,"OfferCreate":20,"TrustSet":10,"NFTokenMint":10,"EscrowCreate":10}'
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import random
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import websockets
|
|
|
|
logger = logging.getLogger("tx_submitter")
|
|
|
|
# Failure kinds already reported at WARNING, so a repeated failure is not
|
|
# logged again at that level.
|
|
_reported_failures: set[str] = set()
|
|
|
|
|
|
def _log_first_failure(key: str, fmt: str, *args: object) -> None:
|
|
"""Log a submission failure loudly the first time its kind is seen.
|
|
|
|
A run in which every transaction fails for one reason -- a refused
|
|
connection, an unfunded account -- used to leave no trace, because these
|
|
messages were DEBUG and CI does not enable DEBUG. Logging every failure
|
|
instead would bury the run in thousands of identical lines, so the first
|
|
of each distinct kind is a WARNING and the rest stay at DEBUG.
|
|
|
|
:param key: Identifies the failure kind, e.g. the engine result code.
|
|
:param fmt: Logging format string.
|
|
:param args: Format arguments.
|
|
"""
|
|
if key in _reported_failures:
|
|
logger.debug(fmt, *args)
|
|
return
|
|
_reported_failures.add(key)
|
|
logger.warning(fmt, *args)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
GENESIS_ACCOUNT = "rHb9CJAWyB4rj91VRWn96DkukG4bwdtyTh"
|
|
GENESIS_SEED = "snoPBrXtMeMyMHUVTgbuqAfg1SUTb"
|
|
|
|
# Amount to fund each test account (100,000 XRP in drops).
|
|
FUND_AMOUNT = "100000000000"
|
|
|
|
# Default transaction mix weights (relative).
|
|
DEFAULT_TX_WEIGHTS: dict[str, int] = {
|
|
"Payment": 40,
|
|
"OfferCreate": 15,
|
|
"OfferCancel": 5,
|
|
"TrustSet": 10,
|
|
"NFTokenMint": 10,
|
|
"NFTokenCreateOffer": 5,
|
|
"EscrowCreate": 5,
|
|
"EscrowFinish": 5,
|
|
"AMMCreate": 3,
|
|
"AMMDeposit": 2,
|
|
}
|
|
|
|
# Number of test accounts to create.
|
|
NUM_TEST_ACCOUNTS = 8
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data classes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Account:
|
|
"""Represents a funded XRPL test account.
|
|
|
|
Attributes:
|
|
name: Human-readable name (e.g., "alice").
|
|
account: Classic address (rXXX...).
|
|
seed: Secret seed for signing.
|
|
sequence: Next available sequence number.
|
|
"""
|
|
|
|
name: str
|
|
account: str
|
|
seed: str
|
|
sequence: int = 0
|
|
|
|
|
|
@dataclass
|
|
class TxStats:
|
|
"""Tracks transaction submission results.
|
|
|
|
Attributes:
|
|
total_submitted: Total transactions sent to the network.
|
|
total_success: Transactions that returned tesSUCCESS or terQUEUED.
|
|
total_errors: Transactions that returned an error engine_result.
|
|
by_type: Per-transaction-type count of submissions.
|
|
errors_by_type: Per-transaction-type count of errors.
|
|
"""
|
|
|
|
total_submitted: int = 0
|
|
total_success: int = 0
|
|
total_errors: int = 0
|
|
by_type: dict[str, int] = field(default_factory=dict)
|
|
errors_by_type: dict[str, int] = field(default_factory=dict)
|
|
|
|
def record(self, tx_type: str, success: bool) -> None:
|
|
"""Record the result of a transaction submission."""
|
|
self.total_submitted += 1
|
|
self.by_type[tx_type] = self.by_type.get(tx_type, 0) + 1
|
|
if success:
|
|
self.total_success += 1
|
|
else:
|
|
self.total_errors += 1
|
|
self.errors_by_type[tx_type] = self.errors_by_type.get(tx_type, 0) + 1
|
|
|
|
def summary(self) -> dict[str, Any]:
|
|
"""Return a summary dict suitable for JSON serialization."""
|
|
return {
|
|
"total_submitted": self.total_submitted,
|
|
"total_success": self.total_success,
|
|
"total_errors": self.total_errors,
|
|
"success_rate_pct": (
|
|
round(self.total_success / self.total_submitted * 100, 2)
|
|
if self.total_submitted
|
|
else 0
|
|
),
|
|
"by_type": self.by_type,
|
|
"errors_by_type": self.errors_by_type,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket RPC helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def ws_request(
|
|
ws: websockets.WebSocketClientProtocol,
|
|
command: str,
|
|
params: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Send a native WebSocket command and return the result payload.
|
|
|
|
Uses rippled's native WebSocket format (``command`` key with flat
|
|
parameters). The response has ``status`` at the top level and the
|
|
actual data payload inside ``result``. This helper unwraps the
|
|
``result`` dict so callers can read fields directly.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection.
|
|
command: RPC command name (e.g., ``account_info``, ``submit``).
|
|
params: Optional flat parameter dict merged into the request.
|
|
|
|
Returns:
|
|
The inner ``result`` dict from the response.
|
|
|
|
Raises:
|
|
RuntimeError: If the request fails or times out.
|
|
"""
|
|
request: dict[str, Any] = {"command": command}
|
|
if params:
|
|
request.update(params)
|
|
await ws.send(json.dumps(request))
|
|
raw = await asyncio.wait_for(ws.recv(), timeout=30.0)
|
|
resp = json.loads(raw)
|
|
|
|
# WS command format: {"status": "success", "result": {...}, "type": "response"}
|
|
# On error: {"status": "error", "error": "...", "error_message": "..."}
|
|
if resp.get("status") == "error":
|
|
logger.warning(
|
|
"%s error: %s — %s",
|
|
command,
|
|
resp.get("error", "unknown"),
|
|
resp.get("error_message", ""),
|
|
)
|
|
return resp.get("result", resp)
|
|
|
|
|
|
async def create_account(ws: websockets.WebSocketClientProtocol, name: str) -> Account:
|
|
"""Create a new account via wallet_propose RPC.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection.
|
|
name: Human-readable name for the account.
|
|
|
|
Returns:
|
|
An Account instance with the generated keypair.
|
|
"""
|
|
result = await ws_request(ws, "wallet_propose")
|
|
if "account_id" not in result:
|
|
raise RuntimeError(
|
|
f"wallet_propose failed: {json.dumps(result, indent=None)[:300]}"
|
|
)
|
|
return Account(
|
|
name=name,
|
|
account=result["account_id"],
|
|
seed=result["master_seed"],
|
|
)
|
|
|
|
|
|
async def fund_account(
|
|
ws: websockets.WebSocketClientProtocol,
|
|
dest: Account,
|
|
genesis_seq: int,
|
|
) -> tuple[bool, int]:
|
|
"""Fund a test account from genesis.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection.
|
|
dest: Destination account to fund.
|
|
genesis_seq: Current genesis account sequence number.
|
|
|
|
Returns:
|
|
Tuple of (success: bool, next_sequence: int).
|
|
"""
|
|
resp = await ws_request(
|
|
ws,
|
|
"submit",
|
|
{
|
|
"secret": GENESIS_SEED,
|
|
"tx_json": {
|
|
"TransactionType": "Payment",
|
|
"Account": GENESIS_ACCOUNT,
|
|
"Destination": dest.account,
|
|
"Amount": FUND_AMOUNT,
|
|
"Sequence": genesis_seq,
|
|
},
|
|
},
|
|
)
|
|
engine_result = resp.get("engine_result", "unknown")
|
|
success = engine_result in ("tesSUCCESS", "terQUEUED")
|
|
if not success:
|
|
# Log the full response to help diagnose submit failures in CI.
|
|
logger.warning(
|
|
"Fund %s failed: engine_result=%s, full response: %s",
|
|
dest.name,
|
|
engine_result,
|
|
json.dumps(resp, indent=None)[:500],
|
|
)
|
|
return success, genesis_seq + 1
|
|
|
|
|
|
async def get_account_sequence(
|
|
ws: websockets.WebSocketClientProtocol, account: str
|
|
) -> int:
|
|
"""Get the current sequence number for an account.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection.
|
|
account: Classic address.
|
|
|
|
Returns:
|
|
Current sequence number.
|
|
"""
|
|
resp = await ws_request(ws, "account_info", {"account": account})
|
|
if "account_data" not in resp:
|
|
# Log full response to diagnose WS API format issues.
|
|
logger.warning(
|
|
"account_info for %s: no account_data, full response: %s",
|
|
account[:12],
|
|
json.dumps(resp, indent=None)[:500],
|
|
)
|
|
return 0
|
|
return resp["account_data"].get("Sequence", 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transaction builders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_payment(sender: Account, receiver: Account) -> dict[str, Any]:
|
|
"""Build an XRP Payment transaction.
|
|
|
|
Args:
|
|
sender: Source account.
|
|
receiver: Destination account.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
amount = str(random.randint(1000, 1000000)) # 0.001 - 1 XRP
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "Payment",
|
|
"Account": sender.account,
|
|
"Destination": receiver.account,
|
|
"Amount": amount,
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_offer_create(sender: Account) -> dict[str, Any]:
|
|
"""Build an OfferCreate transaction (XRP/USD pair).
|
|
|
|
Args:
|
|
sender: Account placing the offer.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "OfferCreate",
|
|
"Account": sender.account,
|
|
"TakerPays": str(random.randint(100000, 10000000)),
|
|
"TakerGets": {
|
|
"currency": "USD",
|
|
"issuer": GENESIS_ACCOUNT,
|
|
"value": str(round(random.uniform(0.1, 100.0), 2)),
|
|
},
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_offer_cancel(sender: Account) -> dict[str, Any]:
|
|
"""Build an OfferCancel transaction.
|
|
|
|
Uses a non-existent offer sequence — will fail gracefully but still
|
|
exercises the tx.process span pipeline.
|
|
|
|
Args:
|
|
sender: Account cancelling the offer.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "OfferCancel",
|
|
"Account": sender.account,
|
|
"OfferSequence": max(1, sender.sequence - 1),
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_trust_set(sender: Account) -> dict[str, Any]:
|
|
"""Build a TrustSet transaction for a USD trust line.
|
|
|
|
Args:
|
|
sender: Account setting the trust line.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "TrustSet",
|
|
"Account": sender.account,
|
|
"LimitAmount": {
|
|
"currency": "USD",
|
|
"issuer": GENESIS_ACCOUNT,
|
|
"value": "1000000",
|
|
},
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_nftoken_mint(sender: Account) -> dict[str, Any]:
|
|
"""Build an NFTokenMint transaction.
|
|
|
|
Args:
|
|
sender: Account minting the NFT.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "NFTokenMint",
|
|
"Account": sender.account,
|
|
"NFTokenTaxon": random.randint(0, 100),
|
|
"Flags": 8, # tfTransferable
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_nftoken_create_offer(sender: Account) -> dict[str, Any]:
|
|
"""Build an NFTokenCreateOffer transaction.
|
|
|
|
Uses a dummy NFTokenID — will fail but exercises the span pipeline.
|
|
|
|
Args:
|
|
sender: Account creating the NFT offer.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "NFTokenCreateOffer",
|
|
"Account": sender.account,
|
|
"NFTokenID": "0" * 64,
|
|
"Amount": str(random.randint(100000, 1000000)),
|
|
"Flags": 1, # tfSellNFToken
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_escrow_create(sender: Account, receiver: Account) -> dict[str, Any]:
|
|
"""Build an EscrowCreate transaction.
|
|
|
|
Creates a time-based escrow that finishes 10 seconds from now.
|
|
|
|
Args:
|
|
sender: Account creating the escrow.
|
|
receiver: Destination account for escrow funds.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
# Ripple epoch offset: 946684800 seconds from Unix epoch
|
|
ripple_time = int(time.time()) - 946684800
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "EscrowCreate",
|
|
"Account": sender.account,
|
|
"Destination": receiver.account,
|
|
"Amount": str(random.randint(100000, 1000000)),
|
|
"FinishAfter": ripple_time + 10,
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_escrow_finish(sender: Account, owner: Account) -> dict[str, Any]:
|
|
"""Build an EscrowFinish transaction.
|
|
|
|
Uses a dummy offer sequence — will likely fail but exercises spans.
|
|
|
|
Args:
|
|
sender: Account finishing the escrow.
|
|
owner: Account that created the escrow.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "EscrowFinish",
|
|
"Account": sender.account,
|
|
"Owner": owner.account,
|
|
"OfferSequence": max(1, owner.sequence - 2),
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_amm_create(sender: Account) -> dict[str, Any]:
|
|
"""Build an AMMCreate transaction (XRP/USD pool).
|
|
|
|
Requires the AMM amendment to be enabled on the network.
|
|
|
|
Args:
|
|
sender: Account creating the AMM pool.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "AMMCreate",
|
|
"Account": sender.account,
|
|
"Amount": str(random.randint(10000000, 100000000)),
|
|
"Amount2": {
|
|
"currency": "USD",
|
|
"issuer": GENESIS_ACCOUNT,
|
|
"value": str(round(random.uniform(10.0, 1000.0), 2)),
|
|
},
|
|
"TradingFee": 500, # 0.5%
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
def build_amm_deposit(sender: Account) -> dict[str, Any]:
|
|
"""Build an AMMDeposit transaction.
|
|
|
|
Args:
|
|
sender: Account depositing into the AMM pool.
|
|
|
|
Returns:
|
|
Transaction JSON and signing secret.
|
|
"""
|
|
return {
|
|
"secret": sender.seed,
|
|
"tx_json": {
|
|
"TransactionType": "AMMDeposit",
|
|
"Account": sender.account,
|
|
"Asset": {"currency": "XRP"},
|
|
"Asset2": {
|
|
"currency": "USD",
|
|
"issuer": GENESIS_ACCOUNT,
|
|
},
|
|
"Amount": str(random.randint(1000000, 10000000)),
|
|
"Flags": 0x00080000, # tfSingleAsset
|
|
"Sequence": sender.sequence,
|
|
},
|
|
}
|
|
|
|
|
|
# Transaction type -> builder function mapping.
|
|
# Each builder takes (accounts: list[Account]) and returns submit params.
|
|
TX_BUILDERS: dict[str, Any] = {
|
|
"Payment": lambda accts: build_payment(accts[0], accts[1]),
|
|
"OfferCreate": lambda accts: build_offer_create(accts[0]),
|
|
"OfferCancel": lambda accts: build_offer_cancel(accts[0]),
|
|
"TrustSet": lambda accts: build_trust_set(accts[2]),
|
|
"NFTokenMint": lambda accts: build_nftoken_mint(accts[3]),
|
|
"NFTokenCreateOffer": lambda accts: build_nftoken_create_offer(accts[3]),
|
|
"EscrowCreate": lambda accts: build_escrow_create(accts[4], accts[1]),
|
|
"EscrowFinish": lambda accts: build_escrow_finish(accts[4], accts[4]),
|
|
"AMMCreate": lambda accts: build_amm_create(accts[5]),
|
|
"AMMDeposit": lambda accts: build_amm_deposit(accts[5]),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main submission loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def setup_accounts(
|
|
ws: websockets.WebSocketClientProtocol,
|
|
) -> list[Account]:
|
|
"""Create and fund test accounts from genesis.
|
|
|
|
Generates NUM_TEST_ACCOUNTS accounts via wallet_propose, then funds
|
|
each with FUND_AMOUNT XRP from genesis.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection to a rippled node.
|
|
|
|
Returns:
|
|
List of funded Account instances.
|
|
"""
|
|
account_names = ["alice", "bob", "carol", "dave", "eve", "frank", "grace", "heidi"]
|
|
|
|
logger.info("Creating %d test accounts...", NUM_TEST_ACCOUNTS)
|
|
accounts: list[Account] = []
|
|
for name in account_names[:NUM_TEST_ACCOUNTS]:
|
|
acct = await create_account(ws, name)
|
|
accounts.append(acct)
|
|
logger.info(" Created %s: %s", name, acct.account)
|
|
|
|
# Get genesis sequence.
|
|
genesis_seq = await get_account_sequence(ws, GENESIS_ACCOUNT)
|
|
logger.info("Genesis sequence: %d", genesis_seq)
|
|
|
|
# Fund all accounts.
|
|
logger.info("Funding test accounts...")
|
|
for acct in accounts:
|
|
success, genesis_seq = await fund_account(ws, acct, genesis_seq)
|
|
if success:
|
|
logger.info(" Funded %s", acct.name)
|
|
else:
|
|
logger.warning(" Failed to fund %s", acct.name)
|
|
|
|
# Wait for funding transactions to be validated.
|
|
logger.info("Waiting 10s for funding transactions to validate...")
|
|
await asyncio.sleep(10)
|
|
|
|
# Refresh sequence numbers for all accounts.
|
|
for acct in accounts:
|
|
try:
|
|
acct.sequence = await get_account_sequence(ws, acct.account)
|
|
logger.info(" %s sequence: %d", acct.name, acct.sequence)
|
|
except Exception as exc:
|
|
logger.warning(" Failed to get sequence for %s: %s", acct.name, exc)
|
|
|
|
return accounts
|
|
|
|
|
|
async def submit_transaction(
|
|
ws: websockets.WebSocketClientProtocol,
|
|
tx_type: str,
|
|
accounts: list[Account],
|
|
stats: TxStats,
|
|
) -> None:
|
|
"""Submit a single transaction of the given type.
|
|
|
|
Selects the appropriate builder, constructs the transaction, submits
|
|
it via the submit RPC, and records the result.
|
|
|
|
Args:
|
|
ws: Open WebSocket connection.
|
|
tx_type: Transaction type name (e.g., "Payment").
|
|
accounts: List of funded test accounts.
|
|
stats: TxStats instance to record results.
|
|
"""
|
|
builder = TX_BUILDERS.get(tx_type)
|
|
if not builder:
|
|
logger.warning("Unknown transaction type: %s", tx_type)
|
|
return
|
|
|
|
try:
|
|
params = builder(accounts)
|
|
# Identify which account is the sender to bump its sequence.
|
|
sender_addr = params["tx_json"]["Account"]
|
|
sender = next((a for a in accounts if a.account == sender_addr), None)
|
|
|
|
resp = await ws_request(ws, "submit", params)
|
|
engine_result = resp.get("engine_result", "unknown")
|
|
success = engine_result in (
|
|
"tesSUCCESS",
|
|
"terQUEUED",
|
|
"tecUNFUNDED_OFFER",
|
|
"tecNO_DST_INSUF_XRP",
|
|
)
|
|
stats.record(tx_type, success)
|
|
|
|
if sender:
|
|
sender.sequence += 1
|
|
|
|
if not success:
|
|
# First occurrence of each distinct result at WARNING, the rest at
|
|
# DEBUG. A run where every transaction failed previously produced
|
|
# no diagnostics at all, because DEBUG is off in CI; logging every
|
|
# failure instead would bury the run in thousands of identical
|
|
# lines.
|
|
_log_first_failure(
|
|
"result:%s" % engine_result,
|
|
"%s result: %s (%s)",
|
|
tx_type,
|
|
engine_result,
|
|
resp.get("engine_result_message", ""),
|
|
)
|
|
except Exception as exc:
|
|
stats.record(tx_type, False)
|
|
# Same reasoning, keyed on the exception type: a connection that is
|
|
# refused for the whole run says it once rather than per transaction.
|
|
_log_first_failure("exc:%s" % type(exc).__name__, "%s error: %s", tx_type, exc)
|
|
|
|
|
|
async def _refresh_sequences(
|
|
ws: websockets.WebSocketClientProtocol,
|
|
accounts: list[Account],
|
|
) -> None:
|
|
"""Re-sync account sequences from the validated ledger.
|
|
|
|
In a consensus network, other nodes' transactions advance sequences
|
|
beyond the submitter's local tracking. Refreshing every ~10 s keeps
|
|
the local counter close to the ledger and prevents tefPAST_SEQ storms.
|
|
"""
|
|
for acct in accounts:
|
|
try:
|
|
seq = await get_account_sequence(ws, acct.account)
|
|
if seq > acct.sequence:
|
|
acct.sequence = seq
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def run_submitter(
|
|
endpoint: str,
|
|
tps: float,
|
|
duration: float,
|
|
weights: dict[str, int],
|
|
) -> TxStats:
|
|
"""Run the transaction submitter against a single endpoint.
|
|
|
|
Args:
|
|
endpoint: WebSocket URL (ws://host:port).
|
|
tps: Target transactions per second.
|
|
duration: Total run time in seconds.
|
|
weights: Transaction type distribution weights.
|
|
|
|
Returns:
|
|
TxStats with aggregated results.
|
|
"""
|
|
stats = TxStats()
|
|
interval = 1.0 / tps if tps > 0 else 0.5
|
|
|
|
ws = await websockets.connect(endpoint, ping_interval=20, ping_timeout=10)
|
|
logger.info("Connected to %s", endpoint)
|
|
|
|
try:
|
|
# Setup test accounts.
|
|
accounts = await setup_accounts(ws)
|
|
if len(accounts) < 6:
|
|
logger.error("Need at least 6 funded accounts, got %d", len(accounts))
|
|
return stats
|
|
|
|
# Build weighted command list.
|
|
tx_types = list(weights.keys())
|
|
tx_weights = [weights[t] for t in tx_types]
|
|
|
|
logger.info(
|
|
"Starting TX submission: tps=%s, duration=%ss, types=%d",
|
|
tps,
|
|
duration,
|
|
len(tx_types),
|
|
)
|
|
|
|
start = time.monotonic()
|
|
last_seq_refresh = start
|
|
seq_refresh_interval = 10.0
|
|
while (time.monotonic() - start) < duration:
|
|
# Periodically re-sync account sequences from the ledger so
|
|
# locally-tracked sequences don't drift behind consensus.
|
|
if (time.monotonic() - last_seq_refresh) >= seq_refresh_interval:
|
|
await _refresh_sequences(ws, accounts)
|
|
last_seq_refresh = time.monotonic()
|
|
|
|
tx_type = random.choices(tx_types, weights=tx_weights, k=1)[0]
|
|
await submit_transaction(ws, tx_type, accounts, stats)
|
|
await asyncio.sleep(interval)
|
|
|
|
# Progress logging every 50 transactions.
|
|
if stats.total_submitted % 50 == 0 and stats.total_submitted > 0:
|
|
elapsed = time.monotonic() - start
|
|
actual_tps = stats.total_submitted / elapsed if elapsed > 0 else 0
|
|
logger.info(
|
|
"Progress: %d submitted, %d success, %d errors, "
|
|
"%.1f TPS (%.0fs elapsed)",
|
|
stats.total_submitted,
|
|
stats.total_success,
|
|
stats.total_errors,
|
|
actual_tps,
|
|
elapsed,
|
|
)
|
|
|
|
finally:
|
|
await ws.close()
|
|
|
|
elapsed = time.monotonic() - start
|
|
logger.info(
|
|
"Submission complete: %d submitted, %d success, %d errors "
|
|
"in %.1fs (%.1f TPS)",
|
|
stats.total_submitted,
|
|
stats.total_success,
|
|
stats.total_errors,
|
|
elapsed,
|
|
stats.total_submitted / elapsed if elapsed > 0 else 0,
|
|
)
|
|
|
|
return stats
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
"""Parse command-line arguments."""
|
|
parser = argparse.ArgumentParser(
|
|
description="Transaction Submitter for rippled telemetry validation",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Basic usage (5 TPS for 2 minutes):
|
|
python3 tx_submitter.py --endpoint ws://localhost:6006 --tps 5 --duration 120
|
|
|
|
# Custom transaction mix:
|
|
python3 tx_submitter.py --endpoint ws://localhost:6006 \\
|
|
--weights '{"Payment": 60, "OfferCreate": 20, "TrustSet": 20}'
|
|
""",
|
|
)
|
|
parser.add_argument(
|
|
"--endpoint",
|
|
type=str,
|
|
default="ws://localhost:6006",
|
|
help="WebSocket endpoint (default: ws://localhost:6006)",
|
|
)
|
|
parser.add_argument(
|
|
"--tps",
|
|
type=float,
|
|
default=5.0,
|
|
help="Target transactions per second (default: 5)",
|
|
)
|
|
parser.add_argument(
|
|
"--duration",
|
|
type=float,
|
|
default=120.0,
|
|
help="Run duration in seconds (default: 120)",
|
|
)
|
|
parser.add_argument(
|
|
"--weights",
|
|
type=str,
|
|
default=None,
|
|
help="JSON string of transaction type weights (overrides defaults)",
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
type=str,
|
|
default=None,
|
|
help="Write JSON summary to this file path",
|
|
)
|
|
parser.add_argument(
|
|
"--verbose",
|
|
action="store_true",
|
|
help="Enable debug logging",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
"""Main entry point for the transaction submitter."""
|
|
args = parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
|
|
)
|
|
|
|
# Parse custom weights if provided.
|
|
weights = DEFAULT_TX_WEIGHTS.copy()
|
|
if args.weights:
|
|
try:
|
|
custom = json.loads(args.weights)
|
|
weights = {k: int(v) for k, v in custom.items()}
|
|
logger.info("Using custom weights: %s", weights)
|
|
except (json.JSONDecodeError, ValueError) as exc:
|
|
logger.error("Invalid --weights JSON: %s", exc)
|
|
sys.exit(1)
|
|
|
|
# Run the submitter.
|
|
stats = asyncio.run(
|
|
run_submitter(
|
|
endpoint=args.endpoint,
|
|
tps=args.tps,
|
|
duration=args.duration,
|
|
weights=weights,
|
|
)
|
|
)
|
|
|
|
summary = stats.summary()
|
|
print(json.dumps(summary, indent=2))
|
|
|
|
if args.output:
|
|
with open(args.output, "w") as f:
|
|
json.dump(summary, f, indent=2)
|
|
logger.info("Summary written to %s", args.output)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|