Chore/optimize api wallet

api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt

@@ -26,22 +26,58 @@ class WebClientConfig(
     private val logbook: Logbook,
     @Value("\${app.auth.url}")
     private val url: String,
+    @Value("\${app.http.client.wiretap.enabled:false}")
+    private val wiretapEnabled: Boolean,
+    @Value("\${app.http.client.general.max-connections:300}")
+    private val generalMaxConnections: Int,
+    @Value("\${app.http.client.general.pending-acquire-max-count:1000}")
+    private val generalPendingAcquireMaxCount: Int,
+    @Value("\${app.http.client.general.max-idle-seconds:30}")
+    private val generalMaxIdleSeconds: Long,
+    @Value("\${app.http.client.general.max-life-seconds:120}")
+    private val generalMaxLifeSeconds: Long,
+    @Value("\${app.http.client.general.pending-acquire-timeout-seconds:30}")
+    private val generalPendingAcquireTimeoutSeconds: Long,
+    @Value("\${app.http.client.general.connect-timeout-millis:5000}")
+    private val generalConnectTimeoutMillis: Int,
+    @Value("\${app.http.client.general.response-timeout-seconds:30}")
+    private val generalResponseTimeoutSeconds: Long,
+    @Value("\${app.http.client.keycloak.max-connections:150}")
+    private val keycloakMaxConnections: Int,
+    @Value("\${app.http.client.keycloak.pending-acquire-max-count:500}")
+    private val keycloakPendingAcquireMaxCount: Int,
+    @Value("\${app.http.client.keycloak.max-idle-seconds:30}")
+    private val keycloakMaxIdleSeconds: Long,
+    @Value("\${app.http.client.keycloak.max-life-seconds:120}")
+    private val keycloakMaxLifeSeconds: Long,
+    @Value("\${app.http.client.keycloak.pending-acquire-timeout-seconds:60}")
+    private val keycloakPendingAcquireTimeoutSeconds: Long,
+    @Value("\${app.http.client.keycloak.connect-timeout-millis:10000}")
+    private val keycloakConnectTimeoutMillis: Int,
+    @Value("\${app.http.client.keycloak.response-timeout-seconds:10}")
+    private val keycloakResponseTimeoutSeconds: Long,
 ) {
     private val provider = ConnectionProvider.builder("apiPool")
-        .maxConnections(150)
-        .pendingAcquireMaxCount(100)
-        .maxIdleTime(Duration.ofSeconds(30))
-        .maxLifeTime(Duration.ofMinutes(2))
-        .pendingAcquireTimeout(Duration.ofSeconds(10))
+        .maxConnections(generalMaxConnections)
+        .pendingAcquireMaxCount(generalPendingAcquireMaxCount)
+        .maxIdleTime(Duration.ofSeconds(generalMaxIdleSeconds))
+        .maxLifeTime(Duration.ofSeconds(generalMaxLifeSeconds))
+        .pendingAcquireTimeout(Duration.ofSeconds(generalPendingAcquireTimeoutSeconds))
         .evictInBackground(Duration.ofMinutes(1))
         .build()
 
-    private val client = HttpClient.create(provider)
-        .wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE)
-        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
-        .responseTimeout(Duration.ofSeconds(30))
-        .keepAlive(true)
-        .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) }
+    private val client = HttpClient.create(provider).let {
+        val configured = if (wiretapEnabled) {
+            it.wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE)
+        } else {
+            it
+        }
+        configured
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, generalConnectTimeoutMillis)
+            .responseTimeout(Duration.ofSeconds(generalResponseTimeoutSeconds))
+            .keepAlive(true)
+            .doOnConnected { conn -> conn.addHandlerLast(LogbookClientHandler(logbook)) }
+    }
 
 
     @Bean("generalWebClient")
