Fix various test flakiness on Windows (#21220)

It looks like there were two sources of flakes:
* In rare cases workers could connect to DB soon, before coordinator
configured WAL. I fix this by starting workers after creating metastore.
We will probably loose ~5-10ms on this, but I think it is worth it.
* Windows is generally very slow with processes, startup/shutdown
sometimes take few seconds. I therefore triple relevant timeouts on
Windows.

In addition:
* Handle more parallel workers startup failure scenarios ~gracefully.
* Add a unique build ID to worker status files.
* Add a bit more info in case of a failure.
* Fix one test with unstable error message order.

Author: ilevkivskyi

mypy/build.py

@@ -91,6 +91,7 @@
 from mypy.defaults import (
     WORKER_CONNECTION_TIMEOUT,
     WORKER_DONE_TIMEOUT,
+    WORKER_SHUTDOWN_TIMEOUT,
     WORKER_START_INTERVAL,
     WORKER_START_TIMEOUT,
 )
@@ -283,6 +284,7 @@ def __init__(self, status_file: str, options_data: str, env: Mapping[str, str])
         ]
         # Return early without waiting, caller must call connect() before using the client.
         self.proc = subprocess.Popen(command, env=env)
+        self.connected = False
 
     def connect(self) -> None:
         end_time = time.time() + WORKER_START_TIMEOUT
@@ -303,18 +305,19 @@ def connect(self) -> None:
                     # verify PIDs reliably.
                     assert pid == self.proc.pid, f"PID mismatch: {pid} vs {self.proc.pid}"
                 self.conn = IPCClient(connection_name, WORKER_CONNECTION_TIMEOUT)
+                self.connected = True
                 return
             except Exception as exc:
                 last_exception = exc
                 break
-        print("Failed to establish connection with worker:", last_exception)
-        sys.exit(2)
+        print(f"Failed to establish connection with worker: {last_exception}")
 
     def close(self) -> None:
-        self.conn.close()
+        if self.connected:
+            self.conn.close()
         # Technically we don't need to wait, but otherwise we will get ResourceWarnings.
         try:
-            self.proc.wait(timeout=1)
+            self.proc.wait(timeout=WORKER_SHUTDOWN_TIMEOUT)
         except subprocess.TimeoutExpired:
             pass
         if os.path.isfile(self.status_file):
