chore: bump emerald-grpc submodule (adds Mova chain refs) (#883)

Bumps emerald-grpc 5191731 -> 111ff26, i.e. p2p-org/emerald-grpc#165, which
adds CHAIN_MOVA__MAINNET = 1170 and CHAIN_MOVA__TESTNET = 10205 to the ChainRef
enum.

Describe.kt:38 and SubscribeChainStatus.kt:31 filter chains with
'Common.ChainRef.forNumber(it.id) != null'. mova's grpcId is 1170, which had no
enum entry, so the chain was dropped from the gRPC surface even though the
mova-nodecore upstream is healthy in all three public-multiregion regions.
Completes #882, which added the chain to the embedded chains.yaml.

Submodule range contains exactly that one commit (+2 lines in
proto/common.proto).

buildSrc/src/main/kotlin/chainsconfig.codegen.gradle.kts

@@ -142,6 +142,7 @@ open class CodeGen(private val config: ChainsConfig) {
             "kadena" -> "BlockchainType.KADENA"
             "avm" -> "BlockchainType.AVM"
             "app" -> "BlockchainType.ETHEREUM"
+            "aptos" -> "BlockchainType.ETHEREUM"
             else -> throw IllegalArgumentException("unknown blockchain type $type")
         }
     }

docs/reference-configuration.adoc

@@ -849,6 +849,18 @@ rpc:
     password: "${ETH_PASSWORD}"
 ----
 
+| `rpc.bearer-auth` + `rpc.bearer-auth.token`
+a| HTTP Bearer token authorization (`Authorization: Bearer <token>` header), if required by the remote server. +
+Cannot be used together with `basic-auth`.
+Value can also reference env variables, for example:
+[source,yaml]
+----
+rpc:
+  url: "https://ethereum.com:8545"
+  bearer-auth:
+    token: "${ETH_TOKEN}"
+----
+
 | `ws.url`
 | WebSocket URL to connect to.
 Optional, but optimizes performance if it's available.
@@ -859,6 +871,9 @@ Optional, but optimizes performance if it's available.
 | `ws.basic-auth` + ...
 | WebSocket Basic Auth configuration, if required by the remote server
 
