diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index 67006f39732..2e37e1b6853 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -176,21 +176,6 @@ def test_forced_ledger_chunk(network, args): return network -def find_snapshot_after_seqno(snapshots_dir, seqno): - for snapshot_name in os.listdir(snapshots_dir): - with ccf.ledger.Snapshot( - os.path.join(snapshots_dir, snapshot_name) - ) as snapshot: - snapshot_seqno = snapshot.get_public_domain().get_seqno() - if snapshot_seqno > seqno: - LOG.info(f"Found a snapshot at {snapshot_seqno} which is after {seqno}") - return snapshot_seqno - - raise RuntimeError( - f"Could not find a snapshot after seqno {seqno} in {snapshots_dir}" - ) - - def find_latest_committed_snapshot_name(network, count=1): assert count > 0, f"Expected positive snapshot count, got {count}" primary, _ = network.find_primary() @@ -231,10 +216,9 @@ def test_forced_snapshot(network, args): # Issue some more transactions network.txs.issue(network, number_txs=5) - snapshots_dir = network.get_committed_snapshots( - primary, target_seqno=hwm_pre_proposal + 1, wait_for_target_seqno=True - ) - find_snapshot_after_seqno(snapshots_dir, hwm_pre_proposal) + snapshot_path = primary.wait_for_snapshot(hwm_pre_proposal + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_public_domain().get_seqno() > hwm_pre_proposal # Do not issue another transaction after this call. The snapshot request # must make all preceding transactions available in a committed chunk even @@ -283,13 +267,10 @@ def issue_governance_txs(count): issue_governance_txs(5) - snapshots_dir = network.get_committed_snapshots( - primary, - target_seqno=hwm_pre_proposal + 1, - force_txs=False, - wait_for_target_seqno=True, - ) - snapshot_seqno = find_snapshot_after_seqno(snapshots_dir, hwm_pre_proposal) + snapshot_path = primary.wait_for_snapshot(hwm_pre_proposal + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + snapshot_seqno = snapshot.get_public_domain().get_seqno() + assert snapshot_seqno > hwm_pre_proposal _, committed_ledger_dirs = primary.get_ledger() ledger = ccf.ledger.Ledger( @@ -334,13 +315,9 @@ def test_snapshot_create_endpoint(network, args): r = c.post("/node/snapshot:create") assert r.status_code == http.HTTPStatus.NO_CONTENT, r - snapshots_dir = network.get_committed_snapshots( - primary, - target_seqno=hwm_pre_request + 1, - force_txs=False, - wait_for_target_seqno=True, - ) - find_snapshot_after_seqno(snapshots_dir, hwm_pre_request) + snapshot_path = primary.wait_for_snapshot(hwm_pre_request + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_public_domain().get_seqno() > hwm_pre_request return network @@ -362,36 +339,34 @@ def test_large_snapshot(network, args): log_capture=[], ) - # Force a snapshot at the following signature + target = network.txs.issue(network, number_txs=1) + # Force a snapshot covering the large entries at the following signature. primary.trigger_snapshot() # Check that there is at least a snapshot larger than args.max_msg_size_bytes - snapshots_dir = network.get_committed_snapshots(primary) + snapshot_path = primary.wait_for_snapshot(target.seqno) extra_data_size_bytes = 10000 # Upper bound on additional snapshot data (e.g. receipt) that is passed separately from the snapshot - for s in os.listdir(snapshots_dir): - snapshot_size = os.stat(os.path.join(snapshots_dir, s)).st_size - if snapshot_size > int(args.max_msg_size_bytes) + extra_data_size_bytes: - # Make sure that large snapshot can be parsed - snapshot = ccf.ledger.Snapshot(os.path.join(snapshots_dir, s)) - assert snapshot.get_len() == snapshot_size - LOG.info( - f"Found snapshot [{snapshot_size}] larger than ring buffer max msg size {args.max_msg_size_bytes}" - ) - return network - - raise RuntimeError( - f"Could not find any snapshot file larger than {args.max_msg_size_bytes}" + snapshot_size = os.path.getsize(snapshot_path) + assert snapshot_size > int(args.max_msg_size_bytes) + extra_data_size_bytes, ( + f"Snapshot {snapshot_path} has size {snapshot_size}, expected more than " + f"{int(args.max_msg_size_bytes) + extra_data_size_bytes}" ) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_len() == snapshot_size + return network def test_snapshot_access(network, args): primary, backups = network.find_nodes() - snapshots_dir = network.get_committed_snapshots(primary) - snapshot_name = ccf.ledger.latest_snapshot(snapshots_dir) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) + snapshot_path = primary.get_snapshots()[-1] + snapshot_name = os.path.basename(snapshot_path) snapshot_index, _ = ccf.ledger.snapshot_index_from_filename(snapshot_name) - with open(os.path.join(snapshots_dir, snapshot_name), "rb") as f: + with open(snapshot_path, "rb") as f: snapshot_data = f.read() for node in (primary, *backups): @@ -558,9 +533,11 @@ def test_snapshot_repr_digest(network, args): """ primary, _ = network.find_nodes() - snapshots_dir = network.get_committed_snapshots(primary) - snapshot_name = ccf.ledger.latest_snapshot(snapshots_dir) - snapshot_path = os.path.join(snapshots_dir, snapshot_name) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) + snapshot_path = primary.get_snapshots()[-1] + snapshot_name = os.path.basename(snapshot_path) with open(snapshot_path, "rb") as f: snapshot_data = f.read() @@ -697,24 +674,15 @@ def test_snapshot_selection(network, args): LOG.info("Creating snapshots") primary, backups = network.find_nodes() - for i in range(3): + for _ in range(max(3, len(backups))): + target = network.txs.issue(network, number_txs=1) primary.trigger_snapshot() - # Snapshot creation and commit takes time. All of the helpers we have to track/poll this - # are expensive, so try a short sleep - time.sleep(1) - - snapshots_dir = network.get_committed_snapshots( - primary, - force_txs=False, - ) + primary.wait_for_snapshot(target.seqno) src_snapshots = [] - for snapshot_name in os.listdir(snapshots_dir): - if ccf.ledger.is_snapshot_file_committed(snapshot_name): - seqno, _ = ccf.ledger.snapshot_index_from_filename(snapshot_name) - src_snapshots.append( - (seqno, snapshot_name, os.path.join(snapshots_dir, snapshot_name)) - ) + for snapshot_path in primary.get_snapshots(): + seqno, _ = ccf.ledger.snapshot_index_from_filename(snapshot_path) + src_snapshots.append((seqno, os.path.basename(snapshot_path), snapshot_path)) src_snapshots.sort() best_snapshot = src_snapshots[-1][1] @@ -1569,8 +1537,10 @@ def test_ledger_chunk_redirect_gap(network, args): commit_seqno = TxID.from_str(r["transaction_id"]).seqno new_node = network.create_node() - # force primary to generate a new snapshot after commit idx - network.get_committed_snapshots() + # Commit a transaction beyond the old boundary before requesting a snapshot. + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) network.join_node( new_node, args.package, @@ -3455,8 +3425,9 @@ def test_join_time_snapshot_fetch_failure(network, args): # Ensure at least one committed snapshot exists so that joining nodes # can be given one (startup_seqno > 0). - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) - network.get_committed_snapshots(primary) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) # Full reconfigure so every remaining node has startup_seqno > 0 # (otherwise a redirect to the primary would let the joiner succeed). @@ -3545,11 +3516,12 @@ def test_error_message_on_failure_to_fetch_snapshot(network, args): ) network.trust_node(new_node, args) - # Issue enough transactions to trigger a new snapshot on the primary. + # Explicitly trigger a snapshot after the new node has joined. # The snapshot_evidence hook on new_node then schedules BackupSnapshotFetch, # which exhausts its 3 attempts (all HTTP 404) and logs "giving up". - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) - network.get_committed_snapshots(primary) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) _assert_snapshot_fetch_failure_messages(new_node, timeout_s=30) @@ -3559,26 +3531,21 @@ def test_backup_snapshot_fetch(network, args): backups = network.find_backups() assert len(backups) > 0, "Expected at least one backup node" - # Issue enough transactions to trigger snapshot generation - # The primary will create a snapshot after snapshot_tx_interval txs - LOG.info("Issuing transactions to trigger snapshot generation") - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() # Wait for committed snapshots on the primary, and use those as expected # snapshot files on backups. LOG.info("Waiting for committed snapshot on primary") - primary_snapshots_dir = network.get_committed_snapshots(primary) + primary.wait_for_snapshot(target.seqno) expected_snapshot_sizes = { - snapshot_name: os.path.getsize( - os.path.join(primary_snapshots_dir, snapshot_name) - ) - for snapshot_name in os.listdir(primary_snapshots_dir) - if ccf.ledger.is_snapshot_file_committed(snapshot_name) + os.path.basename(path): os.path.getsize(path) + for path in primary.get_snapshots(include_read_only=True) } assert ( len(expected_snapshot_sizes) > 0 - ), f"No committed snapshots found in {primary_snapshots_dir}" + ), f"No committed snapshots found on primary {primary.local_node_id}" for backup in backups: backup_snapshots_dir = os.path.join( @@ -3666,7 +3633,10 @@ def assert_no_snapshot_is_present(duration_s=10): ), f"Expected snapshot directory {snapshot_dir} to exist" assert_no_snapshot_is_present() - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) + target = network.txs.issue(network, number_txs=1, msg="X" * 2048) + primary.trigger_snapshot() + snapshot_path = primary.wait_for_snapshot(target.seqno) + assert os.path.getsize(snapshot_path) > 1024, snapshot_path assert_no_snapshot_is_present() expected_log_message = "Failed writing received data to disk/application" out_path, _ = new_node.get_logs() diff --git a/tests/recovery_snapshot_endorsements.py b/tests/recovery_snapshot_endorsements.py index 64c2b18ebe9..6e604db60b1 100644 --- a/tests/recovery_snapshot_endorsements.py +++ b/tests/recovery_snapshot_endorsements.py @@ -138,19 +138,15 @@ def run_recovery_snapshot_endorsements(args): initial_network.start_and_open(args) primary, _ = initial_network.find_primary() - app.LoggingTxs("user0").issue( + target = app.LoggingTxs("user0").issue( initial_network, number_txs=2, send_private=False, send_public=True, wait_for_sync=True, ) - snapshot_trigger = primary.trigger_snapshot() - initial_network.get_committed_snapshots( - primary, - target_seqno=snapshot_trigger.seqno, - wait_for_target_seqno=True, - ) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) app.LoggingTxs("user0").issue( initial_network, number_txs=2,