From effe88109148cf568ee491c61ae75691e5a462f6 Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Tue, 11 Jun 2024 14:49:32 +0300 Subject: [PATCH] Track upstream finalization data (#499) --- emerald-grpc | 2 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 23 +++- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 22 ++- .../configure/GenericUpstreamCreator.kt | 1 + .../dshackle/upstream/ChainResponse.kt | 12 +- .../dshackle/upstream/Multistream.kt | 15 +++ .../emeraldpay/dshackle/upstream/Selector.kt | 72 ++++++++-- .../emeraldpay/dshackle/upstream/Upstream.kt | 3 + .../upstream/bitcoin/BitcoinRpcUpstream.kt | 8 ++ .../ethereum/EthereumCachingReader.kt | 33 +---- .../ethereum/EthereumChainSpecific.kt | 10 +- .../upstream/ethereum/EthereumDirectReader.kt | 26 +++- .../ethereum/EthereumFinalizationDetector.kt | 89 +++++++++++++ .../upstream/ethereum/EthereumLocalReader.kt | 42 +++--- .../upstream/finalization/FinalizationData.kt | 61 +++++++++ .../finalization/FinalizationDetector.kt | 32 +++++ .../upstream/generic/AbstractChainSpecific.kt | 12 +- .../upstream/generic/ChainSpecific.kt | 11 +- .../upstream/generic/GenericUpstream.kt | 38 +++++- .../upstream/grpc/BitcoinGrpcUpstream.kt | 8 ++ .../upstream/grpc/GenericGrpcUpstream.kt | 103 +++++++++------ .../dshackle/upstream/grpc/GrpcHead.kt | 26 ++-- .../dshackle/test/ApiReaderMock.groovy | 2 +- .../dshackle/test/GenericUpstreamMock.groovy | 1 + .../dshackle/upstream/FilteredApisSpec.groovy | 3 +- .../ethereum/EthereumCachingReaderSpec.groovy | 35 +++++ .../ethereum/EthereumLocalReaderSpec.groovy | 44 +++++- .../upstream/grpc/GrpcHeadSpec.groovy | 4 +- .../dshackle/upstream/SelectorTest.kt | 99 ++++++++++++++ .../EthereumFinalizationDetectorTest.kt | 75 +++++++++++ .../upstream/grpc/GenericGrpcUpstreamTest.kt | 125 ++++++++++++++++++ src/test/kotlin/test/MockGrpcServerKt.kt | 40 ++++++ 32 files changed, 933 insertions(+), 144 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetector.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationData.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationDetector.kt create mode 100644 src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt create mode 100644 src/test/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstreamTest.kt create mode 100644 src/test/kotlin/test/MockGrpcServerKt.kt diff --git a/emerald-grpc b/emerald-grpc index a3351cdc..42172430 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit a3351cdcb5448f422dc61726a5186c6a16c8829a +Subproject commit 421724301b1fb74ac35286954129b534e9b22f9e diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 9d65c151..54e1471a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.Global.Companion.nullValue @@ -47,6 +48,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.rpcclient.CallParams import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.rpcclient.ObjectParams @@ -243,6 +245,12 @@ open class NativeCall( result.upstreamId = it.id result.upstreamNodeVersion = it.nodeVersion } + it.finalization?.let { + result.finalization = Common.FinalizationData.newBuilder() + .setHeight(it.height) + .setType(it.type.toProtoFinalizationType()) + .build() + } return result.build() } @@ -426,9 +434,9 @@ open class NativeCall( val resolvedUpstreamData = it.resolvedUpstreamData ?: ctx.upstream.getUpstreamSettingsData() validateResult(result, "local", ctx) if (ctx.nonce != null) { - CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, resolvedUpstreamData?.id ?: ctx.upstream.getId()), resolvedUpstreamData, ctx) + CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, resolvedUpstreamData?.id ?: ctx.upstream.getId()), resolvedUpstreamData, ctx, it.finalization) } else { - CallResult.ok(ctx.id, null, result, null, resolvedUpstreamData, ctx) + CallResult.ok(ctx.id, null, result, null, resolvedUpstreamData, ctx, it.finalization) } } }.switchIfEmpty( @@ -436,6 +444,10 @@ open class NativeCall( ) .onErrorResume { Mono.just(CallResult.fail(ctx.id, ctx.nonce, it, ctx)) + }.doOnNext { + if (it.finalization != null && it.upstreamSettingsData != null) { + ctx.upstream.addFinalization(it.finalization, it.upstreamSettingsData.id) + } } } @@ -700,7 +712,7 @@ open class NativeCall( } } - open class CallResult( + open class CallResult @JvmOverloads constructor( val id: Int, val nonce: Long?, val result: ByteArray?, @@ -709,6 +721,7 @@ open class NativeCall( val upstreamSettingsData: Upstream.UpstreamSettingsData?, val ctx: ValidCallContext?, val stream: Flux? = null, + val finalization: FinalizationData? = null, ) { constructor( @@ -725,6 +738,10 @@ open class NativeCall( return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx) } + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: Upstream.UpstreamSettingsData?, ctx: ValidCallContext?, final: FinalizationData?): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, null, final) + } + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: Upstream.UpstreamSettingsData?, ctx: ValidCallContext?, stream: Flux?): CallResult { return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 5f4f92a7..c8e2a4ac 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -35,7 +35,6 @@ import reactor.core.publisher.Mono class StreamHead( @Autowired private val multistreamHolder: MultistreamHolder, ) { - private val log = LoggerFactory.getLogger(StreamHead::class.java) fun add(requestMono: Mono): Flux { @@ -54,12 +53,20 @@ class StreamHead( fun asProto(ms: Multistream, chain: Chain, block: BlockContainer): BlockchainOuterClass.ChainHead { val msLowerBounds = ms.getLowerBounds() - val lowerBoundsProto = msLowerBounds - .map { - BlockchainOuterClass.LowerBound.newBuilder() - .setLowerBoundTimestamp(it.timestamp) - .setLowerBoundType(toProtoLowerBoundType(it.type)) - .setLowerBoundValue(it.lowerBound) + val lowerBoundsProto = + msLowerBounds + .map { + BlockchainOuterClass.LowerBound.newBuilder() + .setLowerBoundTimestamp(it.timestamp) + .setLowerBoundType(toProtoLowerBoundType(it.type)) + .setLowerBoundValue(it.lowerBound) + .build() + } + val finalizationData = + ms.getFinalizations().map { + Common.FinalizationData.newBuilder() + .setHeight(it.height) + .setType(it.type.toProtoFinalizationType()) .build() } val toOldApi = toOldApi(msLowerBounds) @@ -72,6 +79,7 @@ class StreamHead( .setCurrentLowerSlot(toOldApi.slot) .setCurrentLowerDataTimestamp(toOldApi.timestamp) .addAllLowerBounds(lowerBoundsProto) + .addAllFinalizationData(finalizationData) .setTimestamp(block.timestamp.toEpochMilli()) .setWeight(ByteString.copyFrom(block.difficulty.toByteArray())) .setBlockId(block.hash.toHex()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt index f92d8698..243f4d36 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -84,6 +84,7 @@ open class GenericUpstreamCreator( cs::upstreamRpcModulesDetector, buildMethodsFun, cs::lowerBoundService, + cs::finalizationDetectorBuilder, ) upstream.start() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt index ca45b9e2..23c7c542 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt @@ -18,12 +18,13 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.core.JsonGenerator import com.fasterxml.jackson.databind.JsonSerializer import com.fasterxml.jackson.databind.SerializerProvider +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.dshackle.upstream.stream.Chunk import reactor.core.publisher.Flux import reactor.core.publisher.Mono -class ChainResponse( +class ChainResponse @JvmOverloads constructor( private val result: ByteArray?, val error: ChainCallError?, val id: Id, @@ -33,15 +34,18 @@ class ChainResponse( */ val providedSignature: ResponseSigner.Signature? = null, val resolvedUpstreamData: Upstream.UpstreamSettingsData? = null, + val finalization: FinalizationData? = null, ) { constructor(stream: Flux, id: Int) : - this(null, null, NumberId(id.toLong()), stream, null, null) + this(null, null, NumberId(id.toLong()), stream, null, null, null) - constructor(result: ByteArray?, error: ChainCallError?) : this(result, error, NumberId(0), null) + constructor(result: ByteArray?, error: ChainCallError?) : this(result, error, NumberId(0), null, null) constructor(result: ByteArray?, error: ChainCallError?, resolvedUpstreamData: Upstream.UpstreamSettingsData?) : - this(result, error, NumberId(0), null, null, resolvedUpstreamData) + this(result, error, NumberId(0), null, null, resolvedUpstreamData, null) + constructor(result: ByteArray?, resolvedUpstreamData: Upstream.UpstreamSettingsData?, finalization: FinalizationData) : + this(result, null, NumberId(0), null, null, resolvedUpstreamData, finalization) companion object { private val NULL_VALUE = "null".toByteArray() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 5742c7ce..e4e6515d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -32,6 +32,8 @@ import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallSelector +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import io.micrometer.core.instrument.Gauge @@ -48,6 +50,7 @@ import reactor.core.publisher.Sinks import reactor.core.scheduler.Scheduler import java.time.Duration import java.util.concurrent.ConcurrentHashMap +import kotlin.math.max /** * Aggregation of multiple upstreams responding to a single blockchain @@ -355,6 +358,18 @@ abstract class Multistream( started = true } + override fun getFinalizations(): Collection { + return getAll().flatMap { it.getFinalizations() } + .fold(mutableMapOf()) { acc, data -> + acc[data.type] = max(acc[data.type] ?: 0, data.height) + acc + }.toList().map { FinalizationData(it.second, it.first) } + } + + override fun addFinalization(finalization: FinalizationData, upstreamId: String) { + getAll().find { it.getId() == upstreamId }?.addFinalization(finalization, upstreamId) + } + override fun getLowerBounds(): Collection { return lowerBounds.values } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index a49a2dd5..a69c157d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.MatchesResponse.NotMatchedResponse import io.emeraldpay.dshackle.upstream.MatchesResponse.SameNodeResponse import io.emeraldpay.dshackle.upstream.MatchesResponse.SlotHeightResponse import io.emeraldpay.dshackle.upstream.MatchesResponse.Success +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.lowerbound.fromProtoType import org.apache.commons.lang3.StringUtils import java.util.Collections @@ -41,6 +42,51 @@ class Selector { @JvmStatic val anyLabel = AnyLabelMatcher() + sealed class HeightNumberOrTag { + + companion object { + fun fromHeightSelector(selector: BlockchainOuterClass.HeightSelector): HeightNumberOrTag? { + return when (selector.heightOrNumberCase) { + BlockchainOuterClass.HeightSelector.HeightOrNumberCase.HEIGHTORNUMBER_NOT_SET -> + return if (selector.height == -1L) { + Latest + } else { + Number(selector.height) + } + BlockchainOuterClass.HeightSelector.HeightOrNumberCase.NUMBER -> Number(selector.number) + BlockchainOuterClass.HeightSelector.HeightOrNumberCase.TAG -> when (selector.tag) { + BlockchainOuterClass.BlockTag.SAFE -> Safe + BlockchainOuterClass.BlockTag.LATEST -> Latest + BlockchainOuterClass.BlockTag.PENDING -> Pending + BlockchainOuterClass.BlockTag.FINALIZED -> Finalized + else -> null + } + else -> null + } + } + } + class Number(val num: Long) : HeightNumberOrTag() + object Pending : HeightNumberOrTag() + object Latest : HeightNumberOrTag() + object Safe : HeightNumberOrTag() + object Finalized : HeightNumberOrTag() + + fun getSort(): Sort { + return when (this) { + is Latest -> Sort( + compareByDescending { + it.getHead().getCurrentHeight() + }, + ) + + is Safe -> Sort.safe + is Finalized -> Sort.finalized + + else -> Sort.default + } + } + } + @JvmStatic fun convertToUpstreamFilter(selectors: List): UpstreamFilter { val matcher = selectors @@ -50,15 +96,9 @@ class Selector { SlotMatcher(it.slotHeightSelector.slotHeight) } it.hasHeightSelector() -> { - val height = if (it.heightSelector.height == -1L) { - null - } else { - it.heightSelector.height - } - if (height == null) { - empty - } else { - HeightMatcher(height) + when (val selector = HeightNumberOrTag.fromHeightSelector(it.heightSelector)) { + is HeightNumberOrTag.Number -> HeightMatcher(selector.num) + else -> empty } } else -> empty @@ -71,8 +111,8 @@ class Selector { private fun getSort(selectors: List): Sort { selectors.forEach { selector -> - if (selector.hasHeightSelector() && selector.heightSelector.height == -1L) { - return Sort(compareByDescending { it.getHead().getCurrentHeight() }) + if (selector.hasHeightSelector()) { + return HeightNumberOrTag.fromHeightSelector(selector.heightSelector)?.getSort() ?: Sort.default } else if (selector.hasLowerHeightSelector()) { return Sort( compareBy(nullsLast()) { @@ -180,6 +220,16 @@ class Selector { companion object { @JvmStatic val default = Sort(compareBy { null }) + val safe = Sort( + compareByDescending { up -> + up.getFinalizations().find { it.type == FinalizationType.SAFE_BLOCK }?.height ?: 0L + }, + ) + val finalized = Sort( + compareByDescending { up -> + up.getFinalizations().find { it.type == FinalizationType.FINALIZED_BLOCK }?.height ?: 0L + }, + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index d1ff4841..b4aca8b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import reactor.core.publisher.Flux @@ -49,6 +50,8 @@ interface Upstream : Lifecycle { fun isGrpc(): Boolean fun getLowerBounds(): Collection fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData? + fun getFinalizations(): Collection + fun addFinalization(finalization: FinalizationData, upstreamId: String) fun getUpstreamSettingsData(): UpstreamSettingsData? fun updateLowerBound(lowerBound: Long, type: LowerBoundType) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt index 0d46ed8f..1852588c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import reactor.core.Disposable @@ -81,6 +82,13 @@ open class BitcoinRpcUpstream( return null } + override fun getFinalizations(): Collection { + return emptyList() + } + + override fun addFinalization(finalization: FinalizationData, upstreamId: String) { + } + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { return null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt index e198a324..e10a1710 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -32,7 +32,6 @@ import io.emeraldpay.dshackle.reader.CompoundReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.RekeyingReader import io.emeraldpay.dshackle.reader.SpannedReader -import io.emeraldpay.dshackle.reader.TransformingReader import io.emeraldpay.dshackle.upstream.CachingReader import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -45,6 +44,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import org.apache.commons.collections4.Factory import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono @@ -92,18 +92,8 @@ open class EthereumCachingReader( SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER), ) - fun blocksByHashAsCont(): Reader> { - return CompoundReader( - SpannedReader(CacheWithUpstreamIdReader(RekeyingReader(blockHashToId, caches.getBlocksByHash())), tracer, CACHE_BLOCK_BY_HASH_READER), - SpannedReader(directReader.blockReader, tracer, DIRECT_QUORUM_RPC_READER), - ) - } - - fun blocksByHashParsed(): Reader> { - return TransformingReader( - blocksByHashAsCont(), - extractBlock, - ) + open fun blockByFinalization(): Reader> { + return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER) } open fun blocksByIdAsCont(): Reader> { @@ -117,23 +107,6 @@ open class EthereumCachingReader( ) } - open fun blocksByHeightParsed(): Reader> { - return TransformingReader( - blocksByHeightAsCont(), - extractBlock, - ) - } - - open fun txByHash(): Reader { - return TransformingReader( - CompoundReader( - CacheWithUpstreamIdReader(RekeyingReader(txHashToId, caches.getTxByHash())), - directReader.txReader, - ), - extractTx, - ) - } - open fun logsByHash(): Reader>> { return directReader.logsByHashReader } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt index 400280ce..8776eb3d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource +import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder import io.emeraldpay.dshackle.upstream.generic.GenericUpstream @@ -101,7 +102,14 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { return EthereumLowerBoundService(chain, upstream) } - override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { + override fun finalizationDetectorBuilder(): FinalizationDetector { + return EthereumFinalizationDetector() + } + + override fun upstreamSettingsDetector( + chain: Chain, + upstream: Upstream, + ): UpstreamSettingsDetector { return EthereumUpstreamSettingsDetector(upstream, chain) } 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 a07fb879..dcc90c7c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -32,6 +32,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionReceiptJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import org.apache.commons.collections4.Factory import org.apache.commons.lang3.exception.ExceptionUtils @@ -66,6 +67,7 @@ class EthereumDirectReader( val balanceReader: Reader> val receiptReader: Reader> val logsByHashReader: Reader>> + val blockByFinalizationReader: Reader> init { blockReader = object : Reader> { @@ -105,6 +107,24 @@ class EthereumDirectReader( } } } + + blockByFinalizationReader = object : Reader> { + override fun read(key: FinalizationType): Mono> { + val request = ChainRequest("eth_getBlockByNumber", ListParams(key.toBlockRef(), false)) + val tag = when (key) { + FinalizationType.FINALIZED_BLOCK -> Selector.Companion.HeightNumberOrTag.Finalized + FinalizationType.SAFE_BLOCK -> Selector.Companion.HeightNumberOrTag.Safe + else -> null + } + return readBlock( + request, + key.toString(), + Selector.empty, + tag?.getSort() ?: Selector.Sort.default, + ) + } + } + balanceReader = object : Reader> { override fun read(key: Address): Mono> { val height = up.getHead().getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest" @@ -194,8 +214,9 @@ class EthereumDirectReader( request: ChainRequest, id: String, matcher: Selector.Matcher = Selector.empty, + sort: Selector.Sort = Selector.Sort.default, ): Mono> { - return readWithQuorum(request, matcher) + return readWithQuorum(request, matcher, sort) .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Block not read $id"))) .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200))) .flatMap { result -> @@ -227,6 +248,7 @@ class EthereumDirectReader( private fun readWithQuorum( request: ChainRequest, matcher: Selector.Matcher = Selector.empty, + sort: Selector.Sort = Selector.Sort.default, ): Mono> { return Mono.just(requestReaderFactory) .map { @@ -237,7 +259,7 @@ class EthereumDirectReader( it.create( RequestReaderFactory.ReaderData( up, - Selector.UpstreamFilter(requestMatcher), + Selector.UpstreamFilter(sort, requestMatcher), callMethodsFactory.create().createQuorumFor(request.method), null, tracer, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetector.kt new file mode 100644 index 00000000..930042b6 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetector.kt @@ -0,0 +1,89 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson +import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson +import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException +import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap + +class EthereumFinalizationDetector : FinalizationDetector { + companion object { + private val log = LoggerFactory.getLogger(EthereumFinalizationDetector::class.java) + } + + val data: ConcurrentHashMap = ConcurrentHashMap() + + override fun detectFinalization( + upstream: Upstream, + blockTime: Duration, + ): Flux { + val timer = + Flux.merge( + Flux.just(1), + Flux.interval(blockTime.coerceAtLeast(Duration.ofSeconds(1)).multipliedBy(6)), + ) + return timer.flatMap { + Flux.fromIterable( + listOf( + Pair( + FinalizationType.SAFE_BLOCK, + ChainRequest( + "eth_getBlockByNumber", + ListParams("safe", false), + 1, + ), + ), + Pair( + FinalizationType.FINALIZED_BLOCK, + ChainRequest( + "eth_getBlockByNumber", + ListParams("finalized", false), + 2, + ), + ), + ), + ).flatMap { (type, req) -> + upstream + .getIngressReader() + .read(req) + .flatMap { + it.requireResult().map { result -> + val block = + Global.objectMapper + .readValue(result, BlockJson::class.java) as BlockJson? + if (block != null) { + FinalizationData(block.number, type) + } else { + throw RpcException(RpcResponseError.CODE_INVALID_JSON, "can't parse block data") + } + } + } + }.doOnNext { + addFinalization(it) + }.onErrorResume { + log.error("Error during retrieving — $it") + Flux.empty() + } + } + } + + override fun addFinalization(finalization: FinalizationData) { + data[finalization.type] = maxOf(data[finalization.type], finalization) { a, b -> + ((a?.height ?: 0) - (b?.height ?: 0)).toInt() + } ?: finalization + } + + override fun getFinalizations(): Collection { + return data.values + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt index 1f4a478d..565e3aff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -23,11 +23,12 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.LogsOracle -import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.hex.HexQuantity import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import reactor.core.publisher.Mono import reactor.kotlin.core.publisher.switchIfEmpty @@ -59,12 +60,11 @@ class EthereumLocalReader( // we do not want to serve any requests (except hardcoded) that have nonces from cache return Mono.empty() } - val common = commonRequests(key) - ?.switchIfEmpty { Mono.just(nullValue to null) } - if (common != null) { - return common.map { ChainResponse(it.first, null, it.second) } - } - return Mono.empty() + return commonRequests(key)?.switchIfEmpty { + // we need to explicitly return null to prevent executeOnRemote + // for example + Mono.just(ChainResponse(nullValue, null, null)) + } ?: Mono.empty() } /** @@ -72,7 +72,7 @@ class EthereumLocalReader( * parses JSON into Map. But the purpose of further processing and caching for some of the requests we want * to have actual data types. */ - fun commonRequests(key: ChainRequest): Mono>? { + fun commonRequests(key: ChainRequest): Mono? { val method = key.method val params = key.params if (params is ListParams) { @@ -89,7 +89,7 @@ class EthereumLocalReader( } reader.txByHashAsCont() .read(hash) - .map { it.data.json!! to it.resolvedUpstreamData } + .map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } } method == "eth_getBlockByHash" -> { @@ -106,7 +106,9 @@ class EthereumLocalReader( if (withTx) { null } else { - reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.resolvedUpstreamData } + reader.blocksByIdAsCont().read(hash).map { + ChainResponse(it.data.json, null, it.resolvedUpstreamData) + } } } @@ -126,7 +128,7 @@ class EthereumLocalReader( } reader.receipts() .read(hash) - .map { it.data to it.resolvedUpstreamData } + .map { ChainResponse(it.data, null, it.resolvedUpstreamData) } } method == "drpc_getLogsEstimate" -> { @@ -139,7 +141,7 @@ class EthereumLocalReader( return null } - fun getBlockByNumber(params: List): Mono>? { + fun getBlockByNumber(params: List): Mono? { if (params.size != 2 || params[0] == null || params[1] == null) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") } @@ -168,9 +170,17 @@ class EthereumLocalReader( blockRef == "earliest" -> { number = 0 } - blockRef == "finalized" || blockRef == "safe" || blockRef == "pending" -> { + blockRef == "pending" -> { return null } + blockRef == "finalized" || blockRef == "safe" -> { + val type = FinalizationType.fromBlockRef(blockRef) + return reader + .blockByFinalization().read(type) + .map { + ChainResponse(it.data.json, it.resolvedUpstreamData, FinalizationData(it.data.height, type)) + } + } else -> { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Block number is invalid") } @@ -180,10 +190,10 @@ class EthereumLocalReader( } return reader.blocksByHeightAsCont() - .read(number).map { it.data.json!! to it.resolvedUpstreamData } + .read(number).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) } } - fun getLogsEstimate(params: List): Mono>? { + fun getLogsEstimate(params: List): Mono? { if (logsOracle == null) { throw NotImplementedError() } @@ -234,7 +244,7 @@ class EthereumLocalReader( } return logsOracle.estimate(limit?.toLong(), fromBlock, toBlock, address, topics) - .map { it.toByteArray() to null } + .map { ChainResponse(it.toByteArray(), null, null) } } private fun parseBlockRef(blockRef: String?): Long { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationData.kt new file mode 100644 index 00000000..ed25705f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationData.kt @@ -0,0 +1,61 @@ +package io.emeraldpay.dshackle.upstream.finalization + +import io.emeraldpay.api.proto.Common + +class FinalizationData( + val height: Long, + val type: FinalizationType, +) { + + override fun toString(): String { + return "FinalizationData($height, ${type.toBlockRef()})" + } + override fun equals(other: Any?): Boolean { + return when (other) { + is FinalizationData -> other.height == height && other.type == type + else -> false + } + } +} + +enum class FinalizationType { + UNKNOWN, + SAFE_BLOCK, + FINALIZED_BLOCK, + ; + + companion object { + fun fromBlockRef(v: String): FinalizationType { + return when (v) { + "safe" -> SAFE_BLOCK + "finalized" -> FINALIZED_BLOCK + else -> UNKNOWN + } + } + } + + fun toProtoFinalizationType(): Common.FinalizationType { + return when (this) { + FINALIZED_BLOCK -> Common.FinalizationType.FINALIZATION_FINALIZED_BLOCK + UNKNOWN -> Common.FinalizationType.UNRECOGNIZED + SAFE_BLOCK -> Common.FinalizationType.FINALIZATION_SAFE_BLOCK + } + } + + fun toBlockRef(): String { + return when (this) { + FINALIZED_BLOCK -> "finalized" + SAFE_BLOCK -> "safe" + UNKNOWN -> "unknown" + } + } +} + +fun Common.FinalizationType.fromProtoType(): FinalizationType { + return when (this) { + Common.FinalizationType.FINALIZATION_UNSPECIFIED -> FinalizationType.UNKNOWN + Common.FinalizationType.FINALIZATION_SAFE_BLOCK -> FinalizationType.SAFE_BLOCK + Common.FinalizationType.FINALIZATION_FINALIZED_BLOCK -> FinalizationType.FINALIZED_BLOCK + Common.FinalizationType.UNRECOGNIZED -> FinalizationType.UNKNOWN + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationDetector.kt new file mode 100644 index 00000000..b4aa6ff2 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationDetector.kt @@ -0,0 +1,32 @@ +package io.emeraldpay.dshackle.upstream.finalization + +import io.emeraldpay.dshackle.upstream.Upstream +import reactor.core.publisher.Flux +import java.time.Duration + +interface FinalizationDetector { + fun detectFinalization( + upstream: Upstream, + blockTime: Duration, + ): Flux + + fun getFinalizations(): Collection + + fun addFinalization(finalization: FinalizationData) +} + +class NoopFinalizationDetector : FinalizationDetector { + override fun detectFinalization( + upstream: Upstream, + blockTime: Duration, + ): Flux { + return Flux.empty() + } + + override fun getFinalizations(): Collection { + return emptyList() + } + + override fun addFinalization(finalization: FinalizationData) { + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt index da7c6855..5f7c5363 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt @@ -20,12 +20,13 @@ import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions +import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector +import io.emeraldpay.dshackle.upstream.finalization.NoopFinalizationDetector import org.springframework.cloud.sleuth.Tracer import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler abstract class AbstractChainSpecific : ChainSpecific { - override fun localReaderBuilder( cachingReader: CachingReader, methods: CallMethods, @@ -35,11 +36,18 @@ abstract class AbstractChainSpecific : ChainSpecific { return Mono.just(LocalReader(methods)) } + override fun finalizationDetectorBuilder(): FinalizationDetector { + return NoopFinalizationDetector() + } + override fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder { return { _, _, _ -> NoopCachingReader } } - override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector? { + override fun upstreamSettingsDetector( + chain: Chain, + upstream: Upstream, + ): UpstreamSettingsDetector? { return null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index 6d05f9ce..ec852d43 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -32,6 +32,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallSelector import io.emeraldpay.dshackle.upstream.cosmos.CosmosChainSpecific import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions +import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService import io.emeraldpay.dshackle.upstream.near.NearChainSpecific import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific @@ -45,6 +46,7 @@ import reactor.core.scheduler.Scheduler typealias SubscriptionBuilder = (Multistream) -> EgressSubscription typealias LocalReaderBuilder = (CachingReader, CallMethods, Head, LogsOracle?) -> Mono typealias CachingReaderBuilder = (Multistream, Caches, Factory) -> CachingReader +typealias FinalizationDetectorBuilder = () -> FinalizationDetector interface ChainSpecific { fun getFromHeader(data: ByteArray, upstreamId: String, api: ChainReader): Mono @@ -55,7 +57,14 @@ interface ChainSpecific { fun unsubscribeNewHeadsRequest(subId: String): ChainRequest - fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head, logsOracle: LogsOracle?): Mono + fun finalizationDetectorBuilder(): FinalizationDetector + + fun localReaderBuilder( + cachingReader: CachingReader, + methods: CallMethods, + head: Head, + logsOracle: LogsOracle?, + ): Mono fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index fb379089..ca1863ee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.UpstreamValidatorBuilder import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData @@ -43,11 +44,12 @@ open class GenericUpstream( role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, private val node: QuorumForLabels.QuorumItem?, - chainConfig: ChainsConfig.ChainConfig, + private val chainConfig: ChainsConfig.ChainConfig, connectorFactory: ConnectorFactory, validatorBuilder: UpstreamValidatorBuilder, upstreamSettingsDetectorBuilder: UpstreamSettingsDetectorBuilder, lowerBoundServiceBuilder: LowerBoundServiceBuilder, + finalizationDetectorBuilder: FinalizationDetectorBuilder, ) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig, chain), Lifecycle { constructor( @@ -63,7 +65,8 @@ open class GenericUpstream( upstreamRpcModulesDetectorBuilder: UpstreamRpcModulesDetectorBuilder, buildMethods: (UpstreamsConfig.Upstream<*>, Chain) -> CallMethods, lowerBoundServiceBuilder: LowerBoundServiceBuilder, - ) : this(config.id!!, chain, hash, options, config.role, buildMethods(config, chain), node, chainConfig, connectorFactory, validatorBuilder, upstreamSettingsDetectorBuilder, lowerBoundServiceBuilder) { + finalizationDetectorBuilder: FinalizationDetectorBuilder, + ) : this(config.id!!, chain, hash, options, config.role, buildMethods(config, chain), node, chainConfig, connectorFactory, validatorBuilder, upstreamSettingsDetectorBuilder, lowerBoundServiceBuilder, finalizationDetectorBuilder) { rpcModulesDetector = upstreamRpcModulesDetectorBuilder(this) detectRpcModules(config, buildMethods) } @@ -85,6 +88,9 @@ open class GenericUpstream( private val isUpstreamValid = AtomicBoolean(false) private val clientVersion = AtomicReference(UNKNOWN_CLIENT_VERSION) + private val finalizationDetector = finalizationDetectorBuilder() + private var finalizationDetectorSubscription: Disposable? = null + override fun getHead(): Head { return connector.getHead() } @@ -255,6 +261,8 @@ open class GenericUpstream( detectSettings() detectLowerBlock() + + detectFinalization() } override fun stop() { @@ -272,6 +280,8 @@ open class GenericUpstream( livenessSubscription = null lowerBlockDetectorSubscription?.dispose() lowerBlockDetectorSubscription = null + finalizationDetectorSubscription?.dispose() + finalizationDetectorSubscription = null connector.getHead().stop() } @@ -282,13 +292,31 @@ open class GenericUpstream( } } - private fun detectLowerBlock() { - lowerBlockDetectorSubscription = lowerBoundService.detectLowerBounds() - .subscribe { + override fun getFinalizations(): Collection { + return finalizationDetector.getFinalizations() + } + + override fun addFinalization(finalization: FinalizationData, upstreamId: String) { + if (getId() == upstreamId) { + finalizationDetector.addFinalization(finalization) + } + } + + private fun detectFinalization() { + finalizationDetectorSubscription = + finalizationDetector.detectFinalization(this, chainConfig.expectedBlockTime).subscribe { sendUpstreamStateEvent(UPDATED) } } + private fun detectLowerBlock() { + lowerBlockDetectorSubscription = + lowerBoundService.detectLowerBounds() + .subscribe { + sendUpstreamStateEvent(UPDATED) + } + } + fun getIngressSubscription(): IngressSubscription { return connector.getIngressSubscription() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 42c4b5e1..2bb76356 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -36,6 +36,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType @@ -155,6 +156,13 @@ class BitcoinGrpcUpstream( return emptyList() } + override fun getFinalizations(): Collection { + return emptyList() + } + + override fun addFinalization(finalization: FinalizationData, upstreamId: String) { + } + override fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData? { return null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt index 73633b00..9a620ed6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt @@ -35,6 +35,9 @@ import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType +import io.emeraldpay.dshackle.upstream.finalization.fromProtoType import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType @@ -74,52 +77,65 @@ open class GenericGrpcUpstream( ), GrpcUpstream, Lifecycle { - - private val blockConverter: Function = Function { value -> - val parentHash = - if (value.parentBlockId.isBlank()) { - null - } else { - BlockId.from(BlockHash.from("0x" + value.parentBlockId)) - } - val block = BlockContainer( - value.height, - BlockId.from(BlockHash.from("0x" + value.blockId)), - BigInteger(1, value.weight.toByteArray()), - Instant.ofEpochMilli(value.timestamp), - false, - null, - null, - parentHash, - ) - val lowerBounds = value.lowerBoundsList - .map { LowerBoundData(it.lowerBoundValue, it.lowerBoundTimestamp, it.lowerBoundType.fromProtoType()) } - GrpcHead.GrpcHeadData(block, lowerBounds) - } + private val blockConverter: Function = + Function { value -> + val parentHash = + if (value.parentBlockId.isBlank()) { + null + } else { + BlockId.from(BlockHash.from("0x" + value.parentBlockId)) + } + val block = + BlockContainer( + value.height, + BlockId.from(BlockHash.from("0x" + value.blockId)), + BigInteger(1, value.weight.toByteArray()), + Instant.ofEpochMilli(value.timestamp), + false, + null, + null, + parentHash, + ) + val lowerBounds = + value.lowerBoundsList + .map { LowerBoundData(it.lowerBoundValue, it.lowerBoundTimestamp, it.lowerBoundType.fromProtoType()) } + val finalizationData = + value.finalizationDataList.map { + FinalizationData(it.height, it.type.fromProtoType()) + } + GrpcHead.GrpcHeadData(block, lowerBounds, finalizationData) + } private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) - private val grpcHead = GrpcHead( - getId(), - chain, - this, - remote, - blockConverter, - null, - NoChoiceWithPriorityForkChoice(nodeRating, parentId), - headScheduler, - ) + private val grpcHead = + GrpcHead( + getId(), + chain, + this, + remote, + blockConverter, + null, + NoChoiceWithPriorityForkChoice(nodeRating, parentId), + headScheduler, + ) private var capabilities: Set = emptySet() private val buildInfo: BuildInfo = BuildInfo() private val defaultReader: ChainReader = client.getReader() private val lowerBounds = ConcurrentHashMap() + private var finalizationData = ConcurrentHashMap() override fun start() { - grpcHead.lowerBoundsFlux() - .publishOn(lowerBoundScheduler) - .subscribe { - lowerBounds[it.type] = it + grpcHead.rawDataFlux() + .publishOn(rawDataScheduler) + .subscribe { head -> + head.lowerBounds.forEach { + lowerBounds[it.type] = it + } + head.finalizationData.forEach { + finalizationData[it.type] = it + } sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.UPDATED) } } @@ -205,6 +221,14 @@ open class GenericGrpcUpstream( return lowerBounds[lowerBoundType] } + override fun getFinalizations(): Collection { + return finalizationData.values + } + + override fun addFinalization(finalization: FinalizationData, upstreamId: String) { + finalizationData[finalization.type] = finalization + } + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { return Upstream.UpstreamSettingsData( nodeId(), @@ -214,8 +238,9 @@ open class GenericGrpcUpstream( } companion object { - val lowerBoundScheduler: Scheduler = Schedulers.fromExecutorService( - Executors.newFixedThreadPool(4, CustomizableThreadFactory("grpc-lower-bound-")), - ) + val rawDataScheduler: Scheduler = + Schedulers.fromExecutorService( + Executors.newFixedThreadPool(4, CustomizableThreadFactory("grpc-raw-data-bound-")), + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index b0581a50..493c2486 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.micrometer.core.instrument.Counter @@ -54,10 +55,9 @@ class GrpcHead( private val forkChoice: ForkChoice, headScheduler: Scheduler, ) : AbstractHead(forkChoice, headScheduler, upstreamId = id), Lifecycle { - private var headSubscription: Disposable? = null - private val lowerBoundsSink = Sinks.many().multicast().directBestEffort() + private val rawDataSink = Sinks.many().multicast().directBestEffort() /** * Initiate a new head subscription with connection to the remote @@ -92,14 +92,15 @@ class GrpcHead( log.warn("Head subscription finished: $it") } - var blocks = heads.map(converter) - .doOnNext { - it.lowerBounds.forEach { bound -> lowerBoundsSink.tryEmitNext(bound) } - } - .map { it.block } - .distinctUntilChanged { - it.hash - }.filter { forkChoice.filter(it) } + var blocks = + heads.map(converter) + .doOnNext { + rawDataSink.tryEmitNext(it) + } + .map { it.block } + .distinctUntilChanged { + it.hash + }.filter { forkChoice.filter(it) } if (enhancer != null) { blocks = blocks.flatMap(enhancer) @@ -133,7 +134,7 @@ class GrpcHead( headSubscription?.dispose() } - fun lowerBoundsFlux(): Flux = lowerBoundsSink.asFlux() + fun rawDataFlux(): Flux = rawDataSink.asFlux() val headsCounter = Counter.builder("grpc_head_received") .tag("upstream", id) @@ -143,7 +144,8 @@ class GrpcHead( data class GrpcHeadData( val block: BlockContainer, val lowerBounds: List, + val finalizationData: List, ) { - constructor(block: BlockContainer) : this(block, emptyList()) + constructor(block: BlockContainer) : this(block, emptyList(), emptyList()) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy index c0a51d44..758519d7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy @@ -108,7 +108,7 @@ class ApiReaderMock implements Reader { } error = new ChainCallError(-32601, "Method ${request.method} with ${request.params} is not mocked") } - return new ChainResponse(result, error, ChainResponse.Id.from(request.id), null, null, null) + return new ChainResponse(result, error, ChainResponse.Id.from(request.id), null, null, null, null) } as Callable return Mono.fromCallable(call) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy index 30a51bec..fdfd9a77 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy @@ -77,6 +77,7 @@ class GenericUpstreamMock extends GenericUpstream { io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&validator, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&upstreamSettingsDetector, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&lowerBoundService, + io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&finalizationDetectorBuilder, ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 680a6de2..bf126a0a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -77,7 +77,8 @@ class FilteredApisSpec extends Specification { connectorFactory, cs.&validator, cs.&upstreamSettingsDetector, - cs.&lowerBoundService + cs.&lowerBoundService, + cs.&finalizationDetectorBuilder ) } def matcher = new Selector.LabelMatcher("test", ["foo"]) 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 9a7aaef9..d9be8262 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -12,6 +12,7 @@ import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.ethereum.domain.Address import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash @@ -33,6 +34,40 @@ class EthereumDirectReaderSpec extends Specification { String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String address1 = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d" Upstream.UpstreamSettingsData data = new Upstream.UpstreamSettingsData("test") + def "Reads block by finalization"() { + setup: + def json = new BlockJson().tap { + number = 100 + hash = BlockHash.from(hash1) + timestamp = Instant.now() + totalDifficulty = BigInteger.ONE + parentHash = BlockHash.from(hash1) + transactions = [] + } + def calls = Mock(Factory) { + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + } + EthereumDirectReader reader = new EthereumDirectReader( + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + ) + reader.requestReaderFactory = Mock(RequestReaderFactory) { + 1 * create({ it.upstreamFilter.sort == Selector.Sort.safe }) >> Mock(RequestReader) { + 1 * read(new ChainRequest("eth_getBlockByNumber", new ListParams(["safe", false]))) >> Mono.just( + new RequestReader.Result( + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null) + ) + } + } + when: + def act = reader.blockByFinalizationReader.read(FinalizationType.SAFE_BLOCK) + then: + StepVerifier.create(act) + .expectNextMatches { block -> + block.data.hash.toHexWithPrefix() == hash1 + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } def "Reads block by hash"() { setup: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy index 239a5cb8..39ab8acb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy @@ -10,6 +10,8 @@ import io.emeraldpay.dshackle.upstream.EmptyHead import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import org.apache.commons.collections4.functors.ConstantFactory @@ -84,8 +86,8 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.first.length > 0 - with(Global.objectMapper.readValue(it.first, BlockJson)) { + it.result.length > 0 + with(Global.objectMapper.readValue(it.result, BlockJson)) { number == 101 } } @@ -112,8 +114,8 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.first.length > 0 - with(Global.objectMapper.readValue(it.first, BlockJson)) { + it.result.length > 0 + with(Global.objectMapper.readValue(it.result, BlockJson)) { number == 0 } } @@ -140,13 +142,43 @@ class EthereumLocalReaderSpec extends Specification { then: act != null with(act.block()) { - it.first.length > 0 - with(Global.objectMapper.readValue(it.first, BlockJson)) { + it.result.length > 0 + with(Global.objectMapper.readValue(it.result, BlockJson)) { number == 74735 } } } + def "getBlockByNumber fetches the block by tag"() { + setup: + def head = Stub(Head) {} + def reader = Mock(EthereumCachingReader) { + 1 * blockByFinalization() >> Mock(Reader) { + 1 * read(FinalizationType.SAFE_BLOCK) >> Mono.just( + new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), null) + ) + } + } + def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + def router = new EthereumLocalReader(reader, methods, head, null) + + when: + def act = router.read( + new ChainRequest("eth_getBlockByNumber", + new ListParams("safe", false)) + ) + + then: + act != null + with(act.block()) { + it.result.length > 0 + with(Global.objectMapper.readValue(it.result, BlockJson)) { + number == 74735 + } + it.finalization == new FinalizationData(74735, FinalizationType.SAFE_BLOCK) + } + } + def "getBlockByNumber skips requests with tx bodies"() { setup: def head = Mock(Head) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index 36ba9445..d0260edc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -60,7 +60,7 @@ class GrpcHeadSpec extends Specification { } }) def convert = { BlockchainOuterClass.ChainHead head -> - new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of()) + new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of(), List.of()) } def head = new GrpcHead( "test", @@ -127,7 +127,7 @@ class GrpcHeadSpec extends Specification { } }) def convert = { BlockchainOuterClass.ChainHead head -> - new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of()) + new GrpcHead.GrpcHeadData(TestingCommons.blockForBitcoin(head.height), List.of(), List.of()) } def head = new GrpcHead( "test", diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/SelectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/SelectorTest.kt index 09887da9..a4edb3bc 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/SelectorTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/SelectorTest.kt @@ -1,6 +1,10 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.BlockchainOuterClass.BlockTag +import io.emeraldpay.api.proto.BlockchainOuterClass.HeightSelector +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType import org.junit.jupiter.api.Assertions.assertEquals @@ -53,6 +57,94 @@ class SelectorTest { ) } + @ParameterizedTest + @MethodSource("finalData") + fun `sort with finalization`( + finalizationType: FinalizationType, + finalizationProto: BlockchainOuterClass.BlockTag, + ) { + val up1 = mock { + on { getFinalizations() } doReturn listOf(FinalizationData(1L, finalizationType)) + } + val up2 = mock { + on { getFinalizations() } doReturn listOf(FinalizationData(10L, finalizationType)) + } + val up3 = mock { + on { getFinalizations() } doReturn listOf(FinalizationData(100L, finalizationType)) + } + val up4 = mock { + on { getFinalizations() } doReturn listOf() + } + val ups = listOf(up4, up3, up2, up1) + val requestSelectors = listOf( + BlockchainOuterClass.Selector.newBuilder() + .setHeightSelector( + HeightSelector.newBuilder() + .setTag(finalizationProto), + ) + .build(), + ) + + val upstreamFilter = Selector.convertToUpstreamFilter(requestSelectors) + + val actual = ups.sortedWith(upstreamFilter.sort.comparator) + + assertEquals( + listOf(up3, up2, up1, up4), + actual, + ) + } + + @Test + fun `sort with latest`() { + val mockHead1 = mock { + on { getCurrentHeight() } doReturn 1L + } + val up1 = mock { + on { getHead() } doReturn mockHead1 + } + + val mockHead2 = mock { + on { getCurrentHeight() } doReturn 2L + } + val up2 = mock { + on { getHead() } doReturn mockHead2 + } + + val mockHead3 = mock { + on { getCurrentHeight() } doReturn 3L + } + val up3 = mock { + on { getHead() } doReturn mockHead3 + } + + val mockHead4 = mock { + on { getCurrentHeight() } doReturn null + } + val up4 = mock { + on { getHead() } doReturn mockHead4 + } + + val ups = listOf(up2, up1, up4, up3) + val requestSelectors = listOf( + BlockchainOuterClass.Selector.newBuilder() + .setHeightSelector( + HeightSelector.newBuilder() + .setTag(BlockTag.LATEST), + ) + .build(), + ) + + val upstreamFilter = Selector.convertToUpstreamFilter(requestSelectors) + + val actual = ups.sortedWith(upstreamFilter.sort.comparator) + + assertEquals( + listOf(up3, up2, up1, up4), + actual, + ) + } + @Test fun `preserve the same order if no lower bound type`() { val up1 = mock { @@ -127,5 +219,12 @@ class SelectorTest { of(LowerBoundType.BLOCK, BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOCK), of(LowerBoundType.SLOT, BlockchainOuterClass.LowerBoundType.LOWER_BOUND_SLOT), ) + + @JvmStatic + fun finalData(): List = + listOf( + of(FinalizationType.SAFE_BLOCK, BlockTag.SAFE), + of(FinalizationType.FINALIZED_BLOCK, BlockTag.FINALIZED), + ) } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt new file mode 100644 index 00000000..6e5c210a --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt @@ -0,0 +1,75 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.reader.ChainReader +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson +import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import org.junit.jupiter.api.Assertions +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.mockito.Mockito.mock +import org.mockito.Mockito.`when` +import reactor.core.publisher.Mono +import java.time.Duration +import java.time.Instant + +class EthereumFinalizationDetectorTest { + + private lateinit var upstream: Upstream + private lateinit var chainReader: ChainReader + private lateinit var detector: EthereumFinalizationDetector + + @BeforeEach + fun setUp() { + upstream = mock() + chainReader = mock() + `when`(upstream.getIngressReader()).thenReturn(chainReader) + detector = EthereumFinalizationDetector() + } + + @Test + fun testDetectFinalization() { + `when`(chainReader.read(ChainRequest("eth_getBlockByNumber", ListParams("safe", false), 1))) + .thenReturn( + Mono.just( + ChainResponse( + Global.objectMapper.writeValueAsString( + BlockJson().apply { + number = 1 + timestamp = Instant.now() + }, + ).toByteArray(), + null, + ), + ), + ) + `when`(chainReader.read(ChainRequest("eth_getBlockByNumber", ListParams("finalized", false), 2))) + .thenReturn( + Mono.just( + ChainResponse( + Global.objectMapper.writeValueAsString( + BlockJson().apply { + number = 2 + timestamp = Instant.now() + }, + ).toByteArray(), + null, + ), + ), + ) + + val flux = detector.detectFinalization(upstream, Duration.ofMillis(200)) + flux.take(2).collectList().block() + val result = detector.getFinalizations().toList() + Assertions.assertEquals(2, result.size) + org.assertj.core.api.Assertions.assertThat(result) + .contains(FinalizationData(2L, FinalizationType.FINALIZED_BLOCK)) + .contains(FinalizationData(1L, FinalizationType.SAFE_BLOCK)) + } +} diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstreamTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstreamTest.kt new file mode 100644 index 00000000..2e97562d --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstreamTest.kt @@ -0,0 +1,125 @@ +package io.emeraldpay.dshackle.upstream.grpc + +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainGrpc +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.Chain +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.test.MockGrpcServerKt +import io.emeraldpay.dshackle.upstream.finalization.FinalizationData +import io.emeraldpay.dshackle.upstream.finalization.FinalizationType +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData +import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient +import io.grpc.stub.StreamObserver +import org.junit.jupiter.api.Assertions +import org.junit.jupiter.api.Test +import org.mockito.Mockito +import reactor.core.publisher.Sinks +import reactor.core.scheduler.Schedulers + +class GenericGrpcUpstreamTest { + private val parentId = "testParent" + private val hash: Byte = 0x01 + private val role = UpstreamsConfig.UpstreamRole.PRIMARY + private val headSink = Sinks.many().multicast().directBestEffort() + private val remote = + MockGrpcServerKt().clientForServer( + object : BlockchainGrpc.BlockchainImplBase() { + override fun subscribeHead( + request: Common.Chain, + responseObserver: StreamObserver, + ) { + Thread { + headSink.asFlux().subscribe { data -> + responseObserver.onNext(data) + Thread.sleep(500) + } + }.start() + } + }, + ) + private val client = Mockito.mock(JsonRpcGrpcClient::class.java) + private val nodeRating = 5 + private val overrideLabels = Mockito.mock(UpstreamsConfig.Labels::class.java) + private val headScheduler = Schedulers.single() + + private fun getUpstream(): GrpcUpstream { + return GenericGrpcUpstream( + parentId, + hash, + role, + Chain.LINEA__MAINNET, + remote, + client, + nodeRating, + overrideLabels, + ChainsConfig.ChainConfig.default(), + headScheduler, + ) + } + + @Test + fun start() { + val up = getUpstream() + up.getHead().start() + up.start() + headSink.emitNext( + BlockchainOuterClass.ChainHead.newBuilder() + .setChain(Common.ChainRef.CHAIN_LINEA__MAINNET) + .setHeight(10L) + .setWeight(ByteString.EMPTY) + .setBlockId("a2622ec25e883dd13c1091c18ba717a1a794713baa77b8e68ec6a993045cb50f") + .setTimestamp(0) + .addAllFinalizationData( + mutableListOf( + Common + .FinalizationData + .newBuilder() + .setType(Common.FinalizationType.FINALIZATION_FINALIZED_BLOCK) + .setHeight(8L) + .build(), + ), + ) + .addAllLowerBounds( + mutableListOf( + BlockchainOuterClass.LowerBound + .newBuilder() + .setLowerBoundType(BlockchainOuterClass.LowerBoundType.LOWER_BOUND_TX) + .setLowerBoundTimestamp(0) + .setLowerBoundValue(1L).build(), + ), + ) + .build(), + ) { _, _ -> + true + } + Thread.sleep(100) + Assertions.assertEquals(1, up.getFinalizations().size) + Assertions.assertTrue( + up.getFinalizations() + .contains(FinalizationData(8L, FinalizationType.FINALIZED_BLOCK)), + ) + Assertions.assertTrue( + up.getLowerBounds() + .contains(LowerBoundData(1L, 0L, LowerBoundType.TX)), + ) + } + + @Test + fun getFinalizations() { + val up = getUpstream() + val finalizationData1 = FinalizationData(100L, FinalizationType.FINALIZED_BLOCK) + + val finalizationData2 = FinalizationData(200L, FinalizationType.FINALIZED_BLOCK) + + up.addFinalization(finalizationData1, "upstream1") + up.addFinalization(finalizationData2, "upstream2") + + val finalizations = up.getFinalizations() + Assertions.assertEquals(1, finalizations.size) + Assertions.assertTrue(finalizations.contains(finalizationData2)) + } +} diff --git a/src/test/kotlin/test/MockGrpcServerKt.kt b/src/test/kotlin/test/MockGrpcServerKt.kt new file mode 100644 index 00000000..09b5d895 --- /dev/null +++ b/src/test/kotlin/test/MockGrpcServerKt.kt @@ -0,0 +1,40 @@ +/** + * Copyright (c) 2019 ETCDEV GmbH + * Copyright (c) 2020 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.test + +import io.emeraldpay.api.proto.BlockchainGrpc +import io.emeraldpay.api.proto.ReactorBlockchainGrpc +import io.grpc.inprocess.InProcessChannelBuilder +import io.grpc.inprocess.InProcessServerBuilder +import io.grpc.testing.GrpcCleanupRule +import org.junit.Rule + +class MockGrpcServerKt { + + @get:Rule + val grpcCleanup = GrpcCleanupRule() + + fun clientForServer(impl: BlockchainGrpc.BlockchainImplBase): ReactorBlockchainGrpc.ReactorBlockchainStub { + val serverName = InProcessServerBuilder.generateName() + grpcCleanup.register( + InProcessServerBuilder + .forName(serverName).directExecutor().addService(impl).build().start(), + ) + val channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()) + return ReactorBlockchainGrpc.newReactorStub(channel) + } +}