Introduce timeouts in the UpdateProcessingPipeline

Author: madhead

.deploy/lambda/lib/JProfByBotStack.ts

@@ -131,14 +131,16 @@ export class JProfByBotStack extends cdk.Stack {
             compatibleRuntimes: [lambda.Runtime.JAVA_11],
         });
 
+        const lambdaWebhookTimeout = cdk.Duration.seconds(30);
         const lambdaWebhook = new lambda.Function(this, 'jprof-by-bot-lambda-webhook', {
             functionName: 'jprof-by-bot-lambda-webhook',
             runtime: lambda.Runtime.JAVA_11,
             layers: [
                 layerLibGL,
                 layerLibfontconfig,
             ],
-            timeout: cdk.Duration.seconds(30),
+            timeout: lambdaWebhookTimeout,
+            retryAttempts: 0,
             memorySize: 512,
             code: lambda.Code.fromAsset('../../launchers/lambda/build/libs/jprof_by_bot-launchers-lambda-all.jar'),
             handler: 'by.jprof.telegram.bot.launchers.lambda.JProf',
@@ -157,13 +159,15 @@ export class JProfByBotStack extends cdk.Stack {
                 'STATE_MACHINE_UNPINS': stateMachineUnpin.stateMachineArn,
                 'TOKEN_TELEGRAM_BOT': props.telegramToken,
                 'TOKEN_YOUTUBE_API': props.youtubeToken,
+                'TIMEOUT': lambdaWebhookTimeout.toMilliseconds().toString(),
             },
         });
 
         const lambdaDailyUrbanDictionary = new lambda.Function(this, 'jprof-by-bot-lambda-daily-urban-dictionary', {
             functionName: 'jprof-by-bot-lambda-daily-urban-dictionary',
             runtime: lambda.Runtime.JAVA_11,
             timeout: cdk.Duration.seconds(30),
+            retryAttempts: 0,
             memorySize: 512,
             code: lambda.Code.fromAsset('../../english/urban-dictionary-daily/build/libs/jprof_by_bot-english-urban-dictionary-daily-all.jar'),
             handler: 'by.jprof.telegram.bot.english.urban_dictionary_daily.Handler',

core/src/main/kotlin/by/jprof/telegram/bot/core/UpdateProcessingPipeline.kt

@@ -6,20 +6,29 @@ import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.supervisorScope
+import kotlinx.coroutines.withTimeoutOrNull
 import org.apache.logging.log4j.LogManager
 
 class UpdateProcessingPipeline(
-    private val processors: List<UpdateProcessor>
+    private val processors: List<UpdateProcessor>,
+    private val timeout: Long,
 ) {
     companion object {
         private val logger = LogManager.getLogger(UpdateProcessingPipeline::class.java)!!
     }
 
     fun process(update: Update) = runBlocking {
-        supervisorScope {
-            processors
-                .map { launch(exceptionHandler(it)) { it.process(update) } }
-                .joinAll()
+        withTimeoutOrNull(timeout) {
+            supervisorScope {
+                processors
+                    .map {
+                        launch(exceptionHandler(it)) {
+                            logger.debug("Processing update with ${it::class.simpleName}")
+                            it.process(update)
+                        }
+                    }
+                    .joinAll()
+            }
         }
     }
 

core/src/test/kotlin/by/jprof/telegram/bot/core/UpdateProcessingPipelineTest.kt

@@ -2,47 +2,80 @@ package by.jprof.telegram.bot.core
 
 import dev.inmo.tgbotapi.types.update.abstracts.UnknownUpdate
 import dev.inmo.tgbotapi.types.update.abstracts.Update
+import java.time.Duration
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.random.Random
 import kotlinx.coroutines.delay
 import kotlinx.serialization.json.JsonNull
-import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTimeout
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
-import java.time.Duration
-import java.util.concurrent.atomic.AtomicBoolean
 
 internal class UpdateProcessingPipelineTest {
     @Test
     fun process() {
         val completionFlags = (1..5).map { AtomicBoolean(false) }
         val sut = UpdateProcessingPipeline(
-            completionFlags.mapIndexed { index, atomicBoolean ->
+            processors = completionFlags.mapIndexed { index, atomicBoolean ->
                 Delay((index + 1) * 1000L, atomicBoolean)
-            }
+            },
+            timeout = 10_000,
         )
 
-        Assertions.assertTimeout(Duration.ofMillis(6000)) {
+        assertTimeout(Duration.ofMillis(6000)) {
             sut.process(UnknownUpdate(1L, "", JsonNull))
         }
-        Assertions.assertTrue(completionFlags.all { it.get() })
+        assertTrue(completionFlags.all { it.get() })
     }
 
     @Test
     fun processWithBroken() {
         val completionFlags = (1..5).map { AtomicBoolean(false) }
         val sut = UpdateProcessingPipeline(
-            completionFlags.mapIndexed { index, atomicBoolean ->
+            processors = completionFlags.mapIndexed { index, atomicBoolean ->
                 when (index % 2) {
                     0 -> Delay((index + 1) * 1000L, atomicBoolean)
                     else -> Fail()
                 }
-            }
+            },
+            timeout = 10_000,
         )
 
-        Assertions.assertTimeout(Duration.ofMillis(6000)) {
+        assertTimeout(Duration.ofMillis(6000)) {
             sut.process(UnknownUpdate(1L, "", JsonNull))
         }
         completionFlags.forEachIndexed { index, atomicBoolean ->
             if (index % 2 == 0) {
-                Assertions.assertTrue(atomicBoolean.get())
+                assertTrue(atomicBoolean.get())
+            }
+        }
+    }
+
+    @Test
+    fun processWithTimeout() {
+        val completionFlags = (1..6).map { AtomicBoolean(false) }
+        val sut = UpdateProcessingPipeline(
+            processors = completionFlags.mapIndexed { index, atomicBoolean ->
+                when (index % 3) {
+                    0 -> Delay(Long.MAX_VALUE, atomicBoolean)
+                    1 -> Delay(Random.nextLong(0, 100), atomicBoolean)
+                    else -> Fail()
+                }
+            },
+            timeout = 1000,
+        )
+
+        assertTimeout(Duration.ofMillis(2000)) {
+            sut.process(UnknownUpdate(1L, "", JsonNull))
+        }
+        completionFlags.forEachIndexed { index, atomicBoolean ->
+            if (index % 3 == 1) {
+                assertTrue(atomicBoolean.get())
+            }
+            when (index % 3) {
+                1 -> assertTrue(atomicBoolean.get())
+                else -> assertFalse(atomicBoolean.get())
             }
         }
     }

launchers/lambda/src/main/kotlin/by/jprof/telegram/bot/launchers/lambda/config/env.kt

@@ -16,6 +16,7 @@ const val TABLE_TIMEZONES = "TABLE_TIMEZONES"
 const val TABLE_LANGUAGE_ROOMS = "TABLE_LANGUAGE_ROOMS"
 const val TABLE_URBAN_WORDS_OF_THE_DAY = "TABLE_URBAN_WORDS_OF_THE_DAY"
 const val STATE_MACHINE_UNPINS = "STATE_MACHINE_UNPINS"
+const val TIMEOUT = "TIMEOUT"
 
 val envModule = module {
     listOf(
@@ -37,4 +38,8 @@ val envModule = module {
             System.getenv(variable)!!
         }
     }
+
+    single(named(TIMEOUT)) {
+        System.getenv(TIMEOUT)!!.toLong()
+    }
 }

launchers/lambda/src/main/kotlin/by/jprof/telegram/bot/launchers/lambda/config/pipeline.kt

@@ -28,7 +28,10 @@ import org.koin.dsl.module
 @PreviewFeature
 val pipelineModule = module {
     single {
-        UpdateProcessingPipeline(getAll())
+        UpdateProcessingPipeline(
+            processors = getAll(),
+            timeout = get<Long>(named(TIMEOUT)) - 1000
+        )
     }
 
     single<UpdateProcessor>(named("JEPUpdateProcessor")) {