Adjust retry policies (#199)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt

@@ -78,7 +78,7 @@ class EthereumDirectReader(
                 val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
                 return readWithQuorum(request)
                     .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
-                    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                    .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
                     .flatMap { txbytes ->
                         val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
                         if (tx == null) {
@@ -109,7 +109,7 @@ class EthereumDirectReader(
                             throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value")
                         }
                     }
-                    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+                    .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
                     .doOnNext { value ->
                         balanceCache.put(key, value)
                     }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy

@@ -425,4 +425,59 @@ class EthereumDirectReaderSpec extends Specification {
                 .verify(Duration.ofSeconds(1))
     }
 
+    def "Reads tx by hash with retries - expects an error within 1 sec"() {
+        setup:
+        def up = Mock(Multistream) {
+            4 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            4 * create(_, _, _, _) >> Mock(Reader) {
+                4 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >>>
+                        [Mono.error(new RuntimeException()), Mono.error(new RuntimeException()),
+                         Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
+            }
+        }
+        when:
+        def act = reader.txReader.read(TransactionId.from(hash1))
+        then:
+        StepVerifier.create(act)
+                .expectError()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Reads balance with retries - expects an error within 1 sec"() {
+        setup:
+        def up = Mock(Multistream) {
+            4 * getApiSource(_) >> Stub(ApiSource)
+            1 * getHead() >> Mock(Head) {
+                1 * getCurrentHeight() >> null
+            }
+        }
+        def calls = Mock(Factory) {
+            4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        EthereumDirectReader reader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
+        )
+        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            4 * create(_, _, _, _) >> Mock(Reader) {
+                4 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >>>
+                        [Mono.error(new RuntimeException()), Mono.error(new RuntimeException()),
+                         Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
+            }
+        }
+        when:
+        def act = reader.balanceReader.read(Address.from(address1))
+        then:
+        StepVerifier.create(act)
+                .expectError()
+                .verify(Duration.ofSeconds(1))
+    }
+
 }