From ac5b44332bd5988821db3c49216a2651a3bdfc92 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Thu, 1 Feb 2024 15:51:54 +0400 Subject: [PATCH] Fixes slot, directReader and wsHead bugs (#411) --- .../dshackle/data/BlockContainer.kt | 2 +- .../dshackle/reader/CompoundReader.kt | 1 - .../upstream/ethereum/EthereumDirectReader.kt | 14 ++++-- .../upstream/ethereum/GenericWsHead.kt | 20 +++++---- .../ethereum/GenericWsHeadSpec.groovy | 45 ++++++++++++------- 5 files changed, 53 insertions(+), 29 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt index 08133a53..1a84f6d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt @@ -105,7 +105,7 @@ class BlockContainer @JvmOverloads constructor( fun copyWithRating(nodeRating: Int): BlockContainer { return BlockContainer( - height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating, + height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating, upstreamId, slot, ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt index 22fb1e7b..33644264 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt @@ -42,7 +42,6 @@ class CompoundReader ( rdr.read(key) .timeout(Defaults.timeoutInternal, Mono.empty()) .doOnError { t -> log.warn("Failed to read from $rdr", t) } - .onErrorResume { Mono.empty() } }, 1,) .next() } 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 2d959b2e..64f121c5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash @@ -31,6 +32,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory +import org.apache.commons.lang3.exception.ExceptionUtils import org.slf4j.LoggerFactory import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono @@ -81,7 +83,7 @@ class EthereumDirectReader( override fun read(key: TransactionId): Mono> { val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex())) return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too - .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key"))) + .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Tx not read $key"))) .flatMap { result -> val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java) if (tx == null) { @@ -96,6 +98,8 @@ class EthereumDirectReader( if (tx.data.blockId != null) { caches.cache(Caches.Tag.REQUESTED, tx.data) } + }.onErrorResume { + Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it))) } } } @@ -128,7 +132,7 @@ class EthereumDirectReader( override fun read(key: TransactionId): Mono> { val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex())) return readWithQuorum(request) - .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key"))) + .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Receipt not read $key"))) .flatMap { result -> val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java) if (receipt == null) { @@ -149,6 +153,8 @@ class EthereumDirectReader( result, ) } + }.onErrorResume { + Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it))) } } } @@ -188,7 +194,7 @@ class EthereumDirectReader( matcher: Selector.Matcher = Selector.empty, ): Mono> { return readWithQuorum(request, matcher) - .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id"))) + .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Block not read $id"))) .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200))) .flatMap { result -> val block = objectMapper.readValue(result.data, BlockJson::class.java) as BlockJson? @@ -208,6 +214,8 @@ class EthereumDirectReader( } .doOnNext { block -> caches.cache(Caches.Tag.REQUESTED, block.data) + }.onErrorResume { + Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it))) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt index 05694f24..e13f23b6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt @@ -50,15 +50,12 @@ class GenericWsHead( private var isSyncing = false private var subscription: Disposable? = null + private var headResubSubscription: Disposable? = null private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort() private val headLivenessSink = Sinks.many().multicast().directBestEffort() private var subscriptionId = AtomicReference("") - init { - registerHeadResubscribeFlux() - } - override fun isRunning(): Boolean { return subscription != null } @@ -73,6 +70,10 @@ class GenericWsHead( listenNewHeads(), ) this.subscription = super.follow(heads) + + if (headResubSubscription == null) { + headResubSubscription = registerHeadResubscribeFlux() + } } override fun onNoHeadUpdates() { @@ -86,7 +87,7 @@ class GenericWsHead( this.isSyncing = isSyncing } - fun listenNewHeads(): Flux { + private fun listenNewHeads(): Flux { return subscribe() .map { chainSpecific.parseHeader(it, "unknown") @@ -102,7 +103,8 @@ class GenericWsHead( override fun stop() { super.stop() cancelSub() - noHeadUpdatesSink.tryEmitComplete() + headResubSubscription?.dispose() + headResubSubscription = null } override fun headLiveness(): Flux = headLivenessSink.asFlux() @@ -135,7 +137,7 @@ class GenericWsHead( } } - private fun registerHeadResubscribeFlux() { + private fun registerHeadResubscribeFlux(): Disposable { val connectionStates = wsSubscriptions.connectionInfoFlux() .map { if (it.connectionId == connectionId && it.connectionState == WsConnection.ConnectionState.DISCONNECTED) { @@ -150,10 +152,10 @@ class GenericWsHead( return@map false } - Flux.merge( + return Flux.merge( noHeadUpdatesSink.asFlux(), connectionStates, - ).subscribeOn(wsConnectionResubscribeScheduler) + ).publishOn(wsConnectionResubscribeScheduler) .filter { it && !subscribed && connected && !isSyncing } .subscribe { log.warn("Restart ws head, upstreamId: $upstreamId") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy index 8442f370..e1ccfcdf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.GenericUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.BlockValidator @@ -59,24 +60,32 @@ class GenericWsHeadSpec extends Specification { Global.objectMapper.writeValueAsBytes(it) } - def apiMock = TestingCommons.api() + def reader = Mock(Reader) { + 1 * it.read(new JsonRpcRequest("eth_getBlockByNumber", List.of("latest", false))) >> Mono.empty() + } def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> Flux.empty() } - def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE) - - def res = BlockContainer.from(block) - when: - def act = head.listenNewHeads().blockFirst() - - then: - act == res - 1 * ws.subscribe(_) >> new WsSubscriptions.SubscribeData( Flux.fromIterable([headBlock]), "id", new AtomicReference("") ) + + def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, reader, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE) + + def res = BlockContainer.from(block) + when: + def act = head.getFlux() + + then: + StepVerifier.create(act) + .then { + head.start() + } + .expectNext(res) + .thenCancel() + .verify(Duration.ofSeconds(3)) } def "Restart ethereum ws head"() { @@ -328,7 +337,9 @@ class GenericWsHeadSpec extends Specification { block.uncles = [] block.totalDifficulty = BigInteger.ONE - def apiMock = TestingCommons.api() + def reader = Mock(Reader) { + 1 * it.read(new JsonRpcRequest("eth_getBlockByNumber", List.of("latest", false))) >> Mono.empty() + } def subId = "subId" def ws = Mock(WsSubscriptions) { 1 * it.connectionInfoFlux() >> Flux.empty() @@ -339,15 +350,19 @@ class GenericWsHeadSpec extends Specification { Mono.just(new JsonRpcResponse("".bytes, null)) } - def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE) + def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, reader, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE) when: - def act = head.listenNewHeads() + def act = head.getFlux() then: StepVerifier.create(act) - .expectComplete() - .verify(Duration.ofSeconds(1)) + .then { + head.start() + } + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(3)) } def "If there is ws disconnect then head must emit false its liveness state"() {