From 7cf86129f0654c0ecf048ca6e71d85e8c19213fd Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 4 Aug 2021 17:00:30 -0400 Subject: [PATCH] problem: routes calls to upstreams that quorum wouldnt accept anyway (not-lagging quorum to lagging upstreams) solution: select only possible acceptable upstreams before the call --- .../dshackle/quorum/NotLaggingQuorum.kt | 5 ++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 12 ++++- .../emeraldpay/dshackle/upstream/Selector.kt | 4 ++ .../dshackle/rpc/NativeCallSpec.groovy | 46 +++++++++++++++++++ .../test/MultistreamHolderMock.groovy | 28 +++++++++++ .../dshackle/test/TestingCommons.groovy | 5 ++ 6 files changed, 98 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index aa2cb4f5..80e9050a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -23,6 +23,11 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.infinitape.etherjar.rpc.RpcException import java.util.concurrent.atomic.AtomicReference +/** + * Accepts a response only from a "synced" upstreams, where "maxLag" specifies how many blocks it may be behind to be considered as "synced" + * + * NOTE: NativeCall checks the quorums and applies a HeightSelector if NotLaggingQuorum is enabled for a call + */ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { private val result: AtomicReference = AtomicReference() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 5e8f76d7..ead150c5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream @@ -151,12 +152,19 @@ open class NativeCall( .withMatcher(csm) .forMethod(method) .forLabels(Selector.convertToMatcher(request.selector)) - .build() val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() callQuorum.init(upstream.getHead()) - CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params)) + // for NotLaggingQuorum it makes sense to select compatible upstreams before the call + if (callQuorum is NotLaggingQuorum) { + val lag = callQuorum.maxLag + val minHeight = ((upstream.getHead().getCurrentHeight() ?: 0) - lag).coerceAtLeast(0) + val heightMatcher = Selector.HeightMatcher(minHeight) + matcher.withMatcher(heightMatcher) + } + + CallContext(it.id, upstream, matcher.build(), callQuorum, RawCallDetails(method, params)) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt index 15945945..13780476 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt @@ -115,6 +115,10 @@ class Selector { return matchers.all { it.matches(up) } } + fun getMatchers(): Collection { + return Collections.unmodifiableCollection(matchers) + } + fun getMatcher(type: Class): T? { return matchers.find { type.isAssignableFrom(it.javaClass) } as T? } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 23fe751b..72e4415f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -24,11 +24,16 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods +import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -290,6 +295,47 @@ class NativeCallSpec extends Specification { } } + def "Prepare call adds height selector for not-lagging quorum"() { + setup: + def methods = new ManagedCallMethods( + new DefaultEthereumMethods(Chain.ETHEREUM), + ["foo_bar"] as Set, [] as Set + ) + methods.setQuorum("foo_bar", "not_lagging") + def head = Mock(Head) { + 1 * it.getCurrentHeight() >> 101 + } + def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream()) + multistream.customMethods = methods + multistream.customHead = head + def multistreamHolder = Mock(MultistreamHolder) { + _ * it.observeChains() >> Flux.empty() + } + def nativeCall = new NativeCall(multistreamHolder) + + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.CHAIN_ETHEREUM) + .addItems( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(1) + .setMethod("foo_bar") + ) + .build() + when: + def act = nativeCall.prepareCall(req, multistream) + .collectList().block(Duration.ofSeconds(1)).first() + then: + act.matcher != null + act.matcher instanceof Selector.MultiMatcher + with((Selector.MultiMatcher) act.matcher) { + it.getMatchers().size() >= 1 + it.getMatcher(Selector.HeightMatcher) != null + with(it.getMatcher(Selector.HeightMatcher)) { + it.height == 101 + } + } + } + def "Parse empty params"() { setup: def nativeCall = new NativeCall(Stub(MultistreamHolder)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index c659303f..a98fb797 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -19,9 +19,11 @@ package io.emeraldpay.dshackle.test import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream +import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.MultistreamHolder @@ -100,11 +102,21 @@ class MultistreamHolderMock implements MultistreamHolder { static class EthereumMultistreamMock extends EthereumMultistream { EthereumReader customReader = null + CallMethods customMethods = null + Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { super(chain, upstreams, caches) } + EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { + this(chain, upstreams, Caches.default()) + } + + EthereumMultistreamMock(@NotNull Chain chain, @NotNull EthereumUpstream upstream) { + this(chain, [upstream]) + } + @Override EthereumReader getReader() { if (customReader != null) { @@ -112,6 +124,22 @@ class MultistreamHolderMock implements MultistreamHolder { } return super.getReader() } + + @Override + CallMethods getMethods() { + if (customMethods != null) { + return customMethods + } + return super.getMethods() + } + + @Override + Head getHead() { + if (customHead != null) { + return customHead + } + return super.getHead() + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index d9aaad25..79456618 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream @@ -49,6 +50,10 @@ class TestingCommons { return new EthereumApiMock() } + static EthereumUpstreamMock upstream() { + return new EthereumUpstreamMock(Chain.ETHEREUM, api()) + } + static EthereumUpstreamMock upstream(String id, Reader api) { return new EthereumUpstreamMock(id, Chain.ETHEREUM, api) }