From 07bca4fa22b8ef9e4c3dfed1d9f47913d76e5694 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 9 Aug 2019 23:36:40 -0400 Subject: [PATCH] problem: doesn't track edge cases correctly, like not broadcasted transaction --- .../io/emeraldpay/dshackle/rpc/TrackTx.kt | 348 ++++++++++++------ .../dshackle/rpc/TrackTxSpec.groovy | 171 ++++++++- 2 files changed, 404 insertions(+), 115 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index 920bd37b..a63f84ec 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -5,13 +5,17 @@ import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson 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 @@ -23,9 +27,11 @@ import java.lang.Exception 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 @@ -36,12 +42,21 @@ class TrackTx( @Autowired private val upstreamScheduler: Scheduler ) { - private val ZERO_BLOCK = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000000000") + 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(TrackTx::class.java) private val clients = HashMap>() private val seq = AtomicLong(0) + val notFound = ConcurrentLinkedQueue() + @PostConstruct fun init() { availableChains.observe().subscribe { chain -> @@ -52,175 +67,288 @@ class TrackTx( } } - private fun currentList(chain: Chain): ConcurrentLinkedQueue? { - return clients[chain] + @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 } + ) } - fun add(requestMono: Mono): Flux { - return requestMono.map { request -> - val bus = TopicProcessor.create() - TrackTx.TrackedTx( - Chain.byId(request.chainValue), - bus, - Instant.now(), - TransactionId.from(request.txId), - min(max(1, request.confirmationLimit), 100), - seq.incrementAndGet() - ) - }.filter { - clients.containsKey(it.chain) - }.flatMapMany { tx -> - val current = checkForUpdate(tx).doOnNext{ - currentList(tx.chain)?.add(tx) - }.map(this::asProto) + @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 } + ) + } - val next = Flux.from(tx.bus) - Flux.merge(current, next).doFinally { - currentList(tx.chain)?.removeIf { x -> x.id == tx.id } - }.doOnNext { txp -> - if (txp.confirmations >= tx.maxConfirmations) { - tx.bus.onComplete() - } + 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() + } + } + + fun prepareTracking(request: BlockchainOuterClass.TxStatusRequest): TxDetails { + val chain = Chain.byId(request.chainValue) + if (!clients.containsKey(chain)) { + throw Exception("Unsupported blockchain: ${chain}") + } + val bus = TopicProcessor.create() + val details = TxDetails( + chain, + bus, + Instant.now(), + TransactionId.from(request.txId), + min(max(1, request.confirmationLimit), 100), + seq.incrementAndGet() + ) + 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 } + } + + fun add(requestMono: Mono): Flux { + return requestMono.map { request -> + prepareTracking(request) + }.flatMapMany { tx -> + streamAllUpdates(tx) + } + } + private fun verifyAll(chain: Chain) { - currentList(chain)!! + verifyAll(trackedForChain(chain)!!) + } + + private fun verifyAll(list: Collection) { + list .toFlux() .parallel(8).runOn(upstreamScheduler) - .flatMap { checkForUpdate(it) } + .flatMap { checkForUpdate(it) } .sequential() - .map { Tuples.of(it.bus, asProto(it)) } + .map { Tuples.of(it, asProto(it)) } .subscribe { t -> - notify(t.t1, t.t2) + notify(t.t1.bus, t.t2, t.t1) } } - private fun loadWeight(tx: TrackedTx): Mono { + fun setBlockDetails(tx: TxDetails, block: BlockJson): TxDetails { + return if (block.number != null && block.totalDifficulty != null) { + tx.withStatus( + blockTotalDifficulty = block.totalDifficulty, + blockTime = block.timestamp.toInstant() + ) + } else { + tx.withStatus( + mined = false + ) + } + } + + private fun loadWeight(tx: TxDetails): Mono { val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) return upstream.getApi(Selector.empty) .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) .map { block -> - if (block != null && block.number != null && block.totalDifficulty != null) { - tx.withStatus( - blockTotalDifficulty = block.totalDifficulty, - blockTime = block.timestamp.toInstant() - ) - } else { - tx.withStatus( - mined = false - ) - } + setBlockDetails(tx, block) }.doOnError { t -> log.warn("Failed to update weight", t) } } - private fun checkForUpdate(tx: TrackedTx): Mono { - val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) - val execution = upstream.getApi(Selector.empty) - .executeAndConvert(Commands.eth().getTransaction(tx.txid)) - return execution.flatMap { - if (it.blockNumber != null && it.blockHash != null && it.blockHash != ZERO_BLOCK) { - val updated = tx.withStatus( - blockHash = it.blockHash, - height = it.blockNumber, - found = true, - mined = true, - confirmations = 1 - ) - upstream.getHead().getHead().map { head -> - if (updated.status.height == null || head.number < updated.status.height) { - updated - } else { - updated.withStatus( - confirmations = head.number - updated.status.height + 1 - ) - } - }.flatMap(this::loadWeight) - } else { - Mono.just(tx.withStatus( - found = true, - mined = false - )) - } - }.switchIfEmpty(Mono.just(tx.withStatus(found = false))).filter { current -> - current.status != tx.status + fun updateFromBlock(upstream: Upstream, tx: TxDetails, it: TransactionJson): Mono { + return if (it.blockNumber != null && it.blockHash != null && it.blockHash != ZERO_BLOCK) { + val updated = tx.withStatus( + blockHash = it.blockHash, + height = it.blockNumber, + found = true, + mined = true, + confirmations = 1 + ) + upstream.getHead().getHead().map { head -> + val height = updated.status.height + if (height == null || head.number < height) { + updated + } else { + updated.withStatus( + confirmations = head.number - height + 1 + ) + } + }.doOnError { t -> + log.error("Unable to load head details", t) + }.flatMap(this::loadWeight) + } else { + Mono.just(tx.withStatus( + found = true, + mined = false + )) } } - private fun asProto(tx: TrackedTx): BlockchainOuterClass.TxStatus { + private fun checkForUpdate(tx: TxDetails): Mono { + val initialStatus = tx.status + val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) + val execution = upstream.getApi(Selector.empty) + .executeAndConvert(Commands.eth().getTransaction(tx.txid)) + return execution + .flatMap { updateFromBlock(upstream, tx, it) } + .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()) .setConfirmations(tx.status.confirmations.toInt()) - if (tx.status.found != null) { - data.broadcasted = tx.status.found - } - if (tx.status.mined != null) { - data.mined = tx.status.mined - if (tx.status.mined) { - data.setBlock( - Common.BlockInfo.newBuilder() - .setBlockId(tx.status.blockHash!!.toHex().substring(2)) - .setTimestamp(tx.status.blockTime!!.toEpochMilli()) - .setWeight(ByteString.copyFrom(tx.status.blockTotalDifficulty!!.toByteArray())) - .setHeight(tx.status.height!!) - .setTimestamp(tx.status.blockTime!!.toEpochMilli()) - ) - } + data.broadcasted = tx.status.found + val isMined = tx.status.mined + data.mined = isMined + if (isMined) { + data.setBlock( + Common.BlockInfo.newBuilder() + .setBlockId(tx.status.blockHash!!.toHex().substring(2)) + .setTimestamp(tx.status.blockTime!!.toEpochMilli()) + .setWeight(ByteString.copyFrom(tx.status.blockTotalDifficulty!!.toByteArray())) + .setHeight(tx.status.height!!) + ) } return data.build() } - private fun notify(client: TopicProcessor, data: BlockchainOuterClass.TxStatus) { - client.onNext(data) + 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 TrackedTx(val chain: Chain, - val bus: TopicProcessor, + 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 status: TxStatus = TxStatus()) { + val backref: TrackedTx? = null, + val status: TxStatus = TxStatus(), + val notifiedAt: Instant = Instant.now().minus(Period.ofDays(1)) + ) { - fun withStatus(found: Boolean? = this.status.found, + 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) + + fun withStatus(found: Boolean = this.status.found, height: Long? = this.status.height, - mined: Boolean? = this.status.mined, + mined: Boolean = this.status.mined, blockHash: BlockHash? = this.status.blockHash, blockTime: Instant? = this.status.blockTime, blockTotalDifficulty: BigInteger? = this.status.blockTotalDifficulty, - confirmations: Long = this.status.confirmations) - = TrackedTx( - chain, bus, since, txid, maxConfirmations, id, - this.status.copy(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmations) - ) + confirmations: Long = this.status.confirmations): TxDetails { + return copy(status = this.status.copy(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmations)) + } - fun withCleanStatus() - = TrackedTx( - chain, bus, since, txid, maxConfirmations, id, this.status.clean() - ) + fun withCleanStatus(): TxDetails { + return copy(status = this.status.clean()) + } fun shouldClose(): Boolean { return maxConfirmations <= this.status.confirmations - || since.isBefore(Instant.now().minus(Duration.ofHours(1))) + || since.isBefore(Instant.now().minus(TRACK_TTL)) + || (!status.found && since.isBefore(Instant.now().minus(NOT_FOUND_TRACK_TTL))) + || (!status.mined && since.isBefore(Instant.now().minus(NOT_MINED_TRACK_TTL))) + } + + fun shouldNotify(): Boolean { + return this.notifiedAt.isBefore(Instant.now().minus(PING_PERIOD)) + } + + fun justNotified(): TxDetails { + return notifiedAt(Instant.now()) + } + + fun notifiedAt(time: Instant): TxDetails { + return copy(notifiedAt = time) + } + + fun withBackref(backref: TrackedTx): TxDetails { + return copy(backref = backref) + } + + fun makeCurrent(): TxDetails { + backref?.tx = this + return this } } - class TxStatus(val found: Boolean? = null, + class TxStatus(val found: Boolean = false, val height: Long? = null, - val mined: Boolean? = null, + val mined: Boolean = false, val blockHash: BlockHash? = null, val blockTime: Instant? = null, val blockTotalDifficulty: BigInteger? = null, val confirmations: Long = 0) { - fun copy(found: Boolean? = this.found, + fun copy(found: Boolean = this.found, height: Long? = this.height, - mined: Boolean? = this.mined, + mined: Boolean = this.mined, blockHash: BlockHash? = this.blockHash, blockTime: Instant? = this.blockTime, blockTotalDifficulty: BigInteger? = this.blockTotalDifficulty, @@ -253,7 +381,5 @@ class TrackTx( return result } - } - } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy index c166695f..1a9dcddf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy @@ -3,7 +3,6 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.AggregatedUpstreams import io.emeraldpay.dshackle.upstream.AvailableChains @@ -22,6 +21,7 @@ import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration +import java.time.Instant class TrackTxSpec extends Specification { @@ -37,7 +37,7 @@ class TrackTxSpec extends Specification { trackTx = new TrackTx(upstreams, availableChains, Schedulers.immediate()) } - def start() { + def startTrackTxService() { trackTx.init() availableChains.add(Chain.ETHEREUM) availableChains.add(Chain.TESTNET_KOVAN) @@ -103,7 +103,7 @@ class TrackTxSpec extends Specification { _ * upstreamMock.getHead() >> headMock _ * headMock.getFlux() >> blocksBus _ * headMock.getHead() >> Mono.just(blockHeadJson) - start() + startTrackTxService() when: def flux = trackTx.add(Mono.just(req)) @@ -114,6 +114,122 @@ class TrackTxSpec extends Specification { .verify(Duration.ofSeconds(4)) } + def "Closes 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 upstreamMock = Mock(AggregatedUpstreams) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) + apiMock.answer("eth_getTransactionByHash", [txId], null) + + _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock + _ * upstreamMock.getApi(_) >> apiMock + startTrackTxService() + + when: + def act = StepVerifier.withVirtualTime { + return trackTx.add(Mono.just(req)) + } + 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 } + } + .expectComplete() + .verify(Duration.ofSeconds(4)) + } + + def "Closes for known transaction if not mined"() { + 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 exp2 = BlockchainOuterClass.TxStatus.newBuilder() + .setTxId(txId) + .setBroadcasted(true) + .setMined(false) + .build() + def txJson = new TransactionJson().with { + it.hash = TransactionId.from(txId) + it.nonce = 1 + it + } + + def upstreamMock = Mock(AggregatedUpstreams) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) + apiMock.answerOnce("eth_getTransactionByHash", [txId], null) + apiMock.answer("eth_getTransactionByHash", [txId], txJson) + + _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock + _ * upstreamMock.getApi(_) >> apiMock + startTrackTxService() + + when: + def act = StepVerifier.withVirtualTime { + return trackTx.add(Mono.just(req)) + } + 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) + .expectComplete() + .verify(Duration.ofSeconds(4)) + } + def "Starts to follow new transaction"() { setup: def req = BlockchainOuterClass.TxStatusRequest.newBuilder() @@ -186,7 +302,7 @@ class TrackTxSpec extends Specification { _ * upstreamMock.getHead() >> headMock _ * headMock.getFlux() >> blocksBus _ * headMock.getHead() >> { return Mono.just(headBlock) } - start() + startTrackTxService() def nextBlock = { int i -> return { @@ -214,4 +330,51 @@ class TrackTxSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(4)) } + + def "Tracked after first load"() { + setup: + startTrackTxService() + + 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: + startTrackTxService() + 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 + } }