@@ -346,7 +349,7 @@ def build(
 
     If a flush_errors callback is provided, all error messages will be
     passed to it and the errors and messages fields of BuildResult and
-    CompileError (respectively) will be empty. Otherwise those fields will
+    CompileError (respectively) will be empty. Otherwise, those fields will
     report any error messages.
 
     Args:
@@ -356,6 +359,9 @@ def build(
         (takes precedence over other directories)
       flush_errors: optional function to flush errors after a file is processed
       fscache: optionally a file-system cacher
+      stdout: Output stream to use instead of `sys.stdout`
+      stderr: Error stream to use instead of `sys.stderr`
+      extra_plugins: Plugins to use in addition to those loaded from config
       worker_env: An environment to start parallel build workers (used for tests)
     """
     # If we were not given a flush_errors, we use one that will populate those
@@ -376,14 +382,20 @@ def default_flush_errors(
     stderr = stderr or sys.stderr
     extra_plugins = extra_plugins or []
 
+    # Create metastore before workers to avoid race conditions.
+    metastore = create_metastore(options, parallel_worker=False)
     workers = []
     connect_threads = []
+    # A quasi-unique ID for this specific mypy invocation.
+    build_id = os.urandom(4).hex()
     if options.num_workers > 0:
         # TODO: switch to something more efficient than pickle (also in the daemon).
         pickled_options = pickle.dumps(options.snapshot())
         options_data = b64encode(pickled_options).decode()
         workers = [
-            WorkerClient(f".mypy_worker.{idx}.json", options_data, worker_env or os.environ)
+            WorkerClient(
+                f".mypy_worker.{build_id}.{idx}.json", options_data, worker_env or os.environ
+            )
             for idx in range(options.num_workers)
         ]
         sources_message = SourcesDataMessage(sources=sources)
@@ -394,6 +406,9 @@ def default_flush_errors(
         def connect(wc: WorkerClient, data: bytes) -> None:
             # Start loading sources in each worker as soon as it is up.
             wc.connect()
+            if not wc.connected:
+                # Caller should detect this and fail gracefully.
+                return
             wc.conn.write_bytes(data)
 
         # We don't wait for workers to be ready until they are actually needed.
@@ -414,6 +429,7 @@ def connect(wc: WorkerClient, data: bytes) -> None:
             extra_plugins,
             workers,
             connect_threads,
+            metastore,
         )
         result.errors = messages
         return result
@@ -432,6 +448,8 @@ def connect(wc: WorkerClient, data: bytes) -> None:
         for thread in connect_threads:
             thread.join()
         for worker in workers:
+            if not worker.connected:
+                continue
             try:
                 send(worker.conn, SccRequestMessage(scc_id=None, import_errors={}, mod_data={}))
             except (OSError, IPCException):
@@ -451,6 +469,7 @@ def build_inner(
     extra_plugins: Sequence[Plugin],
     workers: list[WorkerClient],
     connect_threads: list[Thread],
+    metastore: MetadataStore,
 ) -> BuildResult:
     if platform.python_implementation() == "CPython":
         # Run gc less frequently, as otherwise we can spend a large fraction of
@@ -499,6 +518,7 @@ def build_inner(
         fscache=fscache,
         stdout=stdout,
         stderr=stderr,
+        metastore=metastore,
     )
     manager.workers = workers
     if manager.verbosity() >= 2:
@@ -816,6 +836,7 @@ def __init__(
         stderr: TextIO,
         error_formatter: ErrorFormatter | None = None,
         parallel_worker: bool = False,
+        metastore: MetadataStore | None = None,
     ) -> None:
         self.stats: dict[str, Any] = {}  # Values are ints or floats
         # Use in cases where we need to prevent race conditions in stats reporting.
@@ -903,7 +924,9 @@ def __init__(
                 ]
             )
 
-        self.metastore = create_metastore(options, parallel_worker=parallel_worker)
+        if metastore is None:
+            metastore = create_metastore(options, parallel_worker=parallel_worker)
+        self.metastore = metastore
 
         # a mapping from source files to their corresponding shadow files
         # for efficient lookup
@@ -3983,6 +4006,9 @@ def dispatch(
         # Wait for workers since they may be needed at this point.
         for thread in connect_threads:
             thread.join()
+        not_connected = [str(idx) for idx, wc in enumerate(manager.workers) if not wc.connected]
+        if not_connected:
+            raise OSError(f"Cannot connect to build worker(s): {', '.join(not_connected)}")
         process_graph(graph, manager)
         # Update plugins snapshot.
         write_plugins_snapshot(manager)

mypy/defaults.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import sys
 from typing import Final
 
 # Earliest fully supported Python 3.x version. Used as the default Python
@@ -45,8 +46,12 @@
 
 RECURSION_LIMIT: Final = 2**14
 
-WORKER_START_INTERVAL: Final = 0.01
-WORKER_START_TIMEOUT: Final = 3
+# It looks like Windows is slow with processes, causing test flakiness even
+# with our generous timeouts, so we set them higher.
+WORKER_START_INTERVAL: Final = 0.01 if sys.platform != "win32" else 0.03
+WORKER_START_TIMEOUT: Final = 3 if sys.platform != "win32" else 10
+WORKER_SHUTDOWN_TIMEOUT: Final = 1 if sys.platform != "win32" else 3
+
 WORKER_CONNECTION_TIMEOUT: Final = 10
 WORKER_IDLE_TIMEOUT: Final = 600
 WORKER_DONE_TIMEOUT: Final = 600

test-data/unit/check-final.test

@@ -60,8 +60,6 @@ reveal_type(C().x)  # N: Revealed type is "builtins.float"
 [out]
 
 [case testFinalInvalidDefinitions]
-
-# Errors are shown in a different order with the new analyzer.
 from typing import Final, Any
 
 x = y = 1  # type: Final[float]  # E: Invalid final declaration
@@ -432,7 +430,6 @@ y: Final = 3  # E: Cannot redefine an existing name as final
 
 [case testFinalReassignModuleVar3]
 # flags: --disallow-redefinition
-# Error formatting is subtly different with new analyzer.
 from typing import Final
 
 x: Final = 1
@@ -459,16 +456,14 @@ z: Final = 2  # E: Cannot redefine an existing name as final
 z = 3  # E: Cannot assign to final name "z"
 
 [case testFinalReassignModuleReexport]
-
-# Error formatting is subtly different with the new analyzer.
 from typing import Final
 
 from lib import X
 from lib.mod import ID
 
-X = 1  # Error!
-ID: Final = 1  # Two errors!
-ID = 1  # Error!
+X = 1  # E: Cannot assign to final name "X"
+ID: Final = 1  # E: Cannot redefine an existing name as final
+ID = 1  # E: Cannot assign to final name "ID"
 [file lib/__init__.pyi]
 from lib.const import X as X
 
@@ -478,13 +473,8 @@ from lib.const import *
 [file lib/const.pyi]
 from typing import Final
 
-ID: Final  # Error!
+ID: Final  # E: Type in Final[...] can only be omitted if there is an initializer
 X: Final[int]
-[out]
-tmp/lib/const.pyi:3: error: Type in Final[...] can only be omitted if there is an initializer
-main:8: error: Cannot assign to final name "X"
-main:9: error: Cannot redefine an existing name as final
-main:10: error: Cannot assign to final name "ID"
 
 [case testFinalReassignFuncScope]
 from typing import Final