Enhance service bus handling of invalid JSON in receive_message function

CHANGELOG.md

@@ -11,6 +11,7 @@ ENHANCEMENTS:
 * Exclude recovery service vaults from e2e tests ([#4920](https://github.com/microsoft/AzureTRE/issues/4920))
 
 BUG FIXES:
+* Fix to enhance service bus handling of invalid JSON in receive_message function ([#4932](https://github.com/microsoft/AzureTRE/pull/4932))
 * Fix API timeout and name collision failures on workspace creation by checking storage account name availability and improved logging. ([#4946](https://github.com/microsoft/AzureTRE/pull/4946))
 * Fix error handling in airlock processor ([#4929](https://github.com/microsoft/AzureTRE/pull/4929))
 

resource_processor/_version.py

@@ -1 +1 @@
-__version__ = "0.13.3"
+__version__ = "0.13.4"

resource_processor/tests_rp/test_runner.py

@@ -124,10 +124,40 @@ async def test_receive_message(mock_invoke_porter_action, mock_service_bus_clien
     config = {"resource_request_queue": "test_queue"}
 
     await receive_message(mock_service_bus_client_instance, config, keep_running=run_once)
-    mock_receiver.complete_message.assert_called_once()
+    mock_receiver.complete_message.assert_awaited_once()
     mock_service_bus_client_instance.get_queue_receiver.assert_called_once_with(queue_name="test_queue", max_wait_time=1, session_id=ServiceBusSessionFilter.NEXT_AVAILABLE)
 
 
+@pytest.mark.asyncio
+async def test_receive_message_bad_json(mock_service_bus_client, mock_auto_lock_renewer):
+    mock_service_bus_client_instance = mock_service_bus_client.return_value
+
+    # Set up the lock renewer mock correctly
+    mock_renewer = AsyncMock()
+    mock_renewer.register = Mock()
+    mock_auto_lock_renewer.return_value.__aenter__.return_value = mock_renewer
+
+    mock_receiver = AsyncMock()
+    mock_receiver.__aenter__.return_value = mock_receiver
+    mock_receiver.__aexit__.return_value = None
+    mock_receiver.session.session_id = "test_session_id"
+    mock_receiver.__aiter__.return_value = ["invalid_json_string"]
+
+    mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver
+
+    run_once = Mock(side_effect=[True, False])
+
+    config = {"resource_request_queue": "test_queue"}
+
+    await receive_message(mock_service_bus_client_instance, config, keep_running=run_once)
+    mock_receiver.dead_letter_message.assert_awaited_once_with(
+        "invalid_json_string",
+        reason="InvalidJSON",
+        error_description="Expecting value: line 1 column 1 (char 0)"
+    )
+    mock_receiver.complete_message.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_receive_message_unknown_exception(mock_auto_lock_renewer, mock_service_bus_client, mock_logger):
     """Test receiving a message with an unknown exception."""

resource_processor/vmss_porter/runner.py

@@ -73,6 +73,8 @@ async def receive_message(service_bus_client, config: dict, keep_running=lambda:
                             message = json.loads(str(msg))
                         except (json.JSONDecodeError) as e:
                             logger.error(f"Received bad service bus resource request message: {e}")
+                            await receiver.dead_letter_message(msg, reason="InvalidJSON", error_description=str(e))
+                            continue
 
                         with tracer.start_as_current_span("receive_message") as current_span:
                             current_span.set_attribute("resource_id", message["id"])