diff --git a/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptions.kt b/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptions.kt index 16470c70..6a1aae22 100644 --- 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, ) } } diff --git a/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptionsReader.kt b/foundation/src/main/kotlin/io/emeraldpay/dshackle/foundation/ChainOptionsReader.kt index b2b951b8..7c7cfc33 100644 --- 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() { getValueAsBool(values, "disable-log-index-validation")?.let { options.disableLogIndexValidation = it } + getValueAsBool(values, "disable-pending-tx-validation")?.let { + options.disablePendingTxValidation = it + } return options } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt index d193a4ed..a5d03793 100644 --- 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): 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 } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 8d5ab1bf..27b79b62 100644 --- 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 } }, ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt index 5e2189e5..949bb1d7 100644 --- 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, } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt index 3c2a4bc8..6cd99d93 100644 --- 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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidator.kt new file mode 100644 index 00000000..259ed0db --- /dev/null +++ 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 +} + +class NoopPendingTransactionValidator : PendingTransactionValidator { + override fun pendingTxExists(): Flux { + 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 { + 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) + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index c539d261..d5c5ddb3 100644 --- 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() 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() + private val pendingTxSubscription = AtomicReference() 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 { - 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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt index 9d58c153..bb12a6b5 100644 --- 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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt index 7e0a60b1..d8a5be09 100644 --- 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 { + return pendingTxValidator.pendingTxExists() + } + override fun getHead(): Head { return head } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt index 98794195..15c1b2d9 100644 --- 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 { @@ -81,6 +98,10 @@ class GenericWsConnector( return subscriptions } + override fun pendingTxEvents(): Flux { + return pendingTxValidator.pendingTxExists() + } + override fun getHead(): Head { return head } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 2f8b83ed..58c15109 100644 --- 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 ) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/GenericConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/GenericConnectorMock.groovy index f66ddf65..ea92c16a 100644 --- 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 api Head head Flux liveness + Flux pendingTxs GenericConnectorMock(Reader 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 pendingTxEvents() { + return pendingTxs + } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index 452ae5b3..6bde1fda 100644 --- 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() } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidatorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/PendingTransactionValidatorTest.kt new file mode 100644 index 00000000..6568f8cd --- /dev/null +++ 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 { + 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 { + 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 { + 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}}""" + } +}