Add fallback from backjump to backtrack

Author: notatallshaw

src/resolvelib/reporters.py

@@ -60,3 +60,6 @@ def rejecting_candidate(
 
     def pinning(self, candidate: CT) -> None:
         """Called when adding a candidate to the potential solution."""
+
+    def fallback(self, from_: str, to: str) -> None:
+        """Called when falling back from one backtrack strategy to another."""

src/resolvelib/resolvers.py

@@ -11,6 +11,7 @@
     Iterable,
     Mapping,
     NamedTuple,
+    Optional,
 )
 
 from .providers import AbstractProvider
@@ -30,6 +31,9 @@
 
 if TYPE_CHECKING:
     from .providers import Preference
+    from typing_extensions import Literal
+
+    BacktrackStrategy = Literal["backjump_fallback", "backjump", "backtrack"]
 
     class Result(NamedTuple, Generic[RT, CT, KT]):
         mapping: Mapping[KT, CT]
@@ -100,9 +104,12 @@ def __init__(
         self,
         provider: AbstractProvider[RT, CT, KT],
         reporter: BaseReporter[RT, CT, KT],
+        backtrack_strategy: BacktrackStrategy = "backjump_fallback",
     ) -> None:
         self._p = provider
         self._r = reporter
+        self._backtrack_strategy: BacktrackStrategy = backtrack_strategy
+        self._fallback_states: Optional[list[State[RT, CT, KT]]] = None
         self._states: list[State[RT, CT, KT]] = []
 
     @property
@@ -269,6 +276,77 @@ def _attempt_to_pin_criterion(self, name: KT) -> list[Criterion[RT, CT]]:
         # end, signal for backtracking.
         return causes
 
+    def _backtrack_iteration(self) -> tuple[KT, CT, list[tuple[KT, list[CT]]]]:
+        broken_state = self._states.pop()
+        name, candidate = broken_state.mapping.popitem()
+        incompatibilities_from_broken = [
+            (k, list(v.incompatibilities))
+            for k, v in broken_state.criteria.items()
+        ]
+
+        return name, candidate, incompatibilities_from_broken
+
+    def _backjump_iteration(
+        self,
+        causes: list[RequirementInformation[RT, CT]],
+        incompatible_deps: set[KT],
+    ) -> tuple[KT, CT, list[tuple[KT, list[CT]]]]:
+        # Ensure to backtrack to a state that caused the incompatibility
+        incompatible_state = False
+        name, candidate, broken_state = None, None, None
+
+        if (
+            self._backtrack_strategy == "backjump_fallback"
+            and self._fallback_states is None
+        ):
+            fallback_states = [
+                State(
+                    s.mapping.copy(),
+                    s.criteria.copy(),
+                    s.backtrack_causes[:],
+                )
+                for s in self._states
+            ]
+        else:
+            fallback_states = None
+
+        backjump_count = 0
+        while not incompatible_state:
+            backjump_count += 1
+
+            # Retrieve the last candidate pin and known incompatibilities
+            try:
+                broken_state = self._states.pop()
+                name, candidate = broken_state.mapping.popitem()
+            except (IndexError, KeyError):
+                raise ResolutionImpossible(causes)
+            current_dependencies = {
+                self._p.identify(d)
+                for d in self._p.get_dependencies(candidate)
+            }
+            incompatible_state = not current_dependencies.isdisjoint(
+                incompatible_deps
+            )
+
+            # Backup states first time a backjump goes
+            # further than a backtrack would have
+            if (
+                self._backtrack_strategy == "backjump_fallback"
+                and self._fallback_states is None
+                and backjump_count == 2
+            ):
+                self._fallback_states = fallback_states
+
+        if name is None or candidate is None or broken_state is None:
+            raise ResolutionImpossible(causes)
+
+        incompatibilities_from_broken = [
+            (k, list(v.incompatibilities))
+            for k, v in broken_state.criteria.items()
+        ]
+
+        return name, candidate, incompatibilities_from_broken
+
     def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
         """Perform backjumping.
 
@@ -299,6 +377,17 @@ def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
         5a. If this causes Y' to conflict, we need to backtrack again. Make Y'
             the new Z and go back to step 2.
         5b. If the incompatibilities apply cleanly, end backtracking.
+
+        If backtracking each iteraction the the loop will:
+
+        1.  Discard Z.
+        2.  Discard Y but remember its incompatibility information gathered
+            previously, and the failure we're dealing with right now.
+        3.  Push a new state Y' based on X, and apply the incompatibility
+            information from Y to Y'.
+        4a. If this causes Y' to conflict, we need to backtrack again. Make Y'
+            the new Z and go back to step 2.
+        4b. If the incompatibilities apply cleanly, end backtracking.
         """
         incompatible_reqs: Iterable[CT | RT] = itertools.chain(
             (c.parent for c in causes if c.parent is not None),
@@ -309,28 +398,42 @@ def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
             # Remove the state that triggered backtracking.
             del self._states[-1]
 
-            # Ensure to backtrack to a state that caused the incompatibility
-            incompatible_state = False
-            broken_state = self.state
-            while not incompatible_state:
-                # Retrieve the last candidate pin and known incompatibilities.
-                try:
-                    broken_state = self._states.pop()
-                    name, candidate = broken_state.mapping.popitem()
-                except (IndexError, KeyError):
-                    raise ResolutionImpossible(causes) from None
-                current_dependencies = {
-                    self._p.identify(d)
-                    for d in self._p.get_dependencies(candidate)
-                }
-                incompatible_state = not current_dependencies.isdisjoint(
-                    incompatible_deps
-                )
+            name, candidate, incompatibilities_from_broken = None, None, None
 
-            incompatibilities_from_broken = [
-                (k, list(v.incompatibilities))
-                for k, v in broken_state.criteria.items()
-            ]
+            if self._backtrack_strategy in ("backjump", "backjump_fallback"):
+                try:
+                    (
+                        name,
+                        candidate,
+                        incompatibilities_from_broken,
+                    ) = self._backjump_iteration(
+                        causes=causes, incompatible_deps=incompatible_deps
+                    )
+                except ResolutionImpossible:
+                    if (
+                        self._backtrack_strategy == "backjump"
+                        or self._fallback_states is None
+                    ):
+                        raise
+
+                    # Backjumping failed but fallback to backtracking was requested
+                    self._states = self._fallback_states
+                    self._backtrack_strategy = "backtrack"
+                    self._r.fallback("backjump_fallback", "backtrack")
+
+            if self._backtrack_strategy == "backtrack":
+                (
+                    name,
+                    candidate,
+                    incompatibilities_from_broken,
+                ) = self._backtrack_iteration()
+
+            if (
+                name is None
+                or candidate is None
+                or incompatibilities_from_broken is None
+            ):
+                raise ResolutionImpossible(causes)
 
             # Also mark the newly known incompatibility.
             incompatibilities_from_broken.append((name, [candidate]))