From 85caff4537540e7fac7096ea45ebda9b7a57a4dc Mon Sep 17 00:00:00 2001 From: Termina1 Date: Thu, 5 Jan 2023 20:34:56 +0200 Subject: [PATCH 1/4] fix node status subsciption for gRPC upstreams --- .../io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index 93013e66..c746fe2e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -119,9 +119,11 @@ class SubscribeNodeStatus( val statuses = upstream.observeStatus() .distinctUntilChanged() - .takeUntil { it == UpstreamAvailability.UNAVAILABLE } + .takeUntil { + it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc() + } .map { - if (it == UpstreamAvailability.UNAVAILABLE) { + if (it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc()) { onUnavailable.accept(upstream.getId()) // cancel head subscription & reconnections when upstream becomes unavailable cancel.tryEmitNext(true) From 38f8a6817ae90a3a7b1c86b641dea64a492f0229 Mon Sep 17 00:00:00 2001 From: Termina1 Date: Fri, 6 Jan 2023 13:58:01 +0200 Subject: [PATCH 2/4] handle subscription updates for upstream removals separately --- .../dshackle/rpc/SubscribeNodeStatus.kt | 42 +++++++++++-------- .../dshackle/upstream/Multistream.kt | 7 ++++ 2 files changed, 31 insertions(+), 18 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index c746fe2e..a526f0c1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -30,7 +30,7 @@ class SubscribeNodeStatus( fun subscribe(req: Mono): Flux = req.flatMapMany { request -> - val knownUpstreams = ConcurrentHashMap() + val knownUpstreams = ConcurrentHashMap>() val duration = Duration.ofMillis(request.timespan) // send known upstreams details immediately val descriptions = Flux.fromIterable( @@ -51,12 +51,27 @@ class SubscribeNodeStatus( multistreams.all() .flatMap { ms -> ms.getAll().map { up -> - knownUpstreams[up.getId()] = true - subscribeUpstreamUpdates(ms.chain, up, duration) { r -> knownUpstreams.remove(r) } + knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort() + subscribeUpstreamUpdates(ms.chain, up, duration, knownUpstreams[up.getId()]!!) } } ) + // stop removed upstreams update fluxes + val removals = Flux.merge(multistreams.all().map { ms -> + ms.subscribeRemovedUpstreams().mapNotNull { up -> + knownUpstreams[up.getId()]?.let { + it.tryEmitNext(true) + knownUpstreams.remove(up.getId()) + NodeStatusResponse.newBuilder() + .setNodeId(up.getId()) + .setDescription(buildDescription(ms.chain, up)) + .setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight())) + .build() + } + } + }) + // subscribe on head/status updates for just added upstreams val multiStreamUpdates = Flux.merge( multistreams.all() @@ -66,10 +81,10 @@ class SubscribeNodeStatus( it.getId() } .filter { - !knownUpstreams.getOrDefault(it.getId(), false) + !knownUpstreams.contains(it.getId()) } .flatMap { - knownUpstreams[it.getId()] = true + knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort() Flux.concat( Mono.just( NodeStatusResponse.newBuilder() @@ -78,24 +93,22 @@ class SubscribeNodeStatus( .setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight())) .build() ), - subscribeUpstreamUpdates(ms.chain, it, duration) { r -> knownUpstreams.remove(r) } + subscribeUpstreamUpdates(ms.chain, it, duration, knownUpstreams[it.getId()]!!) ) } } ) - Flux.concat(descriptions, Flux.merge(upstreamUpdates, multiStreamUpdates)) + Flux.concat(descriptions, Flux.merge(upstreamUpdates, multiStreamUpdates, removals)) } private fun subscribeUpstreamUpdates( chain: Chain, upstream: Upstream, timespan: Duration, - onUnavailable: Consumer + cancel: Sinks.Many ): Flux { val retry = Sinks.many().multicast().directBestEffort() - val cancel = Sinks.many().multicast().directBestEffort() - val heads = Mono.just(upstream) .repeatWhen { retry.asFlux() @@ -119,15 +132,8 @@ class SubscribeNodeStatus( val statuses = upstream.observeStatus() .distinctUntilChanged() - .takeUntil { - it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc() - } + .takeUntilOther(cancel.asFlux()) .map { - if (it == UpstreamAvailability.UNAVAILABLE && !upstream.isGrpc()) { - onUnavailable.accept(upstream.getId()) - // cancel head subscription & reconnections when upstream becomes unavailable - cancel.tryEmitNext(true) - } NodeStatusResponse.newBuilder() .setNodeId(upstream.getId()) .setStatus(buildStatus(it, upstream.getHead().getCurrentHeight())) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 6305c7c9..c8e460af 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -78,6 +78,9 @@ abstract class Multistream( private val addedUpstreams = Sinks.many() .multicast() .directBestEffort() + private val removedUpstreams = Sinks.many() + .multicast() + .directBestEffort() init { UpstreamAvailability.values().forEach { status -> @@ -342,6 +345,7 @@ abstract class Multistream( eventLock.withLock { if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { removeUpstream(event.upstream.getId()).takeIf { it }?.let { + removedUpstreams.tryEmitNext(event.upstream) log.warn("Upstream ${event.upstream.getId()} with chain $chain has been removed") } } else { @@ -370,6 +374,9 @@ abstract class Multistream( fun subscribeAddedUpstreams(): Flux = addedUpstreams.asFlux() + fun subscribeRemovedUpstreams(): Flux = + removedUpstreams.asFlux() + // -------------------------------------------------------------------------------------------------------- class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) From b769446b0ebd02bd6bc9fd9da3007ef1b7352273 Mon Sep 17 00:00:00 2001 From: Termina1 Date: Mon, 9 Jan 2023 20:19:37 +0200 Subject: [PATCH 3/4] add handler for not able to emit removal event --- .../io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index a526f0c1..f3652c0d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -26,6 +27,7 @@ class SubscribeNodeStatus( companion object { private val RETRY_TIMEOUT = Duration.ofSeconds(10) + private val log = LoggerFactory.getLogger(SubscribeNodeStatus::class.java) } fun subscribe(req: Mono): Flux = @@ -61,7 +63,10 @@ class SubscribeNodeStatus( val removals = Flux.merge(multistreams.all().map { ms -> ms.subscribeRemovedUpstreams().mapNotNull { up -> knownUpstreams[up.getId()]?.let { - it.tryEmitNext(true) + val result = it.tryEmitNext(true) + if (result.isFailure) { + log.warn("Unable to emit event about removal of an upstream - ${result.toString()}") + } knownUpstreams.remove(up.getId()) NodeStatusResponse.newBuilder() .setNodeId(up.getId()) From ed9d39b7332c2b313e40c011fae5db516d121ec9 Mon Sep 17 00:00:00 2001 From: Termina1 Date: Mon, 9 Jan 2023 20:38:15 +0200 Subject: [PATCH 4/4] fix stylw --- .../dshackle/rpc/SubscribeNodeStatus.kt | 32 ++++++++++--------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt index f3652c0d..2ad44345 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt @@ -18,7 +18,6 @@ import reactor.core.publisher.SignalType import reactor.core.publisher.Sinks import java.time.Duration import java.util.concurrent.ConcurrentHashMap -import java.util.function.Consumer @Service class SubscribeNodeStatus( @@ -60,22 +59,25 @@ class SubscribeNodeStatus( ) // stop removed upstreams update fluxes - val removals = Flux.merge(multistreams.all().map { ms -> - ms.subscribeRemovedUpstreams().mapNotNull { up -> - knownUpstreams[up.getId()]?.let { - val result = it.tryEmitNext(true) - if (result.isFailure) { - log.warn("Unable to emit event about removal of an upstream - ${result.toString()}") + val removals = Flux.merge( + multistreams.all() + .map { ms -> + ms.subscribeRemovedUpstreams().mapNotNull { up -> + knownUpstreams[up.getId()]?.let { + val result = it.tryEmitNext(true) + if (result.isFailure) { + log.warn("Unable to emit event about removal of an upstream - $result") + } + knownUpstreams.remove(up.getId()) + NodeStatusResponse.newBuilder() + .setNodeId(up.getId()) + .setDescription(buildDescription(ms.chain, up)) + .setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight())) + .build() + } } - knownUpstreams.remove(up.getId()) - NodeStatusResponse.newBuilder() - .setNodeId(up.getId()) - .setDescription(buildDescription(ms.chain, up)) - .setStatus(buildStatus(UpstreamAvailability.UNAVAILABLE, up.getHead().getCurrentHeight())) - .build() } - } - }) + ) // subscribe on head/status updates for just added upstreams val multiStreamUpdates = Flux.merge(