Live activities clean up - Fix Dynamo AttemptCalloutException (#1805)

* Increase Dynamo timeout config for cleaner lambda

* Update cdk lambda execution timeout.

* tidy

Author: vlbee

cdk/lib/__snapshots__/liveactivities.test.ts.snap

@@ -853,7 +853,7 @@ exports[`The LiveActivities stack matches the snapshot for CODE 1`] = `
             "Value": "CODE",
           },
         ],
-        "Timeout": 180,
+        "Timeout": 660,
       },
       "Type": "AWS::Lambda::Function",
     },

cdk/lib/liveactivities.ts

@@ -198,7 +198,7 @@ export class LiveActivities extends GuStack {
 				fileName: `${app}.jar`,
 				runtime: Runtime.JAVA_11,
 				memorySize: 1024,
-				timeout: Duration.minutes(3),
+				timeout: Duration.minutes(11),
 				onFailure: new SqsDestination(channelCleanerDlq),
 				retryAttempts: 2,
 				environment: {

liveactivities/src/main/scala/com/gu/liveactivities/ChannelCleanUpLambda.scala

@@ -12,14 +12,16 @@ import scala.util.{Failure, Success, Try}
 object ChannelCleanUpLambda extends Lambda with Logging {
 
   private val channelApiClient = new ChannelApiClient(authentication, config.bundleId, config.sendingToProdServer)
-  private val cleanUpService = new ChannelCleanUpService(repository, channelApiClient)
+  private val dynamoRepository = getRepositoryWithCustomTimeouts(10, 40)
+  private val cleanUpService = new ChannelCleanUpService(dynamoRepository, channelApiClient)
 
   // we need a scheduledEvent input here so we have some metadata to capture in the DLQ if there is a failure.
   def handleRequest(input: ScheduledEvent, context: Context): Unit = {
     logger.info("Channel clean up lambda triggered")
 
     val future = for {
       deletedChannels <- cleanUpService.deleteChannelsForEndedBroadcasts()
+      _ = logger.info(s"${deletedChannels.size} channels deleted:  [${deletedChannels.map(_.id).mkString(", ")}]")
       // orphanChannels <- deleteOrphanChannelsInAPNS() // channels that exist in APNS but corresponding mapping is missing in Dynamo
       // staleMappings <- markInactiveStaleMappings // mapping still "LIVE" that missed an end activity update and need to be deleted - Mapi may return channel id otherwise
       // orphanMappings <- markInactiveOrphanMappings // mapping still "active" but channel has been deleted.
@@ -31,7 +33,7 @@ object ChannelCleanUpLambda extends Lambda with Logging {
       }
     } yield cleanupSummary
 
-    Try(Await.result(future, 160.seconds)) match {
+    Try(Await.result(future, 600.seconds)) match {
       case Success(_) => {
         logger.info("Channel clean up completed successfully")
       }

liveactivities/src/main/scala/com/gu/liveactivities/Lambda.scala

@@ -24,23 +24,28 @@ trait Lambda extends Logging{
     DefaultCredentialsProvider.builder.build(),
   )
 
-  val dynamoDbClient =
+  private def buildDbClient(attemptTimeout: Int = 2, timeout: Int = 10): DynamoDbAsyncClient = {
     DynamoDbAsyncClient
       .builder()
       .credentialsProvider(credentialsv2)
       .region(Region.of(config.region))
       .overrideConfiguration(
         ClientOverrideConfiguration.builder()
-          .apiCallAttemptTimeout(Duration.ofSeconds(2)) // limit for one single try
-          .apiCallTimeout(Duration.ofSeconds(10)) // limit for the entire operation
+          .apiCallAttemptTimeout(Duration.ofSeconds(attemptTimeout)) // limit for one single try
+          .apiCallTimeout(Duration.ofSeconds(timeout)) // limit for the entire operation
           .retryStrategy(DefaultRetryStrategy.standardStrategyBuilder()
             .maxAttempts(3)
             .build())
           .build()
       )
       .build()
+  }
 
-  val repository = new LiveActivityChannelRepository(dynamoDbClient, s"mobile-notifications-liveactivities-${config.stage}")
+  private val dynamoTableName = s"mobile-notifications-liveactivities-${config.stage}"
+  val repository = new LiveActivityChannelRepository(buildDbClient(), dynamoTableName)
+  def getRepositoryWithCustomTimeouts(attemptTimeout: Int, timeout: Int): LiveActivityChannelRepository = {
+    new LiveActivityChannelRepository(buildDbClient(attemptTimeout, timeout), dynamoTableName)
+  }
 
   private val eventBusName = s"liveactivities-eventbus-${config.stage}"
   lazy val liveActivityPusher = new LiveActivityPusher(eventBusName, logger)

liveactivities/src/main/scala/com/gu/liveactivities/service/ChannelCleanUpService.scala

@@ -23,7 +23,7 @@ class ChannelCleanUpService(
         hasLastEvent = true,
       )
       _ = logger.info(s"Channels identified for possible deletion: ${channelsForPossibleDeletion.size}")
-      // todo these sequence failures will not currently throw
+      // todo this sequence failures will not currently throw
       _ <- Future.sequence(
         channelsForPossibleDeletion.map { mapping =>
         {
@@ -36,14 +36,19 @@ class ChannelCleanUpService(
               .closeChannel(mapping.channelId)
               .flatMap { _ =>
                 val f = repository.updateMappingActiveChannel(mapping.id, isActive = false)
-                channelsDeleted ++= List(mapping)
-                logger.info(s"Channel successfully marked inactive with channel ID: ${mapping.channelId} for match ID ${mapping.id}")
+                  .map { result =>
+                    channelsDeleted ++= List(mapping)
+                    logger.info(s"Channel successfully marked inactive with channel ID: ${mapping.channelId} for match ID ${mapping.id}")
+                    result
+                  }
                 f
               }
               .recover { case exception =>
-                // todo Handle exception better and throw when needed. Some errorw will be handled by rest of the Cleanup tasks.
+                // todo handle exception better and throw when needed. Some errors will be handled by rest of the Cleanup tasks.
                 logger.error(s"Failed to delete channel with ID ${mapping.channelId} for match ID ${mapping.id} - ${exception.getMessage}")
               }
+
+
           }
         }
         },