CCM-17531: Mesh download out-of-memory exception

.github/actions/acceptance-tests/action.yaml

@@ -77,9 +77,16 @@ runs:
         TEST_TYPE: ${{ inputs.testType }}
         ENVIRONMENT: ${{ inputs.targetEnvironment }}
         PLAYWRIGHT_SHARD: ${{ inputs.shard }}
+    - name: Sanitise shard for artifact name
+      id: shard_label
+      if: ${{ inputs.testType == 'integration' && inputs.shard != '' }}
+      shell: bash
+      run: echo "value=${SHARD//\// of }" >> $GITHUB_OUTPUT
+      env:
+        SHARD: ${{ inputs.shard }}
     - name: Archive integration test results
       if: ${{ inputs.testType == 'integration' }}
       uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
       with:
-        name: Integration test report
+        name: Integration test report${{ inputs.shard != '' && format(' ({0})', steps.shard_label.outputs.value) || '' }}
         path: "tests/playwright/playwright-report"

lambdas/mesh-download/mesh_download/__tests__/test_processor.py

@@ -33,7 +33,6 @@ def setup_mocks():
 
     return config, log, event_publisher, document_store
 
-
 def create_valid_cloud_event():
     """
     Create a valid CloudEvent for testing
@@ -60,7 +59,6 @@ def create_valid_cloud_event():
         }
     }
 
-
 def create_sqs_record(cloud_event=None):
     """
     Create a mock SQS record containing a CloudEvent
@@ -612,3 +610,53 @@ def test_trust_duplicate_publishes_invalid_event_and_acknowledges(self, mock_dat
         assert event_data['failureCode'] == 'DL_CLIV_004'
 
         mesh_message.acknowledge.assert_called_once()
+
+
+class TestValidateFhirContent:
+    """Tests for _validate_fhir_content base64 checks"""
+
+    def make_processor(self):
+        from mesh_download.processor import MeshDownloadProcessor
+        config, log, event_publisher, document_store = setup_mocks()
+        return MeshDownloadProcessor(
+            config=config,
+            log=log,
+            mesh_client=config.mesh_client,
+            download_metric=config.download_metric,
+            internal_duplicate_download_metric=config.internal_duplicate_download_metric,
+            trust_duplicate_download_metric=config.trust_duplicate_download_metric,
+            document_store=document_store,
+            event_publisher=event_publisher,
+        )
+
+    def fhir_with_data(self, data: str) -> str:
+        content = json.loads(create_fhir_content())
+        content['content'][0]['attachment']['data'] = data
+        return json.dumps(content)
+
+    def test_valid_base64_passes(self):
+        """Valid base64 data should not raise"""
+        processor = self.make_processor()
+        # 'SGVsbG8gV29ybGQ=' is 'Hello World' in base64 — valid padding and characters
+        with patch('mesh_download.processor.validate'):
+            processor._validate_fhir_content(self.fhir_with_data('SGVsbG8gV29ybGQ='))
+
+    def test_invalid_characters_raises(self):
+        """Empty base64 data string should raise"""
+        processor = self.make_processor()
+        with pytest.raises(ValueError, match="must not be empty"):
+            processor._validate_fhir_content(self.fhir_with_data(''))
+
+    def test_invalid_padding_raises(self):
+        """base64 string whose length is not a multiple of 4 should raise"""
+        processor = self.make_processor()
+        # Length 5 — not a multiple of 4
+        with pytest.raises(ValueError, match="incorrect padding"):
+            processor._validate_fhir_content(self.fhir_with_data('AAAAA'))
+
+    def test_invalid_characters_raises(self):
+        """base64 string containing characters outside the alphabet should raise"""
+        processor = self.make_processor()
+        # '!' is not in the base64 alphabet
+        with pytest.raises(ValueError, match="invalid characters"):
+            processor._validate_fhir_content(self.fhir_with_data('!AAA'))

lambdas/mesh-download/mesh_download/handler.py

@@ -1,6 +1,5 @@
 """lambda handler for mesh download"""
 
-import json
 from dl_utils import EventPublisher
 
 from .config import Config, log
@@ -25,6 +24,20 @@ def handler(event, context):
         'failed': 0
     }
 
+    def _get_memory_mb() -> dict:
+        current_kb = 0
+        peak_kb = 0
+        with open('/proc/self/status') as f:
+            for line in f:
+                if line.startswith('VmRSS:'):
+                    current_kb = int(line.split()[1])
+                elif line.startswith('VmHWM:'):
+                    peak_kb = int(line.split()[1])
+        return {
+            'current_mb': current_kb / 1024,
+            'peak_mb': peak_kb / 1024,
+        }
+
     try:
         with Config() as config:
             doc_store_config = DocumentStoreConfig(
@@ -76,6 +89,8 @@ def handler(event, context):
                 skipped=processed['skipped'],
                 failed=processed['failed'])
 
