Test mid-recovery election (#6920)

Co-authored-by: Amaury Chamayou <amaury@xargs.fr>
Co-authored-by: Amaury Chamayou <amchamay@microsoft.com>

Author: maxtropets

CMakeLists.txt

@@ -1026,7 +1026,7 @@ if(BUILD_TESTS)
     add_picobench(hash_bench SRCS src/ds/test/hash_bench.cpp)
 
     if(LONG_TESTS)
-      set(ADDITIONAL_RECOVERY_ARGS --with-load)
+      set(ADDITIONAL_RECOVERY_ARGS --with-load --with-election)
 
       add_e2e_test(
         NAME recovery_test_cft_api_0

src/node/node_state.h

@@ -1269,6 +1269,11 @@ namespace ccf
 
       sm.expect(NodeStartupState::readingPrivateLedger);
 
+      LOG_INFO_FMT(
+        "Try end private recovery at {}. Is primary: {}",
+        recovery_v,
+        consensus->is_primary());
+
       if (recovery_v != recovery_store->current_version())
       {
         throw std::logic_error(fmt::format(
@@ -1298,6 +1303,10 @@ namespace ccf
       // Open the service
       if (consensus->can_replicate())
       {
+        LOG_INFO_FMT(
+          "Try end private recovery at {}. Trigger service opening",
+          recovery_v);
+
         auto tx = network.tables->create_tx();
 
         // Clear recovery shares that were submitted to initiate the recovery

tests/infra/node.py

@@ -464,6 +464,9 @@ def get_logs(self):
     def sigterm(self):
         self.remote.sigterm()
 
+    def sigkill(self):
+        self.remote.sigkill()
+
     def is_stopped(self):
         return self.network_state == NodeNetworkState.stopped
 

tests/infra/remote.py

@@ -227,6 +227,9 @@ def _print_stack_trace(self):
     def sigterm(self):
         self.proc.terminate()
 
+    def sigkill(self):
+        self.proc.send_signal(signal.SIGKILL)
+
     def stop(self):
         """
         Disconnect the client, and therefore shut down the command as well.
@@ -666,6 +669,9 @@ def debug_node_cmd(self):
     def sigterm(self):
         self.remote.sigterm()
 
+    def sigkill(self):
+        self.remote.sigkill()
+
     def stop(self):
         try:
             self.remote.stop()

tests/recovery.py

@@ -29,7 +29,7 @@
 from cryptography.hazmat.primitives import serialization
 from ccf.cose import validate_cose_sign1
 from pycose.messages import Sign1Message  # type: ignore
-
+import random
 from loguru import logger as LOG
 
 
@@ -107,10 +107,77 @@ def verify_endorsements_chain(primary, endorsements, pubkey):
         pubkey = serialization.load_der_public_key(next_key_bytes, default_backend())
 
 
+def recover_with_primary_dying(args, recovered_network):
+    # Minimal copy-paste from network.recover() with primary shut down.
+    recovered_network.consortium.activate(recovered_network.find_random_node())
+    recovered_network.consortium.check_for_service(
+        recovered_network.find_random_node(),
+        status=infra.network.ServiceStatus.RECOVERING,
+    )
+    recovered_network.wait_for_all_nodes_to_be_trusted(
+        recovered_network.find_random_node()
+    )
+
+    prev_service_identity = None
+    if args.previous_service_identity_file:
+        prev_service_identity = slurp_file(args.previous_service_identity_file)
+    LOG.info(f"Prev identity: {prev_service_identity}")
+
+    recovered_network.consortium.transition_service_to_open(
+        recovered_network.find_random_node(),
+        previous_service_identity=prev_service_identity,
+    )
+
+    recovered_network.consortium.recover_with_shares(
+        recovered_network.find_random_node()
+    )
+    for node in recovered_network.get_joined_nodes():
+        recovered_network.wait_for_state(
+            node,
+            infra.node.State.READING_PRIVATE_LEDGER.value,
+            timeout=args.ledger_recovery_timeout,
+        )
+
+    retired_primary, _ = recovered_network.find_primary()
+    retired_id = retired_primary.node_id
+
+    LOG.info(f"Force-kill primary {retired_id}")
+    retired_primary.sigkill()
+    recovered_network.nodes.remove(retired_primary)
+
+    primary, _ = recovered_network.find_primary()
+    while not primary or primary.node_id == retired_id:
+        LOG.info("Keep looking for new primary")
+        time.sleep(0.1)
+        primary, _ = recovered_network.find_primary()
+
+    # Ensure new primary has been elected while all nodes are still reading private entries.
+    for node in recovered_network.get_joined_nodes():
+        LOG.info(f"Check state for node id {node.node_id}")
+        with node.client(connection_timeout=1) as c:
+            assert (
+                infra.node.State.READING_PRIVATE_LEDGER.value
+                == c.get("/node/state").body.json()["state"]
+            )
+
+    # Wait for recovery to complete.
+    for node in recovered_network.get_joined_nodes():
+        recovered_network.wait_for_state(
+            node,
+            infra.node.State.PART_OF_NETWORK.value,
+            timeout=args.ledger_recovery_timeout,
+        )
+
+
 @reqs.description("Recover a service")
 @reqs.recover(number_txs=2)
 def test_recover_service(
-    network, args, from_snapshot=True, no_ledger=False, via_recovery_owner=False
+    network,
+    args,
+    from_snapshot=True,
+    no_ledger=False,
+    via_recovery_owner=False,
+    force_election=False,
 ):
     network.save_service_identity(args)
     old_primary, _ = network.find_primary()
@@ -127,6 +194,16 @@ def test_recover_service(
     if from_snapshot:
         snapshots_dir = network.get_committed_snapshots(old_primary)
 
+    if force_election:
+        # Necessary to make recovering private entries taking long enough time
+        # to allow election to happen if primary gets killed. These later get verified post-recovery (logging app verify_tx() thing).
+        network.txs.issue(
+            network,
+            number_txs=10000,
+            send_public=False,
+            msg=str(bytes(random.getrandbits(8) for _ in range(512))),
+        )
+
     # Start health watcher and stop nodes one by one until a recovery has to be staged
     watcher = infra.health_watcher.NetworkHealthWatcher(network, args, verbose=True)
     watcher.start()
@@ -202,7 +279,10 @@ def test_recover_service(
             r = c.get("/node/ready/app")
             assert r.status_code == http.HTTPStatus.SERVICE_UNAVAILABLE.value, r
 
-    recovered_network.recover(args, via_recovery_owner=via_recovery_owner)
+    if force_election:
+        recover_with_primary_dying(args, recovered_network)
+    else:
+        recovered_network.recover(args, via_recovery_owner=via_recovery_owner)
 
     LOG.info("Check that new service view is as expected")
     new_primary, _ = recovered_network.find_primary()
@@ -216,6 +296,16 @@ def test_recover_service(
         r = c.get("/node/ready/gov")
         assert r.status_code == http.HTTPStatus.NO_CONTENT.value, r
         r = c.get("/node/ready/app")
+
+        # Service opening may be slightly delayed due to forced election (if option enabled).
+        app_ready_attempts = 10 if force_election else 0
+        while (
+            r.status_code != http.HTTPStatus.NO_CONTENT.value and app_ready_attempts > 0
+        ):
+            time.sleep(0.1)
+            app_ready_attempts -= 1
+            r = c.get("/node/ready/app")
+
         assert r.status_code == http.HTTPStatus.NO_CONTENT.value, r
 
     return recovered_network
@@ -999,6 +1089,27 @@ def run_recover_snapshot_alone(args):
         return network
 
 
+def run_recovery_with_election(args):
+    """
+    Recover a service but force election during recovery.
+    """
+    if not args.with_election:
+        return
+
+    txs = app.LoggingTxs("user0")
+    with infra.network.network(
+        args.nodes,
+        args.binary_dir,
+        args.debug_nodes,
+        args.perf_nodes,
+        pdb=args.pdb,
+        txs=txs,
+    ) as network:
+        network.start_and_open(args)
+        test_recover_service(network, args, force_election=True)
+        return network
+
+
 def run_recover_via_initial_recovery_owner(args):
     """
     Recover a service using the recovery owner added as part of service creation, without requiring any other recovery members to participate.
@@ -1082,6 +1193,12 @@ def add(parser):
             action="store_true",
             default=False,
         )
+        parser.add_argument(
+            "--with-election",
+            help="If set, the primary gets killed to force election mid-recovery",
+            action="store_true",
+            default=False,
+        )
 
     cr = ConcurrentRunner(add)
 
@@ -1130,4 +1247,13 @@ def add(parser):
         nodes=infra.e2e_args.min_nodes(cr.args, f=0),  # 1 node suffices for recovery
     )
 
+    cr.add(
+        "recovery_with_election",
+        run_recovery_with_election,
+        package="samples/apps/logging/liblogging",
+        nodes=infra.e2e_args.min_nodes(cr.args, f=1),
+        ledger_chunk_bytes="50KB",
+        snapshot_tx_interval=30,
+    )
+
     cr.run()