From 81f825f1c1e5e9c63671494fe5086770df996f97 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 29 Nov 2022 16:04:56 +0400 Subject: [PATCH 1/3] imcrease gRPC message size limit --- build.gradle | 2 +- src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/build.gradle b/build.gradle index 614f96b9..898cab0f 100644 --- a/build.gradle +++ b/build.gradle @@ -36,7 +36,7 @@ group = 'io.emeraldpay.dshackle' // Version schema: // x.x.x for production, following SemVer model // x.x.x-SNAPSHOT for development -version = '0.15.0-SNAPSHOT' +version = '0.16.0-SNAPSHOT' java { sourceCompatibility = JavaVersion.VERSION_13 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index 8923263a..7c9e1e53 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -46,6 +46,7 @@ open class GrpcServer( log.info("Listening Native gRPC on ${mainConfig.host}:${mainConfig.port}") val serverBuilder = NettyServerBuilder .forAddress(InetSocketAddress(mainConfig.host, mainConfig.port)) + .maxInboundMessageSize(Int.MAX_VALUE) .let { if (mainConfig.accessLogConfig.enabled) { it.intercept(accessHandler) @@ -65,7 +66,7 @@ open class GrpcServer( val server = serverBuilder.build() this.server = server - Thread { server.start() }.run() + server.start() log.info("GRPC Server started") } From a55bd2305f6aa7925fceb05c3582f5306c139cd0 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Wed, 30 Nov 2022 18:03:31 +0400 Subject: [PATCH 2/3] imcrease gRPC message size limit --- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index e49a38d6..588d5fdd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -29,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics -import io.grpc.ManagedChannelBuilder import io.grpc.netty.NettyChannelBuilder import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics @@ -70,21 +69,21 @@ class GrpcUpstreams( private val lock = ReentrantLock() fun start(): Flux { - val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) { - NettyChannelBuilder.forAddress(host, port) - // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) - .maxInboundMessageSize(Int.MAX_VALUE) + val chanelBuilder = NettyChannelBuilder.forAddress(host, port) + // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) + .maxInboundMessageSize(Int.MAX_VALUE) + .enableRetry() + .maxRetryAttempts(3) + if (auth != null && StringUtils.isNotEmpty(auth.ca)) { + chanelBuilder .useTransportSecurity() - .enableRetry() - .maxRetryAttempts(3) .sslContext(withTls(auth)) } else { log.warn("Using insecure connection to $host:$port") - ManagedChannelBuilder.forAddress(host, port) - .usePlaintext() + chanelBuilder.usePlaintext() } - val client = ReactorBlockchainGrpc.newReactorStub(channel.build()) + val client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build()) this.client = client val statusSubscription = AtomicReference() From c491831b61bda12773844b6c558070e9032ab461 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Wed, 30 Nov 2022 18:13:14 +0400 Subject: [PATCH 3/3] set gRPC message size limit 32M --- src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt | 1 + src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt | 2 +- .../io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt b/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt index fe9bc976..47870b00 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt @@ -21,6 +21,7 @@ import java.time.Duration class Defaults { companion object { + const val maxMessageSize: Int = 32 * 1024 * 1024 val timeout: Duration = Duration.ofSeconds(60) val timeoutInternal: Duration = timeout.dividedBy(4) val retryConnection: Duration = Duration.ofSeconds(10) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index 7c9e1e53..cb558725 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -46,7 +46,7 @@ open class GrpcServer( log.info("Listening Native gRPC on ${mainConfig.host}:${mainConfig.port}") val serverBuilder = NettyServerBuilder .forAddress(InetSocketAddress(mainConfig.host, mainConfig.port)) - .maxInboundMessageSize(Int.MAX_VALUE) + .maxInboundMessageSize(Defaults.maxMessageSize) .let { if (mainConfig.accessLogConfig.enabled) { it.intercept(accessHandler) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 588d5fdd..ffae18e8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -71,7 +71,7 @@ class GrpcUpstreams( fun start(): Flux { val chanelBuilder = NettyChannelBuilder.forAddress(host, port) // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) - .maxInboundMessageSize(Int.MAX_VALUE) + .maxInboundMessageSize(Defaults.maxMessageSize) .enableRetry() .maxRetryAttempts(3) if (auth != null && StringUtils.isNotEmpty(auth.ca)) {