Keep a sub-proxy's returner config out of the shared load (#70219)

changelog/70219.fixed.md

@@ -0,0 +1 @@
+Stopped a deltaproxy sub-proxy's ``return`` configuration leaking onto its siblings. ``handle_payload`` hands the same publish-load dict to every sub-proxy a job matched, and ``thread_return`` merged ``opts["return"]`` back into ``data["ret"]``, so with ``multiprocessing: False`` a sub-proxy with no returner configured sent its job return to another sub-proxy's returner.

salt/metaproxy/deltaproxy.py

@@ -834,20 +834,29 @@ def thread_return(cls, minion_instance, opts, data):
 
     # Add default returners from minion config
     # Should have been coverted to comma-delimited string already
+    #
+    # ``data`` is the publish load, and ``handle_payload`` hands the *same*
+    # dict to the control proxy and to every sub-proxy the job matched.  With
+    # ``multiprocessing: False`` those all run as threads in one process, so
+    # writing the merged returner list back into ``data["ret"]`` leaks this
+    # sub-proxy's ``return`` config onto its siblings -- a sub-proxy with no
+    # returner configured would send its job return to another one's returner.
+    # Keep the merge local to this job.
+    job_returners = data["ret"]
     if isinstance(opts.get("return"), str):
-        if data["ret"]:
-            data["ret"] = ",".join((data["ret"], opts["return"]))
+        if job_returners:
+            job_returners = ",".join((job_returners, opts["return"]))
         else:
-            data["ret"] = opts["return"]
+            job_returners = opts["return"]
 
     # TODO: make a list? Seems odd to split it this late :/
-    if data["ret"] and isinstance(data["ret"], str):
+    if job_returners and isinstance(job_returners, str):
         if "ret_config" in data:
             ret["ret_config"] = data["ret_config"]
         if "ret_kwargs" in data:
             ret["ret_kwargs"] = data["ret_kwargs"]
         ret["id"] = opts["id"]
-        for returner in set(data["ret"].split(",")):
+        for returner in set(job_returners.split(",")):
             try:
                 returner_str = f"{returner}.returner"
                 if returner_str in minion_instance.returners:

tests/pytests/unit/metaproxy/test_deltaproxy.py

@@ -229,3 +229,53 @@ def test_subproxy_post_master_init_packs_per_minion_grains(
     # control proxy stores the right grains in ``self.deltaproxy_opts``.
     assert result1["proxy_opts"]["grains"]["serial_number"] == "SN-AAA-001"
     assert result2["proxy_opts"]["grains"]["serial_number"] == "SN-BBB-002"
+
+
+def test_thread_return_does_not_write_returners_back_into_the_shared_load(tmp_path):
+    """
+    ``handle_payload`` hands the *same* publish-load dict to the control proxy
+    and to every sub-proxy the job matched.  ``thread_return`` used to merge
+    ``opts["return"]`` back into ``data["ret"]``, so with
+    ``multiprocessing: False`` -- where those all run as threads in one process
+    -- one sub-proxy's returner configuration leaked onto its siblings, and a
+    sub-proxy with no returner configured sent its job return to another
+    sub-proxy's returner.
+    """
+    proc_dir = tmp_path / "proc"
+    proc_dir.mkdir()
+
+    minion_instance = MagicMock()
+    minion_instance.proc_dir = str(proc_dir)
+    minion_instance.connected = False
+
+    # The one shared dict, exactly as the fan-out passes it around.
+    shared_load = {
+        "jid": "20260101000000000002",
+        "fun": "test.ping",
+        "arg": [],
+        "ret": "",
+    }
+
+    class ProxyMinion:
+        """Stand-in for the class deltaproxy names in the process title."""
+
+    configured = {
+        "multiprocessing": False,
+        "id": "minion1",
+        "cachedir": str(tmp_path),
+        "return": "some_returner",
+    }
+    deltaproxy.thread_return(ProxyMinion, minion_instance, configured, shared_load)
+
+    # The sub-proxy that owns the setting must not stamp it on the shared load
+    # the next sub-proxy is about to read.
+    assert shared_load["ret"] == ""
+
+    # And the sibling with no returner of its own still sees nothing.
+    bare = {
+        "multiprocessing": False,
+        "id": "minion2",
+        "cachedir": str(tmp_path),
+    }
+    deltaproxy.thread_return(ProxyMinion, minion_instance, bare, shared_load)
+    assert shared_load["ret"] == ""