diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index 97655aea6e1e..67006f397322 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -3863,12 +3863,7 @@ def run_propose_request_vote(const_args): original_primary, original_term = network.find_primary() LOG.info("Waiting for initial snapshot") - network.get_committed_snapshots( - original_primary, - target_seqno=1, - force_txs=False, - wait_for_target_seqno=True, - ) + original_primary.wait_for_snapshot(1) network.wait_for_node_commit_sync(timeout=16) original_primary.remote.remote.proc.send_signal(signal.SIGTERM) @@ -3916,29 +3911,11 @@ def net_with_min_tx(label, min_tx_interval): net.start_and_open(inner_args) yield net - def get_committed_snapshot_files(net): - primary, _ = net.find_primary() - snapshots_dirs = [ - os.path.join(primary.remote.remote.root, primary.remote.snapshots_dir_name) - ] - if primary.remote.read_only_snapshots_dir_name is not None: - snapshots_dirs.append( - os.path.join( - primary.remote.remote.root, - primary.remote.read_only_snapshots_dir_name, - ) - ) - - snapshots = set() - for snapshots_dir in snapshots_dirs: - if not os.path.isdir(snapshots_dir): - continue - - for snapshot_name in os.listdir(snapshots_dir): - if ccf.ledger.is_snapshot_file_committed(snapshot_name): - snapshots.add(snapshot_name) - - return snapshots + def snapshot_names(node): + return { + os.path.basename(path) + for path in node.get_snapshots(include_read_only=True) + } # Pattern for these tests: # 1. wait for any startup triggered txs to commit and net to settle @@ -3949,15 +3926,15 @@ def get_committed_snapshot_files(net): def run_low(): with net_with_min_tx("_low", 0) as net: time.sleep(1) - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, - ) - baseline = get_committed_snapshot_files(net) + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] + ) + primary.wait_for_snapshot(target.seqno, timeout=5) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( len(final - baseline) >= 8 ), f"With min_tx_interval set to 0 we expect snapshots to be generated at around 1 per second, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." @@ -3965,18 +3942,18 @@ def run_low(): def run_exact(): with net_with_min_tx("_exact", 2) as net: time.sleep(1) - try: - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] ) + try: + primary.wait_for_snapshot(target.seqno, timeout=5) except TimeoutError: pass - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 2 we expect no snapshots to be generated without transactions, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." @@ -3984,38 +3961,32 @@ def run_exact(): def run_high(): with net_with_min_tx("_high", 10) as net: time.sleep(1) - try: - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] ) + try: + primary.wait_for_snapshot(target.seqno, timeout=5) except TimeoutError: pass - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 10 we expect no snapshots to be generated without transactions, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." tx_id = net.txs.issue(net, number_txs=1) - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 10 and we expect no snapshots to be generated with only one extra tx, but got {final} snapshots 10s after a baseline of {baseline}, and in total saw {final - baseline} new snapshots over the test." net.txs.issue(net, number_txs=20) - primary, _ = net.find_primary() - net.get_committed_snapshots( - primary, - target_seqno=tx_id.seqno, - force_txs=False, - wait_for_target_seqno=True, - ) + primary.wait_for_snapshot(tx_id.seqno) with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: futures = [ @@ -4070,8 +4041,10 @@ def run_snapshot_persistence_across_primary_failure(const_args): elapsed = time.time() - start_time snapshots = set() for node in net.nodes: - snapshots_dir = net.get_committed_snapshots(node, force_txs=False) - snapshots = snapshots.union(set(os.listdir(snapshots_dir))) + snapshots.update( + os.path.basename(path) + for path in node.get_snapshots(include_read_only=True) + ) total_snapshots = len(snapshots) diff --git a/tests/governance_history.py b/tests/governance_history.py index 13787f0da8b4..6d9a1edef31f 100644 --- a/tests/governance_history.py +++ b/tests/governance_history.py @@ -4,7 +4,6 @@ import base64 import http import json -import os import ccf.ledger import ccf.read_ledger @@ -250,9 +249,9 @@ def fmt_str(data: bytes) -> str: tables_format_rules=format_rule, ) - snapshot_dir = network.get_committed_snapshots(primary) + snapshot_path = primary.wait_for_snapshot(target_seqno) assert ccf.read_ledger.run( - paths=[os.path.join(snapshot_dir, os.listdir(snapshot_dir)[-1])], + paths=[snapshot_path], print_mode=ccf.read_ledger.PrintMode.Contents, is_snapshot=True, tables_format_rules=format_rule, diff --git a/tests/infra/node.py b/tests/infra/node.py index fc23e2b15e81..b7894c24fc83 100644 --- a/tests/infra/node.py +++ b/tests/infra/node.py @@ -877,6 +877,59 @@ def get_ledger(self): return current_ledger_dir, [committed_ledger_dir] + def get_snapshots(self, *, include_read_only=False) -> list[str]: + """List committed snapshot paths on this node, ordered by snapshot seqno. + + Paths are node-owned: copy them before modifying them or relying on them + surviving cleanup. Read-only startup snapshots are excluded by default. + """ + directories = [self.remote.snapshots_dir_name] + if include_read_only and self.remote.read_only_snapshots_dir_name is not None: + directories.append(self.remote.read_only_snapshots_dir_name) + + snapshots = [] + for directory in directories: + path = os.path.join(self.remote.remote.root, directory) + try: + with os.scandir(path) as entries: + snapshots.extend( + entry.path + for entry in entries + if entry.name.startswith("snapshot_") + and ccf.ledger.is_snapshot_file_committed(entry.name) + and entry.is_file() + ) + except FileNotFoundError: + LOG.debug(f"Snapshot directory does not exist yet: {path}") + + return sorted(snapshots, key=ccf.ledger.snapshot_index_from_filename) + + def wait_for_snapshot(self, target_seqno, timeout=20) -> str: + """Wait for a committed snapshot in this node's writable directory. + + The snapshot state must include target_seqno. This does not emit + transactions, trigger snapshots, or copy files. + """ + LOG.info( + f"Waiting for node {self.local_node_id} snapshot including seqno {target_seqno}" + ) + end_time = time.monotonic() + timeout + while True: + snapshots = self.get_snapshots() + for snapshot in snapshots: + if ccf.ledger.snapshot_index_from_filename(snapshot)[0] >= target_seqno: + LOG.info(f"Found committed snapshot {snapshot}") + return snapshot + + remaining = end_time - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Could not find committed snapshot on node {self.local_node_id} " + f"including seqno {target_seqno} after {timeout}s; " + f"snapshot files: {snapshots}" + ) + time.sleep(min(0.1, remaining)) + def get_committed_snapshots(self, pre_condition_func=lambda src_dir, _: True): ( main_snapshots_dir, diff --git a/tests/recovery_snapshot_endorsements.py b/tests/recovery_snapshot_endorsements.py index e3db52595c44..64c2b18ebe97 100644 --- a/tests/recovery_snapshot_endorsements.py +++ b/tests/recovery_snapshot_endorsements.py @@ -111,14 +111,21 @@ def _copy_ledger_prefix(source_dirs, destination, first_excluded_seqno): assert copied > 0 -def _assert_node_snapshot_unchanged( - network, node, snapshot_name, expected_snapshot_digest -): - snapshots_dir = network.get_committed_snapshots(node, force_txs=False) - snapshot_path = os.path.join(snapshots_dir, snapshot_name) - assert os.path.isfile(snapshot_path), snapshot_path - with open(snapshot_path, "rb") as snapshot_file: - assert hashlib.sha256(snapshot_file.read()).digest() == expected_snapshot_digest +def _assert_node_snapshot_unchanged(node, snapshot_name, expected_snapshot_digest): + snapshot_paths = [ + path + for path in node.get_snapshots(include_read_only=True) + if os.path.basename(path) == snapshot_name + ] + assert ( + snapshot_paths + ), f"Snapshot {snapshot_name} not found on node {node.local_node_id}" + for snapshot_path in snapshot_paths: + with open(snapshot_path, "rb") as snapshot_file: + assert ( + hashlib.sha256(snapshot_file.read()).digest() + == expected_snapshot_digest + ), snapshot_path def run_recovery_snapshot_endorsements(args): @@ -237,7 +244,7 @@ def run_recovery_snapshot_endorsements(args): < logs.index(public_recovery_log) ) _assert_node_snapshot_unchanged( - valid_attempt, valid_primary, snapshot_name, snapshot_digest + valid_primary, snapshot_name, snapshot_digest ) finally: _stop_incomplete_recovery(valid_attempt) @@ -272,7 +279,7 @@ def run_recovery_snapshot_endorsements(args): assert "No usable local snapshot found" in logs assert "Setting startup snapshot seqno" not in logs _assert_node_snapshot_unchanged( - fallback_attempt, fallback_primary, snapshot_name, snapshot_digest + fallback_primary, snapshot_name, snapshot_digest ) finally: _stop_incomplete_recovery(fallback_attempt)