From 549b8b1b5df77766653c281cd771f467094a107e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 30 Apr 2020 23:42:51 -0400 Subject: [PATCH] problem: Ethereum Tx tracking has complex non-reactive logic --- .../dshackle/rpc/TrackEthereumTx.kt | 317 ++++++++---------- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 216 +++++------- 2 files changed, 221 insertions(+), 312 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index d542cf05..4c0cda3b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt +++ b/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 upstreamScheduler: Scheduler + @Autowired private val upstreams: Upstreams ) : 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) - private val clients = HashMap>() - private val seq = AtomicLong(0) + var scheduler: Scheduler = Schedulers.elastic() - val notFound = ConcurrentLinkedQueue() + private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java) override fun isSupported(chain: Chain): Boolean { return BlockchainType.fromBlockchain(chain) == BlockchainType.ETHEREUM && upstreams.isAvailable(chain) } - @PostConstruct - fun init() { - upstreams.observeChains().subscribe { chain -> - clients[chain] = ConcurrentLinkedQueue() - upstreams.getUpstream(chain)?.getHead()?.let { head -> - head.getFlux().subscribe { verifyAll(chain) } + override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux { + val base = prepareTracking(request) + val up = upstreams.getUpstream(base.chain)?.castApi(EthereumApi::class.java) + ?: return Flux.empty() + return update(base) + .defaultIfEmpty(base) + .flatMapMany { + Flux.concat(Mono.just(it), subscribe(it, up)) + .distinctUntilChanged(TxDetails::status) + .map(this@TrackEthereumTx::asProto) + .subscribeOn(scheduler) + } + } + + fun subscribe(base: TxDetails, up: Upstream): Flux { + var latestTx = base + + val untilFound = Mono.just(latestTx) + .subscribeOn(scheduler) + .map { + //replace with the latest value, it may be already found + 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) + } + + return Flux.merge(untilFound, inBlocks) + .takeUntil(TxDetails::shouldClose) + .doOnNext { newTx -> + latestTx = newTx + } + } + + fun onNewBlock(tx: TxDetails, block: BlockContainer): Mono { + val txid = TxId.from(tx.txid) + if (!tx.status.mined) { + val justMined = block.transactions.contains(txid) + return if (justMined) { + Mono.just(tx.withStatus( + mined = true, + found = true, + 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) } } - @Scheduled(fixedDelay = 5000) - fun recheckFreshNotFound() { - val recent = Instant.now().minus(FRESH_TX) - verifyAll( - notFound.filter { - val tx = it.tx - !tx.status.found && tx.since.isAfter(recent) - }.map { it.tx } - ) - } - - @Scheduled(fixedRate = 60000) - fun recheckMatureNotFound() { - val recent = Instant.now().minus(FRESH_TX) - verifyAll( - 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? { - return clients[chain]?.map { it.tx } - } - - fun onFirstUpdate(tx: TxDetails) { - tx.backref?.let { backref -> - clients[tx.chain]?.add(backref) - if (!tx.status.found) { - notFound.add(backref) - } - } - } - - fun onFinished(chain: Chain, txid: Long) { - clients[chain]?.removeIf { x -> x.tx.id == txid } - notFound.removeIf { x -> x.tx.id == txid } - } - - fun onSend(tx: TxDetails) { - val curr = tx.justNotified().makeCurrent() - if (curr.shouldClose()) { - curr.bus.onComplete() - } + private fun update(tx: TxDetails): Mono { + val initialStatus = tx.status + val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream? + ?: 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() + } + .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.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() val details = TxDetails( chain, - bus, Instant.now(), TransactionId.from(request.txId), - min(max(1, request.confirmationLimit), 100), - seq.incrementAndGet() + min(max(1, request.confirmationLimit), 100) ) - val tracked = TrackedTx(details) - return details.withBackref(tracked) - } - - fun streamAllUpdates(tx: TxDetails): Flux { - 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 { - val tx = prepareTracking(request) - return streamAllUpdates(tx) - } - - private fun verifyAll(chain: Chain) { - verifyAll(trackedForChain(chain)!!) - } - - private fun verifyAll(list: Collection) { - 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) - } + return details } fun setBlockDetails(tx: TxDetails, block: BlockJson): TxDetails { @@ -255,27 +238,6 @@ class TrackEthereumTx( } } - private fun checkForUpdate(tx: TxDetails): Mono { - val initialStatus = tx.status - val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream? - ?: 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() - } - .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, 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, val since: Instant, val txid: TransactionId, val maxConfirmations: Int, - val id: Long, - val backref: TrackedTx? = null, - val status: TxStatus = TxStatus(), - val notifiedAt: Instant = Instant.now().minus(Period.ofDays(1)) + val status: TxStatus ) { + 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, - notifiedAt: Instant = this.notifiedAt - ) = TxDetails(chain, bus, since, txid, maxConfirmations, id, backref, status, notifiedAt) + status: TxStatus = this.status + ) = TxDetails(chain, since, txid, maxConfirmations, status) 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 { - return this.notifiedAt.isBefore(Instant.now().minus(PING_PERIOD)) + override fun toString(): String { + return "TxDetails(chain=$chain, txid=$txid, status=$status)" } - fun justNotified(): TxDetails { - return notifiedAt(Instant.now()) + override fun equals(other: Any?): Boolean { + 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 { - return copy(notifiedAt = time) + override fun hashCode(): Int { + 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)" + } + + } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index 9dc4c2f4..78e3c1d5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/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.scheduler.Schedulers +import reactor.core.publisher.Flux 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()) - trackTx.init() + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) 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()) - trackTx.init() + ((EthereumChainUpstreams) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { + _ * 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 { - return trackTx.subscribe(req) - } + def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) + def act = StepVerifier.withVirtualTime( + { trackTx.subscribe(tx, upstreams.getUpstream(Chain.ETHEREUM).castApi(EthereumApi.class)) }, + { scheduler }, + 5) + then: act - .expectNext(exp1) - .then { - 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 } - } + .expectSubscription() + .expectNoEvent(Duration.ofSeconds(20)).as("Waited for updates") .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()) - trackTx.init() + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) + 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 { - return trackTx.subscribe(req) - } + def act = StepVerifier.withVirtualTime({ + return trackTx.subscribe(req).take(2) + }, { scheduler }, 5) then: act - .expectNext(exp1) - .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(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) + .expectSubscription() + .expectNext(exp1).as("Unknown tx") + .expectNext(exp2).as("Found in mempool") .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()) - trackTx.init() + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) 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 - } }