From 1645c334f45682bda19a715b72e54f7d53b41b6f Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Fri, 12 Sep 2025 18:24:07 +0300 Subject: [PATCH 01/23] added new jvmTarget --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 0e7c0ad2d..4c1e6f8f3 100644 --- a/pom.xml +++ b/pom.xml @@ -175,7 +175,7 @@ spring - 1.8 + 17 From c4d3fec78c9034c0ea763b5e50d87652932b8238 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Wed, 17 Sep 2025 19:15:29 +0300 Subject: [PATCH 02/23] fixed --- .gitignore | 2 +- src/main/resources/application.properties | 2 +- test-local-run.http | 2 +- test-on-prem-run.http | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index 259113f73..9212609a9 100644 --- a/.gitignore +++ b/.gitignore @@ -18,7 +18,7 @@ target/ *.iws *.iml *.ipr -./http-client.env.json +http-client.env.json ### NetBeans ### /nbproject/private/ diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 33d51a58b..fb4ea1302 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,5 +1,5 @@ server.address=0.0.0.0 -server.port=8081 +server.port=${SERVER_PORT:8081} server.http2.enabled=true spring.main.allow-bean-definition-overriding=true diff --git a/test-local-run.http b/test-local-run.http index dc8dbeedf..7be0e4f73 100644 --- a/test-local-run.http +++ b/test-local-run.http @@ -12,4 +12,4 @@ Content-Type: application/json ### Stop running test to save time and resources # @timeout 120 -POST http://localhost:4321/test/stop/"{{serviceName}}" \ No newline at end of file +POST http://localhost:1234/test/stop/{{serviceName}} \ No newline at end of file diff --git a/test-on-prem-run.http b/test-on-prem-run.http index dd5765987..cf01bfbb5 100644 --- a/test-on-prem-run.http +++ b/test-on-prem-run.http @@ -16,4 +16,4 @@ Content-Type: application/json ### Stop running test to save credits # @timeout 120 -POST http://77.234.215.138:31234/test/stop/"{{serviceName}}" \ No newline at end of file +POST http://77.234.215.138:31234/test/stop/{{serviceName}} From 484a3db4dae84959dbc786a80dabd48468a62fb6 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Thu, 18 Sep 2025 19:24:39 +0300 Subject: [PATCH 03/23] added rateLimit --- .../payments/logic/PaymentExternalServiceImpl.kt | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 5cb12106a..85fab7a89 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -6,6 +6,8 @@ import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody import org.slf4j.LoggerFactory +import ru.quipy.common.utils.SlidingWindowRateLimiter +import ru.quipy.common.utils.rateLimiter.CustomRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.net.SocketTimeoutException @@ -36,6 +38,11 @@ class PaymentExternalSystemAdapterImpl( private val client = OkHttpClient.Builder().build() + private val rateLimiter = SlidingWindowRateLimiter( + rateLimitPerSec.toLong(), + Duration.ofSeconds(1) + ) + override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { logger.warn("[$accountName] Submitting payment request for payment $paymentId") @@ -47,6 +54,8 @@ class PaymentExternalSystemAdapterImpl( it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } + rateLimiter.tickBlocking() + logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") try { @@ -60,7 +69,7 @@ class PaymentExternalSystemAdapterImpl( mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) } catch (e: Exception) { logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(),false, e.message) + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) } logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") From 0b8d27f5ce5675ddfc69d1c16c1c1494b68da88c Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Tue, 23 Sep 2025 01:10:50 +0300 Subject: [PATCH 04/23] feat: semaphore has been added to sliding window rate limiter --- .gitignore | 3 + .../common/utils/SlidingWindowRateLimiter.kt | 61 ++++++++++++++- .../ru/quipy/payments/logic/OrderPayer.kt | 2 +- .../logic/PaymentExternalServiceImpl.kt | 75 ++++++++++--------- test-local-run.http | 4 +- 5 files changed, 107 insertions(+), 38 deletions(-) diff --git a/.gitignore b/.gitignore index 9212609a9..440730d57 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,6 @@ build/ ### VS Code ### .vscode/ + +grafana/data/* +prometheus/data/* \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt index 6ff3092ab..1eba22a82 100644 --- a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt +++ b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt @@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory import java.time.Duration import java.util.concurrent.Executors import java.util.concurrent.PriorityBlockingQueue +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -67,4 +68,62 @@ class SlidingWindowRateLimiter( companion object { private val logger: Logger = LoggerFactory.getLogger(SlidingWindowRateLimiter::class.java) } -} \ No newline at end of file +} + +class Semaphore(permits: Int) { + private val lock = ReentrantLock() + private val condition = lock.newCondition() + private var availablePermits = permits + + @Throws(InterruptedException::class) + fun acquire() { + lock.withLock { + while (availablePermits <= 0) { + condition.await() + } + availablePermits-- + } + } + + fun tryAcquire(): Boolean { + return lock.withLock { + if (availablePermits > 0) { + availablePermits-- + true + } else { + false + } + } + } + + @Throws(InterruptedException::class) + fun tryAcquire(timeout: Long, unit: TimeUnit): Boolean { + var remainingNanos = unit.toNanos(timeout) + lock.lock() + try { + while (availablePermits <= 0) { + if (remainingNanos <= 0) { + return false + } + remainingNanos = condition.awaitNanos(remainingNanos) + } + availablePermits-- + return true + } finally { + lock.unlock() + } + } + + fun release() { + lock.withLock { + availablePermits++ + condition.signal() + } + } + + fun availablePermits(): Int { + return lock.withLock { + availablePermits + } + } +} diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index a5909b85b..3f047e3bd 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -46,7 +46,7 @@ class OrderPayer { amount ) } - logger.trace("Payment ${createdEvent.paymentId} for order $orderId created.") + logger.trace("Payment {} for order {} created.", createdEvent.paymentId, orderId) paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) } diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 85fab7a89..396d77b29 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -6,8 +6,8 @@ import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody import org.slf4j.LoggerFactory +import ru.quipy.common.utils.Semaphore import ru.quipy.common.utils.SlidingWindowRateLimiter -import ru.quipy.common.utils.rateLimiter.CustomRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.net.SocketTimeoutException @@ -32,12 +32,13 @@ class PaymentExternalSystemAdapterImpl( private val serviceName = properties.serviceName private val accountName = properties.accountName - private val requestAverageProcessingTime = properties.averageProcessingTime private val rateLimitPerSec = properties.rateLimitPerSec private val parallelRequests = properties.parallelRequests private val client = OkHttpClient.Builder().build() + private val semaphore = Semaphore(parallelRequests) + private val rateLimiter = SlidingWindowRateLimiter( rateLimitPerSec.toLong(), Duration.ofSeconds(1) @@ -45,7 +46,7 @@ class PaymentExternalSystemAdapterImpl( override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { logger.warn("[$accountName] Submitting payment request for payment $paymentId") - + logger.info("rate limiter rate limit per sec: {}", rateLimitPerSec) val transactionId = UUID.randomUUID() // Вне зависимости от исхода оплаты важно отметить что она была отправлена. @@ -54,49 +55,55 @@ class PaymentExternalSystemAdapterImpl( it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - rateLimiter.tickBlocking() + semaphore.acquire() + try { - logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") + rateLimiter.tickBlocking() - try { - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - client.newCall(request).execute().use { response -> - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } + logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + try { + val request = Request.Builder().run { + url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") + post(emptyBody) + }.build() - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) - } - } - } catch (e: Exception) { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) + client.newCall(request).execute().use { response -> + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } + + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + it.logProcessing(body.result, now(), transactionId, reason = body.message) } } + } catch (e: Exception) { + when (e) { + is SocketTimeoutException -> { + logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + } + } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + else -> { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = e.message) + } } } } + } finally { + semaphore.release() } } diff --git a/test-local-run.http b/test-local-run.http index 7be0e4f73..6a74b29d0 100644 --- a/test-local-run.http +++ b/test-local-run.http @@ -5,9 +5,9 @@ Content-Type: application/json { "serviceName": "{{serviceName}}", "token": "{{token}}", - "ratePerSecond": 1, + "ratePerSecond": 2, "testCount": 100, - "processingTimeMillis": 80000 + "processingTimeMillis": 60000 } ### Stop running test to save time and resources From c8b02e74933a3b388f2eee05665f8efcc6e40262 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Tue, 23 Sep 2025 01:44:34 +0300 Subject: [PATCH 05/23] refactor: simplify work with semaphore by using new method with try finay in realisation --- .../ru/quipy/common/utils/SlidingWindowRateLimiter.kt | 11 ++++++++--- .../payments/logic/PaymentExternalServiceImpl.kt | 6 +----- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt index 1eba22a82..4df1490ff 100644 --- a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt +++ b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt @@ -96,6 +96,7 @@ class Semaphore(permits: Int) { } } + // работает не чини, так что потом переписать можно будет на промежуточную acquire @Throws(InterruptedException::class) fun tryAcquire(timeout: Long, unit: TimeUnit): Boolean { var remainingNanos = unit.toNanos(timeout) @@ -121,9 +122,13 @@ class Semaphore(permits: Int) { } } - fun availablePermits(): Int { - return lock.withLock { - availablePermits + fun permitTask (task: () -> T): T { + acquire() + try { + return task() + } finally { + release() } } + } diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 396d77b29..aefc58d7b 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -55,9 +55,7 @@ class PaymentExternalSystemAdapterImpl( it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - semaphore.acquire() - try { - + semaphore.permitTask { rateLimiter.tickBlocking() logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") @@ -102,8 +100,6 @@ class PaymentExternalSystemAdapterImpl( } } } - } finally { - semaphore.release() } } From 108d7dbe2ca7aad1a47d50cec136e3669244a690 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 2 Oct 2025 17:45:19 +0300 Subject: [PATCH 06/23] feat: added new metric that show submition rate and processint speed [lab3] --- .../dashboards/ServicesStatistic.json | 114 ++++++++++++++++++ pom.xml | 4 + .../ru/quipy/apigateway/APIController.kt | 12 +- .../common/utils/SlidingWindowRateLimiter.kt | 63 ---------- .../logic/PaymentExternalServiceImpl.kt | 79 +++++++----- .../payments/logic/PaymentServiceImpl.kt | 11 +- test-local-run.http | 2 +- 7 files changed, 186 insertions(+), 99 deletions(-) diff --git a/grafana/provisioning/dashboards/ServicesStatistic.json b/grafana/provisioning/dashboards/ServicesStatistic.json index 684b97269..702a70b38 100644 --- a/grafana/provisioning/dashboards/ServicesStatistic.json +++ b/grafana/provisioning/dashboards/ServicesStatistic.json @@ -1518,6 +1518,120 @@ "title": "(Clients -> Online shop) http requests processed rate (rps)", "type": "timeseries" }, + { + "datasource": { + "type": "prometheus", + "uid": "PBFA97CFB590B2093" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 0 + }, + "id": 101, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "single", + "sort": "none" + } + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "disableTextWrap": false, + "editorMode": "builder", + "expr": "rate(http_requests_served_total[$__rate_interval])", + "fullMetaSearch": false, + "includeNullMetadata": true, + "legendFormat": "__auto", + "range": true, + "refId": "A", + "useBackend": false + }, + { + "datasource": { + "type": "prometheus", + "uid": "PBFA97CFB590B2093" + }, + "disableTextWrap": false, + "editorMode": "builder", + "expr": "rate(payment_service_sanded_total[$__rate_interval])", + "fullMetaSearch": false, + "hide": false, + "includeNullMetadata": true, + "instant": false, + "legendFormat": "__auto", + "range": true, + "refId": "B", + "useBackend": false + } + ], + "title": "New panel", + "type": "timeseries" + }, { "datasource": { "type": "prometheus", diff --git a/pom.xml b/pom.xml index 4c1e6f8f3..204fcd924 100644 --- a/pom.xml +++ b/pom.xml @@ -126,6 +126,10 @@ io.micrometer micrometer-registry-prometheus + + io.micrometer + micrometer-registry-prometheus-simpleclient + org.jetbrains.kotlin kotlin-test diff --git a/src/main/kotlin/ru/quipy/apigateway/APIController.kt b/src/main/kotlin/ru/quipy/apigateway/APIController.kt index 6f23fa18d..43c01db47 100644 --- a/src/main/kotlin/ru/quipy/apigateway/APIController.kt +++ b/src/main/kotlin/ru/quipy/apigateway/APIController.kt @@ -1,5 +1,7 @@ package ru.quipy.apigateway +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry import org.slf4j.Logger import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -9,10 +11,15 @@ import ru.quipy.payments.logic.OrderPayer import java.util.* @RestController -class APIController { +class APIController(@Autowired val meterRegistry: MeterRegistry) { val logger: Logger = LoggerFactory.getLogger(APIController::class.java) + private val orderCounter: Counter = Counter.builder("http_requests_served") + .description("Total number of served http requests for order payment") + .tag("endpoint", "pay_order") + .register(meterRegistry) + @Autowired private lateinit var orderRepository: OrderRepository @@ -21,6 +28,7 @@ class APIController { @PostMapping("/users") fun createUser(@RequestBody req: CreateUserRequest): User { + orderCounter.increment() return User(UUID.randomUUID(), req.name) } @@ -30,6 +38,7 @@ class APIController { @PostMapping("/orders") fun createOrder(@RequestParam userId: UUID, @RequestParam price: Int): Order { + orderCounter.increment() val order = Order( UUID.randomUUID(), userId, @@ -57,6 +66,7 @@ class APIController { @PostMapping("/orders/{orderId}/payment") fun payOrder(@PathVariable orderId: UUID, @RequestParam deadline: Long): PaymentSubmissionDto { val paymentId = UUID.randomUUID() + orderCounter.increment() val order = orderRepository.findById(orderId)?.let { orderRepository.save(it.copy(status = OrderStatus.PAYMENT_IN_PROGRESS)) it diff --git a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt index 4df1490ff..6a3debf57 100644 --- a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt +++ b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt @@ -69,66 +69,3 @@ class SlidingWindowRateLimiter( private val logger: Logger = LoggerFactory.getLogger(SlidingWindowRateLimiter::class.java) } } - -class Semaphore(permits: Int) { - private val lock = ReentrantLock() - private val condition = lock.newCondition() - private var availablePermits = permits - - @Throws(InterruptedException::class) - fun acquire() { - lock.withLock { - while (availablePermits <= 0) { - condition.await() - } - availablePermits-- - } - } - - fun tryAcquire(): Boolean { - return lock.withLock { - if (availablePermits > 0) { - availablePermits-- - true - } else { - false - } - } - } - - // работает не чини, так что потом переписать можно будет на промежуточную acquire - @Throws(InterruptedException::class) - fun tryAcquire(timeout: Long, unit: TimeUnit): Boolean { - var remainingNanos = unit.toNanos(timeout) - lock.lock() - try { - while (availablePermits <= 0) { - if (remainingNanos <= 0) { - return false - } - remainingNanos = condition.awaitNanos(remainingNanos) - } - availablePermits-- - return true - } finally { - lock.unlock() - } - } - - fun release() { - lock.withLock { - availablePermits++ - condition.signal() - } - } - - fun permitTask (task: () -> T): T { - acquire() - try { - return task() - } finally { - release() - } - } - -} diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index aefc58d7b..501f86f6e 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -6,13 +6,14 @@ import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody import org.slf4j.LoggerFactory -import ru.quipy.common.utils.Semaphore import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.net.SocketTimeoutException import java.time.Duration import java.util.* +import java.util.concurrent.Semaphore +import java.util.concurrent.TimeUnit // Advice: always treat time as a Duration @@ -32,18 +33,27 @@ class PaymentExternalSystemAdapterImpl( private val serviceName = properties.serviceName private val accountName = properties.accountName + private val averageProcessTime = properties.averageProcessingTime private val rateLimitPerSec = properties.rateLimitPerSec private val parallelRequests = properties.parallelRequests private val client = OkHttpClient.Builder().build() - private val semaphore = Semaphore(parallelRequests) + private val semaphore = Semaphore(parallelRequests, true) private val rateLimiter = SlidingWindowRateLimiter( rateLimitPerSec.toLong(), Duration.ofSeconds(1) ) + fun deadlineHandler(paymentId: UUID, transactionId: UUID) { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = "Deadline passed") + } + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId") + return + } + override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { logger.warn("[$accountName] Submitting payment request for payment $paymentId") logger.info("rate limiter rate limit per sec: {}", rateLimitPerSec) @@ -55,48 +65,51 @@ class PaymentExternalSystemAdapterImpl( it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - semaphore.permitTask { + try { +// if (semaphore.tryAcquire(averageProcessTime.toMillis(), TimeUnit.MILLISECONDS)) { +// return deadlineHandler(paymentId, transactionId) +// } + semaphore.acquire() rateLimiter.tickBlocking() logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - try { - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - client.newCall(request).execute().use { response -> - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } + val request = Request.Builder().run { + url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") + post(emptyBody) + }.build() + + client.newCall(request).execute().use { response -> + semaphore.release() + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + paymentESService.update(paymentId) { + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } + } + } catch (e: Exception) { + when (e) { + is SocketTimeoutException -> { + logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + it.logProcessing(false, now(), transactionId, reason = "Request timeout.") } } - } catch (e: Exception) { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") - } - } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + else -> { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) - } + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = e.message) } } } diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt index 1c24e5a72..4b1f2d9e0 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt @@ -1,5 +1,7 @@ package ru.quipy.payments.logic +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -15,8 +17,14 @@ import kotlin.concurrent.withLock @Service class PaymentSystemImpl( - private val paymentAccounts: List + private val paymentAccounts: List, + @Autowired val meterRegistry: MeterRegistry, ) : PaymentService { + private val ansCounter: Counter = Counter.builder("payment_service_sanded") + .description("Total number of sanded requests") + .tag("service", "payment_requests") + .register(meterRegistry) + companion object { val logger = LoggerFactory.getLogger(PaymentSystemImpl::class.java) } @@ -24,6 +32,7 @@ class PaymentSystemImpl( override fun submitPaymentRequest(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { for (account in paymentAccounts) { account.performPaymentAsync(paymentId, amount, paymentStartedAt, deadline) + ansCounter.increment() } } } \ No newline at end of file diff --git a/test-local-run.http b/test-local-run.http index 6a74b29d0..aedf30118 100644 --- a/test-local-run.http +++ b/test-local-run.http @@ -6,7 +6,7 @@ Content-Type: application/json "serviceName": "{{serviceName}}", "token": "{{token}}", "ratePerSecond": 2, - "testCount": 100, + "testCount": 500, "processingTimeMillis": 60000 } From f323d4f46efb5564cd5c2cd16e8848ebf565df84 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 16 Oct 2025 18:20:45 +0300 Subject: [PATCH 07/23] new metric --- .../dashboards/ServicesStatistic.json | 146 +----------------- 1 file changed, 4 insertions(+), 142 deletions(-) diff --git a/grafana/provisioning/dashboards/ServicesStatistic.json b/grafana/provisioning/dashboards/ServicesStatistic.json index 702a70b38..9de50eb73 100644 --- a/grafana/provisioning/dashboards/ServicesStatistic.json +++ b/grafana/provisioning/dashboards/ServicesStatistic.json @@ -1582,7 +1582,7 @@ "h": 8, "w": 12, "x": 0, - "y": 0 + "y": 4 }, "id": 101, "options": { @@ -1603,7 +1603,7 @@ { "disableTextWrap": false, "editorMode": "builder", - "expr": "rate(http_requests_served_total[$__rate_interval])", + "expr": "rate(http_requests_served_total{endpoint=~\"pay_order\"}[$__rate_interval])", "fullMetaSearch": false, "includeNullMetadata": true, "legendFormat": "__auto", @@ -1618,7 +1618,7 @@ }, "disableTextWrap": false, "editorMode": "builder", - "expr": "rate(payment_service_sanded_total[$__rate_interval])", + "expr": "rate(payment_service_sanded_total{service=~\"$service\"}[$__rate_interval])", "fullMetaSearch": false, "hide": false, "includeNullMetadata": true, @@ -1629,145 +1629,7 @@ "useBackend": false } ], - "title": "New panel", - "type": "timeseries" - }, - { - "datasource": { - "type": "prometheus", - "uid": "PBFA97CFB590B2093" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "custom": { - "axisBorderShow": false, - "axisCenteredZero": false, - "axisColorMode": "text", - "axisLabel": "", - "axisPlacement": "auto", - "barAlignment": 0, - "barWidthFactor": 0.6, - "drawStyle": "line", - "fillOpacity": 0, - "gradientMode": "none", - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - }, - "insertNulls": false, - "lineInterpolation": "smooth", - "lineStyle": { - "fill": "solid" - }, - "lineWidth": 1, - "pointSize": 5, - "scaleDistribution": { - "type": "linear" - }, - "showPoints": "never", - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "none" - }, - "thresholdsStyle": { - "mode": "off" - } - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - }, - { - "color": "red", - "value": 80 - } - ] - }, - "unit": "ms" - }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "success" - }, - "properties": [ - { - "id": "color", - "value": { - "fixedColor": "green", - "mode": "fixed" - } - } - ] - }, - { - "matcher": { - "id": "byName", - "options": "fail" - }, - "properties": [ - { - "id": "color", - "value": { - "fixedColor": "red", - "mode": "fixed" - } - } - ] - } - ] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 4 - }, - "id": 14, - "options": { - "legend": { - "calcs": [ - "max" - ], - "displayMode": "table", - "placement": "right", - "showLegend": true, - "sortBy": "Max", - "sortDesc": true - }, - "tooltip": { - "mode": "single", - "sort": "none" - } - }, - "pluginVersion": "11.4.0", - "targets": [ - { - "datasource": { - "type": "prometheus", - "uid": "PBFA97CFB590B2093" - }, - "editorMode": "code", - "exemplar": true, - "expr": "rate(http_external_duration_sum{service=~\"$service\"}[1m]) / rate(http_external_duration_count{service=~\"$service\"}[1m])", - "hide": false, - "interval": "", - "legendFormat": "{{method}} - {{result}}", - "range": true, - "refId": "C" - } - ], - "title": "(Clients -> Online shop) http requests duration. Average", + "title": "sending rate & processing speed", "type": "timeseries" }, { From 467884ba42d9e3b54cb787a6fee259b2bbc3aa85 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Fri, 17 Oct 2025 00:35:16 +0300 Subject: [PATCH 08/23] feat: new symaphore with inner control of buffer range --- http-client.env.json | 16 +++ .../ru/quipy/apigateway/APIController.kt | 19 ++- .../common/utils/InstantRateLimitSemaphore.kt | 46 ++++++ .../quipy/common/utils/RateLimitSemaphore.kt | 44 ++++++ .../common/utils/SlidingWindowRateLimiter.kt | 15 +- .../common/utils/TooManyRequestsException.kt | 3 + .../payments/config/PaymentAccountsConfig.kt | 17 ++- .../ru/quipy/payments/logic/OrderPayer.kt | 89 ++++++++++-- .../logic/PaymentExternalServiceImpl.kt | 131 +++++++++++------- test-local-run.http | 6 +- 10 files changed, 309 insertions(+), 77 deletions(-) create mode 100644 http-client.env.json create mode 100644 src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt create mode 100644 src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt create mode 100644 src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt diff --git a/http-client.env.json b/http-client.env.json new file mode 100644 index 000000000..8b0940e6c --- /dev/null +++ b/http-client.env.json @@ -0,0 +1,16 @@ +{ + "Lab4 :: acc-18": { + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "200", + "testCount": "6000", + "processingTimeMillis": "3000" + }, + "Lab4 :: acc-23": { + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "16", + "testCount": "1600", + "processingTimeMillis": "30000" + } +} diff --git a/src/main/kotlin/ru/quipy/apigateway/APIController.kt b/src/main/kotlin/ru/quipy/apigateway/APIController.kt index 43c01db47..9c2b06a97 100644 --- a/src/main/kotlin/ru/quipy/apigateway/APIController.kt +++ b/src/main/kotlin/ru/quipy/apigateway/APIController.kt @@ -5,10 +5,19 @@ import io.micrometer.core.instrument.MeterRegistry import org.slf4j.Logger import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired -import org.springframework.web.bind.annotation.* +import org.springframework.http.HttpHeaders +import org.springframework.http.HttpStatus +import org.springframework.http.ResponseEntity +import org.springframework.web.bind.annotation.ExceptionHandler +import org.springframework.web.bind.annotation.PathVariable +import org.springframework.web.bind.annotation.PostMapping +import org.springframework.web.bind.annotation.RequestBody +import org.springframework.web.bind.annotation.RequestParam +import org.springframework.web.bind.annotation.RestController +import ru.quipy.common.utils.TooManyRequestsException import ru.quipy.orders.repository.OrderRepository import ru.quipy.payments.logic.OrderPayer -import java.util.* +import java.util.UUID @RestController class APIController(@Autowired val meterRegistry: MeterRegistry) { @@ -77,6 +86,12 @@ class APIController(@Autowired val meterRegistry: MeterRegistry) { return PaymentSubmissionDto(createdAt, paymentId) } + @ExceptionHandler(TooManyRequestsException::class) + fun tooManyRequestsExceptionHandler(exception: TooManyRequestsException) = + ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS) + .headers(HttpHeaders().apply { add(HttpHeaders.RETRY_AFTER, exception.delay.toString()) }) + .body(exception.message) + class PaymentSubmissionDto( val timestamp: Long, val transactionId: UUID diff --git a/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt b/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt new file mode 100644 index 000000000..7c8853ca8 --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt @@ -0,0 +1,46 @@ +package ru.quipy.common.utils + +import kotlinx.coroutines.sync.Semaphore +import java.time.Duration +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.floor + +/** + * Обёртка над семафором, которая учитывает сколько семафоров можно взять в единицу времени и если до указанного + * момента осталось меньше, чем limitDuration единиц времени, то ограничивает возможность взятия семафора + */ +class InstantRateLimitSemaphore( + limitDuration: Duration, + private val rateDuration: Duration, + private val unitsPerRate: Int +): RateLimitSemaphore(Semaphore(durationCapacity(limitDuration, rateDuration, unitsPerRate)) ) { + + constructor(limitDuration: Duration, timeUnit: TimeUnit, unitsPerRate: Int) : + this(limitDuration, Duration.ofNanos(timeUnit.toNanos(1)), unitsPerRate) + + /** + * Функция вычисляет допустимый limit взятий семафора на заданный период + */ + override fun limit(volume: Instant): Int = + Duration.between(Instant.now(), volume) + ?.takeIf { it.toNanos() > 0 } + ?.let { durationCapacity(it, rateDuration, unitsPerRate) } + ?: 0 + + companion object { + /** + * Вычисление количества ресурса на Duration при известном объёме на единицу TimeUnit + * @param duration Длительность + * @param rateDuration Длительность периода + * @param unitsPerRate Количество ресурса на период + * @return количество элементов, доступных на duration + */ + private fun durationCapacity( + duration: Duration, + rateDuration: Duration, + unitsPerRate: Int) = + floor(unitsPerRate.toDouble() * duration.toNanos() / rateDuration.toNanos()).toInt() + } + +} diff --git a/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt b/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt new file mode 100644 index 000000000..b96daaef4 --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt @@ -0,0 +1,44 @@ +package ru.quipy.common.utils + +import kotlinx.coroutines.sync.Semaphore +import java.util.concurrent.atomic.AtomicInteger + +abstract class RateLimitSemaphore protected constructor (private val semaphore: Semaphore) { + + val acquires = AtomicInteger() + + fun acquire(value: T): Boolean { + return limitAcquire(limit(value)) + } + + fun release() { + try { + semaphore.release() + acquires.decrementAndGet() + } catch (ex : IllegalStateException) { + throw ex + } + } + + /** + * Функция вычисления limit для заданного объёма + */ + protected abstract fun limit(volume: T): Int + + /** + * Метод берёт семафор, но ограничивая его capacity в пределах limit, который меньше, чем даёт сам семафор + */ + private fun limitAcquire(limit: Int): Boolean { + // Берём семафор + if (limit > 0 && semaphore.tryAcquire()) { + // Если успешно получили, то оцениваем количество захватов + if (acquires.incrementAndGet() <= limit) { + // Если не превысили лимит, то возвращаем, что всё ОК + return true + } + release() + } + return false + } + +} \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt index 6a3debf57..348fb2f3a 100644 --- a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt +++ b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt @@ -9,10 +9,8 @@ import org.slf4j.LoggerFactory import java.time.Duration import java.util.concurrent.Executors import java.util.concurrent.PriorityBlockingQueue -import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicLong -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock +import java.util.concurrent.locks.LockSupport class SlidingWindowRateLimiter( private val rate: Long, @@ -40,6 +38,17 @@ class SlidingWindowRateLimiter( } } + /** + * Пытаемся взять блокировку до заданного момента времени + */ + fun blockingUntil(instant: Long): Boolean { + while(System.currentTimeMillis() < instant) { + if (tick()) return true + LockSupport.parkNanos(Duration.ofMillis(1).toNanos()) + } + return false + } + data class Measure( val value: Long, val timestamp: Long diff --git a/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt b/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt new file mode 100644 index 000000000..8da775dae --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt @@ -0,0 +1,3 @@ +package ru.quipy.common.utils + +class TooManyRequestsException(val delay: Int = 1) : Exception("Too many requests") \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt b/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt index eceb90cff..edaa7a8bf 100644 --- a/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt +++ b/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt @@ -8,12 +8,15 @@ import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import ru.quipy.payments.logic.* +import ru.quipy.payments.logic.PaymentAccountProperties +import ru.quipy.payments.logic.PaymentAggregateState +import ru.quipy.payments.logic.PaymentExternalSystemAdapter +import ru.quipy.payments.logic.PaymentExternalSystemAdapterImpl import java.net.URI import java.net.http.HttpClient import java.net.http.HttpRequest import java.net.http.HttpResponse -import java.util.* +import java.util.UUID @Configuration @@ -36,7 +39,7 @@ class PaymentAccountsConfig { lateinit var allowedAccounts: List @Bean - fun accountAdapters(paymentService: EventSourcingService): List { + fun paymentAccountProperties(): List { val request = HttpRequest.newBuilder() .uri(URI("http://${paymentProviderHostPort}/external/accounts?serviceName=$serviceName&token=$token")) .GET() @@ -51,6 +54,14 @@ class PaymentAccountsConfig { ) .filter { it.accountName in allowedAccounts } .map { it.copy(enabled = true) } + } + + @Bean + fun accountAdapters( + paymentAccountProperties: List, + paymentService: EventSourcingService + ): List { + return paymentAccountProperties .onEach(::println) .map { PaymentExternalSystemAdapterImpl( diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index 3f047e3bd..989026caf 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -5,19 +5,29 @@ import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import ru.quipy.common.utils.CallerBlockingRejectedExecutionHandler +import ru.quipy.common.utils.InstantRateLimitSemaphore import ru.quipy.common.utils.NamedThreadFactory +import ru.quipy.common.utils.TooManyRequestsException import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import java.util.* +import java.time.Duration +import java.time.Instant +import java.util.UUID import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit +import kotlin.math.roundToInt +import kotlin.math.roundToLong @Service -class OrderPayer { +class OrderPayer(paymentAccountProperties: List) { companion object { val logger: Logger = LoggerFactory.getLogger(OrderPayer::class.java) + const val MIN_PARALLEL_PROCESS = 16 + const val MAX_PARALLEL_PROCESS = 256 + const val DELAY_COEFFICIENT = 1.2 + const val MIN_DELAY_ADD_MILLIS = 75L } @Autowired @@ -26,9 +36,15 @@ class OrderPayer { @Autowired private lateinit var paymentService: PaymentService + private val parallelThreads = paymentAccountProperties + .sumOf { it.rateLimitPerSec.coerceAtMost(MAX_PARALLEL_PROCESS) } + private val poolSize = (parallelThreads + 2) + .coerceAtLeast(MIN_PARALLEL_PROCESS) + .coerceAtMost(MAX_PARALLEL_PROCESS) + private val paymentExecutor = ThreadPoolExecutor( - 16, - 16, + poolSize, + poolSize, 0L, TimeUnit.MILLISECONDS, LinkedBlockingQueue(8_000), @@ -36,20 +52,63 @@ class OrderPayer { CallerBlockingRejectedExecutionHandler() ) + private val callsPerMinute = paymentAccountProperties.sumOf { + (TimeUnit.MINUTES.toNanos(1).toDouble() * + it.rateLimitPerSec.coerceAtMost(MAX_PARALLEL_PROCESS) / + it.averageProcessingTime.toNanos()).roundToInt()} + + private val minAverageProcessingTime = paymentAccountProperties.minOf { it.averageProcessingTime } + private val maxAverageProcessingTime = paymentAccountProperties.maxOf { it.averageProcessingTime } + + + /** + * Это предполагаемое время, которое может понадобиться внешнему сервису на выполнение нашего запроса + * Т.е.: если у нас до deathTime остаётся меньше callDelay, то мы его не ставим в очередь, а получаем + * отказ от семафора + */ + val callDelay: Duration = Duration + .ofMillis((minAverageProcessingTime.toMillis() * DELAY_COEFFICIENT) + .roundToLong() + .coerceAtLeast(MIN_DELAY_ADD_MILLIS)) + + val instantRateLimitSemaphore = + Triple( + // Делаем объем по задачам на несколько секунды вперед. + // То есть если есть свободные места в очереди на эти 3 величины обработки запроса и задаче не протухнет + // до того момента, когда сможет выполниться, то мы её ставим в очередь, а если нет, + // то возвращаем TooManyRequests + maxAverageProcessingTime.multipliedBy(3), + TimeUnit.MINUTES, + callsPerMinute + ).let { + logger.info("Create OrderPayer::InstantRateLimitSemaphore(duration=${it.first}, timeUnit=${it.second}, rate=${it.third})") + InstantRateLimitSemaphore(it.first, it.second, it.third) + } + fun processPayment(orderId: UUID, amount: Int, paymentId: UUID, deadline: Long): Long { val createdAt = System.currentTimeMillis() - paymentExecutor.submit { - val createdEvent = paymentESService.create { - it.create( - paymentId, - orderId, - amount - ) - } - logger.trace("Payment {} for order {} created.", createdEvent.paymentId, orderId) + val deadLineTime = Instant.ofEpochMilli(deadline) + if (instantRateLimitSemaphore.acquire(deadLineTime.minus(callDelay))) { + paymentExecutor.submit { + try { + val createdEvent = paymentESService.create { + it.create( + paymentId, + orderId, + amount + ) + } + logger.trace("Payment {} for order {} created.", createdEvent.paymentId, orderId) - paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) + paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) + } finally { + instantRateLimitSemaphore.release() + } + } + } else { + logger.error("Payment: $paymentId retried. Too many requests") + throw TooManyRequestsException() } return createdAt } -} \ No newline at end of file +} diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 501f86f6e..47730e2a5 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -5,13 +5,14 @@ import com.fasterxml.jackson.module.kotlin.registerKotlinModule import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody +import org.slf4j.Logger import org.slf4j.LoggerFactory import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.net.SocketTimeoutException import java.time.Duration -import java.util.* +import java.util.UUID import java.util.concurrent.Semaphore import java.util.concurrent.TimeUnit @@ -25,7 +26,7 @@ class PaymentExternalSystemAdapterImpl( ) : PaymentExternalSystemAdapter { companion object { - val logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) + val logger: Logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) val emptyBody = RequestBody.create(null, ByteArray(0)) val mapper = ObjectMapper().registerKotlinModule() @@ -46,73 +47,79 @@ class PaymentExternalSystemAdapterImpl( Duration.ofSeconds(1) ) - fun deadlineHandler(paymentId: UUID, transactionId: UUID) { + fun deadlineHandler(paymentId: UUID, transactionId: UUID, reason: String) { paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Deadline passed") + it.logProcessing(false, now(), transactionId, reason = "Deadline by reason: $reason") } - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId") - return + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId. Reason: $reason") } override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { - logger.warn("[$accountName] Submitting payment request for payment $paymentId") - logger.info("rate limiter rate limit per sec: {}", rateLimitPerSec) + logger.warn("[$accountName] Try to submit payment request for payment $paymentId") val transactionId = UUID.randomUUID() - - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. - paymentESService.update(paymentId) { - it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) + var acquired = semaphoreRequestAcquire(semaphore, deadline) + // Пытаемся взять блокировку на ограничение параллельных запросов к сервису + if (!acquired) { + deadlineHandler(paymentId, transactionId, "Unable to acquire request semaphore") + return } - try { -// if (semaphore.tryAcquire(averageProcessTime.toMillis(), TimeUnit.MILLISECONDS)) { -// return deadlineHandler(paymentId, transactionId) -// } - semaphore.acquire() - rateLimiter.tickBlocking() - - logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - client.newCall(request).execute().use { response -> - semaphore.release() - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } + // Если блокировка взята, то пытаемся влезть в окно исполнения до возможного момента вызова + if (!rateLimiter.blockingUntil(deadline)) { + deadlineHandler(paymentId, transactionId, "Rate limit exceeded") + return + } + try { + + logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + val request = Request.Builder().run { + url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") + post(emptyBody) + }.build() - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + // Вне зависимости от исхода оплаты важно отметить что она была отправлена. + // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - } - } catch (e: Exception) { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + + client.newCall(request).execute().use { response -> + semaphore.release().also { acquired = false } // Снимаем семафор пораньше + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) } - } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } + } + } catch (e: Exception) { + when (e) { + is SocketTimeoutException -> { + logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + } + } + else -> { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = e.message) + } } } } + } finally { + if (acquired) semaphore.release() } } @@ -122,6 +129,28 @@ class PaymentExternalSystemAdapterImpl( override fun name() = properties.accountName + /** + * Пробуем взять семафор, но не позднее момента протухания запроса (чуть ранее) + */ + private fun semaphoreRequestAcquire(semaphore: Semaphore, epocTime: Long) = + remainingRequestMillis(epocTime).takeIf { it > 0 } // Если ещё есть время на блокировку + ?.let { semaphore.tryAcquire(it, TimeUnit.MILLISECONDS) } + ?: false + + /** + * Сколько миллисекунд осталось до завершения запрос с учётом средней возможной задержки + */ + private fun remainingRequestMillis(epocTime: Long) = + remainingMillis(epocTime)// - averageProcessTime.toMillis() - 75) + + /** + * Сколько миллисекунд осталось до заданного момента + */ + private fun remainingMillis(epocTime: Long) = + System.currentTimeMillis().takeIf { it < epocTime } + ?.let { epocTime - it } + ?: 0 + } -public fun now() = System.currentTimeMillis() \ No newline at end of file +fun now() = System.currentTimeMillis() \ No newline at end of file diff --git a/test-local-run.http b/test-local-run.http index aedf30118..f79c8cac1 100644 --- a/test-local-run.http +++ b/test-local-run.http @@ -5,9 +5,9 @@ Content-Type: application/json { "serviceName": "{{serviceName}}", "token": "{{token}}", - "ratePerSecond": 2, - "testCount": 500, - "processingTimeMillis": 60000 + "ratePerSecond": {{ratePerSecond}}, + "testCount": {{testCount}}, + "processingTimeMillis": {{processingTimeMillis}} } ### Stop running test to save time and resources From 3e846615e2becd24167509066d89f42a27d6b0b8 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 23 Oct 2025 19:37:54 +0300 Subject: [PATCH 09/23] feat: done lab5 and need to impl leaking bucket --- http-client.env.json | 28 ++++++++++++++++++++++++++++ test-on-prem-run.http | 11 +++++------ 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/http-client.env.json b/http-client.env.json index 8b0940e6c..e7410f26d 100644 --- a/http-client.env.json +++ b/http-client.env.json @@ -1,5 +1,6 @@ { "Lab4 :: acc-18": { + "accounts": "acc-18", "serviceName": "m3403-8", "token": "bTgVPIlM5Jbr0T03=8", "ratePerSecond": "200", @@ -7,10 +8,37 @@ "processingTimeMillis": "3000" }, "Lab4 :: acc-23": { + "accounts": "acc-23", "serviceName": "m3403-8", "token": "bTgVPIlM5Jbr0T03=8", "ratePerSecond": "16", "testCount": "1600", "processingTimeMillis": "30000" + }, + "Lab5 :: acc-23 :: v1": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "15", + "testCount": "3000", + "processingTimeMillis": "2500" + }, + "Lab5 :: acc-23 :: v2": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "11", + "testCount": "2200", + "processingTimeMillis": "13000", + "profile": "s_0.7_60" + }, + "Lab5 :: acc-23 :: v3": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "3", + "testCount": "1150", + "processingTimeMillis": "26000", + "runits": "90" } } diff --git a/test-on-prem-run.http b/test-on-prem-run.http index cf01bfbb5..980570dd6 100644 --- a/test-on-prem-run.http +++ b/test-on-prem-run.http @@ -6,12 +6,11 @@ Content-Type: application/json { "serviceName": "{{serviceName}}", "token": "{{token}}", - "branch": "main", - "accounts": "acc-12,acc-20", - "ratePerSecond": 2, - "testCount": 10, - "processingTimeMillis": 80000, - "onPremises": true + "branch": "lab4", + "accounts": "{{accounts}}", + "ratePerSecond": {{ratePerSecond}}, + "testCount": {{testCount}}, + "processingTimeMillis": {{processingTimeMillis}} } ### Stop running test to save credits From 39cebdd751ffb8ee50d3d6b825a0544965cfa0bb Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 30 Oct 2025 11:53:47 +0300 Subject: [PATCH 10/23] feat: added LeakingBucketRateLimiter that normalise rate limit --- .../ru/quipy/payments/logic/PaymentExternalServiceImpl.kt | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 47730e2a5..a4fd496cf 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -7,6 +7,7 @@ import okhttp3.Request import okhttp3.RequestBody import org.slf4j.Logger import org.slf4j.LoggerFactory +import ru.quipy.common.utils.LeakingBucketRateLimiter import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate @@ -42,9 +43,10 @@ class PaymentExternalSystemAdapterImpl( private val semaphore = Semaphore(parallelRequests, true) - private val rateLimiter = SlidingWindowRateLimiter( + private val rateLimiter = LeakingBucketRateLimiter( rateLimitPerSec.toLong(), - Duration.ofSeconds(1) + Duration.ofSeconds(1), + (rateLimitPerSec * 1.2).toInt() // Example bucket size: use something reasonable or make configurable ) fun deadlineHandler(paymentId: UUID, transactionId: UUID, reason: String) { @@ -65,7 +67,7 @@ class PaymentExternalSystemAdapterImpl( } try { // Если блокировка взята, то пытаемся влезть в окно исполнения до возможного момента вызова - if (!rateLimiter.blockingUntil(deadline)) { + if (!rateLimiter.tick()) { deadlineHandler(paymentId, transactionId, "Rate limit exceeded") return } From a9c8176fed71838e32393577ef20979aec3c5218 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 6 Nov 2025 18:38:22 +0300 Subject: [PATCH 11/23] feat: added retries for tests that return fails from bank --- http-client.env.json | 10 +++- .../logic/PaymentExternalServiceImpl.kt | 48 ++++++++++++------- 2 files changed, 40 insertions(+), 18 deletions(-) diff --git a/http-client.env.json b/http-client.env.json index e7410f26d..f819ae55d 100644 --- a/http-client.env.json +++ b/http-client.env.json @@ -40,5 +40,13 @@ "testCount": "1150", "processingTimeMillis": "26000", "runits": "90" - } + }, + "Lab6 :: acc-8 :: v3": { + "accounts": "acc-8", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "7", + "testCount": "800", + "processingTimeMillis": "3500" + } } diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index a4fd496cf..587caa366 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -31,6 +31,8 @@ class PaymentExternalSystemAdapterImpl( val emptyBody = RequestBody.create(null, ByteArray(0)) val mapper = ObjectMapper().registerKotlinModule() + + const val MAX_RETRIES_AMOUNT = 4 } private val serviceName = properties.serviceName @@ -60,6 +62,8 @@ class PaymentExternalSystemAdapterImpl( logger.warn("[$accountName] Try to submit payment request for payment $paymentId") val transactionId = UUID.randomUUID() var acquired = semaphoreRequestAcquire(semaphore, deadline) + var result = false + // Пытаемся взять блокировку на ограничение параллельных запросов к сервису if (!acquired) { deadlineHandler(paymentId, transactionId, "Unable to acquire request semaphore") @@ -80,27 +84,37 @@ class PaymentExternalSystemAdapterImpl( post(emptyBody) }.build() - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. - paymentESService.update(paymentId) { - it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) - } - client.newCall(request).execute().use { response -> - semaphore.release().also { acquired = false } // Снимаем семафор пораньше - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + for (attempt in 1..MAX_RETRIES_AMOUNT) { + var result = false + // Вне зависимости от исхода оплаты важно отметить что она была отправлена. + // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. + paymentESService.update(paymentId) { + it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + client.newCall(request).execute().use { response -> + semaphore.release().also { acquired = false } // Снимаем семафор пораньше + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + + result = body.result + logger.info("[$accountName] Payment passed with result: ${body.result}, and message: ${body.message}, attempt number: $attempt") + + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + paymentESService.update(paymentId) { + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } + } + if (result) { + break } } } catch (e: Exception) { From 1bac33da8475bffde8405f9fe1b00a385f7ae07f Mon Sep 17 00:00:00 2001 From: alevushkin Date: Thu, 27 Nov 2025 17:25:05 +0300 Subject: [PATCH 12/23] fix --- http-client.env.json | 10 +++++++++- src/main/resources/application.properties | 2 +- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/http-client.env.json b/http-client.env.json index f819ae55d..17113fa48 100644 --- a/http-client.env.json +++ b/http-client.env.json @@ -48,5 +48,13 @@ "ratePerSecond": "7", "testCount": "800", "processingTimeMillis": "3500" - } + }, + "Lab8 :: acc-9 :: v3": { + "accounts": "acc-9", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "100", + "testCount": "5000", + "processingTimeMillis": "20000" + } } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index fb4ea1302..50950085a 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -26,5 +26,5 @@ management.endpoints.web.exposure.include=info,health,prometheus,metrics payment.service-name=${PAYMENT_SERVICE_NAME} payment.token=${PAYMENT_TOKEN} -payment.accounts=${PAYMENT_ACCOUNTS:acc-12,acc-20} +payment.accounts=acc-9 payment.hostPort=${PAYMENT_HOST:localhost}:${PAYMENT_PORT:1234} \ No newline at end of file From f41f5328695fb92a6d9aaf71aebed51766758ac3 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 27 Nov 2025 17:46:20 +0300 Subject: [PATCH 13/23] feat --- .../config/EventSourcingLibConfiguration.kt | 17 +- .../logic/PaymentExternalServiceImpl.kt | 206 +++++++++++++----- src/main/resources/application.properties | 10 + 3 files changed, 172 insertions(+), 61 deletions(-) diff --git a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt index 9bcb80d07..873bc99f0 100644 --- a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt +++ b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt @@ -66,12 +66,23 @@ class EventSourcingLibConfiguration { } } - @Bean // hack Jetty to tweak the number of possible https2 streams + @Bean // hack Jetty to tweak the number of possible https2 streams and optimize for high load fun jettyServerCustomizer(): JettyServletWebServerFactory { val jettyServletWebServerFactory = JettyServletWebServerFactory() - val c = JettyServerCustomizer { - (it.connectors[0].getConnectionFactory("h2c") as HTTP2CServerConnectionFactory).maxConcurrentStreams = 10_000_000 + val c = JettyServerCustomizer { server -> + // Настройка HTTP/2 + (server.connectors[0].getConnectionFactory("h2c") as HTTP2CServerConnectionFactory).maxConcurrentStreams = 10_000_000 + + // Оптимизация пула потоков для обработки входящих запросов + // Acceptors и selectors настраиваются через application.properties + server.connectors.forEach { connector -> + val executor = connector.executor + if (executor is org.eclipse.jetty.util.thread.QueuedThreadPool) { + executor.minThreads = 32 + executor.maxThreads = 128 + } + } } jettyServletWebServerFactory.serverCustomizers.add(c) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 587caa366..ba2b9f619 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -2,18 +2,25 @@ package ru.quipy.payments.logic import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.module.kotlin.registerKotlinModule +import okhttp3.Call +import okhttp3.Callback +import okhttp3.Dispatcher import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody +import okhttp3.Response import org.slf4j.Logger import org.slf4j.LoggerFactory import ru.quipy.common.utils.LeakingBucketRateLimiter -import ru.quipy.common.utils.SlidingWindowRateLimiter +import ru.quipy.common.utils.NamedThreadFactory import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate +import java.io.IOException import java.net.SocketTimeoutException import java.time.Duration import java.util.UUID +import java.util.concurrent.CompletableFuture +import java.util.concurrent.Executors import java.util.concurrent.Semaphore import java.util.concurrent.TimeUnit @@ -41,7 +48,26 @@ class PaymentExternalSystemAdapterImpl( private val rateLimitPerSec = properties.rateLimitPerSec private val parallelRequests = properties.parallelRequests - private val client = OkHttpClient.Builder().build() + // Создаем кастомный executor для HTTP клиента + private val httpClientExecutor = Executors.newFixedThreadPool( + parallelRequests.coerceAtLeast(16), + NamedThreadFactory("http-client-$accountName") + ) + + // Создаем executor для обработки ответов + private val responseHandlingExecutor = Executors.newFixedThreadPool( + 16, + NamedThreadFactory("response-handler-$accountName") + ) + + private val dispatcher = Dispatcher(httpClientExecutor).apply { + maxRequests = parallelRequests * 2 + maxRequestsPerHost = parallelRequests * 2 + } + + private val client = OkHttpClient.Builder() + .dispatcher(dispatcher) + .build() private val semaphore = Semaphore(parallelRequests, true) @@ -61,82 +87,146 @@ class PaymentExternalSystemAdapterImpl( override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { logger.warn("[$accountName] Try to submit payment request for payment $paymentId") val transactionId = UUID.randomUUID() - var acquired = semaphoreRequestAcquire(semaphore, deadline) - var result = false - + // Пытаемся взять блокировку на ограничение параллельных запросов к сервису + val acquired = semaphoreRequestAcquire(semaphore, deadline) if (!acquired) { deadlineHandler(paymentId, transactionId, "Unable to acquire request semaphore") return } - try { - // Если блокировка взята, то пытаемся влезть в окно исполнения до возможного момента вызова - if (!rateLimiter.tick()) { - deadlineHandler(paymentId, transactionId, "Rate limit exceeded") - return - } - try { - - logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") + + // Если блокировка взята, то пытаемся влезть в окно исполнения до возможного момента вызова + if (!rateLimiter.tick()) { + semaphore.release() + deadlineHandler(paymentId, transactionId, "Rate limit exceeded") + return + } - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() + logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") + // Вне зависимости от исхода оплаты важно отметить что она была отправлена. + // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. + paymentESService.update(paymentId) { + it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) + } - for (attempt in 1..MAX_RETRIES_AMOUNT) { - var result = false - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. - paymentESService.update(paymentId) { - it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) - } + // Используем асинхронный вызов с CompletableFuture + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, 1) + .thenApplyAsync({ result -> + semaphore.release() + result + }, responseHandlingExecutor) + .exceptionally { exception -> + semaphore.release() + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", exception) + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = exception.message ?: "Unknown error") + } + false + } + } - client.newCall(request).execute().use { response -> - semaphore.release().also { acquired = false } // Снимаем семафор пораньше - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } + private fun performPaymentWithRetry( + paymentId: UUID, + amount: Int, + transactionId: UUID, + paymentStartedAt: Long, + deadline: Long, + attempt: Int + ): CompletableFuture { + if (attempt > MAX_RETRIES_AMOUNT) { + return CompletableFuture.completedFuture(false) + } - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + val request = Request.Builder().run { + url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") + post(emptyBody) + }.build() - result = body.result - logger.info("[$accountName] Payment passed with result: ${body.result}, and message: ${body.message}, attempt number: $attempt") + val future = CompletableFuture() - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + client.newCall(request).enqueue(object : Callback { + override fun onFailure(call: Call, e: IOException) { + responseHandlingExecutor.submit { + when (e) { + is SocketTimeoutException -> { + logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId, attempt: $attempt", e) + if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + // Retry on timeout + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) + .thenAccept { result -> future.complete(result) } + } else { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + } + future.complete(false) + } } - } - if (result) { - break - } - } - } catch (e: Exception) { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") + else -> { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", e) + if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + // Retry on error + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) + .thenAccept { result -> future.complete(result) } + } else { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = e.message) + } + future.complete(false) + } } } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + } + } - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) + override fun onResponse(call: Call, response: Response) { + responseHandlingExecutor.submit { + try { + response.use { + val body = try { + mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } + + logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + + logger.info("[$accountName] Payment passed with result: ${body.result}, and message: ${body.message}, attempt number: $attempt") + + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + paymentESService.update(paymentId) { + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } + + if (body.result) { + future.complete(true) + } else if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + // Retry if payment failed + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) + .thenAccept { result -> future.complete(result) } + } else { + future.complete(false) + } + } + } catch (e: Exception) { + logger.error("[$accountName] Error processing response for txId: $transactionId, payment: $paymentId", e) + if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) + .thenAccept { result -> future.complete(result) } + } else { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = e.message) + } + future.complete(false) } } } } - } finally { - if (acquired) semaphore.release() - } + }) + + return future } override fun price() = properties.price diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 50950085a..392fa2123 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -3,6 +3,16 @@ server.port=${SERVER_PORT:8081} server.http2.enabled=true spring.main.allow-bean-definition-overriding=true +# Jetty configuration for high load +server.jetty.threads.min=32 +server.jetty.threads.max=128 +server.jetty.acceptors=8 +server.jetty.selectors=16 +server.jetty.max-queue-capacity=1000000 + +# Spring MVC async configuration +spring.mvc.async.request-timeout=200000 + # MongoDB properties spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 From 3ac181d00dcee26c8327822b28ccc35cd4f480ec Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 27 Nov 2025 18:30:04 +0300 Subject: [PATCH 14/23] files --- http-client.env.json | 10 +++++++++- test-on-prem-run.http | 2 +- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/http-client.env.json b/http-client.env.json index 17113fa48..3eac6dbd5 100644 --- a/http-client.env.json +++ b/http-client.env.json @@ -49,12 +49,20 @@ "testCount": "800", "processingTimeMillis": "3500" }, - "Lab8 :: acc-9 :: v3": { + "Lab8 :: acc-9": { "accounts": "acc-9", "serviceName": "m3403-8", "token": "bTgVPIlM5Jbr0T03=8", "ratePerSecond": "100", "testCount": "5000", "processingTimeMillis": "20000" + }, + "Lab9 :: acc-12": { + "accounts": "acc-12", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "1000", + "testCount": "200000", + "processingTimeMillis": "50000" } } diff --git a/test-on-prem-run.http b/test-on-prem-run.http index 980570dd6..ef1464dc9 100644 --- a/test-on-prem-run.http +++ b/test-on-prem-run.http @@ -6,7 +6,7 @@ Content-Type: application/json { "serviceName": "{{serviceName}}", "token": "{{token}}", - "branch": "lab4", + "branch": "lab9_M", "accounts": "{{accounts}}", "ratePerSecond": {{ratePerSecond}}, "testCount": {{testCount}}, From 30e3bac88e3e37c5b4854cd4f59e410628f3e166 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 27 Nov 2025 18:38:41 +0300 Subject: [PATCH 15/23] fix --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index cc6f2e042..3d9587cd0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,7 +6,7 @@ RUN mvn dependency:go-offline COPY src src RUN mvn package -FROM openjdk:17-jdk-slim +FROM eclipse-temurin:17-alpine-3.22 COPY --from=build /app/target/*.jar /high-load-course.jar From 782115022100fd07362e884d1b31ba1bda540c6d Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 27 Nov 2025 20:04:09 +0300 Subject: [PATCH 16/23] feat: http added into builder configuration --- .../ru/quipy/payments/logic/PaymentExternalServiceImpl.kt | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index ba2b9f619..b718da86f 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -4,8 +4,10 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.module.kotlin.registerKotlinModule import okhttp3.Call import okhttp3.Callback +import okhttp3.ConnectionPool import okhttp3.Dispatcher import okhttp3.OkHttpClient +import okhttp3.Protocol import okhttp3.Request import okhttp3.RequestBody import okhttp3.Response @@ -65,8 +67,12 @@ class PaymentExternalSystemAdapterImpl( maxRequestsPerHost = parallelRequests * 2 } - private val client = OkHttpClient.Builder() + + private val client = OkHttpClient.Builder() .dispatcher(dispatcher) + .connectionPool(ConnectionPool(parallelRequests, 20, TimeUnit.SECONDS)) + .protocols(listOf(Protocol.HTTP_2)) + .readTimeout(Duration.ofSeconds(30)) .build() private val semaphore = Semaphore(parallelRequests, true) From 8c863f94386df57bc38273ce7770ec3d60318f9f Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Thu, 27 Nov 2025 20:09:34 +0300 Subject: [PATCH 17/23] fix: wrong format --- .../ru/quipy/payments/logic/PaymentExternalServiceImpl.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index b718da86f..4c824e050 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -71,7 +71,7 @@ class PaymentExternalSystemAdapterImpl( private val client = OkHttpClient.Builder() .dispatcher(dispatcher) .connectionPool(ConnectionPool(parallelRequests, 20, TimeUnit.SECONDS)) - .protocols(listOf(Protocol.HTTP_2)) + .protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE)) .readTimeout(Duration.ofSeconds(30)) .build() From ce3a90f0a9d300298d3b6379f5fe8bf8bfcd6ee2 Mon Sep 17 00:00:00 2001 From: Michael Shindarev Date: Mon, 15 Dec 2025 21:38:36 +0300 Subject: [PATCH 18/23] refactored http client --- .../config/EventSourcingLibConfiguration.kt | 4 +- .../ru/quipy/payments/logic/OrderPayer.kt | 7 +- .../logic/PaymentExternalServiceImpl.kt | 253 +++++++----------- src/main/resources/application.properties | 21 +- test-on-prem-run.http | 12 +- 5 files changed, 122 insertions(+), 175 deletions(-) diff --git a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt index 873bc99f0..a2958c354 100644 --- a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt +++ b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt @@ -79,8 +79,8 @@ class EventSourcingLibConfiguration { server.connectors.forEach { connector -> val executor = connector.executor if (executor is org.eclipse.jetty.util.thread.QueuedThreadPool) { - executor.minThreads = 32 - executor.maxThreads = 128 + executor.minThreads = 64 + executor.maxThreads = 200 } } } diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index 989026caf..e13196087 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -24,8 +24,8 @@ class OrderPayer(paymentAccountProperties: List) { companion object { val logger: Logger = LoggerFactory.getLogger(OrderPayer::class.java) - const val MIN_PARALLEL_PROCESS = 16 - const val MAX_PARALLEL_PROCESS = 256 + const val MIN_PARALLEL_PROCESS = 64 + const val MAX_PARALLEL_PROCESS = 512 const val DELAY_COEFFICIENT = 1.2 const val MIN_DELAY_ADD_MILLIS = 75L } @@ -42,12 +42,13 @@ class OrderPayer(paymentAccountProperties: List) { .coerceAtLeast(MIN_PARALLEL_PROCESS) .coerceAtMost(MAX_PARALLEL_PROCESS) + // Увеличена очередь до 50000 для обработки большего количества запросов private val paymentExecutor = ThreadPoolExecutor( poolSize, poolSize, 0L, TimeUnit.MILLISECONDS, - LinkedBlockingQueue(8_000), + LinkedBlockingQueue(50_000), NamedThreadFactory("payment-submission-executor"), CallerBlockingRejectedExecutionHandler() ) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 4c824e050..807ef1b7f 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -2,28 +2,24 @@ package ru.quipy.payments.logic import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.module.kotlin.registerKotlinModule -import okhttp3.Call -import okhttp3.Callback -import okhttp3.ConnectionPool -import okhttp3.Dispatcher -import okhttp3.OkHttpClient -import okhttp3.Protocol -import okhttp3.Request -import okhttp3.RequestBody -import okhttp3.Response import org.slf4j.Logger import org.slf4j.LoggerFactory -import ru.quipy.common.utils.LeakingBucketRateLimiter import ru.quipy.common.utils.NamedThreadFactory +import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate import java.io.IOException -import java.net.SocketTimeoutException +import java.net.URI +import java.net.http.HttpClient +import java.net.http.HttpClient.Version +import java.net.http.HttpRequest +import java.net.http.HttpResponse +import java.net.http.HttpTimeoutException import java.time.Duration import java.util.UUID import java.util.concurrent.CompletableFuture -import java.util.concurrent.Executors -import java.util.concurrent.Semaphore +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit @@ -38,7 +34,6 @@ class PaymentExternalSystemAdapterImpl( companion object { val logger: Logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) - val emptyBody = RequestBody.create(null, ByteArray(0)) val mapper = ObjectMapper().registerKotlinModule() const val MAX_RETRIES_AMOUNT = 4 @@ -50,80 +45,49 @@ class PaymentExternalSystemAdapterImpl( private val rateLimitPerSec = properties.rateLimitPerSec private val parallelRequests = properties.parallelRequests - // Создаем кастомный executor для HTTP клиента - private val httpClientExecutor = Executors.newFixedThreadPool( - parallelRequests.coerceAtLeast(16), - NamedThreadFactory("http-client-$accountName") + private val httpClientExecutor = ThreadPoolExecutor( + 64, + 64, + 0, + TimeUnit.SECONDS, + LinkedBlockingQueue(100000), + NamedThreadFactory("payment-http-client") ) - - // Создаем executor для обработки ответов - private val responseHandlingExecutor = Executors.newFixedThreadPool( - 16, - NamedThreadFactory("response-handler-$accountName") - ) - - private val dispatcher = Dispatcher(httpClientExecutor).apply { - maxRequests = parallelRequests * 2 - maxRequestsPerHost = parallelRequests * 2 - } + private val dbExecutor = ThreadPoolExecutor( + 1000, + 1000, + 0, + TimeUnit.SECONDS, + LinkedBlockingQueue(50000), + NamedThreadFactory("payment-db-callback") + ) - private val client = OkHttpClient.Builder() - .dispatcher(dispatcher) - .connectionPool(ConnectionPool(parallelRequests, 20, TimeUnit.SECONDS)) - .protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE)) - .readTimeout(Duration.ofSeconds(30)) + private val client = HttpClient.newBuilder() + .version(Version.HTTP_2) + .executor(httpClientExecutor) + .connectTimeout(Duration.ofSeconds(3)) .build() - private val semaphore = Semaphore(parallelRequests, true) - - private val rateLimiter = LeakingBucketRateLimiter( - rateLimitPerSec.toLong(), - Duration.ofSeconds(1), - (rateLimitPerSec * 1.2).toInt() // Example bucket size: use something reasonable or make configurable + private val rateLimiter = SlidingWindowRateLimiter( + (rateLimitPerSec * 0.95).toLong(), + Duration.ofSeconds(1) ) - fun deadlineHandler(paymentId: UUID, transactionId: UUID, reason: String) { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Deadline by reason: $reason") - } - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId. Reason: $reason") - } - override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { - logger.warn("[$accountName] Try to submit payment request for payment $paymentId") val transactionId = UUID.randomUUID() - // Пытаемся взять блокировку на ограничение параллельных запросов к сервису - val acquired = semaphoreRequestAcquire(semaphore, deadline) - if (!acquired) { - deadlineHandler(paymentId, transactionId, "Unable to acquire request semaphore") - return - } - - // Если блокировка взята, то пытаемся влезть в окно исполнения до возможного момента вызова - if (!rateLimiter.tick()) { - semaphore.release() - deadlineHandler(paymentId, transactionId, "Rate limit exceeded") - return - } + rateLimiter.tickBlocking() - logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. paymentESService.update(paymentId) { it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - // Используем асинхронный вызов с CompletableFuture performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, 1) .thenApplyAsync({ result -> - semaphore.release() result - }, responseHandlingExecutor) + }, dbExecutor) .exceptionally { exception -> - semaphore.release() logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", exception) paymentESService.update(paymentId) { it.logProcessing(false, now(), transactionId, reason = exception.message ?: "Unknown error") @@ -144,95 +108,86 @@ class PaymentExternalSystemAdapterImpl( return CompletableFuture.completedFuture(false) } - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - val future = CompletableFuture() - - client.newCall(request).enqueue(object : Callback { - override fun onFailure(call: Call, e: IOException) { - responseHandlingExecutor.submit { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId, attempt: $attempt", e) - if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - // Retry on timeout - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) - .thenAccept { result -> future.complete(result) } - } else { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") - } - future.complete(false) - } - } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", e) - if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - // Retry on error - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) - .thenAccept { result -> future.complete(result) } - } else { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) - } - future.complete(false) - } - } - } - } - } - - override fun onResponse(call: Call, response: Response) { - responseHandlingExecutor.submit { + val url = "http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount" + val request = HttpRequest.newBuilder() + .uri(URI.create(url)) + .version(Version.HTTP_2) + .POST(HttpRequest.BodyPublishers.noBody()) + .timeout(Duration.ofSeconds(30)) + .build() + + return client.sendAsync(request, HttpResponse.BodyHandlers.ofString()) + .thenComposeAsync({ response -> + CompletableFuture.supplyAsync({ try { - response.use { - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } - - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") + val body = try { + mapper.readValue(response.body(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.statusCode()}, reason: ${response.body()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } - logger.info("[$accountName] Payment passed with result: ${body.result}, and message: ${body.message}, attempt number: $attempt") + // Убрали warn и info логи для уменьшения overhead в горячем пути - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) - } + // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. + // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + paymentESService.update(paymentId) { + it.logProcessing(body.result, now(), transactionId, reason = body.message) + } - if (body.result) { - future.complete(true) - } else if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - // Retry if payment failed - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) - .thenAccept { result -> future.complete(result) } - } else { - future.complete(false) - } + if (body.result) { + true + } else if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + // Retry if payment failed + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() + } else { + false } } catch (e: Exception) { logger.error("[$accountName] Error processing response for txId: $transactionId, payment: $paymentId", e) if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1) - .thenAccept { result -> future.complete(result) } + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() } else { paymentESService.update(paymentId) { it.logProcessing(false, now(), transactionId, reason = e.message) } - future.complete(false) + false + } + } + }, dbExecutor) + }, dbExecutor) + .exceptionally { exception -> + val cause = exception.cause + val isTimeout = cause is HttpTimeoutException + + if (isTimeout || cause is IOException) { + if (isTimeout) { + logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) + } else { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) + } + + if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + // Retry on timeout or error + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() + } else { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = if (isTimeout) "Request timeout." else exception.message) + } + false + } + } else { + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) + if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { + performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() + } else { + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = exception.message) } + false } } } - }) - - return future } override fun price() = properties.price @@ -241,20 +196,6 @@ class PaymentExternalSystemAdapterImpl( override fun name() = properties.accountName - /** - * Пробуем взять семафор, но не позднее момента протухания запроса (чуть ранее) - */ - private fun semaphoreRequestAcquire(semaphore: Semaphore, epocTime: Long) = - remainingRequestMillis(epocTime).takeIf { it > 0 } // Если ещё есть время на блокировку - ?.let { semaphore.tryAcquire(it, TimeUnit.MILLISECONDS) } - ?: false - - /** - * Сколько миллисекунд осталось до завершения запрос с учётом средней возможной задержки - */ - private fun remainingRequestMillis(epocTime: Long) = - remainingMillis(epocTime)// - averageProcessTime.toMillis() - 75) - /** * Сколько миллисекунд осталось до заданного момента */ diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 392fa2123..2314a0ef3 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -3,12 +3,12 @@ server.port=${SERVER_PORT:8081} server.http2.enabled=true spring.main.allow-bean-definition-overriding=true -# Jetty configuration for high load -server.jetty.threads.min=32 -server.jetty.threads.max=128 -server.jetty.acceptors=8 -server.jetty.selectors=16 -server.jetty.max-queue-capacity=1000000 +# Jetty configuration for high load - максимальные ресурсы +server.jetty.threads.min=128 +server.jetty.threads.max=500 +server.jetty.acceptors=16 +server.jetty.selectors=32 +server.jetty.max-queue-capacity=2000000 # Spring MVC async configuration spring.mvc.async.request-timeout=200000 @@ -24,11 +24,16 @@ event.sourcing.scan-package=ru.quipy event.sourcing.snapshots-enabled=false event.sourcing.sagas-enabled=false -# Postgres event store properties +# Postgres event store properties - максимальные ресурсы для высокой нагрузки spring.datasource.hikari.jdbc-url=jdbc:postgresql://${POSTGRES_ADDRESS:localhost}:${POSTGRES_PORT:65432}/postgres spring.datasource.hikari.username=tiny_es spring.datasource.hikari.password=tiny_es -spring.datasource.hikari.leak-detection-threshold=2000 +spring.datasource.hikari.maximum-pool-size=150 +spring.datasource.hikari.minimum-idle=50 +spring.datasource.hikari.connection-timeout=20000 +spring.datasource.hikari.idle-timeout=300000 +spring.datasource.hikari.max-lifetime=1200000 +spring.datasource.hikari.leak-detection-threshold=5000 management.metrics.web.server.request.autotime.percentiles=0.95 management.metrics.export.prometheus.enabled=true diff --git a/test-on-prem-run.http b/test-on-prem-run.http index ef1464dc9..6947aff80 100644 --- a/test-on-prem-run.http +++ b/test-on-prem-run.http @@ -4,13 +4,13 @@ POST http://77.234.215.138:34321/run Content-Type: application/json { - "serviceName": "{{serviceName}}", - "token": "{{token}}", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", "branch": "lab9_M", - "accounts": "{{accounts}}", - "ratePerSecond": {{ratePerSecond}}, - "testCount": {{testCount}}, - "processingTimeMillis": {{processingTimeMillis}} + "accounts": "acc-12", + "ratePerSecond": "1000", + "testCount": "200000", + "processingTimeMillis": "50000" } ### Stop running test to save credits From 8b41b2c2095e9511fe56890bbc48bd582ae6de41 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Thu, 12 Mar 2026 22:35:52 +0300 Subject: [PATCH 19/23] added pattern hedged requests --- .../logic/PaymentExternalServiceImpl.kt | 156 ++++++++---------- 1 file changed, 67 insertions(+), 89 deletions(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 807ef1b7f..49576a5ff 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -8,19 +8,19 @@ import ru.quipy.common.utils.NamedThreadFactory import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import java.io.IOException import java.net.URI import java.net.http.HttpClient import java.net.http.HttpClient.Version import java.net.http.HttpRequest import java.net.http.HttpResponse -import java.net.http.HttpTimeoutException import java.time.Duration import java.util.UUID import java.util.concurrent.CompletableFuture +import java.util.concurrent.Executors import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean // Advice: always treat time as a Duration @@ -33,17 +33,13 @@ class PaymentExternalSystemAdapterImpl( companion object { val logger: Logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) - val mapper = ObjectMapper().registerKotlinModule() - - const val MAX_RETRIES_AMOUNT = 4 } private val serviceName = properties.serviceName private val accountName = properties.accountName private val averageProcessTime = properties.averageProcessingTime private val rateLimitPerSec = properties.rateLimitPerSec - private val parallelRequests = properties.parallelRequests private val httpClientExecutor = ThreadPoolExecutor( 64, @@ -63,6 +59,8 @@ class PaymentExternalSystemAdapterImpl( NamedThreadFactory("payment-db-callback") ) + private val scheduler = Executors.newScheduledThreadPool(8, NamedThreadFactory("payment-hedge-scheduler")) + private val client = HttpClient.newBuilder() .version(Version.HTTP_2) .executor(httpClientExecutor) @@ -76,17 +74,14 @@ class PaymentExternalSystemAdapterImpl( override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { val transactionId = UUID.randomUUID() - + rateLimiter.tickBlocking() paymentESService.update(paymentId) { it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, 1) - .thenApplyAsync({ result -> - result - }, dbExecutor) + performHedgedPayment(paymentId, amount, transactionId, deadline) .exceptionally { exception -> logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", exception) paymentESService.update(paymentId) { @@ -96,98 +91,84 @@ class PaymentExternalSystemAdapterImpl( } } - private fun performPaymentWithRetry( + /** + * Отправляет первый запрос и если за какое-то вермя не получен ответ, + * отправляет параллельный запрос с тем же transactionId в качестве + * ключа идемпотентности. Короче побеждает тот кто ответил первым + */ + private fun performHedgedPayment( paymentId: UUID, amount: Int, transactionId: UUID, - paymentStartedAt: Long, deadline: Long, - attempt: Int ): CompletableFuture { - if (attempt > MAX_RETRIES_AMOUNT) { - return CompletableFuture.completedFuture(false) + val result = CompletableFuture() + val processed = AtomicBoolean(false) + + fun completeOnce(success: Boolean, reason: String?) { + if (processed.compareAndSet(false, true)) { + paymentESService.update(paymentId) { + it.logProcessing(success, now(), transactionId, reason = reason) + } + result.complete(success) + } } + fun failOnce(ex: Throwable) { + if (processed.compareAndSet(false, true)) { + result.completeExceptionally(ex) + } + } + + fun launchRequest() { + sendSingleRequest(paymentId, amount, transactionId) + .thenAcceptAsync({ (success, reason) -> completeOnce(success, reason) }, dbExecutor) + .exceptionally { ex -> failOnce(ex.cause ?: ex); null } + } + + launchRequest() + + val hedgeDelayMs = averageProcessTime.toMillis() + if (remainingMillis(deadline) > hedgeDelayMs + averageProcessTime.toMillis()) { + scheduler.schedule({ + if (!result.isDone) { + try { + rateLimiter.tickBlocking() + launchRequest() + } catch (e: Exception) { + logger.error("[$accountName] Hedged request failed to start for txId: $transactionId, payment: $paymentId", e) + } + } + }, hedgeDelayMs, TimeUnit.MILLISECONDS) + } + + return result + } + + private fun sendSingleRequest( + paymentId: UUID, + amount: Int, + transactionId: UUID, + ): CompletableFuture> { val url = "http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount" val request = HttpRequest.newBuilder() .uri(URI.create(url)) .version(Version.HTTP_2) .POST(HttpRequest.BodyPublishers.noBody()) + .header("x-idempotency-key", transactionId.toString()) .timeout(Duration.ofSeconds(30)) .build() return client.sendAsync(request, HttpResponse.BodyHandlers.ofString()) - .thenComposeAsync({ response -> - CompletableFuture.supplyAsync({ - try { - val body = try { - mapper.readValue(response.body(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.statusCode()}, reason: ${response.body()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) - } - - // Убрали warn и info логи для уменьшения overhead в горячем пути - - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) - paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) - } - - if (body.result) { - true - } else if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - // Retry if payment failed - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() - } else { - false - } - } catch (e: Exception) { - logger.error("[$accountName] Error processing response for txId: $transactionId, payment: $paymentId", e) - if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() - } else { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) - } - false - } - } - }, dbExecutor) - }, dbExecutor) - .exceptionally { exception -> - val cause = exception.cause - val isTimeout = cause is HttpTimeoutException - - if (isTimeout || cause is IOException) { - if (isTimeout) { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) - } else { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) - } - - if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - // Retry on timeout or error - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() - } else { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = if (isTimeout) "Request timeout." else exception.message) - } - false - } - } else { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId, attempt: $attempt", exception) - if (attempt < MAX_RETRIES_AMOUNT && remainingMillis(deadline) > averageProcessTime.toMillis()) { - performPaymentWithRetry(paymentId, amount, transactionId, paymentStartedAt, deadline, attempt + 1).get() - } else { - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = exception.message) - } - false - } + .thenApplyAsync({ response -> + val body = try { + mapper.readValue(response.body(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment response parse error for txId: $transactionId, payment: $paymentId, code: ${response.statusCode()}, body: ${response.body()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) } - } + Pair(body.result, body.message) + }, dbExecutor) } override fun price() = properties.price @@ -196,9 +177,6 @@ class PaymentExternalSystemAdapterImpl( override fun name() = properties.accountName - /** - * Сколько миллисекунд осталось до заданного момента - */ private fun remainingMillis(epocTime: Long) = System.currentTimeMillis().takeIf { it < epocTime } ?.let { epocTime - it } From d6670e5f86659eaadcae1e8ef6e5252a3c0168b9 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Fri, 13 Mar 2026 00:46:55 +0300 Subject: [PATCH 20/23] removed rateLimiter.TickBlocking --- .../kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 49576a5ff..131fc4716 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -133,7 +133,6 @@ class PaymentExternalSystemAdapterImpl( scheduler.schedule({ if (!result.isDone) { try { - rateLimiter.tickBlocking() launchRequest() } catch (e: Exception) { logger.error("[$accountName] Hedged request failed to start for txId: $transactionId, payment: $paymentId", e) From 20b6097c9f3707a42686e8e15508a52b82dd3dc2 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Fri, 13 Mar 2026 01:34:05 +0300 Subject: [PATCH 21/23] fixed DELAY_COEFFICIENT and MIN_DELAY_ADD_MILLIS --- src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index e13196087..425fd1ec6 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -26,8 +26,8 @@ class OrderPayer(paymentAccountProperties: List) { val logger: Logger = LoggerFactory.getLogger(OrderPayer::class.java) const val MIN_PARALLEL_PROCESS = 64 const val MAX_PARALLEL_PROCESS = 512 - const val DELAY_COEFFICIENT = 1.2 - const val MIN_DELAY_ADD_MILLIS = 75L + const val DELAY_COEFFICIENT = 0.8 + const val MIN_DELAY_ADD_MILLIS = 50L } @Autowired From f4ca120a281f19200f7186307dd24aeedcac26c0 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Fri, 13 Mar 2026 02:10:49 +0300 Subject: [PATCH 22/23] =?UTF-8?q?fixed=20=D0=B1=D0=B0=D0=B3=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=BD=D0=B5=D0=B1=D0=BE=D0=BB=D1=8C=D1=88=D0=B8=D1=85,=20?= =?UTF-8?q?=D1=82=D0=B0=D0=BC=20=D0=BF=D0=BE=20=D1=84=D0=B8=D0=B3=D0=BD?= =?UTF-8?q?=D0=B5=20=D0=BA=D0=BE=D1=80=D0=BE=D1=87=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../logic/PaymentExternalServiceImpl.kt | 40 +++++++++---------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 131fc4716..07521776c 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -20,7 +20,7 @@ import java.util.concurrent.Executors import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger // Advice: always treat time as a Duration @@ -93,7 +93,7 @@ class PaymentExternalSystemAdapterImpl( /** * Отправляет первый запрос и если за какое-то вермя не получен ответ, - * отправляет параллельный запрос с тем же transactionId в качестве + * * то отправляет параллельный запрос с тем же transactionId в качестве * ключа идемпотентности. Короче побеждает тот кто ответил первым */ private fun performHedgedPayment( @@ -103,41 +103,41 @@ class PaymentExternalSystemAdapterImpl( deadline: Long, ): CompletableFuture { val result = CompletableFuture() - val processed = AtomicBoolean(false) + // сколько запросов у нас щас. когда будет 0, то значит что все запросы отправили уже и они вернулись с ответами(дай бог) + val pending = AtomicInteger(0) - fun completeOnce(success: Boolean, reason: String?) { - if (processed.compareAndSet(false, true)) { + fun onSuccess(success: Boolean, reason: String?) { + // result.complete() атомарно вернёт true только первому вызову + if (result.complete(success)) { paymentESService.update(paymentId) { it.logProcessing(success, now(), transactionId, reason = reason) } - result.complete(success) } } - fun failOnce(ex: Throwable) { - if (processed.compareAndSet(false, true)) { + fun onError(ex: Throwable) { + if (pending.decrementAndGet() == 0) { result.completeExceptionally(ex) } } - fun launchRequest() { + fun send() { sendSingleRequest(paymentId, amount, transactionId) - .thenAcceptAsync({ (success, reason) -> completeOnce(success, reason) }, dbExecutor) - .exceptionally { ex -> failOnce(ex.cause ?: ex); null } + .thenAcceptAsync({ (success, reason) -> onSuccess(success, reason) }, dbExecutor) + .exceptionally { ex -> onError(ex.cause ?: ex); null } } - launchRequest() + pending.incrementAndGet() + send() val hedgeDelayMs = averageProcessTime.toMillis() - if (remainingMillis(deadline) > hedgeDelayMs + averageProcessTime.toMillis()) { + if (remainingMillis(deadline) > hedgeDelayMs * 2) { + // Резервируем слот заране, xnj,s если первый упадёт до старта hedge, то + // pending не обнулится раньше + pending.incrementAndGet() scheduler.schedule({ - if (!result.isDone) { - try { - launchRequest() - } catch (e: Exception) { - logger.error("[$accountName] Hedged request failed to start for txId: $transactionId, payment: $paymentId", e) - } - } + if (!result.isDone) send() + else pending.decrementAndGet() // тут типа первый уже ответил }, hedgeDelayMs, TimeUnit.MILLISECONDS) } From 520d19e850fc0f7d8c8d1581dfe67920296efee1 Mon Sep 17 00:00:00 2001 From: aryzhikov Date: Fri, 13 Mar 2026 20:13:30 +0300 Subject: [PATCH 23/23] =?UTF-8?q?=D1=86=D0=B8=D1=84=D0=B5=D1=80=D0=BA?= =?UTF-8?q?=D0=B8=20=D0=BF=D0=BE=D0=BF=D1=80=D0=B0=D0=B2=D0=B8=D0=BB=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ru/quipy/payments/logic/OrderPayer.kt | 7 ++-- .../logic/PaymentExternalServiceImpl.kt | 33 ++++++++++++------- 2 files changed, 24 insertions(+), 16 deletions(-) diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index 425fd1ec6..e610c1bd3 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -26,8 +26,8 @@ class OrderPayer(paymentAccountProperties: List) { val logger: Logger = LoggerFactory.getLogger(OrderPayer::class.java) const val MIN_PARALLEL_PROCESS = 64 const val MAX_PARALLEL_PROCESS = 512 - const val DELAY_COEFFICIENT = 0.8 - const val MIN_DELAY_ADD_MILLIS = 50L + const val DELAY_COEFFICIENT = 0.0 + const val MIN_DELAY_ADD_MILLIS = 10L } @Autowired @@ -100,11 +100,10 @@ class OrderPayer(paymentAccountProperties: List) { ) } logger.trace("Payment {} for order {} created.", createdEvent.paymentId, orderId) - - paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) } finally { instantRateLimitSemaphore.release() } + paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) } } else { logger.error("Payment: $paymentId retried. Too many requests") diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 07521776c..c4fdcfb29 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -68,7 +68,7 @@ class PaymentExternalSystemAdapterImpl( .build() private val rateLimiter = SlidingWindowRateLimiter( - (rateLimitPerSec * 0.95).toLong(), + rateLimitPerSec.toLong(), Duration.ofSeconds(1) ) @@ -122,7 +122,9 @@ class PaymentExternalSystemAdapterImpl( } fun send() { - sendSingleRequest(paymentId, amount, transactionId) + // Таймаут HTTP привязываем к дедлайну — не ждём 30с если платёж уже не успеет + val timeoutMs = remainingMillis(deadline).coerceAtLeast(500L) + sendSingleRequest(paymentId, amount, transactionId, timeoutMs) .thenAcceptAsync({ (success, reason) -> onSuccess(success, reason) }, dbExecutor) .exceptionally { ex -> onError(ex.cause ?: ex); null } } @@ -130,17 +132,23 @@ class PaymentExternalSystemAdapterImpl( pending.incrementAndGet() send() - val hedgeDelayMs = averageProcessTime.toMillis() - if (remainingMillis(deadline) > hedgeDelayMs * 2) { - // Резервируем слот заране, xnj,s если первый упадёт до старта hedge, то - // pending не обнулится раньше - pending.incrementAndGet() - scheduler.schedule({ - if (!result.isDone) send() - else pending.decrementAndGet() // тут типа первый уже ответил - }, hedgeDelayMs, TimeUnit.MILLISECONDS) + fun scheduleHedge(delayMs: Long) { + if (remainingMillis(deadline) > delayMs) { + pending.incrementAndGet() + scheduler.schedule({ + if (!result.isDone) send() + else pending.decrementAndGet() + }, delayMs, TimeUnit.MILLISECONDS) + } } + val avg = averageProcessTime.toMillis() + // 5 попыток равномерно по шкале дедлайна + scheduleHedge((avg * 0.10).toLong().coerceAtLeast(150L)) // ~1300ms остаток + scheduleHedge((avg * 0.22).toLong().coerceAtLeast(280L)) // ~1170ms остаток + scheduleHedge((avg * 0.38).toLong().coerceAtLeast(450L)) // ~1000ms остаток + scheduleHedge((avg * 0.55).toLong().coerceAtLeast(650L)) // ~800ms остаток + return result } @@ -148,6 +156,7 @@ class PaymentExternalSystemAdapterImpl( paymentId: UUID, amount: Int, transactionId: UUID, + timeoutMs: Long = 30_000L, ): CompletableFuture> { val url = "http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount" val request = HttpRequest.newBuilder() @@ -155,7 +164,7 @@ class PaymentExternalSystemAdapterImpl( .version(Version.HTTP_2) .POST(HttpRequest.BodyPublishers.noBody()) .header("x-idempotency-key", transactionId.toString()) - .timeout(Duration.ofSeconds(30)) + .timeout(Duration.ofMillis(timeoutMs)) .build() return client.sendAsync(request, HttpResponse.BodyHandlers.ofString())