problem: Ethereum Tx tracking has complex non-reactive logic

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

@@ -21,6 +21,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.SilentException
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.upstream.AggregatedUpstream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
@@ -36,169 +38,150 @@ import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.core.publisher.TopicProcessor
-import reactor.core.publisher.toFlux
 import reactor.core.scheduler.Scheduler
-import reactor.util.function.Tuples
+import reactor.core.scheduler.Schedulers
+import reactor.util.retry.Retry
 import java.math.BigInteger
 import java.time.Duration
 import java.time.Instant
-import java.time.Period
-import java.util.concurrent.ConcurrentLinkedQueue
-import java.util.concurrent.atomic.AtomicLong
-import javax.annotation.PostConstruct
-import kotlin.collections.HashMap
 import kotlin.math.max
 import kotlin.math.min
 
 @Service
 class TrackEthereumTx(
-        @Autowired private val upstreams: Upstreams,
+        @Autowired private val upstreams: Upstreams
-        @Autowired private val upstreamScheduler: Scheduler
 ) : TrackTx {
 
     companion object {
         private val ZERO_BLOCK = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000000000")
-        private val FRESH_TX = Duration.ofSeconds(60)
         private val TRACK_TTL = Duration.ofHours(1)
         private val NOT_FOUND_TRACK_TTL = Duration.ofMinutes(1)
         private val NOT_MINED_TRACK_TTL = NOT_FOUND_TRACK_TTL.multipliedBy(2)
-        private val PING_PERIOD = Duration.ofMinutes(5)
     }
 
-    private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
+    var scheduler: Scheduler = Schedulers.elastic()
-    private val clients = HashMap<Chain, ConcurrentLinkedQueue<TrackedTx>>()
-    private val seq = AtomicLong(0)
 
-    val notFound = ConcurrentLinkedQueue<TrackedTx>()
+    private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
 
     override fun isSupported(chain: Chain): Boolean {
         return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain)
     }
 