@@ -63,16 +99,17 @@ class WebClientConfig(
     @Bean("keycloakWebClient")
     fun keycloakWebClient(logbook: Logbook): WebClient {
         val provider = ConnectionProvider.builder("keycloakPool")
-            .maxConnections(100)
-            .maxIdleTime(Duration.ofSeconds(30))
-            .maxLifeTime(Duration.ofMinutes(2))
-            .pendingAcquireTimeout(Duration.ofSeconds(60))
+            .maxConnections(keycloakMaxConnections)
+            .pendingAcquireMaxCount(keycloakPendingAcquireMaxCount)
+            .maxIdleTime(Duration.ofSeconds(keycloakMaxIdleSeconds))
+            .maxLifeTime(Duration.ofSeconds(keycloakMaxLifeSeconds))
+            .pendingAcquireTimeout(Duration.ofSeconds(keycloakPendingAcquireTimeoutSeconds))
             .evictInBackground(Duration.ofMinutes(1))
             .build()
 
         val client = HttpClient.create(provider)
-            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
-            .responseTimeout(Duration.ofSeconds(10))
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, keycloakConnectTimeoutMillis)
+            .responseTimeout(Duration.ofSeconds(keycloakResponseTimeoutSeconds))
             .keepAlive(true)
             .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) }
 

api/api-app/src/main/resources/application.yml

@@ -107,10 +107,10 @@ logging:
   level:
     co.nilin: INFO
     org.zalando.logbook: TRACE
-    reactor.netty.pool: DEBUG
-    reactor.netty.http.client: DEBUG
-    org.springframework.web.reactive.function.client: DEBUG
-    co.nilin.opex.api.ports.proxy.impl: DEBUG
+    reactor.netty.pool: WARN
+    reactor.netty.http.client: WARN
+    org.springframework.web.reactive.function.client: WARN
+    co.nilin.opex.api.ports.proxy.impl: INFO
 
 app:
   base:
@@ -152,6 +152,31 @@ app:
   api:
     crypto:
       key: ${api_crypto_key:0e1fd29572ec8c85970d76e3433e96ee}
+  http:
+    client:
+      wiretap:
+        enabled: ${HTTP_CLIENT_WIRETAP_ENABLED:false}
+      general:
+        max-connections: ${HTTP_CLIENT_GENERAL_MAX_CONNECTIONS:300}
+        pending-acquire-max-count: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_MAX_COUNT:1000}
+        max-idle-seconds: ${HTTP_CLIENT_GENERAL_MAX_IDLE_SECONDS:30}
+        max-life-seconds: ${HTTP_CLIENT_GENERAL_MAX_LIFE_SECONDS:120}
+        pending-acquire-timeout-seconds: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_TIMEOUT_SECONDS:30}
+        connect-timeout-millis: ${HTTP_CLIENT_GENERAL_CONNECT_TIMEOUT_MILLIS:5000}
+        response-timeout-seconds: ${HTTP_CLIENT_GENERAL_RESPONSE_TIMEOUT_SECONDS:30}
+      keycloak:
+        max-connections: ${HTTP_CLIENT_KEYCLOAK_MAX_CONNECTIONS:150}
+        pending-acquire-max-count: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_MAX_COUNT:500}
+        max-idle-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_IDLE_SECONDS:30}
+        max-life-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_LIFE_SECONDS:120}
+        pending-acquire-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_TIMEOUT_SECONDS:60}
+        connect-timeout-millis: ${HTTP_CLIENT_KEYCLOAK_CONNECT_TIMEOUT_MILLIS:10000}
+        response-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_RESPONSE_TIMEOUT_SECONDS:10}
+  proxy:
+    market:
+      max-concurrent-requests: ${API_PROXY_MARKET_MAX_CONCURRENT_REQUESTS:64}
+    matching:
+      max-concurrent-requests: ${API_PROXY_MATCHING_MAX_CONCURRENT_REQUESTS:64}
   cors:
     enabled: true
     allowed-origins: ${ALLOWED_ORIGINS:"http://localhost:8110"}
@@ -170,4 +195,3 @@ springdoc:
     display-request-duration: true
     operations-sorter: method
     tags-sorter: alpha
-

api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt

