StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Track upstream finalization data (#499)

Browse Source
This commit is contained in:
Vyacheslav
2024-06-11 14:49:32 +03:00
committed by GitHub
parent 3e898cf325
commit effe881091
32 changed files with 933 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
emerald-grpc

Submodule emerald-grpc updated: a3351cdcb5...421724301b

23
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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<ParsedCallDetails>?,
val stream: Flux<Chunk>? = 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<ParsedCallDetails>?, 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<ParsedCallDetails>?, stream: Flux<Chunk>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream)
}

22
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -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<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
@@ -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())

1
src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt
View File

@@ -84,6 +84,7 @@ open class GenericUpstreamCreator(
cs::upstreamRpcModulesDetector,
buildMethodsFun,
cs::lowerBoundService,
cs::finalizationDetectorBuilder,
)
upstream.start()

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt
View File

@@ -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<Chunk>, 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()

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -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<FinalizationData> {
return getAll().flatMap { it.getFinalizations() }
.fold(mutableMapOf<FinalizationType, Long>()) { 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<LowerBoundData> {
return lowerBounds.values
}

72
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -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<BlockchainOuterClass.Selector>): 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<BlockchainOuterClass.Selector>): 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
},
)
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -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<LowerBoundData>
fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData?
fun getFinalizations(): Collection<FinalizationData>
fun addFinalization(finalization: FinalizationData, upstreamId: String)
fun getUpstreamSettingsData(): UpstreamSettingsData?
fun updateLowerBound(lowerBound: Long, type: LowerBoundType)

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
View File

@@ -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<FinalizationData> {
return emptyList()
}
override fun addFinalization(finalization: FinalizationData, upstreamId: String) {
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {
return null
}

33
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
View File

@@ -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<BlockHash, Result<BlockContainer>> {
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<BlockHash, BlockJson<TransactionRefJson>> {
return TransformingReader(
blocksByHashAsCont(),
extractBlock,
)
open fun blockByFinalization(): Reader<FinalizationType, Result<BlockContainer>> {
return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER)
}
open fun blocksByIdAsCont(): Reader<BlockId, Result<BlockContainer>> {
@@ -117,23 +107,6 @@ open class EthereumCachingReader(
)
}
open fun blocksByHeightParsed(): Reader<Long, BlockJson<TransactionRefJson>> {
return TransformingReader(
blocksByHeightAsCont(),
extractBlock,
)
}
open fun txByHash(): Reader<TransactionId, TransactionJsonSnapshot> {
return TransformingReader(
CompoundReader(
CacheWithUpstreamIdReader(RekeyingReader(txHashToId, caches.getTxByHash())),
directReader.txReader,
),
extractTx,
)
}
open fun logsByHash(): Reader<BlockId, Result<List<TransactionLogJson>>> {
return directReader.logsByHashReader
}

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt
View File

@@ -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)
}

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

