diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 45291ccf..afe64fc6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -21,8 +21,6 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream -import io.emeraldpay.dshackle.upstream.grpc.EthereumGrpcUpstream import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.Mono @@ -47,12 +45,8 @@ class Describe( chainUpstreams.getAll().let { ups -> ups.forEach { up -> val nodes = QuorumForLabels() - if (up is EthereumUpstream) { - nodes.add(up.node) - } else if (up is BitcoinUpstream) { - nodes.add(up.node) - } else if (up is EthereumGrpcUpstream) { - nodes.add(up.getNodes()) + if (up is DefaultUpstream) { + nodes.add(up.getQuorumByLabel()) } nodes.getAll().forEach { node -> val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 726eb779..0ded56fa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams @@ -182,7 +183,7 @@ open class ConfiguredUpstreams( } log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val ethereumUpstream = EthereumUpstream( + val ethereumUpstream = EthereumRpcUpstream( config.id!!, chain, directApi, wsFactoryApi, options, config.role, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt index 77f670fc..66519ba7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt @@ -26,11 +26,15 @@ import kotlin.concurrent.write /** * Summary details over few upstream nodes. Provides aggregate quorum for nodes with particular label */ -class QuorumForLabels { +class QuorumForLabels() { private val lock = ReentrantReadWriteLock() private val nodes = ArrayList() + constructor(node: QuorumItem) : this() { + add(node) + } + fun add(node: QuorumItem) { lock.read { val existing = nodes.find { it.labels == node.labels } @@ -59,6 +63,9 @@ class QuorumForLabels { */ class QuorumItem(val quorum: Int, val labels: UpstreamsConfig.Labels) { companion object { + fun empty(): QuorumItem { + return QuorumItem(0, UpstreamsConfig.Labels()) + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index e47a7e5f..2a53ea29 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -50,8 +50,6 @@ class CurrentMultistreamHolder( private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java) - private val objectMapper: ObjectMapper = Global.objectMapper - private val chainMapping = ConcurrentHashMap() private val chainsBus = TopicProcessor.create() private val callTargets = HashMap() @@ -59,27 +57,32 @@ class CurrentMultistreamHolder( fun update(change: UpstreamChange) { updateLock.withLock { + log.debug("Upstream update: ${change.type} ${change.chain} via ${change.upstream.getId()}") val chain = change.chain - when (BlockchainType.fromBlockchain(chain)) { - BlockchainType.ETHEREUM -> { - val up = change.upstream.cast(EthereumUpstream::class.java) - val current = chainMapping[chain] as Multistream? - val factory = Callable { - EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream + try { + when (BlockchainType.fromBlockchain(chain)) { + BlockchainType.ETHEREUM -> { + val up = change.upstream.cast(EthereumUpstream::class.java) + val current = chainMapping[chain] as Multistream? + val factory = Callable { + EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream + } + processUpdate(change, up, current, factory) } - processUpdate(change, up, current, factory) - } - BlockchainType.BITCOIN -> { - val up = change.upstream.cast(BitcoinUpstream::class.java) - val current = chainMapping[chain] as Multistream? - val factory = Callable { - BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream + BlockchainType.BITCOIN -> { + val up = change.upstream.cast(BitcoinUpstream::class.java) + val current = chainMapping[chain] as Multistream? + val factory = Callable { + BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream + } + processUpdate(change, up, current, factory) + } + else -> { + log.error("Update for unsupported chain: $chain") } - processUpdate(change, up, current, factory) - } - else -> { - log.error("Update for unsupported chain: $chain") } + } catch (e: Throwable) { + log.error("Failed to update upstream", e) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 75a0b6ce..7947740e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.calls.CallMethods import reactor.core.publisher.Flux import reactor.extra.processor.TopicProcessor @@ -28,11 +29,15 @@ abstract class DefaultUpstream( defaultAvail: UpstreamAvailability, private val options: UpstreamsConfig.Options, private val role: UpstreamsConfig.UpstreamRole, - private val targets: CallMethods? + private val targets: CallMethods?, + private val node: QuorumForLabels.QuorumItem? ) : Upstream { constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?) : - this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets) + this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, QuorumForLabels.QuorumItem.empty()) + + constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, node: QuorumForLabels.QuorumItem?) : + this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node) private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail))) private val statusStream: TopicProcessor = TopicProcessor.create() @@ -95,5 +100,12 @@ abstract class DefaultUpstream( return targets ?: throw IllegalStateException("Methods are not set") } + private val quorumByLabel = node?.let { QuorumForLabels(it) } + ?: QuorumForLabels(QuorumForLabels.QuorumItem.empty()) + + open fun getQuorumByLabel(): QuorumForLabels { + return quorumByLabel + } + class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability) } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt index fce79cf2..c3cf873e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt @@ -35,9 +35,9 @@ open class BitcoinUpstream( private val directApi: Reader, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, - val node: QuorumForLabels.QuorumItem, + node: QuorumForLabels.QuorumItem, callMethods: CallMethods -) : DefaultUpstream(id, options, role, callMethods), Lifecycle { +) : DefaultUpstream(id, options, role, callMethods, node), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt new file mode 100644 index 00000000..39e72c6e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -0,0 +1,117 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable +import java.time.Duration + +open class EthereumRpcUpstream( + id: String, + val chain: Chain, + private val directReader: Reader, + private val ethereumWsFactory: EthereumWsFactory? = null, + options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, + private val node: QuorumForLabels.QuorumItem, + targets: CallMethods +) : EthereumUpstream(id, options, role, targets, node), Upstream, CachesEnabled, Lifecycle { + + constructor(id: String, chain: Chain, api: Reader) : + this(id, chain, api, null, + UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD, + QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), + DirectCallMethods()) + + + private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java) + + private val head: Head = this.createHead() + private var validatorSubscription: Disposable? = null + + override fun setCaches(caches: Caches) { + if (head is CachesEnabled) { + head.setCaches(caches) + } + } + + override fun start() { + log.info("Configured for ${chain.chainName}") + + if (getOptions().disableValidation != null && getOptions().disableValidation!!) { + log.warn("Disable validation for upstream ${this.getId()}") + this.setLag(0) + this.setStatus(UpstreamAvailability.OK) + } else { + log.debug("Start validation for upstream ${this.getId()}") + val validator = EthereumUpstreamValidator(this, getOptions()) + validatorSubscription = validator.start() + .subscribe(this::setStatus) + } + } + + override fun isRunning(): Boolean { + return true + } + + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + if (head is Lifecycle) { + head.stop() + } + } + + open fun createHead(): Head { + return if (ethereumWsFactory != null) { + val ws = ethereumWsFactory.create(this).apply { + connect() + } + val wsHead = EthereumWsHead(ws).apply { + start() + } + // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed + val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply { + start() + } + MergedHead(listOf(rpcHead, wsHead)).apply { + start() + } + } else { + log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC") + EthereumRpcHead(getApi()).apply { + start() + } + } + } + + override fun getHead(): Head { + return head + } + + override fun getApi(): Reader { + return directReader + } + + override fun getLabels(): Collection { + return listOf(node.labels) + } + + @Suppress("unchecked") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 813823a8..dbaaaf18 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -16,120 +16,15 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.grpc.Chain -import org.slf4j.LoggerFactory -import org.springframework.context.Lifecycle -import reactor.core.Disposable -import reactor.core.publisher.Mono -import java.time.Duration -open class EthereumUpstream( +abstract class EthereumUpstream( id: String, - val chain: Chain, - private val directReader: Reader, - private val ethereumWsFactory: EthereumWsFactory? = null, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, - val node: QuorumForLabels.QuorumItem, - targets: CallMethods -) : DefaultUpstream(id, options, role, targets), Upstream, CachesEnabled, Lifecycle { - - constructor(id: String, chain: Chain, api: Reader) : - this(id, chain, api, null, - UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD, - QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), - DirectCallMethods()) - - - private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) - - private val head: Head = this.createHead() - private var validatorSubscription: Disposable? = null - - override fun setCaches(caches: Caches) { - if (head is CachesEnabled) { - head.setCaches(caches) - } - } - - override fun start() { - log.info("Configured for ${chain.chainName}") - - if (getOptions().disableValidation != null && getOptions().disableValidation!!) { - log.warn("Disable validation for upstream ${this.getId()}") - this.setLag(0) - this.setStatus(UpstreamAvailability.OK) - } else { - log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) - validatorSubscription = validator.start() - .subscribe(this::setStatus) - } - } - - override fun isRunning(): Boolean { - return true - } - - override fun stop() { - validatorSubscription?.dispose() - validatorSubscription = null - if (head is Lifecycle) { - head.stop() - } - } - - open fun createHead(): Head { - return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create(this).apply { - connect() - } - val wsHead = EthereumWsHead(ws).apply { - start() - } - // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed - val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply { - start() - } - MergedHead(listOf(rpcHead, wsHead)).apply { - start() - } - } else { - log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC") - EthereumRpcHead(getApi()).apply { - start() - } - } - } - - override fun getHead(): Head { - return head - } - - override fun getApi(): Reader { - return directReader - } - - override fun getLabels(): Collection { - return listOf(node.labels) - } - - @Suppress("unchecked") - override fun cast(selfType: Class): T { - if (!selfType.isAssignableFrom(this.javaClass)) { - throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") - } - return this as T - } - -} \ No newline at end of file + targets: CallMethods?, + node: QuorumForLabels.QuorumItem? +) : DefaultUpstream(id, options, role, targets, node) \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 7288f1bb..568e5218 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.upstream.grpc -import com.fasterxml.jackson.databind.ObjectMapper import com.salesforce.reactorgrpc.GrpcRetry import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common @@ -57,11 +56,11 @@ open class EthereumGrpcUpstream( private val chain: Chain, private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: JsonRpcGrpcClient -) : DefaultUpstream( +) : EthereumUpstream( "$parentId/${chain.chainCode}", UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD, - null + null, null ), Lifecycle { private var allLabels: Collection = ArrayList() @@ -176,7 +175,7 @@ open class EthereumGrpcUpstream( ) } - fun getNodes(): QuorumForLabels { + override fun getQuorumByLabel(): QuorumForLabels { return nodes.get() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 5b10b7c6..4e136ec4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -35,7 +36,7 @@ import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher -class EthereumUpstreamMock extends EthereumUpstream { +class EthereumUpstreamMock extends EthereumRpcUpstream { EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 88447fb0..3d809631 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels 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.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.grpc.Chain @@ -45,7 +46,7 @@ class FilteredApisSpec extends Specification { [test: "foo"], [test: "baz"] ].collect { - new EthereumUpstream( + new EthereumRpcUpstream( "test", Chain.ETHEREUM, TestingCommons.api().tap { it.id = "${i++}" },