diff --git a/emerald-grpc b/emerald-grpc index 398c3dd5..ca4852a4 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 398c3dd5b47001751c9b6ab462ae1a7d6c870958 +Subproject commit ca4852a4a439f270ee872f46f558958bcdb0ad5d diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt index 2d079aa9..a7f167df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt @@ -208,7 +208,7 @@ class QuorumRequestReader( private fun setupDefaultResult(key: ChainRequest): Mono { return Mono.just(quorum).flatMap { q -> if (q.isFailed()) { - val resolvedBy = resolvedBy()?.getId() + val resolvedBy = resolvedBy() val err = handleError(q.getError(), key.id, resolvedBy) log.debug("Quorum is failed. Method ${key.method}, message ${err.message}") Mono.error(err) @@ -220,7 +220,7 @@ class QuorumRequestReader( } private fun resolvedBy() = - if (quorum.getResolvedBy().isEmpty()) null else quorum.getResolvedBy().last() + if (quorum.getResolvedBy().isEmpty()) null else quorum.getResolvedBy().last().getUpstreamSettingsData() private fun noResponse(method: String, q: CallQuorum): Mono { return apiControl.upstreamsMatchesResponse()?.run { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt index 8ed57cde..6dd4551d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt @@ -44,7 +44,7 @@ class BroadcastReader( val sig = getSignature(key, it.jsonRpcResponse, it.upstream.getId()) quorum.record(it.jsonRpcResponse, sig, it.upstream) } else { - val err = ChainException(ChainResponse.NumberId(key.id), it.jsonRpcResponse.error!!, it.upstream.getId()) + val err = ChainException(ChainResponse.NumberId(key.id), it.jsonRpcResponse.error!!, it.upstream.getUpstreamSettingsData()) quorum.record(err, null, it.upstream) } quorum @@ -58,7 +58,7 @@ class BroadcastReader( quorum.getResponse()!!.getResult(), quorum.getSignature(), upstreams.size, - quorum.getResolvedBy().first(), + quorum.getResolvedBy().first().getUpstreamSettingsData(), null, ) Mono.just(res) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt index 978fe091..2d91b790 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt @@ -34,9 +34,9 @@ abstract class RequestReader( ) } - protected fun handleError(error: ChainCallError?, id: Int, resolvedBy: String?) = - error?.asException(ChainResponse.NumberId(id), resolvedBy) - ?: ChainException(ChainResponse.NumberId(id), ChainCallError(-32603, "Unhandled Upstream error"), resolvedBy) + protected fun handleError(error: ChainCallError?, id: Int, upstreamSettingsData: Upstream.UpstreamSettingsData?) = + error?.asException(ChainResponse.NumberId(id), upstreamSettingsData) + ?: ChainException(ChainResponse.NumberId(id), ChainCallError(-32603, "Unhandled Upstream error"), upstreamSettingsData) protected fun getSignature(key: ChainRequest, response: ChainResponse, upstreamId: String) = response.providedSignature @@ -50,7 +50,7 @@ abstract class RequestReader( val value: ByteArray, val signature: ResponseSigner.Signature?, val quorum: Int, - val resolvedBy: Upstream?, + val resolvedUpstreamData: Upstream.UpstreamSettingsData?, val stream: Flux?, ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 91b7f9e3..27ce1351 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -43,6 +43,7 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.Selector +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 @@ -122,7 +123,8 @@ open class NativeCall( Mono.just(firstChunk) .map { buildStreamResult(it, callResult.id) - .setUpstreamId(callResult.upstreamId) + .setUpstreamId(callResult.upstreamSettingsData?.id) + .setUpstreamNodeVersion(callResult.upstreamSettingsData?.nodeVersion) .build() }, stream.skip(1).map { buildStreamResult(it, callResult.id).build() }, @@ -224,7 +226,10 @@ open class NativeCall( if (it.nonce != null && it.signature != null) { result.signature = buildSignature(it.nonce, it.signature) } - it.upstreamId ?.let { result.upstreamId = it } + it.upstreamSettingsData?.let { + result.upstreamId = it.id + result.upstreamNodeVersion = it.nodeVersion + } return result.build() } @@ -406,12 +411,12 @@ open class NativeCall( .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, false)) .map { val result = it.getResult() - val upstreamId = it.providedUpstreamId ?: ctx.upstream.getId() + 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, upstreamId), upstreamId, ctx) + CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, resolvedUpstreamData?.id ?: ctx.upstream.getId()), resolvedUpstreamData, ctx) } else { - CallResult.ok(ctx.id, null, result, null, upstreamId, ctx) + CallResult.ok(ctx.id, null, result, null, resolvedUpstreamData, ctx) } } }.switchIfEmpty( @@ -435,13 +440,13 @@ open class NativeCall( return SpannedReader(reader, tracer, RPC_READER) .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, ctx.streamRequest)) .map { - val upId = it.resolvedBy?.getId() ?: ctx.upstream.getId() + val resolvedUpstreamData = it.resolvedUpstreamData ?: ctx.upstream.getUpstreamSettingsData() if (it.stream == null) { val bytes = ctx.resultDecorator.processResult(it) validateResult(bytes, "remote", ctx) - CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx) + CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx) } else { - CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, upId, ctx, it.stream) + CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, resolvedUpstreamData, ctx, it.stream) } } .onErrorResume { t -> @@ -533,8 +538,8 @@ open class NativeCall( } override fun processResult(result: RequestReader.Result): ByteArray { val bytes = result.value - if (bytes.last() == quoteCode && result.resolvedBy != null) { - val suffix = result.resolvedBy.nodeId() + if (bytes.last() == quoteCode && result.resolvedUpstreamData != null) { + val suffix = result.resolvedUpstreamData.nodeId .toUByte() .toString(16).padStart(2, padChar = '0').toByteArray() bytes[bytes.lastIndex] = suffix.first() @@ -648,7 +653,7 @@ open class NativeCall( val message: String, val upstreamError: ChainCallError?, val data: String?, - val upstreamId: String? = null, + val upstreamSettingsData: Upstream.UpstreamSettingsData? = null, ) { companion object { @@ -666,7 +671,7 @@ open class NativeCall( } fun from(t: Throwable): CallError { return when (t) { - is ChainException -> CallError(t.error.code, t.error.message, t.error, getDataAsSting(t.error.details), t.upstreamId) + is ChainException -> CallError(t.error.code, t.error.message, t.error, getDataAsSting(t.error.details), t.upstreamSettingsData) is RpcException -> CallError(t.code, t.rpcMessage, null, getDataAsSting(t.details)) is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error", null, null) else -> { @@ -689,7 +694,7 @@ open class NativeCall( val result: ByteArray?, val error: CallError?, val signature: ResponseSigner.Signature?, - val upstreamId: String?, + val upstreamSettingsData: Upstream.UpstreamSettingsData?, val ctx: ValidCallContext?, val stream: Flux? = null, ) { @@ -701,15 +706,15 @@ open class NativeCall( callError: CallError?, signature: ResponseSigner.Signature?, ctx: ValidCallContext?, - ) : this(id, nonce, result, callError, signature, callError?.upstreamId, ctx) + ) : this(id, nonce, result, callError, signature, callError?.upstreamSettingsData, ctx) companion object { - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?, ctx: ValidCallContext?): CallResult { - return CallResult(id, nonce, result, null, signature, upstreamId, ctx) + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: Upstream.UpstreamSettingsData?, ctx: ValidCallContext?): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx) } - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?, ctx: ValidCallContext?, stream: Flux?): CallResult { - return CallResult(id, nonce, result, null, signature, upstreamId, ctx, stream) + 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) } fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext?): CallResult { 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 58e40a33..7e08a7df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -85,7 +85,7 @@ open class GenericUpstreamCreator( chainConfig, connectorFactory, cs::validator, - cs::labelDetector, + cs::upstreamSettingsDetector, cs::lowerBoundBlockDetector, ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainCallError.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainCallError.kt index 9677f519..b1001a0b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainCallError.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainCallError.kt @@ -36,7 +36,7 @@ data class ChainCallError(val code: Int, val message: String, val details: Any?) return ChainCallUpstreamException(id ?: ChainResponse.NumberId(-1), this) } - fun asException(id: ChainResponse.Id?, upstreamId: String?): ChainException { - return ChainException(id ?: ChainResponse.NumberId(-1), this, upstreamId, false) + fun asException(id: ChainResponse.Id?, upstreamSettingsData: Upstream.UpstreamSettingsData?): ChainException { + return ChainException(id ?: ChainResponse.NumberId(-1), this, upstreamSettingsData, false) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainException.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainException.kt index 2559aa8a..12612226 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainException.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainException.kt @@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException open class ChainException( val id: ChainResponse.Id, val error: ChainCallError, - val upstreamId: String? = null, + val upstreamSettingsData: Upstream.UpstreamSettingsData? = null, writableStackTrace: Boolean = true, cause: Throwable? = null, ) : Exception(error.message, cause, true, writableStackTrace) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt index a0644765..ca45b9e2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt @@ -32,7 +32,7 @@ class ChainResponse( * When making a request through Dshackle protocol a remote may provide its signature with the response, which we keep here */ val providedSignature: ResponseSigner.Signature? = null, - val providedUpstreamId: String? = null, + val resolvedUpstreamData: Upstream.UpstreamSettingsData? = null, ) { constructor(stream: Flux, id: Int) : @@ -40,8 +40,8 @@ class ChainResponse( constructor(result: ByteArray?, error: ChainCallError?) : this(result, error, NumberId(0), null) - constructor(result: ByteArray?, error: ChainCallError?, resolvedBy: String?) : - this(result, error, NumberId(0), null, null, resolvedBy) + constructor(result: ByteArray?, error: ChainCallError?, resolvedUpstreamData: Upstream.UpstreamSettingsData?) : + this(result, error, NumberId(0), null, null, resolvedUpstreamData) companion object { private val NULL_VALUE = "null".toByteArray() @@ -135,7 +135,7 @@ class ChainResponse( } fun copyWithId(id: Id): ChainResponse { - return ChainResponse(result, error, id, stream, providedSignature, providedUpstreamId) + return ChainResponse(result, error, id, stream, providedSignature, resolvedUpstreamData) } override fun equals(other: Any?): Boolean { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 14475d12..dbee254b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -334,6 +334,14 @@ abstract class Multistream( override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData = lowerBlock + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { + return Upstream.UpstreamSettingsData( + nodeId(), + getId(), + UNKNOWN_CLIENT_VERSION, + ) + } + private fun observeUpstreamsStatuses() { subscribeAddedUpstreams() .flatMap { upstream -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index aac568f8..f2625f38 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -45,8 +45,17 @@ interface Upstream : Lifecycle { fun getCapabilities(): Set fun isGrpc(): Boolean fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData + fun getUpstreamSettingsData(): UpstreamSettingsData? fun cast(selfType: Class): T fun nodeId(): Byte + + data class UpstreamSettingsData( + val nodeId: Byte, + val id: String, + val nodeVersion: String, + ) { + constructor(id: String) : this(0, id, UNKNOWN_CLIENT_VERSION) + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LabelsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSettingsDetector.kt similarity index 64% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/LabelsDetector.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSettingsDetector.kt index 8debd08e..688517db 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/LabelsDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSettingsDetector.kt @@ -4,25 +4,45 @@ import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.module.kotlin.readValue import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.reader.ChainReader import org.slf4j.LoggerFactory import reactor.core.publisher.Flux +import reactor.core.publisher.Mono -typealias LabelsDetectorBuilder = (Chain, ChainReader) -> LabelsDetector? -interface LabelsDetector { - fun detectLabels(): Flux> +const val UNKNOWN_CLIENT_VERSION = "unknown" + +typealias UpstreamSettingsDetectorBuilder = (Chain, Upstream) -> UpstreamSettingsDetector? +abstract class UpstreamSettingsDetector( + private val upstream: Upstream, +) { + protected val log = LoggerFactory.getLogger(this::class.java) + + abstract fun detectLabels(): Flux> + + fun detectClientVersion(): Mono { + return upstream.getIngressReader() + .read(clientVersionRequest()) + .flatMap(ChainResponse::requireResult) + .map(::parseClientVersion) + .onErrorResume { + log.warn("Can't detect the client version of upstream ${upstream.getId()}, reason - {}", it.message) + Mono.just(UNKNOWN_CLIENT_VERSION) + } + } + + protected abstract fun clientVersionRequest(): ChainRequest + + protected abstract fun parseClientVersion(data: ByteArray): String } -abstract class BasicEthLabelsDetector( - private val reader: ChainReader, -) : LabelsDetector { - private val log = LoggerFactory.getLogger(this::class.java) - +abstract class BasicEthUpstreamSettingsDetector( + private val upstream: Upstream, +) : UpstreamSettingsDetector(upstream) { protected abstract fun nodeTypeRequest(): NodeTypeRequest protected fun detectNodeType(): Flux?> { val nodeTypeRequest = nodeTypeRequest() - return reader + return upstream + .getIngressReader() .read(nodeTypeRequest.request) .flatMap(ChainResponse::requireResult) .map { Global.objectMapper.readValue(it) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLabelsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLabelsDetector.kt deleted file mode 100644 index 0c6a8f8d..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLabelsDetector.kt +++ /dev/null @@ -1,27 +0,0 @@ -package io.emeraldpay.dshackle.upstream.beaconchain - -import com.fasterxml.jackson.databind.node.NullNode -import io.emeraldpay.dshackle.reader.ChainReader -import io.emeraldpay.dshackle.upstream.BasicEthLabelsDetector -import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.rpcclient.RestParams -import reactor.core.publisher.Flux - -class BeaconChainLabelsDetector( - reader: ChainReader, -) : BasicEthLabelsDetector(reader) { - - override fun nodeTypeRequest(): NodeTypeRequest { - return NodeTypeRequest( - ChainRequest("GET#/eth/v1/node/version", RestParams.emptyParams()), - ) { node -> - node.get("data")?.get("version") ?: NullNode.instance - } - } - - override fun detectLabels(): Flux> { - return Flux.merge( - detectNodeType(), - ) - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt index 4dec9430..d579dd4c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt @@ -11,11 +11,10 @@ import io.emeraldpay.dshackle.config.ChainsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.foundation.ChainOptions -import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.ChainRequest -import io.emeraldpay.dshackle.upstream.LabelsDetector import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.rpcclient.RestParams @@ -56,8 +55,8 @@ object BeaconChainSpecific : AbstractPollChainSpecific() { ) } - override fun labelDetector(chain: Chain, reader: ChainReader): LabelsDetector { - return BeaconChainLabelsDetector(reader) + override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { + return BeaconChainUpstreamSettingsDetector(upstream) } override fun validator( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainUpstreamSettingsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainUpstreamSettingsDetector.kt new file mode 100644 index 00000000..6da9287b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainUpstreamSettingsDetector.kt @@ -0,0 +1,41 @@ +package io.emeraldpay.dshackle.upstream.beaconchain + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.NullNode +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.BasicEthUpstreamSettingsDetector +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.RestParams +import reactor.core.publisher.Flux + +class BeaconChainUpstreamSettingsDetector( + upstream: Upstream, +) : BasicEthUpstreamSettingsDetector(upstream) { + + override fun nodeTypeRequest(): NodeTypeRequest { + return NodeTypeRequest( + clientVersionRequest(), + ) { node -> + node.get("data")?.get("version") ?: NullNode.instance + } + } + + override fun detectLabels(): Flux> { + return Flux.merge( + detectNodeType(), + ) + } + + override fun clientVersionRequest(): ChainRequest { + return ChainRequest("GET#/eth/v1/node/version", RestParams.emptyParams()) + } + + override fun parseClientVersion(data: ByteArray): String { + val node = Global.objectMapper.readValue(data) + + return node.get("data")?.get("version")?.textValue() ?: UNKNOWN_CLIENT_VERSION + } +} 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 097523a7..b925829a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt @@ -76,6 +76,10 @@ open class BitcoinRpcUpstream( return LowerBoundBlockDetector.LowerBlockData.default() } + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { + return null + } + @Suppress("UNCHECKED_CAST") override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { 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 e4df9564..ff323fbf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt @@ -11,11 +11,11 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LabelsDetector import io.emeraldpay.dshackle.upstream.LogsOracle import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallSelector @@ -95,8 +95,8 @@ object EthereumChainSpecific : AbstractPollChainSpecific() { return EthereumLowerBoundBlockDetector(chain, upstream) } - override fun labelDetector(chain: Chain, reader: ChainReader): LabelsDetector { - return EthereumLabelsDetector(reader, chain) + override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { + return EthereumUpstreamSettingsDetector(upstream, chain) } override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { 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 4c0ce13f..6d5d4240 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -17,6 +17,7 @@ import io.emeraldpay.dshackle.upstream.ChainException import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.ethereum.domain.Address @@ -91,7 +92,7 @@ class EthereumDirectReader( Mono.empty() } else { Mono.just( - Result(TxContainer.from(tx, result.data), result.upstreamId), + Result(TxContainer.from(tx, result.data), result.resolvedUpstreamData), ) } } @@ -116,7 +117,7 @@ class EthereumDirectReader( if (str.startsWith("\"") && str.endsWith("\"")) { Result( Wei.from(str.substring(1, str.length - 1)), - it.upstreamId, + it.resolvedUpstreamData, ) } else { throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value") @@ -180,7 +181,7 @@ class EthereumDirectReader( log.debug("Empty logs for block $key") Mono.empty() } else { - Mono.just(Result(logs, it.upstreamId)) + Mono.just(Result(logs, it.resolvedUpstreamData)) } } } @@ -208,7 +209,7 @@ class EthereumDirectReader( Mono.just( Result( BlockContainer.from(block, result.data, "unknown"), - result.upstreamId, + result.resolvedUpstreamData, ), ) } @@ -245,12 +246,12 @@ class EthereumDirectReader( }.flatMap { it.read(request) }.map { - Result(it.value, it.resolvedBy?.getId()) + Result(it.value, it.resolvedUpstreamData) } } data class Result( val data: T, - val upstreamId: String?, + val resolvedUpstreamData: Upstream.UpstreamSettingsData?, ) } 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 9ba2b4e9..1f4a478d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt @@ -23,6 +23,7 @@ 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 @@ -71,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) { @@ -88,7 +89,7 @@ class EthereumLocalReader( } reader.txByHashAsCont() .read(hash) - .map { it.data.json!! to it.upstreamId } + .map { it.data.json!! to it.resolvedUpstreamData } } method == "eth_getBlockByHash" -> { @@ -105,7 +106,7 @@ class EthereumLocalReader( if (withTx) { null } else { - reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.upstreamId } + reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.resolvedUpstreamData } } } @@ -125,7 +126,7 @@ class EthereumLocalReader( } reader.receipts() .read(hash) - .map { it.data to it.upstreamId } + .map { it.data to it.resolvedUpstreamData } } method == "drpc_getLogsEstimate" -> { @@ -138,7 +139,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") } @@ -179,10 +180,10 @@ class EthereumLocalReader( } return reader.blocksByHeightAsCont() - .read(number).map { it.data.json!! to it.upstreamId } + .read(number).map { it.data.json!! to it.resolvedUpstreamData } } - fun getLogsEstimate(params: List): Mono>? { + fun getLogsEstimate(params: List): Mono>? { if (logsOracle == null) { throw NotImplementedError() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt similarity index 64% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetector.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt index 34bcb612..d2f8bacf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt @@ -1,21 +1,21 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.reader.ChainReader -import io.emeraldpay.dshackle.upstream.BasicEthLabelsDetector +import io.emeraldpay.dshackle.upstream.BasicEthUpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.ListParams import reactor.core.publisher.Flux import reactor.core.publisher.Mono const val ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" -class EthereumLabelsDetector( - private val reader: ChainReader, +class EthereumUpstreamSettingsDetector( + private val upstream: Upstream, private val chain: Chain, -) : BasicEthLabelsDetector(reader) { - private val blockNumberReader = EthereumArchiveBlockNumberReader(reader) +) : BasicEthUpstreamSettingsDetector(upstream) { + private val blockNumberReader = EthereumArchiveBlockNumberReader(upstream.getIngressReader()) override fun detectLabels(): Flux> { return Flux.merge( @@ -24,6 +24,18 @@ class EthereumLabelsDetector( ) } + override fun clientVersionRequest(): ChainRequest { + return ChainRequest("web3_clientVersion", ListParams()) + } + + override fun parseClientVersion(data: ByteArray): String { + val version = String(data) + if (version.startsWith("\"") && version.endsWith("\"")) { + return version.substring(1, version.length - 1) + } + return version + } + private fun detectArchiveNode(): Mono> { return Mono.zip( blockNumberReader.readEarliestBlock(chain).flatMap { haveBalance(it) }, @@ -34,7 +46,7 @@ class EthereumLabelsDetector( } private fun haveBalance(blockNumber: String): Mono { - return reader.read( + return upstream.getIngressReader().read( ChainRequest( "eth_getBalance", ListParams(ZERO_ADDRESS, blockNumber), @@ -44,7 +56,7 @@ class EthereumLabelsDetector( override fun nodeTypeRequest(): NodeTypeRequest { return NodeTypeRequest( - ChainRequest("web3_clientVersion", ListParams()), + clientVersionRequest(), ) { node -> node } } } 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 6778a5c0..eeb8227b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt @@ -10,11 +10,12 @@ import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.EmptyEgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LabelsDetector import io.emeraldpay.dshackle.upstream.LogsOracle import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.NoIngressSubscription import io.emeraldpay.dshackle.upstream.NoopCachingReader +import io.emeraldpay.dshackle.upstream.Upstream +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 @@ -37,7 +38,7 @@ abstract class AbstractChainSpecific : ChainSpecific { return { _, _, _ -> NoopCachingReader } } - override fun labelDetector(chain: Chain, reader: ChainReader): LabelsDetector? { + 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 68571203..6a80732e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -19,11 +19,11 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LabelsDetector import io.emeraldpay.dshackle.upstream.LogsOracle import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.beaconchain.BeaconChainSpecific import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -60,7 +60,7 @@ interface ChainSpecific { fun validator(chain: Chain, upstream: Upstream, options: ChainOptions.Options, config: ChainConfig): UpstreamValidator - fun labelDetector(chain: Chain, reader: ChainReader): LabelsDetector? + fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector? fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription 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 71c62728..37e64908 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -13,11 +13,12 @@ import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LabelsDetectorBuilder import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetectorBuilder +import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetectorBuilder import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.UpstreamValidatorBuilder import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult @@ -30,6 +31,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Sinks import java.time.Duration import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference open class GenericUpstream( id: String, @@ -42,7 +44,7 @@ open class GenericUpstream( chainConfig: ChainsConfig.ChainConfig, connectorFactory: ConnectorFactory, validatorBuilder: UpstreamValidatorBuilder, - labelsDetectorBuilder: LabelsDetectorBuilder, + upstreamSettingsDetectorBuilder: UpstreamSettingsDetectorBuilder, lowerBoundBlockDetectorBuilder: LowerBoundBlockDetectorBuilder, ) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig), Lifecycle { @@ -54,12 +56,13 @@ open class GenericUpstream( private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false) protected val connector: GenericConnector = connectorFactory.create(this, chain) private var livenessSubscription: Disposable? = null - private val labelsDetector = labelsDetectorBuilder(chain, this.getIngressReader()) + private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this) private val lowerBoundBlockDetector = lowerBoundBlockDetectorBuilder(chain, this) private val started = AtomicBoolean(false) private val isUpstreamValid = AtomicBoolean(false) + private val clientVersion = AtomicReference(UNKNOWN_CLIENT_VERSION) override fun getHead(): Head { return connector.getHead() @@ -91,6 +94,14 @@ open class GenericUpstream( return lowerBoundBlockDetector.getCurrentLowerBlock() } + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { + return Upstream.UpstreamSettingsData( + nodeId(), + getId(), + clientVersion.get(), + ) + } + @Suppress("UNCHECKED_CAST") override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { @@ -165,8 +176,14 @@ open class GenericUpstream( } } - private fun detectLabels() { - labelsDetector?.detectLabels()?.subscribe { label -> updateLabels(label) } + private fun detectSettings() { + settingsDetector?.detectLabels()?.subscribe { label -> updateLabels(label) } + + settingsDetector?.detectClientVersion() + ?.subscribe { + log.info("Detected node version $it for upstream ${getId()}") + clientVersion.set(it) + } } private fun upstreamStart() { @@ -185,7 +202,7 @@ open class GenericUpstream( }, { log.debug("Error while checking live subscription for ${getId()}", it) },) - detectLabels() + detectSettings() detectLowerBlock() } 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 a4f40448..76bc1466 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -154,6 +154,10 @@ class BitcoinGrpcUpstream( return LowerBoundBlockDetector.LowerBlockData.default() } + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { + return null + } + @Suppress("UNCHECKED_CAST") override fun cast(selfType: Class): T { if (!selfType.isAssignableFrom(this.javaClass)) { 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 b3e72c21..3f535c4a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GenericGrpcUpstream.kt @@ -182,4 +182,8 @@ open class GenericGrpcUpstream( override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData { return LowerBoundBlockDetector.LowerBlockData.default() } + + override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? { + return null + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt index a851b136..394975bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt @@ -13,6 +13,7 @@ import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator @@ -82,6 +83,10 @@ object NearChainSpecific : AbstractPollChainSpecific() { } } + override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { + return NearUpstreamSettingsDetector(upstream) + } + override fun latestBlockRequest(): ChainRequest = // {...} ChainRequest("block", ObjectParams("finality" to "optimistic")) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetector.kt new file mode 100644 index 00000000..685e26d0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetector.kt @@ -0,0 +1,39 @@ +package io.emeraldpay.dshackle.upstream.near + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties +import com.fasterxml.jackson.annotation.JsonProperty +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux + +class NearUpstreamSettingsDetector( + upstream: Upstream, +) : UpstreamSettingsDetector(upstream) { + override fun detectLabels(): Flux> { + return Flux.empty() + } + + override fun clientVersionRequest(): ChainRequest { + return ChainRequest("status", ListParams()) + } + + override fun parseClientVersion(data: ByteArray): String { + return Global.objectMapper.readValue(data).nearVersion.version + } + + @JsonIgnoreProperties(ignoreUnknown = true) + private data class NearVersionResponse( + @JsonProperty("version") + val nearVersion: NearVersion, + ) + + @JsonIgnoreProperties(ignoreUnknown = true) + private data class NearVersion( + @JsonProperty("version") + val version: String, + ) +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index fe6c4e96..c5126b73 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.reader.ChainReader import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.RequestMetrics +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError import io.emeraldpay.dshackle.upstream.signature.ResponseSigner @@ -109,7 +110,7 @@ class JsonRpcGrpcClient( } else { null } - Mono.just(ChainResponse(bytes, null, ChainResponse.NumberId(0), null, signature, resp.upstreamId)) + Mono.just(ChainResponse(bytes, null, ChainResponse.NumberId(0), null, signature, Upstream.UpstreamSettingsData(0, resp.upstreamId, resp.upstreamNodeVersion))) } else { metrics?.fails?.increment() Mono.error( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt index 6979e96d..1c5282ce 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt @@ -13,12 +13,12 @@ import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.DefaultSolanaMethods import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.IngressSubscription -import io.emeraldpay.dshackle.upstream.LabelsDetector import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.SingleCallValidator import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector import io.emeraldpay.dshackle.upstream.UpstreamValidator import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions import io.emeraldpay.dshackle.upstream.generic.AbstractChainSpecific @@ -142,8 +142,8 @@ object SolanaChainSpecific : AbstractChainSpecific() { return SolanaLowerBoundBlockDetector(chain, upstream) } - override fun labelDetector(chain: Chain, reader: ChainReader): LabelsDetector? { - return null + override fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector { + return SolanaUpstreamSettingsDetector(upstream) } override fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetector.kt new file mode 100644 index 00000000..92c23e6c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetector.kt @@ -0,0 +1,33 @@ +package io.emeraldpay.dshackle.upstream.solana + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties +import com.fasterxml.jackson.annotation.JsonProperty +import com.fasterxml.jackson.module.kotlin.readValue +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import reactor.core.publisher.Flux + +class SolanaUpstreamSettingsDetector( + upstream: Upstream, +) : UpstreamSettingsDetector(upstream) { + override fun detectLabels(): Flux> { + return Flux.empty() + } + + override fun clientVersionRequest(): ChainRequest { + return ChainRequest("getVersion", ListParams()) + } + + override fun parseClientVersion(data: ByteArray): String { + return Global.objectMapper.readValue(data).version + } + + @JsonIgnoreProperties(ignoreUnknown = true) + private data class SolanaVersion( + @JsonProperty("solana-core") + val version: String, + ) +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy index bec1d787..d74d7281 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.rpc.NativeSubscribe import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.ethereum.json.RequestJson import io.micrometer.core.instrument.Counter import reactor.core.publisher.Flux @@ -85,7 +86,7 @@ class WebsocketHandlerSpec extends Specification { def "Respond to a single call"() { setup: - def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, "test", null, null) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, new Upstream.UpstreamSettingsData("test"), null, null) def nativeCall = Mock(NativeCall) { 1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 17794b20..f2e81443 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -120,14 +120,10 @@ class NativeCallSpec extends Specification { setup: def quorum = new AlwaysQuorum() - def ups = Mock(Upstream) { - _ * nodeId() >> (byte) 1 - } - def nativeCall = nativeCall() nativeCall.requestReaderFactory = Mock(RequestReaderFactory) { 1 * create(_) >> Mock(RequestReader) { - 1 * read(_) >> Mono.just(new RequestReader.Result("\"foo\"".bytes, null, 1, ups, null)) + 1 * read(_) >> Mono.just(new RequestReader.Result("\"foo\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte)1, "test", "v"), null)) } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, @@ -250,7 +246,7 @@ class NativeCallSpec extends Specification { when: def resp = nativeCall.buildResponse( - new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, new ResponseSigner.Signature("sig1".bytes, "test", 100), "test", null, null) + new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, new ResponseSigner.Signature("sig1".bytes, "test", 100), new Upstream.UpstreamSettingsData("test"), null, null) ) then: resp.id == 1561 @@ -597,9 +593,6 @@ class NativeCallSpec extends Specification { def "Decorate eth_newFilter result"() { setup: - def ups = Mock(Upstream) { - _ * nodeId() >> (byte)255 - } def quorum = new AlwaysQuorum() def methods = new ManagedCallMethods( new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), @@ -616,7 +609,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.requestReaderFactory = Mock(RequestReaderFactory) { 1 * create(_) >> Mock(RequestReader) { - 1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, ups, null)) + 1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte) 255, "", ""), null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, @@ -633,9 +626,6 @@ class NativeCallSpec extends Specification { def "Decorate eth_newFilter result with short nodeId"() { setup: - def ups = Mock(Upstream) { - _ * nodeId() >> (byte)1 - } def quorum = new AlwaysQuorum() def methods = new ManagedCallMethods( new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false), @@ -652,7 +642,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.requestReaderFactory = Mock(RequestReaderFactory) { 1 * create(_) >> Mock(RequestReader) { - 1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, ups, null)) + 1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte) 1, "", ""), null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy index 201d058e..88124754 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/GenericUpstreamMock.groovy @@ -74,7 +74,7 @@ class GenericUpstreamMock extends GenericUpstream { ChainConfig.default(), new ConnectorFactoryMock(api, new EthereumHeadMock()), io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&validator, - io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&labelDetector, + io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&upstreamSettingsDetector, io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.&lowerBoundBlockDetector, ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 5fa2d995..4c63344f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -76,7 +76,7 @@ class FilteredApisSpec extends Specification { ChainsConfig.ChainConfig.default(), connectorFactory, cs.&validator, - cs.&labelDetector, + cs.&upstreamSettingsDetector, cs.&lowerBoundBlockDetector ) } 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 da76fbe0..9a7aaef9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -32,7 +32,7 @@ class EthereumDirectReaderSpec extends Specification { String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String address1 = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d" - Upstream resolver = TestingCommons.upstream() + Upstream.UpstreamSettingsData data = new Upstream.UpstreamSettingsData("test") def "Reads block by hash"() { setup: @@ -54,7 +54,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getBlockByHash", new ListParams([hash1, false]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null) ) } } @@ -81,7 +81,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getBlockByHash", new ListParams([hash1, false]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(null), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(null), null, 1, data, null ) ) } @@ -114,7 +114,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getBlockByNumber", new ListParams(["0x64", false]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null ) ) } @@ -146,7 +146,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getLogs", new ListParams([Map.of("blockHash", hash1)]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes([json]), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes([json]), null, 1, data, null ) ) } @@ -179,7 +179,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getTransactionByHash", new ListParams([hash1]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null ) ) } @@ -212,7 +212,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getTransactionReceipt", new ListParams([hash1]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null ) ) } @@ -246,7 +246,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getTransactionReceipt", new ListParams([hash1]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null ) ) } @@ -271,7 +271,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getTransactionByHash", new ListParams([hash1]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(null), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes(null), null, 1, data, null ) ) } @@ -301,7 +301,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getBalance", new ListParams([address1, "latest"]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, data, null ) ) } @@ -332,7 +332,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RequestReader) { 1 * read(new ChainRequest("eth_getBalance", new ListParams([address1, "0xa8c9bb"]))) >> Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver, null + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, data, null ) ) } @@ -361,7 +361,7 @@ class EthereumDirectReaderSpec extends Specification { } def result = Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -401,7 +401,7 @@ class EthereumDirectReaderSpec extends Specification { } def result = Mono.just( new RequestReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) + Global.objectMapper.writeValueAsBytes(json), null, 1, data, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy similarity index 75% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetectorSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy index 9845489b..f10abaa3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetectorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetectorSpec.groovy @@ -15,7 +15,7 @@ import spock.lang.Specification import java.time.Duration -class EthereumLabelsDetectorSpec extends Specification { +class EthereumUpstreamSettingsDetectorSpec extends Specification { def "Detect labels"() { setup: @@ -27,7 +27,7 @@ class EthereumLabelsDetectorSpec extends Specification { answer("eth_getBalance", ["0x0000000000000000000000000000000000000000", "0x2710"], "") } ) - def detector = new EthereumLabelsDetector(up.getIngressReader(), Chain.ETHEREUM__MAINNET) + def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET) when: def act = detector.detectLabels() @@ -51,7 +51,7 @@ class EthereumLabelsDetectorSpec extends Specification { def "No any label"() { setup: def up = Mock(DefaultUpstream) { - 1 * getIngressReader() >> Mock(Reader) { + 4 * getIngressReader() >> Mock(Reader) { 1 * read(new ChainRequest("web3_clientVersion", new ListParams())) >> Mono.just(new ChainResponse('no/v1.19.3+e8ac1da4/linux-x64/dotnet7.0.8'.getBytes(), null)) 1 * read(new ChainRequest("eth_blockNumber", new ListParams())) >> @@ -62,7 +62,7 @@ class EthereumLabelsDetectorSpec extends Specification { Mono.just(new ChainResponse("".getBytes(), null)) } } - def detector = new EthereumLabelsDetector(up.getIngressReader(), Chain.ETHEREUM__MAINNET) + def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET) when: def act = detector.detectLabels() then: @@ -70,4 +70,22 @@ class EthereumLabelsDetectorSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } + + def "Detect client version"() { + setup: + def up = Mock(DefaultUpstream) { + 2 * getIngressReader() >> Mock(Reader) { + 1 * read(new ChainRequest("web3_clientVersion", new ListParams())) >> + Mono.just(new ChainResponse('"Erigon/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3"'.getBytes(), null)) + } + } + def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET) + when: + def act = detector.detectClientVersion() + then: + StepVerifier.create(act) + .expectNext("Erigon/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallTest.kt index 86f5ed41..31afeb36 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallTest.kt @@ -5,6 +5,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallReplyItem import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallRequest import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import io.emeraldpay.dshackle.upstream.stream.Chunk import org.junit.jupiter.api.Test @@ -31,7 +33,7 @@ class NativeCallTest { ), ) { on { nativeCallResult(request) } doReturn Flux.just( - NativeCall.CallResult.ok(1, null, "0x1".toByteArray(), null, "id", null), + NativeCall.CallResult.ok(1, null, "0x1".toByteArray(), null, Upstream.UpstreamSettingsData("id"), null), ) } @@ -39,6 +41,7 @@ class NativeCallTest { .expectNext( NativeCallReplyItem.newBuilder() .setUpstreamId("id") + .setUpstreamNodeVersion(UNKNOWN_CLIENT_VERSION) .setId(1) .setSucceed(true) .setPayload(ByteString.copyFrom("0x1".toByteArray())) @@ -60,7 +63,7 @@ class NativeCallTest { ), ) { on { nativeCallResult(request) } doReturn Flux.just( - NativeCall.CallResult(1, null, null, NativeCall.CallError(50001, "message", null, null, "upId"), null, null), + NativeCall.CallResult(1, null, null, NativeCall.CallError(50001, "message", null, null, Upstream.UpstreamSettingsData("upId")), null, null), ) } @@ -69,6 +72,7 @@ class NativeCallTest { NativeCallReplyItem.newBuilder() .setUpstreamId("upId") .setId(1) + .setUpstreamNodeVersion(UNKNOWN_CLIENT_VERSION) .setSucceed(false) .setErrorMessage("message") .setItemErrorCode(50001) @@ -95,7 +99,7 @@ class NativeCallTest { ), ) { on { nativeCallResult(request) } doReturn Flux.just( - NativeCall.CallResult.ok(1, null, "".toByteArray(), null, "upId", null, chunks), + NativeCall.CallResult.ok(1, null, "".toByteArray(), null, Upstream.UpstreamSettingsData("upId"), null, chunks), ) } @@ -104,6 +108,7 @@ class NativeCallTest { NativeCallReplyItem.newBuilder() .setUpstreamId("upId") .setId(1) + .setUpstreamNodeVersion(UNKNOWN_CLIENT_VERSION) .setChunked(true) .setSucceed(true) .setPayload(ByteString.copyFrom("0x1".toByteArray())) diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetectorTest.kt new file mode 100644 index 00000000..0cff4de3 --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetectorTest.kt @@ -0,0 +1,46 @@ +package io.emeraldpay.dshackle.upstream.near + +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.rpcclient.ListParams +import org.junit.jupiter.api.Test +import org.mockito.kotlin.doReturn +import org.mockito.kotlin.mock +import reactor.core.publisher.Mono +import reactor.test.StepVerifier + +class NearUpstreamSettingsDetectorTest { + + @Test + fun `detect client version`() { + val reader = mock { + on { read(ChainRequest("status", ListParams())) } doReturn + Mono.just( + ChainResponse( + """ + { + "version": { + "build": "1.38.1", + "rustc_version": "1.75.0", + "version": "1.38.1" + } + } + """.trimIndent().toByteArray(), + null, + ), + ) + } + val up = mock { + on { getIngressReader() } doReturn reader + } + + val detector = NearUpstreamSettingsDetector(up) + + StepVerifier.create(detector.detectClientVersion()) + .expectNext("1.38.1") + .expectComplete() + .verify() + } +} diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetectorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetectorTest.kt new file mode 100644 index 00000000..f6df7011 --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetectorTest.kt @@ -0,0 +1,71 @@ +package io.emeraldpay.dshackle.upstream.solana + +import io.emeraldpay.dshackle.reader.ChainReader +import io.emeraldpay.dshackle.upstream.ChainCallError +import io.emeraldpay.dshackle.upstream.ChainRequest +import io.emeraldpay.dshackle.upstream.ChainResponse +import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.ListParams +import org.junit.jupiter.api.Test +import org.mockito.kotlin.doReturn +import org.mockito.kotlin.mock +import reactor.core.publisher.Mono +import reactor.test.StepVerifier + +class SolanaUpstreamSettingsDetectorTest { + + @Test + fun `detect client version`() { + val reader = mock { + on { read(ChainRequest("getVersion", ListParams())) } doReturn + Mono.just( + ChainResponse( + """ + { + "feature-set": 2891131721, + "solana-core": "1.16.7" + } + """.trimIndent().toByteArray(), + null, + ), + ) + } + val up = mock { + on { getIngressReader() } doReturn reader + } + + val detector = SolanaUpstreamSettingsDetector(up) + + StepVerifier.create(detector.detectClientVersion()) + .expectNext("1.16.7") + .expectComplete() + .verify() + } + + @Test + fun `unknown client if there is an error`() { + val reader = mock { + on { read(ChainRequest("getVersion", ListParams())) } doReturn + Mono.just( + ChainResponse( + null, + ChainCallError( + 1, + "message", + ), + ), + ) + } + val up = mock { + on { getIngressReader() } doReturn reader + } + + val detector = SolanaUpstreamSettingsDetector(up) + + StepVerifier.create(detector.detectClientVersion()) + .expectNext(UNKNOWN_CLIENT_VERSION) + .expectComplete() + .verify() + } +}