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

Detect and send node version in api (#445)

Browse Source
This commit is contained in:
KirillPamPam
2024-03-29 19:25:22 +04:00
committed by GitHub
parent 5fedc28d24
commit 7e9709bb6f
39 changed files with 459 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt
View File

@@ -208,7 +208,7 @@ class QuorumRequestReader(
private fun setupDefaultResult(key: ChainRequest): Mono<Result> {
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<Result> {
return apiControl.upstreamsMatchesResponse()?.run {

4
src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt
View File

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

8
src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt
View File

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

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

@@ -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<ParsedCallDetails>?,
val stream: Flux<Chunk>? = null,
) {
@@ -701,15 +706,15 @@ open class NativeCall(
callError: CallError?,
signature: ResponseSigner.Signature?,
ctx: ValidCallContext<ParsedCallDetails>?,
) : 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<ParsedCallDetails>?): 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<ParsedCallDetails>?): 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<ParsedCallDetails>?, stream: Flux<Chunk>?): 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<ParsedCallDetails>?, stream: Flux<Chunk>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream)
}
fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {

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

@@ -85,7 +85,7 @@ open class GenericUpstreamCreator(
chainConfig,
connectorFactory,
cs::validator,
cs::labelDetector,
cs::upstreamSettingsDetector,
cs::lowerBoundBlockDetector,
)

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainCallError.kt
View File

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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainException.kt
View File

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

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

@@ -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<Chunk>, 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 {

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

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

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

@@ -45,8 +45,17 @@ interface Upstream : Lifecycle {
fun getCapabilities(): Set<Capability>
fun isGrpc(): Boolean
fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData
fun getUpstreamSettingsData(): UpstreamSettingsData?
fun <T : Upstream> cast(selfType: Class<T>): 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)
}
}

40
src/main/kotlin/io/emeraldpay/dshackle/upstream/LabelsDetector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSettingsDetector.kt
View File

@@ -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<Pair<String, String>>
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<Pair<String, String>>
fun detectClientVersion(): Mono<String> {
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<Pair<String, String>?> {
val nodeTypeRequest = nodeTypeRequest()
return reader
return upstream
.getIngressReader()
.read(nodeTypeRequest.request)
.flatMap(ChainResponse::requireResult)
.map { Global.objectMapper.readValue<JsonNode>(it) }

27
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainLabelsDetector.kt
View File

@@ -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<Pair<String, String>> {
return Flux.merge(
detectNodeType(),
)
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt
View File

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

41
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainUpstreamSettingsDetector.kt Normal file
View File

@@ -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<Pair<String, String>> {
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<JsonNode>(data)
return node.get("data")?.get("version")?.textValue() ?: UNKNOWN_CLIENT_VERSION
}
}

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

@@ -76,6 +76,10 @@ open class BitcoinRpcUpstream(
return LowerBoundBlockDetector.LowerBlockData.default()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {
return null
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {

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

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

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

@@ -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<T>(
val data: T,
val upstreamId: String?,
val resolvedUpstreamData: Upstream.UpstreamSettingsData?,
)
}

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

@@ -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<Pair<ByteArray, String?>>? {
fun commonRequests(key: ChainRequest): Mono<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
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<Any?>): Mono<Pair<ByteArray, String?>>? {
fun getBlockByNumber(params: List<Any?>): Mono<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
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<Any?>): Mono<Pair<ByteArray, String?>>? {
fun getLogsEstimate(params: List<Any?>): Mono<Pair<ByteArray, Upstream.UpstreamSettingsData?>>? {
if (logsOracle == null) {
throw NotImplementedError()
}

28
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLabelsDetector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt
View File

@@ -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<Pair<String, String>> {
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<Pair<String, String>> {
return Mono.zip(
blockNumberReader.readEarliestBlock(chain).flatMap { haveBalance(it) },
@@ -34,7 +46,7 @@ class EthereumLabelsDetector(
}
private fun haveBalance(blockNumber: String): Mono<ByteArray> {
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 }
}
}

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

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

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

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

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

@@ -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 <T : Upstream> cast(selfType: Class<T>): 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()
}

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

@@ -154,6 +154,10 @@ class BitcoinGrpcUpstream(
return LowerBoundBlockDetector.LowerBlockData.default()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {
return null
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {

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

@@ -182,4 +182,8 @@ open class GenericGrpcUpstream(
override fun getLowerBlock(): LowerBoundBlockDetector.LowerBlockData {
return LowerBoundBlockDetector.LowerBlockData.default()
}
override fun getUpstreamSettingsData(): Upstream.UpstreamSettingsData? {
return null
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt
View File

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

39
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearUpstreamSettingsDetector.kt Normal file
View File

@@ -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<Pair<String, String>> {
return Flux.empty()
}
override fun clientVersionRequest(): ChainRequest {
return ChainRequest("status", ListParams())
}
override fun parseClientVersion(data: ByteArray): String {
return Global.objectMapper.readValue<NearVersionResponse>(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,
)
}

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

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

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt
View File

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

33
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetector.kt Normal file
View File

@@ -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<Pair<String, String>> {
return Flux.empty()
}
override fun clientVersionRequest(): ChainRequest {
return ChainRequest("getVersion", ListParams())
}
override fun parseClientVersion(data: ByteArray): String {
return Global.objectMapper.readValue<SolanaVersion>(data).version
}
@JsonIgnoreProperties(ignoreUnknown = true)
private data class SolanaVersion(
@JsonProperty("solana-core")
val version: String,
)
}