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

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<ByteArray> = AtomicReference()

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

src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt

@@ -115,6 +115,10 @@ class Selector {
             return matchers.all { it.matches(up) }
         }
 
+        fun getMatchers(): Collection<Matcher> {
+            return Collections.unmodifiableCollection(matchers)
+        }
+
         fun <T : Matcher> getMatcher(type: Class<T>): T? {
             return matchers.find { type.isAssignableFrom(it.javaClass) } as T?
         }

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))

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<EthereumUpstream> upstreams, @NotNull Caches caches) {
             super(chain, upstreams, caches)
         }
 
+        EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumUpstream> 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()
+        }
     }
 
 }

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<JsonRpcRequest, JsonRpcResponse> api) {
         return new EthereumUpstreamMock(id, Chain.ETHEREUM, api)
     }