Make snapshot generation explicit in test setup

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()

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,