@@ -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<Address, Result<Wei>>
val receiptReader: Reader<TransactionId, Result<ByteArray>>
val logsByHashReader: Reader<BlockId, Result<List<TransactionLogJson>>>
val blockByFinalizationReader: Reader<FinalizationType, Result<BlockContainer>>
init {
blockReader = object : Reader<BlockHash, Result<BlockContainer>> {
@@ -105,6 +107,24 @@ class EthereumDirectReader(
}
}
}
blockByFinalizationReader = object : Reader<FinalizationType, Result<BlockContainer>> {
override fun read(key: FinalizationType): Mono<Result<BlockContainer>> {
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<Address, Result<Wei>> {
override fun read(key: Address): Mono<Result<Wei>> {
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<Result<BlockContainer>> {
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<Result<ByteArray>> {
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,

89
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetector.kt Normal file
View File

@@ -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<FinalizationType, FinalizationData> = ConcurrentHashMap()
override fun detectFinalization(
upstream: Upstream,
blockTime: Duration,
): Flux<FinalizationData> {
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<TransactionRefJson>?
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<FinalizationData> {
return data.values
}
}

42
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
View File

@@ -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<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
fun commonRequests(key: ChainRequest): Mono<ChainResponse>? {
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<Any?>): Mono<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
fun getBlockByNumber(params: List<Any?>): Mono<ChainResponse>? {
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<Any?>): Mono<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
fun getLogsEstimate(params: List<Any?>): Mono<ChainResponse>? {
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 {

61
src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationData.kt Normal file
View File

@@ -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
}
}

32
src/main/kotlin/io/emeraldpay/dshackle/upstream/finalization/FinalizationDetector.kt Normal file
View File

@@ -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<FinalizationData>
fun getFinalizations(): Collection<FinalizationData>
fun addFinalization(finalization: FinalizationData)
}
class NoopFinalizationDetector : FinalizationDetector {
override fun detectFinalization(
upstream: Upstream,
blockTime: Duration,
): Flux<FinalizationData> {
return Flux.empty()
}
override fun getFinalizations(): Collection<FinalizationData> {
return emptyList()
}
override fun addFinalization(finalization: FinalizationData) {
}
}

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt
View File

@@ -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
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
View File

@@ -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<ChainReader>
typealias CachingReaderBuilder = (Multistream, Caches, Factory<CallMethods>) -> CachingReader
typealias FinalizationDetectorBuilder = () -> FinalizationDetector
interface ChainSpecific {
fun getFromHeader(data: ByteArray, upstreamId: String, api: ChainReader): Mono<BlockContainer>
@@ -55,7 +57,14 @@ interface ChainSpecific {
fun unsubscribeNewHeadsRequest(subId: String): ChainRequest
fun localReaderBuilder(cachingReader: CachingReader, methods: CallMethods, head: Head, logsOracle: LogsOracle?): Mono<ChainReader>
fun finalizationDetectorBuilder(): FinalizationDetector
fun localReaderBuilder(
cachingReader: CachingReader,
methods: CallMethods,
head: Head,
logsOracle: LogsOracle?,
): Mono<ChainReader>
fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription

38
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
View File

@@ -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<FinalizationData> {
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()
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -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<FinalizationData> {
return emptyList()
}
override fun addFinalization(finalization: FinalizationData, upstreamId: String) {
}
override fun getLowerBound(lowerBoundType: LowerBoundType): LowerBoundData? {
return null
}

103
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt
View File

@@ -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<BlockchainOuterClass.ChainHead, GrpcHead.GrpcHeadData> = 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<BlockchainOuterClass.ChainHead, GrpcHead.GrpcHeadData> =
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<Capability> = emptySet()
private val buildInfo: BuildInfo = BuildInfo()
private val defaultReader: ChainReader = client.getReader()
private val lowerBounds = ConcurrentHashMap<LowerBoundType, LowerBoundData>()
private var finalizationData = ConcurrentHashMap<FinalizationType, FinalizationData>()
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<FinalizationData> {
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-")),
)
}
}

26
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
View File

@@ -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<LowerBoundData>()
private val rawDataSink = Sinks.many().multicast().directBestEffort<GrpcHeadData>()
/**
* 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<LowerBoundData> = lowerBoundsSink.asFlux()
fun rawDataFlux(): Flux<GrpcHeadData> = 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<LowerBoundData>,
val finalizationData: List<FinalizationData>,
) {
constructor(block: BlockContainer) : this(block, emptyList())
constructor(block: BlockContainer) : this(block, emptyList(), emptyList())
}
}

2
src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy
View File

@@ -108,7 +108,7 @@ class ApiReaderMock implements Reader<ChainRequest, ChainResponse> {
}
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<ChainResponse>
return Mono.fromCallable(call)
}

1
src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy
View File

@@ -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)

3
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -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"])

35
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy
View File

@@ -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:

44
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
View File

@@ -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)

4
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy
View File

@@ -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",

99
src/test/kotlin/io/emeraldpay/dshackle/upstream/SelectorTest.kt
View File

@@ -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<Upstream> {
on { getFinalizations() } doReturn listOf(FinalizationData(1L, finalizationType))
}
val up2 = mock<Upstream> {
on { getFinalizations() } doReturn listOf(FinalizationData(10L, finalizationType))
}
val up3 = mock<Upstream> {
on { getFinalizations() } doReturn listOf(FinalizationData(100L, finalizationType))
}
val up4 = mock<Upstream> {
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<Head> {
on { getCurrentHeight() } doReturn 1L
}
val up1 = mock<Upstream> {
on { getHead() } doReturn mockHead1
}
val mockHead2 = mock<Head> {
on { getCurrentHeight() } doReturn 2L
}
val up2 = mock<Upstream> {
on { getHead() } doReturn mockHead2
}
val mockHead3 = mock<Head> {
on { getCurrentHeight() } doReturn 3L
}
val up3 = mock<Upstream> {
on { getHead() } doReturn mockHead3
}
val mockHead4 = mock<Head> {
on { getCurrentHeight() } doReturn null
}
val up4 = mock<Upstream> {
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<Upstream> {
@@ -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<Arguments> =
listOf(
of(FinalizationType.SAFE_BLOCK, BlockTag.SAFE),
of(FinalizationType.FINALIZED_BLOCK, BlockTag.FINALIZED),
)
}
}

75
src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt Normal file
View File

@@ -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<TransactionRefJson>().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<TransactionRefJson>().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))
}
}

125
src/test/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstreamTest.kt Normal file
View File

@@ -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<BlockchainOuterClass.ChainHead>()
private val remote =
MockGrpcServerKt().clientForServer(
object : BlockchainGrpc.BlockchainImplBase() {
override fun subscribeHead(
request: Common.Chain,
responseObserver: StreamObserver<BlockchainOuterClass.ChainHead>,
) {
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))
}
}

40
src/test/kotlin/test/MockGrpcServerKt.kt Normal file
View File

@@ -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)
}
}