@@ -4,8 +4,30 @@ import kotlinx.coroutines.reactor.asCoroutineDispatcher
 import reactor.core.scheduler.Schedulers
 
 object ProxyDispatchers {
+    private fun envInt(name: String, default: Int): Int {
+        val value = System.getenv(name)?.toIntOrNull() ?: return default
+        return if (value > 0) value else default
+    }
 
-    val general = Schedulers.newBoundedElastic(8, 16, "general").asCoroutineDispatcher()
-    val market = Schedulers.newBoundedElastic(8, 16, "market").asCoroutineDispatcher()
-    val wallet = Schedulers.newBoundedElastic(10, 20, "wallet").asCoroutineDispatcher()
+    private val cpu = Runtime.getRuntime().availableProcessors().coerceAtLeast(4)
+    private val defaultThreads = cpu * 4
+    private val defaultQueue = 10_000
+
+    val general = Schedulers.newBoundedElastic(
+        envInt("API_PROXY_GENERAL_THREADS", defaultThreads),
+        envInt("API_PROXY_GENERAL_QUEUE", defaultQueue),
+        "general"
+    ).asCoroutineDispatcher()
+
+    val market = Schedulers.newBoundedElastic(
+        envInt("API_PROXY_MARKET_THREADS", defaultThreads),
+        envInt("API_PROXY_MARKET_QUEUE", defaultQueue),
+        "market"
+    ).asCoroutineDispatcher()
+
+    val wallet = Schedulers.newBoundedElastic(
+        envInt("API_PROXY_WALLET_THREADS", defaultThreads),
+        envInt("API_PROXY_WALLET_QUEUE", defaultQueue),
+        "wallet"
+    ).asCoroutineDispatcher()
 }

api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt

@@ -34,14 +34,19 @@ class MarketUserDataProxyImpl(@Qualifier("generalWebClient") private val webClie
 
     @Value("\${app.market.url}")
     private lateinit var baseUrl: String
+
+    @Value("\${app.proxy.market.max-concurrent-requests:64}")
+    private var marketMaxConcurrentRequests: Int = 64
     private suspend fun <T> retryOnce(backoffMs: Long = 200, block: suspend () -> T): T =
         try {
             block()
         } catch (e: TimeoutException) {
             delay(backoffMs); block()
         }
 
-    private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0)
+    private val mgLimiter by lazy {
+        Semaphore(permits = marketMaxConcurrentRequests, acquiredPermits = 0)
+    }
 
     override suspend fun queryOrder(
         token: String,

api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt

@@ -39,7 +39,13 @@ class MatchingGatewayProxyImpl(@Qualifier("generalWebClient") private val client
 
     @Value("\${app.matching-gateway.url}")
     private lateinit var baseUrl: String
-    private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0) // fair-like behavior
+
+    @Value("\${app.proxy.matching.max-concurrent-requests:64}")
+    private var matchingMaxConcurrentRequests: Int = 64
+
+    private val mgLimiter by lazy {
+        Semaphore(permits = matchingMaxConcurrentRequests, acquiredPermits = 0)
+    }
     override suspend fun createNewOrder(
         uuid: String?,
         pair: String,

device-management/pom.xml

@@ -22,7 +22,7 @@
         <kotlin.version>2.1.0</kotlin.version>
         <spring.version>3.4.2</spring.version>
         <spring-cloud.version>2024.0.0</spring-cloud.version>
-        <error-hanlder.version>1.2.25</error-hanlder.version>
+        <error-hanlder.version>1.2.26</error-hanlder.version>
     </properties>
 
 

pom.xml

@@ -16,7 +16,7 @@
         <kotlin.version>1.9.0</kotlin.version>
         <spring.version>2.7.6</spring.version>
         <spring-cloud.version>2021.0.5</spring-cloud.version>
-        <error-hanlder.version>1.2.25</error-hanlder.version>
+        <error-hanlder.version>1.2.26</error-hanlder.version>
         <interceptor.version>1.0.8</interceptor.version>
         <skip.unit.tests>true</skip.unit.tests>
         <common.version>1.0.1-beta.38</common.version>