From 9af4b95f8f0700e2094eebaee60a4bb314a97055 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Mon, 10 Apr 2023 15:32:47 +0400 Subject: [PATCH] Adjust retry policies (#199) --- .../upstream/ethereum/EthereumDirectReader.kt | 4 +- .../ethereum/EthereumCachingReaderSpec.groovy | 55 +++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) 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 29b819ff..53290dfe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/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) } 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 58ce62a3..005e14e9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/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)) + } + }