#141 Add telegram rate limiting

vityaman-edu · Jun 14, 2024 · cb66333 · cb66333
1 parent 8c533f9
commit cb66333
Show file tree

Hide file tree

Showing 4 changed files with 51 additions and 20 deletions.
diff --git a/.../src/main/kotlin/ru/vityaman/lms/botalka/app/spring/event/homework/SpringTelegramActor.kt b/.../src/main/kotlin/ru/vityaman/lms/botalka/app/spring/event/homework/SpringTelegramActor.kt
@@ -2,14 +2,18 @@ package ru.vityaman.lms.botalka.app.spring.event.homework
 
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.stereotype.Component
 import ru.vityaman.lms.botalka.commons.Consumer
 import ru.vityaman.lms.botalka.core.event.EventConsumingActor
 import ru.vityaman.lms.botalka.core.event.EventSource
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramException
 import ru.vityaman.lms.botalka.core.logging.Slf4jLog
 import ru.vityaman.lms.botalka.core.model.Homework
+import kotlin.random.Random
+import kotlin.time.Duration.Companion.seconds
 
 @Component
 class SpringTelegramActor(
@@ -26,13 +30,39 @@ class SpringTelegramActor(
         mailbox = events,
         consumer = consumer,
         callbacks = EventConsumingActor.Callbacks(
-            onError = { log.warn("Failed to consume: ${it.message}") },
+            onStart = {
+                log.info("Starting...")
+            },
+            onSuccess = {
+                log.info("Successfully sent homework with id ${it.id}")
+                waitSeconds(RELAX_DURATION)
+            },
+            onError = {
+                log.warn("Failed: ${it.message}")
+                if (it !is TelegramException) {
+                    throw it
+                }
+                waitSeconds(RETRY_DURATION)
+            },
         ),
     )
 
     init {
-        scope.launch {
-            logic.run()
-        }
+        scope.launch { logic.run() }
+    }
+
+    suspend fun waitSeconds(range: IntRange) {
+        val duration = Random.nextInt(range.first, range.last).seconds
+        log.warn("Taking a $duration delay...")
+        delay(duration)
+    }
+
+    companion object {
+        val RETRY_DURATION = 30..60
+
+        private const val INSTANCES = 2
+        private const val MESSAGES_PER_MINUTE = 20
+        private const val RELAX_MIN = 60 / MESSAGES_PER_MINUTE * INSTANCES
+        val RELAX_DURATION = RELAX_MIN..RELAX_MIN + 5
     }
 }
diff --git a/botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/event/EventConsumingActor.kt b/botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/event/EventConsumingActor.kt
@@ -1,6 +1,5 @@
 package ru.vityaman.lms.botalka.core.event
 
-import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.onEach
 import ru.vityaman.lms.botalka.commons.Consumer
 import ru.vityaman.lms.botalka.commons.Runnable
@@ -10,23 +9,27 @@ class EventConsumingActor<E>(
     private val consumer: Consumer<E>,
     private val callbacks: Callbacks<E> = Callbacks(),
 ) : Runnable {
+    @Suppress("TooGenericExceptionCaught")
     override suspend fun run() {
-        callbacks.onStart()
-        mailbox.events()
-            .onEach { callbacks.onNext(it.payload) }
-            .onEach { consumer.accept(it.payload) }
-            .onEach { callbacks.onSuccess(it.payload) }
-            .catch { callbacks.onError(it) }
-            .onEach { it.acknowledge() }
-            .collect {}
-        callbacks.onFinish()
+        while (true) {
+            callbacks.onStart()
+            try {
+                mailbox.events()
+                    .onEach { callbacks.onNext(it.payload) }
+                    .onEach { consumer.accept(it.payload) }
+                    .onEach { callbacks.onSuccess(it.payload) }
+                    .onEach { it.acknowledge() }
+                    .collect {}
+            } catch (exception: Exception) {
+                callbacks.onError(exception)
+            }
+        }
     }
 
     data class Callbacks<V>(
         val onStart: suspend () -> Unit = {},
         val onNext: suspend (V) -> Unit = {},
         val onSuccess: suspend (V) -> Unit = {},
-        val onError: suspend (Throwable) -> Unit = {},
-        val onFinish: suspend () -> Unit = {},
+        val onError: suspend (Exception) -> Unit = {},
     )
 }
diff --git a/botalka/src/main/resources/application.yml b/botalka/src/main/resources/application.yml
@@ -74,6 +74,4 @@ external:
 task:
   scheduled:
     publication:
-      precision-seconds: 60
-    notification:
-      retry-after-seconds: 60
+      precision-seconds: 30
diff --git a/infra/fuzzing/custom/homeworks.sh b/infra/fuzzing/custom/homeworks.sh
@@ -8,7 +8,7 @@ MOMENT="$2"
 echo "[fuzz] Got access token: '$TOKEN'"
 echo "[fuzz] Got moment: '$MOMENT'"
 
-for _ in $(seq 1 128); do
+for _ in $(seq 1 64); do
     curl -X 'POST' \
       'http://localhost:8080/api/v1/homework' \
       -H 'accept: application/json' \