StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

problem: unhandled exception when a remote grpc gives an http error

Browse Source 
(cherry picked from commit bae613e8b8)
This commit is contained in:
a10zn8
2022-12-07 21:46:05 +04:00
parent c20d270e44
commit f936c9b999
3 changed files with 137 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
View File

@@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.grpc.StatusRuntimeException
import org.apache.commons.lang3.time.StopWatch
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.util.concurrent.TimeUnit
@@ -32,7 +34,7 @@ import java.util.concurrent.TimeUnit
class JsonRpcGrpcClient(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val metrics: RpcMetrics
private val metrics: RpcMetrics?,
) {
companion object {
@@ -46,11 +48,11 @@ class JsonRpcGrpcClient(
class Executor(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val metrics: RpcMetrics
private val metrics: RpcMetrics?
) : Reader<JsonRpcRequest, JsonRpcResponse> {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
var startTime: Long = 0
val timer = StopWatch()
val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChainValue(chain.id)
@@ -66,39 +68,58 @@ class JsonRpcGrpcClient(
req.addItems(reqItem.build())
return Mono.just(key)
.doOnNext {
startTime = System.nanoTime()
}.flatMap {
.doOnNext { timer.start() }
.flatMap {
stub.nativeCall(req.build())
.single()
.flatMap { resp ->
if (resp.succeed) {
val bytes = resp.payload.toByteArray()
val signature = if (resp.hasSignature()) {
extractSignature(resp.signature)
} else {
null
}
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
} else {
metrics.fails.increment()
Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
resp.errorMessage
)
)
}
}
.onErrorResume(::handleError)
.flatMap(::handleResponse)
}
.doOnNext {
if (startTime > 0) {
val now = System.nanoTime()
metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
if (timer.isStarted) {
metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
}
}
}
fun handleResponse(resp: BlockchainOuterClass.NativeCallReplyItem): Mono<JsonRpcResponse> =
if (resp.succeed) {
val bytes = resp.payload.toByteArray()
val signature = if (resp.hasSignature()) {
extractSignature(resp.signature)
} else {
null
}
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
} else {
metrics?.fails?.increment()
Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
resp.errorMessage
)
)
}
fun handleError(t: Throwable): Mono<BlockchainOuterClass.NativeCallReplyItem> {
metrics?.fails?.increment()
return when (t) {
is StatusRuntimeException -> Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
"Remote status code: ${t.status.code.name}"
)
)
else -> Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
"Other connection error"
)
)
}
}
fun extractSignature(resp: NativeCallReplySignature?): ResponseSigner.Signature? {
if (resp == null || resp.signature == null || resp.signature.isEmpty || resp.upstreamId == null || resp.upstreamId.isEmpty()) {
return null

8
src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy
View File

@@ -26,14 +26,6 @@ class MockGrpcServer {
GrpcCleanupRule grpcCleanup = new GrpcCleanupRule()
ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(ReactorBlockchainGrpc.BlockchainImplBase impl) {
String serverName = InProcessServerBuilder.generateName()
grpcCleanup.register(InProcessServerBuilder
.forName(serverName).directExecutor().addService(impl).build().start());
def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build())
return ReactorBlockchainGrpc.newReactorStub(channel)
}
ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(BlockchainGrpc.BlockchainImplBase impl){
String serverName = InProcessServerBuilder.generateName()
grpcCleanup.register(InProcessServerBuilder

88
src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClientSpec.groovy Normal file
View File

@@ -0,0 +1,88 @@
package io.emeraldpay.dshackle.upstream.rpcclient
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainGrpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.test.MockGrpcServer
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.emeraldpay.dshackle.Chain
import io.grpc.stub.StreamObserver
import spock.lang.Specification
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
class JsonRpcGrpcClientSpec extends Specification {
def "Makes a request"() {
setup:
def mockGrpc = new MockGrpcServer()
def requested = new AtomicReference<BlockchainOuterClass.NativeCallRequest>()
def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
requested.set(request)
responseObserver.onNext(BlockchainOuterClass.NativeCallReplyItem.newBuilder()
.setId(1)
.setSucceed(true)
.setPayload(ByteString.copyFromUtf8("\"hello world!\""))
.build())
responseObserver.onCompleted()
}
})
def client = new JsonRpcGrpcClient(
grpc, Chain.BITCOIN, null
).getReader()
when:
def act = client.read(
new JsonRpcRequest("test", [])
).block(Duration.ofSeconds(1))
then:
!act.hasError()
act.resultAsProcessedString == "hello world!"
requested.get() == BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_BITCOIN)
.addAllItems([
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("test")
.setPayload(ByteString.copyFromUtf8("[]"))
.build()
])
.build()
}
def "Return error on HTTP error"() {
setup:
def mockGrpc = new MockGrpcServer()
def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
@Override
void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
responseObserver.onError(new IllegalStateException("fail"))
}
})
def client = new JsonRpcGrpcClient(
grpc, Chain.BITCOIN, null
).getReader()
when:
client.read(
new JsonRpcRequest("test", [])
).block(Duration.ofSeconds(1))
then:
def t = thrown(RpcException)
with(t.error) {
message == "Remote status code: UNKNOWN"
it.code == RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR
}
}
}