+| `ws.bearer-auth` + `ws.bearer-auth.token`
+| WebSocket Bearer token authorization, if required by the remote server
+
 | `ws.frameSize`
 | WebSocket frame size limit.
 Ex `1kb`, `1024` (same as `1kb), `2mb`, etc.

foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptions.kt

@@ -20,6 +20,7 @@ class ChainOptions {
         val disableBoundValidation: Boolean = false,
         val valdateErigonBug: Boolean,
         val disableLogIndexValidation: Boolean = false,
+        val disablePendingTxValidation: Boolean = false,
     )
 
     data class DefaultOptions(
@@ -43,7 +44,8 @@ class ChainOptions {
         var disableLivenessSubscriptionValidation: Boolean? = null,
         var disableBoundValidation: Boolean? = null,
         var validateErigonBug: Boolean? = null,
-        var disableLogIndexValidation: Boolean? = null
+        var disableLogIndexValidation: Boolean? = null,
+        var disablePendingTxValidation: Boolean? = null
     ) {
         companion object {
             @JvmStatic
@@ -76,6 +78,7 @@ class ChainOptions {
             copy.disableBoundValidation = overwrites.disableBoundValidation ?: this.disableBoundValidation
             copy.validateErigonBug = overwrites.validateErigonBug ?: this.validateErigonBug
             copy.disableLogIndexValidation = overwrites.disableLogIndexValidation ?: this.disableLogIndexValidation
+            copy.disablePendingTxValidation = overwrites.disablePendingTxValidation ?: this.disablePendingTxValidation
             return copy
         }
 
@@ -97,6 +100,7 @@ class ChainOptions {
                 this.disableBoundValidation ?: false,
                 this.validateErigonBug ?: true,
                 this.disableLogIndexValidation ?: false,
+                this.disablePendingTxValidation ?: false,
             )
     }
 }

foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptionsReader.kt

@@ -61,6 +61,9 @@ class ChainOptionsReader : YamlConfigReader<ChainOptions.PartialOptions>() {
         getValueAsBool(values, "disable-log-index-validation")?.let {
             options.disableLogIndexValidation = it
         }
+        getValueAsBool(values, "disable-pending-tx-validation")?.let {
+            options.disablePendingTxValidation = it
+        }
         return options
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/config/AuthConfig.kt

@@ -33,6 +33,10 @@ class AuthConfig {
         val password: String,
     ) : ClientAuth()
 
+    class ClientBearerAuth(
+        val token: String,
+    ) : ClientAuth()
+
     class ClientTlsAuth(
         var ca: String? = null,
         var certificate: String? = null,

src/main/kotlin/io/emeraldpay/dshackle/config/AuthConfigReader.kt

@@ -40,6 +40,18 @@ class AuthConfigReader : YamlConfigReader<AuthConfig>() {
         }
     }
 
+    fun readClientBearerAuth(node: MappingNode?): AuthConfig.ClientBearerAuth? {
+        return getMapping(node, "bearer-auth")?.let { authNode ->
+            val token = getValueAsString(authNode, "token")
+            if (token != null) {
+                AuthConfig.ClientBearerAuth(token)
+            } else {
+                log.warn("Bearer auth is not fully configured, token is required")
+                null
+            }
+        }
+    }
+
     fun readClientTls(node: MappingNode?): AuthConfig.ClientTlsAuth? {
         return getMapping(node, "tls")?.let { authNode ->
             val auth = AuthConfig.ClientTlsAuth()

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -136,12 +136,14 @@ data class UpstreamsConfig(
         constructor(url: URI) : this(url, DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE)
 
         var basicAuth: AuthConfig.ClientBasicAuth? = null
+        var bearerAuth: AuthConfig.ClientBearerAuth? = null
         var tls: AuthConfig.ClientTlsAuth? = null
     }
 
     data class WsEndpoint(val url: URI) {
         var origin: URI? = null
         var basicAuth: AuthConfig.ClientBasicAuth? = null
+        var bearerAuth: AuthConfig.ClientBearerAuth? = null
         var frameSize: Int? = null
         var msgSize: Int? = null
         var connections: Int = 1

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -147,6 +147,7 @@ class UpstreamsConfigReader(
             getValueAsString(node, "url")?.let { url ->
                 val http = UpstreamsConfig.HttpEndpoint(URI(url), DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE)
                 http.basicAuth = authConfigReader.readClientBasicAuth(node)
+                http.bearerAuth = readBearerAuth(node, http.basicAuth, url)
                 http.tls = authConfigReader.readClientTls(node)
                 connection.esplora = http
             }
@@ -173,6 +174,19 @@ class UpstreamsConfigReader(
         return connection
     }
 
+    private fun readBearerAuth(
+        node: MappingNode?,
+        basicAuth: AuthConfig.ClientBasicAuth?,
+        url: String,
+    ): AuthConfig.ClientBearerAuth? {
+        val bearerAuth = authConfigReader.readClientBearerAuth(node)
+        if (bearerAuth != null && basicAuth != null) {
+            log.warn("Both basic-auth and bearer-auth are configured for $url, basic-auth is used")
+            return null
+        }
+        return bearerAuth
+    }
+
     private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? {
         return getMapping(connConfigNode, "rpc")?.let { node ->
             val maxConnections = getValueAsInt(node, "max-connections") ?: DEFAULT_MAX_CONNECTIONS
@@ -181,6 +195,7 @@ class UpstreamsConfigReader(
             getValueAsString(node, "url")?.let { url ->
                 val http = UpstreamsConfig.HttpEndpoint(URI(url), maxConnections, queueSize)
                 http.basicAuth = authConfigReader.readClientBasicAuth(node)
+                http.bearerAuth = readBearerAuth(node, http.basicAuth, url)
                 http.tls = authConfigReader.readClientTls(node)
                 http
             }
@@ -224,6 +239,7 @@ class UpstreamsConfigReader(
                     ws.origin = URI(origin)
                 }
                 ws.basicAuth = authConfigReader.readClientBasicAuth(node)
+                ws.bearerAuth = readBearerAuth(node, ws.basicAuth, url)
 
                 getValueAsBytes(node, "frameSize")?.let {
                     if (it < 65_535) {

src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt

@@ -32,11 +32,12 @@ class ChainEventMapper {
     }
 
     fun mapCapabilities(capabilities: Collection<Capability>): BlockchainOuterClass.ChainEvent {
-        val caps = capabilities.map {
+        val caps = capabilities.filter { it != Capability.WS_PENDING_TX }.map {
             when (it) {
                 Capability.RPC -> BlockchainOuterClass.Capabilities.CAP_CALLS
                 Capability.BALANCE -> BlockchainOuterClass.Capabilities.CAP_BALANCE
                 Capability.WS_HEAD -> BlockchainOuterClass.Capabilities.CAP_WS_HEAD
+                else -> null
             }
         }
 

src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt

@@ -63,11 +63,12 @@ class Describe(
                             }
                         capabilities.addAll(chainUpstreams.getCapabilities())
                         chainDescription.addAllCapabilities(
-                            capabilities.map {
+                            capabilities.filter { it != Capability.WS_PENDING_TX }.map {
                                 when (it) {
                                     Capability.RPC -> BlockchainOuterClass.Capabilities.CAP_CALLS
                                     Capability.BALANCE -> BlockchainOuterClass.Capabilities.CAP_BALANCE
                                     Capability.WS_HEAD -> BlockchainOuterClass.Capabilities.CAP_WS_HEAD
+                                    else -> null
                                 }
                             },
                         )

src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt

@@ -52,7 +52,7 @@ class BitcoinUpstreamCreator(
                     fileResolver.resolve(ca).readBytes()
                 }
             }
-            EsploraClient(endpoint.url, endpoint.basicAuth, tls)
+            EsploraClient(endpoint.url, endpoint.basicAuth, tls, endpoint.bearerAuth)
         }
 
         val extractBlock = ExtractBlock()

src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt

@@ -84,6 +84,7 @@ open class GenericConnectorFactoryCreator(
                 monitoringCfg.nettyMetricsConfig.enabled,
                 httpScheduler,
                 customHeaders,
+                conn.bearerAuth,
             )
         }
     }
@@ -106,6 +107,7 @@ open class GenericConnectorFactoryCreator(
             ).apply {
                 config = endpoint
                 basicAuth = endpoint.basicAuth
+                bearerAuth = endpoint.bearerAuth
                 this.customHeaders = customHeaders
             }
             val wsApi = WsConnectionPoolFactory(

src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt

@@ -21,6 +21,7 @@ class BasicHttpFactory(
     private val nettyMetricsEnabled: Boolean,
     private val httpScheduler: Scheduler,
     private val customHeaders: Map<String, String> = emptyMap(),
+    private val bearerAuth: AuthConfig.ClientBearerAuth? = null,
 ) : HttpFactory {
     private val log = LoggerFactory.getLogger(this::class.java)
 
@@ -34,11 +35,14 @@ class BasicHttpFactory(
             Tag.of("chain", chain.chainCode),
         )
         val metrics = RequestMetrics(
-            Timer.builder("upstream.rpc.conn")
+            { method ->
-                .description("Request time through a HTTP JSON RPC connection")
+                Timer.builder("upstream.rpc.conn")
-                .tags(metricsTags)
+                    .description("Request time through a HTTP JSON RPC connection")
-                .publishPercentileHistogram()
+                    .tags(metricsTags)
-                .register(Metrics.globalRegistry),
+                    .tag("method", method ?: "unknown")
+                    .publishPercentileHistogram()
+                    .register(Metrics.globalRegistry)
+            },
             Counter.builder("upstream.rpc.fail")
                 .description("Number of failures of HTTP JSON RPC requests")
                 .tags(metricsTags)
@@ -47,8 +51,8 @@ class BasicHttpFactory(
         )
 
         if (chain.type.apiType == ApiType.REST) {
-            return RestHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, chain, basicAuth, tls, customHeaders)
+            return RestHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, chain, basicAuth, tls, customHeaders, bearerAuth)
         }
-        return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls, customHeaders)
+        return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls, customHeaders, bearerAuth)
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt

@@ -4,4 +4,5 @@ enum class Capability {
     RPC,
     BALANCE,
     WS_HEAD,
+    WS_PENDING_TX,
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt

@@ -28,6 +28,7 @@ abstract class HttpReader(
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
     customHeaders: Map<String, String> = emptyMap(),
+    bearerAuth: AuthConfig.ClientBearerAuth? = null,
 ) : ChainReader {
 
     constructor() : this("", 1500, 1000, null)
@@ -66,6 +67,13 @@ abstract class HttpReader(
             build = build.headers(headers)
         }
 
+        if (basicAuth == null) {
+            bearerAuth?.let { auth ->
+                val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, "Bearer ${auth.token}") }
+                build = build.headers(headers)
+            }
+        }
+
         if (customHeaders.isNotEmpty()) {
             val headers = Consumer { h: HttpHeaders ->
                 customHeaders.forEach { (key, value) ->
@@ -101,7 +109,7 @@ abstract class HttpReader(
 
     open fun onStop() {
         if (metrics != null) {
-            Metrics.globalRegistry.remove(metrics.timer)
+            metrics.registeredTimers().forEach { Metrics.globalRegistry.remove(it) }
             Metrics.globalRegistry.remove(metrics.fails)
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestMetrics.kt

@@ -17,9 +17,20 @@ package io.emeraldpay.dshackle.upstream
 
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Timer
+import java.util.concurrent.ConcurrentHashMap
 
 class RequestMetrics(
-    val timer: Timer,
+    private val timerFactory: (String?) -> Timer,
     val fails: Counter,
     val nettyMetricsEnabled: Boolean,
-)
+) {
+    private val timerCache = ConcurrentHashMap<String, Timer>()
+
+    constructor(timer: Timer, fails: Counter, nettyMetricsEnabled: Boolean) :
+        this({ _ -> timer }, fails, nettyMetricsEnabled)
+
+    fun timer(method: String? = null): Timer =
+        timerCache.computeIfAbsent(method ?: "") { timerFactory(method) }
+
+    fun registeredTimers(): Collection<Timer> = timerCache.values
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecific.kt

@@ -1,6 +1,7 @@
 package io.emeraldpay.dshackle.upstream.aztec
 
 import com.fasterxml.jackson.databind.JsonNode
+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
@@ -8,6 +9,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.foundation.ChainOptions.Options
 import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.upstream.ChainException
 import io.emeraldpay.dshackle.upstream.ChainRequest
 import io.emeraldpay.dshackle.upstream.GenericSingleCallValidator
 import io.emeraldpay.dshackle.upstream.SingleValidator
@@ -21,16 +23,36 @@ import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import java.math.BigInteger
+import java.time.Duration
 import java.time.Instant
 
 object AztecChainSpecific : AbstractPollChainSpecific() {
     private val log = LoggerFactory.getLogger(AztecChainSpecific::class.java)
 
-    // node_getL2Tips reshaped between Aztec versions:
+    private const val METHOD_NOT_FOUND = -32601
+
+    // Aztec v5 (v5.0.0-rc.1) renamed the tips RPC: node_getL2Tips -> node_getChainTips.
+    // Older nodes (incl. current mainnet) expose only the legacy name; v5+ nodes expose
+    // only the new one. We probe the legacy method first (most upstreams are still on it)
+    // and fall back to the new one on "method not found", remembering the working method
+    // per upstream so we stop probing the dead one on every poll.
+    private val LEGACY_TIPS_REQUEST = ChainRequest("node_getL2Tips", ListParams())
+    private val CHAIN_TIPS_REQUEST = ChainRequest("node_getChainTips", ListParams())
+
+    // Bounded so per-upstream entries can't accumulate without limit (e.g. across config
+    // reloads): a hard size cap plus idle expiry evict stale ids, and an evicted entry just
+    // costs one re-probe. Only upstreams that actually fall back take a slot — legacy-only
+    // ones keep using the default and never populate it.
+    private val workingTipsRequest = Caffeine.newBuilder()
+        .maximumSize(1024)
+        .expireAfterAccess(Duration.ofHours(1))
+        .build<String, ChainRequest>()
+
+    // The tips response reshaped between Aztec versions:
     //   v3 (and earlier): {proposed: {number, hash}, proven: {number, hash}, checkpointed: {number, hash}}
-    //   v4: proven/finalized/checkpointed each became {block: {number, hash}, checkpoint: {number, hash}}
+    //   v4/v5: proven/finalized/checkpointed each became {block: {number, hash}, checkpoint: {number, hash}}
-    // proposed stayed flat. We always look at the v4 nested path first and fall back
+    // proposed stayed flat across all versions. We look at the flat path first and fall back
-    // to the flat v3 path so an upstream on either version is parsed correctly.
+    // to the nested path so an upstream on any version is parsed correctly.
     private val PROPOSED_NUMBER = arrayOf("proposed.number", "proposed.block.number")
     private val PROPOSED_HASH = arrayOf("proposed.hash", "proposed.block.hash")
 
@@ -122,8 +144,43 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
         return AztecLowerBoundService(chain, upstream)
     }
 
-    override fun latestBlockRequest(): ChainRequest =
+    override fun latestBlockRequest(): ChainRequest = LEGACY_TIPS_REQUEST
-        ChainRequest("node_getL2Tips", ListParams())
+
+    // Try the per-upstream remembered method (legacy by default); on "method not found"
+    // fall back to the other one and remember whichever succeeds, so subsequent polls go
+    // straight to the working method. Any other error propagates as before.
+    override fun getLatestBlock(api: ChainReader, upstreamId: String): Mono<BlockContainer> {
+        val preferred = workingTipsRequest.getIfPresent(upstreamId) ?: LEGACY_TIPS_REQUEST
+        val fallback = if (preferred === LEGACY_TIPS_REQUEST) CHAIN_TIPS_REQUEST else LEGACY_TIPS_REQUEST
+        return fetchTips(api, upstreamId, preferred)
+            .onErrorResume { err ->
+                if (isMethodNotFound(err)) {
+                    log.info(
+                        "Aztec upstream {} does not support {}, falling back to {}",
+                        upstreamId,
+                        preferred.method,
+                        fallback.method,
+                    )
+                    fetchTips(api, upstreamId, fallback)
+                        .doOnNext { workingTipsRequest.put(upstreamId, fallback) }
+                } else {
+                    Mono.error(err)
+                }
+            }
+    }
+
+    private fun fetchTips(api: ChainReader, upstreamId: String, request: ChainRequest): Mono<BlockContainer> {
+        return api.read(request).flatMap {
+            parseBlock(it.getResult(), upstreamId, api)
+        }
+    }
+
+    private fun isMethodNotFound(err: Throwable): Boolean {
+        if (err is ChainException && err.error.code == METHOD_NOT_FOUND) {
+            return true
+        }
+        return err.message?.contains("method not found", ignoreCase = true) ?: false
+    }
 
     override fun upstreamSettingsDetector(
         chain: Chain,

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt

@@ -34,10 +34,11 @@ import java.security.cert.X509Certificate
 import java.util.Base64
 import java.util.function.Consumer
 
-class EsploraClient(
+class EsploraClient @JvmOverloads constructor(
     private val url: URI,
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
+    bearerAuth: AuthConfig.ClientBearerAuth? = null,
 ) {
 
     companion object {
@@ -62,6 +63,13 @@ class EsploraClient(
             build = build.headers(headers)
         }
 
+        if (basicAuth == null) {
+            bearerAuth?.let { auth ->
+                val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, "Bearer ${auth.token}") }
+                build = build.headers(headers)
+            }
+        }
+
         tlsCAAuth?.let { auth ->
             val cf = CertificateFactory.getInstance("X.509")
             val cert = cf.generateCertificate(ByteArrayInputStream(auth)) as X509Certificate

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultAztecMethods.kt

@@ -16,18 +16,24 @@ class DefaultAztecMethods : CallMethods {
         "node_getBlockNumber",
         "node_getProvenBlockNumber",
         "node_getL2Tips",
+        "node_getChainTips",
         "node_getBlock",
         "node_getBlocks",
+        "node_getBlockData",
         "node_getBlockHeader",
         "node_getBlockByArchive",
         "node_getBlockByHash",
         "node_getBlockHeaderByArchive",
 
-        // Checkpoints
+        // Checkpoints / consensus
         "node_getCheckpointNumber",
+        "node_getCheckpoint",
         "node_getCheckpointedBlockNumber",
         "node_getCheckpointedBlocks",
         "node_getCheckpoints",
+        "node_getCheckpointsData",
+        "node_getCheckpointAttestationsForSlot",
+        "node_getProposalsForSlot",
 
         // Transactions
         "node_sendTx",
@@ -45,6 +51,11 @@ class DefaultAztecMethods : CallMethods {
         "node_getWorldStateSyncStatus",
         "node_findLeavesIndexes",
 
+        // Sync status
+        "node_getSyncedL1Timestamp",
+        "node_getSyncedL2EpochNumber",
+        "node_getSyncedL2SlotNumber",
+
         // Sibling paths
         "node_getNullifierSiblingPath",
         "node_getNoteHashSiblingPath",
@@ -65,12 +76,14 @@ class DefaultAztecMethods : CallMethods {
         "node_getL1ToL2MessageCheckpoint",
         "node_isL1ToL2MessageSynced",
         "node_getL2ToL1Messages",
+        "node_getL2ToL1MembershipWitness",
 
         // Logs
         "node_getPrivateLogs",
         "node_getPrivateLogsByTags",
         "node_getPublicLogs",
         "node_getPublicLogsByTagsFromContract",
+        "node_getPublicLogsByTags",
         "node_getContractClassLogs",
         "node_getLogsByTags",
 
@@ -86,12 +99,14 @@ class DefaultAztecMethods : CallMethods {
         "node_getVersion",
         "node_getChainId",
         "node_getL1ContractAddresses",
+        "node_getL1Constants",
         "node_getProtocolContractAddresses",
         "node_getEncodedEnr",
 
         // Fees
         "node_getCurrentBaseFees",
         "node_getCurrentMinFees",
+        "node_getPredictedMinFees",
         "node_getMaxPriorityFees",
 
         // Validators

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt

@@ -11,6 +11,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
 import io.emeraldpay.dshackle.upstream.ethereum.hex.Hex32
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
 import org.slf4j.LoggerFactory
@@ -72,7 +73,7 @@ open class EthereumEgressSubscription(
         } else {
             listOf()
         }
-        return if (pendingTxesSource != null) {
+        return if (pendingTxesSource != null && pendingTxesSource !is NoPendingTxes && upstream.getCapabilities().contains(Capability.WS_PENDING_TX)) {
             subs.plus(listOf(METHOD_PENDING_TXES, METHOD_DRPC_PENDING_TXES))
         } else {
             subs

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSettingsDetector.kt

@@ -8,6 +8,7 @@ import io.emeraldpay.dshackle.upstream.ChainRequest
 import io.emeraldpay.dshackle.upstream.ChainResponse
 import io.emeraldpay.dshackle.upstream.NodeTypeRequest
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -15,7 +16,6 @@ import java.util.concurrent.atomic.AtomicInteger
 
 const val ZERO_ADDRESS = "0x0000000000000000000000000000000000000000"
 const val HL_NATIVE_TX_FROM_MAINNET = "0x2222222222222222222222222222222222222222"
-const val HL_NATIVE_TX_FROM_TESTNET = "0x6ed35e7d6de4b45f4efb8a91eff31afa49362569"
 
 class EthereumUpstreamSettingsDetector(
     private val _upstream: Upstream,
@@ -179,15 +179,17 @@ class EthereumUpstreamSettingsDetector(
         Some clients on hyperliquid don't include system topup transactions, set either one of labels
      */
     private fun detectHlNativeTx(): Flux<Pair<String, String>> {
-        // Only run HL native tx detection on Hyperliquid chains
         if (chain != Chain.HYPERLIQUID__MAINNET && chain != Chain.HYPERLIQUID__TESTNET) {
             return Flux.empty()
         }
-        val hlNativeTxFrom = when (chain) {
+        // prefer the explicit ?hl= flag from the upstream URL
-            Chain.HYPERLIQUID__MAINNET -> HL_NATIVE_TX_FROM_MAINNET
+        hlNativeTxLabelsFromUrl()?.let { return Flux.fromIterable(it) }
-            Chain.HYPERLIQUID__TESTNET -> HL_NATIVE_TX_FROM_TESTNET
+
-            else -> return Flux.empty()
+        // no ?hl= flag: block-scan fallback (reliable only on mainnet)
+        if (chain != Chain.HYPERLIQUID__MAINNET) {
+            return Flux.empty()
         }
+        val hlNativeTxFrom = HL_NATIVE_TX_FROM_MAINNET
         if (detectCounter.get() % 5 != 1) {
             return Flux.empty() // reduce frequency of detection
         }
@@ -255,6 +257,19 @@ class EthereumUpstreamSettingsDetector(
         }
     }
 
+    // maps the upstream URL's ?hl= flag to routing labels, or null if absent
+    private fun hlNativeTxLabelsFromUrl(): List<Pair<String, String>>? {
+        val url = (upstream as? GenericUpstream)?.getRpcConnectionUrl()?.toString() ?: return null
+        // ?hl=false => serves native txs (include); ?hl=true => hl-node compliant (exclude)
+        return when {
+            url.contains(Regex("[?&]hl=false\\b")) ->
+                listOf("include_hl_native_tx" to "true", "exclude_hl_native_tx" to "false")
+            url.contains(Regex("[?&]hl=true\\b")) ->
+                listOf("exclude_hl_native_tx" to "true", "include_hl_native_tx" to "false")
+            else -> null
+        }
+    }
+
     private fun detectArchiveNode(notArchived: Boolean): Mono<Pair<String, String>> {
         if (notArchived) {
             return Mono.empty()

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidator.kt

@@ -0,0 +1,66 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.upstream.ChainRequest
+import io.emeraldpay.dshackle.upstream.ChainResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.time.Duration
+
+const val BASE_TX_LIMIT = 1000L
+
+interface PendingTransactionValidator {
+    fun pendingTxExists(): Flux<Boolean>
+}
+
+class NoopPendingTransactionValidator : PendingTransactionValidator {
+    override fun pendingTxExists(): Flux<Boolean> {
+        return Flux.just(true)
+    }
+}
+
+class PendingTransactionValidatorImpl(
+    private val upstreamId: String,
+    private val directReader: ChainReader,
+    private val interval: Duration,
+    private val txLimit: Long,
+) : PendingTransactionValidator {
+    private val log = LoggerFactory.getLogger(this::class.java)
+
+    override fun pendingTxExists(): Flux<Boolean> {
+        return Flux.interval(
+            Duration.ofSeconds(15),
+            interval,
+        )
+            .flatMap {
+                directReader.read(ChainRequest("txpool_content", ListParams()))
+                    .flatMap(ChainResponse::requireResult)
+                    .map {
+                        val node = Global.objectMapper.readTree(it)
+                        val pendingTxsNode = node.get("pending")
+                        val queuedTxsNode = node.get("queued")
+
+                        val pendingTxsCount = if (pendingTxsNode != null) {
+                            pendingTxsNode.fieldNames().asSequence().toList().size
+                        } else {
+                            0
+                        }
+                        val queuedTxsCount = if (queuedTxsNode != null) {
+                            queuedTxsNode.fieldNames().asSequence().toList().size
+                        } else {
+                            0
+                        }
+
+                        ((pendingTxsCount + queuedTxsCount) >= txLimit)
+                    }
+                    .timeout(Duration.ofSeconds(30))
+                    .onErrorResume {
+                        log.error("unable to read txs from txpool_content of upstream {}", upstreamId, it)
+                        Mono.just(false)
+                    }
+            }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt

@@ -21,6 +21,7 @@ open class WsConnectionFactory(
 ) {
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
+    var bearerAuth: AuthConfig.ClientBearerAuth? = null
     var config: UpstreamsConfig.WsEndpoint? = null
     var customHeaders: Map<String, String> = emptyMap()
 
@@ -47,7 +48,7 @@ open class WsConnectionFactory(
     }
 
     open fun createWsConnection(connIndex: Int = 0): WsConnection =
-        WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), scheduler, eventsScheduler, customHeaders).also { ws ->
+        WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), scheduler, eventsScheduler, customHeaders, bearerAuth).also { ws ->
             config?.frameSize?.let {
                 ws.frameSize = it
             }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt

@@ -67,6 +67,7 @@ open class WsConnectionImpl(
     private val scheduler: Scheduler,
     private val eventsScheduler: Scheduler,
     private val customHeaders: Map<String, String> = emptyMap(),
+    private val bearerAuth: AuthConfig.ClientBearerAuth? = null,
 ) : AutoCloseable, WsConnection, Cloneable {
 
     companion object {
@@ -227,6 +228,11 @@ open class WsConnectionImpl(
                     val base64password = Base64.getEncoder().encodeToString(tmp.toByteArray())
                     headers.add(HttpHeaderNames.AUTHORIZATION, "Basic $base64password")
                 }
+                if (basicAuth == null) {
+                    bearerAuth?.let { auth ->
+                        headers.add(HttpHeaderNames.AUTHORIZATION, "Bearer ${auth.token}")
+                    }
+                }
                 customHeaders.forEach { (key, value) ->
                     headers.add(key, value)
                 }
@@ -404,7 +410,7 @@ open class WsConnectionImpl(
         return Mono.from(onResponse.asMono()).or(failOnDisconnect)
             .doOnSubscribe { sendRpc(request) }
             .take(Defaults.timeout)
-            .doOnNext { requestMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) }
+            .doOnNext { requestMetrics?.timer()?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) }
             .doOnError { requestMetrics?.fails?.increment() }
             .map { it.copyWithId(ChainResponse.Id.from(originalId)) }
             .switchIfEmpty(
@@ -424,7 +430,7 @@ open class WsConnectionImpl(
             it.close()
             Metrics.globalRegistry.remove(it)
         }
-        requestMetrics?.timer?.let {
+        requestMetrics?.registeredTimers()?.forEach {
             it.close()
             Metrics.globalRegistry.remove(it)
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt

@@ -43,6 +43,7 @@ import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Sinks
 import reactor.core.scheduler.Schedulers
+import java.net.URI
 import java.time.Duration
 import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicBoolean
@@ -101,6 +102,8 @@ open class GenericUpstream(
     ) {
         rpcMethodsDetector = upstreamRpcMethodsDetectorBuilder(this, config)
         detectRpcMethods(config, buildMethods)
+        rpcConnectionUrl = (config.connection as? UpstreamsConfig.RpcConnection)
+            ?.let { it.rpc?.url ?: it.ws?.url }
     }
 
     private val validator: UpstreamValidator? = validatorBuilder(chain, this, getOptions(), chainConfig, versionRules)
@@ -110,6 +113,8 @@ open class GenericUpstream(
     private val settingsDetectorSubscription = AtomicReference<Disposable?>()
 
     private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(getOptions().disableLivenessSubscriptionValidation)
+    private val hasPendingTxs = AtomicBoolean(getOptions().disablePendingTxValidation)
+
     protected val connector: GenericConnector = connectorFactory.create(this, chain)
         .also { upConnector ->
             Gauge.builder("upstream_head", upConnector.getHead()) {
@@ -120,9 +125,13 @@ open class GenericUpstream(
                 .register(Metrics.globalRegistry)
         }
     private val livenessSubscription = AtomicReference<Disposable?>()
+    private val pendingTxSubscription = AtomicReference<Disposable?>()
     private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this)
     private var rpcMethodsDetector: UpstreamRpcMethodsDetector? = null
 
+    // configured RPC/WS URL (carries query flags like ?hl=)
+    private var rpcConnectionUrl: URI? = null
+
     private val lowerBoundService = lowerBoundServiceBuilder(chain, this)
 
     private val started = AtomicBoolean(false)
@@ -150,11 +159,14 @@ open class GenericUpstream(
 
     // outdated, looks like applicable only for bitcoin and our ws_head trick
     override fun getCapabilities(): Set<Capability> {
-        return if (hasLiveSubscriptionHead.get()) {
+        val caps = mutableSetOf(Capability.RPC, Capability.BALANCE)
-            setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD)
+        if (hasLiveSubscriptionHead.get()) {
-        } else {
+            caps.add(Capability.WS_HEAD)
-            setOf(Capability.RPC, Capability.BALANCE)
         }
+        if (hasPendingTxs.get()) {
+            caps.add(Capability.WS_PENDING_TX)
+        }
+        return caps
     }
 
     override fun isGrpc(): Boolean {
@@ -178,6 +190,8 @@ open class GenericUpstream(
         )
     }
 
+    fun getRpcConnectionUrl(): URI? = rpcConnectionUrl
+
     @Suppress("UNCHECKED_CAST")
     override fun <T : Upstream> cast(selfType: Class<T>): T {
         if (!selfType.isAssignableFrom(this.javaClass)) {
@@ -358,6 +372,14 @@ open class GenericUpstream(
                 ),
             )
         }
+        if (!getOptions().disablePendingTxValidation) {
+            pendingTxSubscription.set(
+                connector.pendingTxEvents().subscribe {
+                    hasPendingTxs.set(it)
+                    sendUpstreamStateEvent(UPDATED)
+                },
+            )
+        }
         detectSettings()
 
         if (!getOptions().disableBoundValidation) {
@@ -380,6 +402,7 @@ open class GenericUpstream(
         lowerBlockDetectorSubscription.getAndSet(null)?.dispose()
         finalizationDetectorSubscription.getAndSet(null)?.dispose()
         settingsDetectorSubscription.getAndSet(null)?.dispose()
+        pendingTxSubscription.getAndSet(null)?.dispose()
         connector.getHead().stop()
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt

@@ -15,4 +15,6 @@ interface GenericConnector : Lifecycle {
     fun getIngressReader(): ChainReader
 
     fun getIngressSubscription(): IngressSubscription
+
+    fun pendingTxEvents(): Flux<Boolean>
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt

@@ -13,11 +13,15 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.NoIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.AlwaysHeadLivenessValidator
+import io.emeraldpay.dshackle.upstream.ethereum.BASE_TX_LIMIT
 import io.emeraldpay.dshackle.upstream.ethereum.GenericWsHead
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidatorImpl
 import io.emeraldpay.dshackle.upstream.ethereum.NoHeadLivenessValidator
+import io.emeraldpay.dshackle.upstream.ethereum.NoopPendingTransactionValidator
+import io.emeraldpay.dshackle.upstream.ethereum.PendingTransactionValidator
+import io.emeraldpay.dshackle.upstream.ethereum.PendingTransactionValidatorImpl
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
@@ -58,6 +62,7 @@ class GenericRpcConnector(
     private val head: Head
     private val liveness: HeadLivenessValidator
     private val jsonRpcWsClient: JsonRpcWsClient?
+    private val pendingTxValidator: PendingTransactionValidator
 
     companion object {
         private val log = LoggerFactory.getLogger(GenericRpcConnector::class.java)
@@ -136,6 +141,16 @@ class GenericRpcConnector(
                 )
             }
         }
+        pendingTxValidator = if (chain == Chain.BASE__MAINNET) {
+            PendingTransactionValidatorImpl(
+                upstream.getId(),
+                getIngressReader(),
+                Duration.ofMinutes(5),
+                BASE_TX_LIMIT,
+            )
+        } else {
+            NoopPendingTransactionValidator()
+        }
 
         liveness = if (connectorType != RPC_ONLY && isSpecialChain(chain)) {
             AlwaysHeadLivenessValidator()
@@ -186,6 +201,10 @@ class GenericRpcConnector(
         return ingressSubscription ?: NoIngressSubscription()
     }
 
+    override fun pendingTxEvents(): Flux<Boolean> {
+        return pendingTxValidator.pendingTxExists()
+    }
+
     override fun getHead(): Head {
         return head
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt

@@ -6,9 +6,13 @@ import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.IngressSubscription
+import io.emeraldpay.dshackle.upstream.ethereum.BASE_TX_LIMIT
 import io.emeraldpay.dshackle.upstream.ethereum.GenericWsHead
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidatorImpl
+import io.emeraldpay.dshackle.upstream.ethereum.NoopPendingTransactionValidator
+import io.emeraldpay.dshackle.upstream.ethereum.PendingTransactionValidator
+import io.emeraldpay.dshackle.upstream.ethereum.PendingTransactionValidatorImpl
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl
@@ -36,6 +40,8 @@ class GenericWsConnector(
     private val head: GenericWsHead
     private val subscriptions: IngressSubscription
     private val liveness: HeadLivenessValidatorImpl
+    private val pendingTxValidator: PendingTransactionValidator
+
     init {
         pool = wsFactory.create(upstream)
         reader = JsonRpcWsClient(pool)
@@ -54,6 +60,17 @@ class GenericWsConnector(
         )
         liveness = HeadLivenessValidatorImpl(head, expectedBlockTime, headLivenessScheduler, upstream.getId())
         subscriptions = chainSpecific.makeIngressSubscription(chain, wsSubscriptions)
+
+        pendingTxValidator = if (chain == Chain.BASE__MAINNET) {
+            PendingTransactionValidatorImpl(
+                upstream.getId(),
+                getIngressReader(),
+                Duration.ofMinutes(10),
+                BASE_TX_LIMIT,
+            )
+        } else {
+            NoopPendingTransactionValidator()
+        }
     }
 
     override fun headLivenessEvents(): Flux<HeadLivenessState> {
@@ -81,6 +98,10 @@ class GenericWsConnector(
         return subscriptions
     }
 
+    override fun pendingTxEvents(): Flux<Boolean> {
+        return pendingTxValidator.pendingTxExists()
+    }
+
     override fun getHead(): Head {
         return head
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt

@@ -36,7 +36,8 @@ class RestHttpReader(
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
     customHeaders: Map<String, String> = emptyMap(),
-) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth, customHeaders) {
+    bearerAuth: AuthConfig.ClientBearerAuth? = null,
+) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth, customHeaders, bearerAuth) {
 
     private val parser = ResponseRpcParser()
     private val requestParser = RestRequestParser
@@ -59,7 +60,7 @@ class RestHttpReader(
             .flatMap(this::execute)
             .doOnNext {
                 if (startTime.isStarted) {
-                    metrics?.timer?.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
+                    metrics?.timer(key.method)?.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
                 }
             }
             .handle { it, sink ->

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt

@@ -114,7 +114,7 @@ class JsonRpcGrpcClient(
                 }
                 .doOnNext {
                     if (timer.isStarted) {
-                        metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
+                        metrics?.timer()?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
                     }
                 }
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt

@@ -36,7 +36,7 @@ import java.util.function.Function
 /**
  * JSON RPC client
  */
-class JsonRpcHttpReader(
+class JsonRpcHttpReader @JvmOverloads constructor(
     target: String,
     maxConnections: Int,
     queueSize: Int,
@@ -45,7 +45,8 @@ class JsonRpcHttpReader(
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null,
     customHeaders: Map<String, String> = emptyMap(),
-) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth, customHeaders) {
+    bearerAuth: AuthConfig.ClientBearerAuth? = null,
+) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth, customHeaders, bearerAuth) {
 
     private val parser = ResponseRpcParser()
     private val streamParser = JsonRpcStreamParser()
@@ -92,7 +93,7 @@ class JsonRpcHttpReader(
             .flatMap(this@JsonRpcHttpReader::execute)
             .doOnNext {
                 if (startTime.isStarted) {
-                    metrics?.timer?.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
+                    metrics?.timer(key.method)?.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
                 }
             }
             .transform(asJsonRpcResponse(key))

src/test/groovy/io/emeraldpay/dshackle/config/AuthConfigReaderSpec.groovy

@@ -35,6 +35,29 @@ class AuthConfigReaderSpec extends Specification {
         act.password == "258fe4149c199ad8f2811a68f20154fc"
     }
 
+    def "Read bearer-auth for client"() {
+        setup:
+        def yaml =
+                "bearer-auth:\n" +
+                        "  token: 9c199ad8f281f20154fc258fe41a6814"
+        when:
+        def act = reader.readClientBearerAuth(reader.readNode(yaml))
+        then:
+        act != null
+        act.token == "9c199ad8f281f20154fc258fe41a6814"
+    }
+
+    def "Read bearer-auth without token as null"() {
+        setup:
+        def yaml =
+                "bearer-auth:\n" +
+                        "  something: else"
+        when:
+        def act = reader.readClientBearerAuth(reader.readNode(yaml))
+        then:
+        act == null
+    }
+
     def "Read tls for client"() {
         setup:
         def yaml =

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -75,6 +75,38 @@ class UpstreamsConfigReaderSpec extends Specification {
         }
     }
 
+    def "Parse config with bearer auth"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-bearer-auth.yaml")
+        when:
+        def act = reader.readInternal(config)
+        then:
+        act != null
+        act.upstreams.size() == 2
+        with(act.upstreams.get(0)) {
+            id == "local"
+            connection instanceof UpstreamsConfig.RpcConnection
+            with((UpstreamsConfig.RpcConnection) connection) {
+                rpc.basicAuth == null
+                rpc.bearerAuth != null
+                rpc.bearerAuth.token == "9c199ad8f281f20154fc258fe41a6814"
+                ws != null
+                ws.basicAuth == null
+                ws.bearerAuth != null
+                ws.bearerAuth.token == "258fe4149c199ad8f2811a68f20154fc"
+            }
+        }
+        with(act.upstreams.get(1)) {
+            id == "both-auth"
+            connection instanceof UpstreamsConfig.RpcConnection
+            with((UpstreamsConfig.RpcConnection) connection) {
+                // when both are configured basic-auth wins and bearer-auth is dropped
+                rpc.basicAuth != null
+                rpc.bearerAuth == null
+            }
+        }
+    }
+
     def "Parse websocket-only config"() {
         setup:
         def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-ws-only.yaml")
@@ -636,7 +668,7 @@ class UpstreamsConfigReaderSpec extends Specification {
         def options = partialOptions.buildOptions()
         then:
         options == new ChainOptions.Options(
-                false, false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true, true, 1_000_000, false, false, true, false
+                false, false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true, true, 1_000_000, false, false, true, false, false
         )
     }
 }

src/test/groovy/io/emeraldpay/dshackle/test/GenericConnectorMock.groovy

@@ -14,11 +14,13 @@ class GenericConnectorMock implements GenericConnector {
     Reader<ChainRequest, ChainResponse> api
     Head head
     Flux<HeadLivenessState> liveness
+    Flux<Boolean> pendingTxs
 
     GenericConnectorMock(Reader<ChainRequest, ChainResponse> api, Head head) {
         this.api = api
         this.head = head
         this.liveness = Flux.just(HeadLivenessState.NON_CONSECUTIVE)
+        this.pendingTxs = Flux.just(false)
     }
 
     @Override
@@ -51,4 +53,9 @@ class GenericConnectorMock implements GenericConnector {
     IngressSubscription getIngressSubscription() {
         return NoEthereumIngressSubscription.DEFAULT
     }
+
+    @Override
+    Flux<Boolean> pendingTxEvents() {
+        return pendingTxs
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy

@@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.upstream.ethereum
 
 
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
 import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
@@ -216,6 +218,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
         when:
         def up3 = TestingCommons.upstream("test")
         up3.getConnectorMock().setLiveness(Flux.just(HeadLivenessState.OK))
+        up3.getConnectorMock().setPendingTxs(Flux.just(true))
         up3.stop()
         up3.start()
         def ethereumSubscribe3 = new EthereumEgressSubscription(TestingCommons.multistream(up3) as GenericMultistream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
@@ -230,5 +233,23 @@ class EthereumEgressSubscriptionSpec extends Specification {
         then:
         ethereumSubscribe4.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_NEW_HEADS].toSet()
 
+        when:
+        def up5 = TestingCommons.upstream("test")
+        up5.getConnectorMock().setLiveness(Flux.just(HeadLivenessState.OK))
+        up5.stop()
+        up5.start()
+        def ethereumSubscribe5 = new EthereumEgressSubscription(TestingCommons.multistream(up5) as GenericMultistream, Schedulers.boundedElastic(), Stub(PendingTxesSource))
+        then:
+        ethereumSubscribe5.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS].toSet()
+
+        when:
+        def up6 = TestingCommons.upstream("test")
+        up6.getConnectorMock().setLiveness(Flux.just(HeadLivenessState.OK))
+        up6.getConnectorMock().setPendingTxs(Flux.just(true))
+        up6.stop()
+        up6.start()
+        def ethereumSubscribe6 = new EthereumEgressSubscription(TestingCommons.multistream(up6) as GenericMultistream, Schedulers.boundedElastic(), new NoPendingTxes())
+        then:
+        ethereumSubscribe6.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS].toSet()
     }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
 
+import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.ChainException
 import io.emeraldpay.dshackle.upstream.ChainRequest
@@ -69,6 +70,28 @@ class JsonRpcHttpReaderSpec extends Specification {
         new String(act.result) == '"0x98de45"'
     }
 
+    def "Make a request with bearer auth"() {
+        setup:
+        def bearerAuth = new AuthConfig.ClientBearerAuth("test-token-123")
+        JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${mockServer.port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:], bearerAuth)
+        def resp = '{' +
+                '  "jsonrpc": "2.0",' +
+                '  "result": "0x98de45",' +
+                '  "error": null,' +
+                '  "id": 15' +
+                '}'
+        mockServer.when(
+                HttpRequest.request().withHeader("authorization", "Bearer test-token-123")
+        ).respond(
+                HttpResponse.response(resp)
+        )
+        when:
+        def act = client.read(new ChainRequest("test", new ListParams())).block(Duration.ofSeconds(5))
+        then:
+        act.error == null
+        new String(act.result) == '"0x98de45"'
+    }
+
     def "Produces RPC Exception on error status code"() {
         setup:
         def client = new JsonRpcHttpReader("localhost:${mockServer.port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:])

src/test/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecificTest.kt

@@ -0,0 +1,146 @@
+package io.emeraldpay.dshackle.upstream.aztec
+
+import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.upstream.ChainCallError
+import io.emeraldpay.dshackle.upstream.ChainCallUpstreamException
+import io.emeraldpay.dshackle.upstream.ChainRequest
+import io.emeraldpay.dshackle.upstream.ChainResponse
+import org.assertj.core.api.Assertions
+import org.junit.jupiter.api.Test
+import reactor.core.publisher.Mono
+import java.util.concurrent.atomic.AtomicInteger
+
+// v5 (v5.0.0-rc.1) node_getChainTips response: proposed stays flat, the rest nested.
+private val chainTipsResponse = """
+    {
+        "proposed": {"number": 12345, "hash": "0xaaaa"},
+        "checkpointed": {"block": {"number": 12340, "hash": "0xbbbb"}, "checkpoint": {"number": 100, "hash": "0x1111"}},
+        "proven": {"block": {"number": 12330, "hash": "0xcccc"}, "checkpoint": {"number": 99, "hash": "0x2222"}},
+        "finalized": {"block": {"number": 12320, "hash": "0xdddd"}, "checkpoint": {"number": 98, "hash": "0x3333"}}
+    }
+""".trimIndent()
+
+// legacy node_getL2Tips response (v3-era flat proposed)
+private val l2TipsResponse = """
+    {
+        "proposed": {"number": 999, "hash": "0x0999"},
+        "proven": {"number": 990, "hash": "0x0990"},
+        "checkpointed": {"number": 980, "hash": "0x0980"}
+    }
+""".trimIndent()
+
+// defensive: some versions nest proposed under .block
+private val nestedProposedResponse = """
+    {
+        "proposed": {"block": {"number": 777, "hash": "0x0777"}}
+    }
+""".trimIndent()
+
+private fun methodNotFound(method: String) =
+    Mono.error<ChainResponse>(
+        ChainCallUpstreamException(
+            ChainResponse.NumberId(1),
+            ChainCallError(-32601, "Method not found: $method"),
+        ),
+    )
+
+class AztecChainSpecificTest {
+
+    @Test
+    fun latestBlockRequestUsesL2Tips() {
+        Assertions.assertThat(AztecChainSpecific.latestBlockRequest().method)
+            .isEqualTo("node_getL2Tips")
+    }
+
+    @Test
+    fun parseBlockReadsFlatProposed() {
+        val result = AztecChainSpecific.parseBlock(
+            chainTipsResponse.toByteArray(),
+            "up-flat",
+            noopReader(),
+        ).block()!!
+
+        Assertions.assertThat(result.height).isEqualTo(12345L)
+        Assertions.assertThat(result.hash.toHex()).contains("aaaa")
+    }
+
+    @Test
+    fun parseBlockReadsNestedProposed() {
+        val result = AztecChainSpecific.parseBlock(
+            nestedProposedResponse.toByteArray(),
+            "up-nested",
+            noopReader(),
+        ).block()!!
+
+        Assertions.assertThat(result.height).isEqualTo(777L)
+        Assertions.assertThat(result.hash.toHex()).contains("0777")
+    }
+
+    @Test
+    fun getLatestBlockUsesL2TipsWhenAvailable() {
+        val calls = mutableListOf<String>()
+        val reader = object : ChainReader {
+            override fun read(key: ChainRequest): Mono<ChainResponse> {
+                calls += key.method
+                return Mono.just(ChainResponse(l2TipsResponse.toByteArray(), null))
+            }
+        }
+
+        val result = AztecChainSpecific.getLatestBlock(reader, "up-legacy").block()!!
+
+        Assertions.assertThat(result.height).isEqualTo(999L)
+        Assertions.assertThat(calls).containsExactly("node_getL2Tips")
+    }
+
+    @Test
+    fun getLatestBlockFallsBackToChainTipsAndCaches() {
+        val calls = mutableListOf<String>()
+        val reader = object : ChainReader {
+            override fun read(key: ChainRequest): Mono<ChainResponse> {
+                calls += key.method
+                return when (key.method) {
+                    "node_getL2Tips" -> methodNotFound("node_getL2Tips")
+                    "node_getChainTips" -> Mono.just(ChainResponse(chainTipsResponse.toByteArray(), null))
+                    else -> Mono.error(IllegalStateException("unexpected ${key.method}"))
+                }
+            }
+        }
+
+        // first poll: probes legacy, falls back to v5
+        val first = AztecChainSpecific.getLatestBlock(reader, "up-v5").block()!!
+        Assertions.assertThat(first.height).isEqualTo(12345L)
+        Assertions.assertThat(calls).containsExactly("node_getL2Tips", "node_getChainTips")
+
+        // second poll: must hit the cached working method directly, no dead probe
+        calls.clear()
+        val second = AztecChainSpecific.getLatestBlock(reader, "up-v5").block()!!
+        Assertions.assertThat(second.height).isEqualTo(12345L)
+        Assertions.assertThat(calls).containsExactly("node_getChainTips")
+    }
+
+    @Test
+    fun getLatestBlockDoesNotFallBackOnOtherErrors() {
+        val attempts = AtomicInteger(0)
+        val reader = object : ChainReader {
+            override fun read(key: ChainRequest): Mono<ChainResponse> {
+                attempts.incrementAndGet()
+                return Mono.error(
+                    ChainCallUpstreamException(
+                        ChainResponse.NumberId(1),
+                        ChainCallError(-32000, "internal error"),
+                    ),
+                )
+            }
+        }
+
+        val thrown = runCatching { AztecChainSpecific.getLatestBlock(reader, "up-err").block() }
+        Assertions.assertThat(thrown.isFailure).isTrue()
+        // only the primary method is attempted; no fallback probe on a non-method-not-found error
+        Assertions.assertThat(attempts.get()).isEqualTo(1)
+    }
+
+    private fun noopReader() = object : ChainReader {
+        override fun read(key: ChainRequest): Mono<ChainResponse> =
+            Mono.error(IllegalStateException("not expected"))
+    }
+}

src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidatorTest.kt

@@ -0,0 +1,211 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.upstream.ChainCallError
+import io.emeraldpay.dshackle.upstream.ChainRequest
+import io.emeraldpay.dshackle.upstream.ChainResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import java.time.Duration
+
+class PendingTransactionValidatorTest {
+
+    @Test
+    fun `noop validator always emits true`() {
+        val validator = NoopPendingTransactionValidator()
+
+        StepVerifier.create(validator.pendingTxExists())
+            .expectNext(true)
+            .expectComplete()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `emits true when pending plus queued exceeds limit`() {
+        val reader = mockReader(
+            response = txpoolContent(pendingAddresses = 6, queuedAddresses = 0),
+        )
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(true)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `emits true when pending plus queued is at limit`() {
+        val reader = mockReader(
+            response = txpoolContent(pendingAddresses = 3, queuedAddresses = 2),
+        )
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(true)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `counts both pending and queued buckets`() {
+        // 3 pending + 4 queued = 7 > limit of 5
+        val reader = mockReader(
+            response = txpoolContent(pendingAddresses = 3, queuedAddresses = 4),
+        )
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(true)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `handles missing pending field`() {
+        val reader = mockReader(response = """{"queued": {"0xaa": {}, "0xbb": {}}}""")
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        // Only 2 queued (no pending node) -> below limit -> false
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(false)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `handles missing queued field`() {
+        val reader = mockReader(response = """{"pending": {"0xaa": {}, "0xbb": {}, "0xcc": {}}}""")
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 2,
+        )
+
+        // 3 pending (no queued node) > limit 2 -> true
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(true)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `emits false on rpc error`() {
+        val reader = mock<ChainReader> {
+            on { read(ChainRequest("txpool_content", ListParams())) } doReturn
+                Mono.just(ChainResponse(null, ChainCallError(-32000, "method not supported")))
+        }
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(false)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `continues polling after an error response`() {
+        val errorResponse = Mono.just(ChainResponse(null, ChainCallError(-32000, "method not supported")))
+        val okResponse = Mono.just(
+            ChainResponse(
+                txpoolContent(pendingAddresses = 10, queuedAddresses = 0).toByteArray(),
+                null,
+            ),
+        )
+        val reader = mock<ChainReader> {
+            on { read(ChainRequest("txpool_content", ListParams())) } doReturn errorResponse doReturn okResponse
+        }
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(false) // first poll: error -> false
+            .expectNoEvent(Duration.ofSeconds(30))
+            .expectNext(true) // second poll: success
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    @Test
+    fun `polls at configured interval`() {
+        val reader = mockReader(
+            response = txpoolContent(pendingAddresses = 10, queuedAddresses = 0),
+        )
+        val validator = PendingTransactionValidatorImpl(
+            upstreamId = "test-upstream",
+            directReader = reader,
+            interval = Duration.ofSeconds(30),
+            txLimit = 5,
+        )
+
+        StepVerifier.withVirtualTime { validator.pendingTxExists() }
+            .expectSubscription()
+            .expectNoEvent(Duration.ofSeconds(15))
+            .expectNext(true)
+            .expectNoEvent(Duration.ofSeconds(30))
+            .expectNext(true)
+            .expectNoEvent(Duration.ofSeconds(30))
+            .expectNext(true)
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
+    }
+
+    private fun mockReader(response: String): ChainReader =
+        mock<ChainReader> {
+            on { read(ChainRequest("txpool_content", ListParams())) } doReturn
+                Mono.just(ChainResponse(response.toByteArray(), null))
+        }
+
+    private fun txpoolContent(pendingAddresses: Int, queuedAddresses: Int): String {
+        val pending = (0 until pendingAddresses).joinToString(",") { """"0xp$it": {}""" }
+        val queued = (0 until queuedAddresses).joinToString(",") { """"0xq$it": {}""" }
+        return """{"pending": {$pending}, "queued": {$queued}}"""
+    }
+}

src/test/resources/configs/upstreams-bearer-auth.yaml

@@ -0,0 +1,27 @@
+version: v1
+
+upstreams:
+  - id: local
+    chain: ethereum
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+          bearer-auth:
+            token: 9c199ad8f281f20154fc258fe41a6814
+        ws:
+          url: "ws://localhost:8546"
+          origin: "http://localhost"
+          bearer-auth:
+            token: 258fe4149c199ad8f2811a68f20154fc
+  - id: both-auth
+    chain: ethereum
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+          basic-auth:
+            username: 4fc258fe41a68149c199ad8f281f2015
+            password: 1a68f20154fc258fe4149c199ad8f281
+          bearer-auth:
+            token: 9c199ad8f281f20154fc258fe41a6814