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

Bypass getBlockByNumber cache (#679)

Browse Source 
Co-authored-by: Кирилл <zhitelevkm@m2.ru>
This commit is contained in:
KirillPamPam
2025-07-09 14:28:24 +04:00
committed by GitHub
parent 2ad9b3db3f
commit e644ead250
3 changed files with 4 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -65,7 +65,7 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
methods: CallMethods,
head: Head,
): Mono<ChainReader> {
return Mono.just(EthereumLocalReader(cachingReader as EthereumCachingReader, methods, head))
return Mono.just(EthereumLocalReader(cachingReader as EthereumCachingReader, methods))
}
override fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription {

63
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
View File

@@ -21,18 +21,12 @@ import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.ChainReader
import io.emeraldpay.dshackle.upstream.ChainRequest
import io.emeraldpay.dshackle.upstream.ChainResponse
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.hex.HexQuantity
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.switchIfEmpty
import java.math.BigInteger
/**
* Reader for JSON RPC requests. Verifies if the method is allowed, transforms if necessary, and calls EthereumReader for data.
@@ -44,7 +38,6 @@ import java.math.BigInteger
class EthereumLocalReader(
private val reader: EthereumCachingReader,
private val methods: CallMethods,
private val head: Head,
) : ChainReader {
override fun read(key: ChainRequest): Mono<ChainResponse> {
@@ -111,10 +104,6 @@ class EthereumLocalReader(
}
}
method == "eth_getBlockByNumber" -> {
getBlockByNumber(params.list, key.upstreamFilter)
}
method == "eth_getTransactionReceipt" -> {
if (params.list.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
@@ -135,56 +124,4 @@ class EthereumLocalReader(
}
return null
}
fun getBlockByNumber(params: List<Any?>, upstreamFilter: Selector.UpstreamFilter): Mono<ChainResponse>? {
if (params.size != 2 || params[0] == null || params[1] == null) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val number: Long
val withTx = params[1].toString().toBoolean()
if (withTx) {
// with Tx request much more efficient in remote call
return null
}
try {
val blockRef = params[0].toString()
when {
blockRef.startsWith("0x") -> {
val quantity = HexQuantity.from(blockRef) ?: throw IllegalArgumentException()
number = quantity.value.let {
if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) {
it.toLong()
} else {
throw IllegalArgumentException()
}
}
}
blockRef == "latest" -> {
number = head.getCurrentHeight() ?: return null
}
blockRef == "earliest" -> {
number = 0
}
blockRef == "pending" -> {
return null
}
blockRef == "finalized" || blockRef == "safe" -> {
val type = FinalizationType.fromBlockRef(blockRef)
return reader
.blockByFinalization().read(type)
.map {
ChainResponse(it.data.json, it.resolvedUpstreamData, FinalizationData(it.data.height, type))
}
}
else -> {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Block number is invalid")
}
}
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number")
}
return reader.blocksByHeightAsCont(upstreamFilter)
.read(number).map { ChainResponse(it.data.json, null, it.resolvedUpstreamData) }
}
}

152
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
View File

@@ -1,22 +1,12 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.EmptyHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ChainRequest
import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
import org.apache.commons.collections4.functors.ConstantFactory
import reactor.core.publisher.Mono
import spock.lang.Specification
import java.time.Duration
@@ -33,8 +23,7 @@ class EthereumLocalReaderSpec extends Specification {
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)),
TestingCommons.tracerMock()
),
methods,
new EmptyHead(),
methods
)
when:
def act = router.read(new ChainRequest("eth_coinbase", new ListParams())).block(Duration.ofSeconds(1))
@@ -52,8 +41,7 @@ class EthereumLocalReaderSpec extends Specification {
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)),
TestingCommons.tracerMock()
),
methods,
new EmptyHead(),
methods
)
when:
def act = router.read(new ChainRequest("eth_getTransactionByHash", new ListParams(["test"]), 10))
@@ -61,138 +49,4 @@ class EthereumLocalReaderSpec extends Specification {
then:
act == null
}
def "getBlockByNumber with latest uses latest id"() {
setup:
def head = Mock(Head) {
1 * getCurrentHeight() >> 101L
}
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(101L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), List.of())
)
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
def router = new EthereumLocalReader(reader, methods, head)
when:
def act = router.getBlockByNumber(["latest", false], Selector.UpstreamFilter.default)
then:
act != null
with(act.block()) {
it.result.length > 0
with(Global.objectMapper.readValue(it.result, BlockJson)) {
number == 101
}
}
}
def "getBlockByNumber with earliest uses 0 block"() {
setup:
def head = Stub(Head) {}
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(0L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), List.of())
)
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
def router = new EthereumLocalReader(reader, methods, head)
when:
def act = router.getBlockByNumber(["earliest", false], Selector.UpstreamFilter.default)
then:
act != null
with(act.block()) {
it.result.length > 0
with(Global.objectMapper.readValue(it.result, BlockJson)) {
number == 0
}
}
}
def "getBlockByNumber fetches the block"() {
setup:
def head = Stub(Head) {}
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
1 * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> Mock(Reader) {
1 * read(74735L) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), List.of())
)
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
def router = new EthereumLocalReader(reader, methods, head)
when:
def act = router.getBlockByNumber(["0x123ef", false], Selector.UpstreamFilter.default)
then:
act != null
with(act.block()) {
it.result.length > 0
with(Global.objectMapper.readValue(it.result, BlockJson)) {
number == 74735
}
}
}
def "getBlockByNumber fetches the block by tag"() {
setup:
def head = Stub(Head) {}
def reader = Mock(EthereumCachingReader) {
1 * blockByFinalization() >> Mock(Reader) {
1 * read(FinalizationType.SAFE_BLOCK) >> Mono.just(
new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), List.of())
)
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
def router = new EthereumLocalReader(reader, methods, head)
when:
def act = router.read(
new ChainRequest("eth_getBlockByNumber",
new ListParams("safe", false))
)
then:
act != null
with(act.block()) {
it.result.length > 0
with(Global.objectMapper.readValue(it.result, BlockJson)) {
number == 74735
}
it.finalization == new FinalizationData(74735, FinalizationType.SAFE_BLOCK)
}
}
def "getBlockByNumber skips requests with tx bodies"() {
setup:
def head = Mock(Head)
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * txByHashAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
_ * blocksByHeightAsCont(Selector.UpstreamFilter.default) >> new EmptyReader<>()
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
def router = new EthereumLocalReader(reader, methods, head)
when:
def act = router.getBlockByNumber(["0x0", true], Selector.UpstreamFilter.default)
then:
act == null
}
}