Update methods based on upstream availability (#216)

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<UpstreamAvailability>()
+        .directBestEffort<UpstreamChangeState>()
 
     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<UpstreamAvailability> {
         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
+    )
 }

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<Upstream>()
+    private val stateStream = Sinks.many()
+        .multicast()
+        .directBestEffort<UpstreamChangeState>()
 
     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<Upstream> =
         removedUpstreams.asFlux()
 
+    fun subscribeStateChanges(): Flux<UpstreamChangeState> =
+        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
+    )
 }

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

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
     }
 

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
     }
 

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<Disposable>()
+        val statusSubscriptions = mutableMapOf<Chain, Disposable>()
 
         return Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
             .flatMap {
@@ -131,19 +130,19 @@ class GrpcUpstreams(
             }.flatMap { value ->
                 processDescription(value)
             }.doOnNext {
-                val subscription = client.subscribeStatus(
+                val sub = statusSubscriptions[it.chain]
-                    StatusRequest.newBuilder()
+                if (sub == null || sub.isDisposed) {
-                        .addChains(Common.ChainRef.forNumber(it.chain.id)).build()
+                    val subscription = client.subscribeStatus(
-                ).subscribeOn(chainStatusScheduler)
+                        StatusRequest.newBuilder()
-                    .subscribe { value ->
+                            .addChains(Common.ChainRef.forNumber(it.chain.id)).build()
-                        val chain = Chain.byId(value.chain.number)
+                    ).subscribeOn(chainStatusScheduler)
-                        if (chain != Chain.UNSPECIFIED) {
+                        .subscribe { value ->
-                            known[chain]?.onStatus(value)
+                            val chain = Chain.byId(value.chain.number)
+                            if (chain != Chain.UNSPECIFIED) {
+                                known[chain]?.onStatus(value)
+                            }
                         }
-                    }
+                    statusSubscriptions[it.chain] = subscription
-                statusSubscription.updateAndGet { prev ->
-                    prev?.dispose()
-                    subscription
                 }
             }.doOnError { t ->
                 log.error("Failed to process update from gRPC upstream $id", t)

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<EthereumPosUpstream>(), 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<EthereumPosUpstream> upstreams, @NotNull Caches caches) {

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<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+            }
+
+            @Override
+            void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> 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<String> methods) {
+        return BlockchainOuterClass.DescribeChain.newBuilder()
+                .setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
+                .addAllSupportedMethods(methods)
+                .build()
+    }
 }