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

generic upstreams introduced (#317)

Browse Source 
- added base implementation of upstreams/multistreams
- connectors were generalised, chain specific logic was extracted to ChainSpecific class
- StarkNet network support based on generic upstreams added
This commit is contained in:
a10zn8
2023-10-09 23:17:16 +04:00
committed by GitHub
parent 222e046e32
commit f3bf34768a
56 changed files with 864 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,7 +1,7 @@
package io.emeraldpay.dshackle
enum class BlockchainType {
BITCOIN, EVM_POW, EVM_POS;
BITCOIN, EVM_POW, EVM_POS, STARKNET;
companion object {
@@ -10,12 +10,16 @@ enum class BlockchainType {
)
val bitcoin = setOf(Chain.BITCOIN__MAINNET, Chain.BITCOIN__TESTNET)
val starknet = setOf(Chain.STARKNET__MAINNET, Chain.STARKNET__TESTNET, Chain.STARKNET__TESTNET_2)
@JvmStatic
fun from(chain: Chain): BlockchainType {
return if (pow.contains(chain)) {
EVM_POW
} else if (bitcoin.contains(chain)) {
BITCOIN
} else if (starknet.contains(chain)) {
STARKNET
} else {
EVM_POS
}

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

@@ -17,7 +17,7 @@
package io.emeraldpay.dshackle.config
import io.emeraldpay.dshackle.foundation.ChainOptions
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode
import java.net.URI
import java.util.Arrays
import java.util.Locale
@@ -58,23 +58,11 @@ data class UpstreamsConfig(
open class UpstreamConnection
open class RpcConnection(
open var rpc: HttpEndpoint? = null,
) : UpstreamConnection()
class GrpcConnection : UpstreamConnection() {
var host: String? = null
var port: Int = 0
var auth: AuthConfig.ClientTlsAuth? = null
var tokenAuth: AuthConfig.ClientTokenAuth? = null
var upstreamRating: Int = 0
}
data class EthereumConnection(
override var rpc: HttpEndpoint? = null,
data class RpcConnection(
var rpc: HttpEndpoint? = null,
var ws: WsEndpoint? = null,
var connectorMode: String? = null,
) : RpcConnection(rpc) {
) : UpstreamConnection() {
fun resolveMode(): ConnectorMode {
return if (connectorMode == null) {
@@ -91,14 +79,22 @@ data class UpstreamsConfig(
}
}
class GrpcConnection : UpstreamConnection() {
var host: String? = null
var port: Int = 0
var auth: AuthConfig.ClientTlsAuth? = null
var tokenAuth: AuthConfig.ClientTokenAuth? = null
var upstreamRating: Int = 0
}
data class BitcoinConnection(
override var rpc: HttpEndpoint? = null,
var rpc: HttpEndpoint? = null,
var esplora: HttpEndpoint? = null,
var zeroMq: BitcoinZeroMq? = null,
) : RpcConnection()
) : UpstreamConnection()
data class EthereumPosConnection(
var execution: EthereumConnection? = null,
var execution: RpcConnection? = null,
var upstreamRating: Int = 0,
) : UpstreamConnection()

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

@@ -86,9 +86,13 @@ class UpstreamsConfigReader(
getList<MappingNode>(input, "upstreams")?.value?.forEach { upNode ->
val connNode = getMapping(upNode, "connection")
if (hasAny(connNode, "ethereum")) {
if (hasAny(connNode, "generic")) {
readUpstream(config, upNode) {
readEthereumConnection(getMapping(connNode, "ethereum")!!)
readRpcConnection(getMapping(connNode, "generic")!!)
}
} else if (hasAny(connNode, "ethereum")) {
readUpstream(config, upNode) {
readRpcConnection(getMapping(connNode, "ethereum")!!)
}
} else if (hasAny(connNode, "bitcoin")) {
readUpstream(config, upNode) {
@@ -175,7 +179,7 @@ class UpstreamsConfigReader(
private fun readEthereumPosConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumPosConnection {
val connection = UpstreamsConfig.EthereumPosConnection()
getMapping(connConfigNode, "execution")?.let {
connection.execution = readEthereumConnection(it)
connection.execution = readRpcConnection(it)
}
getValueAsInt(connConfigNode, "upstream-rating")?.let {
connection.upstreamRating = it
@@ -183,8 +187,8 @@ class UpstreamsConfigReader(
return connection
}
private fun readEthereumConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumConnection {
val connection = UpstreamsConfig.EthereumConnection()
private fun readRpcConnection(connConfigNode: MappingNode): UpstreamsConfig.RpcConnection {
val connection = UpstreamsConfig.RpcConnection()
.apply { rpc = readRpcConfig(connConfigNode) }
getValueAsString(connConfigNode, "connector-mode")?.let {

26
src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt
View File

@@ -1,6 +1,10 @@
package io.emeraldpay.dshackle.config.context
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.BlockchainType.BITCOIN
import io.emeraldpay.dshackle.BlockchainType.EVM_POS
import io.emeraldpay.dshackle.BlockchainType.EVM_POW
import io.emeraldpay.dshackle.BlockchainType.STARKNET
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
@@ -8,6 +12,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory
import org.springframework.cloud.sleuth.Tracer
@@ -30,13 +35,28 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
.filterNot { it == Chain.UNSPECIFIED }
.map { chain ->
when (BlockchainType.from(chain)) {
BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler, tracer)
BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler, tracer)
BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory, headScheduler)
EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler, tracer)
EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler, tracer)
BITCOIN -> bitcoinMultistream(chain, cachesFactory, headScheduler)
STARKNET -> genericMultistream(chain, cachesFactory, headScheduler)
}
}
}
private fun genericMultistream(
chain: Chain,
cachesFactory: CachesFactory,
headScheduler: Scheduler,
): Multistream {
val name = "multi-$chain"
return GenericMultistream(
chain,
CopyOnWriteArrayList(),
cachesFactory.getCaches(chain),
headScheduler,
).also { register(it, name) }
}
private fun ethereumMultistream(
chain: Chain,
cachesFactory: CachesFactory,

7
src/main/kotlin/io/emeraldpay/dshackle/data/BlockId.kt
View File

@@ -47,7 +47,12 @@ class BlockId(
} else {
id
}
val bytes = Hex.decode(clean)
val even = if (clean.length % 2 != 0) {
"0$clean"
} else {
clean
}
val bytes = Hex.decode(even)
return BlockId(bytes)
}
}

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

@@ -19,15 +19,25 @@ package io.emeraldpay.dshackle.startup
import brave.grpc.GrpcTracing
import com.google.common.annotations.VisibleForTesting
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.BlockchainType.BITCOIN
import io.emeraldpay.dshackle.BlockchainType.EVM_POS
import io.emeraldpay.dshackle.BlockchainType.EVM_POW
import io.emeraldpay.dshackle.BlockchainType.STARKNET
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.AuthorizationConfig
import io.emeraldpay.dshackle.config.ChainsConfig
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.config.CompressionConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig.BitcoinConnection
import io.emeraldpay.dshackle.config.UpstreamsConfig.EthereumPosConnection
import io.emeraldpay.dshackle.config.UpstreamsConfig.HttpEndpoint
import io.emeraldpay.dshackle.config.UpstreamsConfig.RpcConnection
import io.emeraldpay.dshackle.foundation.ChainOptions
import io.emeraldpay.dshackle.foundation.ChainOptions.Options
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
@@ -45,13 +55,14 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionFactory
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionFactory
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcAuthContext
import io.grpc.ClientInterceptor
@@ -128,27 +139,37 @@ open class ConfiguredUpstreams(
.merge(up.options ?: ChainOptions.PartialOptions())
.buildOptions()
val upstream = when (BlockchainType.from(chain)) {
BlockchainType.EVM_POS -> {
EVM_POS -> {
buildEthereumPosUpstream(
up.nodeId,
up.cast(UpstreamsConfig.EthereumPosConnection::class.java),
up.cast(EthereumPosConnection::class.java),
chain,
options,
chainConfig,
)
}
BlockchainType.EVM_POW -> {
EVM_POW -> {
buildEthereumUpstream(
up.nodeId,
up.cast(UpstreamsConfig.EthereumConnection::class.java),
up.cast(RpcConnection::class.java),
chain,
options,
chainConfig,
)
}
BlockchainType.BITCOIN -> {
BITCOIN -> {
buildBitcoinUpstream(
up.cast(UpstreamsConfig.BitcoinConnection::class.java),
up.cast(BitcoinConnection::class.java),
chain,
options,
chainConfig,
)
}
STARKNET -> {
buildStarknetUpstream(
up.nodeId,
up.cast(RpcConnection::class.java),
chain,
options,
chainConfig,
@@ -204,12 +225,66 @@ open class ConfiguredUpstreams(
}
}
private fun buildStarknetUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<RpcConnection>,
chain: Chain,
options: Options,
chainConfig: ChainConfig,
): Upstream? {
if (config.connection == null) {
log.warn("Upstream doesn't have connection configuration")
return null
}
val connection = config.connection!!
val connectorFactory = buildConnectorFactory(
config.id!!,
connection,
chain,
NoChoiceWithPriorityForkChoice(0, config.id!!),
BlockValidator.ALWAYS_VALID,
chainConfig,
)
if (connectorFactory == null) {
return null
}
val methods = buildMethods(config, chain)
val hashUrl = connection.let {
if (it.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
}
val hash = getHash(nodeId, hashUrl!!)
val upstream = GenericUpstream(
config.id!!,
chain,
hash,
options,
config.role,
methods,
QuorumForLabels.QuorumItem(1, config.labels),
chainConfig,
connectorFactory,
eventPublisher,
)
upstream.start()
if (!upstream.isRunning) {
log.debug("Upstream ${upstream.getId()} is not running, it can't be added")
return null
}
return upstream
}
private fun buildEthereumPosUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumPosConnection>,
config: UpstreamsConfig.Upstream<EthereumPosConnection>,
chain: Chain,
options: ChainOptions.Options,
chainConf: ChainsConfig.ChainConfig,
options: Options,
chainConf: ChainConfig,
): Upstream? {
val conn = config.connection!!
val execution = conn.execution
@@ -218,11 +293,10 @@ open class ConfiguredUpstreams(
return null
}
val urls = ArrayList<URI>()
val connectorFactory = buildEthereumConnectorFactory(
val connectorFactory = buildConnectorFactory(
config.id!!,
execution,
chain,
urls,
NoChoiceWithPriorityForkChoice(conn.upstreamRating, config.id!!),
BlockValidator.ALWAYS_VALID,
chainConf,
@@ -258,13 +332,13 @@ open class ConfiguredUpstreams(
}
private fun buildBitcoinUpstream(
config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
config: UpstreamsConfig.Upstream<BitcoinConnection>,
chain: Chain,
options: ChainOptions.Options,
chainConf: ChainsConfig.ChainConfig,
options: Options,
chainConf: ChainConfig,
): Upstream? {
val conn = config.connection!!
val httpFactory = buildHttpFactory(conn)
val httpFactory = buildHttpFactory(conn.rpc)
if (httpFactory == null) {
log.warn("Upstream doesn't have API configuration")
return null
@@ -302,21 +376,18 @@ open class ConfiguredUpstreams(
private fun buildEthereumUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
config: UpstreamsConfig.Upstream<RpcConnection>,
chain: Chain,
options: ChainOptions.Options,
chainConf: ChainsConfig.ChainConfig,
options: Options,
chainConf: ChainConfig,
): Upstream? {
val conn = config.connection!!
val urls = ArrayList<URI>()
val methods = buildMethods(config, chain)
val connectorFactory = buildEthereumConnectorFactory(
val connectorFactory = buildConnectorFactory(
config.id!!,
conn,
chain,
urls,
MostWorkForkChoice(),
EthereumBlockValidator(),
chainConf,
@@ -388,26 +459,26 @@ open class ConfiguredUpstreams(
.subscribe(eventPublisher::publishEvent)
}
private fun buildHttpFactory(conn: UpstreamsConfig.RpcConnection, urls: ArrayList<URI>? = null): HttpRpcFactory? {
return conn.rpc?.let { endpoint ->
val tls = conn.rpc?.tls?.let { tls ->
private fun buildHttpFactory(conn: HttpEndpoint?, urls: ArrayList<URI>? = null): HttpRpcFactory? {
return conn?.let { endpoint ->
val tls = conn?.tls?.let { tls ->
tls.ca?.let { ca ->
fileResolver.resolve(ca).readBytes()
}
}
urls?.add(endpoint.url)
HttpRpcFactory(endpoint.url.toString(), conn.rpc?.basicAuth, tls)
HttpRpcFactory(endpoint.url.toString(), conn?.basicAuth, tls)
}
}
private fun buildWsFactory(
id: String,
chain: Chain,
conn: UpstreamsConfig.EthereumConnection,
conn: RpcConnection,
urls: ArrayList<URI>? = null,
): EthereumWsConnectionPoolFactory? {
): WsConnectionPoolFactory? {
return conn.ws?.let { endpoint ->
val wsConnectionFactory = EthereumWsConnectionFactory(
val wsConnectionFactory = WsConnectionFactory(
id,
chain,
endpoint.url,
@@ -417,7 +488,7 @@ open class ConfiguredUpstreams(
config = endpoint
basicAuth = endpoint.basicAuth
}
val wsApi = EthereumWsConnectionPoolFactory(
val wsApi = WsConnectionPoolFactory(
id,
endpoint.connections,
wsConnectionFactory,
@@ -427,20 +498,20 @@ open class ConfiguredUpstreams(
}
}
private fun buildEthereumConnectorFactory(
private fun buildConnectorFactory(
id: String,
conn: UpstreamsConfig.EthereumConnection,
conn: RpcConnection,
chain: Chain,
urls: ArrayList<URI>,
forkChoice: ForkChoice,
blockValidator: BlockValidator,
chainsConf: ChainsConfig.ChainConfig,
): EthereumConnectorFactory? {
chainsConf: ChainConfig,
): GenericConnectorFactory? {
val urls = ArrayList<URI>()
val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
val httpFactory = buildHttpFactory(conn, urls)
val httpFactory = buildHttpFactory(conn.rpc, urls)
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val connectorFactory =
EthereumConnectorFactory(
GenericConnectorFactory(
conn.resolveMode(),
wsFactoryApi,
httpFactory,

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

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultStarknetMethods
import org.springframework.stereotype.Component
@Component
@@ -20,7 +21,7 @@ class CallTargetsHolder {
BlockchainType.EVM_POW -> DefaultEthereumMethods(chain)
BlockchainType.BITCOIN -> DefaultBitcoinMethods()
BlockchainType.EVM_POS -> DefaultEthereumMethods(chain)
else -> throw IllegalStateException("Unsupported chain: $chain")
BlockchainType.STARKNET -> DefaultStarknetMethods(chain)
}
callTargets[chain] = created
return created

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

@@ -1,29 +0,0 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.data.BlockContainer
class DistanceExtractor {
sealed class ChainDistance {
data class Distance(val dist: Long) : ChainDistance()
object Fork : ChainDistance()
}
companion object {
fun extractPowDistance(top: BlockContainer, curr: BlockContainer): ChainDistance {
return when {
curr.height > top.height -> if (curr.difficulty >= top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork
curr.height == top.height -> if (curr.difficulty == top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork
else -> ChainDistance.Distance(top.height - curr.height)
}
}
fun extractPriorityDistance(top: BlockContainer, curr: BlockContainer): ChainDistance {
return when {
(curr.parentHash != null && curr.height - top.height == 1L) ->
if (curr.parentHash == top.hash) ChainDistance.Distance(0) else ChainDistance.Fork
curr.height == top.height -> if (curr.hash == top.hash) ChainDistance.Distance(0) else ChainDistance.Fork
else -> ChainDistance.Distance((top.height - curr.height).coerceAtLeast(0))
}
}
}
}

31
src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
View File

@@ -31,11 +31,12 @@ import java.time.Duration
* other upstreams.
*/
typealias Extractor = (top: BlockContainer, curr: BlockContainer) -> DistanceExtractor.ChainDistance
abstract class HeadLagObserver(
class HeadLagObserver(
private val master: Head,
private val followers: Collection<Upstream>,
private val distanceExtractor: Extractor,
private val lagObserverScheduler: Scheduler,
private val forkDistance: Long,
private val throttling: Duration = Duration.ofSeconds(5),
) : Lifecycle {
@@ -102,5 +103,31 @@ abstract class HeadLagObserver(
}
}
abstract fun forkDistance(top: BlockContainer, curr: BlockContainer): Long
fun forkDistance(top: BlockContainer, curr: BlockContainer) = forkDistance
}
class DistanceExtractor {
sealed class ChainDistance {
data class Distance(val dist: Long) : ChainDistance()
object Fork : ChainDistance()
}
companion object {
fun extractPowDistance(top: BlockContainer, curr: BlockContainer): ChainDistance {
return when {
curr.height > top.height -> if (curr.difficulty >= top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork
curr.height == top.height -> if (curr.difficulty == top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork
else -> ChainDistance.Distance(top.height - curr.height)
}
}
fun extractPriorityDistance(top: BlockContainer, curr: BlockContainer): ChainDistance {
return when {
(curr.parentHash != null && curr.height - top.height == 1L) ->
if (curr.parentHash == top.hash) ChainDistance.Distance(0) else ChainDistance.Fork
curr.height == top.height -> if (curr.hash == top.hash) ChainDistance.Distance(0) else ChainDistance.Fork
else -> ChainDistance.Distance((top.height - curr.height).coerceAtLeast(0))
}
}
}
}

40
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt
View File

@@ -1,40 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HeadLagObserver
import io.emeraldpay.dshackle.upstream.Upstream
import org.slf4j.LoggerFactory
import reactor.core.scheduler.Scheduler
class BitcoinHeadLagObserver(
master: Head,
followers: Collection<Upstream>,
headScheduler: Scheduler,
) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) {
companion object {
private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java)
}
override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long {
// TODO fetch actual blocks
return 3
}
}

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

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.EgressSubscription
import io.emeraldpay.dshackle.upstream.EmptyEgressSubscription
import io.emeraldpay.dshackle.upstream.EmptyHead
@@ -153,7 +154,7 @@ open class BitcoinMultistream(
}
override fun makeLagObserver(): HeadLagObserver {
return BitcoinHeadLagObserver(head, sourceUpstreams, headScheduler)
return HeadLagObserver(head, sourceUpstreams, DistanceExtractor::extractPowDistance, headScheduler, 3)
}
override fun start() {

127
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultStarknetMethods.kt Normal file
View File

@@ -0,0 +1,127 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.calls
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.BroadcastQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.quorum.MaximumValueQuorum
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.emeraldpay.dshackle.quorum.NotNullQuorum
import io.emeraldpay.etherjar.rpc.RpcException
/**
* Default configuration for Ethereum based RPC. Defines optimal Quorum strategies for different methods, and provides
* hardcoded results for base methods, such as `net_version`, `web3_clientVersion` and similar
*/
class DefaultStarknetMethods(
private val chain: Chain,
) : CallMethods {
private val nonce = setOf(
"starknet_getNonce",
)
private val add = setOf(
"starknet_addDeployAccountTransaction",
"starknet_addDeclareTransaction",
"starknet_addInvokeTransaction",
)
private val anyResponseMethods = listOf(
"starknet_estimateFee",
"starknet_estimateMessageFee",
"starknet_specVersion",
)
private val nonNull = listOf(
"starknet_getBlockWithTxHashes",
"starknet_getBlockWithTxs",
"starknet_getTransactionStatus",
"starknet_getTransactionByHash",
"starknet_getTransactionByBlockIdAndIndex",
"starknet_getTransactionReceipt",
"starknet_getBlockTransactionCount",
)
private val firstValueMethods = listOf(
"starknet_call",
"starknet_getEvents",
"starknet_getStateUpdate",
"starknet_getStorageAt",
"starknet_getClass",
"starknet_getClassHashAt",
"starknet_getClassAt",
)
private val nonLagging = listOf(
"starknet_blockNumber",
"starknet_blockHashAndNumber",
)
private val hardcodedMethods = listOf(
"starknet_chainId",
"starknet_syncing",
)
private val allowedMethods: List<String> = anyResponseMethods + firstValueMethods + nonNull + nonce +
add + nonLagging
override fun createQuorumFor(method: String): CallQuorum {
return when {
nonce.contains(method) -> MaximumValueQuorum()
add.contains(method) -> BroadcastQuorum()
hardcodedMethods.contains(method) -> AlwaysQuorum()
nonLagging.contains(method) -> NotLaggingQuorum(1)
nonNull.contains(method) -> NotNullQuorum()
firstValueMethods.contains(method) -> AlwaysQuorum()
anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
else -> AlwaysQuorum()
}
}
override fun isCallable(method: String): Boolean {
return allowedMethods.contains(method)
}
override fun isHardcoded(method: String): Boolean {
return hardcodedMethods.contains(method)
}
override fun executeHardcoded(method: String): ByteArray {
val json = when (method) {
"starknet_chainId" -> "\"${chain.chainId}\""
"starknet_syncing" -> {
"false"
}
else -> throw RpcException(-32601, "Method not found")
}
return json.toByteArray()
}
override fun getGroupMethods(groupName: String): Set<String> =
when (groupName) {
"default" -> getSupportedMethods()
else -> emptyList()
}.toSet()
override fun getSupportedMethods(): Set<String> {
return allowedMethods.plus(hardcodedMethods).toSortedSet()
}
}

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt Normal file
View File

@@ -0,0 +1,14 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
object EthereumChainSpecific : ChainSpecific {
override fun parseBlock(data: JsonRpcResponse, upstreamId: String): BlockContainer {
return BlockContainer.fromEthereumJson(data.getResult(), upstreamId)
}
override fun latestBlockRequest() = JsonRpcRequest("eth_getBlockByNumber", listOf("latest", false))
}

41
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt
View File

@@ -1,41 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2020 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HeadLagObserver
import io.emeraldpay.dshackle.upstream.Upstream
import org.slf4j.LoggerFactory
import reactor.core.scheduler.Scheduler
class EthereumHeadLagObserver(
master: Head,
followers: Collection<Upstream>,
headScheduler: Scheduler,
) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance, headScheduler) {
companion object {
private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java)
}
override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long {
// TODO look for common ancestor? though it may be a corruption
return 6
}
}

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

@@ -33,9 +33,9 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator.ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator.ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator.ValidateUpstreamSettingsResult.UPSTREAM_VALID
import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector
import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
import org.springframework.context.ApplicationEventPublisher
import org.springframework.context.Lifecycle
import reactor.core.Disposable
@@ -57,7 +57,7 @@ open class EthereumLikeRpcUpstream(
private val eventPublisher: ApplicationEventPublisher?,
) : EthereumLikeUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(chain, this, getOptions(), chainConfig.callLimitContract)
protected val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance)
protected val connector: GenericConnector = connectorFactory.create(this, chain, skipEnhance)
private val labelsDetector = EthereumLabelsDetector(this.getIngressReader(), chain)
private var hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false)

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

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.DynamicMergedHead
import io.emeraldpay.dshackle.upstream.EgressSubscription
import io.emeraldpay.dshackle.upstream.EmptyHead
@@ -145,9 +146,10 @@ open class EthereumMultistream(
head.removeHead(upstreamId)
}
override fun makeLagObserver(): HeadLagObserver {
return EthereumHeadLagObserver(head, upstreams as Collection<Upstream>, headScheduler)
}
override fun makeLagObserver(): HeadLagObserver =
HeadLagObserver(head, upstreams, DistanceExtractor::extractPowDistance, headScheduler, 6).apply {
start()
}
override fun isRunning(): Boolean {
return super.isRunning() || reader.isRunning()

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -28,6 +28,8 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
import io.emeraldpay.dshackle.upstream.generic.GenericHead
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.domain.BlockHash
@@ -48,7 +50,8 @@ class EthereumWsHead(
private val wsConnectionResubscribeScheduler: Scheduler,
private val headScheduler: Scheduler,
private val upstream: DefaultUpstream,
) : DefaultEthereumHead(upstream.getId(), forkChoice, blockValidator, headScheduler), Lifecycle {
chainSpecific: ChainSpecific,
) : GenericHead(upstream.getId(), forkChoice, blockValidator, headScheduler, chainSpecific), Lifecycle {
private var connectionId: String? = null
private var subscribed = false

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt
View File

@@ -11,7 +11,7 @@ import io.micrometer.core.instrument.Timer
import reactor.core.scheduler.Scheduler
import java.net.URI
open class EthereumWsConnectionFactory(
open class WsConnectionFactory(
private val id: String,
private val chain: Chain,
private val uri: URI,

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

@@ -38,7 +38,7 @@ import kotlin.concurrent.write
* By default, it adds a new connection every 5 second until it reaches the target number.
*/
class WsConnectionMultiPool(
private val ethereumWsConnectionFactory: EthereumWsConnectionFactory,
private val wsConnectionFactory: WsConnectionFactory,
private val upstream: DefaultUpstream,
private val connections: Int,
) : WsConnectionPool {
@@ -110,7 +110,7 @@ class WsConnectionMultiPool(
SCHEDULE_FULL
} else {
current.add(
ethereumWsConnectionFactory.createWsConnection(connIndex++) {
wsConnectionFactory.createWsConnection(connIndex++) {
if (isUnavailable()) {
upstream.setStatus(UpstreamAvailability.UNAVAILABLE)
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPoolFactory.kt
View File

@@ -18,10 +18,10 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.upstream.DefaultUpstream
class EthereumWsConnectionPoolFactory(
class WsConnectionPoolFactory(
private val id: String,
private val connections: Int,
private val ethereumWsConnectionFactory: EthereumWsConnectionFactory,
private val wsConnectionFactory: WsConnectionFactory,
) {
fun create(upstream: DefaultUpstream): WsConnectionPool {
@@ -29,9 +29,9 @@ class EthereumWsConnectionPoolFactory(
"Creating instance for different upstream. ${upstream.getId()} != id"
}
return if (connections > 1) {
WsConnectionMultiPool(ethereumWsConnectionFactory, upstream, connections)
WsConnectionMultiPool(wsConnectionFactory, upstream, connections)
} else {
WsConnectionSinglePool(ethereumWsConnectionFactory, upstream)
WsConnectionSinglePool(wsConnectionFactory, upstream)
}
}
}

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

@@ -20,10 +20,10 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import reactor.core.publisher.Flux
class WsConnectionSinglePool(
ethereumWsConnectionFactory: EthereumWsConnectionFactory,
wsConnectionFactory: WsConnectionFactory,
private val upstream: DefaultUpstream,
) : WsConnectionPool {
private val connection = ethereumWsConnectionFactory.createWsConnection {
private val connection = wsConnectionFactory.createWsConnection {
upstream.setStatus(UpstreamAvailability.UNAVAILABLE)
}

24
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt
View File

@@ -1,24 +0,0 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HeadLagObserver
import io.emeraldpay.dshackle.upstream.Upstream
import org.slf4j.LoggerFactory
import reactor.core.scheduler.Scheduler
class EthereumPosHeadLagObserver(
master: Head,
followers: Collection<Upstream>,
headScheduler: Scheduler,
) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance, headScheduler) {
companion object {
private val log = LoggerFactory.getLogger(EthereumPosHeadLagObserver::class.java)
}
override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long {
return 6
}
}

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

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.DynamicMergedHead
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
@@ -107,7 +108,7 @@ open class EthereumPosMultiStream(
}
override fun makeLagObserver(): HeadLagObserver =
EthereumPosHeadLagObserver(head, upstreams, headScheduler).apply {
HeadLagObserver(head, upstreams, DistanceExtractor::extractPriorityDistance, headScheduler, 6).apply {
start()
}

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt Normal file
View File

@@ -0,0 +1,25 @@
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.BlockchainType.STARKNET
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific
interface ChainSpecific {
fun parseBlock(data: JsonRpcResponse, upstreamId: String): BlockContainer
fun latestBlockRequest(): JsonRpcRequest
}
object ChainSpecificRegistry {
fun resolve(chain: Chain): ChainSpecific {
if (BlockchainType.from(chain) == STARKNET) {
return StarknetChainSpecific
}
return EthereumChainSpecific
}
}

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericHead.kt
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.data.BlockContainer
@@ -22,24 +22,22 @@ import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
open class DefaultEthereumHead(
open class GenericHead(
protected val upstreamId: String,
forkChoice: ForkChoice,
blockValidator: BlockValidator,
private val headScheduler: Scheduler,
private val chainSpecific: ChainSpecific,
) : Head, AbstractHead(forkChoice, headScheduler, blockValidator, 60_000, upstreamId) {
fun getLatestBlock(api: JsonRpcReader): Mono<BlockContainer> {
return api.read(JsonRpcRequest("eth_getBlockByNumber", listOf("latest", false)))
return api.read(chainSpecific.latestBlockRequest())
.subscribeOn(headScheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
.map {
BlockContainer.fromEthereumJson(it.getResult(), upstreamId)
}
.map { chainSpecific.parseBlock(it, upstreamId) }
.onErrorResume { err ->
log.error("Failed to fetch latest block: ${err.message} $upstreamId", err)
Mono.empty()

112
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt Normal file
View File

@@ -0,0 +1,112 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2020 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.ChainFees
import io.emeraldpay.dshackle.upstream.DistanceExtractor
import io.emeraldpay.dshackle.upstream.DynamicMergedHead
import io.emeraldpay.dshackle.upstream.EgressSubscription
import io.emeraldpay.dshackle.upstream.EmptyEgressSubscription
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HeadLagObserver
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
@Suppress("UNCHECKED_CAST")
open class GenericMultistream(
chain: Chain,
val upstreams: MutableList<GenericUpstream>,
caches: Caches,
private val headScheduler: Scheduler,
) : Multistream(chain, upstreams as MutableList<Upstream>, caches) {
private var head: DynamicMergedHead = DynamicMergedHead(
PriorityForkChoice(),
"Multistream of ${chain.chainCode}",
headScheduler,
)
init {
this.init()
}
override fun init() {
if (upstreams.size > 0) {
upstreams.forEach { addHead(it) }
}
super.init()
}
override fun start() {
super.start()
head.start()
onHeadUpdated(head)
}
override fun addHead(upstream: Upstream) {
val newHead = upstream.getHead()
if (newHead is Lifecycle && !newHead.isRunning()) {
newHead.start()
}
head.addHead(upstream)
}
override fun removeHead(upstreamId: String) {
head.removeHead(upstreamId)
}
override fun makeLagObserver(): HeadLagObserver =
HeadLagObserver(head, upstreams, DistanceExtractor::extractPriorityDistance, headScheduler, 6).apply {
start()
}
override fun getHead(): Head {
return head
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return upstreams.flatMap { it.getLabels() }
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return this as T
}
override fun getLocalReader(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(LocalReader(getMethods()))
}
override fun getEgressSubscription(): EgressSubscription {
return EmptyEgressSubscription()
}
override fun getFeeEstimation(): ChainFees {
throw NotImplementedError()
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericRpcHead.kt
View File

@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
@@ -25,14 +25,15 @@ import reactor.core.publisher.Flux
import reactor.core.scheduler.Scheduler
import java.time.Duration
class EthereumRpcHead(
class GenericRpcHead(
private val api: JsonRpcReader,
forkChoice: ForkChoice,
upstreamId: String,
blockValidator: BlockValidator,
private val headScheduler: Scheduler,
private val chainSpecific: ChainSpecific,
private val interval: Duration = Duration.ofSeconds(10),
) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle {
) : GenericHead(upstreamId, forkChoice, blockValidator, headScheduler, chainSpecific), Lifecycle {
private var refreshSubscription: Disposable? = null
private var isSyncing = false

101
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt Normal file
View File

@@ -0,0 +1,101 @@
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.ChainsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig.Labels
import io.emeraldpay.dshackle.foundation.ChainOptions
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent.ChangeType.UPDATED
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
import org.springframework.context.ApplicationEventPublisher
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import java.util.concurrent.atomic.AtomicBoolean
class GenericUpstream(
id: String,
val chain: Chain,
hash: Byte,
options: ChainOptions.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?,
val chainConfig: ChainsConfig.ChainConfig,
connectorFactory: ConnectorFactory,
private val eventPublisher: ApplicationEventPublisher?,
) : DefaultUpstream(id, hash, null, UpstreamAvailability.OK, options, role, targets, node, chainConfig), Lifecycle {
private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false)
private val connector: GenericConnector = connectorFactory.create(this, chain, true)
private var livenessSubscription: Disposable? = null
override fun getHead(): Head {
return connector.getHead()
}
override fun getIngressReader(): JsonRpcReader {
return connector.getIngressReader()
}
override fun getLabels(): Collection<Labels> {
return node?.let { listOf(it.labels) } ?: emptyList()
}
override fun getSubscriptionTopics(): List<String> {
// should be implemented in next iterations
// starknet doesn't have any subscriptions at all
// polkadot serves subscriptions like separate json-rpc methods
return emptyList()
}
// outdated, looks like applicable only for bitcoin and our ws_head trick
override fun getCapabilities(): Set<Capability> {
return if (hasLiveSubscriptionHead.get()) {
setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD)
} else {
setOf(Capability.RPC, Capability.BALANCE)
}
}
override fun isGrpc(): Boolean {
// this implementation works only with statically configured upstreams
return false
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return this as T
}
override fun start() {
log.info("Configured for ${chain.chainName}")
connector.start()
livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({
hasLiveSubscriptionHead.set(it)
eventPublisher?.publishEvent(UpstreamChangeEvent(chain, this, UPDATED))
}, {
log.debug("Error while checking live subscription for ${getId()}", it)
},)
}
override fun stop() {
livenessSubscription?.dispose()
livenessSubscription = null
connector.stop()
}
override fun isRunning() = connector.isRunning()
}

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/LocalReader.kt Normal file
View File

@@ -0,0 +1,22 @@
package io.emeraldpay.dshackle.upstream.generic
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import reactor.core.publisher.Mono
class LocalReader(private val methods: CallMethods) : JsonRpcReader {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
if (methods.isHardcoded(key.method)) {
return Mono.just(methods.executeHardcoded(key.method))
.map { JsonRpcResponse(it, null) }
}
if (!methods.isCallable(key.method)) {
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
return Mono.empty()
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/ConnectorFactory.kt
View File

@@ -1,16 +1,14 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
package io.emeraldpay.dshackle.upstream.generic.connectors
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
interface ConnectorFactory {
fun create(
upstream: DefaultUpstream,
validator: EthereumUpstreamValidator,
chain: Chain,
skipEnhance: Boolean,
): EthereumConnector
): GenericConnector
fun isValid(): Boolean
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt
View File

@@ -1,4 +1,4 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
package io.emeraldpay.dshackle.upstream.generic.connectors
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.Head
@@ -6,7 +6,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import reactor.core.publisher.Flux
interface EthereumConnector : Lifecycle {
interface GenericConnector : Lifecycle {
fun getHead(): Head
fun hasLiveSubscriptionHead(): Flux<Boolean>

30
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactory.kt
View File

@@ -1,22 +1,22 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
package io.emeraldpay.dshackle.upstream.generic.connectors
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.HttpFactory
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_ONLY
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.WS_ONLY
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_ONLY
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.WS_ONLY
import reactor.core.scheduler.Scheduler
import java.time.Duration
open class EthereumConnectorFactory(
open class GenericConnectorFactory(
private val connectorType: ConnectorMode,
private val wsFactory: EthereumWsConnectionPoolFactory?,
private val wsFactory: WsConnectionPoolFactory?,
private val httpFactory: HttpFactory?,
private val forkChoice: ForkChoice,
private val blockValidator: BlockValidator,
@@ -47,12 +47,12 @@ open class EthereumConnectorFactory(
override fun create(
upstream: DefaultUpstream,
validator: EthereumUpstreamValidator,
chain: Chain,
skipEnhance: Boolean,
): EthereumConnector {
): GenericConnector {
val specific = ChainSpecificRegistry.resolve(chain)
if (wsFactory != null && connectorType == WS_ONLY) {
return EthereumWsConnector(
return GenericWsConnector(
wsFactory,
upstream,
forkChoice,
@@ -61,12 +61,13 @@ open class EthereumConnectorFactory(
wsConnectionResubscribeScheduler,
headScheduler,
expectedBlockTime,
specific,
)
}
if (httpFactory == null) {
throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set")
}
return EthereumRpcConnector(
return GenericRpcConnector(
connectorType,
httpFactory.create(upstream.getId(), chain),
wsFactory,
@@ -77,6 +78,7 @@ open class EthereumConnectorFactory(
wsConnectionResubscribeScheduler,
headScheduler,
expectedBlockTime,
specific,
)
}
@@ -88,7 +90,7 @@ open class EthereumConnectorFactory(
;
companion object {
val values = values().map { it.name }.toSet()
val values = entries.map { it.name }.toSet()
fun parse(value: String): ConnectorMode {
val upper = value.uppercase()
if (!values.contains(upper)) {

33
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt
View File

@@ -1,4 +1,4 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
package io.emeraldpay.dshackle.upstream.generic.connectors
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
@@ -9,29 +9,30 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator
import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_ONLY
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.WS_ONLY
import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
import io.emeraldpay.dshackle.upstream.generic.GenericRpcHead
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_ONLY
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.WS_ONLY
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.scheduler.Scheduler
import java.time.Duration
class EthereumRpcConnector(
class GenericRpcConnector(
connectorType: ConnectorMode,
private val directReader: JsonRpcReader,
wsFactory: EthereumWsConnectionPoolFactory?,
wsFactory: WsConnectionPoolFactory?,
upstream: DefaultUpstream,
forkChoice: ForkChoice,
blockValidator: BlockValidator,
@@ -39,14 +40,15 @@ class EthereumRpcConnector(
wsConnectionResubscribeScheduler: Scheduler,
headScheduler: Scheduler,
expectedBlockTime: Duration,
) : EthereumConnector, CachesEnabled {
chainSpecific: ChainSpecific,
) : GenericConnector, CachesEnabled {
private val id = upstream.getId()
private val pool: WsConnectionPool?
private val head: Head
private val liveness: HeadLivenessValidator
companion object {
private val log = LoggerFactory.getLogger(EthereumRpcConnector::class.java)
private val log = LoggerFactory.getLogger(GenericRpcConnector::class.java)
}
override fun hasLiveSubscriptionHead(): Flux<Boolean> {
@@ -59,7 +61,7 @@ class EthereumRpcConnector(
head = when (connectorType) {
RPC_ONLY -> {
log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator, headScheduler)
GenericRpcHead(getIngressReader(), forkChoice, id, blockValidator, headScheduler, chainSpecific)
}
WS_ONLY -> {
@@ -77,15 +79,17 @@ class EthereumRpcConnector(
wsConnectionResubscribeScheduler,
headScheduler,
upstream,
chainSpecific,
)
// receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead =
EthereumRpcHead(
GenericRpcHead(
getIngressReader(),
AlwaysForkChoice(),
id,
blockValidator,
headScheduler,
chainSpecific,
Duration.ofSeconds(30),
)
MergedHead(listOf(rpcHead, wsHead), forkChoice, headScheduler, "Merged for $id")
@@ -101,6 +105,7 @@ class EthereumRpcConnector(
wsConnectionResubscribeScheduler,
headScheduler,
upstream,
chainSpecific,
)
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt
View File

@@ -1,24 +1,25 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
package io.emeraldpay.dshackle.upstream.generic.connectors
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
import reactor.core.publisher.Flux
import reactor.core.scheduler.Scheduler
import java.time.Duration
class EthereumWsConnector(
wsFactory: EthereumWsConnectionPoolFactory,
class GenericWsConnector(
wsFactory: WsConnectionPoolFactory,
upstream: DefaultUpstream,
forkChoice: ForkChoice,
blockValidator: BlockValidator,
@@ -26,7 +27,8 @@ class EthereumWsConnector(
wsConnectionResubscribeScheduler: Scheduler,
headScheduler: Scheduler,
expectedBlockTime: Duration,
) : EthereumConnector {
chainSpecific: ChainSpecific,
) : GenericConnector {
private val pool: WsConnectionPool
private val reader: JsonRpcReader
private val head: EthereumWsHead
@@ -45,6 +47,7 @@ class EthereumWsConnector(
wsConnectionResubscribeScheduler,
headScheduler,
upstream,
chainSpecific,
)
liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler, upstream.getId())
subscriptions = EthereumWsIngressSubscription(wsSubscriptions)

42
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt Normal file
View File

@@ -0,0 +1,42 @@
package io.emeraldpay.dshackle.upstream.starknet
import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonProperty
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import java.math.BigInteger
import java.time.Instant
object StarknetChainSpecific : ChainSpecific {
override fun parseBlock(data: JsonRpcResponse, upstreamId: String): BlockContainer {
val raw = data.getResult()
val block = Global.objectMapper.readValue(raw, StarknetBlock::class.java)
return BlockContainer(
height = block.number,
hash = BlockId.from(block.hash),
difficulty = BigInteger.ZERO,
timestamp = block.timestamp,
full = false,
json = raw,
parsed = block,
transactions = emptyList(),
upstreamId = upstreamId,
parentHash = BlockId.from(block.parent),
)
}
override fun latestBlockRequest(): JsonRpcRequest = JsonRpcRequest("starknet_getBlockWithTxHashes", listOf("latest"))
}
@JsonIgnoreProperties(ignoreUnknown = true)
data class StarknetBlock(
@JsonProperty("block_hash") var hash: String,
@JsonProperty("block_number") var number: Long,
@JsonProperty("timestamp") var timestamp: Instant,
@JsonProperty("parent_hash") var parent: String,
)

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

@@ -45,8 +45,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(0)) {
id == "local"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection)connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection)connection) {
rpc != null
rpc.url == new URI("http://localhost:8545")
ws != null
@@ -61,8 +61,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(1)) {
id == "infura"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection)connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection)connection) {
rpc.url == new URI("https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2")
rpc.basicAuth != null
with((AuthConfig.ClientBasicAuth) rpc.basicAuth) {
@@ -86,8 +86,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(0)) {
id == "local"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection) connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection) connection) {
rpc == null
ws != null
ws.url == new URI("ws://localhost:8546")
@@ -111,8 +111,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(0)) {
id == "local"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection) connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection) connection) {
rpc == null
ws != null
ws.url == new URI("ws://localhost:8546")
@@ -278,8 +278,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(0)) {
id == "local"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection)connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection)connection) {
rpc != null
rpc.url == new URI("http://localhost:8545")
ws == null
@@ -345,14 +345,14 @@ class UpstreamsConfigReaderSpec extends Specification {
def "Invalidate wrong ids"() {
expect:
!reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>(id: id))
!reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.RpcConnection>(id: id))
where:
id << ["", null, "a", "ab", "!ab", "foo bar", "foo@bar", "123test", "_test", "test/test"]
}
def "Accept good ids"() {
expect:
reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>(id: id))
reader.isValid(new UpstreamsConfig.Upstream<UpstreamsConfig.RpcConnection>(id: id))
where:
id << ["test", "test_test", "test-test", "test123", "test1test", "foo_bar_12"]
}
@@ -477,8 +477,8 @@ class UpstreamsConfigReaderSpec extends Specification {
with(act.upstreams.get(0)) {
id == "local"
chain == "ethereum"
connection instanceof UpstreamsConfig.EthereumConnection
with((UpstreamsConfig.EthereumConnection) connection) {
connection instanceof UpstreamsConfig.RpcConnection
with((UpstreamsConfig.RpcConnection) connection) {
rpc != null
rpc.url == new URI("https://localhost:8546")
ws != null

10
src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy
View File

@@ -5,10 +5,8 @@ import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -25,7 +23,7 @@ class ConnectorFactoryMock implements ConnectorFactory {
return true
}
EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain, boolean skipEnhance) {
return new EthereumConnectorMock(api, head)
GenericConnector create(DefaultUpstream upstream, Chain chain, boolean skipEnhance) {
return new GenericConnectorMock(api, head)
}
}

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

@@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.test
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.foundation.ChainOptions
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
@@ -91,8 +90,8 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream {
this.ethereumHeadMock.nextBlock(block)
}
EthereumConnectorMock getConnectorMock() {
return this.connector as EthereumConnectorMock
GenericConnectorMock getConnectorMock() {
return this.connector as GenericConnectorMock
}
void setBlocks(Publisher<BlockContainer> blocks) {

8
src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy → src/test/groovy/io/emeraldpay/dshackle/test/GenericConnectorMock.groovy
View File

@@ -4,19 +4,17 @@ import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Flux
class EthereumConnectorMock implements EthereumConnector {
class GenericConnectorMock implements GenericConnector {
Reader<JsonRpcRequest, JsonRpcResponse> api
Head head
Flux<Boolean> liveness
EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
GenericConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
this.api = api
this.head = head
this.liveness = Flux.just(false)

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

@@ -24,7 +24,7 @@ import io.emeraldpay.dshackle.test.EthereumApiStub
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import reactor.core.scheduler.Schedulers
import reactor.test.StepVerifier
@@ -53,8 +53,8 @@ class FilteredApisSpec extends Specification {
def httpFactory = Mock(HttpFactory) {
create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" }
}
def connectorFactory = new EthereumConnectorFactory(
EthereumConnectorFactory.ConnectorMode.RPC_ONLY,
def connectorFactory = new GenericConnectorFactory(
GenericConnectorFactory.ConnectorMode.RPC_ONLY,
null,
httpFactory,
new MostWorkForkChoice(),

19
src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy
View File

@@ -74,7 +74,8 @@ class HeadLagObserverSpec extends Specification {
1 * up2.setLag(1)
1 * up2.setLag(0)
HeadLagObserver observer = new TestHeadLagObserver(master, [up1, up2])
HeadLagObserver observer = new HeadLagObserver(master, [up1, up2], DistanceExtractor.@Companion::extractPowDistance,
Schedulers.boundedElastic(), 11, Duration.ofNanos(1))
when:
def act = observer.subscription().take(Duration.ofMillis(5000))
@@ -88,7 +89,8 @@ class HeadLagObserverSpec extends Specification {
def "Probes until there is no difference"() {
setup:
Head master = Mock()
HeadLagObserver observer = new TestHeadLagObserver(master, [])
HeadLagObserver observer = new HeadLagObserver(master, [], DistanceExtractor.@Companion::extractPowDistance,
Schedulers.boundedElastic(), 11, Duration.ofNanos(1))
Upstream up = Mock()
def blocks = [100, 101, 102].collect { i ->
@@ -112,17 +114,4 @@ class HeadLagObserverSpec extends Specification {
.expectNext(Tuples.of(0L, up))
.verifyComplete()
}
class TestHeadLagObserver extends HeadLagObserver {
TestHeadLagObserver(@NotNull Head master, @NotNull Collection<? extends Upstream> followers) {
super(master, followers, DistanceExtractor.@Companion::extractPowDistance,
Schedulers.boundedElastic(), Duration.ofNanos(1))
}
@Override
long forkDistance(@NotNull BlockContainer top, @NotNull BlockContainer curr) {
return 11
}
}
}

4
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
View File

@@ -15,10 +15,8 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.IngressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.etherjar.hex.Hex32

12
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy
View File

@@ -70,7 +70,7 @@ class EthereumWsHeadSpec extends Specification {
1 * it.connectionInfoFlux() >> Flux.empty()
}
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.listenNewHeads().blockFirst()
@@ -111,7 +111,7 @@ class EthereumWsHeadSpec extends Specification {
]
}
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.getFlux()
@@ -165,7 +165,7 @@ class EthereumWsHeadSpec extends Specification {
]
}
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.getFlux()
@@ -205,7 +205,7 @@ class EthereumWsHeadSpec extends Specification {
]
}
def head = new EthereumWsHead( new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead( new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.getFlux()
@@ -244,7 +244,7 @@ class EthereumWsHeadSpec extends Specification {
]
}
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.getFlux()
@@ -297,7 +297,7 @@ class EthereumWsHeadSpec extends Specification {
]
}
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream)
def head = new EthereumWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
when:
def act = head.getFlux()

5
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy → src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericHeadSpec.groovy
View File

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.generic.GenericHead
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
@@ -30,9 +31,9 @@ import spock.lang.Specification
import java.time.Instant
class DefaultEthereumHeadSpec extends Specification {
class GenericHeadSpec extends Specification {
DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic())
GenericHead head = new GenericHead("upstream", new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic(), EthereumChainSpecific.INSTANCE)
ObjectMapper objectMapper = Global.objectMapper
BlockHash parent = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915210")

8
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy
View File

@@ -35,10 +35,10 @@ class WsConnectionImplRealSpec extends Specification {
server = new MockWSServer(port)
server.start()
Thread.sleep(SLEEP)
conn = new EthereumWsConnectionPoolFactory(
conn = new WsConnectionPoolFactory(
"test",
1,
new EthereumWsConnectionFactory(
new WsConnectionFactory(
"test",
Chain.ETHEREUM__MAINNET,
"ws://localhost:${port}".toURI(),
@@ -115,10 +115,10 @@ class WsConnectionImplRealSpec extends Specification {
def up = Mock(DefaultUpstream) {
_ * getId() >> "test"
}
conn = new EthereumWsConnectionPoolFactory(
conn = new WsConnectionPoolFactory(
"test",
1,
new EthereumWsConnectionFactory(
new WsConnectionFactory(
"test",
Chain.ETHEREUM__MAINNET,
"ws://localhost:${port}".toURI(),

12
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy
View File

@@ -36,10 +36,10 @@ class WsConnectionImplSpec extends Specification {
def "Makes a RPC call"() {
setup:
def wsf = new EthereumWsConnectionPoolFactory(
def wsf = new WsConnectionPoolFactory(
"test",
1,
new EthereumWsConnectionFactory(
new WsConnectionFactory(
"test",
Chain.ETHEREUM__MAINNET,
new URI("http://localhost"),
@@ -71,10 +71,10 @@ class WsConnectionImplSpec extends Specification {
def "Makes a RPC call - return null"() {
setup:
def wsf = new EthereumWsConnectionPoolFactory(
def wsf = new WsConnectionPoolFactory(
"test",
1,
new EthereumWsConnectionFactory(
new WsConnectionFactory(
"test",
Chain.ETHEREUM__MAINNET,
new URI("http://localhost"),
@@ -104,10 +104,10 @@ class WsConnectionImplSpec extends Specification {
def "Makes a RPC call - return error"() {
setup:
def wsf = new EthereumWsConnectionPoolFactory(
def wsf = new WsConnectionPoolFactory(
"test",
1,
new EthereumWsConnectionFactory(
new WsConnectionFactory(
"test",
Chain.ETHEREUM__MAINNET,
new URI("http://localhost"),

6
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPoolSpec.groovy
View File

@@ -29,7 +29,7 @@ class WsConnectionMultiPoolSpec extends Specification {
1 * it.connectionInfoFlux() >> Flux.empty()
}
def up = Mock(DefaultUpstream)
def factory = Mock(EthereumWsConnectionFactory)
def factory = Mock(WsConnectionFactory)
def pool = new WsConnectionMultiPool(factory, up, 3)
pool.scheduler = Stub(ScheduledExecutorService)
@@ -53,7 +53,7 @@ class WsConnectionMultiPoolSpec extends Specification {
1 * it.connectionInfoFlux() >> Flux.empty()
}
def up = Mock(DefaultUpstream)
def factory = Mock(EthereumWsConnectionFactory)
def factory = Mock(WsConnectionFactory)
def pool = new WsConnectionMultiPool(factory, up, 3)
pool.scheduler = Stub(ScheduledExecutorService)
@@ -106,7 +106,7 @@ class WsConnectionMultiPoolSpec extends Specification {
1 * it.connectionInfoFlux() >> Flux.empty()
}
def up = Mock(DefaultUpstream)
def factory = Mock(EthereumWsConnectionFactory)
def factory = Mock(WsConnectionFactory)
def pool = new WsConnectionMultiPool(factory, up, 3)
pool.scheduler = Stub(ScheduledExecutorService)

2
src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt
View File

@@ -100,7 +100,7 @@ class IntegrationTest {
validateChain = false
}
connection = UpstreamsConfig.EthereumPosConnection().apply {
execution = UpstreamsConfig.EthereumConnection().apply {
execution = UpstreamsConfig.RpcConnection().apply {
rpc = UpstreamsConfig.HttpEndpoint(
URI.create(
"http://" + ganacheContainer.getHost() + ":" + ganacheContainer.getMappedPort(8545) + "/",

16
src/test/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigTest.kt
View File

@@ -1,6 +1,6 @@
package io.emeraldpay.dshackle.config
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
@@ -14,25 +14,25 @@ internal class UpstreamsConfigTest {
fun data(): Stream<Arguments> {
return Stream.of(
Arguments.of(
UpstreamsConfig.EthereumConnection(),
UpstreamsConfig.RpcConnection(),
ConnectorMode.RPC_ONLY,
),
Arguments.of(
UpstreamsConfig.EthereumConnection()
UpstreamsConfig.RpcConnection()
.apply {
ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546"))
},
ConnectorMode.WS_ONLY,
),
Arguments.of(
UpstreamsConfig.EthereumConnection()
UpstreamsConfig.RpcConnection()
.apply {
ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546"))
},
ConnectorMode.WS_ONLY,
),
Arguments.of(
UpstreamsConfig.EthereumConnection()
UpstreamsConfig.RpcConnection()
.apply {
connectorMode = "RPC_REQUESTS_WITH_WS_HEAD"
ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546"))
@@ -40,7 +40,7 @@ internal class UpstreamsConfigTest {
ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD,
),
Arguments.of(
UpstreamsConfig.EthereumConnection()
UpstreamsConfig.RpcConnection()
.apply {
connectorMode = "RPC_REQUESTS_WITH_MIXED_HEAD"
ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546"))
@@ -48,7 +48,7 @@ internal class UpstreamsConfigTest {
ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD,
),
Arguments.of(
UpstreamsConfig.EthereumConnection()
UpstreamsConfig.RpcConnection()
.apply {
rpc = UpstreamsConfig.HttpEndpoint(URI("http://localhost:8546"))
ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546"))
@@ -61,7 +61,7 @@ internal class UpstreamsConfigTest {
@ParameterizedTest
@MethodSource("data")
fun testKeepForwarded(input: UpstreamsConfig.EthereumConnection, expected: ConnectorMode) {
fun testKeepForwarded(input: UpstreamsConfig.RpcConnection, expected: ConnectorMode) {
assertEquals(expected, input.resolveMode())
}
}

33
src/test/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecificTest.kt Normal file
View File

@@ -0,0 +1,33 @@
package io.emeraldpay.dshackle.upstream.starknet
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.assertj.core.api.Assertions
import org.junit.jupiter.api.Test
val example = """
{
"status": "ACCEPTED_ON_L2",
"block_hash": "0x46fa6638dc7fae06cece980ce4195436a79ef314ca49d99e0cef552d6f13c4e",
"parent_hash": "0x7cc1e178c848b0bdfa047a414d9a8bee4f6cb76f25f312f2d42e058ea91d78b",
"block_number": 304789,
"new_root": "0x71ba085a6fb172ccb206347eec04c9c171e770c947968fc9df31297e4cee145",
"timestamp": 1696802363,
"sequencer_address": "0x1176a1bd84444c89232ec27754698e5d2e7e1a7f1539f12027f28b23ec9f3d8",
"transactions": [
"0x3303c7acaa3e6efc0cd30f0e4be41a4df1117958f0bdc32cd0759b3664922ed"
]
}
""".trimIndent()
class StarknetChainSpecificTest {
@Test
fun parseResponse() {
val result = StarknetChainSpecific.parseBlock(JsonRpcResponse.ok(example), "1")
Assertions.assertThat(result.height).isEqualTo(304789)
Assertions.assertThat(result.hash).isEqualTo(BlockId.from("046fa6638dc7fae06cece980ce4195436a79ef314ca49d99e0cef552d6f13c4e"))
Assertions.assertThat(result.upstreamId).isEqualTo("1")
Assertions.assertThat(result.parentHash).isEqualTo(BlockId.from("07cc1e178c848b0bdfa047a414d9a8bee4f6cb76f25f312f2d42e058ea91d78b"))
}
}