problem: sporadic unit-test failure

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

@@ -209,11 +209,11 @@ class TrackEthereumTx(
                 }
     }
 
-    fun updateFromBlock(upstream: Upstream<EthereumApi>, tx: TxDetails, it: TransactionJson): Mono<TxDetails> {
+    fun updateFromBlock(upstream: Upstream<EthereumApi>, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
-        return if (it.blockNumber != null && it.blockHash != null && it.blockHash != ZERO_BLOCK) {
+        return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) {
             val updated = tx.withStatus(
-                    blockHash = it.blockHash,
+                    blockHash = blockTx.blockHash,
-                    height = it.blockNumber,
+                    height = blockTx.blockNumber,
                     found = true,
                     mined = true,
                     confirmations = 1

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy

@@ -303,30 +303,23 @@ class TrackEthereumTxSpec extends Specification {
             apiMock.answer("eth_getBlockByHash", [block.hash.toHex(), false], block)
         }
 
-        def nextBlock = { int i ->
+        upstreamMock.blocks = Flux.fromIterable(blocks)
-            return {
+                .map { block ->
-                println("block $i");
+                    BlockContainer.from(block, TestingCommons.objectMapper())
-                upstreamMock.nextBlock(BlockContainer.from(blocks[i], TestingCommons.objectMapper()))
+                }
-            } as Runnable
-        }
 
         when:
         def flux = trackTx.subscribe(req)
         then:
         StepVerifier.create(flux)
                 .expectNext(exp1.build()).as("Just empty")
-                .then(nextBlock(1))
                 .expectNext(exp1.setBroadcasted(true).build()).as("Found in mempool")
-                .then(nextBlock(2))
                 .expectNext(exp2.setConfirmations(1).build()).as("Mined")
-                .then(nextBlock(3))
+                .expectNext(exp2.setConfirmations(2).build()).as("Confirmed 2")
-                .expectNext(exp2.setConfirmations(2).build())
+                .expectNext(exp2.setConfirmations(3).build()).as("Confirmed 3")
-                .then(nextBlock(4))
+                .expectNext(exp2.setConfirmations(4).build()).as("Confirmed 4")
-                .expectNext(exp2.setConfirmations(3).build())
-                .then(nextBlock(5))
-                .expectNext(exp2.setConfirmations(4).build())
                 .expectComplete()
-                .verify(Duration.ofSeconds(4))
+                .verify(Duration.ofSeconds(1))
     }
 
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.test
 
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.Head
+import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
@@ -25,6 +26,7 @@ import reactor.core.publisher.TopicProcessor
 class EthereumHeadMock implements Head {
 
     private TopicProcessor<BlockContainer> bus = TopicProcessor.create()
+    private Publisher<BlockContainer> predefined = null
     private BlockContainer latest
 
     void nextBlock(BlockContainer block) {
@@ -33,8 +35,20 @@ class EthereumHeadMock implements Head {
         bus.onNext(block)
     }
 
+    void setPredefined(Publisher<BlockContainer> predefined) {
+        this.predefined = Flux.from(predefined)
+                .publish()
+                .refCount(1)
+                .doOnNext { latest = it }
+        // keep the current block as latest, because getFlux is also used to get the current height
+    }
+
     @Override
     Flux<BlockContainer> getFlux() {
-        return Flux.concat(Mono.justOrEmpty(latest), bus).distinctUntilChanged()
+        if (predefined != null) {
+            return Flux.concat(Mono.justOrEmpty(latest), Flux.from(predefined))
+        } else {
+            return Flux.concat(Mono.justOrEmpty(latest), bus).distinctUntilChanged()
+        }
     }
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy

@@ -29,6 +29,7 @@ import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.jetbrains.annotations.NotNull
+import org.reactivestreams.Publisher
 
 class EthereumUpstreamMock extends EthereumUpstream {
 
@@ -58,6 +59,10 @@ class EthereumUpstreamMock extends EthereumUpstream {
         ethereumHeadMock.nextBlock(block)
     }
 
+    void setBlocks(Publisher<BlockContainer> blocks) {
+        ethereumHeadMock.predefined = blocks
+    }
+
     @Override
     Head createHead() {
         return ethereumHeadMock