diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 867cbd67..156a2770 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -493,8 +493,6 @@ abstract class Multistream( abstract fun getHead(mather: Selector.Matcher): Head - abstract fun getEnrichedHead(mather: Selector.Matcher): Head - // -------------------------------------------------------------------------------------------------------- class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 64d5fa6d..29556787 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -125,10 +125,6 @@ open class BitcoinMultistream( return head } - override fun getEnrichedHead(mather: Matcher): Head { - TODO("Not yet implemented") - } - override fun getLabels(): Collection { return sourceUpstreams.flatMap { it.getLabels() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt index 78daa36a..82326108 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt @@ -159,7 +159,7 @@ class EthereumCallSelector( } } - private fun blockByHash(blockHash: String): Mono { + fun blockByHash(blockHash: String): Mono { return try { caches.getLastHeightByHash() .read(BlockId.from(blockHash)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt deleted file mode 100644 index 3cfdfdf1..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt +++ /dev/null @@ -1,100 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import com.google.common.cache.CacheBuilder -import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.etherjar.domain.BlockHash -import reactor.core.Disposable -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.publisher.Sinks -import reactor.core.scheduler.Scheduler -import java.time.Duration - -class EnrichedMergedHead constructor( - private val sources: Iterable, - private val referenceHead: Head, - private val headScheduler: Scheduler, - private val api: Reader, -) : Head, Lifecycle { - - private val enrichedBlocks = CacheBuilder.newBuilder() - .maximumSize(10) - .build() - private val enrichedPromises = CacheBuilder.newBuilder() - .maximumSize(10) - .build>() - private var cacheSub: Disposable? = null - - private fun getEnrichBlockMono(id: BlockId): Mono { - val block = enrichedBlocks.getIfPresent(id) - return if (block != null) { - Mono.just(block) - } else { - enrichedPromises.get(id) { - Sinks.one() - }.asMono() - } - } - - override fun getFlux(): Flux { - return referenceHead.getFlux().concatMap { block -> - if (block.enriched) { - Mono.just(block) - } else { - Mono.firstWithValue( - getEnrichBlockMono(block.hash), - Mono.just(block) - .delayElement(Duration.ofSeconds(1)) - .flatMap { - EthereumBlockEnricher.enrich(BlockHash(block.hash.value), api, headScheduler) - }, - ) - } - } - } - - override fun onBeforeBlock(handler: Runnable) {} - - override fun getCurrentHeight(): Long? { - return referenceHead.getCurrentHeight() - } - - override fun getCurrentSlotHeight(): Long? { - return referenceHead.getCurrentSlotHeight() - } - - override fun isRunning(): Boolean { - return cacheSub != null - } - - override fun start() { - cacheSub?.dispose() - sources.forEach { head -> - if (head is Lifecycle && !head.isRunning()) { - head.start() - } - } - if (referenceHead is Lifecycle && !referenceHead.isRunning()) { - referenceHead.start() - } - cacheSub = Flux.merge(sources.map { it.getFlux() }).subscribe { block -> - if (block.enriched) { - enrichedBlocks.put(block.hash, block) - enrichedPromises.get(block.hash) { Sinks.one() }.tryEmitValue(block) - } - } - } - - override fun stop() { - cacheSub?.dispose() - cacheSub = null - } - - override fun onSyncingNode(isSyncing: Boolean) {} - - override fun headLiveness(): Flux = Flux.empty() -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt deleted file mode 100644 index cbd45fea..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt +++ /dev/null @@ -1,29 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.Defaults -import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.etherjar.domain.BlockHash -import reactor.core.publisher.Mono -import reactor.core.scheduler.Scheduler -import reactor.retry.Repeat -import java.time.Duration - -class EthereumBlockEnricher { - companion object { - fun enrich(blockHash: BlockHash, api: Reader, scheduler: Scheduler): Mono { - return Mono.just(blockHash) - .flatMap { hash -> - api.read(hash) - .subscribeOn(scheduler) - .timeout(Defaults.timeoutInternal, Mono.empty()) - }.repeatWhenEmpty { n -> - Repeat.times(5) - .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) - .apply(n) - } - .timeout(Defaults.timeout, Mono.empty()) - .onErrorResume { Mono.empty() } - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt index 7a506650..5dc0a75b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -43,6 +43,7 @@ import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.domain.Wei +import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory import org.springframework.cloud.sleuth.Tracer @@ -133,6 +134,10 @@ open class EthereumCachingReader( ) } + open fun logsByHash(): Reader>> { + return directReader.logsByHashReader + } + open fun txByHashAsCont(): Reader> { return CompoundReader( CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)), diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index b7af956e..2d959b2e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -16,6 +16,7 @@ import io.emeraldpay.dshackle.reader.RpcReaderFactory import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -26,6 +27,7 @@ import io.emeraldpay.etherjar.domain.Wei import io.emeraldpay.etherjar.hex.HexQuantity import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError +import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory @@ -59,6 +61,7 @@ class EthereumDirectReader( val txReader: Reader> val balanceReader: Reader> val receiptReader: Reader> + val logsByHashReader: Reader>> init { blockReader = object : Reader> { @@ -120,6 +123,7 @@ class EthereumDirectReader( } } } + receiptReader = object : Reader> { override fun read(key: TransactionId): Mono> { val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex())) @@ -148,6 +152,33 @@ class EthereumDirectReader( } } } + + logsByHashReader = object : Reader>> { + override fun read(key: BlockId): Mono>> { + val request = JsonRpcRequest( + "eth_getLogs", + listOf( + mapOf( + "blockHash" to key.toHexWithPrefix(), + ), + ), + ) + return EthereumCallSelector(caches).blockByHash(key.toHexWithPrefix()) + .defaultIfEmpty(Selector.empty).flatMap { matcher -> + readWithQuorum(request, matcher) + .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Logs not read $key"))) + .flatMap { + val logs = objectMapper.readValue(it.data, Array::class.java)?.toList() + if (logs == null) { + log.debug("Empty logs for block $key") + Mono.empty() + } else { + Mono.just(Result(logs, it.upstreamId)) + } + } + } + } + } } @Suppress("UNCHECKED_CAST") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt index 20b6ba14..3ce019f5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -17,7 +17,6 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector @@ -51,7 +50,7 @@ class ConnectBlockUpdates( fun connect() = connect(Selector.empty) override fun connect(matcher: Selector.Matcher): Flux { return connected.computeIfAbsent(matcher.describeInternal()) { key -> - extract(upstream.getEnrichedHead(matcher)) + extract(upstream.getHead(matcher)) .publishOn(scheduler) .publish() .refCount(1, Duration.ofSeconds(60)) @@ -104,35 +103,31 @@ class ConnectBlockUpdates( * Produce updates for transactions when a block is replaces with a different one on the same height. */ fun whenReplaced(prev: BlockContainer, source: String): Flux { - return Flux.fromIterable(prev.transactions).map { + return Flux.just( Update( prev.hash, prev.height, UpdateType.DROP, - it, source, - ) - } + ), + ) } fun extractUpdates(block: BlockContainer): Flux { - return Flux.fromIterable(block.transactions) - .map { - Update( - block.hash, - block.height, - UpdateType.NEW, - it, - block.upstreamId, - ) - } + return Flux.just( + Update( + block.hash, + block.height, + UpdateType.NEW, + block.upstreamId, + ), + ) } data class Update( val blockHash: BlockId, val blockNumber: Long, val type: UpdateType, - val transactionId: TxId, val upstreamId: String, ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index 5fa72d26..9388150b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -17,16 +17,14 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import com.google.common.cache.CacheBuilder import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage import io.emeraldpay.etherjar.hex.HexData -import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson +import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -34,7 +32,7 @@ import reactor.kotlin.core.publisher.switchIfEmpty import java.util.concurrent.TimeUnit class ProduceLogs( - private val receipts: Reader>, + private val logs: Reader>>, private val chain: Chain, ) { @@ -43,17 +41,13 @@ class ProduceLogs( } constructor(upstream: Multistream) : - this((upstream.getCachingReader() as EthereumCachingReader).receipts(), (upstream as Multistream).chain) - - private val objectMapper = Global.objectMapper + this((upstream.getCachingReader() as EthereumCachingReader).logsByHash(), (upstream as Multistream).chain) // need to keep history of recent messages in case they get removed. cannot rely on // any other cache or upstream because if when it gets removed it's unavailable in any other source private val oldMessages = CacheBuilder.newBuilder() - // in general a block with its events can be replaced in ~90 seconds, but in case of a big network disturbance - // it can be much longer. here we keep events up to an hour - .expireAfterWrite(60, TimeUnit.MINUTES) - .maximumSize(90000) + .expireAfterWrite(10, TimeUnit.MINUTES) + .maximumSize(30) .build>() fun produce(block: Flux): Flux { @@ -67,10 +61,10 @@ class ProduceLogs( } fun produceRemoved(update: ConnectBlockUpdates.Update): Flux { - val old = oldMessages.getIfPresent(LogReference(update.blockHash, update.transactionId)) + val old = oldMessages.getIfPresent(LogReference(update.blockHash)) if (old == null) { log.warn( - "No old message to produce removal messages for tx ${update.transactionId} " + + "No old message to produce removal messages " + "at block ${update.blockHash} for chain ${chain.chainName}", ) return Flux.empty() @@ -80,48 +74,33 @@ class ProduceLogs( } fun produceAdded(update: ConnectBlockUpdates.Update): Flux { - return receipts.read(update.transactionId) - .switchIfEmpty { - log.warn("Cannot find receipt for tx ${update.transactionId} for chain ${chain.chainName}") - Mono.empty() - } - .map { it.data } - .flatMapMany { jsonBytes -> - // receipt could be a null, like when the original block was replaced, etc. - // so just skip it as Flux.empty - val receipt = objectMapper.readValue(jsonBytes, TransactionReceiptJson::class.java) - ?: return@flatMapMany Flux.empty() - try { - val messages = receipt.logs - .map { txlog -> - LogMessage( - txlog.address, - txlog.blockHash, - txlog.blockNumber, - txlog.data ?: HexData.empty(), - txlog.logIndex, - txlog.topics, - txlog.transactionHash, - txlog.transactionIndex, - false, - update.upstreamId, + return logs.read(update.blockHash).switchIfEmpty { + log.warn("Cannot find receipt for block ${update.blockHash} for chain ${chain.chainName}") + Mono.empty() + }.map { + it.data + }.flatMapMany { + val messages = it.map { log -> + LogMessage( + log.address, + log.blockHash, + log.blockNumber, + log.data ?: HexData.empty(), + log.logIndex, + log.topics, + log.transactionHash, + log.transactionIndex, + false, + update.upstreamId, - ) - } - oldMessages.put(LogReference(update.blockHash, update.transactionId), messages) - Flux.fromIterable(messages) - } catch (t: Throwable) { - log.warn( - "Invalid Receipt ${update.transactionId} for chain ${chain.chainName}. " + - "${t.javaClass}: ${t.message}", - ) - Flux.empty() - } + ) } + oldMessages.put(LogReference(update.blockHash), messages) + Flux.fromIterable(messages) + } } private data class LogReference( val block: BlockId, - val tx: TxId, ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt index cad026be..8f97e601 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericMultistream.kt @@ -21,9 +21,7 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.JsonRpcReader -import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.CachingReader import io.emeraldpay.dshackle.upstream.DistanceExtractor import io.emeraldpay.dshackle.upstream.DynamicMergedHead @@ -39,11 +37,8 @@ import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Selector.Matcher import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallSelector -import io.emeraldpay.dshackle.upstream.ethereum.EnrichedMergedHead -import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream -import io.emeraldpay.etherjar.domain.BlockHash import org.springframework.util.ConcurrentReferenceHashMap import org.springframework.util.ConcurrentReferenceHashMap.ReferenceType.WEAK import reactor.core.publisher.Flux @@ -157,27 +152,6 @@ open class GenericMultistream( return head } - override fun getEnrichedHead(mather: Selector.Matcher): Head = - filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ -> - upstreams.filter { mather.matches(it) } - .apply { - log.debug("Found $size upstreams matching [${mather.describeInternal()}]") - }.let { - val selected = it.map { source -> source.getHead() } - EnrichedMergedHead( - selected, - getHead(), - headScheduler, - object : - Reader { - override fun read(key: BlockHash): Mono { - return (cachingReader as EthereumCachingReader).blocksByHashAsCont().read(key).map { res -> res.data } - } - }, - ) - } - } - override fun getLabels(): Collection { return upstreams.flatMap { it.getLabels() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy deleted file mode 100644 index bbeffef3..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHeadSpec.groovy +++ /dev/null @@ -1,185 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.test.ApiReaderMock -import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.AbstractHead -import io.emeraldpay.dshackle.upstream.BlockValidator -import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.etherjar.domain.BlockHash -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.publisher.Sinks -import reactor.core.scheduler.Schedulers -import reactor.test.StepVerifier -import spock.lang.Specification -import java.time.Duration - -class EnrichedMergedHeadSpec extends Specification { - - def "ensures that heads are running on start"() { - setup: - def head1 = Mock(TestHead) { - _ * isRunning() >> false - _ * getFlux() >> Flux.empty() - } - def head2 = Mock(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.empty() - } - def head3 = Mock(TestHead) { - _ * isRunning() >> false - _ * getFlux() >> Flux.empty() - } - - def api = new ApiReaderMock() - when: - def merged = new EnrichedMergedHead([head1, head2], head3, Schedulers.boundedElastic(), new BlockReader(api)) - merged.start() - - then: - 1 * head3.start() - 1 * head1.start() - merged.isRunning() - } - - def "if enriched block comes from reference head we pass it along instantly"() { - setup: - def block = TestingCommons.enrichedBlockForEthereum(100) - def api = new ApiReaderMock() - def head = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.just(block) - } - when: - def merge = new EnrichedMergedHead([], head, Schedulers.boundedElastic(), new BlockReader(api)) - - then: - StepVerifier.create(merge.getFlux()) - .then { merge.start() } - .expectNext(block) - .thenCancel() - .verify(Duration.ofMillis(100)) - block.enriched - } - - def "enriched block arrived in sources before reference block"() { - setup: - def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) - def block = TestingCommons.blockForEthereum(100) - Sinks.Many refSink = Sinks.many().multicast().directBestEffort() - def headRef = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> refSink.asFlux() - } - def headSource = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.just(enrichedBlock) - } - when: - def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(new ApiReaderMock())) - then: - StepVerifier.create(merge.getFlux()) - .then { merge.start() } - .expectNoEvent(Duration.ofMillis(100)) - .then { refSink.tryEmitNext(block) } - .expectNext(enrichedBlock) - .thenCancel() - .verify(Duration.ofMillis(200)) - } - - def "enriched block arrived in source after reference block, but before deadline"() { - setup: - def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) - def block = TestingCommons.blockForEthereum(100) - Sinks.Many sourceSink = Sinks.many().multicast().directBestEffort() - def headRef = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.just(block) - } - def headSource = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> sourceSink.asFlux() - } - when: - def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(new ApiReaderMock())) - then: - StepVerifier.create(merge.getFlux()) - .then { merge.start() } - .expectNoEvent(Duration.ofMillis(600)) - .then { sourceSink.tryEmitNext(enrichedBlock) } - .expectNext(enrichedBlock) - .thenCancel() - .verify(Duration.ofSeconds(1)) - } - - def "enriched blocks does not arrive before deadline"() { - setup: - def enrichedBlock = TestingCommons.enrichedBlockForEthereum(100) - def block = TestingCommons.blockForEthereum(100) - def headRef = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.just(block) - } - def headSource = Stub(TestHead) { - _ * isRunning() >> true - _ * getFlux() >> Flux.just(block) - } - def api = new ApiReaderMock().tap { - answer("eth_getBlockByHash", [block.hash.toHexWithPrefix(), false], enrichedBlock.toBlock()) - } - when: - def merge = new EnrichedMergedHead([headSource], headRef, Schedulers.boundedElastic(), new BlockReader(api)) - then: - StepVerifier.create(merge.getFlux()) - .then { merge.start() } - .expectNoEvent(Duration.ofSeconds(1)) - .expectNext(enrichedBlock) - .thenCancel() - .verify(Duration.ofMillis(1200)) - } - - class BlockReader implements Reader { - private ApiReaderMock mockApi - BlockReader(ApiReaderMock api) { - mockApi = api - } - - Mono read(BlockHash hash) { - return mockApi.read(new JsonRpcRequest("eth_getBlockByHash", [hash.toHex(), false])) - .map { - def t = it - def a = 1 - return it - } - .flatMap(JsonRpcResponse::requireResult) - .map { BlockContainer.fromEthereumJson(it, "test") } - } - } - - class TestHead extends AbstractHead implements Lifecycle { - - TestHead() { - super(new MostWorkForkChoice(), Schedulers.boundedElastic(), new BlockValidator.AlwaysValid(), 100_000) - } - - @Override - void start() { - - } - - @Override - void stop() { - - } - - @Override - boolean isRunning() { - return false - } - } -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index 43d541ce..a7a6c5ec 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache +import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.DefaultContainer import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.RpcReaderFactory @@ -17,6 +18,7 @@ import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.domain.Wei import io.emeraldpay.etherjar.rpc.json.TransactionJson +import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import org.apache.commons.collections4.Factory import reactor.core.publisher.Mono @@ -128,6 +130,38 @@ class EthereumDirectReaderSpec extends Specification { .verify(Duration.ofSeconds(1)) } + def "Reads logs by block hash"() { + setup: + def json = new TransactionLogJson().tap { + address = Address.from(address1) + blockHash = BlockHash.from(hash1) + } + def calls = Mock(Factory) { + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET, false) + } + EthereumDirectReader reader = new EthereumDirectReader( + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + ) + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { + 1 * read(new JsonRpcRequest("eth_getLogs", [Map.of("blockHash", hash1)])) >> Mono.just( + new RpcReader.Result( + Global.objectMapper.writeValueAsBytes([json]), null, 1, resolver + ) + ) + } + } + when: + def act = reader.logsByHashReader.read(BlockId.from(hash1)) + then: + StepVerifier.create(act) + .expectNextMatches { logs -> + logs.data[0].blockHash == BlockHash.from(hash1) + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + def "Reads tx"() { setup: def json = new TransactionJson().tap { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy index ef1ce1b9..3069a0ac 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy @@ -45,27 +45,17 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) when: def act = connectBlockUpdates.extractUpdates(block) .collectList().block(Duration.ofSeconds(3)) then: - act.size() == 2 + act.size() == 1 with(act[0]) { it.blockNumber == 13412871 it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") - it.type == ConnectBlockUpdates.UpdateType.NEW - } - with(act[1]) { - it.blockNumber == 13412871 - it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") it.type == ConnectBlockUpdates.UpdateType.NEW } } @@ -79,27 +69,17 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) when: def act = connectBlockUpdates.whenReplaced(block, "ConnectBlockUpdatesSpec") .collectList().block(Duration.ofSeconds(3)) then: - act.size() == 2 + act.size() == 1 with(act[0]) { it.blockNumber == 13412871 it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") - it.type == ConnectBlockUpdates.UpdateType.DROP - } - with(act[1]) { - it.blockNumber == 13412871 - it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") it.type == ConnectBlockUpdates.UpdateType.DROP } } @@ -113,10 +93,7 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) def block2 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") @@ -124,10 +101,7 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) def block3 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0xdb1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7") @@ -135,10 +109,7 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE parentHash = parent timestamp = Instant.now() - transactions = [ - new TransactionRefJson(TransactionId.from("0x9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af6c88df9d65ccc")), - new TransactionRefJson(TransactionId.from("0xdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe25c241a64e7ce536f")) - ] + transactions = [] }) when: @@ -177,10 +148,7 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) def block2 = BlockContainer.from(new BlockJson().tap { hash = BlockHash.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") @@ -188,10 +156,7 @@ class ConnectBlockUpdatesSpec extends Specification { totalDifficulty = BigInteger.ONE timestamp = Instant.now() parentHash = parent - transactions = [ - new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), - new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) - ] + transactions = [] }) when: @@ -200,29 +165,15 @@ class ConnectBlockUpdatesSpec extends Specification { .collectList().block(Duration.ofSeconds(1)) then: - act.size() == 4 + act.size() == 2 with(act[0]) { it.blockNumber == 13412871 it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") it.type == ConnectBlockUpdates.UpdateType.DROP } with(act[1]) { - it.blockNumber == 13412871 - it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") - it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") - it.type == ConnectBlockUpdates.UpdateType.DROP - } - with(act[2]) { it.blockNumber == 13412871 it.blockHash == BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") - it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") - it.type == ConnectBlockUpdates.UpdateType.NEW - } - with(act[3]) { - it.blockNumber == 13412871 - it.blockHash == BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") - it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") it.type == ConnectBlockUpdates.UpdateType.NEW } } @@ -233,7 +184,7 @@ class ConnectBlockUpdatesSpec extends Specification { 1 * getFlux() >> Flux.never() } def up = Mock(GenericMultistream) { - 1 * getEnrichedHead(Selector.empty) >> head + 1 * getHead(Selector.empty) >> head } def connectBlockUpdates = new ConnectBlockUpdates(up, Schedulers.boundedElastic()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy index fbbec844..839f25ba 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy @@ -20,6 +20,11 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.hex.HexData +import io.emeraldpay.etherjar.rpc.json.TransactionLogJson import reactor.core.publisher.Mono import spock.lang.Specification @@ -27,28 +32,28 @@ import java.time.Duration class ProduceLogsSpec extends Specification { + def logs = [new TransactionLogJson().tap { + blockHash = BlockHash.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") + address = Address.from("0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d") + blockNumber = 1 + logIndex = 1 + data = HexData.empty() + transactionIndex = 1 + topics = [] + transactionHash = TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec") + }] + def "Produce added as nothing with no logs"() { setup: - String receipt = '{\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "blockNumber": "0x1",\n' + - ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "logs": [\n' + - ' ],\n' + - ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + - ' "transactionIndex": "0x0"\n' + - ' }' - - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) + def logs = Mock(Reader) { + 1 * it.read(BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da")) >> + Mono.just(new EthereumDirectReader.Result<>([], null)) } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) + def producer = new ProduceLogs(logs, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), - 13412871, + 1, ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"), "upstream" ) when: @@ -59,67 +64,18 @@ class ProduceLogsSpec extends Specification { act.size() == 0 } - def "Produce added as nothing with null receipt"() { + def "Produce added with logs"() { setup: - String receipt = 'null' - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) + def logs = Mock(Reader) { + 1 * it.read(BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da")) >> + Mono.just(new EthereumDirectReader.Result<>(logs, null)) } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) + def producer = new ProduceLogs(logs, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), - 13412871, + 1, ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"), - "upstream" - ) - when: - def act = producer.produceAdded(update) - .collectList().block(Duration.ofSeconds(1)) - - then: - act.size() == 0 - } - - def "Produce added with single log"() { - setup: - String receipt = '{\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "blockNumber": "0x1",\n' + - ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "logs": [\n' + - ' {\n' + - ' "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "0x00000000000000000000000099897cb0e667d354b920fc38a40a5100b2a01566"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000000000000007505d91f0",\n' + - ' "blockNumber": "0xc7f3b4",\n' + - ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + - ' "transactionIndex": "0x0",\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "logIndex": "0x0",\n' + - ' "removed": false\n' + - ' }' + - ' ],\n' + - ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + - ' "transactionIndex": "0x0"\n' + - ' }' - - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) - } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) - def update = new ConnectBlockUpdates.Update( - BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), - 13412871, - ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"), "upstream" ) when: @@ -128,244 +84,27 @@ class ProduceLogsSpec extends Specification { then: act.size() == 1 - with(act[0]) { - it.transactionHash.toHex() == "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af" - } - } - - def "Produce added with no data"() { - setup: - String receipt = '{\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "blockNumber": "0x1",\n' + - ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "logs": [\n' + - ' {\n' + - ' "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"\n' + - ' ],\n' + - ' "blockNumber": "0xc7f3b4",\n' + - ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + - ' "transactionIndex": "0x0",\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "logIndex": "0x0",\n' + - ' "removed": false\n' + - ' }' + - ' ],\n' + - ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + - ' "transactionIndex": "0x0"\n' + - ' }' - - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) - } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) - def update = new ConnectBlockUpdates.Update( - BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), - 13412871, - ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"), - "upstream" - ) - when: - def act = producer.produceAdded(update) - .collectList().block(Duration.ofSeconds(1)) - - then: - act.size() == 1 - with(act[0]) { - it.transactionHash.toHex() == "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af" - // Geth actually renders it as null, so this check may be wrong - it.data != null && it.data.size == 0 - } - } - - def "Produce added with multiple logs"() { - setup: - String receipt = '{\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "blockNumber": "0x1",\n' + - ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "logs": [\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x00000000000000000000000075b8c48bdb04d426aed57b36bb835ad2dc321c30"\n' + - ' ],\n' + - ' "data": "0x0000000000000000000000000000000000000000000000013e7ec767db370000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb4",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000000023636b7d513f0000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb5",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000011316d590258fba0000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb6",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + - ' ],\n' + - ' "data": "0x0000000000000000000000000000000000000000000014188a101e403a500000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb7",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x1b46b72c5280f30fbe8a958b4f3c348fd0fd2e55",\n' + - ' "topics": [\n' + - ' "0x90890809c654f11d6e72a28fa60149770a0d11ec6c92319d6ceb2bb0a4ea1a15",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000011478b7c30abc300000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb8",\n' + - ' "removed": false\n' + - ' }' + - ' ],\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57"\n' + - ' }' - - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) - } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) - def update = new ConnectBlockUpdates.Update( - BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), - 13412871, - ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), - "upstream" - ) - when: - def act = producer.produceAdded(update) - .collectList().block(Duration.ofSeconds(1)) - - then: - act.size() == 5 - act*.transactionHash.every { it.toHex() == "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec" } - act*.logIndex == [180, 181, 182, 183, 184] + act*.address.every { it.toHex() == "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d" } act*.removed.every { !it } } def "Produce removed"() { setup: - String receipt = '{\n' + - ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + - ' "blockNumber": "0x1",\n' + - ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + - ' "logs": [\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x00000000000000000000000075b8c48bdb04d426aed57b36bb835ad2dc321c30"\n' + - ' ],\n' + - ' "data": "0x0000000000000000000000000000000000000000000000013e7ec767db370000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb4",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000000023636b7d513f0000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb5",\n' + - ' "removed": false\n' + - ' },\n' + - ' {\n' + - ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + - ' "topics": [\n' + - ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + - ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + - ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + - ' ],\n' + - ' "data": "0x00000000000000000000000000000000000000000000011316d590258fba0000",\n' + - ' "blockNumber": "0xccc493",\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57",\n' + - ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + - ' "logIndex": "0xb6",\n' + - ' "removed": false\n' + - ' }\n' + - ' ],\n' + - ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + - ' "transactionIndex": "0x57"\n' + - ' }' - - def receipts = Mock(Reader) { - 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> - Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) + def logs = Mock(Reader) { + 1 * it.read(BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da")) >> + Mono.just(new EthereumDirectReader.Result<>(logs, null)) } - def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) + def producer = new ProduceLogs(logs, Chain.ETHEREUM__MAINNET) def update1 = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, ConnectBlockUpdates.UpdateType.NEW, - TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), "upstream" ) def update2 = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, ConnectBlockUpdates.UpdateType.DROP, - TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), "upstream" ) when: @@ -376,9 +115,9 @@ class ProduceLogsSpec extends Specification { .collectList().block(Duration.ofSeconds(1)) then: - act.size() == 3 + act.size() == 1 act*.transactionHash.every { it.toHex() == "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec" } - act*.logIndex == [180, 181, 182] + act*.logIndex == [1] act*.removed.every { it } } }