StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

configure method groups

Browse Source
This commit is contained in:
Maksim Fomenkov
2022-12-12 15:46:40 +03:00
parent e144a9b0de
commit c6ffceaff8
16 changed files with 269 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -76,6 +76,7 @@ open class UpstreamsConfig {
var connection: T? = null
val labels = Labels()
var methods: Methods? = null
var methodGroups: MethodGroups? = null
var role: UpstreamRole = UpstreamRole.PRIMARY
@Suppress("UNCHECKED_CAST")
@@ -182,7 +183,12 @@ open class UpstreamsConfig {
class Methods(
val enabled: Set<Method>,
val disabled: Set<Method>
val disabled: Set<Method>,
)
class MethodGroups(
val enabled: Set<String>,
val disabled: Set<String>,
)
class Method(

10
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -255,6 +255,7 @@ class UpstreamsConfigReader(
upstream.nodeId = getValueAsInt(upNode, "node-id")
upstream.options = tryReadOptions(upNode)
upstream.methods = tryReadMethods(upNode)
upstream.methodGroups = tryReadMethodGroups(upNode)
getValueAsBool(upNode, "enabled")?.let {
upstream.isEnabled = it
}
@@ -330,6 +331,15 @@ class UpstreamsConfigReader(
}
}
internal fun tryReadMethodGroups(upNode: MappingNode): UpstreamsConfig.MethodGroups? {
return getMapping(upNode, "method-groups")?.let {
UpstreamsConfig.MethodGroups(
enabled = getListOfString(it, "enabled")?.toSet() ?: emptySet(),
disabled = getListOfString(it, "disabled")?.toSet() ?: emptySet()
)
}
}
internal fun readOptions(values: MappingNode): UpstreamsConfig.Options {
val options = UpstreamsConfig.Options()
getValueAsInt(values, "min-peers")?.let {

12
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -139,13 +139,15 @@ open class ConfiguredUpstreams(
}
fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
return if (config.methods != null) {
return if (config.methods != null || config.methodGroups != null) {
ManagedCallMethods(
callTargets.getDefaultMethods(chain),
config.methods!!.enabled.map { it.name }.toSet(),
config.methods!!.disabled.map { it.name }.toSet()
delegate = callTargets.getDefaultMethods(chain),
enabled = config.methods?.enabled?.map { it.name }?.toSet() ?: emptySet(),
disabled = config.methods?.disabled?.map { it.name }?.toSet() ?: emptySet(),
groupsEnabled = config.methodGroups?.enabled ?: emptySet(),
groupsDisabled = config.methodGroups?.disabled ?: emptySet()
).also {
config.methods!!.enabled.forEach { m ->
config.methods?.enabled?.forEach { m ->
if (m.quorum != null) {
it.setQuorum(m.name, m.quorum)
}

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

@@ -72,4 +72,7 @@ class AggregatedCallMethods(
it.isHardcoded(method)
}?.executeHardcoded(method) ?: throw IllegalStateException("No hardcoded for $method")
}
override fun getGroupMethods(groupName: String): Set<String> =
delegates.map { it.getGroupMethods(groupName) }.firstOrNull() ?: emptySet()
}

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

@@ -60,4 +60,9 @@ interface CallMethods {
fun isAvailable(method: String): Boolean {
return isCallable(method) || isHardcoded(method)
}
/**
* Returns list of methods conforming the methods group
*/
fun getGroupMethods(groupName: String): Set<String>
}

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

@@ -89,4 +89,6 @@ class DefaultBitcoinMethods : CallMethods {
else -> throw RpcException(-32601, "Method not found")
}
}
override fun getGroupMethods(groupName: String): Set<String> = emptySet()
}

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

@@ -47,6 +47,18 @@ class DefaultEthereumMethods(
"eth_newBlockFilter",
"eth_newPendingTransactionFilter",
)
val traceMethods = listOf(
"trace_call",
"trace_callMany",
"trace_rawTransaction",
"trace_replayBlockTransactions",
"trace_replayTransaction",
"trace_block",
"trace_filter",
"trace_get",
"trace_transaction",
)
}
private val anyResponseMethods = listOf(
@@ -106,8 +118,7 @@ class DefaultEthereumMethods(
allowedMethods = anyResponseMethods +
firstValueMethods +
specialMethods +
headVerifiedMethods +
filterMethods -
headVerifiedMethods -
chainUnsupportedMethods(chain) +
getChainSpecificMethods(chain)
}
@@ -297,6 +308,13 @@ class DefaultEthereumMethods(
return json.toByteArray()
}
override fun getGroupMethods(groupName: String): Set<String> =
when (groupName) {
"filter" -> filterMethods
"trace" -> traceMethods
else -> emptyList()
}.toSet()
override fun getSupportedMethods(): Set<String> {
return allowedMethods.plus(hardcodedMethods).toSortedSet()
}

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

@@ -47,4 +47,6 @@ open class DirectCallMethods(private val methods: Set<String>) : CallMethods {
override fun executeHardcoded(method: String): ByteArray {
return "unsupported".toByteArray()
}
override fun getGroupMethods(groupName: String): Set<String> = emptySet()
}

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

@@ -34,7 +34,9 @@ import java.util.Collections
class ManagedCallMethods(
private val delegate: CallMethods,
private val enabled: Set<String>,
private val disabled: Set<String>
disabled: Set<String>,
groupsEnabled: Set<String>,
groupsDisabled: Set<String>
) : CallMethods {
companion object {
@@ -44,13 +46,15 @@ class ManagedCallMethods(
}
}
private val delegated = delegate.getSupportedMethods().sorted()
private val delegated = delegate.getSupportedMethods().associateWith { true }
private val allGroupEnabled = groupsEnabled.flatMap { delegate.getGroupMethods(it) }
private val allGroupDisabled = groupsDisabled.flatMap { delegate.getGroupMethods(it) }
private val allAllowed: Set<String> = Collections.unmodifiableSet(
enabled + delegated - disabled
delegated.keys + allGroupEnabled - allGroupDisabled.toSet() + enabled - disabled
)
private val quorum: MutableMap<String, Factory<CallQuorum>> = HashMap()
private val staticResponse: MutableMap<String, String> = HashMap()
private val redefined = delegated.filter(enabled::contains).sorted()
private val redefined = delegated.keys.filter(enabled::contains).associateWith { true }
init {
enabled.forEach { m ->
@@ -87,11 +91,11 @@ class ManagedCallMethods(
}
private fun isDelegated(method: String): Boolean {
return Collections.binarySearch(delegated, method) >= 0
return delegated[method] ?: false
}
private fun isRedefined(method: String): Boolean {
return Collections.binarySearch(redefined, method) >= 0
return redefined[method] ?: false
}
override fun isCallable(method: String): Boolean {
@@ -121,4 +125,7 @@ class ManagedCallMethods(
}
return delegate.executeHardcoded(method)
}
override fun getGroupMethods(groupName: String): Set<String> =
delegate.getGroupMethods(groupName)
}

27
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -410,9 +410,28 @@ class UpstreamsConfigReaderSpec extends Specification {
then:
act != null
act.upstreams.size() == 2
act.upstreams[0].nodeId == 1
act.upstreams[0].id == "has_node_id"
act.upstreams[1].nodeId == null
act.upstreams[1].id == "has_no_node_id"
with(act.upstreams.get(0)) {
nodeId == 1
id == "has_node_id"
}
with(act.upstreams.get(1)) {
nodeId == null
id == "has_no_node_id"
}
}
def "Parse method groups"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-method-groups.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 1
with(act.upstreams.get(0)) {
id == "enable_filter_methods"
methodGroups.enabled.first() == "filter"
methodGroups.disabled.first() == "trace"
}
}
}

42
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -354,7 +354,7 @@ class NativeCallSpec extends Specification {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["foo_bar"] as Set, [] as Set
["foo_bar"] as Set, [] as Set, [] as Set, [] as Set
)
methods.setQuorum("foo_bar", "not_lagging")
def head = Mock(Head) {
@@ -395,7 +395,7 @@ class NativeCallSpec extends Specification {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_newFilter"] as Set, [] as Set
["eth_newFilter"] as Set, [] as Set, [] as Set, [] as Set
)
methods.setQuorum("eth_newFilter", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
@@ -426,9 +426,8 @@ class NativeCallSpec extends Specification {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_getFilterChanges"] as Set, [] as Set
["eth_getFilterChanges"] as Set, [] as Set, [] as Set, [] as Set
)
methods.setQuorum("eth_getFilterChanges", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
@@ -457,9 +456,8 @@ class NativeCallSpec extends Specification {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_uninstallFilter"] as Set, [] as Set
["eth_uninstallFilter"] as Set, [] as Set, [] as Set, [] as Set
)
methods.setQuorum("eth_uninstallFilter", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
@@ -553,14 +551,24 @@ class NativeCallSpec extends Specification {
def "Decorate eth_newFilter result"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
[] as Set, [] as Set, ["filter"] as Set, [] as Set
)
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream(
TestingCommons.api(), methods
))
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)255)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null)
@@ -575,14 +583,24 @@ class NativeCallSpec extends Specification {
def "Decorate eth_newFilter result with short nodeId"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
[] as Set, [] as Set, ["filter"] as Set, [] as Set
)
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream(
TestingCommons.api(), methods
))
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)1)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null)

22
src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy
View File

@@ -104,4 +104,26 @@ class ConfiguredUpstreamsSpec extends Specification {
h4 == (byte)8
h5 == (byte)-128
}
def "Supporting method groups"() {
setup:
def callTargetsHolder = new CallTargetsHolder()
def configurer = new ConfiguredUpstreams(
Stub(FileResolver),
Stub(UpstreamsConfig),
callTargetsHolder,
Mock(ApplicationEventPublisher)
)
def methodsGroup = new UpstreamsConfig.MethodGroups(
["filter"] as Set,
[] as Set
)
def upstream = new UpstreamsConfig.Upstream()
upstream.methodGroups = methodsGroup
when:
def act = configurer.buildMethods(upstream, Chain.ETHEREUM)
then:
act instanceof ManagedCallMethods
act.supportedMethods.findAll {it.containsIgnoreCase("filter")}.size() == 6
}
}

5
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.CallTargetsHolder
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.EthereumPosMultiStream
@@ -74,6 +75,10 @@ class TestingCommons {
return new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods))
}
static EthereumPosRpcUpstreamMock upstream(Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods callMethods) {
return new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api, callMethods)
}
static Multistream multistream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
return multistream(upstream(api))
}

