Merge resource state results into managing minion and add tests

For state functions (state.apply, state.highstate, state.sls, etc.) run
against a wildcard glob, resource jobs are now executed inline by the
managing minion and their results folded into a single response block,
producing one Summary section instead of one per resource.
Key changes:
- salt/utils/minions.py: add _MERGE_RESOURCE_FUNS; check_minions accepts
  fun= and skips resource augmentation for merge-mode functions so the
  master does not wait for separate per-resource responses
- salt/master.py: pass fun= to check_minions at publish time
- salt/minion.py: add _MERGE_RESOURCE_FUNS and _prefix_resource_state_key
  (refactored to @staticmethod); _handle_payload skips separate resource
  dispatch for merge-mode functions; _thread_return executes resources
  inline and merges prefixed state keys into the managing minion's ret,
  reporting Result: False for unsupported functions and missing loaders
- salt/modules/sshresource_state.py: highstate() returns a no_|-states
  state dict (Result: False) when the top file has no match; _exec_state_pkg
  recovers valid state dicts from SSHCommandExecutionError instead of
  re-raising and losing the result
Tests:
- test_minions_resources.py: merge-fun skips augmentation, non-merge and
  no-fun still augment, compound targets unaffected
- test_minion_resources.py: _MERGE_RESOURCE_FUNS membership and sync
  invariant, _prefix_resource_state_key correctness and fallback, merge
  block no-loader/unsupported/dict/string branches
- test_sshresource_state.py (new): empty-chunks no-top-file return,
  SSHCommandExecutionError recovery and re-raise cases, normal envelope path

Author: dwoz

salt/master.py

@@ -1654,20 +1654,21 @@ def _mine_flush(self, load):
 
     def _register_resources(self, load):
         """
-        Store the resource list for a minion in the ``minion_resources`` cache
-        bank.  Called by the minion on startup via ``cmd: "_register_resources"``
+        Update the flat resource index for a minion and persist it to the
+        cache.  Called by the minion on startup via ``cmd: "_register_resources"``
         so that the master knows which resource IDs each minion manages.
 
-        This allows :meth:`salt.utils.minions.CkMinions._augment_with_resources`
-        to include resource IDs when expanding glob / non-compound targets, making
-        ``salt '*' test.ping`` return results for both the minion and its resources.
+        Uses :func:`salt.utils.minions._update_resource_index` which atomically
+        updates both the in-process index (so this worker sees the change
+        immediately) and the single flat ``resource_index`` cache file (so
+        other workers pick it up within their TTL window).
         """
         load = self.__verify_load(load, ("id", "resources"))
         if load is False:
             return {}
         if self.opts.get("minion_data_cache", True):
