From 9ca2c921b52d8cd2e89a61382b978523c9385f29 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 4 Apr 2025 17:09:08 +0300 Subject: [PATCH] support null topics and topic position in logs filter (#649) --- .../ethereum/EthereumEgressSubscription.kt | 9 ++-- .../ethereum/subscribe/ConnectLogs.kt | 15 +++---- .../EthereumEgressSubscriptionSpec.groovy | 6 +-- .../ethereum/subscribe/ConnectLogsSpec.groovy | 45 ++++++++++++++++++- 4 files changed, 55 insertions(+), 20 deletions(-) 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 76d3a695..04ac6c70 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt @@ -72,7 +72,7 @@ open class EthereumEgressSubscription( data class LogsRequest( val address: List
, - val topics: List, + val topics: List, ) fun readLogsRequest(params: Map): LogsRequest { @@ -98,7 +98,7 @@ open class EthereumEgressSubscription( } else { emptyList() } - val topics: List = if (params.containsKey("topics")) { + val topics: List = if (params.containsKey("topics")) { when (val topics = params["topics"]) { is String -> try { listOf(Hex32.from(topics)) @@ -106,15 +106,16 @@ open class EthereumEgressSubscription( log.debug("Ignore invalid topic: $topics with error ${t.message}") emptyList() } - is Collection<*> -> topics.mapNotNull { topic -> + is Collection<*> -> topics.map { topic -> try { when (topic) { + null -> null is Collection<*> -> topic.firstOrNull()?.toString()?.let { Hex32.from(it) } else -> topic?.toString()?.let { Hex32.from(it) } } } catch (t: Throwable) { log.debug("Ignore invalid topic: $topic with error ${t.message}") - null + throw IllegalArgumentException("Invalid topic: $topic") } } null -> emptyList() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt index a8710086..b76931c6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt @@ -44,7 +44,7 @@ open class ConnectLogs( return produceLogs.produce(connectBlockUpdates.connect(matcher)) } - open fun create(addresses: List
, topics: List): SubscriptionConnect { + open fun create(addresses: List
, topics: List): SubscriptionConnect { return object : SubscriptionConnect { override fun connect(matcher: Selector.Matcher): Flux { // shortcut to the whole output if we don't have any filters @@ -58,22 +58,17 @@ open class ConnectLogs( } } - fun filtered(addresses: List
, topics: List): Function, Flux> { + fun filtered(addresses: List
, selectedTopics: List): Function, Flux> { // sort search criteria to use binary search later val sortedAddresses: List
= addresses.sortedWith(ADDR_COMPARATOR) - val sortedTopics: List = topics.sortedWith(TOPIC_COMPARATOR) return Function { logs -> logs.filter { val goodAddress = sortedAddresses.isEmpty() || sortedAddresses.binarySearch(it.address, ADDR_COMPARATOR) >= 0 val goodTopic = when { - sortedTopics.isEmpty() -> true - it.topics.size < sortedTopics.size -> false - else -> sortedTopics.indices.all { index -> - it.topics[index].let { logTopic -> - sortedTopics.binarySearch(logTopic, TOPIC_COMPARATOR) >= 0 - } - } + selectedTopics.isEmpty() -> true + it.topics.size < selectedTopics.size -> false + else -> selectedTopics.zip(it.topics).all { (selectedTopic, logTopic) -> selectedTopic == null || selectedTopic == logTopic } } goodAddress && goodTopic } 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 17e6565a..222f98d4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -129,10 +129,7 @@ class EthereumEgressSubscriptionSpec extends Specification { ]) then: - act.address == [] - act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") - ] + thrown(IllegalArgumentException) } def "read multi topic logs request"() { @@ -195,6 +192,7 @@ class EthereumEgressSubscriptionSpec extends Specification { ] act.topics == [ Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + null, Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") ] } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy index ddde0152..36e05682 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy @@ -104,6 +104,23 @@ class ConnectLogsSpec extends Specification { "upstream" ) + def log6 = new LogMessage( + Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"), + BlockHash.empty(), + 100L, + HexData.empty(), + 1L, + [ + Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"), + Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"), + Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641") + ], + TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), + 1L, + false, + "upstream" + ) + def "Filter is empty"() { setup: def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic()) @@ -174,7 +191,7 @@ class ConnectLogsSpec extends Specification { def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic()) when: def input = Flux.fromIterable([ - log1, log2, log3, log4, log5 + log1, log2, log3, log4, log5, log6 ]) def act = input.transform(connectLogs.filtered( @@ -187,7 +204,31 @@ class ConnectLogsSpec extends Specification { .collectList().block() then: - act.size() == 1 + act.size() == 2 act[0] == log5 + act[1] == log6 + } + + def "Filter by address and second topics"() { + setup: + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic()) + when: + def input = Flux.fromIterable([ + log1, log2, log3, log4, log5, log6 + ]) + + def act = input.transform(connectLogs.filtered( + [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], + [ + Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"), + null, + Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641"), + ] + )) + .collectList().block() + + then: + act.size() == 1 + act[0] == log6 } } \ No newline at end of file