Unify error handling in pgjsonb returner and event_return

Two asymmetries existed in the pgjsonb writer functions:

`returner(ret)` only caught `salt.exceptions.SaltMasterError`, which
`_get_serv` raises only on connection failure (`OperationalError`).
A `psycopg2.DatabaseError` raised during the actual `cur.execute(...)`
into `salt_returns` (FK violation, NOT NULL violation, deadlock,
DataError) propagated to the caller. The publish path in `_prep_pub`
catches it via `except Exception -> log.critical`, but the
syndic-aggregate path at master.py:1699-1701 does not, so a single
bad row could escape into a master subprocess.

`event_return(events)` had no error handling at all. The event-bus
caller in `EventReturn._flush_event_single` already wraps it in
`except Exception -> log.error`, so the master does not crash, but
the entire `event_queue` is dropped on any failure with no local
context about how many events were lost.

Add a `psycopg2.DatabaseError` catch to both functions; log via
`log.exception` so the traceback is preserved. Tighten the existing
SaltMasterError messages to include jid/id (for returns) or the
batch size (for events) so operators can correlate the drop to the
lost payload. Behaviour is unchanged on the publish path; the
syndic-aggregate path is now properly contained.

While here, drop `event_return`'s positional `events` argument to
`_get_serv`. The `ret` parameter is the return-dict of one returner
call; passing a list of events was a copy-paste leftover from
`returner(ret)`. Today `salt.returners.get_returner_options` is
permissive enough that the result is the same as `ret=None`, but
the line has always been semantically wrong and is misleading to
read.

Add four behavioural tests:

- `test_returner_logs_on_connection_failure_without_raising`
- `test_returner_logs_on_database_error_without_raising`
- `test_event_return_inserts_each_event_into_salt_events` --
  happy-path coverage that did not exist; also pins
  `_get_serv(commit=True)` (no positional `events`).
- `test_event_return_logs_on_database_error_without_raising`

Depends on #69049 (which routed the `_get_serv`/`_purge_jobs`/
`_archive_jobs` errors through Salt's logger); this PR builds on
that branch and should be merged after it.

Refs: #69058

Author: co-cy

changelog/69058.fixed.md

@@ -0,0 +1,11 @@
+Fixed `salt.returners.pgjsonb.returner` letting any non-connection
+`psycopg2.DatabaseError` propagate to the caller — including the
+syndic-aggregate publish path in `salt/master.py` which had no outer
+catch — so a single bad row could escape into a master subprocess.
+`event_return` had no error handling at all and a database failure
+during a flush propagated similarly. Both functions now catch
+`SaltMasterError` and `psycopg2.DatabaseError` locally, log a
+contextual message (jid/id for returns, batch size for events), and
+drop the affected payload. While here, fix `event_return` passing
+the events list as the positional `ret` argument to `_get_serv`,
+which was a copy-paste leftover from `returner(ret)`.

salt/returners/pgjsonb.py

@@ -301,8 +301,15 @@ def returner(ret):
             )
     except salt.exceptions.SaltMasterError:
         log.critical(
-            "Could not store return with pgjsonb returner. PostgreSQL server"
-            " unavailable."
+            "pgjsonb: PostgreSQL unavailable, dropping return for jid=%s id=%s",
+            ret.get("jid"),
+            ret.get("id"),
+        )
+    except psycopg2.DatabaseError:
+        log.exception(
+            "pgjsonb: failed to store return for jid=%s id=%s",
+            ret.get("jid"),
+            ret.get("id"),
         )
 
 
