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

Merge pull request #33 from p2p-org/eth_filters_support

Browse Source 
eth_filter methods support
This commit is contained in:
MaxFomenkov
2022-11-09 02:45:35 +03:00
committed by GitHub
parent 2a7542e54c fc1b9b9b07
commit 605ed8cad5
35 changed files with 544 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/04-upstream-config.adoc
View File

@@ -51,6 +51,7 @@ cluster:
min-peers: 2
upstreams:
- id: us-nodes
node-id: 1
chain: auto
connection:
grpc:
@@ -61,6 +62,7 @@ cluster:
certificate: client.crt
key: client.p8.key
- id: infura-eth
node-id: 2
chain: ethereum
role: fallback
labels:
@@ -80,6 +82,7 @@ cluster:
username: ${INFURA_USER}
password: ${INFURA_PASSWD}
- id: ethereum-pos
node-id: 3
chain: ropsten
connection:
ethereum-pos:
@@ -110,6 +113,12 @@ In the example above we have:
** label `[provider: infura]` is set for that particular upstream, which can be selected during a request.For example for some requests you may want to use nodes with that label only, i.e. _"send that tx to infura nodes only"_, or _"read only from archive node, with label [archive: true]"_
** upstream validation (peers, sync status, etc) is disabled for that particular upstream
=== Nodes
`[node-id: 1]` is numeric node identifier defined in a range [1..255] and used to forward
`eth_getFilterChanges` request to the node where one of `eth_newFilter`, `eth_newBlockFilter` or `eth_newPendingTransactionFilter` methods was executed (because filter is s stateful method).
*It's kindly recommended* to strictly associate _node-id_ parameter with a physical node and keep it during any configuration changes
=== Roles and Fallback upstream
By default, the Dshackle connects to each upstream in a Round-Robin basis, i.e. sequentially one by one.
@@ -206,6 +215,10 @@ Dshackle currently supports
- `eth_getUncleByBlockNumberAndIndex`
- `eth_feeHistory`
- `eth_getLogs`
- `eth_getFilterChanges`
- `eth_newFilter`
- `eth_newBlockFilter`
- `eth_newPendingTransactionFilter`
.Plus following methods are answered directly by Dshackle
- `net_version`

