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

Merge pull request #80 from p2p-org/ability_to_turn_off_local_router

Browse Source 
disable local router caches
This commit is contained in:
a10zn8
2022-12-12 17:56:03 +04:00
committed by GitHub
parent e144a9b0de 2c666eed99
commit 6a252029f7
13 changed files with 53 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/config/CacheConfig.kt
View File

@@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.config
class CacheConfig {
var requestsCacheEnabled = true
var redis: Redis? = null
class Redis(

5
src/main/kotlin/io/emeraldpay/dshackle/config/CacheConfigReader.kt
View File

@@ -33,6 +33,9 @@ class CacheConfigReader : YamlConfigReader(), ConfigReader<CacheConfig> {
override fun read(input: MappingNode?): CacheConfig? {
return getMapping(input, "cache")?.let { node ->
val config = CacheConfig()
getValueAsBool(node, "requests-cache-enabled")?.let {
config.requestsCacheEnabled = it
}
getMapping(node, "redis")?.let { redisNode ->
val redis = CacheConfig.Redis()
val enabled = getValueAsBool(redisNode, "enabled") ?: true
@@ -52,7 +55,7 @@ class CacheConfigReader : YamlConfigReader(), ConfigReader<CacheConfig> {
config.redis = redis
}
}
if (config.redis == null) {
if (config.redis == null && config.requestsCacheEnabled) {
return null
}
config

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
@@ -55,7 +56,8 @@ import java.util.concurrent.atomic.AtomicInteger
@Service
open class NativeCall(
private val multistreamHolder: MultistreamHolder,
private val signer: ResponseSigner
private val signer: ResponseSigner,
cacheConfig: CacheConfig
) {
private val log = LoggerFactory.getLogger(NativeCall::class.java)
@@ -63,6 +65,8 @@ open class NativeCall(
private val nullValue: ByteArray = "null".toByteArray()
private val localRouterEnabled = cacheConfig.requestsCacheEnabled
var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
private val ethereumCallSelectors = EnumMap<Chain, EthereumCallSelector>(Chain::class.java)
@@ -121,9 +125,6 @@ open class NativeCall(
result.setErrorMessage(error.message).setErrorCode(error.id)
}
} else {
if (it.result == null || it.result.isEmpty() || nullValue.contentEquals(it.result)) {
log.warn("Empty result [${it.result}] on building response, method ${it.ctx?.payload?.method}, params ${it.ctx?.payload?.params}")
}
result.payload = ByteString.copyFrom(it.result)
}
if (it.nonce != null && it.signature != null) {
@@ -273,7 +274,7 @@ open class NativeCall(
if (method in DefaultEthereumMethods.newFilterMethods) CreateFilterDecorator() else NoneResultDecorator()
fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getRoutedApi(ctx.matcher)
return ctx.upstream.getRoutedApi(localRouterEnabled)
.flatMap { api ->
api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
.flatMap(JsonRpcResponse::requireResult)

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

@@ -148,7 +148,7 @@ abstract class Multistream(
/**
* Finds an API that leverages caches and other optimizations/transformations of the request.
*/
abstract fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
abstract fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -106,7 +106,7 @@ open class BitcoinMultistream(
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(callRouter)
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -154,8 +154,8 @@ open class EthereumMultistream(
return this as T
}
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead()))
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
}
override fun getSubscribe(): EthereumSubscribe {

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
View File

@@ -39,7 +39,8 @@ import java.math.BigInteger
class LocalCallRouter(
private val reader: EthereumCachingReader,
private val methods: CallMethods,
private val head: Head
private val head: Head,
private val localEnabled: Boolean
) : Reader<JsonRpcRequest, JsonRpcResponse> {
companion object {
@@ -56,7 +57,9 @@ class LocalCallRouter(
return Mono.just(methods.executeHardcoded(key.method))
.map { JsonRpcResponse(it, null) }
}
if (!localEnabled) {
return Mono.empty()
}
if (!methods.isCallable(key.method)) {
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
View File

@@ -147,8 +147,8 @@ open class EthereumPosMultiStream(
return this as T
}
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead()))
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
}
override fun getSubscribe(): EthereumSubscribe {

10
src/test/groovy/io/emeraldpay/dshackle/config/CacheConfigReaderSpec.groovy
View File

@@ -47,4 +47,14 @@ class CacheConfigReaderSpec extends Specification {
//later may be not null if we support something else besides Redis
act == null
}
def "Local read disabled"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("cache-local-router-disabled.yaml")
when:
def act = reader.read(config)
then:
!act.requestsCacheEnabled
}
}

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

@@ -21,10 +21,10 @@ import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
import io.emeraldpay.dshackle.test.MultistreamHolderMock
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
@@ -52,14 +52,19 @@ class NativeCallSpec extends Specification {
ObjectMapper objectMapper = Global.objectMapper
def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null) {
def nativeCall(MultistreamHolder upstreams = null, ResponseSigner signer = null, Boolean enableCache = true) {
if (upstreams == null) {
upstreams = Stub(MultistreamHolder)
}
if (signer == null) {
signer = Stub(ResponseSigner)
}
new NativeCall(upstreams, signer)
def config = new CacheConfig()
config.requestsCacheEnabled = enableCache
new NativeCall(upstreams, signer, config)
}
def "Tries router first"() {

3
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -327,8 +327,9 @@ class MultistreamSpec extends Specification {
super(chain, upstreams, caches)
}
@NotNull
@Override
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(@NotNull Selector.Matcher matcher) {
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(boolean localEnabled) {
return null
}

13
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy
View File

@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.EmptyHead
@@ -29,7 +30,8 @@ class LocalCallRouterSpec extends Specification {
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
),
methods,
new EmptyHead()
new EmptyHead(),
true
)
when:
def act = router.read(new JsonRpcRequest("eth_coinbase", [])).block(Duration.ofSeconds(1))
@@ -47,7 +49,8 @@ class LocalCallRouterSpec extends Specification {
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
),
methods,
new EmptyHead()
new EmptyHead(),
true
)
when:
def act = router.read(new JsonRpcRequest("eth_getTransactionByHash", ["test"], 10))
@@ -69,7 +72,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head)
def router = new LocalCallRouter(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["latest", false])
@@ -95,7 +98,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head)
def router = new LocalCallRouter(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["earliest", false])
@@ -121,7 +124,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head)
def router = new LocalCallRouter(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["0x123ef", false])

2
src/test/resources/cache-local-router-disabled.yaml Normal file
View File

@@ -0,0 +1,2 @@
cache:
requests-cache-enabled: false