From 66232ccb2847b09fd4aa78e656f1d1e9d1e0ff4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9amus=20=C3=93=20Ceanainn?= Date: Thu, 4 Jun 2026 01:50:53 +0100 Subject: [PATCH 1/3] Ignore errors when attempting to restore messages if the message was acked --- kombu/transport/virtual/base.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/kombu/transport/virtual/base.py b/kombu/transport/virtual/base.py index 4c3ede5140..221ae670f1 100644 --- a/kombu/transport/virtual/base.py +++ b/kombu/transport/virtual/base.py @@ -260,14 +260,18 @@ def restore_unacked(self): while delivered: try: - _, message = pop_message() + key, message = pop_message() except KeyError: # pragma: no cover break try: restore(message) except BaseException as exc: - errors.append((exc, message)) + if key not in self._dirty: + # Another thread may have acked the message after the earlier '_flush' call. + # This may cause the restore attempt to fail (e.g. in SQS). + # If restore fails, we only care about errors for messages that have not been 'acked'. + errors.append((exc, message)) delivered.clear() return errors From 8d1476b4decce15acde95c84adb996e8c5821ae9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9amus=20=C3=93=20Ceanainn?= Date: Thu, 4 Jun 2026 16:24:43 +0100 Subject: [PATCH 2/3] Add test for new code --- t/unit/transport/virtual/test_base.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/t/unit/transport/virtual/test_base.py b/t/unit/transport/virtual/test_base.py index 0c3dd1f08e..d8e321f5d3 100644 --- a/t/unit/transport/virtual/test_base.py +++ b/t/unit/transport/virtual/test_base.py @@ -410,6 +410,22 @@ def test_restore_unacked_raises_BaseException(self): assert errors[0][1] == 1 assert not q._delivered + def test_restore_unacked_ignores_raised_exceptions_when_acked(self): + q = self.channel.qos + q._flush = Mock() + q._delivered = {1: 1} + + def mock_restore_raises_exceptions_due_to_acked_message(*args, **kwargs): + q._dirty = {1} # acked dirty message + raise SystemExit(1) + + q.channel._restore = Mock() + q.channel._restore.side_effect = mock_restore_raises_exceptions_due_to_acked_message + + errors = q.restore_unacked() + assert not errors + assert not q._delivered + @patch('kombu.transport.virtual.base.emergency_dump_state') @patch(PRINT_FQDN) def test_restore_unacked_once_when_unrestored(self, print_, From 6fd1cc62d3e57123ac52305795eb16657951df0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9amus=20=C3=93=20Ceanainn?= Date: Tue, 9 Jun 2026 00:12:25 +0200 Subject: [PATCH 3/3] Use q.ack(1) instead of assigning to _dirty directly --- t/unit/transport/virtual/test_base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/t/unit/transport/virtual/test_base.py b/t/unit/transport/virtual/test_base.py index d8e321f5d3..bf2493dca2 100644 --- a/t/unit/transport/virtual/test_base.py +++ b/t/unit/transport/virtual/test_base.py @@ -416,7 +416,7 @@ def test_restore_unacked_ignores_raised_exceptions_when_acked(self): q._delivered = {1: 1} def mock_restore_raises_exceptions_due_to_acked_message(*args, **kwargs): - q._dirty = {1} # acked dirty message + q.ack(1) # simulate concurrent ack of the delivered message raise SystemExit(1) q.channel._restore = Mock()