solution: upgrade Reactor libs

build.gradle

@@ -90,9 +90,9 @@ dependencies {
     implementation "org.springframework.security:spring-security-web:$springSecurtyVersion"
     implementation "org.springframework.security:spring-security-config:$springSecurtyVersion"
     implementation "io.projectreactor:reactor-core:$reactorVersion"
-    implementation "io.projectreactor.netty:reactor-netty:1.0.7"
+    implementation "io.projectreactor.netty:reactor-netty:1.0.11"
-    implementation 'io.projectreactor.addons:reactor-extra:3.4.3'
+    implementation 'io.projectreactor.addons:reactor-extra:3.4.5'
-    implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.1.3'
+    implementation 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.1.4'
     implementation "com.salesforce.servicelibs:reactor-grpc-stub:$reactiveGrpcVersion"
     implementation 'io.micrometer:micrometer-registry-prometheus:1.5.6'
     implementation 'io.lettuce:lettuce-core:5.2.2.RELEASE'

gradle.properties

@@ -6,11 +6,11 @@ protocVersion=3.9.0
 slf4jVersion=1.7.25
 jacksonVersion=2.11.0
 grpcVersion=1.38.0
-reactiveGrpcVersion=1.0.1
+reactiveGrpcVersion=1.2.0
 springBootVersion=2.4.5
 springVersion=5.3.6
 springSecurtyVersion=5.4.6
-reactorVersion=3.4.5
+reactorVersion=3.4.10
 nettyVersion=4.1.65.Final
 # Our Libs
 etherjarVersion=0.11.1

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -41,6 +41,7 @@ import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import reactor.core.publisher.*
+import reactor.kotlin.core.publisher.toMono
 import java.lang.Exception
 import java.util.*
 

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt

@@ -58,7 +58,7 @@ class TrackEthereumTx(
         private val NOT_MINED_TRACK_TTL = NOT_FOUND_TRACK_TTL.multipliedBy(2)
     }
 
-    var scheduler: Scheduler = Schedulers.elastic()
+    var scheduler: Scheduler = Schedulers.boundedElastic()
 
     private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt

@@ -20,7 +20,8 @@ import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.extra.processor.TopicProcessor
+import reactor.core.publisher.Sinks
+import reactor.core.scheduler.Schedulers
 import java.util.concurrent.atomic.AtomicReference
 
 abstract class AbstractHead : Head {
@@ -30,10 +31,16 @@ abstract class AbstractHead : Head {
     }
 
     private val head = AtomicReference<BlockContainer>(null)
-    private val stream: TopicProcessor<BlockContainer> = TopicProcessor.create()
+    private var stream = Sinks.many().multicast().directBestEffort<BlockContainer>()
+    private var completed = false
     private val beforeBlockHandlers = ArrayList<Runnable>()
 
     fun follow(source: Flux<BlockContainer>): Disposable {
+        if (completed) {
+            // if stream was already completed it cannot accept messages (with FAIL_TERMINATED), so needs to be recreated
+            stream = Sinks.many().multicast().directBestEffort<BlockContainer>()
+            completed = false
+        }
         return source
                 .distinctUntilChanged {
                     it.hash
@@ -42,11 +49,13 @@ abstract class AbstractHead : Head {
                     curr == null || curr.difficulty < block.difficulty
                 }
                 .doFinally {
-                    // close internal stream if upstream is finished, otherwise it gets stuck
+                    // close internal stream if upstream is finished, otherwise it gets stuck,
-                    // but technically is should never happen during normal work, only when the Head
+                    // but technically it should never happen during normal work, only when the Head
                     // is stopping
-                    stream.onComplete()
+                    completed = true
+                    stream.tryEmitComplete()
                 }
+                .subscribeOn(Schedulers.boundedElastic())
                 .subscribe { block ->
                     notifyBeforeBlock()
                     val prev = head.getAndUpdate { curr ->
@@ -58,7 +67,10 @@ abstract class AbstractHead : Head {
                     }
                     if (prev == null || prev.hash != block.hash) {
                         log.debug("New block ${block.height} ${block.hash}")
-                        stream.onNext(block)
+                        val result = stream.tryEmitNext(block)
+                        if (result.isFailure && result != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) {
+                            log.warn("Failed to dispatch block: $result as ${this.javaClass}")
+                        }
                     }
                 }
     }
@@ -80,7 +92,7 @@ abstract class AbstractHead : Head {
     override fun getFlux(): Flux<BlockContainer> {
         return Flux.concat(
                 Mono.justOrEmpty(head.get()),
-                Flux.from(stream)
+                stream.asFlux()
         ).onBackpressureLatest()
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt

@@ -34,7 +34,7 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Repository
 import reactor.core.publisher.Flux
-import reactor.extra.processor.TopicProcessor
+import reactor.core.publisher.Sinks
 import java.util.*
 import java.util.concurrent.Callable
 import java.util.concurrent.ConcurrentHashMap
@@ -49,7 +49,9 @@ open class CurrentMultistreamHolder(
     private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java)
 
     private val chainMapping = ConcurrentHashMap<Chain, Multistream>()
-    private val chainsBus = TopicProcessor.create<Chain>()
+    private val chainsBus = Sinks.many()
+            .multicast()
+            .directBestEffort<Chain>()
     private val callTargets = HashMap<Chain, CallMethods>()
     private val updateLock = ReentrantLock()
 
@@ -99,7 +101,7 @@ open class CurrentMultistreamHolder(
                 created.addUpstream(up)
                 created.start()
                 chainMapping[chain] = created
-                chainsBus.onNext(chain)
+                chainsBus.tryEmitNext(chain)
             } else {
                 if (up is CachesEnabled) {
                     up.setCaches(current.caches)
@@ -124,7 +126,7 @@ open class CurrentMultistreamHolder(
     override fun observeChains(): Flux<Chain> {
         return Flux.concat(
                 Flux.fromIterable(getAvailable()),
-                Flux.from(chainsBus)
+                chainsBus.asFlux()
         )
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -22,7 +22,6 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Sinks
-import reactor.extra.processor.TopicProcessor
 import java.util.concurrent.atomic.AtomicReference
 
 abstract class DefaultUpstream(

src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt

@@ -23,8 +23,8 @@ import io.micrometer.core.instrument.DistributionSummary
 import io.micrometer.core.instrument.Metrics
 import io.micrometer.core.instrument.Tag
 import org.reactivestreams.Subscriber
-import reactor.core.publisher.EmitterProcessor
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
 import java.time.Duration
 import java.util.*
 import java.util.concurrent.locks.Lock
@@ -81,7 +81,7 @@ class FilteredApis(
     private val standardUpstreams: List<Upstream>
     private val standardWithFallback: List<Upstream>
 
-    private val control = EmitterProcessor.create<Boolean>(32, false)
+    private val control = Sinks.many().unicast().onBackpressureBuffer<Boolean>()
 
     init {
         delay = if (jitter > 0) {
@@ -161,19 +161,19 @@ class FilteredApis(
         }
 
         result.filter { up -> up.isAvailable() && matcher.matches(up) }
-                .zipWith(control)
+                .zipWith(control.asFlux())
                 .map { it.t1 }
                 .subscribe(subscriber)
     }
 
     override fun resolve() {
-        control.onComplete()
+        control.tryEmitComplete()
     }
 
     override fun request(tries: Int) {
         //TODO check the buffer size before submitting
         repeat(tries) {
-            control.onNext(true)
+            control.tryEmitNext(true)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt

@@ -64,7 +64,7 @@ abstract class HeadLagObserver(
     fun probeFollowers(top: BlockContainer): Flux<Tuple2<Long, Upstream>> {
         return Flux.fromIterable(followers)
                 .parallel(followers.size)
-                .flatMap { up -> mapLagging(top, up, getCurrentBlocks(up)).subscribeOn(Schedulers.elastic()) }
+                .flatMap { up -> mapLagging(top, up, getCurrentBlocks(up)).subscribeOn(Schedulers.boundedElastic()) }
                 .sequential()
                 .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) }
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt

@@ -43,6 +43,11 @@ class MergedHead(
     }
 
     override fun stop() {
+        sources.forEach { head ->
+            if (head is Lifecycle && head.isRunning) {
+                head.stop()
+            }
+        }
         subscription?.dispose()
         subscription = null
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt

@@ -69,8 +69,8 @@ class EthereumWsFactory(
 
         private val topic = Sinks
                 .many()
-                .unicast()
+                .multicast()
-                .onBackpressureBuffer<BlockContainer>()
+                .directBestEffort<BlockContainer>()
         private var keepConnection = true
         private var connection: Disposable? = null
 
@@ -126,7 +126,6 @@ class EthereumWsFactory(
                                     .compress(false)
                                     .build()
                     )
-
                     .uri(uri)
                     .handle { inbound, outbound ->
                         val consumer = inbound.aggregateFrames()
@@ -163,11 +162,14 @@ class EthereumWsFactory(
                                 }
 
 
-                        outbound.sendString(Mono.just(START_REQUEST).doOnError {
+                        outbound.sendString(Mono.just(START_REQUEST)
-                            println("!!!!!!!")
+                                .doOnError { log.warn("Failed to start WS subscription. ${it.javaClass}: ${it.message}") })
-                        })
                                 .then(consumer.then())
-                    }.subscribe()
+                    }
+                    .doOnError {
+                        println(it)
+                    }
+                    .subscribe()
         }
 
         fun onNewBlock(block: BlockJson<TransactionRefJson>) {

src/main/resources/log4j2.xml

@@ -24,6 +24,13 @@
             <AppenderRef ref="STDERR" level="warn"/>
         </Logger>
 
+        <!-- Reactor Netty produces warnings that are ok,
+             ex. when Dshackle closes a connection too fast and Reactor Netty HTTPClient doesn't like that -->
+        <Logger name="reactor.netty.http.client" level="error" additivity="false">
+            <AppenderRef ref="STDOUT"/>
+            <AppenderRef ref="STDERR" level="warn"/>
+        </Logger>
+
         <Root level="warn" additivity="false">
             <AppenderRef ref="STDOUT"/>
             <AppenderRef ref="STDERR" level="warn"/>

src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy

@@ -24,6 +24,7 @@ import spock.lang.Specification
 
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.Executors
 
 class AbstractHeadSpec extends Specification {
 
@@ -44,9 +45,9 @@ class AbstractHeadSpec extends Specification {
             called = true
         }
         def act = head.flux
-        source.tryEmitNext(blocks[0])
         then:
         StepVerifier.create(act)
+                .then { source.tryEmitNext(blocks[0]) }
                 .expectNext(blocks[0])
                 .then {
                     assert called
@@ -69,9 +70,9 @@ class AbstractHeadSpec extends Specification {
         when:
         head.follow(source.asFlux())
         def act = head.flux
-        source.tryEmitNext(blocks[0])
         then:
         StepVerifier.create(act)
+                .then { source.tryEmitNext(blocks[0]) }
                 .expectNext(blocks[0])
                 .then { source.tryEmitNext(blocks[1]) }
                 .expectNext(blocks[1])
@@ -97,9 +98,9 @@ class AbstractHeadSpec extends Specification {
         when:
         head.follow(source.asFlux())
         def act = head.flux
-        source.tryEmitNext(blocks[0])
         then:
         StepVerifier.create(act)
+                .then { source.tryEmitNext(blocks[0]) }
                 .expectNext(blocks[0])
                 .then { source.tryEmitNext(blocks[1]) }
                 .expectNext(blocks[1])

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy

@@ -52,12 +52,10 @@ class EthereumWsFactorySpec extends Specification {
 
         when:
         def act = Flux.from(ws.getFlux())
-        new Thread({
-            ws.onNewBlock(block)
-        }).run()
 
         then:
         StepVerifier.create(act)
+                .then { ws.onNewBlock(block) }
                 .expectNext(BlockContainer.from(block))
                 .thenCancel()
                 .verify(Duration.ofSeconds(1))

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -87,7 +87,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        upstream.start()
+        new Thread({ Thread.sleep(50); upstream.start() }).start()
         def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
         then:
         callData.chain == Chain.ETHEREUM.id
@@ -145,7 +145,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        upstream.start()
+        new Thread({ Thread.sleep(50); upstream.start() }).start()
         def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
         then:
         upstream.status == UpstreamAvailability.OK
@@ -207,7 +207,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        upstream.start()
+        new Thread({ Thread.sleep(50); upstream.start() }).start()
         finished.get()
         def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
         then:

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy

@@ -66,13 +66,13 @@ class GrpcHeadSpec extends Specification {
         when:
         def act = head.getFlux()
                 .take(3)
-        head.start(client)
 
         then:
         StepVerifier.create(act)
-                .expectNext(TestingCommons.blockForBitcoin(10))
+                .then { head.start(client) }
-                .expectNext(TestingCommons.blockForBitcoin(11))
+                .expectNext(TestingCommons.blockForBitcoin(10)).as("block 10")
-                .expectNext(TestingCommons.blockForBitcoin(12))
+                .expectNext(TestingCommons.blockForBitcoin(11)).as("block 11")
+                .expectNext(TestingCommons.blockForBitcoin(12)).as("block 12")
                 .expectComplete()
                 .verify(Duration.ofSeconds(5))
     }