18
src/test/groovy/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethodsSpec.groovy
View File

@@ -69,4 +69,22 @@ class DefaultEthereumMethodsSpec extends Specification {
"eth_getRootHash"]
Chain.OPTIMISM | ["rollup_gasPrices"]
}
def "Has no filter methods by default"() {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
when:
def act = methods.getSupportedMethods().findAll { it.containsIgnoreCase("filter") }
then:
act.isEmpty()
}
def "Has no trace methods by default"() {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
when:
def act = methods.getSupportedMethods().findAll { it.containsIgnoreCase("trace") }
then:
act.isEmpty()
}
}

91
src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy
View File

@@ -34,6 +34,8 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
new DirectCallMethods(["eth_test2", "foo_bar"] as Set),
["eth_test"] as Set,
[] as Set,
[] as Set,
[] as Set
)
when:
@@ -47,6 +49,8 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
new DirectCallMethods(["eth_test2"] as Set),
["eth_test"] as Set,
[] as Set,
[] as Set,
[] as Set
)
when:
@@ -60,7 +64,9 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
new DirectCallMethods(["eth_test2", "foo_bar"] as Set),
["eth_test"] as Set,
["foo_bar"] as Set
["foo_bar"] as Set,
[] as Set,
[] as Set
)
when:
def act = managed.getSupportedMethods()
@@ -78,7 +84,9 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
delegate,
["eth_test"] as Set,
["foo_bar"] as Set
["foo_bar"] as Set,
[] as Set,
[] as Set
)
when:
def act = managed.createQuorumFor("eth_test")
@@ -92,6 +100,8 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_test", "eth_foo", "eth_bar"] as Set,
[] as Set,
[] as Set,
[] as Set
)
managed.setQuorum("eth_test", "not_empty")
@@ -116,6 +126,8 @@ class ManagedCallMethodsSpec extends Specification {
def managed = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_test"] as Set,
[] as Set,
[] as Set,
[] as Set
)
def parallel = Executors.newFixedThreadPool(16)
@@ -133,4 +145,79 @@ class ManagedCallMethodsSpec extends Specification {
instances.size() == 50
ids.toSet().size() == 50
}
def "Test enable method group"() {
setup:
def managed = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
[] as Set,
[] as Set,
["filter"] as Set,
[] as Set
)
when:
def act = managed.getSupportedMethods()
then:
act.containsAll([
"eth_getFilterChanges",
"eth_getFilterLogs",
"eth_uninstallFilter",
"eth_newFilter",
"eth_newBlockFilter",
"eth_newPendingTransactionFilter"
])
}
def "Test enable method group minus one"() {
setup:
def managed = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
[] as Set,
["eth_newPendingTransactionFilter"] as Set,
["filter"] as Set,
[] as Set
)
when:
def act = managed.getSupportedMethods()
then:
act.containsAll([
"eth_getFilterChanges",
"eth_getFilterLogs",
"eth_uninstallFilter",
"eth_newFilter",
"eth_newBlockFilter",
])
!act.contains("eth_newPendingTransactionFilter")
}
def "Test disabled group not disable enabled method"() {
setup:
def managed = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_newPendingTransactionFilter"] as Set,
[] as Set,
[] as Set,
["filter"] as Set
)
when:
def act = managed.getSupportedMethods()
then:
with(act.findAll {it in [
"eth_getFilterChanges",
"eth_getFilterLogs",
"eth_uninstallFilter",
"eth_newFilter",
"eth_newBlockFilter",
"eth_newPendingTransactionFilter"
]}) {
size() == 1
first() == "eth_newPendingTransactionFilter"
}
}
}

13
src/test/resources/upstreams-method-groups.yaml Normal file
View File

@@ -0,0 +1,13 @@
version: v1
upstreams:
- id: enable_filter_methods
chain: ethereum
method-groups:
enabled:
- filter
disabled:
- trace
connection:
ethereum:
rpc:
url: "http://localhost:8545"