New reactive endpoint (#514)

2024-07-01 15:49:35 +04:00
parent a5590f6603
commit eddaf98238
22 changed files with 1277 additions and 233 deletions
--- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy
@@ -92,4 +92,9 @@ class EthereumHeadMock implements Head {
    Flux<Boolean> headLiveness() {
        return Flux.empty()
    }
+
+    @Override
+    BlockContainer getCurrent() {
+        return null
+    }
 }
--- a/src/test/kotlin/io/emeraldpay/dshackle/rpc/SubscribeChainStatusTest.kt
+++ b/src/test/kotlin/io/emeraldpay/dshackle/rpc/SubscribeChainStatusTest.kt
@@ -0,0 +1,319 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.CachingReader
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.EgressSubscription
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.HeadLagObserver
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import io.emeraldpay.dshackle.upstream.state.MultistreamStateEvent
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.spy
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import reactor.test.StepVerifier
+import java.math.BigInteger
+import java.time.Duration
+import java.time.Instant
+
+class SubscribeChainStatusTest {
+    private val chainEventMapper = ChainEventMapper()
+
+    @Test
+    fun `terminate stream if an error is thrown`() {
+        val head = mock<Head> {
+            on { getCurrent() } doReturn null
+            on { getFlux() } doReturn Flux.error(IllegalStateException())
+        }
+        val ms = mock<Multistream> {
+            on { chain } doReturn Chain.ETHEREUM__MAINNET
+            on { getHead() } doReturn head
+            on { stateEvents() } doReturn Flux.empty()
+        }
+        val msHolder = mock<MultistreamHolder> {
+            on { all() } doReturn listOf(ms)
+        }
+        val subscribeChainStatus = SubscribeChainStatus(msHolder, chainEventMapper)
+
+        StepVerifier.create(subscribeChainStatus.chainStatuses())
+            .expectSubscription()
+            .expectError()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    @Test
+    fun `first full event if there is already an ms head`() {
+        val head = mock<Head> {
+            on { getCurrent() } doReturn head(550)
+            on { getFlux() } doReturn Flux.empty()
+        }
+        val ms = spy<TestMultistream> {
+            on { getHead() } doReturn head
+            on { stateEvents() } doReturn Flux.empty()
+        }
+        val msHolder = mock<MultistreamHolder> {
+            on { all() } doReturn listOf(ms)
+        }
+        val subscribeChainStatus = SubscribeChainStatus(msHolder, chainEventMapper)
+
+        StepVerifier.create(subscribeChainStatus.chainStatuses())
+            .expectSubscription()
+            .expectNext(response(true))
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    @Test
+    fun `first full event with awaiting a head from a head stream`() {
+        val head = mock<Head> {
+            on { getCurrent() } doReturn null
+            on { getFlux() } doReturn Flux.just(head(550))
+        }
+        val ms = spy<TestMultistream> {
+            on { getHead() } doReturn head
+            on { stateEvents() } doReturn Flux.empty()
+        }
+        val msHolder = mock<MultistreamHolder> {
+            on { all() } doReturn listOf(ms)
+        }
+        val subscribeChainStatus = SubscribeChainStatus(msHolder, chainEventMapper)
+
+        StepVerifier.create(subscribeChainStatus.chainStatuses())
+            .expectSubscription()
+            .expectNext(response(true))
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    @Test
+    fun `first full event with awaiting a head from a head stream and then state events`() {
+        val head = mock<Head> {
+            on { getCurrent() } doReturn null
+            on { getFlux() } doReturn Flux.just(head(550))
+        }
+        val ms = spy<TestMultistream> {
+            on { getHead() } doReturn head
+            on { stateEvents() } doReturn Flux.just(
+                listOf(
+                    MultistreamStateEvent.StatusEvent(UpstreamAvailability.OK),
+                    MultistreamStateEvent.MethodsEvent(setOf("superMethod")),
+                    MultistreamStateEvent.SubsEvent(listOf("heads")),
+                    MultistreamStateEvent.CapabilitiesEvent(setOf(Capability.BALANCE)),
+                    MultistreamStateEvent.LowerBoundsEvent(listOf(LowerBoundData(800, 1000, LowerBoundType.STATE))),
+                    MultistreamStateEvent.FinalizationEvent(listOf(FinalizationData(30, FinalizationType.SAFE_BLOCK))),
+                    MultistreamStateEvent.NodeDetailsEvent(listOf(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "val"))))),
+                ),
+            )
+        }
+        val msHolder = mock<MultistreamHolder> {
+            on { all() } doReturn listOf(ms)
+        }
+        val subscribeChainStatus = SubscribeChainStatus(msHolder, chainEventMapper)
+
+        StepVerifier.create(subscribeChainStatus.chainStatuses())
+            .expectSubscription()
+            .expectNext(response(true))
+            .expectNext(response(false))
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    @Test
+    fun `first full event with awaiting a head from a head stream and then head events`() {
+        val head = mock<Head> {
+            on { getCurrent() } doReturn null
+            on { getFlux() } doReturn Flux.just(head(550), head(600))
+        }
+        val ms = spy<TestMultistream> {
+            on { getHead() } doReturn head
+            on { stateEvents() } doReturn Flux.empty()
+        }
+        val msHolder = mock<MultistreamHolder> {
+            on { all() } doReturn listOf(ms)
+        }
+        val subscribeChainStatus = SubscribeChainStatus(msHolder, chainEventMapper)
+
+        StepVerifier.create(subscribeChainStatus.chainStatuses())
+            .expectSubscription()
+            .expectNext(response(true))
+            .expectNext(
+                BlockchainOuterClass.SubscribeChainStatusResponse.newBuilder()
+                    .setChainDescription(
+                        BlockchainOuterClass.ChainDescription.newBuilder()
+                            .setChain(Common.ChainRef.CHAIN_ETHEREUM__MAINNET)
+                            .addChainEvent(chainEventMapper.mapHead(head(600)))
+                            .build(),
+                    )
+                    .build(),
+            )
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    private fun head(height: Long): BlockContainer {
+        return BlockContainer(
+            height,
+            BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"),
+            BigInteger.ONE,
+            Instant.ofEpochSecond(1719485864),
+            false,
+            null,
+            null,
+            BlockId.from("0xa6af163aab691919c595e2a466f0a6b01f1dff8cfd9631dee811df57064c2d32"),
+            emptyList(),
+        )
+    }
+
+    private fun response(headEvent: Boolean): BlockchainOuterClass.SubscribeChainStatusResponse {
+        return BlockchainOuterClass.SubscribeChainStatusResponse.newBuilder()
+            .apply {
+                if (headEvent) {
+                    setBuildInfo(
+                        BlockchainOuterClass.BuildInfo.newBuilder()
+                            .setVersion("DEV")
+                            .build(),
+                    )
+                }
+            }
+            .setChainDescription(
+                BlockchainOuterClass.ChainDescription.newBuilder()
+                    .setChain(Common.ChainRef.CHAIN_ETHEREUM__MAINNET)
+                    .addChainEvent(chainEventMapper.chainStatus(UpstreamAvailability.OK))
+                    .apply {
+                        if (headEvent) {
+                            addChainEvent(chainEventMapper.mapHead(head(550)))
+                        }
+                    }
+                    .addChainEvent(chainEventMapper.supportedMethods(setOf("superMethod")))
+                    .addChainEvent(chainEventMapper.supportedSubs(listOf("heads")))
+                    .addChainEvent(chainEventMapper.mapCapabilities(setOf(Capability.BALANCE)))
+                    .addChainEvent(
+                        chainEventMapper.mapLowerBounds(
+                            listOf(LowerBoundData(800, 1000, LowerBoundType.STATE)),
+                        ),
+                    )
+                    .addChainEvent(
+                        chainEventMapper.mapFinalizationData(
+                            listOf(FinalizationData(30, FinalizationType.SAFE_BLOCK)),
+                        ),
+                    )
+                    .addChainEvent(
+                        chainEventMapper.mapNodeDetails(
+                            listOf(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "val")))),
+                        ),
+                    )
+                    .build(),
+            )
+            .apply {
+                if (headEvent) {
+                    setFullResponse(true)
+                }
+            }
+            .build()
+    }
+
+    private open class TestMultistream : Multistream(Chain.ETHEREUM__MAINNET, mock<Caches>(), null, Schedulers.single()) {
+        companion object {
+            private const val UNIMPLEMENTED = "UNIMPLEMENTED"
+        }
+
+        override fun getMethods(): CallMethods {
+            val callMethods = mock<CallMethods> {
+                on { getSupportedMethods() } doReturn setOf("superMethod")
+            }
+            return callMethods
+        }
+
+        override fun getUpstreams(): MutableList<out Upstream> {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun addUpstreamInternal(u: Upstream) {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun getLocalReader(): Mono<ChainReader> {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun addHead(upstream: Upstream) {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun removeHead(upstreamId: String) {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun makeLagObserver(): HeadLagObserver {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun getCachingReader(): CachingReader? {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun getHead(mather: Selector.Matcher): Head {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun getHead(): Head {
+            return mock<Head>()
+        }
+
+        override fun getEgressSubscription(): EgressSubscription {
+            val sub = mock<EgressSubscription> {
+                on { getAvailableTopics() } doReturn listOf("heads")
+            }
+            return sub
+        }
+
+        override fun getLabels(): Collection<UpstreamsConfig.Labels> {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun <T : Upstream> cast(selfType: Class<T>): T {
+            throw IllegalStateException(UNIMPLEMENTED)
+        }
+
+        override fun getStatus(): UpstreamAvailability {
+            return UpstreamAvailability.OK
+        }
+
+        override fun getCapabilities(): Set<Capability> {
+            return setOf(Capability.BALANCE)
+        }
+
+        override fun getLowerBounds(): Collection<LowerBoundData> {
+            return listOf(LowerBoundData(800, 1000, LowerBoundType.STATE))
+        }
+
+        override fun getFinalizations(): Collection<FinalizationData> {
+            return listOf(FinalizationData(30, FinalizationType.SAFE_BLOCK))
+        }
+
+        override fun getQuorumLabels(): List<QuorumForLabels.QuorumItem> {
+            return listOf(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "val"))))
+        }
+    }
+}
--- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt
+++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFinalizationDetectorTest.kt
@@ -10,66 +10,55 @@ import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionRefJson
 import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
 import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
 import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
-import org.junit.jupiter.api.Assertions
-import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import org.mockito.Mockito.mock
-import org.mockito.Mockito.`when`
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
 import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
 import java.time.Duration
 import java.time.Instant

 class EthereumFinalizationDetectorTest {

-    private lateinit var upstream: Upstream
-    private lateinit var chainReader: ChainReader
-    private lateinit var detector: EthereumFinalizationDetector
-
-    @BeforeEach
-    fun setUp() {
-        upstream = mock()
-        chainReader = mock()
-        `when`(upstream.getIngressReader()).thenReturn(chainReader)
-        detector = EthereumFinalizationDetector()
-    }
-
    @Test
    fun testDetectFinalization() {
-        `when`(chainReader.read(ChainRequest("eth_getBlockByNumber", ListParams("safe", false), 1)))
-            .thenReturn(
-                Mono.just(
-                    ChainResponse(
-                        Global.objectMapper.writeValueAsString(
-                            BlockJson<TransactionRefJson>().apply {
-                                number = 1
-                                timestamp = Instant.now()
-                            },
-                        ).toByteArray(),
-                        null,
-                    ),
-                ),
-            )
-        `when`(chainReader.read(ChainRequest("eth_getBlockByNumber", ListParams("finalized", false), 2)))
-            .thenReturn(
-                Mono.just(
-                    ChainResponse(
-                        Global.objectMapper.writeValueAsString(
-                            BlockJson<TransactionRefJson>().apply {
-                                number = 2
-                                timestamp = Instant.now()
-                            },
-                        ).toByteArray(),
-                        null,
-                    ),
-                ),
-            )
+        val reader = mock<ChainReader> {
+            on {
+                read(ChainRequest("eth_getBlockByNumber", ListParams("safe", false), 1))
+            } doReturn response(1) doReturn response(5) doReturn response(3)
+            on {
+                read(ChainRequest("eth_getBlockByNumber", ListParams("finalized", false), 2))
+            } doReturn response(2) doReturn response(10) doReturn response(5)
+        }
+        val upstream = mock<Upstream> {
+            on { getIngressReader() } doReturn reader
+        }
+        val detector = EthereumFinalizationDetector()

-        val flux = detector.detectFinalization(upstream, Duration.ofMillis(200))
-        flux.take(2).collectList().block()
-        val result = detector.getFinalizations().toList()
-        Assertions.assertEquals(2, result.size)
-        org.assertj.core.api.Assertions.assertThat(result)
-            .contains(FinalizationData(2L, FinalizationType.FINALIZED_BLOCK))
-            .contains(FinalizationData(1L, FinalizationType.SAFE_BLOCK))
+        StepVerifier.withVirtualTime { detector.detectFinalization(upstream, Duration.ofMillis(200)) }
+            .expectSubscription()
+            .thenAwait(Duration.ofSeconds(0))
+            .expectNext(FinalizationData(1L, FinalizationType.SAFE_BLOCK))
+            .expectNext(FinalizationData(2L, FinalizationType.FINALIZED_BLOCK))
+            .thenAwait(Duration.ofSeconds(15))
+            .expectNext(FinalizationData(5L, FinalizationType.SAFE_BLOCK))
+            .expectNext(FinalizationData(10L, FinalizationType.FINALIZED_BLOCK))
+            .thenAwait(Duration.ofSeconds(15))
+            .expectNoEvent(Duration.ofMillis(100))
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
    }
+
+    private fun response(blockNumber: Long) =
+        Mono.just(
+            ChainResponse(
+                Global.objectMapper.writeValueAsString(
+                    BlockJson<TransactionRefJson>().apply {
+                        number = blockNumber
+                        timestamp = Instant.now()
+                    },
+                ).toByteArray(),
+                null,
+            ),
+        )
 }
--- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/state/MultistreamStateHandlerTest.kt
+++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/state/MultistreamStateHandlerTest.kt
@@ -0,0 +1,93 @@
+package io.emeraldpay.dshackle.upstream.state
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
+
+class MultistreamStateHandlerTest {
+    private val stateHandler = MultistreamStateHandler
+
+    @ParameterizedTest
+    @MethodSource("states")
+    fun `compare states`(
+        newState: MultistreamState.CurrentMultistreamState,
+        expectedEvents: Collection<MultistreamStateEvent>,
+    ) {
+        val oldState = state()
+        val events = stateHandler.compareStates(oldState, newState)
+
+        assertThat(events).isEqualTo(expectedEvents)
+    }
+
+    companion object {
+        @JvmStatic
+        fun states(): List<Arguments> =
+            listOf(
+                Arguments.of(
+                    state(),
+                    listOf<MultistreamStateEvent>(),
+                ),
+                Arguments.of(
+                    state(status = UpstreamAvailability.UNAVAILABLE),
+                    listOf(MultistreamStateEvent.StatusEvent(UpstreamAvailability.UNAVAILABLE)),
+                ),
+                Arguments.of(
+                    state(methods = setOf("otherCall")),
+                    listOf(MultistreamStateEvent.MethodsEvent(setOf("otherCall"))),
+                ),
+                Arguments.of(
+                    state(subs = setOf("newSub")),
+                    listOf(MultistreamStateEvent.SubsEvent(setOf("newSub"))),
+                ),
+                Arguments.of(
+                    state(caps = setOf(Capability.BALANCE)),
+                    listOf(MultistreamStateEvent.CapabilitiesEvent(setOf(Capability.BALANCE))),
+                ),
+                Arguments.of(
+                    state(lowerBounds = setOf(LowerBoundData(90, 90, LowerBoundType.STATE))),
+                    listOf(MultistreamStateEvent.LowerBoundsEvent(setOf(LowerBoundData(90, 90, LowerBoundType.STATE)))),
+                ),
+                Arguments.of(
+                    state(finalizationData = setOf(FinalizationData(80, FinalizationType.SAFE_BLOCK))),
+                    listOf(MultistreamStateEvent.FinalizationEvent(setOf(FinalizationData(80, FinalizationType.SAFE_BLOCK)))),
+                ),
+                Arguments.of(
+                    state(quorumForLabels = setOf(QuorumForLabels.QuorumItem(2, UpstreamsConfig.Labels.fromMap(mapOf("test1" to "val1"))))),
+                    listOf(
+                        MultistreamStateEvent.NodeDetailsEvent(
+                            setOf(QuorumForLabels.QuorumItem(2, UpstreamsConfig.Labels.fromMap(mapOf("test1" to "val1")))),
+                        ),
+                    ),
+                ),
+            )
+
+        private fun state(
+            status: UpstreamAvailability = UpstreamAvailability.OK,
+            methods: Set<String> = setOf("method"),
+            subs: Set<String> = setOf("sub"),
+            caps: Set<Capability> = setOf(Capability.RPC),
+            lowerBounds: Set<LowerBoundData> = setOf(LowerBoundData(55, 55, LowerBoundType.STATE)),
+            finalizationData: Set<FinalizationData> = setOf(FinalizationData(20, FinalizationType.SAFE_BLOCK)),
+            quorumForLabels: Set<QuorumForLabels.QuorumItem> = setOf(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "val")))),
+        ): MultistreamState.CurrentMultistreamState {
+            return MultistreamState.CurrentMultistreamState(
+                status,
+                methods,
+                subs,
+                caps,
+                lowerBounds,
+                finalizationData,
+                quorumForLabels,
+            )
+        }
+    }
+}
--- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/state/MultistreamStateTest.kt
+++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/state/MultistreamStateTest.kt
@@ -0,0 +1,161 @@
+package io.emeraldpay.dshackle.upstream.state
+
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
+import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.doReturn
+import org.mockito.kotlin.mock
+import reactor.test.StepVerifier
+import java.time.Duration
+
+class MultistreamStateTest {
+
+    @Test
+    fun `update state and send events`() {
+        val up1 = upstream(
+            UpstreamAvailability.OK,
+            true,
+            callMethods(setOf("eth_call", "super_call")),
+            setOf(Capability.RPC),
+            QuorumForLabels(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "value")))),
+            listOf(LowerBoundData(55, LowerBoundType.STATE), LowerBoundData(99, LowerBoundType.BLOCK)),
+            listOf(FinalizationData(32, FinalizationType.SAFE_BLOCK), FinalizationData(80, FinalizationType.FINALIZED_BLOCK)),
+        )
+        val up2 = upstream(
+            UpstreamAvailability.UNAVAILABLE,
+            false,
+            callMethods(setOf("one_more_call")),
+            setOf(Capability.BALANCE),
+            QuorumForLabels(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("new" to "old")))),
+            listOf(LowerBoundData(22, LowerBoundType.STATE), LowerBoundData(5, LowerBoundType.BLOCK)),
+            listOf(FinalizationData(3200, FinalizationType.SAFE_BLOCK), FinalizationData(1180, FinalizationType.FINALIZED_BLOCK)),
+        )
+        val up3 = upstream(
+            UpstreamAvailability.LAGGING,
+            true,
+            callMethods(setOf("super_duper_call")),
+            setOf(Capability.WS_HEAD),
+            QuorumForLabels(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("megaTest" to "valTest")))),
+            listOf(LowerBoundData(40, LowerBoundType.STATE), LowerBoundData(400, LowerBoundType.BLOCK)),
+            listOf(FinalizationData(70, FinalizationType.SAFE_BLOCK), FinalizationData(15, FinalizationType.FINALIZED_BLOCK)),
+        )
+        val up4 = upstream(
+            UpstreamAvailability.OK,
+            true,
+            callMethods(setOf("super_duper_call", "yet_another_call")),
+            setOf(Capability.WS_HEAD),
+            QuorumForLabels(QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("megaTest" to "valTest")))),
+            listOf(LowerBoundData(1, LowerBoundType.STATE), LowerBoundData(1, LowerBoundType.BLOCK)),
+            listOf(FinalizationData(990, FinalizationType.SAFE_BLOCK), FinalizationData(880, FinalizationType.FINALIZED_BLOCK)),
+        )
+
+        val state = MultistreamState {}
+
+        StepVerifier.create(state.stateEvents())
+            .then { state.updateState(listOf(up1, up2, up3), listOf("heads", "notHeads")) }
+            .assertNext {
+                assertThat(it).hasSize(7)
+                assertThat(it.toList())
+                    .usingRecursiveFieldByFieldElementComparatorIgnoringFields("lowerBounds.timestamp")
+                    .hasSameElementsAs(
+                        listOf(
+                            MultistreamStateEvent.StatusEvent(UpstreamAvailability.OK),
+                            MultistreamStateEvent.MethodsEvent(setOf("eth_call", "super_call", "super_duper_call").toHashSet()),
+                            MultistreamStateEvent.SubsEvent(listOf("heads", "notHeads")),
+                            MultistreamStateEvent.CapabilitiesEvent(setOf(Capability.WS_HEAD, Capability.RPC).toHashSet()),
+                            MultistreamStateEvent.LowerBoundsEvent(
+                                setOf(
+                                    LowerBoundData(40, LowerBoundType.STATE),
+                                    LowerBoundData(99, LowerBoundType.BLOCK),
+                                ).toHashSet(),
+                            ),
+                            MultistreamStateEvent.FinalizationEvent(
+                                setOf(
+                                    FinalizationData(70, FinalizationType.SAFE_BLOCK),
+                                    FinalizationData(80, FinalizationType.FINALIZED_BLOCK),
+                                ).toHashSet(),
+                            ),
+                            MultistreamStateEvent.NodeDetailsEvent(
+                                listOf(
+                                    QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "value"))),
+                                    QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("megaTest" to "valTest"))),
+                                ),
+                            ),
+                        ),
+                    )
+            }
+            .then { state.updateState(listOf(up1, up2, up3), listOf("heads", "notHeads")) }
+            .assertNext {
+                assertThat(it).hasSize(0)
+            }
+            .then { state.updateState(listOf(up1, up2, up3, up4), listOf("heads", "notHeads")) }
+            .assertNext {
+                assertThat(it).hasSize(4)
+                assertThat(it.toList())
+                    .usingRecursiveFieldByFieldElementComparatorIgnoringFields("lowerBounds.timestamp")
+                    .hasSameElementsAs(
+                        listOf(
+                            MultistreamStateEvent.MethodsEvent(setOf("eth_call", "yet_another_call", "super_call", "super_duper_call").toHashSet()),
+                            MultistreamStateEvent.LowerBoundsEvent(
+                                setOf(
+                                    LowerBoundData(1, LowerBoundType.STATE),
+                                    LowerBoundData(1, LowerBoundType.BLOCK),
+                                ).toHashSet(),
+                            ),
+                            MultistreamStateEvent.FinalizationEvent(
+                                setOf(
+                                    FinalizationData(990, FinalizationType.SAFE_BLOCK),
+                                    FinalizationData(880, FinalizationType.FINALIZED_BLOCK),
+                                ).toHashSet(),
+                            ),
+                            MultistreamStateEvent.NodeDetailsEvent(
+                                listOf(
+                                    QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(mapOf("test" to "value"))),
+                                    QuorumForLabels.QuorumItem(2, UpstreamsConfig.Labels.fromMap(mapOf("megaTest" to "valTest"))),
+                                ),
+                            ),
+                        ),
+                    )
+            }
+            .thenCancel()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    private fun upstream(
+        status: UpstreamAvailability,
+        isAvailable: Boolean,
+        callMethods: CallMethods,
+        capabilities: Set<Capability>,
+        quorumForLabels: QuorumForLabels,
+        lowerBounds: Collection<LowerBoundData>,
+        finalizationData: Collection<FinalizationData>,
+    ): DefaultUpstream {
+        val upstream = mock<DefaultUpstream> {
+            on { isAvailable() } doReturn isAvailable
+            on { getMethods() } doReturn callMethods
+            on { getCapabilities() } doReturn capabilities
+            on { getQuorumByLabel() } doReturn quorumForLabels
+            on { getLowerBounds() } doReturn lowerBounds
+            on { getFinalizations() } doReturn finalizationData
+            on { getStatus() } doReturn status
+        }
+
+        return upstream
+    }
+
+    private fun callMethods(methods: Set<String>): CallMethods {
+        val callMethods = mock<CallMethods> {
+            on { getSupportedMethods() } doReturn methods
+        }
+        return callMethods
+    }
+}