Isolate the active-HighState stack per execution context (#63056)

changelog/63056.fixed.md

@@ -0,0 +1 @@
+Fixed a race in concurrent state/orchestration renders where the active-HighState stack was shared on the class, so parallel reactor renders corrupted one another and failed with ``IndexError`` (empty pydsl render stack) or ``KeyError: '__env__'`` (spurious conflicting-ID). The stack and the cached pydsl top-file matches are now isolated per execution context.

salt/client/ssh/state.py

@@ -140,9 +140,7 @@ def __init__(
 
         self._pydsl_all_decls = {}
         self._pydsl_render_stack = []
-
-    def push_active(self):
-        salt.state.HighState.stack.append(self)
+        self._pydsl_sls_matches = None
 
     def load_dynamic(self, matches):
         """

salt/state.py

@@ -11,6 +11,16 @@
       }
 """
 
+try:
+    import contextvars
+except (ImportError, SyntaxError):
+    # Some salt-ssh targets (notably Python 3.6, which lacks a stdlib
+    # contextvars) resolve ``import contextvars`` to the thin's bundled backport,
+    # which pulls in immutables/typing_extensions that can be missing or
+    # syntactically incompatible on the target. Degrade to a shared stack there
+    # -- salt-ssh runs one execution per target, so the per-context isolation
+    # contextvars provides (#63056) is not needed.
+    contextvars = None
 import copy
 import datetime
 import fnmatch
@@ -3910,6 +3920,52 @@ def __init__(self, opts):
         self.avail = self.__gather_avail()
         self.building_highstate = HashableOrderedDict()
 
+    # Fallback shared stack, used only when contextvars is unavailable (see the
+    # guarded import at the top of the module). salt-ssh runs one execution per
+    # target, so a shared stack there is safe.
+    _shared_active_stack = []
+
+    @classmethod
+    def _active_stack(cls):
+        if _active_highstates is None:
+            return BaseHighState._shared_active_stack
+        # The active-HighState stack for the current execution context, created
+        # lazily on first use. A default= on the ContextVar would share one
+        # list object across every context, defeating the isolation, so the
+        # per-context list is set explicitly here instead.
+        try:
+            return _active_highstates.get()
+        except LookupError:
+            stack = []
+            _active_highstates.set(stack)
+            return stack
+
+    def push_active(self):
+        self._active_stack().append(self)
+
+    @classmethod
+    def clear_active(cls):
+        # Nuclear option
+        #
+        # Blow away the active-HighState stack for the current execution
+        # context. Used primarily by the test runner but also useful in custom
+        # wrappers of the HighState class, to reset the stack to a fresh state.
+        if _active_highstates is None:
+            BaseHighState._shared_active_stack.clear()
+        else:
+            _active_highstates.set([])
+
+    @classmethod
+    def pop_active(cls):
+        cls._active_stack().pop()
+
+    @classmethod
+    def get_active(cls):
+        try:
+            return cls._active_stack()[-1]
+        except IndexError:
+            return None
+
     def __gather_avail(self):
         """
         Lazily gather the lists of available sls data from the master
@@ -4863,10 +4919,10 @@ def merge_included_states(self, highstate, state, errors):
                         " conflicting ID is '{}' and is found in SLS"
                         " '{}:{}' and SLS '{}:{}'".format(
                             id_,
-                            highstate[id_]["__env__"],
-                            highstate[id_]["__sls__"],
-                            state[id_]["__env__"],
-                            state[id_]["__sls__"],
+                            highstate[id_].get("__env__"),
+                            highstate[id_].get("__sls__"),
+                            state[id_].get("__env__"),
+                            state[id_].get("__sls__"),
                         )
                     )
         try:
@@ -5083,15 +5139,27 @@ def __exit__(self, *_):
         self.destroy()
 
 
+# The stack of active HighState objects during a state run is kept per
+# execution context rather than on the class, so concurrent runs -- e.g.
+# reactor orchestrations rendered in parallel reactor worker threads -- each
+# get their own stack instead of corrupting a shared class-level list
+# (#63056). This mirrors salt.loader's loader_ctxvar.
+if contextvars is not None:
+    _active_highstates = contextvars.ContextVar("salt_active_highstates")
+else:
+    _active_highstates = None
+
+
 class HighState(BaseHighState):
     """
     Generate and execute the salt "High State". The High State is the
     compound state derived from a group of template files stored on the
     salt master or in the local cache.
     """
 
-    # a stack of active HighState objects during a state.highstate run
-    stack = []
+    # The stack of active HighState objects during a state run is stored per
+    # execution context in the module-level ``_active_highstates`` ContextVar;
+    # see ``_active_stack`` and the push/pop/get/clear accessors below.
 
     def __init__(
         self,
@@ -5151,28 +5219,10 @@ def __init__(
         # a stack of current rendering Sls objects, maintained and used by the pydsl renderer.
         self._pydsl_render_stack = []
 
-    def push_active(self):
-        self.stack.append(self)
-
-    @classmethod
-    def clear_active(cls):
-        # Nuclear option
-        #
-        # Blow away the entire stack. Used primarily by the test runner but also
-        # useful in custom wrappers of the HighState class, to reset the stack
-        # to a fresh state.
-        cls.stack = []
-
-    @classmethod
-    def pop_active(cls):
-        cls.stack.pop()
-
-    @classmethod
-    def get_active(cls):
-        try:
-            return cls.stack[-1]
-        except IndexError:
-            return None
+        # cached top-file matches for pydsl includes, computed once per run.
+        # Held on the instance (not a module global) so concurrent runs do not
+        # share one another's matches (#63056).
+        self._pydsl_sls_matches = None
 
     def destroy(self):
         if not self.preserve_client:

salt/utils/pydsl.py

@@ -103,9 +103,6 @@ def __getattr__(self, name):
         return self.get(name)
 
 
-SLS_MATCHES = None
-
-
 class Sls:
     def __init__(self, sls, saltenv, rendered_sls):
         self.name = sls
@@ -146,9 +143,13 @@ def include(self, *sls_names, **kws):
 
         HIGHSTATE = HighState.get_active()
 
-        global SLS_MATCHES
-        if SLS_MATCHES is None:
-            SLS_MATCHES = HIGHSTATE.top_matches(HIGHSTATE.get_top())
+        # Cache the top-file matches on the active HighState instead of a module
+        # global so concurrent runs don't share each other's matches (#63056).
+        sls_matches = HIGHSTATE._pydsl_sls_matches
+        if sls_matches is None:
+            sls_matches = HIGHSTATE._pydsl_sls_matches = HIGHSTATE.top_matches(
+                HIGHSTATE.get_top()
+            )
 
         highstate = self.included_highstate
         slsmods = []  # a list of pydsl sls modules rendered.
@@ -159,7 +160,7 @@ def include(self, *sls_names, **kws):
                     sls
                 )  # needed in case the starting sls uses the pydsl renderer.
                 histates, errors = HIGHSTATE.render_state(
-                    sls, saltenv, self.rendered_sls, SLS_MATCHES
+                    sls, saltenv, self.rendered_sls, sls_matches
                 )
                 HIGHSTATE.merge_included_states(highstate, histates, errors)
                 if errors:

tests/pytests/unit/state/test_active_highstate_stack.py

@@ -0,0 +1,157 @@
+"""
+Tests for the per-execution-context active-HighState stack (#63056).
+
+Concurrent state runs -- e.g. reactor orchestrations rendered in parallel
+reactor worker threads -- previously shared a single class-level
+``HighState.stack`` list. One run's ``push_active`` was therefore visible to
+another, so ``HighState.get_active`` could return the wrong HighState in the
+middle of a render (surfacing downstream as ``IndexError`` popping an empty
+pydsl render stack, or ``KeyError: '__env__'`` from a spuriously detected
+conflicting ID). The stack is now isolated per execution context via a
+ContextVar.
+"""
+
+import threading
+
+import salt.state
+from tests.support.mock import patch
+
+
+class _Marker(salt.state.HighState):
+    # A cheap stand-in that skips HighState's heavy __init__ but inherits the
+    # real push/pop/get/clear accessors under test.
+    def __init__(self, tag):  # pylint: disable=super-init-not-called
+        self.tag = tag
+
+
+def test_active_stack_push_pop_get_clear():
+    HighState = salt.state.HighState
+    HighState.clear_active()
+    assert HighState.get_active() is None
+
+    a = _Marker("a")
+    b = _Marker("b")
+
+    a.push_active()
+    assert HighState.get_active() is a
+    b.push_active()
+    assert HighState.get_active() is b
+    b.pop_active()
+    assert HighState.get_active() is a
+    a.pop_active()
+    assert HighState.get_active() is None
+
+    # clear_active() resets the stack for the current context.
+    a.push_active()
+    HighState.clear_active()
+    assert HighState.get_active() is None
+
+
+def test_active_stack_isolated_across_threads():
+    HighState = salt.state.HighState
+    HighState.clear_active()
+
+    results = {}
+    # Two barriers make the failure deterministic on a *shared* stack: every
+    # thread pushes before any reads (so the shared top holds both markers),
+    # and every thread reads before any pops (so a fast pop can't restore the
+    # reader's own marker by luck). With a shared stack both threads then read
+    # whichever marker was pushed last -- two identical tags, never {A, B}.
+    both_pushed = threading.Barrier(2)
+    both_read = threading.Barrier(2)
+
+    def worker(tag):
+        marker = _Marker(tag)
+        marker.push_active()
+        both_pushed.wait(timeout=10)
+        try:
+            active = HighState.get_active()
+            results[tag] = None if active is None else active.tag
+            both_read.wait(timeout=10)
+        finally:
+            marker.pop_active()
+
+    threads = [
+        threading.Thread(target=worker, args=("A",)),
+        threading.Thread(target=worker, args=("B",)),
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=15)
+
+    # Each thread sees only the HighState it pushed, despite the concurrent
+    # push from the other thread. On the old shared stack both threads would
+    # read whichever marker was pushed last.
+    assert results == {"A": "A", "B": "B"}
+
+
+def test_sshhighstate_shares_active_stack_with_highstate():
+    """
+    salt-ssh's ``SSHHighState`` subclasses ``BaseHighState`` (not ``HighState``)
+    and its state wrapper calls ``st_.push_active()`` on every run. The
+    active-stack accessors therefore have to live on ``BaseHighState``: when
+    they lived on ``HighState``, ``SSHHighState`` had no ``push_active`` and
+    every salt-ssh state execution raised
+    ``'SSHHighState' object has no attribute 'push_active'``.
+
+    Also pins the cross-subclass contract the pydsl renderer depends on: a
+    non-``HighState`` ``BaseHighState`` subclass that pushes itself is visible
+    to ``HighState.get_active()``.
+    """
+    from salt.client.ssh.state import SSHHighState
+
+    assert issubclass(SSHHighState, salt.state.BaseHighState)
+    assert not issubclass(SSHHighState, salt.state.HighState)
+    for name in ("push_active", "pop_active", "get_active", "clear_active"):
+        assert hasattr(SSHHighState, name), f"SSHHighState lost {name}"
+    # Inherited from the shared base, not redefined per subclass.
+    assert SSHHighState.push_active is salt.state.BaseHighState.push_active
+
+    class _SSHMarker(SSHHighState):
+        # Skip SSHHighState's heavy __init__; only the inherited accessors are
+        # under test.
+        def __init__(self):  # pylint: disable=super-init-not-called
+            pass
+
+    salt.state.BaseHighState.clear_active()
+    try:
+        marker = _SSHMarker()
+        marker.push_active()
+        assert salt.state.HighState.get_active() is marker
+        marker.pop_active()
+        assert salt.state.HighState.get_active() is None
+    finally:
+        salt.state.BaseHighState.clear_active()
+
+
+def test_active_stack_falls_back_when_contextvars_unavailable():
+    """
+    On a salt-ssh target without a usable ``contextvars`` (e.g. Python 3.6,
+    where the module is only available through the thin's backport and can pull
+    in an incompatible ``typing_extensions``), ``import contextvars`` is guarded
+    and ``_active_highstates`` is ``None``. The active-stack accessors must then
+    degrade to a shared class-level list instead of raising -- salt-ssh runs a
+    single execution per target, so a shared stack is safe there.
+    """
+    HighState = salt.state.HighState
+    with patch.object(salt.state, "_active_highstates", None):
+        salt.state.BaseHighState._shared_active_stack.clear()
+        HighState.clear_active()
+        assert HighState.get_active() is None
+
+        a = _Marker("a")
+        b = _Marker("b")
+        a.push_active()
+        assert HighState.get_active() is a
+        b.push_active()
+        assert HighState.get_active() is b
+        b.pop_active()
+        assert HighState.get_active() is a
+        a.pop_active()
+        assert HighState.get_active() is None
+
+        a.push_active()
+        HighState.clear_active()
+        assert HighState.get_active() is None
+    salt.state.BaseHighState._shared_active_stack.clear()