Merge pull request #659 from ianco/master

Check for fatal ledger errors on dispatch task completion

aries_cloudagent/core/conductor.py

@@ -274,6 +274,13 @@ def dispatch_complete(self, message: InboundMessage, completed: CompletedTask):
             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
+            ):
+                LOGGER.error("Shutdown with %s", str(completed.exc_info[1]))
+                self.admin_server.notify_fatal_error()
+            else:
+                LOGGER.error("DON'T Shutdown with %s", str(completed.exc_info[1]))
         self.inbound_transport_manager.dispatch_complete(message, completed)
 
     async def get_stats(self) -> dict:

aries_cloudagent/core/tests/test_conductor.py

@@ -576,3 +576,24 @@ async def test_webhook_router(self):
             mock_enqueue.assert_called_once_with(
                 test_topic, test_payload, test_endpoint, test_attempts
             )
+
+    async def test_dispatch_complete_fatal_x(self):
+        builder: ContextBuilder = StubContextBuilder(self.test_settings)
+        conductor = test_module.Conductor(builder)
+
+        message_body = "{}"
+        receipt = MessageReceipt(direct_response_mode="snail mail")
+        message = InboundMessage(message_body, receipt)
+        mock_task = async_mock.MagicMock(
+            exc_info=(test_module.LedgerTransactionError, ("Ledger is wobbly"), "..."),
+            ident="abc",
+            timing={
+                "queued": 1234567890,
+                "unqueued": 1234567899,
+                "started": 1234567901,
+                "ended": 1234567999,
+            },
+        )
+
+        await conductor.setup()
+        conductor.dispatch_complete(message, mock_task)