From 9df6d8221afb9445d7845a81190f3ed8282cadb4 Mon Sep 17 00:00:00 2001 From: Lukas Holecek Date: Fri, 30 Oct 2026 09:08:55 +0000 Subject: [PATCH] Avoid crashing listener on failed policy matching An HTTP error while determining applicable policies for a subject, e.g. Koji rejecting an invalid NVR, crashed the listener. With the Kafka backend the message was never committed, so the listener kept crashing on it after every restart, blocking all messages behind it in the partition and failing to send decision updates. Assisted-by: GLM 5.3 JIRA: RHELWF-14872 --- greenwave/listeners/base.py | 15 ++++- greenwave/tests/test_listeners.py | 100 +++++++++++++++++++++++++++++- 2 files changed, 111 insertions(+), 4 deletions(-) diff --git a/greenwave/listeners/base.py b/greenwave/listeners/base.py index a75d7e2f..1430e733 100644 --- a/greenwave/listeners/base.py +++ b/greenwave/listeners/base.py @@ -379,9 +379,18 @@ def _publish_decision_change( policy_attributes["product_version"] = product_version policies = self.app.config["policies"] - contexts_product_versions = applicable_decision_context_product_version_pairs( - policies, **policy_attributes - ) + try: + contexts_product_versions = ( + applicable_decision_context_product_version_pairs( + policies, **policy_attributes + ) + ) + except (HTTPError, HTTPException): + self.app.logger.exception( + "Failed to find applicable policies for data=%s", policy_attributes + ) + self._inc(decision_failed_counter.labels(decision_context="")) + return for decision_context, pv in sorted(contexts_product_versions): old_decision, decision = self._old_and_new_decisions( diff --git a/greenwave/tests/test_listeners.py b/greenwave/tests/test_listeners.py index a849a8fd..5d9bf563 100644 --- a/greenwave/tests/test_listeners.py +++ b/greenwave/tests/test_listeners.py @@ -6,9 +6,10 @@ import pytest import stomp from requests.exceptions import HTTPError -from werkzeug.exceptions import NotFound +from werkzeug.exceptions import BadGateway, NotFound from greenwave.app_factory import create_app +from greenwave.config import TestingConfig from greenwave.listeners.resultsdb import ResultsDBListener from greenwave.listeners.waiverdb import WaiverDBListener from greenwave.monitor import ( @@ -86,6 +87,30 @@ def body(self): return json.dumps(self.message) +class DummyKafkaMessage: + """ + Simple stand-in for confluent_kafka.Message, as returned by + KafkaBus.poll(). + """ + + def __init__(self, body, offset=67818): + self._body = body + self._offset = offset + + def error(self): + return None + + def value(self): + return self._body + + def offset(self): + return self._offset + + +class KafkaTestingConfig(TestingConfig): + LISTENER_BACKEND = "kafka" + + def resultsdb_listener(policies=POLICIES_DEFAULT): listener = ResultsDBListener(config_obj=CONFIG_NAME) listener.app.config["policies"] = Policy.safe_load_all(policies) @@ -545,6 +570,79 @@ def retrieve_decision(data, _config): assert len(mock_connection.send.mock_calls) == 0 +def test_listener_survives_failed_policy_matching( + mock_retrieve_scm_from_koji, + mock_retrieve_yaml_remote_rule, + mock_retrieve_decision, + mock_retrieve_results, + mock_connection, + koji_proxy, +): + """ + Test that the listener keeps processing messages when applicable + policies cannot be determined for a subject, e.g. when Koji rejects + an invalid NVR such as a bare package name. + """ + mock_retrieve_scm_from_koji.side_effect = BadGateway( + 'Failed to get Koji build for "pki-core": invalid format: pki-core (code: 1000)' + ) + + listener = resultsdb_listener(POLICIES_WITH_REMOTE_RULE) + + with listener.app.app_context(): + listener.on_message(DummyMessage(nvr="pki-core")) + + # No decision update is published for the invalid subject. + assert len(mock_connection.send.mock_calls) == 0 + + # The listener still processes the next message. + gating_yaml = dedent(""" + --- !Policy + product_versions: [fedora-rawhide] + decision_context: test_context + rules: + - !PassingTestCaseRule {test_case_name: dist.rpmdeplint} + """) + mock_retrieve_yaml_remote_rule.return_value = gating_yaml + mock_retrieve_scm_from_koji.side_effect = None + mock_retrieve_scm_from_koji.return_value = ( + "rpms", + DUMMY_NVR, + "c3c47a08a66451cb9686c49f040776ed35a0d1bb", + ) + + with listener.app.app_context(): + listener.on_message(DummyMessage()) + + assert len(mock_connection.send.mock_calls) == 1 + + +def test_kafka_listener_commits_message_on_failed_policy_matching( + mock_retrieve_scm_from_koji, + koji_proxy, +): + """ + Test that the Kafka consumer commits a message even when applicable + policies cannot be determined for its subject, e.g. when Koji rejects + an invalid NVR such as a bare package name. Not committing such a + message would make the consumer loop fail on it again after every + restart, blocking all messages behind it in the partition. + """ + mock_retrieve_scm_from_koji.side_effect = BadGateway( + 'Failed to get Koji build for "pki-core": invalid format: pki-core (code: 1000)' + ) + + listener = ResultsDBListener(config_obj=KafkaTestingConfig) + listener.app.config["policies"] = Policy.safe_load_all(POLICIES_WITH_REMOTE_RULE) + listener._kafka_bus = mock.Mock() + + message = DummyKafkaMessage(DummyMessage(nvr="pki-core").body) + listener._handle_kafka_message(message) + + listener._kafka_bus.publish.assert_not_called() + listener._kafka_bus.commit.assert_called_once_with(message) + + def test_guess_product_version(mock_connection, koji_proxy): listener = resultsdb_listener()