fix: Fixing proxy_restart

`start_proxy` starts a process that proxifies the node in the
background, and immediatelly returns control to the main process.
If the main process then kills the node, it sets `local_stopped` to 1.
If `start_proxy` hasn't gotten to `check_finish` yet, `check_finish`
will observe `local_stopped` set to 1, and stop the server.
`start_proxy` will then proceed to calling `server.serve_forever`, which
will crash.
I'm fixing it by making `start_proxy` actually wait until it gets to
`serve_forever` before returning control

pytest/lib/proxy.py

@@ -185,7 +185,7 @@ async def stop_server(server):
 
 def check_finish(server, global_stopped, local_stopped, error):
     loop = asyncio.get_running_loop()
-    if 0 == global_stopped.value and 0 == local_stopped.value and 0 == error.value:
+    if 0 == global_stopped.value and 0 >= local_stopped.value and 0 == error.value:
         loop.call_later(1, check_finish, server,
                         global_stopped, local_stopped, error)
     else:
@@ -200,7 +200,7 @@ async def bridge(reader, writer, handler_fn, global_stopped, local_stopped, erro
     logging.debug(f"Start bridge. port={_MY_PORT} bridge_id={bridge_id}")
 
     try:
-        while 0 == global_stopped.value and 0 == local_stopped.value and 0 == error.value:
+        while 0 == global_stopped.value and 0 >= local_stopped.value and 0 == error.value:
             header = await reader.read(4)
             if not header:
                 writer.close()
@@ -261,7 +261,7 @@ async def handle_connection(outer_reader, outer_writer, inner_port, outer_port,
     except asyncio.CancelledError:
         logging.debug(
             f"Cancelled Error (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:
+        if local_stopped.value <= 0:
             global_stopped.value = 1
     except ConnectionRefusedError:
         logging.debug(
@@ -299,13 +299,14 @@ async def start_connection(reader, writer):
                 await asyncio.sleep(1)
 
         check_finish(server, global_stopped, local_stopped, error)
+        local_stopped.value = 0
 
         async with server:
             await server.serve_forever()
     except asyncio.CancelledError:
         logging.debug(
             f"Cancelled Error (listener). port={_MY_PORT} global_stopped={global_stopped.value} local_stopped={local_stopped.value} error={error.value}")
-        if local_stopped.value == 0:
+        if local_stopped.value <= 0:
             global_stopped.value = 1
     except:
         logging.debug(
@@ -326,14 +327,22 @@ def proxify_node(node, ps, handler, global_stopped, error, proxy):
 
     def start_proxy():
         # local_stopped denotes the current of state of the proxy:
+        # -1: The proxy hasn't started yet
         # 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)
+        local_stopped = multiprocessing.Value('i', -1)
         p = multiprocessing.Process(target=start_server, args=(
             inner_port, outer_port, handler, global_stopped, local_stopped, error))
         p.start()
         ps.append(p)
+        for attempt in range(3):
+            if local_stopped.value == 0:
+                break
+            time.sleep(1)
+        else:
+            error.value = 1
+            assert False, "The proxy failed to start after 3 seconds"
         return local_stopped
 
     node.port = outer_port