Fix archive detection (#561)
This commit is contained in:
@@ -25,7 +25,7 @@ class Defaults {
|
|||||||
const val maxMetadataSize = 16384
|
const val maxMetadataSize = 16384
|
||||||
val timeout: Duration = Duration.ofSeconds(60)
|
val timeout: Duration = Duration.ofSeconds(60)
|
||||||
val timeoutInternal: Duration = timeout.dividedBy(4)
|
val timeoutInternal: Duration = timeout.dividedBy(4)
|
||||||
val internalCallsTimeout = Duration.ofSeconds(3)
|
val internalCallsTimeout = Duration.ofSeconds(5)
|
||||||
val retryConnection: Duration = Duration.ofSeconds(10)
|
val retryConnection: Duration = Duration.ofSeconds(10)
|
||||||
val grpcServerKeepAliveTime: Long = 15 // seconds
|
val grpcServerKeepAliveTime: Long = 15 // seconds
|
||||||
val grpcServerKeepAliveTimeout: Long = 5
|
val grpcServerKeepAliveTimeout: Long = 5
|
||||||
|
|||||||
@@ -19,11 +19,14 @@ class EthereumUpstreamSettingsDetector(
|
|||||||
private val chain: Chain,
|
private val chain: Chain,
|
||||||
) : BasicEthUpstreamSettingsDetector(_upstream) {
|
) : BasicEthUpstreamSettingsDetector(_upstream) {
|
||||||
private val blockNumberReader = EthereumArchiveBlockNumberReader(upstream.getIngressReader())
|
private val blockNumberReader = EthereumArchiveBlockNumberReader(upstream.getIngressReader())
|
||||||
|
private val notArchived = upstream
|
||||||
|
.getLabels()
|
||||||
|
.find { it.getOrDefault("archive", "") == "false" } != null
|
||||||
|
|
||||||
override fun internalDetectLabels(): Flux<Pair<String, String>> {
|
override fun internalDetectLabels(): Flux<Pair<String, String>> {
|
||||||
return Flux.merge(
|
return Flux.merge(
|
||||||
detectNodeType(),
|
detectNodeType(),
|
||||||
detectArchiveNode(),
|
detectArchiveNode(notArchived),
|
||||||
detectGasLabels(),
|
detectGasLabels(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -88,8 +91,8 @@ class EthereumUpstreamSettingsDetector(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun detectArchiveNode(): Mono<Pair<String, String>> {
|
private fun detectArchiveNode(notArchived: Boolean): Mono<Pair<String, String>> {
|
||||||
if (upstream.getLabels().firstOrNull { it.getOrDefault("archive", "") == "false" } != null) {
|
if (notArchived) {
|
||||||
return Mono.empty()
|
return Mono.empty()
|
||||||
}
|
}
|
||||||
return Mono.zip(
|
return Mono.zip(
|
||||||
|
|||||||
@@ -38,7 +38,7 @@ class AggregatedUpstreamValidator(
|
|||||||
.timeout(internalCallsTimeout)
|
.timeout(internalCallsTimeout)
|
||||||
.onErrorResume {
|
.onErrorResume {
|
||||||
log.error("Error during upstream validation for {}, reason - {}", upstream.getId(), it.message)
|
log.error("Error during upstream validation for {}, reason - {}", upstream.getId(), it.message)
|
||||||
Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR)
|
Mono.just(ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -76,6 +76,10 @@ class TestingCommons {
|
|||||||
return new GenericUpstreamMock(Chain.ETHEREUM__MAINNET, api)
|
return new GenericUpstreamMock(Chain.ETHEREUM__MAINNET, api)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static GenericUpstreamMock upstream(Reader<ChainRequest, ChainResponse> api, Map<String, String> labels) {
|
||||||
|
return new GenericUpstreamMock("test-1", Chain.ETHEREUM__MAINNET, api, labels)
|
||||||
|
}
|
||||||
|
|
||||||
static GenericUpstreamMock upstream(Reader<ChainRequest, ChainResponse> api, String method) {
|
static GenericUpstreamMock upstream(Reader<ChainRequest, ChainResponse> api, String method) {
|
||||||
return upstream(api, [method])
|
return upstream(api, [method])
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -84,6 +84,29 @@ class EthereumUpstreamSettingsDetectorSpec extends Specification {
|
|||||||
.verify(Duration.ofSeconds(1))
|
.verify(Duration.ofSeconds(1))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def "No archive label if it is set manually"() {
|
||||||
|
setup:
|
||||||
|
def up = TestingCommons.upstream(
|
||||||
|
new ApiReaderMock().tap {
|
||||||
|
answer("web3_clientVersion", [], "Bor/v0.4.0/linux-amd64/go1.19.10")
|
||||||
|
answer("eth_blockNumber", [], "0x10df3e5")
|
||||||
|
},
|
||||||
|
Map.of("archive", "false")
|
||||||
|
)
|
||||||
|
def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET)
|
||||||
|
|
||||||
|
when:
|
||||||
|
def act = detector.internalDetectLabels()
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectNext(
|
||||||
|
new Pair<String, String>("client_type", "bor"),
|
||||||
|
new Pair<String, String>("client_version", "v0.4.0"),
|
||||||
|
)
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(1))
|
||||||
|
}
|
||||||
|
|
||||||
def "Only default label"() {
|
def "Only default label"() {
|
||||||
setup:
|
setup:
|
||||||
def up = Mock(DefaultUpstream) {
|
def up = Mock(DefaultUpstream) {
|
||||||
@@ -128,6 +151,7 @@ class EthereumUpstreamSettingsDetectorSpec extends Specification {
|
|||||||
1 * read(new ChainRequest("web3_clientVersion", new ListParams())) >>
|
1 * read(new ChainRequest("web3_clientVersion", new ListParams())) >>
|
||||||
Mono.just(new ChainResponse('"Erigon/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3"'.getBytes(), null))
|
Mono.just(new ChainResponse('"Erigon/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3"'.getBytes(), null))
|
||||||
}
|
}
|
||||||
|
1 * getLabels() >> List.of()
|
||||||
}
|
}
|
||||||
def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET)
|
def detector = new EthereumUpstreamSettingsDetector(up, Chain.ETHEREUM__MAINNET)
|
||||||
when:
|
when:
|
||||||
|
|||||||
Reference in New Issue
Block a user