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

Filter and sort upstreams if latest (#476)

Browse Source
This commit is contained in:
KirillPamPam
2024-05-15 14:50:24 +04:00
committed by GitHub
parent de4e16309b
commit b468b4039d
12 changed files with 90 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt
View File

@@ -68,16 +68,16 @@ interface RequestReaderFactory {
class Default : RequestReaderFactory {
override fun create(data: ReaderData): RequestReader {
if (data.quorum is MaximumValueQuorum || data.quorum is BroadcastQuorum) {
return BroadcastReader(data.multistream.getAll(), data.matcher, data.signer, data.quorum, data.tracer)
return BroadcastReader(data.multistream.getAll(), data.upstreamFilter.matcher, data.signer, data.quorum, data.tracer)
}
val apis = data.multistream.getApiSource(data.matcher)
val apis = data.multistream.getApiSource(data.upstreamFilter)
return QuorumRequestReader(apis, data.quorum, data.signer, data.tracer)
}
}
data class ReaderData(
val multistream: Multistream,
val matcher: Selector.Matcher,
val upstreamFilter: Selector.UpstreamFilter,
val quorum: CallQuorum,
val signer: ResponseSigner?,
val tracer: Tracer,

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

@@ -331,12 +331,12 @@ open class NativeCall(
),
)
}
val requestMatcher = requestItem.selectorsList
val upstreamFilter = requestItem.selectorsList
.takeIf { it.isNotEmpty() }
?.run { Mono.just(Selector.convertToMatcher(this, upstream.getHead())) }
?.run { Selector.convertToUpstreamFilter(this) }
// for ethereum the actual block needed for the call may be specified in the call parameters
val callSpecificMatcher: Mono<Selector.Matcher> =
requestMatcher ?: upstream.callSelector?.getMatcher(method, params, upstream.getHead(), passthrough) ?: Mono.empty()
upstreamFilter?.matcher?.let { Mono.just(it) } ?: upstream.callSelector?.getMatcher(method, params, upstream.getHead(), passthrough) ?: Mono.empty()
return callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm ->
val matcher = Selector.Builder()
.withMatcher(csm)
@@ -363,7 +363,7 @@ open class NativeCall(
requestItem.id,
nonce,
upstream,
matcher.build(),
Selector.UpstreamFilter(upstreamFilter?.sort ?: Selector.Sort.default, matcher.build()),
callQuorum,
parsedCallDetails(requestItem),
requestDecorator,
@@ -433,7 +433,7 @@ open class NativeCall(
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
val reader = requestReaderFactory.create(
ReaderData(ctx.upstream, ctx.matcher, ctx.callQuorum, signer, tracer),
ReaderData(ctx.upstream, ctx.upstreamFilter, ctx.callQuorum, signer, tracer),
)
val counter = reader.attempts()
@@ -572,7 +572,7 @@ open class NativeCall(
val id: Int,
val nonce: Long?,
val upstream: Multistream,
val matcher: Selector.Matcher,
val upstreamFilter: Selector.UpstreamFilter,
val callQuorum: CallQuorum,
val payload: T,
val requestDecorator: RequestDecorator,
@@ -587,13 +587,13 @@ open class NativeCall(
id: Int,
nonce: Long?,
upstream: Multistream,
matcher: Selector.Matcher,
upstreamFilter: Selector.UpstreamFilter,
callQuorum: CallQuorum,
payload: T,
requestId: String,
requestCount: Int,
) : this(
id, nonce, upstream, matcher, callQuorum, payload,
id, nonce, upstream, upstreamFilter, callQuorum, payload,
NoneRequestDecorator(), NoneResultDecorator(), null, false, requestId, requestCount,
)
@@ -613,13 +613,13 @@ open class NativeCall(
fun <X> withPayload(payload: X): ValidCallContext<X> {
return ValidCallContext(
id, nonce, upstream, matcher, callQuorum, payload,
id, nonce, upstream, upstreamFilter, callQuorum, payload,
requestDecorator, resultDecorator, forwardedSelector, streamRequest, requestId, requestCount,
)
}
fun getApis(): ApiSource {
return upstream.getApiSource(matcher)
return upstream.getApiSource(upstreamFilter)
}
}

14
src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
View File

@@ -38,6 +38,7 @@ class FilteredApis(
matcher: Selector.Matcher,
private val pos: Int,
private val retries: Int,
sort: Selector.Sort = Selector.Sort.default,
) : ApiSource {
private val internalMatcher: Selector.Matcher
@@ -69,6 +70,13 @@ class FilteredApis(
pos: Int,
) : this(chain, allUpstreams, matcher, pos, DEFAULT_RETRY_LIMIT)
constructor(
chain: Chain,
allUpstreams: List<Upstream>,
upstreamFilter: Selector.UpstreamFilter,
pos: Int,
) : this(chain, allUpstreams, upstreamFilter.matcher, pos, DEFAULT_RETRY_LIMIT, upstreamFilter.sort)
constructor(
chain: Chain,
allUpstreams: List<Upstream>,
@@ -79,12 +87,12 @@ class FilteredApis(
it.getRole() == UpstreamsConfig.UpstreamRole.PRIMARY
}.let {
startFrom(it, pos)
}
}.sortedWith(sort.comparator)
private val secondaryUpstreams: List<Upstream> = allUpstreams.filter {
it.getRole() == UpstreamsConfig.UpstreamRole.SECONDARY
}.let {
startFrom(it, pos)
}
}.sortedWith(sort.comparator)
private val standardWithFallback: List<Upstream>
private val counter: AtomicInteger = AtomicInteger(0)
@@ -98,7 +106,7 @@ class FilteredApis(
it.getRole() == UpstreamsConfig.UpstreamRole.FALLBACK
}.let {
startFrom(it, pos)
}
}.sortedWith(sort.comparator)
standardWithFallback = emptyList<Upstream>()
.plus(primaryUpstreams)
.plus(secondaryUpstreams)

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

@@ -206,12 +206,12 @@ abstract class Multistream(
/**
* Get a source for direct APIs
*/
open fun getApiSource(matcher: Selector.Matcher): ApiSource {
open fun getApiSource(upstreamFilter: Selector.UpstreamFilter): ApiSource {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
}
return FilteredApis(chain, getUpstreams(), matcher, i)
return FilteredApis(chain, getUpstreams(), upstreamFilter, i)
}
/**

30
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -41,15 +41,19 @@ class Selector {
val anyLabel = AnyLabelMatcher()
@JvmStatic
fun convertToMatcher(selectors: List<BlockchainOuterClass.Selector>, head: Head): Matcher {
return selectors
fun convertToUpstreamFilter(selectors: List<BlockchainOuterClass.Selector>): UpstreamFilter {
val matcher = selectors
.map {
when {
it.hasSlotHeightSelector() -> {
SlotMatcher(it.slotHeightSelector.slotHeight)
}
it.hasHeightSelector() -> {
val height = if (it.heightSelector.height == -1L) head.getCurrentHeight() else it.heightSelector.height
val height = if (it.heightSelector.height == -1L) {
null
} else {
it.heightSelector.height
}
if (height == null) {
empty
} else {
@@ -61,6 +65,10 @@ class Selector {
}.run {
MultiMatcher(this)
}
val sort = selectors.firstOrNull { it.hasHeightSelector() && it.heightSelector.height == -1L }
?.let { Sort(compareByDescending { it.getHead().getCurrentHeight() }) }
?: Sort.default
return UpstreamFilter(sort, matcher)
}
@JvmStatic
@@ -153,6 +161,15 @@ class Selector {
}
}
data class Sort(
val comparator: Comparator<Upstream>,
) {
companion object {
@JvmStatic
val default = Sort(compareBy { null })
}
}
abstract class Matcher {
fun matches(up: Upstream): Boolean = matchesWithCause(up).matched()
@@ -161,6 +178,13 @@ class Selector {
abstract fun describeInternal(): String
}
data class UpstreamFilter(
val sort: Sort,
val matcher: Matcher,
) {
constructor(matcher: Matcher) : this(Sort.default, matcher)
}
data class MultiMatcher(
private val matchers: Collection<Matcher>,
) : Matcher() {

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

@@ -92,7 +92,7 @@ open class BitcoinMultistream(
* Finds an API that executed directly on a remote.
*/
open fun getDirectApi(matcher: Selector.Matcher): Mono<ChainReader> {
val apis = getApiSource(matcher)
val apis = getApiSource(Selector.UpstreamFilter(matcher))
apis.request(1)
return Mono.from(apis)
.map(Upstream::getIngressReader)

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

@@ -18,7 +18,7 @@ class RemoteUnspentReader(
)
override fun read(key: Address): Mono<List<SimpleUnspent>> {
val apis = upstreams.getApiSource(selector)
val apis = upstreams.getApiSource(Selector.UpstreamFilter(selector))
apis.request(1)
return Mono.empty()
}

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

@@ -237,7 +237,7 @@ class EthereumDirectReader(
it.create(
RequestReaderFactory.ReaderData(
up,
requestMatcher,
Selector.UpstreamFilter(requestMatcher),
callMethodsFactory.create().createQuorumFor(request.method),
null,
tracer,

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

@@ -80,7 +80,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(
1, null, upstream, Selector.empty, new AlwaysQuorum(),
1, null, upstream, new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1
)
@@ -101,7 +101,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(
15, null, upstream, Selector.empty, new AlwaysQuorum(),
15, null, upstream, new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1
)
@@ -126,7 +126,7 @@ class NativeCallSpec extends Specification {
1 * read(_) >> Mono.just(new RequestReader.Result("\"foo\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte)1, "test", "v"), null))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), new Selector.UpstreamFilter(Selector.empty), quorum,
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
@@ -148,7 +148,7 @@ class NativeCallSpec extends Specification {
1 * read(new ChainRequest("eth_test", new ListParams(), 10)) >> Mono.empty()
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), new Selector.UpstreamFilter(Selector.empty), quorum,
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
@@ -174,7 +174,7 @@ class NativeCallSpec extends Specification {
)
}
}
def call = new NativeCall.ValidCallContext(12, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(12, 10, TestingCommons.multistream(TestingCommons.api()), new Selector.UpstreamFilter(Selector.empty), quorum,
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
@@ -420,9 +420,9 @@ class NativeCallSpec extends Specification {
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act.matcher != null
act.matcher instanceof Selector.MultiMatcher
with((Selector.MultiMatcher) act.matcher) {
act.upstreamFilter.matcher != null
act.upstreamFilter.matcher instanceof Selector.MultiMatcher
with((Selector.MultiMatcher) act.upstreamFilter.matcher) {
it.getMatchers().size() >= 1
it.getMatcher(Selector.HeightMatcher) != null
with(it.getMatcher(Selector.HeightMatcher)) {
@@ -528,7 +528,7 @@ class NativeCallSpec extends Specification {
def "Parse empty params"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
def act = nativeCall.parseParams(ctx)
@@ -541,7 +541,7 @@ class NativeCallSpec extends Specification {
def "Parse none params"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
def act = nativeCall.parseParams(ctx)
@@ -554,7 +554,7 @@ class NativeCallSpec extends Specification {
def "Parse single param"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams(false)), "reqId", 1)
when:
def act = nativeCall.parseParams(ctx)
@@ -567,7 +567,7 @@ class NativeCallSpec extends Specification {
def "Parse multi param"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams(false, 123)), "reqId", 1)
when:
def act = nativeCall.parseParams(ctx)
@@ -580,7 +580,7 @@ class NativeCallSpec extends Specification {
def "Decorate eth_getFilterUpdates params"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_getFilterUpdates", new ListParams("0xabcd")),
new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null, false, "reqId", 1)
when:
@@ -612,7 +612,7 @@ class NativeCallSpec extends Specification {
1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte) 255, "", ""), null))
}
}
def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, multistream, new Selector.UpstreamFilter(Selector.empty), quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", new ListParams()),
new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, false, "reqId", 1)
@@ -645,7 +645,7 @@ class NativeCallSpec extends Specification {
1 * read(_) >> Mono.just(new RequestReader.Result("\"0xab\"".bytes, null, 1, new Upstream.UpstreamSettingsData((byte) 1, "", ""), null))
}
}
def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum,
def call = new NativeCall.ValidCallContext(1, 10, multistream, new Selector.UpstreamFilter(Selector.empty), quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", new ListParams()),
new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, false, "reqId", 1)
@@ -667,7 +667,7 @@ class NativeCallSpec extends Specification {
def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(10, null,
upstream,
Selector.empty, new AlwaysQuorum(),
new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
nativeCall.fetch(ctx)
@@ -684,7 +684,7 @@ class NativeCallSpec extends Specification {
def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(10, null,
upstream,
Selector.empty, new AlwaysQuorum(),
new Selector.UpstreamFilter(Selector.empty), new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", new ListParams()), "reqId", 1)
when:
def act = nativeCall.fetch(ctx)

16
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -86,7 +86,7 @@ class FilteredApisSpec extends Specification {
it.setStatus(UpstreamAvailability.OK)
}
when:
def iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 0, 0)
def iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 0, 0, Selector.Sort.default)
iter.request(10)
then:
StepVerifier.create(iter)
@@ -97,7 +97,7 @@ class FilteredApisSpec extends Specification {
.verify(Duration.ofSeconds(1))
when:
iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 1, 0)
iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 1, 0, Selector.Sort.default)
iter.request(10)
then:
StepVerifier.create(iter)
@@ -108,7 +108,7 @@ class FilteredApisSpec extends Specification {
.verify(Duration.ofSeconds(1))
when:
iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 1, 2)
iter = new FilteredApis(Chain.ETHEREUM__MAINNET, upstreams, matcher, 1, 2, Selector.Sort.default)
iter.request(10)
then:
StepVerifier.create(iter)
@@ -134,7 +134,7 @@ class FilteredApisSpec extends Specification {
TestingCommons.upstream(it)
}
when:
def act = new FilteredApis(Chain.ETHEREUM__MAINNET, ups, Selector.empty, 2, 0)
def act = new FilteredApis(Chain.ETHEREUM__MAINNET, ups, Selector.empty, 2, 0, Selector.Sort.default)
act.request(10)
then:
StepVerifier.create(act)
@@ -153,7 +153,7 @@ class FilteredApisSpec extends Specification {
TestingCommons.upstream(it)
}
when:
def act = new FilteredApis(Chain.ETHEREUM__MAINNET, ups, Selector.empty, 2, 0)
def act = new FilteredApis(Chain.ETHEREUM__MAINNET, ups, Selector.empty, 2, 0, Selector.Sort.default)
act.request(3)
then:
StepVerifier.create(act)
@@ -212,7 +212,7 @@ class FilteredApisSpec extends Specification {
when:
def act = new FilteredApis(Chain.ETHEREUM__MAINNET,
[] + fallback + standard,
Selector.empty, 0, 1)
Selector.empty, 0, 1, Selector.Sort.default)
act.request(10)
then:
StepVerifier.create(act)
@@ -248,7 +248,7 @@ class FilteredApisSpec extends Specification {
when:
def act = new FilteredApis(Chain.ETHEREUM__MAINNET,
[] + fallback + standard + secondary,
Selector.empty, 0, 2)
Selector.empty, 0, 2, Selector.Sort.default)
act.request(11)
then:
StepVerifier.create(act)
@@ -285,7 +285,7 @@ class FilteredApisSpec extends Specification {
when:
def act = new FilteredApis(Chain.ETHEREUM__MAINNET,
[] + lagging + ok,
Selector.empty, 0, 1)
Selector.empty, 0, 1, Selector.Sort.default)
act.request(4)
then:
StepVerifier.create(act)

16
src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy
View File

@@ -42,9 +42,9 @@ class SelectorSpec extends Specification {
)
.build()
when:
def act = Selector.convertToMatcher(List.of(slotHeightSelector), Stub(Head))
def act = Selector.convertToUpstreamFilter(List.of(slotHeightSelector))
then:
act == new Selector.MultiMatcher(List.of(new Selector.SlotMatcher(10000)))
act.matcher == new Selector.MultiMatcher(List.of(new Selector.SlotMatcher(10000)))
}
def "Convert height selector"() {
@@ -57,16 +57,13 @@ class SelectorSpec extends Specification {
)
.build()
when:
def act = Selector.convertToMatcher(List.of(heightSelector), Stub(Head))
def act = Selector.convertToUpstreamFilter(List.of(heightSelector))
then:
act == new Selector.MultiMatcher(List.of(new Selector.HeightMatcher(10000)))
act.matcher == new Selector.MultiMatcher(List.of(new Selector.HeightMatcher(10000)))
}
def "Convert height selector with latest"() {
setup:
def head = Mock(Head) {
1 * getCurrentHeight() >> 15000
}
def heightSelector = BlockchainOuterClass.Selector.newBuilder()
.setHeightSelector(
BlockchainOuterClass.HeightSelector.newBuilder()
@@ -75,9 +72,10 @@ class SelectorSpec extends Specification {
)
.build()
when:
def act = Selector.convertToMatcher(List.of(heightSelector), head)
def act = Selector.convertToUpstreamFilter(List.of(heightSelector))
then:
act == new Selector.MultiMatcher(List.of(new Selector.HeightMatcher(15000)))
act.matcher == new Selector.MultiMatcher(List.of(Selector.empty))
act.sort != Selector.Sort.default
}
def "Convert LABEL match"() {

4
src/test/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactoryTest.kt
View File

@@ -34,7 +34,7 @@ class RequestReaderFactoryTest {
Arguments.of(
RequestReaderFactory.ReaderData(
ms,
Selector.empty,
Selector.UpstreamFilter(Selector.empty),
MaximumValueQuorum(),
null,
tracer,
@@ -43,7 +43,7 @@ class RequestReaderFactoryTest {
Arguments.of(
RequestReaderFactory.ReaderData(
ms,
Selector.empty,
Selector.UpstreamFilter(Selector.empty),
BroadcastQuorum(),
null,
tracer,