Enhance service bus handling of invalid JSON in receive_message function

CHANGELOG.md

@@ -17,6 +17,7 @@ ENHANCEMENTS:
 * Update the version of `super-linter` used in the `build_validation_develop` workflow to 8.7.0 ([#4957](https://github.com/microsoft/AzureTRE/issues/4957))
 
 BUG FIXES:
+* Fix to enhance service bus handling of invalid JSON in receive_message function ([#4932](https://github.com/microsoft/AzureTRE/pull/4932))
 * Fix workspace deletion when backup is enabled for the base, unrestricted and airlock-import-review workspaces by adding a `delete_backups_on_uninstall` flag and a pre-teardown backup cleanup (`remove_backup.sh`) that stops protection and either deletes or retains the Recovery Services Vault, so deletion works with Azure secure-by-default soft delete ([#4962](https://github.com/microsoft/AzureTRE/issues/4962))
 * Fix Nexus shared service security: fetch admin password from Key Vault at runtime via managed identity (IMDS) instead of embedding it in the VM Run Command script content. Fix `deploy_nexus_container.sh` short-circuit path to fail loudly if the container does not start. (`sonatype-nexus` 3.10.0) ([#4983](https://github.com/microsoft/AzureTRE/pull/4983))
 * Fix UI TypeScript deprecation warning by updating `moduleResolution` to `bundler` in `tsconfig.json`. ([#4968](https://github.com/microsoft/AzureTRE/issues/4968))

resource_processor/_version.py

@@ -1 +1 @@
-__version__ = "0.13.5"
+__version__ = "0.13.6"

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"])