Merge pull request #67 from p2p-org/fix-managed-methods-from-upstream

fix managed methods reusing old quorums

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -229,7 +229,7 @@ open class NativeCall(
                 .forMethod(method)
                 .forLabels(Selector.convertToMatcher(request.selector))
 
-            val callQuorum = availableMethods.getQuorumFor(method) // can be null in tests
+            val callQuorum = availableMethods.createQuorumFor(method) // can be null in tests
             callQuorum.init(upstream.getHead())
 
             // for NotLaggingQuorum it makes sense to select compatible upstreams before the call

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

@@ -37,10 +37,10 @@ class AggregatedCallMethods(
     /**
      * Finds first delegate that has Allowed that method and returns its Quorum
      */
-    override fun getQuorumFor(method: String): CallQuorum {
+    override fun createQuorumFor(method: String): CallQuorum {
         return delegates.find {
             it.isCallable(method) || it.isHardcoded(method)
-        }?.getQuorumFor(method) ?: throw IllegalStateException("No executor delegate for $method")
+        }?.createQuorumFor(method) ?: throw IllegalStateException("No executor delegate for $method")
     }
 
     /**

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

@@ -24,9 +24,11 @@ import io.emeraldpay.dshackle.quorum.CallQuorum
 interface CallMethods {
 
     /**
+     * For a stateful CallQuorum it _MUST CREATE_ a new instance of each time to avoid using a shared state between different requests
+     *
      * @return CallQuorum configured for the specified method
      */
-    fun getQuorumFor(method: String): CallQuorum
+    fun createQuorumFor(method: String): CallQuorum
 
     /**
      * Check if the method can be called on an upstream. Doesn't include Hardcoded methods

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

@@ -59,7 +59,7 @@ class DefaultBitcoinMethods : CallMethods {
     private val allowedMethods =
         (freshMethods + anyResponseMethods + headVerifiedMethods + broadcastMethods).sorted()
 
-    override fun getQuorumFor(method: String): CallQuorum {
+    override fun createQuorumFor(method: String): CallQuorum {
         return when {
             Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum()
             Collections.binarySearch(anyResponseMethods, method) >= 0 -> NonEmptyQuorum()

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

@@ -112,7 +112,7 @@ class DefaultEthereumMethods(
             getChainSpecificMethods(chain)
     }
 
-    override fun getQuorumFor(method: String): CallQuorum {
+    override fun createQuorumFor(method: String): CallQuorum {
         return when {
             filterMethods.contains(method) -> NotLaggingQuorum(1)
             hardcodedMethods.contains(method) -> AlwaysQuorum()

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

@@ -28,7 +28,7 @@ open class DirectCallMethods(private val methods: Set<String>) : CallMethods {
     constructor() : this(emptySet())
     constructor(methods: Collection<String>) : this(methods.toSet())
 
-    override fun getQuorumFor(method: String): CallQuorum {
+    override fun createQuorumFor(method: String): CallQuorum {
         return AlwaysQuorum()
     }
 

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

@@ -42,7 +42,7 @@ class EthereumCallSelector(
             "eth_getBalance",
             "eth_getCode",
             "eth_getTransactionCount",
-            // no "eth_getStorageAt" because it's has different structure, and therefore separate logic
+            // no "eth_getStorageAt" because it has different structure, and therefore separate logic
             "eth_call"
         ).sorted()
     }

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

@@ -21,6 +21,7 @@ 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.apache.commons.collections4.Factory
 import org.slf4j.LoggerFactory
 import java.io.IOException
 import java.util.Collections
@@ -38,14 +39,16 @@ class ManagedCallMethods(
 
     companion object {
         private val log = LoggerFactory.getLogger(ManagedCallMethods::class.java)
-        private val defaultQuorum = AlwaysQuorum()
+        private val defaultQuorum: Factory<CallQuorum> = Factory<CallQuorum> {
+            AlwaysQuorum()
+        }
     }
 
     private val delegated = delegate.getSupportedMethods().sorted()
     private val allAllowed: Set<String> = Collections.unmodifiableSet(
         enabled + delegated - disabled
     )
-    private val quorum: MutableMap<String, CallQuorum> = HashMap()
+    private val quorum: MutableMap<String, Factory<CallQuorum>> = HashMap()
     private val staticResponse: MutableMap<String, String> = HashMap()
     private val redefined = delegated.filter(enabled::contains).sorted()
 
@@ -57,9 +60,9 @@ class ManagedCallMethods(
 
     fun setQuorum(method: String, quorumId: String) {
         val quorum = when (quorumId) {
-            "always" -> AlwaysQuorum()
+            "always" -> Factory<CallQuorum> { AlwaysQuorum() }
-            "no-lag", "not-lagging", "no_lag", "not_lagging" -> NotLaggingQuorum(0)
+            "no-lag", "not-lagging", "no_lag", "not_lagging" -> Factory<CallQuorum> { NotLaggingQuorum(0) }
-            "not-empty", "not_empty", "non-empty", "non_empty" -> NonEmptyQuorum()
+            "not-empty", "not_empty", "non-empty", "non_empty" -> Factory<CallQuorum> { NonEmptyQuorum() }
             else -> {
                 log.warn("Unknown quorum: $quorumId for custom method $method")
                 return
@@ -72,13 +75,13 @@ class ManagedCallMethods(
         this.staticResponse[method] = response
     }
 
-    override fun getQuorumFor(method: String): CallQuorum {
+    override fun createQuorumFor(method: String): CallQuorum {
         return when {
-            isDelegated(method) && !isRedefined(method) -> delegate.getQuorumFor(method)
+            isDelegated(method) && !isRedefined(method) -> delegate.createQuorumFor(method)
-            enabled.contains(method) -> quorum[method] ?: defaultQuorum
+            enabled.contains(method) -> quorum[method]?.create() ?: defaultQuorum.create()
             else -> {
                 log.warn("Getting quorum for unknown method")
-                defaultQuorum
+                defaultQuorum.create()
             }
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt

@@ -132,8 +132,12 @@ class EthereumDirectReader(
      */
     private fun readWithQuorum(request: JsonRpcRequest): Mono<ByteArray> {
         return quorumReaderFactory
-            // we do not use Signer for internal requests because it doesn't make much sense
+            .create(
-            .create(up.getApiSource(Selector.empty), callMethodsFactory.create().getQuorumFor(request.method), null)
+                up.getApiSource(Selector.empty),
+                callMethodsFactory.create().createQuorumFor(request.method),
+                // we do not use Signer for internal requests because it doesn't make much sense
+                null
+            )
             .read(request)
             .map { it.value }
     }

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

@@ -1,7 +1,6 @@
 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.CallTargetsHolder
@@ -36,7 +35,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def act = configurer.buildMethods(upstream, Chain.ETHEREUM)
         then:
         act instanceof ManagedCallMethods
-        act.getQuorumFor("foo_bar") instanceof NonEmptyQuorum
+        act.createQuorumFor("foo_bar") instanceof NonEmptyQuorum
     }
 
     def "Got static response from extra methods"() {

src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy

@@ -58,9 +58,9 @@ class MultistreamSpec extends Specification {
         act.isCallable("eth_test1")
         act.isCallable("eth_test2")
         act.isCallable("eth_test3")
-        act.getQuorumFor("eth_test1") instanceof AlwaysQuorum
+        act.createQuorumFor("eth_test1") instanceof AlwaysQuorum
-        act.getQuorumFor("eth_test2") instanceof AlwaysQuorum
+        act.createQuorumFor("eth_test2") instanceof AlwaysQuorum
-        act.getQuorumFor("eth_test3") instanceof AlwaysQuorum
+        act.createQuorumFor("eth_test3") instanceof AlwaysQuorum
     }
 
     def "Filter Best Status accepts any input when none available "() {

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

@@ -31,11 +31,11 @@ class AggregatedCallMethodsSpec extends Specification {
         def delegate2 = Mock(CallMethods) {
             _ * getSupportedMethods() >> ["eth_test", "foo_bar"]
             1 * isCallable("eth_test") >> true
-            1 * getQuorumFor("eth_test") >> quorum
+            1 * createQuorumFor("eth_test") >> quorum
         }
         def aggregate = new AggregatedCallMethods([delegate1, delegate2])
         when:
-        def act = aggregate.getQuorumFor("eth_test")
+        def act = aggregate.createQuorumFor("eth_test")
         then:
         act == quorum
     }

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

@@ -18,14 +18,15 @@ package io.emeraldpay.dshackle.upstream.calls
 
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
+import io.emeraldpay.dshackle.quorum.CallQuorum
 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.dshackle.Chain
 import spock.lang.Specification
 
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
 class ManagedCallMethodsSpec extends Specification {
 
     def "Gets quorum for enabled method"() {
@@ -36,7 +37,7 @@ class ManagedCallMethodsSpec extends Specification {
                 [] as Set
         )
         when:
-        def act = managed.getQuorumFor("eth_test")
+        def act = managed.createQuorumFor("eth_test")
         then:
         act instanceof AlwaysQuorum
     }
@@ -72,7 +73,7 @@ class ManagedCallMethodsSpec extends Specification {
         def delegated = ["eth_test", "eth_test2"] as Set
         def delegate = Mock(CallMethods) {
             _ * it.getSupportedMethods() >> delegated
-            0 * it.getQuorumFor("eth_test") >> new BroadcastQuorum()
+            0 * it.createQuorumFor("eth_test") >> new BroadcastQuorum()
         }
         def managed = new ManagedCallMethods(
                 delegate,
@@ -80,7 +81,7 @@ class ManagedCallMethodsSpec extends Specification {
                 ["foo_bar"] as Set
         )
         when:
-        def act = managed.getQuorumFor("eth_test")
+        def act = managed.createQuorumFor("eth_test")
         then:
         act != null
         act instanceof AlwaysQuorum
@@ -96,18 +97,40 @@ class ManagedCallMethodsSpec extends Specification {
         managed.setQuorum("eth_test", "not_empty")
         managed.setQuorum("eth_foo", "not_lagging")
         when:
-        def act = managed.getQuorumFor("eth_test")
+        def act = managed.createQuorumFor("eth_test")
         then:
         act instanceof NonEmptyQuorum
 
         when:
-        act = managed.getQuorumFor("eth_foo")
+        act = managed.createQuorumFor("eth_foo")
         then:
         act instanceof NotLaggingQuorum
 
         when:
-        act = managed.getQuorumFor("eth_bar")
+        act = managed.createQuorumFor("eth_bar")
         then:
         act instanceof AlwaysQuorum
     }
+
+    def "Doesn't reuse same instance"() {
+        def managed = new ManagedCallMethods(
+                new DefaultEthereumMethods(Chain.ETHEREUM),
+                ["eth_test"] as Set,
+                [] as Set
+        )
+        def parallel = Executors.newFixedThreadPool(16)
+        when:
+        List<CallQuorum> instances = []
+        50.times {
+            instances << managed.createQuorumFor("eth_test")
+        }
+        parallel.shutdown()
+        parallel.awaitTermination(5, TimeUnit.SECONDS)
+
+        def ids = instances.collect { System.identityHashCode(it) }
+
+        then:
+        instances.size() == 50
+        ids.toSet().size() == 50
+    }
 }