mirror of
https://github.com/Xahau/xahaud.git
synced 2026-07-23 15:10:10 +00:00
test(export): cover share stream and signer recovery
This commit is contained in:
Nicholas Dudfield
@@ -53,6 +53,12 @@ tests:
|
||||
- rng_poll_ms=333
|
||||
- n4:no_export_sig_hash=true
|
||||
|
||||
- name: export_share_subscription
|
||||
script: .testnet/scenarios/export/export_share_subscription.py
|
||||
|
||||
- name: export_two_member_committee_recovery
|
||||
script: .testnet/scenarios/export/export_committee_recovery.py
|
||||
|
||||
# CE + Export: 1 node suppressed, 4/5 = 80% quorum, should succeed
|
||||
- name: export_ce_one_node_down
|
||||
script: .testnet/scenarios/export/export_quorum.py
|
||||
|
||||
126
.testnet/scenarios/export/export_committee_recovery.py
Normal file
126
.testnet/scenarios/export/export_committee_recovery.py
Normal file
@@ -0,0 +1,126 @@
|
||||
""":descr: a 2-of-5 Export committee recovers without a new intent
|
||||
|
||||
The intent selects only validators n0 and n4, so qC is 2. Validator n4 is
|
||||
stopped before admission: the network still validates the intent with 4/5, but
|
||||
one selected share cannot form a witness. Restarting n4 must republish its share
|
||||
for the same live latch and complete the witness without resubmitting Export.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from export_helpers import (
|
||||
EXPORT_RETRY_LEDGER_WINDOW,
|
||||
assert_export_latch,
|
||||
bitmap_positions,
|
||||
find_export_signature_witness,
|
||||
find_export_txns,
|
||||
require_export,
|
||||
submit_direct_export,
|
||||
wait_for_export_signature_witness,
|
||||
)
|
||||
|
||||
|
||||
async def scenario(ctx, log):
|
||||
await require_export(ctx, log)
|
||||
await ctx.fund_accounts({"alice": 10000, "bob": 1000})
|
||||
alice = ctx.account("alice")
|
||||
bob = ctx.account("bob")
|
||||
|
||||
if not ctx.stop_node(4):
|
||||
raise AssertionError("Failed to stop selected validator n4")
|
||||
await ctx.wait_for_nodes_down(nodes=[4], timeout=30)
|
||||
log("Stopped selected validator n4; selected committee is n0+n4")
|
||||
|
||||
current = ctx.validated_ledger_index(0)
|
||||
result = await submit_direct_export(
|
||||
ctx,
|
||||
log,
|
||||
{
|
||||
"TransactionType": "Export",
|
||||
"Fee": "1000000",
|
||||
"ExportedTxn": {
|
||||
"TransactionType": "Payment",
|
||||
"Account": alice.address,
|
||||
"Destination": bob.address,
|
||||
"Amount": "1000000",
|
||||
"Fee": "10",
|
||||
"Sequence": 0,
|
||||
"TicketSequence": 1,
|
||||
"FirstLedgerSequence": current + 1,
|
||||
"LastLedgerSequence": current + EXPORT_RETRY_LEDGER_WINDOW,
|
||||
"Flags": 2147483648,
|
||||
"SigningPubKey": "",
|
||||
},
|
||||
},
|
||||
alice.wallet,
|
||||
committee_node_ids=[0, 4],
|
||||
)
|
||||
if result.get("engine_result") != "tesSUCCESS":
|
||||
raise AssertionError(f"Export intent failed: {result}")
|
||||
|
||||
origin = result.get("hash")
|
||||
origin_seq = int(result.get("ledger_index"))
|
||||
if not origin:
|
||||
raise AssertionError(f"Validated Export missing hash: {result}")
|
||||
|
||||
latches = assert_export_latch(
|
||||
ctx,
|
||||
alice.address,
|
||||
log,
|
||||
origin_hash=origin,
|
||||
expect_witness=False,
|
||||
)
|
||||
selected = bitmap_positions(latches[0]["ExportCommittee"])
|
||||
if len(selected) != 2:
|
||||
raise AssertionError(f"Expected a 2-member committee, got {selected}")
|
||||
|
||||
await ctx.wait_for_ledger(origin_seq + 1, node_id=0, timeout=30)
|
||||
if find_export_signature_witness(ctx, origin_seq + 1, origin):
|
||||
raise AssertionError("Witness formed while one of two selected signers was down")
|
||||
assert_export_latch(
|
||||
ctx,
|
||||
alice.address,
|
||||
log,
|
||||
origin_hash=origin,
|
||||
expect_witness=False,
|
||||
)
|
||||
log("No witness with only one selected signer; original latch remains pending")
|
||||
|
||||
if not ctx.start_node(4):
|
||||
raise AssertionError("Failed to restart selected validator n4")
|
||||
await ctx.wait_for_nodes(
|
||||
lambda node_id: ctx.rpc.server_info(node_id) is not None,
|
||||
nodes=[4],
|
||||
timeout=30,
|
||||
poll_interval=0.5,
|
||||
name="selected-export-validator-up",
|
||||
)
|
||||
log("Restarted selected validator n4")
|
||||
|
||||
witness = await wait_for_export_signature_witness(
|
||||
ctx, log, origin, after_ledger=origin_seq
|
||||
)
|
||||
contributors = bitmap_positions(witness["EntropyContributors"])
|
||||
if contributors != selected:
|
||||
raise AssertionError(
|
||||
f"Recovered witness contributors {contributors} != selected {selected}"
|
||||
)
|
||||
if len(witness["_WitnessSigners"]) != 2:
|
||||
raise AssertionError(
|
||||
f"Recovered witness has {len(witness['_WitnessSigners'])} signers, need 2"
|
||||
)
|
||||
witness_seq = int(witness["LedgerSequence"])
|
||||
for seq in range(origin_seq + 1, witness_seq + 1):
|
||||
if any(tx.get("Account") == alice.address for tx in find_export_txns(ctx, seq)):
|
||||
raise AssertionError(
|
||||
f"A second Export was submitted before recovery in ledger {seq}"
|
||||
)
|
||||
assert_export_latch(
|
||||
ctx,
|
||||
alice.address,
|
||||
log,
|
||||
origin_hash=origin,
|
||||
expect_witness=True,
|
||||
)
|
||||
log(f"Same Export {origin} completed after selected validator recovery")
|
||||
log("PASS")
|
||||
@@ -2,7 +2,13 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from xahaud_scripts.testnet.config import _unl_report_index, feature_name_to_hash
|
||||
import json
|
||||
|
||||
from xahaud_scripts.testnet.config import (
|
||||
_decode_node_public_key,
|
||||
_unl_report_index,
|
||||
feature_name_to_hash,
|
||||
)
|
||||
|
||||
EXPORT_RETRY_LEDGER_WINDOW = 5
|
||||
EXPORT_PUBLICATION_LEDGER_WINDOW = 5
|
||||
@@ -65,7 +71,29 @@ def find_export_txns(ctx, seq):
|
||||
return [tx for tx in txns if tx.get("TransactionType") == "Export"]
|
||||
|
||||
|
||||
def export_authority(ctx, *, require_unl_report=True):
|
||||
def _validator_master_keys_by_node(ctx):
|
||||
"""Return generated validator master keys keyed by testnet node id."""
|
||||
network = json.loads((ctx.base_dir / "network.json").read_text())
|
||||
return {
|
||||
int(node["id"]): _decode_node_public_key(node["public_key"])
|
||||
for node in network["nodes"]
|
||||
}
|
||||
|
||||
|
||||
def bitmap_positions(bitmap):
|
||||
"""Return the selected bit positions from a hex bitmap or bytes."""
|
||||
raw = bytes.fromhex(bitmap) if isinstance(bitmap, str) else bytes(bitmap)
|
||||
return {
|
||||
byte_index * 8 + bit_index
|
||||
for byte_index, byte in enumerate(raw)
|
||||
for bit_index in range(8)
|
||||
if byte & (1 << bit_index)
|
||||
}
|
||||
|
||||
|
||||
def export_authority(
|
||||
ctx, *, require_unl_report=True, committee_node_ids=None
|
||||
):
|
||||
"""Build the explicit authority declaration for a direct Export.
|
||||
|
||||
Direct clients pin the validated parent ledger and select members from its
|
||||
@@ -92,8 +120,34 @@ def export_authority(ctx, *, require_unl_report=True):
|
||||
raise AssertionError(f"UNLReport active universe unavailable: {report}")
|
||||
active = [{}]
|
||||
|
||||
committee = bytearray((len(active) + 7) // 8)
|
||||
for index in range(len(active)):
|
||||
active_keys = set()
|
||||
for entry in active:
|
||||
validator = entry.get("ActiveValidator", entry)
|
||||
key = validator.get("PublicKey")
|
||||
if not key:
|
||||
raise AssertionError(f"Malformed UNLReport validator entry: {entry}")
|
||||
active_keys.add(key.upper())
|
||||
active_keys = sorted(active_keys, key=bytes.fromhex)
|
||||
|
||||
if committee_node_ids is None:
|
||||
selected_positions = range(len(active_keys))
|
||||
else:
|
||||
masters = _validator_master_keys_by_node(ctx)
|
||||
selected_positions = []
|
||||
for node_id in committee_node_ids:
|
||||
if node_id not in masters:
|
||||
raise AssertionError(f"Unknown testnet validator node n{node_id}")
|
||||
try:
|
||||
selected_positions.append(active_keys.index(masters[node_id]))
|
||||
except ValueError as exc:
|
||||
raise AssertionError(
|
||||
f"Validator n{node_id} is absent from the active UNLReport"
|
||||
) from exc
|
||||
if not selected_positions:
|
||||
raise AssertionError("Export committee must select at least one validator")
|
||||
|
||||
committee = bytearray((len(active_keys) + 7) // 8)
|
||||
for index in selected_positions:
|
||||
committee[index // 8] |= 1 << (index % 8)
|
||||
|
||||
return {
|
||||
@@ -175,7 +229,16 @@ async def wait_for_validated_transaction(
|
||||
raise AssertionError(f"Transaction {tx_hash} did not validate by ledger {target}")
|
||||
|
||||
|
||||
async def submit_direct_export(ctx, log, tx, wallet, *, timeout=60, max_rebases=2):
|
||||
async def submit_direct_export(
|
||||
ctx,
|
||||
log,
|
||||
tx,
|
||||
wallet,
|
||||
*,
|
||||
timeout=60,
|
||||
max_rebases=2,
|
||||
committee_node_ids=None,
|
||||
):
|
||||
"""Submit a direct Export, rebasing after a validated parent mismatch."""
|
||||
for attempt in range(max_rebases + 1):
|
||||
current = ctx.validated_ledger_index(0)
|
||||
@@ -183,7 +246,9 @@ async def submit_direct_export(ctx, log, tx, wallet, *, timeout=60, max_rebases=
|
||||
raise AssertionError("Validated ledger unavailable before Export")
|
||||
|
||||
candidate = dict(tx)
|
||||
candidate.update(export_authority(ctx))
|
||||
candidate.update(
|
||||
export_authority(ctx, committee_node_ids=committee_node_ids)
|
||||
)
|
||||
candidate["LastLedgerSequence"] = current + EXPORT_RETRY_LEDGER_WINDOW
|
||||
result = await ctx.submit_and_wait(candidate, wallet, timeout=timeout)
|
||||
if result.get("engine_result") != "tecEXPORT_UNIVERSE_MISMATCH":
|
||||
|
||||
193
.testnet/scenarios/export/export_share_subscription.py
Normal file
193
.testnet/scenarios/export/export_share_subscription.py
Normal file
@@ -0,0 +1,193 @@
|
||||
""":descr: subscribe over a real WebSocket to post-validation Export shares
|
||||
|
||||
The subscriber opens before the Export is submitted. It proves the stream
|
||||
publishes a quorum of independently attributable shares for the exact validated
|
||||
origin and that those same signature records form the later ledger witness.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
|
||||
import websockets
|
||||
from xahaud_scripts.testnet.config import _decode_node_public_key
|
||||
|
||||
from export_helpers import (
|
||||
EXPORT_RETRY_LEDGER_WINDOW,
|
||||
assert_export_latch,
|
||||
bitmap_positions,
|
||||
require_export,
|
||||
submit_direct_export,
|
||||
wait_for_export_signature_witness,
|
||||
)
|
||||
|
||||
|
||||
def _witness_records(witness):
|
||||
positions = sorted(bitmap_positions(witness["EntropyContributors"]))
|
||||
signers = witness["_WitnessSigners"]
|
||||
if len(positions) != len(signers):
|
||||
raise AssertionError("Witness bitmap and signer count differ")
|
||||
return {
|
||||
(
|
||||
position,
|
||||
signer["SigningPubKey"].upper(),
|
||||
signer["TxnSignature"].upper(),
|
||||
)
|
||||
for position, signer in zip(positions, signers, strict=True)
|
||||
}
|
||||
|
||||
|
||||
async def scenario(ctx, log):
|
||||
await require_export(ctx, log)
|
||||
await ctx.fund_accounts({"alice": 10000, "bob": 1000})
|
||||
alice = ctx.account("alice")
|
||||
bob = ctx.account("bob")
|
||||
|
||||
network = json.loads((ctx.base_dir / "network.json").read_text())
|
||||
node0 = next(node for node in network["nodes"] if int(node["id"]) == 0)
|
||||
ws_url = f"ws://127.0.0.1:{node0['port_ws']}"
|
||||
events = []
|
||||
|
||||
async with websockets.connect(ws_url, open_timeout=10) as websocket:
|
||||
await websocket.send(
|
||||
json.dumps(
|
||||
{
|
||||
"id": 1,
|
||||
"command": "subscribe",
|
||||
"streams": ["export_signatures"],
|
||||
}
|
||||
)
|
||||
)
|
||||
ack = json.loads(await asyncio.wait_for(websocket.recv(), timeout=10))
|
||||
if ack.get("status") != "success":
|
||||
raise AssertionError(f"export_signatures subscription failed: {ack}")
|
||||
log("Subscribed to export_signatures over WebSocket")
|
||||
|
||||
async def receive_events():
|
||||
while True:
|
||||
event = json.loads(await websocket.recv())
|
||||
if event.get("stream") == "export_signatures":
|
||||
events.append(event)
|
||||
|
||||
reader = asyncio.create_task(receive_events())
|
||||
try:
|
||||
current = ctx.validated_ledger_index(0)
|
||||
result = await submit_direct_export(
|
||||
ctx,
|
||||
log,
|
||||
{
|
||||
"TransactionType": "Export",
|
||||
"Fee": "1000000",
|
||||
"ExportedTxn": {
|
||||
"TransactionType": "Payment",
|
||||
"Account": alice.address,
|
||||
"Destination": bob.address,
|
||||
"Amount": "1000000",
|
||||
"Fee": "10",
|
||||
"Sequence": 0,
|
||||
"TicketSequence": 1,
|
||||
"FirstLedgerSequence": current + 1,
|
||||
"LastLedgerSequence": current
|
||||
+ EXPORT_RETRY_LEDGER_WINDOW,
|
||||
"Flags": 2147483648,
|
||||
"SigningPubKey": "",
|
||||
},
|
||||
},
|
||||
alice.wallet,
|
||||
)
|
||||
if result.get("engine_result") != "tesSUCCESS":
|
||||
raise AssertionError(f"Export failed: {result}")
|
||||
|
||||
origin = result.get("hash")
|
||||
origin_seq = int(result.get("ledger_index"))
|
||||
if not origin:
|
||||
raise AssertionError(f"Validated Export missing hash: {result}")
|
||||
origin_ledger = ctx.ledger(origin_seq) or {}
|
||||
origin_hash = origin_ledger.get("ledger_hash") or origin_ledger.get(
|
||||
"ledger", {}
|
||||
).get("hash")
|
||||
if not origin_hash:
|
||||
raise AssertionError(
|
||||
f"Validated origin ledger {origin_seq} missing hash"
|
||||
)
|
||||
|
||||
latches = assert_export_latch(
|
||||
ctx,
|
||||
alice.address,
|
||||
log,
|
||||
origin_hash=origin,
|
||||
expect_witness=False,
|
||||
)
|
||||
selected = bitmap_positions(latches[0]["ExportCommittee"])
|
||||
quorum = (4 * len(selected) + 4) // 5
|
||||
|
||||
witness = await wait_for_export_signature_witness(
|
||||
ctx, log, origin, after_ledger=origin_seq
|
||||
)
|
||||
expected_records = _witness_records(witness)
|
||||
|
||||
deadline = asyncio.get_running_loop().time() + 10
|
||||
while asyncio.get_running_loop().time() < deadline:
|
||||
matching = [event for event in events if event.get("origin_txid") == origin]
|
||||
observed_records = {
|
||||
(
|
||||
int(event["universe_position"]),
|
||||
_decode_node_public_key(event["signing_key"]),
|
||||
event["signature"].upper(),
|
||||
)
|
||||
for event in matching
|
||||
}
|
||||
if expected_records <= observed_records:
|
||||
break
|
||||
await asyncio.sleep(0.1)
|
||||
else:
|
||||
raise AssertionError(
|
||||
"WebSocket stream did not publish every signature used by "
|
||||
f"the witness: expected={expected_records}, "
|
||||
f"observed={observed_records}"
|
||||
)
|
||||
|
||||
unique_positions = set()
|
||||
for event in matching:
|
||||
if event.get("type") != "exportSignatureReceived":
|
||||
raise AssertionError(f"Unexpected Export stream event: {event}")
|
||||
if event.get("version") != 1:
|
||||
raise AssertionError(f"Unexpected Export share version: {event}")
|
||||
if event.get("owner") != alice.address:
|
||||
raise AssertionError(f"Export stream owner mismatch: {event}")
|
||||
if int(event.get("origin_ledger_seq", 0)) != origin_seq:
|
||||
raise AssertionError(f"Export stream origin sequence mismatch: {event}")
|
||||
if event.get("origin_ledger_hash") != origin_hash:
|
||||
raise AssertionError(f"Export stream origin hash mismatch: {event}")
|
||||
if event.get("trigger_txid") != origin:
|
||||
raise AssertionError(f"Export stream trigger mismatch: {event}")
|
||||
position = int(event.get("universe_position", -1))
|
||||
if position not in selected:
|
||||
raise AssertionError(f"Unselected validator streamed a share: {event}")
|
||||
unique_positions.add(position)
|
||||
|
||||
if len(unique_positions) < quorum:
|
||||
raise AssertionError(
|
||||
f"Stream exposed only {len(unique_positions)} selected positions; "
|
||||
f"need quorum {quorum}"
|
||||
)
|
||||
witness_seq = int(witness["LedgerSequence"])
|
||||
if not any(
|
||||
int(event.get("ledger_index", 0)) < witness_seq
|
||||
for event in matching
|
||||
):
|
||||
raise AssertionError(
|
||||
"No Export share stream event preceded the ledger witness"
|
||||
)
|
||||
log(
|
||||
f"WebSocket exposed {len(unique_positions)} selected shares; "
|
||||
f"witness used {len(expected_records)}"
|
||||
)
|
||||
finally:
|
||||
reader.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await reader
|
||||
|
||||
log("PASS")
|
||||
Reference in New Issue
Block a user