Fix retries

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

@@ -146,7 +146,7 @@ class EthereumDirectReader(
     private fun readBlock(request: JsonRpcRequest, id: String): Mono<BlockContainer> {
         return readWithQuorum(request)
             .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id")))
-            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
+            .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
             .flatMap { blockbytes ->
                 val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>?
                 if (block == null) {
@@ -164,14 +164,18 @@ class EthereumDirectReader(
      * Read from an Upstream applying a Quorum specific for that request
      */
     private fun readWithQuorum(request: JsonRpcRequest): Mono<ByteArray> {
-        return quorumReaderFactory
+        return Mono.just(quorumReaderFactory)
-            .create(
+            .map {
-                up.getApiSource(Selector.empty),
+                it.create(
-                callMethodsFactory.create().createQuorumFor(request.method),
+                    up.getApiSource(Selector.empty),
-                // we do not use Signer for internal requests because it doesn't make much sense
+                    callMethodsFactory.create().createQuorumFor(request.method),
-                null
+                    // we do not use Signer for internal requests because it doesn't make much sense
-            )
+                    null
-            .read(request)
+                )
-            .map { it.value }
+            }.flatMap {
+                it.read(request)
+            }.map {
+                it.value
+            }
     }
 }

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

@@ -334,4 +334,88 @@ class EthereumDirectReaderSpec extends Specification {
                 .verify(Duration.ofSeconds(1))
     }
 
+    def "Reads block by hash with retries"() {
+        setup:
+        def json = new BlockJson().tap {
+            number = 100
+            hash = BlockHash.from(hash1)
+            timestamp = Instant.now()
+            totalDifficulty = BigInteger.ONE
+            transactions = []
+        }
+        def up = Mock(Multistream) {
+            3 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        def result = Mono.just(
+                new QuorumRpcReader.Result(
+                        Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null)
+        )
+        EthereumDirectReader ethereumDirectReader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            2 * create(_, _, _) >> Mock(Reader) {
+                2 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >>>
+                        [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
+            }
+            1 * create(_, _, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> result
+            }
+        }
+        when:
+        def act = ethereumDirectReader.blockReader.read(BlockHash.from(hash1))
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHexWithPrefix() == hash1
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Reads block by number with retries"() {
+        setup:
+        def json = new BlockJson().tap {
+            number = 100
+            hash = BlockHash.from(hash1)
+            timestamp = Instant.now()
+            totalDifficulty = BigInteger.ONE
+            transactions = []
+        }
+        def up = Mock(Multistream) {
+            3 * getApiSource(_) >> Stub(ApiSource)
+        }
+        def calls = Mock(Factory) {
+            3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        def result = Mono.just(
+                new QuorumRpcReader.Result(
+                        Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null)
+        )
+        EthereumDirectReader ethereumDirectReader = new EthereumDirectReader(
+                up, Caches.default(), new CurrentBlockCache(), calls
+        )
+        ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            2 * create(_, _, _) >> Mock(Reader) {
+                2 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >>>
+                        [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
+            }
+            1 * create(_, _, _) >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> result
+            }
+        }
+        when:
+        def act = ethereumDirectReader.blockByHeightReader.read(100)
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches { block ->
+                    block.hash.toHexWithPrefix() == hash1
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
 }