From b255bed3a1336d246c3abdd1f72ddc7ac2afce87 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 23 Feb 2020 23:07:31 -0500 Subject: [PATCH] solution: use caching for WS head --- .../dshackle/upstream/ChainUpstreams.kt | 2 +- .../dshackle/upstream/ethereum/EthereumApi.kt | 6 +- .../upstream/ethereum/EthereumHeadMerge.kt | 23 +++--- .../upstream/ethereum/EthereumRpcHead.kt | 10 +++ .../upstream/ethereum/EthereumUpstream.kt | 5 +- .../dshackle/upstream/ethereum/EthereumWs.kt | 47 +++++++----- .../upstream/ethereum/EthereumWsHead.kt | 8 +- .../dshackle/test/EthereumApiMock.groovy | 38 ++++++---- .../upstream/ethereum/EthereumWsSpec.groovy | 73 +++++++++++++++++++ 9 files changed, 165 insertions(+), 47 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 3e425959..a9d67a62 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -89,7 +89,7 @@ open class ChainUpstreams ( upstream.setLag(0) upstream.getHead() } else { - val newHead = EthereumHeadMerge(upstreams.map { it.getHead().getFlux() }).apply { + val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply { this.start() } val lagObserver = HeadLagObserver(newHead, upstreams).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt index fcba2a7f..3d2c8f04 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt @@ -35,6 +35,10 @@ abstract class EthereumApi( abstract fun execute(id: Int, method: String, params: List): Mono + fun execute(rpcCall: RpcCall): Mono { + return execute(0, rpcCall.method, rpcCall.params as List) + } + fun executeAndConvert(rpcCall: RpcCall): Mono { val convertToJS = java.util.function.Function> { resp -> val inputStream: InputStream = resp.inputStream() @@ -42,7 +46,7 @@ abstract class EthereumApi( if (jsonValue == null) Mono.empty() else Mono.just(jsonValue) } - return execute(0, rpcCall.method, rpcCall.params as List) + return execute(rpcCall) .flatMap(convertToJS) .map(rpcCall.converter::apply) .doOnError { err -> log.debug("Failed to read from upstream", err) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt index eeb84a84..d2484732 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt @@ -15,20 +15,15 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson -import org.reactivestreams.Publisher -import org.slf4j.LoggerFactory +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import java.util.concurrent.atomic.AtomicReference class EthereumHeadMerge( - private val fluxes: Iterable>> -): DefaultEthereumHead(), Lifecycle { + private val sources: Iterable +): DefaultEthereumHead(), Lifecycle, CachesEnabled { private var subscription: Disposable? = null @@ -37,11 +32,19 @@ class EthereumHeadMerge( } override fun start() { - subscription = super.follow(Flux.merge(fluxes)) + subscription = super.follow(Flux.merge(sources.map { it.getFlux() })) } override fun stop() { subscription?.dispose() } + override fun setCaches(caches: Caches) { + sources.forEach { + if (it is CachesEnabled) { + it.setCaches(caches) + } + } + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index 7b846e35..1e092f16 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -16,9 +16,17 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.reader.EmptyReader +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.CachingEthereumApi +import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.ReactorBatch +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import org.springframework.scheduling.concurrent.CustomizableThreadFactory @@ -52,6 +60,8 @@ class EthereumRpcHead( .timeout(Defaults.timeout, Mono.error(Exception("Block number not received"))) } .flatMap { + //fetching by Block Height here, critical to use same upstream, + //different upstreams may have different blocks on the same height api.rpcClient .execute(Commands.eth().getBlock(it)) .subscribeOn(scheduler) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index f2929b6a..bcfadbac 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -52,6 +52,9 @@ open class EthereumUpstream( override fun setCaches(caches: Caches) { api.caches = caches; + if (head is CachesEnabled) { + head.setCaches(caches) + } } override fun getId(): String { @@ -92,7 +95,7 @@ open class EthereumUpstream( val rpc = EthereumRpcHead(api, Duration.ofSeconds(30)).apply { this.start() } - EthereumHeadMerge(listOf(rpc.getFlux(), ws.getFlux())).apply { + EthereumHeadMerge(listOf(rpc, ws)).apply { this.start() } } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt index d0bdaaf1..fffb3eae 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt @@ -16,8 +16,12 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.infinitape.etherjar.domain.TransactionId +import io.emeraldpay.dshackle.reader.EmptyReader +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson @@ -34,7 +38,7 @@ class EthereumWs( private val uri: URI, private val origin: URI, private val api: EthereumApi -) { +): CachesEnabled { private val log = LoggerFactory.getLogger(EthereumWs::class.java) private val topic = TopicProcessor @@ -43,6 +47,8 @@ class EthereumWs( .build() var basicAuth: UpstreamsConfig.BasicAuth? = null + private var blockCache: Reader> = EmptyReader() + fun connect() { log.info("Connecting to WebSocket: $uri") val clientBuilder = WebsocketClient.newBuilder() @@ -54,24 +60,27 @@ class EthereumWs( val client = clientBuilder.build() try { client.connect() + client.onNewBlock(this::onNewBlock) } catch (e: Exception) { log.error("Failed to connect to websocket at $uri. Error: ${e.message}") - return } - client.onNewBlock { - if (it.totalDifficulty == null || it.transactions == null) { - Mono.just(it.hash).flatMap { hash -> - api.executeAndConvert(Commands.eth().getBlock(hash)) - }.repeatWhenEmpty { n -> - Repeat.times(10) - .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250)) - .apply(n) - } - .timeout(Defaults.timeout, Mono.empty()) - .subscribe(topic::onNext) - } else { - topic.onNext(it) - } + } + + fun onNewBlock(block: BlockJson) { + if (block.totalDifficulty == null || block.transactions == null) { + Mono.just(block.hash).flatMap { hash -> + // first check in cache, if empty then check api + blockCache.read(hash) + .switchIfEmpty(api.executeAndConvert(Commands.eth().getBlock(hash))) + }.repeatWhenEmpty { n -> + Repeat.times(10) + .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250)) + .apply(n) + } + .timeout(Defaults.timeout, Mono.empty()) + .subscribe(topic::onNext) + } else { + topic.onNext(block) } } @@ -79,4 +88,8 @@ class EthereumWs( return Flux.from(this.topic) .onBackpressureLatest() } + + override fun setCaches(caches: Caches) { + blockCache = caches.getBlocksByHash() + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 29f6c0d9..7a5394b9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -15,13 +15,15 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable class EthereumWsHead( private val ws: EthereumWs -): DefaultEthereumHead(), Lifecycle { +): DefaultEthereumHead(), Lifecycle, CachesEnabled { private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) @@ -40,4 +42,8 @@ class EthereumWsHead( subscription = null } + override fun setCaches(caches: Caches) { + ws.setCaches(caches) + } + } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index e5d9cac3..f5f62883 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -32,6 +32,8 @@ import org.slf4j.Logger import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import java.util.concurrent.Callable + class EthereumApiMock extends DirectEthereumApi { private static final Logger log = LoggerFactory.getLogger(this) @@ -55,26 +57,30 @@ class EthereumApiMock extends DirectEthereumApi { @Override Mono execute(int id, @NotNull String method, @NotNull List params) { - def predefined = predefined.find { it.isSame(id, method, params) } - ResponseJson json = new ResponseJson(id: id) - if (predefined != null) { - if (predefined.exception != null) { + Callable call = { + def predefined = predefined.find { it.isSame(id, method, params) } + ResponseJson json = new ResponseJson(id: id) + if (predefined != null) { + if (predefined.exception != null) { + predefined.onCalled() + predefined.print() + throw predefined.exception + } + if (predefined.result instanceof RpcResponseError) { + json.error = predefined.result + } else { + json.result = predefined.result + } predefined.onCalled() predefined.print() - return Mono.error(predefined.exception) - } - if (predefined.result instanceof RpcResponseError) { - json.error = predefined.result } else { - json.result = predefined.result + log.error("Method ${method} with ${params} is not mocked") + json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked") } - } else { - log.error("Method ${method} with ${params} is not mocked") - json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked") - } - predefined.onCalled() - predefined.print() - return Mono.just(objectMapper.writeValueAsBytes(json)) + byte[] result = objectMapper.writeValueAsBytes(json) + return result + } as Callable + return Mono.fromCallable(call) } def nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsSpec.groovy new file mode 100644 index 00000000..af9a8709 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsSpec.groovy @@ -0,0 +1,73 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.HeightCache +import io.emeraldpay.dshackle.test.TestingCommons +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import io.infinitape.etherjar.rpc.ws.WebsocketClient +import reactor.core.publisher.Mono +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration +import java.time.Instant +import java.time.temporal.ChronoUnit + +class EthereumWsSpec extends Specification { + + def "Uses cache to fetch block"() { + setup: + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) + def apiMock = TestingCommons.api(rpcClient) + def ws = new EthereumWs(new URI("http://localhost"), new URI("http://localhost"), apiMock) + def blocksCache = Mock(BlocksMemCache) + def caches = Caches.newBuilder().setBlockByHash(blocksCache).build() + ws.setCaches(caches) + + def block = new BlockJson() + block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") + block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + + when: + ws.onNewBlock(block) + + then: + 1 * blocksCache.read(BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")) >> Mono.just(block) + StepVerifier.create(ws.flux.take(1)) + .expectNext(block) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Fetch block if cache is empty"() { + setup: + ReactorRpcClient rpcClient = Stub(ReactorRpcClient) + def apiMock = TestingCommons.api(rpcClient) + def ws = new EthereumWs(new URI("http://localhost"), new URI("http://localhost"), apiMock) + def blocksCache = Mock(BlocksMemCache) + def caches = Caches.newBuilder().setBlockByHash(blocksCache).build() + ws.setCaches(caches) + + def block = new BlockJson() + block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") + block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + block.transactions = [] + block.uncles = [] + + apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) + + when: + ws.onNewBlock(block) + + then: + 1 * blocksCache.read(_) >> Mono.empty() + StepVerifier.create(ws.flux.take(1)) + .expectNext(block) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } +}