problem: tx broadcast breaks call (and causes timeout) if it's already broadcasted

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
     }

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) {
+    override fun recordError(response: ByteArray?,  errorMessage: String?, upstream: Upstream) {
-        // can be "message: known transaction: TXID" or "message: Nonce too low"
+        // 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

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 {

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) {
     }
 
 }

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++
     }
 

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()
     }

src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt

@@ -43,8 +43,12 @@ abstract class ValueAwareQuorum<T>(
         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)
 
 }

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}")

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, _)
+    }
 }

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")