diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index eb90c74e..a41cb04e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -146,7 +146,7 @@ class EthereumDirectReader( private fun readBlock(request: JsonRpcRequest, id: String): Mono { 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? 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 { - return quorumReaderFactory - .create( - up.getApiSource(Selector.empty), - callMethodsFactory.create().createQuorumFor(request.method), - // we do not use Signer for internal requests because it doesn't make much sense - null - ) - .read(request) - .map { it.value } + return Mono.just(quorumReaderFactory) + .map { + it.create( + up.getApiSource(Selector.empty), + callMethodsFactory.create().createQuorumFor(request.method), + // we do not use Signer for internal requests because it doesn't make much sense + null + ) + }.flatMap { + it.read(request) + }.map { + it.value + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index c5c9c8bf..48f55904 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/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)) + } + }