Aggregate the requisites first and then aggregate the states.

salt/state.py

@@ -858,13 +858,16 @@ def _aggregate_requisites(self, low, chunks):
         Aggregate the requisites
         """
         requisites = {}
-        low_state = low["state"]
         for chunk in chunks:
             # if the state function in the chunk matches
             # the state function in the low we're looking at
             # and __agg__ is True, add the requisites from the
             # chunk to those in the low.
-            if chunk["state"] == low["state"] and chunk.get("__agg__"):
+            if (
+                chunk["state"] == low["state"]
+                and chunk.get("__agg__")
+                and low["name"] != chunk["name"]
+            ):
                 for req in frozenset.union(
                     *[STATE_REQUISITE_KEYWORDS, STATE_REQUISITE_IN_KEYWORDS]
                 ):
@@ -889,13 +892,23 @@ def _mod_aggregate(self, low, running, chunks):
             return low
         if low["state"] in agg_opt and not low.get("__agg__"):
             agg_fun = "{}.mod_aggregate".format(low["state"])
+            if "loader_cache" not in self.state_con:
+                self.state_con["loader_cache"] = {}
+            if (
+                agg_fun in self.state_con["loader_cache"]
+                and not self.state_con["loader_cache"][agg_fun]
+            ):
+                return low
             if agg_fun in self.states:
+                self.state_con["loader_cache"][agg_fun] = True
                 try:
-                    low = self.states[agg_fun](low, chunks, running)
-                    low = self._aggregate_requisites(low, chunks)
                     low["__agg__"] = True
+                    low = self._aggregate_requisites(low, chunks)
+                    low = self.states[agg_fun](low, chunks, running)
                 except TypeError:
                     log.error("Failed to execute aggregate for state %s", low["state"])
+            else:
+                self.state_con["loader_cache"][agg_fun] = False
         return low
 
     def _run_check(self, low_data):
@@ -2772,11 +2785,11 @@ def check_requisite(self, low, running, chunks, pre=False):
                         req = {"id": req}
                     req = trim_req(req)
                     found = False
+                    req_key = next(iter(req))
+                    req_val = req[req_key]
+                    if req_val is None:
+                        continue
                     for chunk in chunks:
-                        req_key = next(iter(req))
-                        req_val = req[req_key]
-                        if req_val is None:
-                            continue
                         if req_key == "sls":
                             # Allow requisite tracking of entire sls files
                             if fnmatch.fnmatch(chunk["__sls__"], req_val):

tests/pytests/unit/state/test_state_compiler.py

@@ -934,3 +934,96 @@ def test_aggregate_requisites():
             "/tmp/install-tmux",
             "google-cloud-repo",
         ]
+
+
+def test_mod_aggregate():
+    """
+    Test to ensure that the requisites are included in the aggregated low state.
+    """
+    # The low that is returned from _mod_aggregrate
+    low = {
+        "state": "pkg",
+        "name": "sl",
+        "__sls__": "test.62439",
+        "__env__": "base",
+        "__id__": "sl",
+        "require_in": [OrderedDict([("file", "/tmp/foo")])],
+        "order": 10002,
+        "aggregate": True,
+        "fun": "installed",
+    }
+
+    # Chunks that have been processed through the pkg mod_aggregate function
+    chunks = [
+        {
+            "state": "file",
+            "name": "/tmp/foo",
+            "__sls__": "test.62439",
+            "__env__": "base",
+            "__id__": "/tmp/foo",
+            "content": "This is some content",
+            "order": 10000,
+            "require": [{"pkg": "sl"}],
+            "fun": "managed",
+        },
+        {
+            "state": "pkg",
+            "name": "figlet",
+            "__sls__": "test.62439",
+            "__env__": "base",
+            "__id__": "figlet",
+            "__agg__": True,
+            "require": [OrderedDict([("file", "/tmp/foo")])],
+            "order": 10001,
+            "aggregate": True,
+            "fun": "installed",
+        },
+        {
+            "state": "pkg",
+            "name": "sl",
+            "__sls__": "test.62439",
+            "__env__": "base",
+            "__id__": "sl",
+            "require_in": [OrderedDict([("file", "/tmp/foo")])],
+            "order": 10002,
+            "aggregate": True,
+            "fun": "installed",
+        },
+    ]
+
+    running = {}
+
+    mock_pkg_mod_aggregate = {
+        "state": "pkg",
+        "name": "sl",
+        "__sls__": "test.62439",
+        "__env__": "base",
+        "__id__": "sl",
+        "require_in": [OrderedDict([("file", "/tmp/foo")])],
+        "order": 10002,
+        "fun": "installed",
+        "__agg__": True,
+        "pkgs": ["figlet", "sl"],
+    }
+
+    with patch("salt.state.State._gather_pillar") as state_patch:
+        minion_opts = salt.config.DEFAULT_MINION_OPTS.copy()
+        state_obj = salt.state.State(minion_opts)
+        with patch.dict(
+            state_obj.states,
+            {"pkg.mod_aggregate": MagicMock(return_value=mock_pkg_mod_aggregate)},
+        ):
+            low_ret = state_obj._mod_aggregate(low, running, chunks)
+
+            # Ensure the low returned contains require
+            assert "require_in" in low_ret
+
+            # Ensure all the requires from pkg states are in low
+            assert low_ret["require_in"] == [OrderedDict([("file", "/tmp/foo")])]
+
+            # Ensure that the require requisite from the
+            # figlet state doesn't find its way into this state
+            assert "require" not in low_ret
+
+            # Ensure pkgs were aggregated
+            assert low_ret["pkgs"] == ["figlet", "sl"]