diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt
index 7317b01a..f267688a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt
@@ -33,7 +33,7 @@ class SubscribeNodeStatus(
             multistreams.all()
                 .flatMap { ms ->
                     ms.getAll().map { up ->
-                        knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
+                        knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort()
                         subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
                     }
                 }
@@ -76,7 +76,7 @@ class SubscribeNodeStatus(
                             !knownUpstreams.contains(it.getId())
                         }
                         .flatMap {
-                            knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
+                            knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort()
                             Flux.concat(
                                 Mono.just(
                                     NodeStatusResponse.newBuilder()
@@ -100,7 +100,6 @@ class SubscribeNodeStatus(
         cancel: Sinks.Many<Boolean>
     ): Flux<NodeStatusResponse> {
         val statuses = upstream.observeStatus()
-            .distinctUntilChanged()
             .takeUntilOther(cancel.asFlux())
             .map {
                 NodeStatusResponse.newBuilder()
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
index 5c674544..3b52ecb7 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
@@ -129,7 +129,7 @@ abstract class DefaultUpstream(
                 Status(nLag, curr.avail, statusByLag(nLag, curr.avail))
             }.also {
                 statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
-                log.trace("Status of upstream [$id] changed to [$it], requested change lag to [$lag]")
+                log.debug("Status of upstream [$id] changed to [$it], requested change lag to [$lag]")
             }
         }
     }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
index 0233583b..4184bf0a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
@@ -351,8 +351,12 @@ 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")
+                        try {
+                            removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
+                        } catch (e: Sinks.EmissionException) {
+                            log.error("error during event processing $event", e)
+                        }
                     }
                 } else {
                     if (event.upstream is CachesEnabled) {
@@ -362,8 +366,12 @@ abstract class Multistream(
                         if (!started) {
                             start()
                         }
-                        addedUpstreams.tryEmitNext(event.upstream)
-                        log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
+                        try {
+                            addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                            log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
+                        } catch (e: Sinks.EmissionException) {
+                            log.error("error during event processing $event", e)
+                        }
                     }
                 }
             }