@@ -313,15 +320,23 @@ def event_return(events):
     Requires that configuration be enabled via 'event_return'
     option in master config.
     """
-    with _get_serv(events, commit=True) as cur:
-        for event in events:
-            tag = event.get("tag", "")
-            data = event.get("data", "")
-            sql = """INSERT INTO salt_events (tag, data, master_id, alter_time)
-                     VALUES (%s, %s, %s, to_timestamp(%s))"""
-            cur.execute(
-                sql, (tag, psycopg2.extras.Json(data), __opts__["id"], time.time())
-            )
+    try:
+        with _get_serv(commit=True) as cur:
+            for event in events:
+                tag = event.get("tag", "")
+                data = event.get("data", "")
+                sql = """INSERT INTO salt_events (tag, data, master_id, alter_time)
+                         VALUES (%s, %s, %s, to_timestamp(%s))"""
+                cur.execute(
+                    sql,
+                    (tag, psycopg2.extras.Json(data), __opts__["id"], time.time()),
+                )
+    except salt.exceptions.SaltMasterError:
+        log.critical(
+            "pgjsonb: PostgreSQL unavailable, dropping %d event(s)", len(events)
+        )
+    except psycopg2.DatabaseError:
+        log.exception("pgjsonb: failed to store %d event(s)", len(events))
 
 
 def save_load(jid, load, minions=None):

tests/pytests/unit/returners/test_pgjsonb.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+import salt.exceptions
 import salt.returners.pgjsonb as pgjsonb
 from tests.support.mock import MagicMock, call, patch
 
@@ -209,3 +210,110 @@ def test__get_serv_logs_via_salt_logger_and_reraises_on_yield_error(caplog):
 
     fake_cursor.execute.assert_any_call("ROLLBACK")
     assert any("_get_serv" in r.message for r in caplog.records)
+
+
+@pytest.mark.skipif(not pgjsonb.HAS_PG, reason="psycopg2 not installed")
+def test_returner_logs_on_connection_failure_without_raising(caplog):
+    """When the database is unreachable, ``returner`` must not propagate
+    the SaltMasterError. The drop is logged at CRITICAL with jid and id
+    so operators can correlate it to the lost return."""
+    serv = MagicMock(side_effect=salt.exceptions.SaltMasterError("pg down"))
+    with patch.object(pgjsonb, "_get_serv", serv):
+        with caplog.at_level(logging.CRITICAL, logger="salt.returners.pgjsonb"):
+            pgjsonb.returner(
+                {
+                    "fun": "test.ping",
+                    "jid": "20260505000000000001",
+                    "id": "minion-1",
+                    "return": True,
+                }
+            )
+
+    assert any(
+        "PostgreSQL unavailable" in r.message
+        and "20260505000000000001" in r.message
+        and "minion-1" in r.message
+        for r in caplog.records
+    )
+
+
+@pytest.mark.skipif(not pgjsonb.HAS_PG, reason="psycopg2 not installed")
+def test_returner_logs_on_database_error_without_raising(caplog):
+    """A DatabaseError during the INSERT into ``salt_returns`` must not
+    propagate; ``returner`` logs and drops the return so that one bad row
+    cannot escape into syndic-aggregate paths that lack an outer catch."""
+    cur = MagicMock()
+    cur.execute.side_effect = psycopg2.DatabaseError("bad row")
+    serv = MagicMock()
+    serv.return_value.__enter__.return_value = cur
+
+    with patch.object(pgjsonb, "_get_serv", serv):
+        with caplog.at_level(logging.ERROR, logger="salt.returners.pgjsonb"):
+            pgjsonb.returner(
+                {
+                    "fun": "test.ping",
+                    "jid": "20260505000000000002",
+                    "id": "minion-2",
+                    "return": True,
+                }
+            )
+
+    assert any(
+        "failed to store return" in r.message
+        and "20260505000000000002" in r.message
+        and "minion-2" in r.message
+        for r in caplog.records
+    )
+
+
+@pytest.mark.skipif(not pgjsonb.HAS_PG, reason="psycopg2 not installed")
+def test_event_return_inserts_each_event_into_salt_events():
+    """``event_return`` writes one row to ``salt_events`` per queued event,
+    carrying the event tag and the master id. Also pins that ``_get_serv``
+    is opened with ``commit=True`` only -- passing the events list as the
+    positional ``ret`` argument was a copy-paste leftover from
+    ``returner(ret)``."""
+    events = [
+        {"tag": "salt/job/1/new", "data": {"jid": "1", "fun": "test.ping"}},
+        {"tag": "salt/auth", "data": {"id": "minion-1", "act": "accept"}},
+    ]
+    cur = MagicMock()
+    serv = MagicMock()
+    serv.return_value.__enter__.return_value = cur
+
+    with patch.object(pgjsonb, "_get_serv", serv):
+        with patch.dict(pgjsonb.__opts__, {"id": "master-A"}):
+            with patch("salt.returners.pgjsonb.time.time", return_value=1700000000.0):
+                pgjsonb.event_return(events)
+
+    serv.assert_called_once_with(commit=True)
+
+    assert cur.execute.call_count == len(events)
+    for executed_call, event in zip(cur.execute.call_args_list, events):
+        sql, params = executed_call.args
+        assert "INSERT INTO salt_events" in sql
+        tag, _data_json, master_id, ts = params
+        assert tag == event["tag"]
+        assert master_id == "master-A"
+        assert ts == 1700000000.0
+
+
+@pytest.mark.skipif(not pgjsonb.HAS_PG, reason="psycopg2 not installed")
+def test_event_return_logs_on_database_error_without_raising(caplog):
+    """A DatabaseError mid-batch must not propagate out of ``event_return``;
+    the queue length is logged so operators see how many events were lost."""
+    events = [{"tag": f"tag-{i}", "data": {}} for i in range(3)]
+    cur = MagicMock()
+    cur.execute.side_effect = psycopg2.DatabaseError("bad event")
+    serv = MagicMock()
+    serv.return_value.__enter__.return_value = cur
+
+    with patch.object(pgjsonb, "_get_serv", serv):
+        with patch.dict(pgjsonb.__opts__, {"id": "master-A"}):
+            with caplog.at_level(logging.ERROR, logger="salt.returners.pgjsonb"):
+                pgjsonb.event_return(events)
+
+    assert any(
+        "failed to store" in r.message and "3 event" in r.message
+        for r in caplog.records
+    )