-            self.masterapi.cache.store(
-                "minion_resources", load["id"], load["resources"]
+            salt.utils.minions._update_resource_index(
+                self.masterapi.cache, load["id"], load["resources"]
             )
             log.debug(
                 "Registered resources for minion '%s': %s",
@@ -2366,7 +2367,10 @@ async def publish(self, clear_load):
         delimiter = extra.get("delimiter", DEFAULT_TARGET_DELIM)
 
         _res = self.ckminions.check_minions(
-            clear_load["tgt"], clear_load.get("tgt_type", "glob"), delimiter
+            clear_load["tgt"],
+            clear_load.get("tgt_type", "glob"),
+            delimiter,
+            fun=clear_load.get("fun"),
         )
         minions = _res.get("minions", list())
         missing = _res.get("missing", list())

salt/minion.py

@@ -2868,6 +2868,129 @@ def _thread_return(cls, minion_instance, opts, data):
                 ret["retcode"] = salt.defaults.exitcodes.EX_GENERIC
                 ret["out"] = "nested"
 
+            # -------------------------------------------------------------------
+            # Merge-mode: for state functions the managing minion runs each
+            # resource's function inline here and folds the results into its own
+            # state dict.  The operator then sees ONE combined block + ONE Summary
+            # section instead of a separate block per resource.
+            # -------------------------------------------------------------------
+            if (
+                not data.get("resource_target")
+                and data.get("fun") in cls._MERGE_RESOURCE_FUNS
+                and data.get("resource_targets")
+                and isinstance(ret.get("return"), dict)
+            ):
+                import salt.loader.context as _loader_ctx  # noqa: PLC0415
+
+                _prefix_state_key = minion_instance._prefix_resource_state_key
+
+                run_num_base = (
+                    max(
+                        (
+                            v.get("__run_num__", 0)
+                            for v in ret["return"].values()
+                            if isinstance(v, dict)
+                        ),
+                        default=0,
+                    )
+                    + 1
+                )
+
+                for resource in data["resource_targets"]:
+                    rid = resource["id"]
+                    rtype = resource["type"]
+                    resource_loader = getattr(
+                        minion_instance, "resource_loaders", {}
+                    ).get(rtype)
+                    if resource_loader is None:
+                        ret["return"][f"no_|-{rid}_|-{rid}_|-None"] = {
+                            "result": False,
+                            "comment": (
+                                f"No resource loader for type '{rtype}'. "
+                                "Ensure the resource module exists."
+                            ),
+                            "name": rid,
+                            "changes": {},
+                            "__run_num__": run_num_base,
+                        }
+                        run_num_base += 1
+                        if ret.get("retcode") == salt.defaults.exitcodes.EX_OK:
+                            ret["retcode"] = salt.defaults.exitcodes.EX_GENERIC
+                        continue
+
+                    if function_name not in resource_loader:
+                        # Function not implemented for this resource type —
+                        # same message the separate-job path would return.
+                        resource_return = (
+                            f"Function '{function_name}' is not supported for "
+                            f"resource type '{rtype}'. Implement it in a "
+                            f"'{rtype}resource_*' execution module."
+                        )
+                    else:
+                        token = _loader_ctx.resource_ctxvar.set(resource)
+                        try:
+                            resource_return = minion_instance._execute_job_function(
+                                function_name,
+                                function_args,
+                                executors,
+                                opts,
+                                data,
+                                functions=resource_loader,
+                            )
+                        except Exception as exc:  # pylint: disable=broad-except
+                            log.error(
+                                "Inline resource execution for '%s' raised: %s",
+                                rid,
+                                exc,
+                                exc_info=True,
+                            )
+                            resource_return = (
+                                f"ERROR running {function_name} for '{rid}': {exc}"
+                            )
+                        finally:
+                            _loader_ctx.resource_ctxvar.reset(token)
+
+                    if isinstance(resource_return, dict):
+                        for state_id, state_val in resource_return.items():
+                            if isinstance(state_val, dict):
+                                entry = dict(state_val)
+                                entry["__run_num__"] = run_num_base
+                            else:
+                                entry = {
+                                    "result": True,
+                                    "comment": str(state_val),
+                                    "name": f"[{rid}]",
+                                    "changes": {},
+                                    "__run_num__": run_num_base,
+                                }
+                            run_num_base += 1
+                            ret["return"][_prefix_state_key(state_id, rid)] = entry
+                        r_retcode = resource_loader.pack["__context__"].get(
+                            "retcode", 0
+                        )
+                        if (
+                            r_retcode
+                            and ret.get("retcode") == salt.defaults.exitcodes.EX_OK
+                        ):
+                            ret["retcode"] = r_retcode
+                    else:
+                        # String result means the resource couldn't fulfill the
+                        # operation (e.g. "not supported" for dummy resources).
+                        # Mark as False — the state was NOT applied, so reporting
+                        # True would silently mask unactioned resources.
+                        ret["return"][f"no_|-{rid}_|-{rid}_|-None"] = {
+                            "result": False,
+                            "comment": str(resource_return),
+                            "name": rid,
+                            "changes": {},
+                            "__run_num__": run_num_base,
+                        }
+                        run_num_base += 1
+                        if ret.get("retcode") == salt.defaults.exitcodes.EX_OK:
+                            ret["retcode"] = salt.defaults.exitcodes.EX_GENERIC
+
+                ret["success"] = ret.get("retcode") == salt.defaults.exitcodes.EX_OK
+
             if isinstance(ret["return"], dict) and ret["return"].get("__no_return__"):
                 # This is used to suppress the return for queued jobs
                 # The job will be executed later and will return then
@@ -4379,14 +4502,20 @@ async def _handle_payload(self, payload):
                 if load.get("minion_is_target", True):
                     await self._handle_decoded_payload(load)
 
-                for resource in load.get("resource_targets", []):
-                    resource_load = dict(load)
-                    resource_load["resource_target"] = resource
-                    # Flag so _handle_decoded_payload_impl can bypass JID
-                    # deduplication — resource jobs share the parent JID by
-                    # design but are independent executions.
-                    resource_load["resource_job"] = True
-                    await self._handle_decoded_payload(resource_load)
+                # For merge-mode functions (state.apply etc.) resources are
+                # executed inline inside _thread_return and folded into the
+                # managing minion's own response.  Dispatching them as
+                # separate jobs would send duplicate responses the master is
+                # no longer waiting for.
+                if load.get("fun") not in self._MERGE_RESOURCE_FUNS:
+                    for resource in load.get("resource_targets", []):
+                        resource_load = dict(load)
+                        resource_load["resource_target"] = resource
+                        # Flag so _handle_decoded_payload_impl can bypass JID
+                        # deduplication — resource jobs share the parent JID by
+                        # design but are independent executions.
+                        resource_load["resource_job"] = True
+                        await self._handle_decoded_payload(resource_load)
 
             elif self.opts["zmq_filtering"]:
                 # In the filtering enabled case, we'd like to know when minion sees something it shouldn't
@@ -4473,6 +4602,36 @@ def _is_pure_resource_target(self, load):
         }
     )
 