+        log.info("Memory after processing SQS event", memory=_get_memory_mb())
+
         return {"batchItemFailures": batch_item_failures}
 
     except Exception as exc:

lambdas/mesh-download/mesh_download/processor.py

@@ -1,4 +1,5 @@
 import json
+import re
 from datetime import datetime, timezone
 from uuid import uuid4
 
@@ -58,6 +59,23 @@ def _parse_and_validate_event(self, sqs_record):
 
     def _validate_fhir_content(self, content):
         json_content = json.loads(content)
+
+        for item in json_content.get('content', []):
+            b64 = item.get('attachment', {}).get('data')
+            if b64 is not None:
+                if len(b64) == 0:
+                    raise ValueError(
+                        "Attachment base64 data must not be empty"
+                    )
+                if len(b64) % 4 != 0:
+                    raise ValueError(
+                        f"Attachment base64 data has incorrect padding (length {len(b64)} is not a multiple of 4)"
+                    )
+                if not re.fullmatch(r'[A-Za-z0-9+/]*={0,2}', b64):
+                    raise ValueError(
+                        "Attachment base64 data contains invalid characters"
+                    )
+
         validate(json_content)
 
     def _handle_download(self, event, logger):
@@ -83,7 +101,11 @@ def _handle_download(self, event, logger):
         try:
             self._validate_fhir_content(content)
         except Exception as e:
-            logger.error("FHIR content is invalid", error=str(e))
+            logger.error(
+                "FHIR content is invalid",
+                error=str(e),
+                mesh_message_id=data.meshMessageId
+            )
             self._publish_message_invalid_event(incoming_event=event, failure_code='DL_CLIV_005')
             message.acknowledge()
             logger.info("Acknowledged message")

lambdas/mesh-download/requirements.txt

@@ -3,7 +3,7 @@ structlog>=21.5.0
 pydantic>=2.0.0
 boto3>=1.28.62
 pyopenssl>=24.2.1
-nhs-notify-digital-letters-onboarding @ git+https://github.com/NHSDigital/nhs-notify-digital-letters-onboarding@0.1.0
+nhs-notify-digital-letters-onboarding @ git+https://github.com/NHSDigital/nhs-notify-digital-letters-onboarding@0.2.0
 -e ../../src/digital-letters-events
 -e ../../utils/py-mock-mesh
 -e ../../utils/py-utils

lambdas/pdm-uploader-lambda/src/__tests__/app/upload-to-pdm.test.ts

@@ -95,6 +95,7 @@ describe('UploadToPdm', () => {
           message: error.message,
           name: error.name,
         }),
+        messageReference: mockEvent.data.messageReference,
       });
       expect(mockPdmClient.createDocumentReference).not.toHaveBeenCalled();
     });
@@ -113,6 +114,7 @@ describe('UploadToPdm', () => {
         err: expect.objectContaining({
           message: error.message,
         }),
+        messageReference: mockEvent.data.messageReference,
       });
       expect(mockPdmClient.createDocumentReference).not.toHaveBeenCalled();
     });
@@ -131,6 +133,7 @@ describe('UploadToPdm', () => {
           message: error.message,
           name: error.name,
         }),
+        messageReference: mockEvent.data.messageReference,
       });
     });
   });

lambdas/pdm-uploader-lambda/src/app/upload-to-pdm.ts

@@ -35,6 +35,7 @@ export class UploadToPdm {
     } catch (error) {
       this.logger.error({
         description: 'Error sending request to PDM',
+        messageReference: event.data.messageReference,
         err:
           error instanceof Error
             ? { message: error.message, name: error.name, stack: error.stack }

utils/utils/src/__tests__/pdm-client/pdm-client.test.ts

@@ -173,7 +173,11 @@ describe('PdmClient', () => {
       expect(mockLogger.error).toHaveBeenCalledWith({
         description: 'Failed sending PDM request',
         requestId: mockRequestId,
-        err: mockError,
+        err: expect.objectContaining({
+          message: mockError.message,
+          name: mockError.name,
+          stack: mockError.stack,
+        }),
       });
     });
 

utils/utils/src/pdm-client/pdm-client.ts

@@ -82,7 +82,10 @@ export class PdmClient implements IPdmClient {
       this.logger.error({
         description: 'Failed sending PDM request',
         requestId,
-        err: error,
+        err:
+          error instanceof Error
+            ? { message: error.message, name: error.name, stack: error.stack }
+            : error,
       });
 
       throw error;