1
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -69,6 +69,7 @@ open class UpstreamsConfig {
class Upstream<T : UpstreamConnection> {
var id: String? = null
var nodeId: Int? = null
var chain: String? = null
var options: Options? = null
var isEnabled = true

14
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -31,6 +31,7 @@ class UpstreamsConfigReader(
private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
private val authConfigReader = AuthConfigReader()
private val knownNodeIds: MutableSet<Int> = HashSet()
fun read(input: InputStream): UpstreamsConfig? {
val configNode = readNode(input)
@@ -236,11 +237,22 @@ class UpstreamsConfigReader(
log.warn("Invalid id: $id")
return false
}
return true
return upstream.nodeId?.let {
if (it !in 1..255) {
log.warn("Invalid node-id: $it. Must be in range [1, 255].")
false
} else if (!knownNodeIds.add(it)) {
log.warn("Duplicated node-id: $it. Must be in unique.")
false
} else {
true
}
} ?: true
}
internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
upstream.id = getValueAsString(upNode, "id")
upstream.nodeId = getValueAsInt(upNode, "node-id")
upstream.options = tryReadOptions(upNode)
upstream.methods = tryReadMethods(upNode)
getValueAsBool(upNode, "enabled")?.let {

6
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import java.util.concurrent.ConcurrentLinkedQueue
open class AlwaysQuorum : CallQuorum {
@@ -28,6 +29,7 @@ open class AlwaysQuorum : CallQuorum {
private var result: ByteArray? = null
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
override fun init(head: Head) {
}
@@ -48,6 +50,7 @@ open class AlwaysQuorum : CallQuorum {
result = response
resolved = true
sig = signature
resolvers.add(upstream)
return true
}
@@ -64,6 +67,9 @@ open class AlwaysQuorum : CallQuorum {
return rpcError
}
override fun getResolvedBy(): List<Upstream> =
resolvers.toList()
override fun toString(): String {
return "Quorum: Accept Any"
}

1
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -34,4 +34,5 @@ interface CallQuorum {
fun getSignature(): ResponseSigner.Signature?
fun getResult(): ByteArray?
fun getError(): JsonRpcError?
fun getResolvedBy(): Collection<Upstream>
}

5
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicReference
/**
@@ -34,6 +35,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
private val failed = AtomicReference(false)
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
override fun init(head: Head) {
}
@@ -51,6 +53,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
if (!lagging) {
result.set(response)
sig = signature
resolvers.add(upstream)
return true
}
return false
@@ -75,6 +78,8 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
override fun toString(): String {
return "Quorum: late <= $maxLag blocks"
}

7
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -117,9 +117,9 @@ class QuorumRpcReader(
return Function { quorumResult ->
quorumResult
.filter { it.isResolved() } // return nothing if not resolved
.map {
.map { quorum ->
// TODO find actual quorum number
QuorumRpcReader.Result(it.getResult()!!, it.getSignature(), 1)
Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy().map { it.nodeId() })
}
.switchIfEmpty(defaultResult)
}
@@ -197,6 +197,7 @@ class QuorumRpcReader(
class Result(
val value: ByteArray,
val signature: ResponseSigner.Signature?,
val quorum: Int
val quorum: Int,
val resolvers: Collection<Byte>
)
}

6
src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
import java.util.concurrent.ConcurrentLinkedQueue
abstract class ValueAwareQuorum<T>(
val clazz: Class<T>
@@ -30,6 +31,7 @@ abstract class ValueAwareQuorum<T>(
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
private var rpcError: JsonRpcError? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
fun extractValue(response: ByteArray, clazz: Class<T>): T? {
return Global.objectMapper.readValue(response.inputStream(), clazz)
@@ -39,6 +41,7 @@ abstract class ValueAwareQuorum<T>(
try {
val value = extractValue(response, clazz)
recordValue(response, value, signature, upstream)
resolvers.add(upstream)
} catch (e: RpcException) {
recordError(response, e.rpcMessage, signature, upstream)
} catch (e: Exception) {
@@ -59,4 +62,7 @@ abstract class ValueAwareQuorum<T>(
override fun getError(): JsonRpcError? {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
}

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

@@ -102,7 +102,8 @@ open class NativeCall(
}
fun parseParams(it: ValidCallContext<RawCallDetails>): ValidCallContext<ParsedCallDetails> {
val params = extractParams(it.payload.params)
val rawParams = extractParams(it.payload.params)
val params = it.requestDecorator.processRequest(rawParams)
return it.withPayload(ParsedCallDetails(it.payload.method, params))
}
@@ -234,17 +235,28 @@ open class NativeCall(
matcher.withMatcher(heightMatcher)
}
val nonce = requestItem.nonce.let { if (it == 0L) null else it }
val requestDecorator = getRequestDecorator(requestItem.method)
val resultDecorator = getResultDecorator(requestItem.method)
ValidCallContext(
requestItem.id,
nonce,
upstream,
matcher.build(),
callQuorum,
RawCallDetails(method, params)
RawCallDetails(method, params),
requestDecorator,
resultDecorator
)
}
}
private fun getRequestDecorator(method: String): RequestDecorator =
if (method == "eth_getFilterChanges") GetFilterUpdatesDecorator() else NoneRequestDecorator()
private fun getResultDecorator(method: String): ResultDecorator =
if (CreateFilterDecorator.createFilterMethods.contains(method)) CreateFilterDecorator() else NoneResultDecorator()
fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getRoutedApi(ctx.matcher)
.flatMap { api ->
@@ -283,7 +295,8 @@ open class NativeCall(
return reader
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce))
.map {
CallResult(ctx.id, ctx.nonce, it.value, null, it.signature)
val bytes = ctx.resultDecorator.processResult(it)
CallResult(ctx.id, ctx.nonce, bytes, null, it.signature)
}
.onErrorResume { t ->
val failure = when (t) {
@@ -350,14 +363,67 @@ open class NativeCall(
fun getError(): CallError
}
interface ResultDecorator {
fun processResult(result: QuorumRpcReader.Result): ByteArray
}
open class NoneResultDecorator : ResultDecorator {
override fun processResult(result: QuorumRpcReader.Result): ByteArray = result.value
}
open class CreateFilterDecorator : ResultDecorator {
companion object {
const val quoteCode = '"'.code.toByte()
val createFilterMethods = listOf("eth_getFilterChanges", "eth_newFilter", "eth_newBlockFilter")
}
override fun processResult(result: QuorumRpcReader.Result): ByteArray {
val bytes = result.value
if (bytes.last() == quoteCode) {
val suffix = result.resolvers.first().toUByte().toString(16).padStart(2, padChar = '0').toByteArray()
bytes[bytes.lastIndex] = suffix.first()
return bytes + suffix.last() + quoteCode
}
return bytes
}
}
interface RequestDecorator {
fun processRequest(request: List<Any>): List<Any>
}
open class NoneRequestDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> = request
}
open class GetFilterUpdatesDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> {
val filterId = request.first().toString()
val sanitized = filterId.substring(0, filterId.lastIndex - 1)
return listOf(sanitized)
}
}
open class ValidCallContext<T>(
val id: Int,
val nonce: Long?,
val upstream: Multistream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T
val payload: T,
val requestDecorator: RequestDecorator,
val resultDecorator: ResultDecorator
) : CallContext {
constructor(
id: Int,
nonce: Long?,
upstream: Multistream,
matcher: Selector.Matcher,
callQuorum: CallQuorum,
payload: T
) : this(id, nonce, upstream, matcher, callQuorum, payload, NoneRequestDecorator(), NoneResultDecorator())
override fun isValid(): Boolean {
return true
}
@@ -371,7 +437,7 @@ open class NativeCall(
}
fun <X> withPayload(payload: X): ValidCallContext<X> {
return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload)
return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload, requestDecorator, resultDecorator)
}
fun getApis(): ApiSource {

92
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.startup
import com.google.common.annotations.VisibleForTesting
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.UpstreamsConfig
@@ -53,7 +54,9 @@ import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.net.URI
import java.util.concurrent.atomic.AtomicInteger
import java.util.function.Function
import javax.annotation.PostConstruct
import kotlin.math.abs
@Repository
open class ConfiguredUpstreams(
@@ -65,6 +68,8 @@ open class ConfiguredUpstreams(
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
private var seq = AtomicInteger(0)
private val hashes: MutableMap<Byte, Boolean> = HashMap()
@PostConstruct
fun start() {
log.debug("Starting upstreams")
@@ -77,7 +82,7 @@ open class ConfiguredUpstreams(
log.debug("Start upstream ${up.id}")
if (up.connection is UpstreamsConfig.GrpcConnection) {
val options = up.options ?: UpstreamsConfig.Options()
buildGrpcUpstream(up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
} else {
val chain = Global.chainById(up.chain)
if (chain == Chain.UNSPECIFIED) {
@@ -88,14 +93,22 @@ open class ConfiguredUpstreams(
.merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
val upstream = when (BlockchainType.from(chain)) {
BlockchainType.ETHEREUM -> {
buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
buildEthereumUpstream(up.nodeId, up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
}
BlockchainType.BITCOIN -> {
buildBitcoinUpstream(up.cast(UpstreamsConfig.BitcoinConnection::class.java), chain, options)
}
BlockchainType.ETHEREUM_POS -> {
buildEthereumPosUpstream(up.cast(UpstreamsConfig.EthereumPosConnection::class.java), chain, options)
buildEthereumPosUpstream(
up.nodeId,
up.cast(UpstreamsConfig.EthereumPosConnection::class.java),
chain,
options
)
}
else -> {
log.error("Chain is unsupported: ${up.chain}")
return@forEach
@@ -156,6 +169,7 @@ open class ConfiguredUpstreams(
}
private fun buildEthereumPosUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumPosConnection>,
chain: Chain,
options: UpstreamsConfig.Options
@@ -167,13 +181,26 @@ open class ConfiguredUpstreams(
return null
}
val urls = ArrayList<URI>()
val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating), BlockValidator.ALWAYS_VALID)
val connectorFactory = buildEthereumConnectorFactory(
config.id!!,
execution,
chain,
urls,
NoChoiceWithPriorityForkChoice(conn.upstreamRating),
BlockValidator.ALWAYS_VALID
)
val methods = buildMethods(config, chain)
if (connectorFactory == null) {
return null
}
val hashUrl = conn.execution!!.let {
if (it.preferHttp) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
}
val hash = getHash(nodeId, hashUrl!!)
val upstream = EthereumPosRpcUpstream(
config.id!!,
hash,
chain,
options, config.role,
methods,
@@ -227,6 +254,7 @@ open class ConfiguredUpstreams(
}
private fun buildEthereumUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
chain: Chain,
options: UpstreamsConfig.Options
@@ -236,12 +264,22 @@ open class ConfiguredUpstreams(
val urls = ArrayList<URI>()
val methods = buildMethods(config, chain)
val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice(), EthereumBlockValidator())
val connectorFactory = buildEthereumConnectorFactory(
config.id!!,
conn,
chain,
urls,
MostWorkForkChoice(),
EthereumBlockValidator()
)
if (connectorFactory == null) {
return null
}
val hashUrl = if (conn.preferHttp) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url
val upstream = EthereumRpcUpstream(
config.id!!,
getHash(nodeId, hashUrl!!),
chain,
options, config.role,
methods,
@@ -253,12 +291,15 @@ open class ConfiguredUpstreams(
}
private fun buildGrpcUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
options: UpstreamsConfig.Options
) {
val endpoint = config.connection!!
val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}")
val ds = GrpcUpstreams(
config.id!!,
hash,
config.role,
endpoint.host!!,
endpoint.port,
@@ -289,7 +330,12 @@ open class ConfiguredUpstreams(
}
}
private fun buildWsFactory(id: String, chain: Chain, conn: UpstreamsConfig.EthereumConnection, urls: ArrayList<URI>? = null): EthereumWsFactory? {
private fun buildWsFactory(
id: String,
chain: Chain,
conn: UpstreamsConfig.EthereumConnection,
urls: ArrayList<URI>? = null
): EthereumWsFactory? {
return conn.ws?.let { endpoint ->
val wsApi = EthereumWsFactory(
id, chain,
@@ -305,15 +351,45 @@ open class ConfiguredUpstreams(
}
}
private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice, blockValidator: BlockValidator): EthereumConnectorFactory? {
private fun buildEthereumConnectorFactory(
id: String,
conn: UpstreamsConfig.EthereumConnection,
chain: Chain,
urls: ArrayList<URI>,
forkChoice: ForkChoice,
blockValidator: BlockValidator
): EthereumConnectorFactory? {
val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
val httpFactory = buildHttpFactory(conn, urls)
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator)
val connectorFactory =
EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator)
if (!connectorFactory.isValid()) {
log.warn("Upstream configuration is invalid (probably no http endpoint)")
return null
}
return connectorFactory
}
@VisibleForTesting
private fun getHash(nodeId: Int?, obj: Any): Byte =
nodeId?.toByte() ?: (obj.hashCode() % 255).let {
if (it == 0) 1 else it
}.let { nonZeroHash ->
listOf<Function<Int, Int>>(
Function { i -> i },
Function { i -> (-i) },
Function { i -> 127 - abs(i) },
Function { i -> abs(i) - 128 },
).map {
it.apply(nonZeroHash).toByte()
}.firstOrNull {
hashes[it] != true
}?.let {
hashes[it] = true
it
} ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
it != 0 && hashes[it.toByte()] != true
}.toByte()
}
}

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

@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicReference
abstract class DefaultUpstream(
private val id: String,
private val hash: Byte,
defaultLag: Long,
defaultAvail: UpstreamAvailability,
private val options: UpstreamsConfig.Options,
@@ -36,12 +37,14 @@ abstract class DefaultUpstream(
constructor(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?
) :
this(
id,
hash,
Long.MAX_VALUE,
UpstreamAvailability.UNAVAILABLE,
options,
@@ -52,12 +55,13 @@ abstract class DefaultUpstream(
constructor(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
node: QuorumForLabels.QuorumItem?
) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
this(id, hash, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
private val statusStream = Sinks.many()
@@ -139,6 +143,8 @@ abstract class DefaultUpstream(
return targets ?: throw IllegalStateException("Methods are not set")
}
override fun nodeId(): Byte = hash
private val quorumByLabel = node?.let { QuorumForLabels(it) }
?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())

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

@@ -282,6 +282,8 @@ abstract class Multistream(
return false
}
override fun nodeId(): Byte = 0
fun printStatus() {
var height: Long? = null
try {

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

@@ -396,4 +396,22 @@ class Selector {
return "Matcher: ${describeInternal()}"
}
}
class SameNodeMatcher(private val upstreamHash: Byte) : Matcher {
override fun matches(up: Upstream): Boolean =
up.nodeId() == upstreamHash
override fun describeInternal(): String =
"upstream node-id=${upstreamHash.toUByte()}"
override fun toString(): String {
return "Matcher: ${describeInternal()}"
}
override fun equals(other: Any?): Boolean {
if (other === this) return true
if (other !is SameNodeMatcher) return false
return other.upstreamHash == upstreamHash
}
}
}

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

@@ -40,4 +40,6 @@ interface Upstream {
fun isGrpc(): Boolean
fun <T : Upstream> cast(selfType: Class<T>): T
fun nodeId(): Byte
}

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

@@ -31,7 +31,7 @@ abstract class BitcoinUpstream(
callMethods: CallMethods,
node: QuorumForLabels.QuorumItem,
val esploraClient: EsploraClient? = null
) : DefaultUpstream(id, options, role, callMethods, node) {
) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node) {
constructor(
id: String,

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt
View File

@@ -70,7 +70,14 @@ class DefaultEthereumMethods(
"eth_feeHistory"
)
private val allowedMethods = anyResponseMethods + firstValueMethods + specialMethods + headVerifiedMethods
private val filterMethods = listOf(
"eth_getFilterChanges",
"eth_newFilter",
"eth_newBlockFilter",
"eth_newPendingTransactionFilter",
)
private val allowedMethods = anyResponseMethods + firstValueMethods + specialMethods + headVerifiedMethods + filterMethods
private val hardcodedMethods = listOf(
"net_version",
@@ -88,6 +95,7 @@ class DefaultEthereumMethods(
override fun getQuorumFor(method: String): CallQuorum {
return when {
filterMethods.contains(method) -> AlwaysQuorum()
hardcodedMethods.contains(method) -> AlwaysQuorum()
firstValueMethods.contains(method) -> AlwaysQuorum()
anyResponseMethods.contains(method) -> NotLaggingQuorum(4)

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt
View File

@@ -57,10 +57,26 @@ class EthereumCallSelector(
return blockTagSelector(params, 1, head)
} else if (method == "eth_getStorageAt") {
return blockTagSelector(params, 2, head)
} else if (method == "eth_getFilterChanges") {
return sameUpstreamMatcher(params)
}
return Mono.empty()
}
private fun sameUpstreamMatcher(params: String): Mono<Selector.Matcher> {
val list = objectMapper.readerFor(Any::class.java).readValues<Any>(params).readAll()
if (list.isEmpty()) {
return Mono.empty()
}
val filterId = list[0].toString()
if (filterId.length < 4) {
return Mono.just(Selector.SameNodeMatcher(0.toByte()))
}
val hashHex = filterId.substring(filterId.length - 2)
val nodeId = hashHex.toInt(16)
return Mono.just(Selector.SameNodeMatcher(nodeId.toByte()))
}
private fun blockTagSelector(params: String, pos: Int, head: Head): Mono<Selector.Matcher> {
val list = objectMapper.readerFor(Any::class.java).readValues<Any>(params).readAll()
if (list.size < pos + 1) {

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

@@ -36,13 +36,14 @@ import reactor.core.Disposable
open class EthereumRpcUpstream(
id: String,
hash: Byte,
val chain: Chain,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?,
connectorFactory: ConnectorFactory
) : EthereumUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
) : EthereumUpstream(id, hash, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java)
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)

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

@@ -24,11 +24,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
abstract class EthereumUpstream(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?
) : DefaultUpstream(id, options, role, targets, node) {
) : DefaultUpstream(id, hash, options, role, targets, node) {
private val capabilities = if (options.providesBalance != false) {
setOf(Capability.RPC, Capability.BALANCE)

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

@@ -36,13 +36,14 @@ import reactor.core.Disposable
open class EthereumPosRpcUpstream(
id: String,
hash: Byte,
val chain: Chain,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?,
connectorFactory: ConnectorFactory
) : EthereumPosUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
) : EthereumPosUpstream(id, hash, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
private val log = LoggerFactory.getLogger(EthereumPosRpcUpstream::class.java)
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)

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

@@ -24,11 +24,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
abstract class EthereumPosUpstream(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?
) : DefaultUpstream(id, options, role, targets, node) {
) : DefaultUpstream(id, hash, options, role, targets, node) {
private val capabilities = if (options.providesBalance != false) {
setOf(Capability.RPC, Capability.BALANCE)

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

@@ -51,6 +51,7 @@ import java.util.function.Function
open class EthereumGrpcUpstream(
private val parentId: String,
hash: Byte,
role: UpstreamsConfig.UpstreamRole,
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
@@ -58,6 +59,7 @@ open class EthereumGrpcUpstream(
overrideLabels: UpstreamsConfig.Labels?
) : EthereumUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
hash,
UpstreamsConfig.Options.getDefaults(),
role,
null,

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

@@ -51,6 +51,7 @@ import java.util.function.Function
open class EthereumPosGrpcUpstream(
private val parentId: String,
hash: Byte,
role: UpstreamsConfig.UpstreamRole,
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
@@ -59,6 +60,7 @@ open class EthereumPosGrpcUpstream(
overrideLabels: UpstreamsConfig.Labels?
) : EthereumPosUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
hash,
UpstreamsConfig.Options.getDefaults(),
role,
null, null

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -52,6 +52,7 @@ import kotlin.concurrent.withLock
class GrpcUpstreams(
private val id: String,
private val hash: Byte,
private val role: UpstreamsConfig.UpstreamRole,
private val host: String,
private val port: Int,
@@ -206,7 +207,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient, labels)
val created = EthereumGrpcUpstream(id, hash, role, chain, client!!, rpcClient, labels)
created.timeout = this.timeout
known[chain] = created
created.start()
@@ -222,7 +223,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating, labels)
val created = EthereumPosGrpcUpstream(id, hash, role, chain, client!!, rpcClient, nodeRating, labels)
created.timeout = this.timeout
known[chain] = created
created.start()

14
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -401,4 +401,18 @@ class UpstreamsConfigReaderSpec extends Specification {
act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY
act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.PRIMARY
}
def "Parse node id"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-node-id.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 2
act.upstreams[0].nodeId == 1
act.upstreams[0].id == "has_node_id"
act.upstreams[1].nodeId == null
act.upstreams[1].id == "has_no_node_id"
}
}

122
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -118,7 +118,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1))
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList((byte) 1)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
@@ -395,6 +395,68 @@ class NativeCallSpec extends Specification {
}
}
def "Prepare call adds decorator for eth_newFilter"() {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_newFilter"] as Set, [] as Set
)
methods.setQuorum("eth_newFilter", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("eth_newFilter")
)
.build()
when:
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act instanceof NativeCall.ValidCallContext
act.resultDecorator instanceof NativeCall.CreateFilterDecorator
}
def "Prepare call adds decorator for eth_getFilterChanges"() {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_getFilterChanges"] as Set, [] as Set
)
methods.setQuorum("eth_getFilterChanges", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("eth_getFilterChanges")
)
.build()
when:
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act instanceof NativeCall.ValidCallContext
act.requestDecorator instanceof NativeCall.GetFilterUpdatesDecorator
}
def "Parse empty params"() {
setup:
def nativeCall = nativeCall()
@@ -447,6 +509,64 @@ class NativeCallSpec extends Specification {
act.payload.method == "eth_test"
}
def "Decorate eth_getFilterUpdates params"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.NoneResultDecorator())
when:
def act = nativeCall.parseParams(ctx)
then:
act.id == 1
act.payload.params == ["0xab"]
act.payload.method == "eth_getFilterUpdates"
}
def "Decorate eth_newFilter result"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)255)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
when:
def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
def act = objectMapper.readValue(resp.result, Object)
then:
act == "0xabff"
resp.nonce == 10
}
def "Decorate eth_newFilter result with short nodeId"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)1)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
when:
def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
def act = objectMapper.readValue(resp.result, Object)
then:
act == "0xab01"
resp.nonce == 10
}
@Ignore
//TODO
def "Calls cache before remote"() {

33
src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy
View File

@@ -58,4 +58,37 @@ class ConfiguredUpstreamsSpec extends Specification {
act instanceof ManagedCallMethods
new String(act.executeHardcoded("foo_bar")) == "\"static_response\""
}
def "Calculate node-id"() {
setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
)
expect:
configurer.getHash(node, src) == expected
where:
node | src | expected
1 | "" | 1
9 | "hohoho" | 9
null | "hohoho" | 120
}
def "Calculate node-id conflicting results"() {
setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
)
when:
def h1 = configurer.getHash(null, "hohoho")
def h2 = configurer.getHash(null, "hohoho")
def h3 = configurer.getHash(null, "hohoho")
def h4 = configurer.getHash(null, "hohoho")
def h5 = configurer.getHash(null, "hohoho")
then:
h1 == (byte)120
h2 == (byte)-120
h3 == (byte)-9
h4 == (byte)8
h5 == (byte)-128
}
}

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

@@ -69,7 +69,7 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels) {
super(id, chain,
super(id, (byte)id.hashCode(), chain,
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.PRIMARY,
methods,

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

@@ -60,7 +60,7 @@ class EthereumRpcUpstreamMock extends EthereumRpcUpstream {
}
EthereumRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
super(id, chain,
super(id, id.hashCode().byteValue(), chain,
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.PRIMARY,
methods,

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

@@ -51,6 +51,7 @@ class FilteredApisSpec extends Specification {
def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
new EthereumRpcUpstream(
"test",
(byte)123,
Chain.ETHEREUM,
new UpstreamsConfig.Options(),
UpstreamsConfig.UpstreamRole.PRIMARY,

23
src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy
View File

@@ -349,4 +349,27 @@ class SelectorSpec extends Specification {
!act
}
def "Matches same nodeId"() {
setup:
def up = Mock(Upstream) {
nodeId() >> (byte)5
}
def matcher = new Selector.SameNodeMatcher((byte)5)
when:
def act = matcher.matches(up)
then:
act
}
def "Not matches nodeId"() {
setup:
def up = Mock(Upstream) {
nodeId() >> (byte)5
}
def matcher = new Selector.SameNodeMatcher((byte)1)
when:
def act = matcher.matches(up)
then:
!act
}
}

29
src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy
View File

@@ -179,4 +179,33 @@ class EthereumCallSelectorSpec extends Specification {
then:
act == new Selector.HeightMatcher(100)
}
def "Get same matcher for getFilterChanges method"() {
setup:
def callSelector = new EthereumCallSelector(Mock(Reader))
def head = Mock(Head)
expect:
callSelector.getMatcher("eth_getFilterChanges", param, head).block()
== new Selector.SameNodeMatcher((byte)hash)
where:
param | hash
'["0xff09"]' | 9
'["0xff"]' | 255
'[""]' | 0
'["0x0"]' | 0
}
def "Get empty matcher for getFilterChanges method without params"() {
setup:
def callSelector = new EthereumCallSelector(Mock(Reader))
def head = Mock(Head)
when:
def act = callSelector.getMatcher("eth_getFilterChanges", "[]", head).block()
then:
act == null
}
}

16
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy
View File

@@ -30,6 +30,7 @@ class EthereumDirectReaderSpec extends Specification {
String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String address1 = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d"
List<Byte> resolvers = Collections.singletonList((byte)1)
def "Reads block by hash"() {
setup:
@@ -53,8 +54,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
)
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers)
)
}
}
@@ -84,7 +84,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers
)
)
}
@@ -119,7 +119,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers
)
)
}
@@ -155,7 +155,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers
)
)
}
@@ -186,7 +186,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers
)
)
}
@@ -217,7 +217,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers
)
)
}
@@ -249,7 +249,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers
)
)
}

8
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy
View File

@@ -50,6 +50,8 @@ class EthereumGrpcUpstreamSpec extends Specification {
Counter.builder("test2").register(TestingCommons.meterRegistry)
)
def hash = (byte)123
def "Subscribe to head"() {
setup:
def callData = [:]
@@ -81,7 +83,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -139,7 +141,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -201,7 +203,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
finished.complete(true)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))

34
src/test/resources/upstreams-node-id.yaml Normal file
View File

@@ -0,0 +1,34 @@
version: v1
upstreams:
- id: has_node_id
node-id: 1
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
- id: has_no_node_id
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
ws:
url: "ws://localhost:8546"
- id: conflicted_node_id
node-id: 1
chain: ethereum
connection:
prefer-http: true
ethereum:
rpc:
url: "http://localhost:9545"
ws:
url: "ws://localhost:9546"
- id: invalid_node_id
node-id: 256
chain: ethereum
connection:
grpc:
host: "localhost"
port: 2449