Fix: [AEA-0000] - Add an upper time limit for the lambda runtime (#1846)

## Summary

- Routine Change

### Details

Lambdas have a timeout - heavy traffic was causing the lambdas to
continually process the queue, and exceed that timeout as they never
observed an empty one. This just adds a time limit to how long a lambda
can be processing the queue before it gives up and yields. Note that the
old behaviour works, but throws a timeout error that we don't want to
see.

Author: wildjames

SAMtemplates/alarms/main.yaml

@@ -18,6 +18,9 @@ Parameters:
 
   ConvertRequestToFhirFormatFunctionName:
     Type: String
+
+  NotifyProcessorFunctionName:
+    Type: String
 
   EnableAlerts:
     Type: String
@@ -309,3 +312,40 @@ Resources:
         - !ImportValue lambda-resources:SlackAlertsSnsTopicArn
       OKActions:
         - !ImportValue lambda-resources:SlackAlertsSnsTopicArn
+
+  NotifyProcessorTimeoutsMetricFilter:
+    Type: AWS::Logs::MetricFilter
+    Properties:
+      FilterName: NotifyProcessorTimeouts
+      FilterPattern: !Sub '{ ($.level = "INFO") && ($.function_name = "${NotifyProcessorFunctionName}") && ($.message = %drainAndProcess timed out; exiting before queue is empty%) }'
+      LogGroupName:
+        Fn::ImportValue: !Sub ${StackName}:functions:${NotifyProcessorFunctionName}:LambdaLogGroupName
+      MetricTransformations:
+        - MetricNamespace: LambdaLogFilterMetrics
+          MetricName: TimeoutCount
+          MetricValue: 1
+          Unit: Count
+          Dimensions:
+            - Key: FunctionName
+              Value: $.function_name
+
+  NotifyProcessorTimeoutsAlarm:
+    Type: AWS::CloudWatch::Alarm
+    Properties:
+      AlarmDescription: Count of NotifyProcessor invocations that timed out
+      AlarmName: !Sub ${StackName}_NotifyProcessor_Timeouts
+      Namespace: LambdaLogFilterMetrics
+      MetricName: TimeoutCount
+      Period: 60 # seconds
+      EvaluationPeriods: 1
+      Statistic: Sum
+      ComparisonOperator: GreaterThanOrEqualToThreshold
+      Threshold: 1
+      TreatMissingData: notBreaching
+      ActionsEnabled: !Ref EnableAlerts
+      AlarmActions:
+        - !ImportValue lambda-resources:SlackAlertsSnsTopicArn
+      InsufficientDataActions:
+        - !ImportValue lambda-resources:SlackAlertsSnsTopicArn
+      OKActions:
+        - !ImportValue lambda-resources:SlackAlertsSnsTopicArn

SAMtemplates/functions/main.yaml

@@ -392,6 +392,7 @@ Resources:
     Type: AWS::Serverless::Function
     Properties:
       FunctionName: !Sub ${StackName}-NotifyProcessor
+      Timeout: 900 # 15 minutes
       CodeUri: ../../packages/
       Handler: nhsNotifyLambda.handler
       Role: !GetAtt NotifyProcessorResources.Outputs.LambdaRoleArn

SAMtemplates/main_template.yaml

@@ -193,5 +193,6 @@ Resources:
         UpdatePrescriptionStatusFunctionName: !GetAtt Functions.Outputs.UpdatePrescriptionStatusFunctionName
         PrescriptionStatusUpdatesTableName: !GetAtt Tables.Outputs.PrescriptionStatusUpdatesTableName
         ConvertRequestToFhirFormatFunctionName: !GetAtt Functions.Outputs.ConvertRequestToFhirFormatFunctionName
+        NotifyProcessorFunctionName: !GetAtt Functions.Outputs.NotifyProcessorFunctionName
         DynamoDBUtilizationPercentageThreshold: !Ref DynamoDBUtilizationPercentageThreshold
         EnableAlerts: !Ref EnableAlerts

packages/nhsNotifyLambda/src/nhsNotifyLambda.ts

@@ -20,6 +20,8 @@ import {
 
 const logger = new Logger({serviceName: "nhsNotify"})
 
+const MAX_QUEUE_RUNTIME = 14*60*1000 // 14 minutes, to avoid Lambda timeout issues (timeout is 15 minutes)
+
 /**
  * Process a single batch of SQS messages: filter, notify, persist, and clean up.
  */
@@ -28,7 +30,6 @@ async function processBatch(
   routingId: string
 ): Promise<void> {
   if (messages.length === 0) {
-    console.log("No messages to process")
     logger.info("No messages to process")
     return
   }
@@ -80,14 +81,22 @@ function logSuppression(suppressedCount: number, total: number): void {
 }
 
 /**
- * Drain the queue until empty, processing each batch.
+ * Drain the queue until empty or the MAX_QUEUE_RUNTIME has passed, processing each batch.
  */
 async function drainAndProcess(routingId: string): Promise<void> {
+  const start = Date.now()
   let empty = false
   while (!empty) {
+    if (Date.now() - start >= MAX_QUEUE_RUNTIME) {
+      logger.warn("drainAndProcess timed out; exiting before queue is empty",
+        {maxRuntimeMilliseconds: MAX_QUEUE_RUNTIME}
+      )
+      break
+    }
+
     const {messages, isEmpty} = await drainQueue(logger, 100)
     empty = isEmpty
-    console.log(messages)
+
     await processBatch(messages, routingId)
   }
 }

packages/nhsNotifyLambda/tests/testNhsNotifyLambda.test.ts

@@ -39,13 +39,15 @@ jest.unstable_mockModule(
 
 const mockInfo = jest.fn()
 const mockError = jest.fn()
+const mockWarn = jest.fn()
 jest.unstable_mockModule(
   "@aws-lambda-powertools/logger",
   async () => ({
     __esModule: true,
     Logger: jest.fn().mockImplementation(() => ({
       info: mockInfo,
-      error: mockError
+      error: mockError,
+      warn: mockWarn
     }))
   })
 )
@@ -282,4 +284,35 @@ describe("Unit test for NHS Notify lambda handler", () => {
         {suppressedCount: 1, totalFetched: 1}
       )
   })
+
+  it("Stops draining after 14 minutes", async () => {
+    const msg = constructPSUDataItemMessage()
+
+    // call returns a non‐empty batch so the loop should continue
+    mockDrainQueue.mockImplementation(() =>
+      Promise.resolve({messages: [msg], isEmpty: false})
+    )
+
+    const nowSpy = jest.spyOn(Date, "now")
+      .mockImplementationOnce(() => 0) // start time
+      .mockImplementationOnce(() => (14 * 60 * 1000) + 1)
+
+    // Happy‐path for everything else
+    mockCheckCooldownForUpdate.mockReturnValueOnce(Promise.resolve(true))
+    mockMakeBatchNotifyRequest.mockReturnValueOnce(Promise.resolve([
+      {...msg, success: true, notifyMessageId: "m1"}
+    ]))
+    mockAddPrescriptionMessagesToNotificationStateStore.mockReturnValueOnce(() => Promise.resolve())
+    mockRemoveSQSMessages.mockReturnValueOnce(() => Promise.resolve())
+
+    await expect(lambdaHandler(mockEventBridgeEvent)).resolves.not.toThrow()
+
+    // Calls a warning
+    expect(mockWarn).toHaveBeenCalledTimes(1)
+
+    // Since the timer advances before the promise resolves, this never gets called
+    expect(mockDrainQueue).toHaveBeenCalledTimes(0)
+
+    nowSpy.mockRestore()
+  })
 })