solution: allow configuring a quorum for custom methods

docs/reference-configuration.adoc

@@ -514,6 +514,7 @@ configuration, and may be omitted for most of the situations.
   methods:
     enabled:
       - name: "parity_trace"
+        quorum: "not_empty"
     disabled:
       - name: "admin_shutdown"
   connection:
@@ -556,7 +557,18 @@ See link:09-quorum-and-selectors.adoc[Quorum and Selectors]
 
 | `methods`
 | no
-| Enable or disable additional JSON RPC methods that are provided by that particular upstream
+| Enable (`enabled`) or disable (`disabled`) additional JSON RPC methods that are provided by that particular upstream
+
+| `methods.enabled.name`, `methods.disabled.name`
+| yes
+| Name of the RPC method to enable/disable.
+
+| `methods.enabled.quorum`
+| no
+| Set quorum criteria to accept a response.
+`always` (default) - accept any response;
+`not_empty` - accept not _null_ value, otherwise retry another upstream;
+`not_lagging` - accept response only from a fully synced upstream.
 
 | `connection.ethereum`
 | yes
@@ -574,9 +586,9 @@ See link:09-quorum-and-selectors.adoc[Quorum and Selectors]
 | Option | Description
 
 | `rpc.url`
-a| HTTP URL to connect to. This is required for a connection. +
+a| HTTP URL to connect to.This is required for a connection. +
-   URL can be configured with Environment Variable placeholders `${ENV_VAR_NAME}`. +
+URL can be configured with Environment Variable placeholders `${ENV_VAR_NAME}`. +
-   Example: `https://kovan.infura.io/v3/${INFURA_USER}`
+Example: `https://kovan.infura.io/v3/${INFURA_USER}`
 
 | `rpc.basic-auth` + `rpc.basic-auth.username`, `rpc.basic-auth.password`
 a| HTTP Basic Auth configuration, if required by the remote server. +

src/main/kotlin/io/emeraldpay/dshackle/FileResolver.kt

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle
 
 import java.io.File
 
