From 37e192be14d17f020a5264dc44dd9211baff567c Mon Sep 17 00:00:00 2001 From: "Gary T. Giesen" Date: Fri, 10 Jul 2026 09:25:32 -0400 Subject: [PATCH] Fix threaded-minion wrong retcode by owning the loader per job With multiprocessing:False every job runs against one shared minion_instance. _thread_return calls gen_modules() at the top of every job, which rebuilds all loaders and rebinds minion_instance.functions to a brand-new LazyLoader whose __context__ is a fresh, empty dict. A module writes its retcode into the __context__ of the loader generation it ran under (e.g. cmdmod via NamedLoaderContext), but _thread_return read the retcode back from minion_instance.functions at completion. When a sibling job's gen_modules() rebound functions between a job's write and its read, the read landed on the new empty context and returned EX_OK, so a failed command was delivered to the master as a success. Make each job own the loader it wrote its retcode into: - gen_modules() now returns the loader generation it built. - _thread_return / _thread_multi_return capture that return value and use it for the function lookup, the retcode reset, and the retcode read, instead of re-reading the shared minion_instance.functions attribute. - _execute_job_function takes the captured loader (defaulting to self.functions) and uses it for the lookup and the retcode reset. - sys.reload_modules is bound to a new _reload_modules wrapper that reloads and returns None, so the reload job stays serializable on the wire (a LazyLoader is not). - The proxy and delta-proxy thread_return / thread_multi_return copies capture minion_instance.functions once per job and use it consistently, closing the same write/read desync there. Adds a direct-altitude regression test that drives the real Minion._thread_return with the poison sibling reload forced into the write->read window and asserts a failed job is delivered as failed (and a passing job as passing), plus a test that sys.reload_modules still returns a serializable None. Fixes #61830 --- changelog/61830.fixed.md | 1 + salt/metaproxy/deltaproxy.py | 26 ++++--- salt/metaproxy/proxy.py | 24 ++++--- salt/minion.py | 65 ++++++++++------- tests/pytests/unit/test_minion.py | 115 ++++++++++++++++++++++++++++++ 5 files changed, 189 insertions(+), 42 deletions(-) create mode 100644 changelog/61830.fixed.md diff --git a/changelog/61830.fixed.md b/changelog/61830.fixed.md new file mode 100644 index 000000000000..c6faec70d1bf --- /dev/null +++ b/changelog/61830.fixed.md @@ -0,0 +1 @@ +Fixed threaded minion jobs (``multiprocessing: False``) reporting the wrong retcode, including a failed command returning success. Every job rebuilds and rebinds the shared ``minion_instance.functions`` at its top via ``gen_modules()``; a sibling job's rebind landing between a job's retcode write and its read made ``_thread_return`` read a fresh, empty ``__context__`` and deliver ``EX_OK``. ``gen_modules()`` now returns the loader generation it built, and ``_thread_return``/``_thread_multi_return`` capture it and read the retcode from the loader the job owns instead of the shared attribute. The ``sys.reload_modules`` binding was pointed at a new ``_reload_modules`` wrapper so it keeps returning ``None`` on the wire. The proxy and delta-proxy job runners got the same loader-capture treatment. diff --git a/salt/metaproxy/deltaproxy.py b/salt/metaproxy/deltaproxy.py index 7cae077faf67..27e96ed95cb7 100644 --- a/salt/metaproxy/deltaproxy.py +++ b/salt/metaproxy/deltaproxy.py @@ -655,7 +655,11 @@ def thread_return(cls, minion_instance, opts, data): if f"{executor}.allow_missing_func" in minion_instance.executors ] ) - if function_name in minion_instance.functions or allow_missing_funcs is True: + # Own the loader for the length of this job so the func lookup, the retcode + # reset and the retcode read all resolve against the same generation, even if + # a concurrent sys.reload_modules rebinds minion_instance.functions. See #61830. + functions = minion_instance.functions + if function_name in functions or allow_missing_funcs is True: try: minion_blackout_violation = False if minion_instance.connected and minion_instance.opts["pillar"].get( @@ -687,15 +691,15 @@ def thread_return(cls, minion_instance, opts, data): "saltutil.refresh_pillar allowed in blackout mode." ) - if function_name in minion_instance.functions: - func = minion_instance.functions[function_name] + if function_name in functions: + func = functions[function_name] args, kwargs = salt.minion.load_args_and_kwargs(func, data["arg"], data) else: # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True func = function_name args, kwargs = data["arg"], data - minion_instance.functions.pack["__context__"]["retcode"] = 0 - minion_instance.functions.pack["__opts__"] = opts + functions.pack["__context__"]["retcode"] = 0 + functions.pack["__opts__"] = opts if isinstance(executors, str): executors = [executors] elif not isinstance(executors, list) or not executors: @@ -737,7 +741,7 @@ def thread_return(cls, minion_instance, opts, data): else: ret["return"] = return_data - retcode = minion_instance.functions.pack["__context__"].get( + retcode = functions.pack["__context__"].get( "retcode", salt.defaults.exitcodes.EX_OK ) if retcode == salt.defaults.exitcodes.EX_OK: @@ -897,6 +901,10 @@ def thread_multi_return(cls, minion_instance, opts, data): else: ret = {"return": {}, "retcode": {}, "success": {}} + # Own the loader for the length of this job so every func's lookup, retcode + # reset and retcode read resolve against the same generation, even if a + # concurrent sys.reload_modules rebinds minion_instance.functions. See #61830. + functions = minion_instance.functions for ind in range(0, num_funcs): if not multifunc_ordered: ret["success"][data["fun"][ind]] = False @@ -930,15 +938,15 @@ def thread_multi_return(cls, minion_instance, opts, data): "saltutil.refresh_pillar allowed in blackout mode." ) - func = minion_instance.functions[data["fun"][ind]] + func = functions[data["fun"][ind]] args, kwargs = salt.minion.load_args_and_kwargs( func, data["arg"][ind], data ) - minion_instance.functions.pack["__context__"]["retcode"] = 0 + functions.pack["__context__"]["retcode"] = 0 key = ind if multifunc_ordered else data["fun"][ind] ret["return"][key] = func(*args, **kwargs) - retcode = minion_instance.functions.pack["__context__"].get("retcode", 0) + retcode = functions.pack["__context__"].get("retcode", 0) if retcode == 0: # No nonzero retcode in __context__ dunder. Check if return # is a dictionary with a "result" or "success" key. diff --git a/salt/metaproxy/proxy.py b/salt/metaproxy/proxy.py index d9a2848054ae..8f7d606b78d1 100644 --- a/salt/metaproxy/proxy.py +++ b/salt/metaproxy/proxy.py @@ -418,7 +418,11 @@ def thread_return(cls, minion_instance, opts, data): if f"{executor}.allow_missing_func" in minion_instance.executors ] ) - if function_name in minion_instance.functions or allow_missing_funcs is True: + # Own the loader for the length of this job so the func lookup, the retcode + # reset and the retcode read all resolve against the same generation, even if + # a concurrent sys.reload_modules rebinds minion_instance.functions. See #61830. + functions = minion_instance.functions + if function_name in functions or allow_missing_funcs is True: try: minion_blackout_violation = False if minion_instance.connected and minion_instance.opts["pillar"].get( @@ -450,14 +454,14 @@ def thread_return(cls, minion_instance, opts, data): "saltutil.refresh_pillar allowed in blackout mode." ) - if function_name in minion_instance.functions: - func = minion_instance.functions[function_name] + if function_name in functions: + func = functions[function_name] args, kwargs = salt.minion.load_args_and_kwargs(func, data["arg"], data) else: # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True func = function_name args, kwargs = data["arg"], data - minion_instance.functions.pack["__context__"]["retcode"] = 0 + functions.pack["__context__"]["retcode"] = 0 if isinstance(executors, str): executors = [executors] elif not isinstance(executors, list) or not executors: @@ -497,7 +501,7 @@ def thread_return(cls, minion_instance, opts, data): else: ret["return"] = return_data - retcode = minion_instance.functions.pack["__context__"].get( + retcode = functions.pack["__context__"].get( "retcode", salt.defaults.exitcodes.EX_OK ) if retcode == salt.defaults.exitcodes.EX_OK: @@ -651,6 +655,10 @@ def thread_multi_return(cls, minion_instance, opts, data): else: ret = {"return": {}, "retcode": {}, "success": {}} + # Own the loader for the length of this job so every func's lookup, retcode + # reset and retcode read resolve against the same generation, even if a + # concurrent sys.reload_modules rebinds minion_instance.functions. See #61830. + functions = minion_instance.functions for ind in range(0, num_funcs): if not multifunc_ordered: ret["success"][data["fun"][ind]] = False @@ -684,15 +692,15 @@ def thread_multi_return(cls, minion_instance, opts, data): "saltutil.refresh_pillar allowed in blackout mode." ) - func = minion_instance.functions[data["fun"][ind]] + func = functions[data["fun"][ind]] args, kwargs = salt.minion.load_args_and_kwargs( func, data["arg"][ind], data ) - minion_instance.functions.pack["__context__"]["retcode"] = 0 + functions.pack["__context__"]["retcode"] = 0 key = ind if multifunc_ordered else data["fun"][ind] ret["return"][key] = func(*args, **kwargs) - retcode = minion_instance.functions.pack["__context__"].get("retcode", 0) + retcode = functions.pack["__context__"].get("retcode", 0) if retcode == 0: # No nonzero retcode in __context__ dunder. Check if return # is a dictionary with a "result" or "success" key. diff --git a/salt/minion.py b/salt/minion.py index 229ec76956d4..10ca161838c7 100644 --- a/salt/minion.py +++ b/salt/minion.py @@ -512,34 +512,47 @@ def gen_modules(self, initial_load=False, context=None): ).compile_pillar() self.utils = salt.loader.utils(self.opts, context=context) - self.functions = salt.loader.minion_mods( + functions = salt.loader.minion_mods( self.opts, utils=self.utils, context=context ) + self.functions = functions self.serializers = salt.loader.serializers(self.opts) self.returners = salt.loader.returners( - self.opts, functions=self.functions, context=context + self.opts, functions=functions, context=context ) self.proxy = salt.loader.proxy( - self.opts, functions=self.functions, returners=self.returners + self.opts, functions=functions, returners=self.returners ) # TODO: remove self.function_errors = {} # Keep the funcs clean self.states = salt.loader.states( self.opts, - functions=self.functions, + functions=functions, utils=self.utils, serializers=self.serializers, context=context, ) - self.rend = salt.loader.render( - self.opts, functions=self.functions, context=context - ) + self.rend = salt.loader.render(self.opts, functions=functions, context=context) # self.matcher = Matcher(self.opts, self.functions) self.matchers = salt.loader.matchers(self.opts) - self.functions["sys.reload_modules"] = self.gen_modules + functions["sys.reload_modules"] = self._reload_modules self.executors = salt.loader.executors( - self.opts, functions=self.functions, proxy=self.proxy, context=context + self.opts, functions=functions, proxy=self.proxy, context=context ) + # Return the loader generation this call built so that a threaded job + # (multiprocessing=False) can own the exact loader it wrote its retcode + # into, rather than re-reading the shared self.functions attribute which + # a sibling job's gen_modules() may have rebound. See issue #61830. + return functions + + def _reload_modules(self, initial_load=False, context=None): + """ + The ``sys.reload_modules`` execution function. gen_modules() returns the + rebuilt loader (a LazyLoader, which is not serializable), so it cannot be + bound to ``sys.reload_modules`` directly -- the returned value would be + put on the wire. This wrapper reloads the modules and returns None. + """ + self.gen_modules(initial_load=initial_load, context=context) @staticmethod def process_schedule(minion, loop_interval): @@ -2613,12 +2626,19 @@ def run_func(minion_instance, opts, data): run_func(minion_instance, opts, data) def _execute_job_function( - self, function_name, function_args, executors, opts, data + self, function_name, function_args, executors, opts, data, functions=None ): """ Executes a function within a job given it's name, the args and the executors. It also checks if the function is allowed to run if 'blackout mode' is enabled. + + ``functions`` is the loader generation this job owns (returned by + gen_modules()); the func lookup and the retcode reset must use it rather + than the shared self.functions, which a sibling job may have rebound. It + defaults to self.functions for callers that do not thread a loader. """ + if functions is None: + functions = self.functions minion_blackout_violation = False if self.connected and self.opts["pillar"].get("minion_blackout", False): whitelist = self.opts["pillar"].get("minion_blackout_whitelist", []) @@ -2643,14 +2663,14 @@ def _execute_job_function( "saltutil.refresh_pillar allowed in blackout mode." ) - if function_name in self.functions: - func = self.functions[function_name] + if function_name in functions: + func = functions[function_name] args, kwargs = load_args_and_kwargs(func, function_args, data) else: # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True func = function_name args, kwargs = function_args, data - self.functions.pack["__context__"]["retcode"] = 0 + functions.pack["__context__"]["retcode"] = 0 if isinstance(executors, str): executors = [executors] @@ -2680,7 +2700,7 @@ def _thread_return(cls, minion_instance, opts, data): This method should be used as a threading target, start the actual minion side execution. """ - minion_instance.gen_modules() + functions = minion_instance.gen_modules() fn_ = os.path.join(minion_instance.proc_dir, str(data["jid"])) if opts.get("multiprocessing", True): @@ -2709,13 +2729,10 @@ def _thread_return(cls, minion_instance, opts, data): ] ) try: - if ( - function_name in minion_instance.functions - or allow_missing_funcs is True - ): + if function_name in functions or allow_missing_funcs is True: try: return_data = minion_instance._execute_job_function( - function_name, function_args, executors, opts, data + function_name, function_args, executors, opts, data, functions ) log.info( "Job %s execution finished, return_data: %s", @@ -2743,7 +2760,7 @@ def _thread_return(cls, minion_instance, opts, data): else: ret["return"] = return_data - retcode = minion_instance.functions.pack["__context__"].get( + retcode = functions.pack["__context__"].get( "retcode", salt.defaults.exitcodes.EX_OK ) if retcode == salt.defaults.exitcodes.EX_OK: @@ -2924,7 +2941,7 @@ def _thread_multi_return(cls, minion_instance, opts, data): This method should be used as a threading target, start the actual minion side execution. """ - minion_instance.gen_modules() + functions = minion_instance.gen_modules() fn_ = os.path.join(minion_instance.proc_dir, str(data["jid"])) if opts.get("multiprocessing", True): @@ -2977,14 +2994,12 @@ def _thread_multi_return(cls, minion_instance, opts, data): ret["success"][function_name] = False try: return_data = minion_instance._execute_job_function( - function_name, function_args, executors, opts, data + function_name, function_args, executors, opts, data, functions ) key = ind if multifunc_ordered else data["fun"][ind] ret["return"][key] = return_data - retcode = minion_instance.functions.pack["__context__"].get( - "retcode", 0 - ) + retcode = functions.pack["__context__"].get("retcode", 0) if retcode == 0: # No nonzero retcode in __context__ dunder. Check if return # is a dictionary with a "result" or "success" key. diff --git a/tests/pytests/unit/test_minion.py b/tests/pytests/unit/test_minion.py index 70831a9d5f40..c316297d1cf3 100644 --- a/tests/pytests/unit/test_minion.py +++ b/tests/pytests/unit/test_minion.py @@ -9,11 +9,13 @@ import pytest +import salt.defaults.exitcodes import salt.ext.tornado import salt.ext.tornado.gen import salt.ext.tornado.testing import salt.minion import salt.modules.test as test_mod +import salt.payload import salt.syspaths import salt.utils.crypt import salt.utils.event as event @@ -1093,6 +1095,119 @@ def compile_pillar(self): minion.destroy() +def test_thread_return_retcode_owns_captured_loader_61830(minion_opts): + """ + #61830: with multiprocessing=False every threaded job shares one + minion_instance and rebuilds+rebinds minion_instance.functions at its top + (gen_modules). A sibling job's gen_modules() that rebinds functions between a + job's retcode write and its retcode read made the pre-fix _thread_return read + a fresh, empty __context__ and deliver EX_OK -- a failed command reported + SUCCESS. The fix makes _thread_return capture the loader gen_modules() + returns and read the retcode from that loader. + + This drives the REAL Minion._thread_return end to end. test.retcode writes + the retcode into its loader's __context__ exactly like a production module, + and the poison sibling reload is forced into the exact write->read window + (the only thing _thread_return does there is log.info("... execution + finished")). + """ + minion_opts["multiprocessing"] = False + minion_opts["file_client"] = "local" + minion_opts["grains"] = {} + minion_opts["pillar"] = {} + io_loop = salt.ext.tornado.ioloop.IOLoop() + io_loop.make_current() + minion = salt.minion.Minion(minion_opts, io_loop=io_loop) + try: + minion.gen_modules() + minion.connected = True + proc_dir = os.path.join(minion_opts["cachedir"], "proc") + os.makedirs(proc_dir, exist_ok=True) + minion.proc_dir = proc_dir + + delivered = [] + minion._return_pub = lambda ret, *args, **kwargs: delivered.append(ret) + + original_info = salt.minion.log.info + state = {"fired": False} + + def poison_info(msg, *args, **kwargs): + # Fire the sibling gen_modules() rebind once, in the write->read + # window, so the shared minion.functions is a DIFFERENT empty loader + # by the time _thread_return reads the retcode back. + if not state["fired"] and "execution finished" in str(msg): + state["fired"] = True + minion.gen_modules() + return original_info(msg, *args, **kwargs) + + def run(code): + delivered.clear() + state["fired"] = False + data = { + "jid": f"20260101000000{code:06d}", + "fun": "test.retcode", + "arg": [code], + "ret": "", + } + with patch.object(salt.minion.log, "info", poison_info): + salt.minion.Minion._thread_return(minion, minion.opts, data) + assert state[ + "fired" + ], "sibling reload never fired in the write->read window" + return delivered[-1] + + # A FAILED job (retcode 42) under the forced interleave must be delivered + # as failed. Pre-fix this delivered retcode=0/success=True (the bug). + ret = run(42) + assert ret["retcode"] == 42 + assert ret["success"] is False + # The retcode came from the loader the job owns, NOT the shared attribute: + # the sibling reload left minion.functions with an empty __context__. + assert ( + minion.functions.pack["__context__"].get( + "retcode", salt.defaults.exitcodes.EX_OK + ) + == salt.defaults.exitcodes.EX_OK + ) + + # Inverse must-not: a SUCCEEDING job (retcode 0) under the same forced + # interleave must still be delivered as success -- the fix must not flip a + # genuinely-passing job to failed. + ret = run(0) + assert ret["retcode"] == salt.defaults.exitcodes.EX_OK + assert ret["success"] is True + finally: + minion.destroy() + + +def test_sys_reload_modules_returns_none_and_is_serializable_61830(minion_opts): + """ + #61830 follow-on: gen_modules() now returns the rebuilt loader so a threaded + job can own it. A LazyLoader is not serializable, so sys.reload_modules must + not be bound to gen_modules directly (its return would be put on the wire). + It is bound to _reload_modules, which reloads the modules and returns None. + """ + minion_opts["file_client"] = "local" + io_loop = salt.ext.tornado.ioloop.IOLoop() + io_loop.make_current() + minion = salt.minion.Minion(minion_opts, io_loop=io_loop) + try: + minion.gen_modules() + reload_fn = minion.functions["sys.reload_modules"] + result = reload_fn() + assert result is None + # The wire payload must round-trip without raising. + salt.payload.dumps({"return": result}) + # gen_modules() itself returns the loader (not serializable) -- which is + # exactly why sys.reload_modules needs the None-returning wrapper. + loader = minion.gen_modules() + assert loader is minion.functions + with pytest.raises(TypeError): + salt.payload.dumps({"return": loader}) + finally: + minion.destroy() + + def test_minion_manage_schedule(minion_opts): """ Tests that the manage_schedule will call the add function, adding