diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 8ae1caf2..5c4a24a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -54,7 +54,7 @@ abstract class DefaultUpstream( private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail))) private val statusStream = Sinks.many() .multicast() - .directBestEffort() + .directBestEffort() init { if (id.length < 3 || !id.matches(Regex("[a-zA-Z][a-zA-Z0-9_-]+[a-zA-Z0-9]"))) { @@ -67,9 +67,14 @@ abstract class DefaultUpstream( } fun onStatus(value: BlockchainOuterClass.ChainStatus) { + this.onStatus(value, false) + } + + fun onStatus(value: BlockchainOuterClass.ChainStatus, stateChanged: Boolean = false) { val available = value.availability setStatus( - if (available != null) UpstreamAvailability.fromGrpc(available.number) else UpstreamAvailability.UNAVAILABLE + if (available != null) UpstreamAvailability.fromGrpc(available.number) else UpstreamAvailability.UNAVAILABLE, + stateChanged ) } @@ -78,10 +83,16 @@ abstract class DefaultUpstream( } open fun setStatus(avail: UpstreamAvailability) { + this.setStatus(avail, false) + } + + open fun setStatus(avail: UpstreamAvailability, stateChanged: Boolean = false) { status.updateAndGet { curr -> Status(curr.lag, avail, statusByLag(curr.lag, avail)) }.also { - statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + statusStream.emitNext( + UpstreamChangeState(it.status, stateChanged) + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } log.trace("Status of upstream [$id] changed to [$it], requested change status to [$avail]") } } @@ -103,7 +114,17 @@ abstract class DefaultUpstream( override fun observeStatus(): Flux { return statusStream.asFlux() - .distinctUntilChanged() + .distinctUntilChanged( + { it }, + { prev, current -> + if (current.stateChanged) { + false + } else { + prev.status == current.status + } + } + ) + .map { it.status } } override fun setLag(lag: Long) { @@ -111,7 +132,9 @@ abstract class DefaultUpstream( status.updateAndGet { curr -> Status(nLag, curr.avail, statusByLag(nLag, curr.avail)) }.also { - statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + statusStream.emitNext( + UpstreamChangeState(it.status, false) + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } log.trace("Status of upstream [$id] changed to [$it], requested change lag to [$lag]") } } @@ -147,4 +170,9 @@ abstract class DefaultUpstream( } data class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability) + + private data class UpstreamChangeState( + val status: UpstreamAvailability, + val stateChanged: Boolean + ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index af8b272d..252ccdd1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -38,6 +38,7 @@ import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks +import reactor.util.function.Tuples import java.time.Duration import java.time.Instant import java.util.concurrent.atomic.AtomicReference @@ -81,6 +82,9 @@ abstract class Multistream( private val removedUpstreams = Sinks.many() .multicast() .directBestEffort() + private val stateStream = Sinks.many() + .multicast() + .directBestEffort() init { UpstreamAvailability.values().forEach { status -> @@ -266,9 +270,41 @@ abstract class Multistream( // print status _change_ every 15 seconds, at most; otherwise prints it on interval of 30 seconds .sample(Duration.ofSeconds(15)) .subscribe { printStatus() } + + observeUpstreamsStatuses() + started = true } + private fun observeUpstreamsStatuses() { + stateStream.asFlux() + .distinctUntilChanged( + { it }, + { prev, current -> + prev.status == current.status || prev.equals(current) + } + ).subscribe { + upstreams.filter { it.isAvailable() }.map { it.getMethods() }.let { + callMethods = AggregatedCallMethods(it) + } + } + + subscribeAddedUpstreams() + .filter { !it.isGrpc() } + .distinctUntilChanged { + it.getId() + }.map { + Tuples.of(it.getId(), it.observeStatus()) + } + .subscribe { pair -> + pair.t2.subscribe { status -> + stateStream.emitNext( + UpstreamChangeState(pair.t1, status) + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + } + } + } + override fun stop() { cacheSubscription?.dispose() cacheSubscription = null @@ -411,6 +447,9 @@ abstract class Multistream( fun subscribeRemovedUpstreams(): Flux = removedUpstreams.asFlux() + fun subscribeStateChanges(): Flux = + stateStream.asFlux() + abstract fun makeLagObserver(): HeadLagObserver // -------------------------------------------------------------------------------------------------------- @@ -435,4 +474,9 @@ abstract class Multistream( return curr == t } } + + data class UpstreamChangeState( + val upId: String, + val status: UpstreamAvailability + ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 54ab83b6..8bbb6424 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -166,7 +166,7 @@ class BitcoinGrpcUpstream( val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { capabilities = newCapabilities } - conf.status?.let { status -> onStatus(status) } + conf.status?.let { status -> onStatus(status, upstreamStatusChanged) } return buildInfoChanged || upstreamStatusChanged } } 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 2dddb790..1ef105b3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -151,7 +151,7 @@ open class EthereumGrpcUpstream( val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { capabilities = newCapabilities } - conf.status?.let { status -> onStatus(status) } + conf.status?.let { status -> onStatus(status, upstreamStatusChanged) } return buildInfoChanged || upstreamStatusChanged } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 80847e39..ef1aa8ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -118,7 +118,7 @@ open class EthereumPosGrpcUpstream( val upstreamStatusChanged = (upstreamStatus.update(conf) || (newCapabilities != capabilities)).also { capabilities = newCapabilities } - conf.status?.let { status -> onStatus(status) } + conf.status?.let { status -> onStatus(status, upstreamStatusChanged) } return buildInfoChanged || upstreamStatusChanged } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index ecdc9a30..bb01da3e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -55,7 +55,6 @@ import reactor.core.scheduler.Scheduler import java.io.IOException import java.time.Duration import java.util.concurrent.Executor -import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -114,7 +113,7 @@ class GrpcUpstreams( } this.client = client - val statusSubscription = AtomicReference() + val statusSubscriptions = mutableMapOf() return Flux.interval(Duration.ZERO, Duration.ofSeconds(20)) .flatMap { @@ -131,19 +130,19 @@ class GrpcUpstreams( }.flatMap { value -> processDescription(value) }.doOnNext { - val subscription = client.subscribeStatus( - StatusRequest.newBuilder() - .addChains(Common.ChainRef.forNumber(it.chain.id)).build() - ).subscribeOn(chainStatusScheduler) - .subscribe { value -> - val chain = Chain.byId(value.chain.number) - if (chain != Chain.UNSPECIFIED) { - known[chain]?.onStatus(value) + val sub = statusSubscriptions[it.chain] + if (sub == null || sub.isDisposed) { + val subscription = client.subscribeStatus( + StatusRequest.newBuilder() + .addChains(Common.ChainRef.forNumber(it.chain.id)).build() + ).subscribeOn(chainStatusScheduler) + .subscribe { value -> + val chain = Chain.byId(value.chain.number) + if (chain != Chain.UNSPECIFIED) { + known[chain]?.onStatus(value) + } } - } - statusSubscription.updateAndGet { prev -> - prev?.dispose() - subscription + statusSubscriptions[it.chain] = subscription } }.doOnError { t -> log.error("Failed to process update from gRPC upstream $id", t) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index eb1651e7..4d5c8353 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -23,16 +23,17 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.grpc.EthereumPosGrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash -import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux @@ -252,6 +253,63 @@ class MultistreamSpec extends Specification { !act } + def "Change ms methods based on upstream availability"() { + setup: + def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"])) + def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) + def ms = new EthereumPosMultiStream(Chain.ETHEREUM, new ArrayList(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) + when: + ms.onUpstreamChange( + new UpstreamChangeEvent(Chain.ETHEREUM, up1, UpstreamChangeEvent.ChangeType.ADDED) + ) + ms.onUpstreamChange( + new UpstreamChangeEvent(Chain.ETHEREUM, up2, UpstreamChangeEvent.ChangeType.ADDED) + ) + def states = ms.subscribeStateChanges() + then: + StepVerifier.create(states) + .then { + up1.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_OK)) + up2.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_OK)) + } + .expectNext(new Multistream.UpstreamChangeState(up1.getId(), UpstreamAvailability.OK)) + .expectNext(new Multistream.UpstreamChangeState(up2.getId(), UpstreamAvailability.OK)) + .then { + assert ms.getMethods().supportedMethods == Set.of("eth_test1", "eth_test2", "eth_test3") + } + .then { + up1.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_SYNCING)) + } + .expectNext(new Multistream.UpstreamChangeState(up1.getId(), UpstreamAvailability.SYNCING)) + .then { + assert ms.getMethods().supportedMethods == Set.of("eth_test1", "eth_test2") + } + .then { + up1.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_OK)) + } + .expectNext(new Multistream.UpstreamChangeState(up1.getId(), UpstreamAvailability.OK)) + .then { + assert ms.getMethods().supportedMethods == Set.of("eth_test1", "eth_test2", "eth_test3") + } + .then { + up1.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_OK)) + } + .expectNextCount(0) + .then { + up2.onStatus(status(BlockchainOuterClass.AvailabilityEnum.AVAIL_OK)) + } + .expectNextCount(0) + .thenCancel() + .verify(Duration.ofSeconds(3)) + + } + + private BlockchainOuterClass.ChainStatus status(BlockchainOuterClass.AvailabilityEnum status) { + return BlockchainOuterClass.ChainStatus.newBuilder() + .setAvailability(status) + .build() + } + class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index 2eae2dd0..f943bf6c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -30,14 +30,15 @@ import io.emeraldpay.dshackle.test.MockGrpcServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.BuildInfo import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.etherjar.domain.BlockHash -import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.grpc.stub.StreamObserver import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Timer import reactor.core.scheduler.Schedulers +import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration @@ -254,4 +255,61 @@ class EthereumGrpcUpstreamSpec extends Specification { h.hash == BlockId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") h.height == 650247 } + + def "Send update status if methods were changed"() { + setup: + def chain = Chain.ETHEREUM + def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + } + + @Override + void subscribeHead(Common.Chain request, StreamObserver responseObserver) { + } + }) + def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null, ChainsConfig.ChainConfig.default(), Schedulers.parallel()) + upstream.setLag(0) + upstream.setStatus(UpstreamAvailability.OK) + when: + def statuses = upstream.observeStatus() + then: + StepVerifier.create(statuses) + .then { + upstream.update( + describe(["eth_getBlockByHash"]), + BlockchainOuterClass.BuildInfo.newBuilder() + .setVersion(buildInfo.version) + .build(), + ) + } + .expectNext(UpstreamAvailability.OK) + .then { + upstream.update( + describe(["eth_getBlockByHash"]), + BlockchainOuterClass.BuildInfo.newBuilder() + .setVersion(buildInfo.version) + .build(), + ) + } + .expectNextCount(0) + .then { + upstream.update( + describe(["eth_getBlockByHash", "eth_getBlockByHash1"]), + BlockchainOuterClass.BuildInfo.newBuilder() + .setVersion(buildInfo.version) + .build(), + ) + } + .expectNext(UpstreamAvailability.OK) + .thenCancel() + .verify(Duration.ofSeconds(3)) + } + + private BlockchainOuterClass.DescribeChain describe(List methods) { + return BlockchainOuterClass.DescribeChain.newBuilder() + .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId)) + .addAllSupportedMethods(methods) + .build() + } }