-class FileResolver(
+open class FileResolver(
         private val baseDir: File
 ) {
 

src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt

@@ -33,7 +33,7 @@ import kotlin.system.exitProcess
 
 
 @Repository
-class CachesFactory(
+open class CachesFactory(
         @Autowired private val cacheConfig: CacheConfig
 ) {
 

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -23,7 +23,7 @@ import java.util.*
 import kotlin.collections.ArrayList
 import kotlin.collections.HashMap
 
-class UpstreamsConfig {
+open class UpstreamsConfig {
     var defaultOptions: MutableList<DefaultOptions> = ArrayList<DefaultOptions>()
     var upstreams: MutableList<Upstream<*>> = ArrayList<Upstream<*>>()
 
@@ -172,6 +172,7 @@ class UpstreamsConfig {
     )
 
     class Method(
-            val name: String
+            val name: String,
+            val quorum: String? = null
     )
 }

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -232,7 +232,8 @@ class UpstreamsConfigReader(
             val enabled = getList<MappingNode>(mnode, "enabled")?.value?.map { m ->
                 getValueAsString(m, "name")?.let { name ->
                     UpstreamsConfig.Method(
-                            name = name
+                            name = name,
+                            quorum = getValueAsString(m, "quorum")
                     )
                 }
             }?.filterNotNull()?.toSet() ?: emptySet()

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -130,12 +130,18 @@ open class ConfiguredUpstreams(
         return defaultOptions
     }
 
-    private fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
+    fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
         return if (config.methods != null) {
             ManagedCallMethods(currentUpstreams.getDefaultMethods(chain),
                     config.methods!!.enabled.map { it.name }.toSet(),
                     config.methods!!.disabled.map { it.name }.toSet()
-            )
+            ).also {
+                config.methods!!.enabled.forEach { m ->
+                    if (m.quorum != null) {
+                        it.setQuorum(m.name, m.quorum)
+                    }
+                }
+            }
         } else {
             currentUpstreams.getDefaultMethods(chain)
         }

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

@@ -42,7 +42,7 @@ import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
 @Repository
-class CurrentMultistreamHolder(
+open class CurrentMultistreamHolder(
         @Autowired private val cachesFactory: CachesFactory
 ) : MultistreamHolder {
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt

@@ -18,8 +18,11 @@ package io.emeraldpay.dshackle.upstream.calls
 
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
+import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
+import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import org.slf4j.LoggerFactory
 import java.util.*
+import kotlin.collections.HashMap
 
 /**
  * Wrapper on top of another configuration, that may disable or enable additional methods on top of it.
@@ -34,20 +37,41 @@ class ManagedCallMethods(
 
     companion object {
         private val log = LoggerFactory.getLogger(ManagedCallMethods::class.java)
+        private val defaultQuorum = AlwaysQuorum()
     }
 
     private val delegated = delegate.getSupportedMethods().sorted()
     private val allAllowed: Set<String> = Collections.unmodifiableSet(
             enabled + delegated - disabled
     )
+    private val quorum: MutableMap<String, CallQuorum> = HashMap()
+
+    init {
+        enabled.forEach { m ->
+            quorum[m] = defaultQuorum
+        }
+    }
+
+    fun setQuorum(method: String, quorumId: String) {
+        val quorum = when (quorumId) {
+            "always" -> AlwaysQuorum()
+            "no-lag", "not-lagging", "no_lag", "not_lagging" -> NotLaggingQuorum(0)
+            "not-empty", "not_empty", "non-empty", "non_empty" -> NonEmptyQuorum()
+            else -> {
+                log.warn("Unknown quorum: $quorumId for custom method $method")
+                return
+            }
+        }
+        this.quorum[method] = quorum
+    }
 
     override fun getQuorumFor(method: String): CallQuorum {
         return when {
             Collections.binarySearch(delegated, method) >= 0 -> delegate.getQuorumFor(method)
-            enabled.contains(method) -> AlwaysQuorum()
+            enabled.contains(method) -> quorum[method] ?: defaultQuorum
             else -> {
                 log.warn("Getting quorum for unknown method")
-                AlwaysQuorum()
+                defaultQuorum
             }
         }
     }

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -233,6 +233,29 @@ class UpstreamsConfigReaderSpec extends Specification {
         }
     }
 
+    def "Parse config with methods and quorum"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-methods-quorum.yaml")
+        when:
+        def act = reader.read(config)
+        then:
+        act != null
+        with(act.upstreams.get(0)) {
+            methods != null
+            with(methods) {
+                enabled.size() == 2
+                with(enabled[0]) {
+                    it.name == "custom_foo"
+                    it.quorum == "not_lagging"
+                }
+                with(enabled[1]) {
+                    it.name == "custom_bar"
+                    it.quorum == "not_empty"
+                }
+            }
+        }
+    }
+
     def "Parse config with invalid ids"() {
         setup:
         def config = this.class.getClassLoader().getResourceAsStream("upstreams-no-id.yaml")

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -0,0 +1,38 @@
+package io.emeraldpay.dshackle.startup
+
+import io.emeraldpay.dshackle.FileResolver
+import io.emeraldpay.dshackle.cache.CachesFactory
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
+import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
+import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
+import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
+import io.emeraldpay.grpc.Chain
+import spock.lang.Specification
+
+class ConfiguredUpstreamsSpec extends Specification {
+
+    def "Applied  quorum to extra methods"() {
+        setup:
+        def currentUpstreams = Mock(CurrentMultistreamHolder) {
+            _ * getDefaultMethods(Chain.ETHEREUM) >> new DefaultEthereumMethods(Chain.ETHEREUM)
+        }
+        def configurer = new ConfiguredUpstreams(
+                currentUpstreams, Stub(FileResolver), Stub(UpstreamsConfig), Stub(CachesFactory)
+        )
+        def methods = new UpstreamsConfig.Methods(
+                [
+                        new UpstreamsConfig.Method("foo_bar", null),
+                        new UpstreamsConfig.Method("foo_bar", "not_empty")
+                ] as Set,
+                [] as Set
+        )
+        def upstream = new UpstreamsConfig.Upstream()
+        upstream.methods = methods
+        when:
+        def act = configurer.buildMethods(upstream, Chain.ETHEREUM)
+        then:
+        act instanceof ManagedCallMethods
+        act.getQuorumFor("foo_bar") instanceof NonEmptyQuorum
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy

@@ -18,8 +18,12 @@ package io.emeraldpay.dshackle.upstream.calls
 
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
+import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
+import io.emeraldpay.dshackle.quorum.NonceQuorum
+import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
+import io.emeraldpay.grpc.Chain
 import spock.lang.Specification
 
 class ManagedCallMethodsSpec extends Specification {
@@ -81,4 +85,29 @@ class ManagedCallMethodsSpec extends Specification {
         act != null
         act instanceof BroadcastQuorum
     }
+
+    def "Use custom quorum if provided"() {
+        setup:
+        def managed = new ManagedCallMethods(
+                new DefaultEthereumMethods(Chain.ETHEREUM),
+                ["eth_test", "eth_foo", "eth_bar"] as Set,
+                [] as Set
+        )
+        managed.setQuorum("eth_test", "not_empty")
+        managed.setQuorum("eth_foo", "not_lagging")
+        when:
+        def act = managed.getQuorumFor("eth_test")
+        then:
+        act instanceof NonEmptyQuorum
+
+        when:
+        act = managed.getQuorumFor("eth_foo")
+        then:
+        act instanceof NotLaggingQuorum
+
+        when:
+        act = managed.getQuorumFor("eth_bar")
+        then:
+        act instanceof AlwaysQuorum
+    }
 }

src/test/resources/upstreams-methods-quorum.yaml

@@ -0,0 +1,18 @@
+version: v1
+
+upstreams:
+  - id: local
+    chain: ethereum
+    methods:
+      enabled:
+        - name: custom_foo
+          quorum: not_lagging
+        - name: custom_bar
+          quorum: not_empty
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+        ws:
+          url: "ws://localhost:8546"
+          origin: "http://localhost"