-    @PostConstruct
+    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
-    fun init() {
+        val base = prepareTracking(request)
-        upstreams.observeChains().subscribe { chain ->
+        val up = upstreams.getUpstream(base.chain)?.castApi(EthereumApi::class.java)
-            clients[chain] = ConcurrentLinkedQueue()
+                ?: return Flux.empty()
-            upstreams.getUpstream(chain)?.getHead()?.let { head ->
+        return update(base)
-                head.getFlux().subscribe { verifyAll(chain) }
+                .defaultIfEmpty(base)
-            }
+                .flatMapMany {
+                    Flux.concat(Mono.just(it), subscribe(it, up))
+                            .distinctUntilChanged(TxDetails::status)
+                            .map(this@TrackEthereumTx::asProto)
+                            .subscribeOn(scheduler)
                 }
     }
 
-    @Scheduled(fixedDelay = 5000)
+    fun subscribe(base: TxDetails, up: Upstream<EthereumApi>): Flux<TxDetails> {
-    fun recheckFreshNotFound() {
+        var latestTx = base
-        val recent = Instant.now().minus(FRESH_TX)
+
-        verifyAll(
+        val untilFound = Mono.just(latestTx)
-                notFound.filter {
+                .subscribeOn(scheduler)
-                    val tx = it.tx
+                .map {
-                    !tx.status.found && tx.since.isAfter(recent)
+                    //replace with the latest value, it may be already found
-                }.map { it.tx }
+                    latestTx
+                }
+                .flatMap { latest ->
+                    if (!latest.status.found) {
+                        update(latest).defaultIfEmpty(latestTx)
+                    } else {
+                        Mono.just(latest)
+                    }
+                }
+                .flatMap { received ->
+                    if (!received.status.found) {
+                        Mono.error(SilentException("Retry not found"))
+                    } else {
+                        Mono.just(received)
+                    }
+                }
+                .retryWhen(
+                        Retry.fixedDelay(10, Duration.ofSeconds(2))
                 )
+                .onErrorResume { Mono.empty() }
+
+        val inBlocks = up.getHead().getFlux()
+                .subscribeOn(scheduler)
+                .flatMap { block ->
+                    onNewBlock(latestTx, block)
                 }
 
-    @Scheduled(fixedRate = 60000)
+        return Flux.merge(untilFound, inBlocks)
-    fun recheckMatureNotFound() {
+                .takeUntil(TxDetails::shouldClose)
-        val recent = Instant.now().minus(FRESH_TX)
+                .doOnNext { newTx ->
-        verifyAll(
+                    latestTx = newTx
-                notFound.filter {
-                    !it.tx.status.found && it.tx.since.isBefore(recent)
-                }.map { it.tx }
-        )
-    }
-
-    fun recheckNotFound() {
-        verifyAll(notFound.map { it.tx })
-    }
-
-    @Scheduled(fixedRate = 60000, initialDelay = 10000)
-    fun cleanupNotFound() {
-        notFound.removeIf {
-            it.tx.shouldClose() || it.tx.status.found
                 }
     }
 
-    private fun trackedForChain(chain: Chain): List<TxDetails>? {
+    fun onNewBlock(tx: TxDetails, block: BlockContainer): Mono<TxDetails> {
-        return clients[chain]?.map { it.tx }
+        val txid = TxId.from(tx.txid)
-    }
+        if (!tx.status.mined) {
-
+            val justMined = block.transactions.contains(txid)
-    fun onFirstUpdate(tx: TxDetails) {
+            return if (justMined) {
-        tx.backref?.let { backref ->
+                Mono.just(tx.withStatus(
-            clients[tx.chain]?.add(backref)
+                        mined = true,
-            if (!tx.status.found) {
+                        found = true,
-                notFound.add(backref)
+                        confirmations = 1,
+                        height = block.height,
+                        blockTime = block.timestamp,
+                        blockTotalDifficulty = block.difficulty,
+                        blockHash = BlockHash(block.hash.value)
+                ))
+            } else {
+                update(tx)
             }
+        } else {
+            //verify if it's still on chain
+            //TODO head is supposed to erase block when it was replaced, so can safely recalc here
+            return update(tx)
         }
     }
 
-    fun onFinished(chain: Chain, txid: Long) {
+    private fun update(tx: TxDetails): Mono<TxDetails> {
-        clients[chain]?.removeIf { x -> x.tx.id == txid }
+        val initialStatus = tx.status
-        notFound.removeIf { x -> x.tx.id == txid }
+        val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream<EthereumApi>?
+                ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain))
+        val execution = upstream.getApi(Selector.empty)
+                .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) }
+        return execution
+                .onErrorResume(RpcException::class.java) { t ->
+                    log.warn("Upstream error, ignoring. {}", t.rpcMessage)
+                    Mono.empty<TransactionJson>()
                 }
-
+                .flatMap { updateFromBlock(upstream, tx, it) }
-    fun onSend(tx: TxDetails) {
+                .doOnError { t ->
-        val curr = tx.justNotified().makeCurrent()
+                    log.error("Failed to load tx block", t)
-        if (curr.shouldClose()) {
+                }
-            curr.bus.onComplete()
+                .switchIfEmpty(Mono.just(tx.withStatus(found = false)))
+                .filter { current ->
+                    initialStatus != current.status || current.shouldClose()
                 }
     }
 
     fun prepareTracking(request: BlockchainOuterClass.TxStatusRequest): TxDetails {
         val chain = Chain.byId(request.chainValue)
-        if (BlockchainType.fromBlockchain(chain) != BlockchainType.ETHEREUM) {
+        if (!isSupported(chain)) {
             throw SilentException.UnsupportedBlockchain(request.chainValue)
         }
-        if (!clients.containsKey(chain)) {
-            throw SilentException.UnsupportedBlockchain(chain)
-        }
-        val bus = TopicProcessor.create<Notification>()
         val details = TxDetails(
                 chain,
-                bus,
                 Instant.now(),
                 TransactionId.from(request.txId),
-                min(max(1, request.confirmationLimit), 100),
+                min(max(1, request.confirmationLimit), 100)
-                seq.incrementAndGet()
         )
-        val tracked = TrackedTx(details)
+        return details
-        return details.withBackref(tracked)
-    }
-
-    fun streamAllUpdates(tx: TxDetails): Flux<BlockchainOuterClass.TxStatus> {
-        val current = checkForUpdate(tx)
-                .doOnNext(this::onFirstUpdate)
-                .map { Notification(it, asProto(it)) }
-        val updates = Flux.from(tx.bus)
-
-        return Flux.concat(current, updates)
-                .doFinally { onFinished(tx.chain, tx.id) }
-                .doOnNext { onSend(it.tx) }
-                .map { it.proto }
-    }
-
-    override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
-        val tx = prepareTracking(request)
-        return streamAllUpdates(tx)
-    }
-
-    private fun verifyAll(chain: Chain) {
-        verifyAll(trackedForChain(chain)!!)
-    }
-
-    private fun verifyAll(list: Collection<TxDetails>) {
-        list
-                .toFlux()
-                .parallel(8).runOn(upstreamScheduler)
-                .flatMap { checkForUpdate(it) }
-                .sequential()
-                .map { Tuples.of(it, asProto(it)) }
-                .subscribe { t ->
-                    notify(t.t1.bus, t.t2, t.t1)
-                }
     }
 
     fun setBlockDetails(tx: TxDetails, block: BlockJson<TransactionRefJson>): TxDetails {
@@ -255,27 +238,6 @@ class TrackEthereumTx(
         }
     }
 
-    private fun checkForUpdate(tx: TxDetails): Mono<TxDetails> {
-        val initialStatus = tx.status
-        val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream<EthereumApi>?
-                ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain))
-        val execution = upstream.getApi(Selector.empty)
-                .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) }
-        return execution
-                .onErrorResume(RpcException::class.java) { t ->
-                    log.warn("Upstream error, ignoring. {}", t.rpcMessage)
-                    Mono.empty<TransactionJson>()
-                }
-                .flatMap { updateFromBlock(upstream, tx, it) }
-                .doOnError { t ->
-                    log.error("Failed to load tx block", t)
-                }
-                .switchIfEmpty(Mono.just(tx.withStatus(found = false)))
-                .filter { current ->
-                    initialStatus != current.status || current.shouldNotify() || current.shouldClose()
-                }
-    }
-
     private fun asProto(tx: TxDetails): BlockchainOuterClass.TxStatus {
         val data = BlockchainOuterClass.TxStatus.newBuilder()
                 .setTxId(tx.txid.toHex())
@@ -296,35 +258,22 @@ class TrackEthereumTx(
         return data.build()
     }
 
-    private fun notify(client: TopicProcessor<Notification>, data: BlockchainOuterClass.TxStatus, tx: TxDetails) {
-        try {
-            client.onNext(Notification(tx, data))
-        } catch (t: Throwable) {
-            log.warn("Failed to put to bus", t)
-        }
-    }
-
-    class Notification(val tx: TxDetails, val proto: BlockchainOuterClass.TxStatus)
-
-    class TrackedTx(var tx: TxDetails)
-
     class TxDetails(val chain: Chain,
-                    val bus: TopicProcessor<Notification>,
                     val since: Instant,
                     val txid: TransactionId,
                     val maxConfirmations: Int,
-                    val id: Long,
+                    val status: TxStatus
-                    val backref: TrackedTx? = null,
-                    val status: TxStatus = TxStatus(),
-                    val notifiedAt: Instant = Instant.now().minus(Period.ofDays(1))
     ) {
 
+        constructor(chain: Chain,
+                    since: Instant,
+                    txid: TransactionId,
+                    maxConfirmations: Int) : this(chain, since, txid, maxConfirmations, TxStatus())
+
         fun copy(
                 since: Instant = this.since,
-                backref: TrackedTx? = this.backref,
+                status: TxStatus = this.status
-                status: TxStatus = this.status,
+        ) = TxDetails(chain, since, txid, maxConfirmations, status)
-                notifiedAt: Instant = this.notifiedAt
-        ) = TxDetails(chain, bus, since, txid, maxConfirmations, id, backref, status, notifiedAt)
 
         fun withStatus(found: Boolean = this.status.found,
                        height: Long? = this.status.height,
@@ -336,10 +285,6 @@ class TrackEthereumTx(
             return copy(status = this.status.copy(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmations))
         }
 
-        fun withCleanStatus(): TxDetails {
-            return copy(status = this.status.clean())
-        }
-
         fun shouldClose(): Boolean {
             return maxConfirmations <= this.status.confirmations
                     || since.isBefore(Instant.now().minus(TRACK_TTL))
@@ -347,26 +292,31 @@ class TrackEthereumTx(
                     || (!status.mined && since.isBefore(Instant.now().minus(NOT_MINED_TRACK_TTL)))
         }
 
-        fun shouldNotify(): Boolean {
+        override fun toString(): String {
-            return this.notifiedAt.isBefore(Instant.now().minus(PING_PERIOD))
+            return "TxDetails(chain=$chain, txid=$txid, status=$status)"
         }
 
-        fun justNotified(): TxDetails {
+        override fun equals(other: Any?): Boolean {
-            return notifiedAt(Instant.now())
+            if (this === other) return true
+            if (other !is TxDetails) return false
+
+            if (chain != other.chain) return false
+            if (since != other.since) return false
+            if (txid != other.txid) return false
+            if (maxConfirmations != other.maxConfirmations) return false
+            if (status != other.status) return false
+
+            return true
         }
 
-        fun notifiedAt(time: Instant): TxDetails {
+        override fun hashCode(): Int {
-            return copy(notifiedAt = time)
+            var result = chain.hashCode()
+            result = 31 * result + since.hashCode()
+            result = 31 * result + txid.hashCode()
+            result = 31 * result + status.hashCode()
+            return result
         }
 
-        fun withBackref(backref: TrackedTx): TxDetails {
-            return copy(backref = backref)
-        }
-
-        fun makeCurrent(): TxDetails {
-            backref?.tx = this
-            return this
-        }
     }
 
     class TxStatus(val found: Boolean = false,
@@ -412,5 +362,10 @@ class TrackEthereumTx(
             return result
         }
 
+        override fun toString(): String {
+            return "TxStatus(found=$found, height=$height, mined=$mined, blockHash=$blockHash, blockTime=$blockTime, blockTotalDifficulty=$blockTotalDifficulty, confirmations=$confirmations)"
+        }
+
+
     }
 }

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

@@ -20,9 +20,15 @@ import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.test.UpstreamsMock
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
@@ -30,9 +36,10 @@ import io.infinitape.etherjar.rpc.ReactorRpcClient
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
-import reactor.core.publisher.Mono
+import reactor.core.publisher.Flux
-import reactor.core.scheduler.Schedulers
 import reactor.test.StepVerifier
+import reactor.test.scheduler.VirtualTimeScheduler
+import spock.lang.Ignore
 import spock.lang.Specification
 
 import java.time.Duration
@@ -96,8 +103,7 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
         def upstreamMock = TestingCommons.upstream(apiMock)
         Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
-        trackTx.init()
 
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
         apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson)
@@ -112,54 +118,36 @@ class TrackEthereumTxSpec extends Specification {
                 .verify(Duration.ofSeconds(4))
     }
 
-    def "Closes for unknown transaction"() {
+    def "Wait for unknown transaction"() {
         setup:
-        def req = BlockchainOuterClass.TxStatusRequest.newBuilder()
-                .setChain(chain)
-                .setConfirmationLimit(6)
-                .setTxId(txId)
-                .build()
-        def exp1 = BlockchainOuterClass.TxStatus.newBuilder()
-                .setTxId(txId)
-                .setBroadcasted(false)
-                .setMined(false)
-                .build()
-
         def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
         def upstreamMock = TestingCommons.upstream(apiMock)
         Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
+        ((EthereumChainUpstreams) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) {
-        trackTx.init()
+            _ * getFlux() >> Flux.empty()
+        }
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
+        def scheduler = VirtualTimeScheduler.create(true)
+        trackTx.scheduler = scheduler
 
         apiMock.answer("eth_getTransactionByHash", [txId], null)
 
         when:
-        def act = StepVerifier.withVirtualTime {
+        def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
-            return trackTx.subscribe(req)
+        def act = StepVerifier.withVirtualTime(
-        }
+                { trackTx.subscribe(tx, upstreams.getUpstream(Chain.ETHEREUM).castApi(EthereumApi.class)) },
+                { scheduler },
+                5)
+
         then:
         act
-                .expectNext(exp1)
+                .expectSubscription()
-                .then {
+                .expectNoEvent(Duration.ofSeconds(20)).as("Waited for updates")
-                    assert trackTx.notFound.any { it.tx.txid.toHex() == txId }
-                }
-                .expectNoEvent(Duration.ofSeconds(30))
-                .then {
-                    def track = trackTx.notFound.iterator().next()
-                    track.tx
-                            .copy(Instant.now() - Duration.ofHours(1), track, track.tx.status, Instant.now() - Duration.ofMinutes(15))
-                            .makeCurrent()
-                    trackTx.recheckNotFound()
-                }
-                .expectNext(exp1)
-                .then {
-                    assert !trackTx.notFound.any { it.tx.txid.toHex() == txId }
-                }
                 .expectComplete()
-                .verify(Duration.ofSeconds(4))
+                .verify(Duration.ofSeconds(3))
     }
 
-    def "Closes for known transaction if not mined"() {
+    def "Known transaction when not mined"() {
         setup:
         def req = BlockchainOuterClass.TxStatusRequest.newBuilder()
                 .setChain(chain)
@@ -185,49 +173,70 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
         def upstreamMock = TestingCommons.upstream(apiMock)
         Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
-        trackTx.init()
+        def scheduler = VirtualTimeScheduler.create(true)
+        trackTx.scheduler = scheduler
 
         apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
 
         when:
-        def act = StepVerifier.withVirtualTime {
+        def act = StepVerifier.withVirtualTime({
-            return trackTx.subscribe(req)
+            return trackTx.subscribe(req).take(2)
-        }
+        }, { scheduler }, 5)
         then:
         act
-                .expectNext(exp1)
+                .expectSubscription()
-                .then {
+                .expectNext(exp1).as("Unknown tx")
-                    assert trackTx.notFound.any { it.tx.txid.toHex() == txId }
+                .expectNext(exp2).as("Found in mempool")
-                }
-                .expectNoEvent(Duration.ofSeconds(20))
-
-                .then {
-                    def track = trackTx.notFound.iterator().next()
-                    track.tx
-                            .copy(Instant.now() - Duration.ofSeconds(119), track, track.tx.status, Instant.now() - Duration.ofMinutes(15))
-                            .makeCurrent()
-                    trackTx.recheckNotFound()
-                }
-                .expectNext(exp2)
-                .then {
-                    assert trackTx.notFound.any { it.tx.txid.toHex() == txId }
-                }
-                .expectNoEvent(Duration.ofSeconds(20))
-
-                .then {
-                    def track = trackTx.notFound.iterator().next()
-                    track.tx
-                            .copy(Instant.now() - Duration.ofSeconds(121), track, track.tx.status, Instant.now() - Duration.ofMinutes(15))
-                            .makeCurrent()
-                    trackTx.recheckNotFound()
-                }
-                .expectNext(exp2)
                 .expectComplete()
                 .verify(Duration.ofSeconds(4))
     }
 
+    def "New block makes tx mined"() {
+        setup:
+        def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
+        def upstreamMock = TestingCommons.upstream(apiMock)
+        Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
+
+        def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
+        def block = new BlockContainer(
+                100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes,
+                [TxId.from(txId)]
+        )
+
+        when:
+        def act = trackTx.onNewBlock(tx, block)
+        then:
+        StepVerifier.create(act)
+                .expectNext(tx.withStatus(true, 100, true, BlockHash.from(txId), block.timestamp, BigInteger.ONE, 1))
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "New block without current tx requires a call"() {
+        setup:
+        def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
+        def upstreamMock = TestingCommons.upstream(apiMock)
+        Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
+
+        def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
+        def block = new BlockContainer(
+                100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes,
+                [TxId.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22")]
+        )
+        apiMock.answer("eth_getTransactionByHash", [txId], null)
+
+        when:
+        def act = trackTx.onNewBlock(tx, block)
+        then:
+        StepVerifier.create(act)
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
     def "Starts to follow new transaction"() {
         setup:
         def req = BlockchainOuterClass.TxStatusRequest.newBuilder()
@@ -284,11 +293,11 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
         def upstreamMock = TestingCommons.upstream(apiMock)
         Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams)
-        trackTx.init()
 
         apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
         apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted)
+        apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted)
         apiMock.answer("eth_getTransactionByHash", [txId], txJsonMined)
         blocks.forEach { block ->
             apiMock.answer("eth_getBlockByHash", [block.hash.toHex(), false], block)
@@ -305,11 +314,11 @@ class TrackEthereumTxSpec extends Specification {
         def flux = trackTx.subscribe(req)
         then:
         StepVerifier.create(flux)
-                .expectNext(exp1.build())
+                .expectNext(exp1.build()).as("Just empty")
                 .then(nextBlock(1))
-                .expectNext(exp1.setBroadcasted(true).build())
+                .expectNext(exp1.setBroadcasted(true).build()).as("Found in mempool")
                 .then(nextBlock(2))
-                .expectNext(exp2.setConfirmations(1).build())
+                .expectNext(exp2.setConfirmations(1).build()).as("Mined")
                 .then(nextBlock(3))
                 .expectNext(exp2.setConfirmations(2).build())
                 .then(nextBlock(4))
@@ -320,59 +329,4 @@ class TrackEthereumTxSpec extends Specification {
                 .verify(Duration.ofSeconds(4))
     }
 
-    def "Tracked after first load"() {
-        setup:
-        def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
-        def upstreamMock = TestingCommons.upstream(apiMock)
-        Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
-        trackTx.init()
-
-        def req = BlockchainOuterClass.TxStatusRequest.newBuilder()
-                .setChain(chain)
-                .setConfirmationLimit(6)
-                .setTxId(txId)
-                .build()
-        def tx = trackTx.prepareTracking(req)
-        when:
-        trackTx.onFirstUpdate(tx)
-        then:
-        trackTx.notFound.any { it.tx.txid == TransactionId.from(txId) }
-        trackTx.trackedForChain(Chain.ETHEREUM).any { it.txid == TransactionId.from(txId) }
-    }
-
-    def "Update of last notified keeps everything else"() {
-        setup:
-        def apiMock = TestingCommons.api(Stub(ReactorRpcClient))
-        def upstreamMock = TestingCommons.upstream(apiMock)
-        Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock)
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.immediate())
-        trackTx.init()
-
-        def req = BlockchainOuterClass.TxStatusRequest.newBuilder()
-                .setChain(chain)
-                .setConfirmationLimit(6)
-                .setTxId(txId)
-                .build()
-        def tx = trackTx.prepareTracking(req)
-        when:
-        tx = tx.withStatus(true, 100,
-                true, BlockHash.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22"), Instant.now(),
-                1000 as BigInteger, 15L)
-        then:
-        tx.status.found
-        tx.notifiedAt.isBefore(Instant.now() - Duration.ofSeconds(5))
-        when:
-        def c = tx.justNotified()
-        then:
-        tx.notifiedAt.isBefore(Instant.now() - Duration.ofSeconds(5))
-        c.notifiedAt.isAfter(Instant.now() - Duration.ofSeconds(5))
-        c.status.found
-        c.status.height == 100
-        c.status.mined
-        c.status.blockHash.toHex() == "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22"
-        c.status.blockTime.isAfter(Instant.now() - Duration.ofSeconds(5))
-        c.status.blockTotalDifficulty == 1000 as BigInteger
-        c.status.confirmations == 15L
-    }
 }