✨ Don't shutdown on ledger error (openwallet-foundation#3636)

* ✨ Don't shutdown on ledger error

Signed-off-by: ff137 <[email protected]>

* ✅ Fix tests

Signed-off-by: ff137 <[email protected]>

* 🧪 Don't skip test

Signed-off-by: ff137 <[email protected]>

* 🧪 Cover StorageNotFoundError scenario

Signed-off-by: ff137 <[email protected]>

* 🧪 Fix confusing unawaited coroutine warning

Signed-off-by: ff137 <[email protected]>

---------

Signed-off-by: ff137 <[email protected]>
Co-authored-by: Stephen Curran <[email protected]>

Author: ff137

acapy_agent/core/conductor.py

@@ -74,7 +74,7 @@
 from ..wallet.anoncreds_upgrade import upgrade_wallet_to_anoncreds_if_requested
 from ..wallet.did_info import DIDInfo
 from .dispatcher import Dispatcher
-from .error import StartupError
+from .error import ProfileError, StartupError
 from .oob_processor import OobMessageProcessor
 from .util import SHUTDOWN_EVENT_TOPIC, STARTUP_EVENT_TOPIC
 
@@ -559,31 +559,31 @@ def inbound_message_router(
                 lambda completed: self.dispatch_complete(message, completed),
             )
         except (LedgerConfigError, LedgerTransactionError) as e:
-            LOGGER.error("Shutdown on ledger error %s", str(e))
-            if self.admin_server:
-                self.admin_server.notify_fatal_error()
+            LOGGER.error("Ledger error occurred in message handler: %s", str(e))
             raise
 
     def dispatch_complete(self, message: InboundMessage, completed: CompletedTask):
         """Handle completion of message dispatch."""
         if completed.exc_info:
-            LOGGER.exception("Exception in message handler:", exc_info=completed.exc_info)
-            if isinstance(completed.exc_info[1], LedgerConfigError) or isinstance(
-                completed.exc_info[1], LedgerTransactionError
-            ):
+            exc_class, exc, _ = completed.exc_info
+            if isinstance(exc, (LedgerConfigError, LedgerTransactionError)):
                 LOGGER.error(
-                    "%shutdown on ledger error %s",
-                    "S" if self.admin_server else "No admin server to s",
-                    str(completed.exc_info[1]),
+                    "Ledger error occurred in message handler: %s",
+                    str(exc),
+                    exc_info=completed.exc_info,
                 )
-                if self.admin_server:
-                    self.admin_server.notify_fatal_error()
-            else:
+            elif isinstance(exc, (ProfileError, StorageNotFoundError)):
                 LOGGER.error(
-                    "DON'T shutdown on %s %s",
-                    completed.exc_info[0].__name__,
-                    str(completed.exc_info[1]),
+                    "Storage error occurred in message handler: %s: %s",
+                    exc_class.__name__,
+                    str(exc),
+                    exc_info=completed.exc_info,
+                )
+            else:
+                LOGGER.exception(
+                    "Exception in message handler:", exc_info=completed.exc_info
                 )
+
         self.inbound_transport_manager.dispatch_complete(message, completed)
 
     async def get_stats(self) -> dict:
@@ -656,9 +656,9 @@ def handle_not_returned(self, profile: Profile, outbound: OutboundMessage):
         try:
             self.dispatcher.run_task(self.queue_outbound(profile, outbound))
         except (LedgerConfigError, LedgerTransactionError) as e:
-            LOGGER.error("Shutdown on ledger error %s", str(e))
-            if self.admin_server:
-                self.admin_server.notify_fatal_error()
+            LOGGER.error(
+                "Ledger error occurred while handling failed delivery: %s", str(e)
+            )
             raise
 
     async def queue_outbound(
@@ -688,9 +688,9 @@ async def queue_outbound(
                 LOGGER.exception("Error preparing outbound message for transmission")
                 return OutboundSendStatus.UNDELIVERABLE
             except (LedgerConfigError, LedgerTransactionError) as e:
-                LOGGER.error("Shutdown on ledger error %s", str(e))
-                if self.admin_server:
-                    self.admin_server.notify_fatal_error()
+                LOGGER.error(
+                    "Ledger error occurred while preparing outbound message: %s", str(e)
+                )
                 raise
             del conn_mgr
         # Find oob/connectionless target we can send the message to

acapy_agent/core/tests/test_conductor.py

@@ -1,12 +1,10 @@
 from unittest import IsolatedAsyncioTestCase
 
-import pytest
-
-from ...connections.base_manager import BaseConnectionManager
 from ...admin.base_server import BaseAdminServer
 from ...askar.profile import AskarProfileManager
 from ...config.base_context import ContextBuilder
 from ...config.injection_context import InjectionContext
+from ...connections.base_manager import BaseConnectionManager
 from ...connections.models.conn_record import ConnRecord
 from ...connections.models.connection_target import ConnectionTarget
 from ...connections.models.diddoc import DIDDoc, PublicKey, PublicKeyType, Service
@@ -587,9 +585,7 @@ async def test_inbound_message_handler_ledger_x(self):
             mock.patch.object(
                 conductor.dispatcher, "queue_message", autospec=True
             ) as mock_dispatch_q,
-            mock.patch.object(
-                conductor.admin_server, "notify_fatal_error", mock.MagicMock()
-            ) as mock_notify,
+            mock.patch.object(test_module, "LOGGER", mock.MagicMock()) as mock_logger,
         ):
             mock_dispatch_q.side_effect = test_module.LedgerConfigError("ledger down")
 
@@ -603,7 +599,7 @@ async def test_inbound_message_handler_ledger_x(self):
                 )
 
             mock_dispatch_q.assert_called_once()
