SERVER-111964 replicaset_transition_to_and_from_csrs should be resilient to text fixture restarts (#42359)

GitOrigin-RevId: d04760388192d63eb8b1e00bf00a220e2dfd5a8d

buildscripts/resmokelib/testing/fixtures/replicaset.py

@@ -178,6 +178,9 @@ class ReplicaSetFixture(interface.ReplFixture, interface._DockerComposeInterface
         # teared down earlier, it must be skipped during those final checks.
         self.removeshard_teardown_marker = False
         self.removeshard_teardown_mutex = Lock()
+        # Track the number of times the fixture has been teared down. This can be used as a restart
+        # indicator for hooks that need to reset state upon a restart.
+        self.teardown_counter = 0
 
     def setup(self):
         """Set up the replica set."""
@@ -736,6 +739,7 @@ class ReplicaSetFixture(interface.ReplFixture, interface._DockerComposeInterface
 
         if teardown_handler.was_successful():
             self.logger.info("Successfully stopped all members of the replica set.")
+            self.teardown_counter += 1
         else:
             self.logger.error("Stopping the replica set fixture failed.")
             raise self.fixturelib.ServerFailure(teardown_handler.get_error_message())

buildscripts/resmokelib/testing/hooks/replicaset_transition_to_and_from_csrs.py

@@ -44,6 +44,7 @@ class ContinuousTransition(interface.Hook):
         interface.Hook.__init__(self, hook_logger, fixture, ContinuousTransition.DESCRIPTION)
 
         self._fixture = fixture
+        self._restart_counters = []
         self._rs_fixtures = []
         self._transition_interval_min_secs = float(transition_interval_min_ms) / 1000
         self._transition_interval_max_secs = float(transition_interval_max_ms) / 1000
@@ -102,6 +103,10 @@ class ContinuousTransition(interface.Hook):
 
     def before_test(self, test, test_report):
         """Before test."""
+        for idx, fixture in enumerate(self._rs_fixtures):
+            if fixture.teardown_counter != self._restart_counters[idx]:
+                self._transition_thread.reset(idx)
+                self._restart_counters[idx] = fixture.teardown_counter
         self.logger.info("Resuming the transition thread.")
         self._transition_thread.pause()
         self._transition_thread.resume()
@@ -122,6 +127,7 @@ class ContinuousTransition(interface.Hook):
             raise ValueError(
                 "Transition to and from CSRS hook cannot be specified on any fixture other than a ReplicaSetFixture."
             )
+        self._restart_counters.append(fixture.teardown_counter)
         self._rs_fixtures.append(fixture)
 
 
@@ -267,6 +273,9 @@ class _TransitionThread(threading.Thread):
         """Resume the thread."""
         self.__lifecycle.mark_test_started()
 
+    def reset(self, idx):
+        self._current_states[idx] = 0
+
     def _check_thread(self):
         if not self.is_alive():
             msg = "The transition thread is not running."