+    # Functions where resource results are merged into the managing minion's
+    # own response rather than dispatched as independent jobs.  This produces
+    # ONE combined block + Summary section per managing minion instead of
+    # separate blocks per resource, matching how any other minion looks.
+    _MERGE_RESOURCE_FUNS = frozenset(
+        {
+            "state.apply",
+            "state.highstate",
+            "state.sls",
+            "state.sls_id",
+            "state.single",
+        }
+    )
+
+    @staticmethod
+    def _prefix_resource_state_key(sid, rid):
+        """Re-label the ID/name components of a state result key with rid.
+
+        Key format: {module}_|-{id}_|-{name}_|-{function}
+        Only comps[1] and comps[2] (id and name) are prefixed so the
+        highstate formatter still reads {comps[0]}.{comps[3]} correctly,
+        preserving ``Function: pkg.installed`` while showing ``ID: node1 curl``.
+        """
+        parts = sid.split("_|-", 3)
+        if len(parts) == 4:
+            parts[1] = f"{rid} {parts[1]}"
+            parts[2] = f"{rid} {parts[2]}"
+            return "_|-".join(parts)
+        return f"no_|-{rid}_|-{rid}_|-None"
+
     def _resolve_resource_targets(self, load):
         """
         Return the list of per-resource dicts ``{"id": ..., "type": ...}`` that

salt/modules/sshresource_state.py

@@ -281,6 +281,24 @@ def highstate(test=None, **kwargs):
             if not isinstance(chunk, dict):
                 return chunks_or_errors
 
+        if not chunks_or_errors:
+            # Top file has no match for this resource ID — no SSH round-trip needed.
+            # Return a state dict using the same key format salt uses for a regular
+            # minion's "No Top file" entry so the merged output is consistent.
+            rid = _resource_id()
+            return {
+                "no_|-states_|-states_|-None": {
+                    "result": False,
+                    "comment": (
+                        f"No Top file or master_tops data matches found for"
+                        f" resource '{rid}'."
+                    ),
+                    "name": "states",
+                    "changes": {},
+                    "__run_num__": 0,
+                }
+            }
+
         file_refs = salt.client.ssh.state.lowstate_file_refs(
             chunks_or_errors,
             _merge_extra_filerefs(
@@ -436,16 +454,28 @@ def _exec_state_pkg(opts, trans_tar, test):
         except OSError:
             pass
 
-    # parse_ret returns data["local"] = {"jid": ..., "return": {states}, "retcode": N}
-    # The minion dispatcher expects the state dict directly (not the envelope).
-    envelope = salt.client.ssh.wrapper.parse_ret(stdout, stderr, retcode)
+    # parse_ret raises SSHCommandExecutionError on any non-zero retcode, even
+    # when the remote ran states and produced a valid result dict (e.g. some
+    # states failed → retcode 2).  Catch that case and surface the result dict
+    # normally so operators see the full state tree rather than raw JSON.
+    try:
+        envelope = salt.client.ssh.wrapper.parse_ret(stdout, stderr, retcode)
+    except salt.client.ssh.wrapper.SSHCommandExecutionError as exc:
+        local = (exc.parsed or {}).get("local", {})
+        if isinstance(local.get("return"), dict):
+            ret = local["return"]
+            __context__["retcode"] = local.get(  # pylint: disable=undefined-variable
+                "retcode", salt.defaults.exitcodes.EX_STATE_FAILURE
+            )
+            return ret
+        raise
+
     if isinstance(envelope, dict) and "return" in envelope:
         ret = envelope["return"]
-        # Propagate non-zero retcode into context so caller can signal failure.
         remote_retcode = envelope.get("retcode", 0)
         if remote_retcode:
-            __context__["retcode"] = (
-                remote_retcode  # pylint: disable=undefined-variable
+            __context__["retcode"] = (  # pylint: disable=undefined-variable
+                remote_retcode
             )
         return ret
     return envelope