Verify indirect deps reachable on incremental run (#20735)

Fixes #19477

This is another surprising correctness hole in the indirect dependencies
logic. We implicitly assume everywhere that indirect dependencies are
subset of transitive ones. Although it is true when the indirect
dependencies are initially calculated, there is no guarantee this will
stay true after an incremental update where some import structure has
changed.

Currently, we only handle the situations where an indirect dependency is
removed from the graph completely, but if it is still in the graph, but
now brought in by a _different_ dependent, this may cause out-of-order
cache loading, and thus a bug or a crash. It is interesting that
although in theory it is kind of a big deal, in practice it is very hard
to create such scenario. So far I found only two complicated scenarios
(see tests).

Implementation note: although this additional check is needed for rare
edge cases, it is quite computationally expensive. I tried few options
(including only computing import deltas), but the only viable/robust
option seems to be an honest recursive check with some caching. I tested
this on `torch` (it has ~1000 small SCCs plus single SCC with around
1000 modules), and the performance penalty is negligible there. So
hopefully it should be fine even for large code bases.

Author: ilevkivskyi

mypy/build.py

@@ -870,6 +870,9 @@ def __init__(
         # Mapping from SCC id to corresponding SCC instance. This is populated
         # in process_graph().
         self.scc_by_id: dict[int, SCC] = {}
+        # Mapping from module id to the SCC it belongs to. This is populated
+        # in process_graph().
+        self.scc_by_mod_id: dict[str, SCC] = {}
         # Global topological order for SCCs. This exists to make order of processing
         # SCCs more predictable.
         self.top_order: list[int] = []
@@ -892,6 +895,8 @@ def __init__(
         # raw parsed trees not analyzed with mypy. We use these to find absolute
         # location of a symbol used as a location for an error message.
         self.extra_trees: dict[str, MypyFile] = {}
+        # Cache for transitive dependency check (expensive).
+        self.transitive_deps_cache: dict[tuple[int, int], bool] = {}
 
     def dump_stats(self) -> None:
         if self.options.dump_build_stats:
@@ -1203,6 +1208,35 @@ def wait_for_done_workers(self) -> tuple[list[SCC], bool, dict[str, tuple[str, l
             results,
         )
 
+    def is_transitive_scc_dep(self, from_scc_id: int, to_scc_id: int) -> bool:
+        """Check if one SCC is a (transitive) dependency of another."""
+        edge = (from_scc_id, to_scc_id)
+        if (cached := self.transitive_deps_cache.get(edge)) is not None:
+            return cached
+        todo = self.scc_by_id[from_scc_id].deps
+        seen = set()
+        while todo:
+            more = set()
+            # Breadth-first search seems to be better here, because all
+            # "lower-level" SCCs are processed and some may be cached.
+            for dep in todo:
+                seen.add(dep)
+                if dep == to_scc_id:
+                    self.transitive_deps_cache[edge] = True
+                    return True
+                if cached := self.transitive_deps_cache.get((dep, to_scc_id)):
+                    self.transitive_deps_cache[edge] = True
+                    return True
+                elif cached is None:
+                    more |= self.scc_by_id[dep].deps
+            todo = more
+        self.transitive_deps_cache[edge] = False
+        for dep in seen:
+            # We negative-cache all intermediate lookups, thus
+            # trading time for space.
+            self.transitive_deps_cache[(dep, to_scc_id)] = False
+        return False
+
 
 def deps_to_json(x: dict[str, set[str]]) -> bytes:
     return json_dumps({k: list(v) for k, v in x.items()})
@@ -1841,6 +1875,7 @@ def write_cache(
     dep_prios: list[int],
     dep_lines: list[int],
     old_interface_hash: bytes,
+    trans_dep_hash: bytes,
     source_hash: str,
     ignore_all: bool,
     manager: BuildManager,
@@ -1957,6 +1992,7 @@ def write_cache(
         dep_prios=dep_prios,
         dep_lines=dep_lines,
         interface_hash=interface_hash,
+        trans_dep_hash=trans_dep_hash,
         version_id=manager.version_id,
         ignore_all=ignore_all,
         plugin_data=plugin_data,
@@ -2175,6 +2211,12 @@ class State:
     # Contains a hash of the public interface in incremental mode
     interface_hash: bytes = b""
 
+    # Hash of import structure that this module depends on. It is not 1:1 with
+    # transitive dependencies set, but if two hashes are equal, transitive
+    # dependencies are guaranteed to be identical. Some expensive checks can be
+    # skipped if this value is unchanged for a module.
+    trans_dep_hash: bytes = b""
+
     # Options, specialized for this file
     options: Options
 
@@ -2322,15 +2364,15 @@ def new_state(
             if temporary:
                 state.load_tree(temporary=True)
             if not manager.use_fine_grained_cache():
-                # Special case: if there were a previously missing package imported here
+                # Special case: if there were a previously missing package imported here,
                 # and it is not present, then we need to re-calculate dependencies.
                 # This is to support patterns like this:
                 #     from missing_package import missing_module  # type: ignore
                 # At first mypy doesn't know that `missing_module` is a module
                 # (it may be a variable, a class, or a function), so it is not added to
                 # suppressed dependencies. Therefore, when the package with module is added,
                 # we need to re-calculate dependencies.
-                # NOTE: see comment below for why we skip this in fine grained mode.
+                # NOTE: see comment below for why we skip this in fine-grained mode.
                 if exist_added_packages(suppressed, manager, options):
                     state.parse_file()  # This is safe because the cache is anyway stale.
                     state.compute_dependencies()
@@ -2350,6 +2392,7 @@ def new_state(
                 # We don't need parsed trees in coordinator process, we parse only to
                 # compute dependencies.
                 state.tree = None
+                del manager.ast_cache[id]
 
         return state
 
@@ -3012,6 +3055,7 @@ def write_cache(self) -> tuple[CacheMeta, str] | None:
             dep_prios,
             dep_lines,
             self.interface_hash,
+            self.trans_dep_hash,
             self.source_hash,
             self.ignore_all,
             self.manager,
@@ -3774,6 +3818,27 @@ def order_ascc_ex(graph: Graph, ascc: SCC) -> list[str]:
     return scc
 
 
+def verify_transitive_deps(ascc: SCC, graph: Graph, manager: BuildManager) -> str | None:
+    """Verify all indirect dependencies of this SCC are still reachable via direct ones.
+
+    Return first unreachable dependency id, or None.
+    """
+    for id in ascc.mod_ids:
+        st = graph[id]
+        assert st.meta is not None, "Must be called on fresh SCCs only"
+        if st.trans_dep_hash == st.meta.trans_dep_hash:
+            # Import graph unchanged, skip this module.
+            continue
+        for dep in st.dependencies:
+            if st.priorities.get(dep) == PRI_INDIRECT:
+                dep_scc_id = manager.scc_by_mod_id[dep].id
+                if dep_scc_id == ascc.id:
+                    continue
+                if not manager.is_transitive_scc_dep(ascc.id, dep_scc_id):
+                    return dep
+    return None
+
+
 def find_stale_sccs(
     sccs: list[SCC], graph: Graph, manager: BuildManager
 ) -> tuple[list[SCC], list[SCC]]:
@@ -3782,7 +3847,8 @@ def find_stale_sccs(
     Fresh SCCs are those where:
     * We have valid cache files for all modules in the SCC.
     * There are no changes in dependencies (files removed from/added to the build).
-    * The interface hashes of direct dependents matches those recorded in the cache.
+    * The interface hashes of dependencies matches those recorded in the cache.
+    * All indirect dependencies are still reachable via direct ones.
     The first and second conditions are verified by is_fresh().
     """
     stale_sccs = []
@@ -3799,6 +3865,15 @@ def find_stale_sccs(
                     stale_deps.add(dep)
         fresh = fresh and not stale_deps
 
+        # Verify the invariant that indirect dependencies are a subset of transitive direct
+        # dependencies. Note: the case where indirect dependency is removed from the graph
+        # completely is already handled above.
+        stale_indirect = None
+        if fresh:
+            stale_indirect = verify_transitive_deps(ascc, graph, manager)
+            if stale_indirect is not None:
+                fresh = False
+
         if fresh:
             fresh_msg = "fresh"
         elif stale_scc:
@@ -3807,8 +3882,11 @@ def find_stale_sccs(
                 fresh_msg += f" ({' '.join(sorted(stale_scc))})"
             if stale_deps:
                 fresh_msg += f" with stale deps ({' '.join(sorted(stale_deps))})"
-        else:
+        elif stale_deps:
             fresh_msg = f"stale due to deps ({' '.join(sorted(stale_deps))})"
+        else:
+            assert stale_indirect is not None
+            fresh_msg = f"stale due to stale indirect dep(s): first {stale_indirect}"
 
         scc_str = " ".join(ascc.mod_ids)
         if fresh:
@@ -3860,6 +3938,9 @@ def process_graph(graph: Graph, manager: BuildManager) -> None:
     scc_by_id = {scc.id: scc for scc in sccs}
     manager.scc_by_id = scc_by_id
     manager.top_order = [scc.id for scc in sccs]
+    for scc in sccs:
+        for mod_id in scc.mod_ids:
+            manager.scc_by_mod_id[mod_id] = scc
 
     # Broadcast SCC structure to the parallel workers, since they don't compute it.
     sccs_message = SccsDataMessage(sccs=sccs)
@@ -3904,8 +3985,8 @@ def process_graph(graph: Graph, manager: BuildManager) -> None:
             # type-checking this is already done and results should be empty here.
             if not manager.workers:
                 assert not results
-            for id, (interface_cache, errors) in results.items():
-                new_hash = bytes.fromhex(interface_cache)
+            for id, (interface_hash, errors) in results.items():
+                new_hash = bytes.fromhex(interface_hash)
                 if new_hash != graph[id].interface_hash:
                     graph[id].mark_interface_stale()
                     graph[id].interface_hash = new_hash
@@ -3917,6 +3998,7 @@ def process_graph(graph: Graph, manager: BuildManager) -> None:
                 if not scc_by_id[dependent].not_ready_deps:
                     not_ready.remove(scc_by_id[dependent])
                     ready.append(scc_by_id[dependent])
+    manager.trace(f"Transitive deps cache size: {sys.getsizeof(manager.transitive_deps_cache)}")
 
 
 def order_ascc(graph: Graph, ascc: AbstractSet[str], pri_max: int = PRI_INDIRECT) -> list[str]:
@@ -4168,6 +4250,11 @@ def sorted_components(graph: Graph) -> list[SCC]:
             scc.size_hint = sum(graph[mid].size_hint for mid in scc.mod_ids)
             for dep in scc_dep_map[scc]:
                 dep.direct_dependents.append(scc.id)
+            # We compute dependencies hash here since we know no direct
+            # dependencies will be added or suppressed after this point.
+            trans_dep_hash = transitive_dep_hash(scc, graph)
+            for id in scc.mod_ids:
+                graph[id].trans_dep_hash = trans_dep_hash
         res.extend(sorted_ready)
     return res
 
@@ -4201,6 +4288,21 @@ def deps_filtered(graph: Graph, vertices: AbstractSet[str], id: str, pri_max: in
     ]
 
 
+def transitive_dep_hash(scc: SCC, graph: Graph) -> bytes:
+    """Compute stable snapshot of transitive import structure for given SCC."""
+    all_direct_deps = {
+        dep
+        for id in scc.mod_ids
+        for dep in graph[id].dependencies
+        if graph[id].priorities.get(dep) != PRI_INDIRECT
+    }
+    trans_dep_hash_map = {
+        dep_id: "" if dep_id in scc.mod_ids else graph[dep_id].trans_dep_hash.hex()
+        for dep_id in all_direct_deps
+    }
+    return hash_digest_bytes(json_dumps(trans_dep_hash_map))
+
+
 def missing_stubs_file(cache_dir: str) -> str:
     return os.path.join(cache_dir, "missing_stubs")
 

mypy/cache.py

@@ -69,7 +69,7 @@
 from mypy_extensions import u8
 
 # High-level cache layout format
-CACHE_VERSION: Final = 2
+CACHE_VERSION: Final = 3
 
 SerializedError: _TypeAlias = tuple[str | None, int | str, int, int, int, str, str, str | None]
 
@@ -95,6 +95,7 @@ def __init__(
         dep_lines: list[int],
         dep_hashes: list[bytes],
         interface_hash: bytes,
+        trans_dep_hash: bytes,
         error_lines: list[SerializedError],
         version_id: str,
         ignore_all: bool,
@@ -117,6 +118,7 @@ def __init__(
         # dep_hashes list is aligned with dependencies only
         self.dep_hashes = dep_hashes  # list of interface_hash for dependencies
         self.interface_hash = interface_hash  # hash representing the public interface
+        self.trans_dep_hash = trans_dep_hash  # hash of import structure (transitive)
         self.error_lines = error_lines
         self.version_id = version_id  # mypy version for cache invalidation
         self.ignore_all = ignore_all  # if errors were ignored
@@ -138,6 +140,7 @@ def serialize(self) -> dict[str, Any]:
             "dep_lines": self.dep_lines,
             "dep_hashes": [dep.hex() for dep in self.dep_hashes],
             "interface_hash": self.interface_hash.hex(),
+            "trans_dep_hash": self.trans_dep_hash.hex(),
             "error_lines": self.error_lines,
             "version_id": self.version_id,
             "ignore_all": self.ignore_all,
@@ -165,6 +168,7 @@ def deserialize(cls, meta: dict[str, Any], data_file: str) -> CacheMeta | None:
                 dep_lines=meta["dep_lines"],
                 dep_hashes=[bytes.fromhex(dep) for dep in meta["dep_hashes"]],
                 interface_hash=bytes.fromhex(meta["interface_hash"]),
+                trans_dep_hash=bytes.fromhex(meta["trans_dep_hash"]),
                 error_lines=[tuple(err) for err in meta["error_lines"]],
                 version_id=meta["version_id"],
                 ignore_all=meta["ignore_all"],
@@ -191,6 +195,7 @@ def write(self, data: WriteBuffer) -> None:
         write_int_list(data, self.dep_lines)
         write_bytes_list(data, self.dep_hashes)
         write_bytes(data, self.interface_hash)
+        write_bytes(data, self.trans_dep_hash)
         write_errors(data, self.error_lines)
         write_str(data, self.version_id)
         write_bool(data, self.ignore_all)
@@ -219,6 +224,7 @@ def read(cls, data: ReadBuffer, data_file: str) -> CacheMeta | None:
                 dep_lines=read_int_list(data),
                 dep_hashes=read_bytes_list(data),
                 interface_hash=read_bytes(data),
+                trans_dep_hash=read_bytes(data),
                 error_lines=read_errors(data),
                 version_id=read_str(data),
                 ignore_all=read_bool(data),

mypy/semanal_main.py

@@ -31,7 +31,6 @@
 from itertools import groupby
 from typing import TYPE_CHECKING, Final, TypeAlias as _TypeAlias
 
-import mypy.build
 import mypy.state
 from mypy.checker import FineGrainedDeferredNode
 from mypy.errors import Errors
@@ -416,11 +415,6 @@ def semantic_analyze_target(
         )
         if isinstance(node, Decorator):
             infer_decorator_signature_if_simple(node, analyzer)
-    for dep in analyzer.imports:
-        state.add_dependency(dep)
-        priority = mypy.build.PRI_LOW
-        if priority <= state.priorities.get(dep, priority):
-            state.priorities[dep] = priority
 
     # Clear out some stale data to avoid memory leaks and astmerge
     # validity check confusion

mypy/test/testgraph.py

@@ -64,9 +64,10 @@ def test_sorted_components(self) -> None:
             "d": State.new_state("d", None, "pass", manager),
             "b": State.new_state("b", None, "import c", manager),
             "c": State.new_state("c", None, "import b, d", manager),
+            "builtins": State.new_state("builtins", None, "", manager),
         }
         res = [scc.mod_ids for scc in sorted_components(graph)]
-        assert_equal(res, [{"d"}, {"c", "b"}, {"a"}])
+        assert_equal(res, [{"builtins"}, {"d"}, {"c", "b"}, {"a"}])
 
     def test_order_ascc(self) -> None:
         manager = self._make_manager()
@@ -75,9 +76,10 @@ def test_order_ascc(self) -> None:
             "d": State.new_state("d", None, "def f(): import a", manager),
             "b": State.new_state("b", None, "import c", manager),
             "c": State.new_state("c", None, "import b, d", manager),
+            "builtins": State.new_state("builtins", None, "", manager),
         }
         res = [scc.mod_ids for scc in sorted_components(graph)]
-        assert_equal(res, [frozenset({"a", "d", "c", "b"})])
-        ascc = res[0]
+        assert_equal(res, [{"builtins"}, {"a", "d", "c", "b"}])
+        ascc = res[1]
         scc = order_ascc(graph, ascc)
         assert_equal(scc, ["d", "c", "b", "a"])