From 5eec6d928ada69c3d4e3d46baa4444345ccf8fc9 Mon Sep 17 00:00:00 2001 From: Maksym Yankin Date: Mon, 13 Jul 2026 17:58:48 +0300 Subject: [PATCH] [19.0][FIX] edi_queue_oca: propagate failed jobs to exchanges When an EDI queue job reached its maximum retries, only the queue job was marked as failed while the related exchange remained pending. Propagate terminal send, receive, and process job failures to the corresponding exchange error state, including the exception message and traceback. --- edi_queue_oca/models/__init__.py | 1 + edi_queue_oca/models/edi_exchange_record.py | 27 +++++ edi_queue_oca/models/queue_job.py | 28 +++++ edi_queue_oca/tests/test_backend_jobs.py | 108 +++++++++++++++++++- 4 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 edi_queue_oca/models/queue_job.py diff --git a/edi_queue_oca/models/__init__.py b/edi_queue_oca/models/__init__.py index e1a8e889b..e6c8d6714 100644 --- a/edi_queue_oca/models/__init__.py +++ b/edi_queue_oca/models/__init__.py @@ -1,3 +1,4 @@ from . import edi_exchange_record from . import edi_exchange_type from . import edi_backend +from . import queue_job diff --git a/edi_queue_oca/models/edi_exchange_record.py b/edi_queue_oca/models/edi_exchange_record.py index c8ddeff7d..aa2769a7e 100644 --- a/edi_queue_oca/models/edi_exchange_record.py +++ b/edi_queue_oca/models/edi_exchange_record.py @@ -66,6 +66,33 @@ def delayable(self, **kw): def _job_retry_params(self): return {} + def _mark_failed_from_queue_job(self, job): + """Set the EDI error matching a terminal queue job failure. + + :param job: failed ``queue.job`` record + """ + failure_mapping = { + "action_exchange_process": ("input_processed_error", "process_ko"), + "action_exchange_receive": ("input_receive_error", "receive_ko"), + "action_exchange_send": ("output_error_on_send", "send_ko"), + } + failure = failure_mapping.get(job.method_name) + if not failure: + return + state, message_key = failure + for record in self: + state_changed = record.edi_exchange_state != state + record.write( + { + "edi_exchange_state": state, + "exchange_error": job.exc_message, + "exchange_error_traceback": job.exc_info, + "exchanged_on": fields.Datetime.now(), + } + ) + if state_changed: + record._notify_error(message_key) + def _compute_related_queue_jobs_count(self): for rec in self: # TODO: We should refactor the object field on queue_job to use jsonb field diff --git a/edi_queue_oca/models/queue_job.py b/edi_queue_oca/models/queue_job.py new file mode 100644 index 000000000..d8bfe8da9 --- /dev/null +++ b/edi_queue_oca/models/queue_job.py @@ -0,0 +1,28 @@ +# Copyright 2026 Camptocamp SA +# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl). + +from odoo import models + + +class QueueJob(models.Model): + _inherit = "queue.job" + + def write(self, vals): + result = super().write(vals) + if vals.get("state") == "failed": + self._mark_related_edi_exchanges_failed() + return result + + def _mark_related_edi_exchanges_failed(self): + """Propagate terminal EDI job failures to their exchange records.""" + supported_methods = { + "action_exchange_process", + "action_exchange_receive", + "action_exchange_send", + } + jobs = self.filtered( + lambda job: job.model_name == "edi.exchange.record" + and job.method_name in supported_methods + ) + for job in jobs: + job.records.sudo()._mark_failed_from_queue_job(job) diff --git a/edi_queue_oca/tests/test_backend_jobs.py b/edi_queue_oca/tests/test_backend_jobs.py index 61ac106bd..f8b149a6e 100644 --- a/edi_queue_oca/tests/test_backend_jobs.py +++ b/edi_queue_oca/tests/test_backend_jobs.py @@ -94,6 +94,7 @@ def test_output(self): self.assertEqual(created, self._get_related_jobs(record)) def test_output_fail_retry(self): + """Test a retryable send failure keeps the exchange pending.""" job_counter = self.job_counter() vals = { "model": self.partner._name, @@ -106,8 +107,113 @@ def test_output_fail_retry(self): job_counter.search_created() with mock.patch.object(type(self.backend), "_exchange_send") as mocked: mocked.side_effect = ReqConnectionError("Connection broken") - with self.assertRaises(RetryableJobError): + with self.assertRaisesRegex(RetryableJobError, "Connection broken"): job.perform() + self.assertEqual(record.edi_exchange_state, "output_pending") + + def test_failed_send_job_marks_exchange_as_error(self): + """Test a terminal send job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_output", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "output_pending", + }, + ) + record._set_file_content("ABC") + job = record.with_delay().action_exchange_send() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Connection broken", + "exc_info": "Traceback of the connection failure", + } + ) + + self.assertEqual(record.edi_exchange_state, "output_error_on_send") + self.assertEqual(record.exchange_error, "Connection broken") + self.assertEqual( + record.exchange_error_traceback, "Traceback of the connection failure" + ) + self.assertTrue(record.exchanged_on) + + def test_failed_receive_job_marks_exchange_as_error(self): + """Test a terminal receive job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_input", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "input_pending", + }, + ) + job = record.with_delay().action_exchange_receive() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Receive failed", + "exc_info": "Traceback for receive", + } + ) + + self.assertEqual(record.edi_exchange_state, "input_receive_error") + self.assertEqual(record.exchange_error, "Receive failed") + self.assertEqual(record.exchange_error_traceback, "Traceback for receive") + + def test_failed_process_job_marks_exchange_as_error(self): + """Test a terminal process job failure marks its exchange as failed.""" + record = self.backend.create_record( + "test_csv_input", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "input_received", + }, + ) + job = record.with_delay().action_exchange_process() + + job.db_record().write( + { + "state": "failed", + "exc_message": "Process failed", + "exc_info": "Traceback for process", + } + ) + + self.assertEqual(record.edi_exchange_state, "input_processed_error") + self.assertEqual(record.exchange_error, "Process failed") + self.assertEqual(record.exchange_error_traceback, "Traceback for process") + + def test_unsupported_failed_jobs_do_not_mark_exchange_as_error(self): + """Test generate and non-exchange jobs do not alter the exchange state.""" + record = self.backend.create_record( + "test_csv_output", + { + "model": self.partner._name, + "res_id": self.partner.id, + "edi_exchange_state": "output_pending", + }, + ) + jobs = ( + record.with_delay().action_exchange_generate(), + self.backend.with_delay().exchange_send(record), + ) + + for job in jobs: + job.db_record().write( + { + "state": "failed", + "exc_message": "Unsupported job failed", + "exc_info": "Unsupported job traceback", + } + ) + + self.assertEqual(record.edi_exchange_state, "output_pending") + self.assertFalse(record.exchange_error) + self.assertFalse(record.exchange_error_traceback) def test_input(self): job_counter = self.job_counter()