fix(proxy_restart): Consistent way to restart nodes with proxy (#3272)

On proxy restart wait for the previous instance to get closed before starting a new one

fixes #3134 


Test plan
=======
sanity/proxy_restart.py pass consistently

http://nayduck.eastus.cloudapp.azure.com:3000/#/run/285

pytest/lib/cluster.py

@@ -18,6 +18,7 @@
 import traceback
 import uuid
 import network
+import logging
 from proxy import NodesProxy
 
 os.environ["ADVERSARY_CONSENT"] = "1"
@@ -302,6 +303,12 @@ def rpc_addr(self):
         return ("127.0.0.1", self.rpc_port)
 
     def start(self, boot_key, boot_node_addr):
+        if self._proxy_local_stopped is not None:
+            while self._proxy_local_stopped.value != 2:
+                logging.debug(f'Waiting for previous proxy instance to close')
+                time.sleep(1)
+
+
         env = os.environ.copy()
         env["RUST_BACKTRACE"] = "1"
         env["RUST_LOG"] = "actix_web=warn,mio=warn,tokio_util=warn,actix_server=warn,actix_http=warn," + env.get("RUST_LOG", "debug")

pytest/lib/proxy.py

@@ -192,6 +192,7 @@ def check_finish(server, global_stopped, local_stopped, error):
         logging.debug(
             f"Stopping server. port={_MY_PORT}, global_stopped={global_stopped.value}, local_stopped={local_stopped.value}, error={error.value}")
         server.close()
+        local_stopped.value = 2
 
 
 async def bridge(reader, writer, handler_fn, global_stopped, local_stopped, error):
@@ -265,8 +266,6 @@ async def handle_connection(outer_reader, outer_writer, inner_port, outer_port,
     except ConnectionRefusedError:
         logging.debug(
             f"ConnectionRefusedError (handle_connection). port={_MY_PORT} connection_id={connection_id} global_stopped={global_stopped.value} local_stopped={local_stopped.value} error={error.value}")
-        if local_stopped.value == 0:
-            global_stopped.value = 1
     except:
         logging.debug(
             f"Other Error (handle_connection). port={_MY_PORT} connection_id={connection_id} global_stopped={global_stopped.value} local_stopped={local_stopped.value} error={error.value}")
@@ -326,6 +325,10 @@ def proxify_node(node, ps, handler, global_stopped, error, proxy):
     outer_port = inner_port + 100
 
     def start_proxy():
+        # local_stopped denotes the current of state of the proxy:
+        # 0: The proxy is running
+        # 1: The proxy is running but should be closed soon
+        # 2: The proxy is closed
         local_stopped = multiprocessing.Value('i', 0)
         p = multiprocessing.Process(target=start_server, args=(
             inner_port, outer_port, handler, global_stopped, local_stopped, error))

pytest/tests/sanity/proxy_restart.py

@@ -8,13 +8,12 @@
 from peer import *
 from proxy import ProxyHandler
 
-TIMEOUT = 30
+TIMEOUT = 40
 TARGET_HEIGHT = 20
 
 nodes = start_cluster(2, 0, 1, None, [], {}, ProxyHandler)
 
 nodes[1].kill()
-time.sleep(2)
 
 nodes[1].start(nodes[0].node_key.pk, nodes[0].addr())
 started = time.time()