-            mock_notify.assert_called_once()
+            mock_logger.error.assert_called_once()
 
     async def test_outbound_message_handler_return_route(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings)
@@ -830,18 +826,10 @@ async def test_handle_nots(self):
                     conductor.dispatcher, "run_task", mock.MagicMock()
                 ) as mock_run_task,
             ):
-                # Normally this should be a coroutine mock; however, the coroutine
-                # is awaited by dispatcher.run_task, which is mocked here. MagicMock
-                # to prevent unawaited coroutine warning.
-                mock_conn_mgr.return_value.get_connection_targets = mock.MagicMock()
                 mock_run_task.side_effect = test_module.BaseConnectionManagerError()
                 await conductor.queue_outbound(conductor.root_profile, message)
-                mock_outbound_mgr.return_value.enqueue_message.assert_not_called()
 
                 message.connection_id = None
-                mock_outbound_mgr.return_value.enqueue_message.side_effect = (
-                    test_module.OutboundDeliveryError()
-                )
                 await conductor.queue_outbound(conductor.root_profile, message)
                 mock_run_task.assert_called_once()
 
@@ -870,7 +858,6 @@ async def test_handle_outbound_queue(self):
 
         await conductor.queue_outbound(conductor.root_profile, message)
 
-    @pytest.mark.skip("This test has a bad mock that isn't awaited")
     async def test_handle_not_returned_ledger_x(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings_admin)
         conductor = test_module.Conductor(builder)
@@ -898,9 +885,7 @@ async def test_handle_not_returned_ledger_x(self):
             mock.patch.object(
                 conductor.dispatcher, "run_task", mock.MagicMock()
             ) as mock_dispatch_run,
-            mock.patch.object(
-                conductor.admin_server, "notify_fatal_error", mock.MagicMock()
-            ) as mock_notify,
+            mock.patch.object(conductor, "queue_outbound", mock.MagicMock()),
         ):
             mock_dispatch_run.side_effect = test_module.LedgerConfigError(
                 "No such ledger"
@@ -917,7 +902,6 @@ async def test_handle_not_returned_ledger_x(self):
                 conductor.handle_not_returned(conductor.root_profile, message)
 
             mock_dispatch_run.assert_called_once()
-            mock_notify.assert_called_once()
 
     async def test_queue_outbound_ledger_x(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings_admin)
@@ -949,14 +933,8 @@ async def test_queue_outbound_ledger_x(self):
             mock.patch.object(
                 conductor.dispatcher, "run_task", mock.MagicMock()
             ) as mock_dispatch_run,
-            mock.patch.object(
-                conductor.admin_server, "notify_fatal_error", mock.MagicMock()
-            ) as mock_notify,
+            mock.patch.object(test_module, "LOGGER", mock.MagicMock()) as mock_logger,
         ):
-            # Normally this should be a coroutine mock; however, the coroutine
-            # is awaited by dispatcher.run_task, which is mocked here. MagicMock
-            # to prevent unawaited coroutine warning.
-            conn_mgr.get_connection_targets = mock.MagicMock()
             mock_dispatch_run.side_effect = test_module.LedgerConfigError(
                 "No such ledger"
             )
@@ -972,7 +950,7 @@ async def test_queue_outbound_ledger_x(self):
                 await conductor.queue_outbound(conductor.root_profile, message)
 
             mock_dispatch_run.assert_called_once()
-            mock_notify.assert_called_once()
+            mock_logger.error.assert_called_once()
 
     async def test_admin(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings)
@@ -1217,7 +1195,7 @@ async def test_dispatch_complete_non_fatal_x(self):
         message_body = "{}"
         receipt = MessageReceipt(direct_response_mode="snail mail")
         message = InboundMessage(message_body, receipt)
-        exc = KeyError("sample exception")
+        exc = StorageNotFoundError("sample exception")
         mock_task = mock.MagicMock(
             exc_info=(type(exc), exc, exc.__traceback__),
             ident="abc",
@@ -1255,7 +1233,7 @@ async def test_dispatch_complete_non_fatal_x(self):
             conductor.dispatch_complete(message, mock_task)
             mock_notify.assert_not_called()
 
-    async def test_dispatch_complete_fatal_x(self):
+    async def test_dispatch_complete_ledger_error_x(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings_admin)
         conductor = test_module.Conductor(builder)
 
@@ -1294,11 +1272,9 @@ async def test_dispatch_complete_fatal_x(self):
             }
             await conductor.setup()
 
-        with mock.patch.object(
-            conductor.admin_server, "notify_fatal_error", mock.MagicMock()
-        ) as mock_notify:
+        with mock.patch.object(test_module, "LOGGER", mock.MagicMock()) as mock_logger:
             conductor.dispatch_complete(message, mock_task)
-            mock_notify.assert_called_once_with()
+            mock_logger.error.assert_called_once()
 
     async def test_clear_default_mediator(self):
         builder: ContextBuilder = StubContextBuilder(self.test_settings)