problem: doesn't handle upstream error for additionally allowed methods; allowing an existing method redefines its logic
rel: #67
This commit is contained in:
@@ -36,7 +36,7 @@ open class AlwaysQuorum: CallQuorum {
|
||||
}
|
||||
|
||||
override fun isFailed(): Boolean {
|
||||
return false
|
||||
return rpcError != null
|
||||
}
|
||||
|
||||
override fun record(response: ByteArray, upstream: Upstream): Boolean {
|
||||
|
||||
@@ -16,10 +16,8 @@
|
||||
*/
|
||||
package io.emeraldpay.dshackle.quorum
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper
|
||||
import io.emeraldpay.dshackle.upstream.Head
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.infinitape.etherjar.rpc.JacksonRpcConverter
|
||||
|
||||
open class BroadcastQuorum(
|
||||
val quorum: Int = 3
|
||||
@@ -33,11 +31,11 @@ open class BroadcastQuorum(
|
||||
}
|
||||
|
||||
override fun isResolved(): Boolean {
|
||||
return calls >= quorum
|
||||
return calls >= quorum && txid != null
|
||||
}
|
||||
|
||||
override fun isFailed(): Boolean {
|
||||
return false
|
||||
return calls >= quorum && getError() != null
|
||||
}
|
||||
|
||||
override fun getResult(): ByteArray? {
|
||||
|
||||
@@ -62,7 +62,7 @@ class QuorumRpcReader(
|
||||
?: RpcException(-32000, "Unknown Upstream error")
|
||||
)
|
||||
} else {
|
||||
log.warn("Did get any result from upstream. Method [${key.method}] using [$q]")
|
||||
log.warn("Did not get any result from upstream. Method [${key.method}] using [$q]")
|
||||
Mono.empty<Result>()
|
||||
}
|
||||
}
|
||||
@@ -118,7 +118,7 @@ class QuorumRpcReader(
|
||||
}
|
||||
}
|
||||
.doOnNext {
|
||||
if (!it.isResolved()) {
|
||||
if (!it.isResolved() && !it.isFailed()) {
|
||||
log.debug("No quorum for ${key.method} using [${quorum}]. Error: ${it.getError()?.message ?: ""}")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -84,7 +84,7 @@ class DefaultEthereumMethods(
|
||||
return when {
|
||||
hardcodedMethods.contains(method) -> AlwaysQuorum()
|
||||
firstValueMethods.contains(method) -> AlwaysQuorum()
|
||||
anyResponseMethods.contains(method) -> NotLaggingQuorum(6)
|
||||
anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
|
||||
headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
|
||||
specialMethods.contains(method) -> {
|
||||
when (method) {
|
||||
|
||||
@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.calls
|
||||
|
||||
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
|
||||
import io.emeraldpay.dshackle.quorum.CallQuorum
|
||||
import org.slf4j.LoggerFactory
|
||||
import java.util.*
|
||||
|
||||
/**
|
||||
@@ -31,15 +32,23 @@ class ManagedCallMethods(
|
||||
private val disabled: Set<String>
|
||||
): CallMethods {
|
||||
|
||||
companion object {
|
||||
private val log = LoggerFactory.getLogger(ManagedCallMethods::class.java)
|
||||
}
|
||||
|
||||
private val delegated = delegate.getSupportedMethods().sorted()
|
||||
private val allAllowed: Set<String> = Collections.unmodifiableSet(
|
||||
enabled + delegate.getSupportedMethods() - disabled
|
||||
enabled + delegated - disabled
|
||||
)
|
||||
|
||||
override fun getQuorumFor(method: String): CallQuorum {
|
||||
return if (enabled.contains(method)) {
|
||||
AlwaysQuorum()
|
||||
} else {
|
||||
delegate.getQuorumFor(method)
|
||||
return when {
|
||||
Collections.binarySearch(delegated, method) >= 0 -> delegate.getQuorumFor(method)
|
||||
enabled.contains(method) -> AlwaysQuorum()
|
||||
else -> {
|
||||
log.warn("Getting quorum for unknown method")
|
||||
AlwaysQuorum()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
/**
|
||||
* Copyright (c) 2021 EmeraldPay, Inc
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package io.emeraldpay.dshackle.quorum
|
||||
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
|
||||
import spock.lang.Specification
|
||||
|
||||
class AlwaysQuorumSpec extends Specification {
|
||||
|
||||
def "Failed if error received"() {
|
||||
setup:
|
||||
def quorum = new AlwaysQuorum()
|
||||
def up = Stub(Upstream)
|
||||
when:
|
||||
quorum.record(new JsonRpcException(1, "test"), up)
|
||||
then:
|
||||
quorum.isFailed()
|
||||
!quorum.isResolved()
|
||||
quorum.getError() != null
|
||||
with(quorum.getError()) {
|
||||
message == "test"
|
||||
}
|
||||
}
|
||||
|
||||
def "Resolved if result received"() {
|
||||
setup:
|
||||
def quorum = new AlwaysQuorum()
|
||||
def up = Stub(Upstream)
|
||||
when:
|
||||
quorum.record("123".bytes, up)
|
||||
then:
|
||||
quorum.isResolved()
|
||||
quorum.getResult() == "123".bytes
|
||||
!quorum.isFailed()
|
||||
}
|
||||
}
|
||||
@@ -93,4 +93,31 @@ class BroadcastQuorumSpec extends Specification {
|
||||
q.isResolved()
|
||||
objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"
|
||||
}
|
||||
|
||||
def "Failed if error received 3+ times"() {
|
||||
setup:
|
||||
def quorum = new BroadcastQuorum(3)
|
||||
def up = Stub(Upstream)
|
||||
when:
|
||||
quorum.record(new JsonRpcException(1, "test 1"), up)
|
||||
then:
|
||||
!quorum.isFailed()
|
||||
!quorum.isResolved()
|
||||
|
||||
when:
|
||||
quorum.record(new JsonRpcException(1, "test 2"), up)
|
||||
then:
|
||||
!quorum.isFailed()
|
||||
!quorum.isResolved()
|
||||
|
||||
when:
|
||||
quorum.record(new JsonRpcException(1, "test 3"), up)
|
||||
then:
|
||||
quorum.isFailed()
|
||||
!quorum.isResolved()
|
||||
quorum.getError() != null
|
||||
with(quorum.getError()) {
|
||||
message == "test 3"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.test.TestingCommons
|
||||
import io.emeraldpay.dshackle.upstream.FilteredApis
|
||||
import io.emeraldpay.dshackle.upstream.Selector
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.infinitape.etherjar.rpc.RpcException
|
||||
@@ -59,12 +60,11 @@ class QuorumRpcReaderSpec extends Specification {
|
||||
.verify(Duration.ofSeconds(1))
|
||||
}
|
||||
|
||||
def "always-quorum - retry upstream error"() {
|
||||
def "always-quorum - return upstream error"() {
|
||||
setup:
|
||||
def api = Mock(Reader) {
|
||||
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
|
||||
Mono.just(JsonRpcResponse.error(1, "test")),
|
||||
Mono.just(JsonRpcResponse.ok("1"))
|
||||
1 * read(new JsonRpcRequest("eth_test", [])) >>> [
|
||||
Mono.just(JsonRpcResponse.error(1, "test"))
|
||||
]
|
||||
}
|
||||
def up = Mock(Upstream) {
|
||||
@@ -85,8 +85,9 @@ class QuorumRpcReaderSpec extends Specification {
|
||||
|
||||
then:
|
||||
StepVerifier.create(act)
|
||||
.expectNext("1")
|
||||
.expectComplete()
|
||||
.expectErrorMatches {
|
||||
it instanceof JsonRpcException && ((JsonRpcException) it).error.message == "test"
|
||||
}
|
||||
.verify(Duration.ofSeconds(1))
|
||||
}
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package io.emeraldpay.dshackle.upstream.calls
|
||||
|
||||
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
|
||||
import io.emeraldpay.dshackle.quorum.BroadcastQuorum
|
||||
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
|
||||
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
|
||||
import spock.lang.Specification
|
||||
@@ -26,7 +27,7 @@ class ManagedCallMethodsSpec extends Specification {
|
||||
def "Gets quorum for enabled method"() {
|
||||
setup:
|
||||
def managed = new ManagedCallMethods(
|
||||
new DirectCallMethods(),
|
||||
new DirectCallMethods(["eth_test2", "foo_bar"] as Set),
|
||||
["eth_test"] as Set,
|
||||
[] as Set
|
||||
)
|
||||
@@ -61,4 +62,23 @@ class ManagedCallMethodsSpec extends Specification {
|
||||
then:
|
||||
act.sort() == ["eth_test", "eth_test2"].sort()
|
||||
}
|
||||
|
||||
def "Use quorum from delegate if it supports the method"() {
|
||||
setup:
|
||||
def delegated = ["eth_test", "eth_test2"] as Set
|
||||
def delegate = Mock(CallMethods) {
|
||||
_ * it.getSupportedMethods() >> delegated
|
||||
1 * it.getQuorumFor("eth_test") >> new BroadcastQuorum()
|
||||
}
|
||||
def managed = new ManagedCallMethods(
|
||||
delegate,
|
||||
["eth_test"] as Set,
|
||||
["foo_bar"] as Set
|
||||
)
|
||||
when:
|
||||
def act = managed.getQuorumFor("eth_test")
|
||||
then:
|
||||
act != null
|
||||
act instanceof BroadcastQuorum
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user