Logs subscription via eth_getLogs (#356)

move retrieving logs subscription to eth_getLogs by hash
2023-12-04 17:09:32 +02:00
parent 9696f06732
commit 0a47da602a
14 changed files with 157 additions and 769 deletions
--- 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)
--- 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<UpstreamsConfig.Labels> {
        return sourceUpstreams.flatMap { it.getLabels() }
    }
--- 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<Selector.Matcher> {
+    fun blockByHash(blockHash: String): Mono<Selector.Matcher> {
        return try {
            caches.getLastHeightByHash()
                .read(BlockId.from(blockHash))
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EnrichedMergedHead.kt
@@ -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<Head>,
-    private val referenceHead: Head,
-    private val headScheduler: Scheduler,
-    private val api: Reader<BlockHash, BlockContainer>,
-) : Head, Lifecycle {
-
-    private val enrichedBlocks = CacheBuilder.newBuilder()
-        .maximumSize(10)
-        .build<BlockId, BlockContainer>()
-    private val enrichedPromises = CacheBuilder.newBuilder()
-        .maximumSize(10)
-        .build<BlockId, Sinks.One<BlockContainer>>()
-    private var cacheSub: Disposable? = null
-
-    private fun getEnrichBlockMono(id: BlockId): Mono<BlockContainer> {
-        val block = enrichedBlocks.getIfPresent(id)
-        return if (block != null) {
-            Mono.just(block)
-        } else {
-            enrichedPromises.get(id) {
-                Sinks.one()
-            }.asMono()
-        }
-    }
-
-    override fun getFlux(): Flux<BlockContainer> {
-        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<Boolean> = Flux.empty()
-}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockEnricher.kt
@@ -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<BlockHash, BlockContainer>, scheduler: Scheduler): Mono<BlockContainer> {
-            return Mono.just(blockHash)
-                .flatMap { hash ->
-                    api.read(hash)
-                        .subscribeOn(scheduler)
-                        .timeout(Defaults.timeoutInternal, Mono.empty())
-                }.repeatWhenEmpty { n ->
-                    Repeat.times<Any>(5)
-                        .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
-                        .apply(n)
-                }
-                .timeout(Defaults.timeout, Mono.empty())
-                .onErrorResume { Mono.empty() }
-        }
-    }
-}
--- 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<BlockId, Result<List<TransactionLogJson>>> {
+        return directReader.logsByHashReader
+    }
+
    open fun txByHashAsCont(): Reader<TxId, Result<TxContainer>> {
        return CompoundReader(
            CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)),
--- 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<TransactionId, Result<TxContainer>>
    val balanceReader: Reader<Address, Result<Wei>>
    val receiptReader: Reader<TransactionId, Result<ByteArray>>
+    val logsByHashReader: Reader<BlockId, Result<List<TransactionLogJson>>>

    init {
        blockReader = object : Reader<BlockHash, Result<BlockContainer>> {
@@ -120,6 +123,7 @@ class EthereumDirectReader(
                    }
            }
        }
+
        receiptReader = object : Reader<TransactionId, Result<ByteArray>> {
            override fun read(key: TransactionId): Mono<Result<ByteArray>> {
                val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex()))
@@ -148,6 +152,33 @@ class EthereumDirectReader(
                    }
            }
        }
+
+        logsByHashReader = object : Reader<BlockId, Result<List<TransactionLogJson>>> {
+            override fun read(key: BlockId): Mono<Result<List<TransactionLogJson>>> {
+                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<TransactionLogJson>::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")
--- 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<Update> {
        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<Update> {
-        return Flux.fromIterable(prev.transactions).map {
+        return Flux.just(
            Update(
                prev.hash,
                prev.height,
                UpdateType.DROP,
-                it,
                source,
-            )
-        }
+            ),
+        )
    }

    fun extractUpdates(block: BlockContainer): Flux<Update> {
-        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,
    )

--- 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<TxId, Result<ByteArray>>,
+    private val logs: Reader<BlockId, Result<List<TransactionLogJson>>>,
    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<LogReference, List<LogMessage>>()

    fun produce(block: Flux<ConnectBlockUpdates.Update>): Flux<LogMessage> {
@@ -67,10 +61,10 @@ class ProduceLogs(
    }

    fun produceRemoved(update: ConnectBlockUpdates.Update): Flux<LogMessage> {
-        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<LogMessage> {
-        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<LogMessage>()
-                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<LogMessage>()
-                }
+                )
            }
+            oldMessages.put(LogReference(update.blockHash), messages)
+            Flux.fromIterable(messages)
+        }
    }

    private data class LogReference(
        val block: BlockId,
-        val tx: TxId,
    )
 }
--- 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<BlockHash, BlockContainer> {
-                            override fun read(key: BlockHash): Mono<BlockContainer> {
-                                return (cachingReader as EthereumCachingReader).blocksByHashAsCont().read(key).map { res -> res.data }
-                            }
-                        },
-                    )
-                }
-        }
-
    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
        return upstreams.flatMap { it.getLabels() }
    }