problem: doesn't retry on upstream error

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -134,6 +134,7 @@ class NativeCall(
                 .flatMap { api ->
                     api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) }
                 }
+                .retry(3)
                 .reduce(ctx.callQuorum, {res, a ->
                     if (res.record(a.t1, a.t2)) {
                         repeatControl.onComplete()

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -32,6 +32,7 @@ import reactor.test.StepVerifier
 import spock.lang.Specification
 
 import java.time.Duration
+import java.util.concurrent.TimeoutException
 
 class NativeCallSpec extends Specification {
 
@@ -300,4 +301,32 @@ class NativeCallSpec extends Specification {
         1 * cacheMock.execute(10, "eth_test", []) >> Mono.just('{"result": "foo"}'.bytes)
         new String(act.block().payload) == '{"result": "foo"}'
     }
+
+    def "Retries on error"() {
+        setup:
+        def quorum = Spy(new AlwaysQuorum())
+
+        def upstreams = Stub(Upstreams)
+        RpcClient rpcClient = Stub(RpcClient)
+        def apiMock = TestingCommons.api(rpcClient)
+        apiMock.upstream = Stub(Upstream)
+
+        apiMock.answer("eth_test", [], null, 1, new TimeoutException("test 1"))
+        apiMock.answer("eth_test", [], null, 1, new TimeoutException("test 2"))
+        apiMock.answerOnce("eth_test", [], "bar")
+
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock),
+                Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
+
+
+        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: "bar"]
+        1 * quorum.record(_, _)
+        1 * quorum.getResult()
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy

@@ -45,8 +45,9 @@ class EthereumApiMock extends DirectEthereumApi {
         return answer(method, params, result, 1)
     }
 
-    EthereumApiMock answer(@NotNull String method, List<Object> params, Object result, Integer limit = null) {
+    EthereumApiMock answer(@NotNull String method, List<Object> params, Object result,
-        predefined << new PredefinedResponse(method: method, params: params, result: result, limit: limit)
+                           Integer limit = null, Throwable exception = null) {
+        predefined << new PredefinedResponse(method: method, params: params, result: result, limit: limit, exception: exception)
         return this
     }
 
@@ -55,6 +56,11 @@ class EthereumApiMock extends DirectEthereumApi {
         def predefined = predefined.find { it.isSame(id, method, params) }
         ResponseJson json = new ResponseJson<Object, Integer>(id: id)
         if (predefined != null) {
+            if (predefined.exception != null) {
+                predefined.onCalled()
+                predefined.print()
+                throw predefined.exception
+            }
             json.result = predefined.result
         } else {
             log.error("Method ${method} with ${params} is not mocked")
@@ -84,6 +90,7 @@ class EthereumApiMock extends DirectEthereumApi {
         List params
         Object result
         Integer limit
+        Throwable exception
 
         boolean isSame(int id, String method, List<?> params) {
             if (limit != null) {