StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

problem: doesn't track edge cases correctly, like not broadcasted transaction

Browse Source
This commit is contained in:
Igor Artamonov
2019-08-09 23:36:40 -04:00
parent 5660f4ca09
commit 07bca4fa22
2 changed files with 404 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

348
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
View File

@@ -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<Chain, ConcurrentLinkedQueue<TrackedTx>>()
private val seq = AtomicLong(0)
val notFound = ConcurrentLinkedQueue<TrackedTx>()
@PostConstruct
fun init() {
availableChains.observe().subscribe { chain ->
@@ -52,175 +67,288 @@ class TrackTx(
}
}
private fun currentList(chain: Chain): ConcurrentLinkedQueue<TrackedTx>? {
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<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
return requestMono.map { request ->
val bus = TopicProcessor.create<BlockchainOuterClass.TxStatus>()
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<TxDetails>? {
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<Notification>()
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<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 }
}
fun add(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
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<TxDetails>) {
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<TrackedTx> {
fun setBlockDetails(tx: TxDetails, block: BlockJson<TransactionId>): 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<TxDetails> {
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<TrackedTx> {
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<TxDetails> {
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<TxDetails> {
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<BlockchainOuterClass.TxStatus>, data: BlockchainOuterClass.TxStatus) {
client.onNext(data)
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 TrackedTx(val chain: Chain,
val bus: TopicProcessor<BlockchainOuterClass.TxStatus>,
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 = 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
}
}
}

171
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy
View File

@@ -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
}
}