newPedningTransactions validator (#864)

2026-05-26 17:13:31 +04:00
parent f93f350a29
commit 998e1ac933
15 changed files with 382 additions and 9 deletions
--- a/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptions.kt
+++ b/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,
            )
    }
 }
--- a/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptionsReader.kt
+++ b/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
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt
+++ b/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
            }
        }

--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
+++ b/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
                                }
                            },
                        )
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt
@@ -4,4 +4,5 @@ enum class Capability {
    RPC,
    BALANCE,
    WS_HEAD,
+    WS_PENDING_TX,
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt
+++ b/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
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidator.kt
+++ b/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)
+                    }
+            }
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
@@ -110,6 +110,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,6 +122,7 @@ 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

@@ -150,11 +153,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()) {
-            setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD)
-        } else {
-            setOf(Capability.RPC, Capability.BALANCE)
+        val caps = mutableSetOf(Capability.RPC, Capability.BALANCE)
+        if (hasLiveSubscriptionHead.get()) {
+            caps.add(Capability.WS_HEAD)
        }
+        if (hasPendingTxs.get()) {
+            caps.add(Capability.WS_PENDING_TX)
+        }
+        return caps
    }

    override fun isGrpc(): Boolean {
@@ -358,6 +364,14 @@ open class GenericUpstream(
                ),
            )
        }
+        if (!getOptions().disablePendingTxValidation) {
+            pendingTxSubscription.set(
+                connector.pendingTxEvents().subscribe {
+                    hasPendingTxs.set(it)
+                    sendUpstreamStateEvent(UPDATED)
+                },
+            )
+        }
        detectSettings()

        if (!getOptions().disableBoundValidation) {
@@ -380,6 +394,7 @@ open class GenericUpstream(
        lowerBlockDetectorSubscription.getAndSet(null)?.dispose()
        finalizationDetectorSubscription.getAndSet(null)?.dispose()
        settingsDetectorSubscription.getAndSet(null)?.dispose()
+        pendingTxSubscription.getAndSet(null)?.dispose()
        connector.getHead().stop()
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt
+++ b/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>
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt
+++ b/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
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt
+++ b/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
    }
--- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
@@ -636,7 +636,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
        )
    }
 }
--- a/src/test/groovy/io/emeraldpay/dshackle/test/GenericConnectorMock.groovy
+++ b/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
+    }
 }
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
+++ b/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()
    }
 }
--- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidatorTest.kt
+++ b/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}}"""
+    }
+}