diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index 84ae4462..f54e06fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson @@ -39,6 +40,9 @@ open class AlwaysQuorum: CallQuorum { return true } + override fun record(error: RpcException, upstream: Upstream) { + } + override fun getResult(): ByteArray? { return result } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index 88e970fe..026d76cb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -50,8 +50,8 @@ open class BroadcastQuorum( } } - override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { - // can be "message: known transaction: TXID" or "message: Nonce too low" + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + // can be "message: known transaction: TXID", "Transaction with the same hash was already imported" or "message: Nonce too low" calls++ if (result == null) { result = response diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index 80e7ec3c..018d3016 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import reactor.util.function.Tuple2 @@ -30,6 +31,7 @@ interface CallQuorum { fun isResolved(): Boolean fun record(response: ByteArray, upstream: Upstream): Boolean + fun record(error: RpcException, upstream: Upstream) fun getResult(): ByteArray? companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt index 11f7174c..d801be5f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson @@ -48,7 +49,10 @@ open class NonEmptyQuorum( return result } - override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + } + + override fun record(error: RpcException, upstream: Upstream) { } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt index 401faee0..d41b5d17 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import java.util.concurrent.locks.ReentrantLock @@ -64,7 +65,11 @@ open class NonceQuorum( return result } - override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { + override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + errors++ + } + + override fun record(error: RpcException, upstream: Upstream) { errors++ } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index b016a1cf..10e82e53 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import java.util.concurrent.atomic.AtomicReference @@ -42,6 +43,10 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { return false } + override fun record(error: RpcException, upstream: Upstream) { + } + + override fun getResult(): ByteArray { return result.get() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt index 216504d9..ffc8c8f8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt @@ -43,8 +43,12 @@ abstract class ValueAwareQuorum( return isResolved(); } + override fun record(error: RpcException, upstream: Upstream) { + recordError(null, error.rpcMessage, upstream) + } + abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream) - abstract fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) + abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index cdcdf0dd..10b8588d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.RpcException import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -123,17 +124,28 @@ class NativeCall( var failures = 0 return Flux.from(apis) .flatMap { api -> - api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) } + val upstream = api.upstream!! + api.execute(ctx.id, ctx.payload.method, ctx.payload.params) + // on error notify quorum, it may use error message or other details + .doOnError { err -> + if (err is RpcException) { + ctx.callQuorum.record(err, upstream) + } + } + .map { Tuples.of(it, upstream) } } .retry { failures++ - if (failures <= 3) { + if (ctx.callQuorum.isResolved()) { + false + } else if (failures < 3) { apis.request(1) true } else { false } } + // record all correct responses until quorum reached .reduce(ctx.callQuorum, {res, a -> if (res.record(a.t1, a.t2)) { apis.resolve() @@ -142,6 +154,14 @@ class NativeCall( } res }) + // if last call resulted in error it's still possible that request was resolved correctly. i.e. for BroadcastQuorum + .onErrorResume { err -> + if (ctx.callQuorum.isResolved()) { + Mono.just(ctx.callQuorum) + } else { + Mono.error(err) + } + } .doOnNext { if (!it.isResolved()) { log.debug("No quorum for ${ctx.payload.method} as ${ctx.callQuorum}") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 9e793f95..95dc0924 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.quorum.BroadcastQuorum import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.upstream.CachingEthereumApi @@ -27,6 +28,9 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError +import io.infinitape.etherjar.rpc.RpcResponseException import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification @@ -110,10 +114,11 @@ class NativeCallSpec extends Specification { when: def t1 = System.currentTimeMillis() - nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) def delta = System.currentTimeMillis() - t1 then: - delta >= 100 + delta > 95 // should be 100, but sometimes gives less ??? + new String(resp.payload) == '{"jsonrpc":"2.0","id":1,"result":"bar"}' } def "One call has no pause"() { @@ -329,4 +334,35 @@ class NativeCallSpec extends Specification { 1 * quorum.record(_, _) 1 * quorum.getResult() } + + def "Send raw retries 3 times"() { + setup: + def quorum = Spy(new BroadcastQuorum(TestingCommons.rpcConverter(), 3)) + + def upstreams = Stub(Upstreams) + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) + + apiMock.answer("eth_sendRawTransaction", ["0x1234"], + "0x4b66b555df9faed6f0711f2104d183736c8e2dc7434626dd2622e243f041d41b", 1) + apiMock.answer("eth_sendRawTransaction", ["0x1234"], null, 10, + new RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Transaction with the same hash was already imported")) +// apiMock.answer("eth_sendRawTransaction", ["0x1234"], +// new RpcResponseError(RpcResponseError.CODE_INVALID_REQUEST, "Transaction with the same hash was already imported"), 10) + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), + Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_sendRawTransaction", ["0x1234"])) + + + when: + def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def act = objectMapper.readValue(resp.payload, Map) + then: + act == [jsonrpc:"2.0", id:1, result: "0x4b66b555df9faed6f0711f2104d183736c8e2dc7434626dd2622e243f041d41b"] + 1 * quorum.record(_ as byte[], _) + 2 * quorum.record(_ as RpcException, _) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 54dbbe28..275a32a7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.ResponseJson import org.jetbrains.annotations.NotNull @@ -60,9 +61,13 @@ class EthereumApiMock extends DirectEthereumApi { if (predefined.exception != null) { predefined.onCalled() predefined.print() - throw predefined.exception + return Mono.error(predefined.exception) + } + if (predefined.result instanceof RpcResponseError) { + json.error = predefined.result + } else { + json.result = predefined.result } - json.result = predefined.result } else { log.error("Method ${method} with ${params} is not mocked") json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked")