problem: ERC20 Balancer subsription checks balancer on each block, even if no transfer happened

splix · splix · commit e1170dd055a6 · 2021-11-24T19:26:34.000-05:00
solution: use ERC20 Logs to determine changes to ERC20 balance
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
@@ -1,19 +1,30 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.TokensConfig
-import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
-import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.domain.Address
+import io.emeraldpay.etherjar.domain.EventId
 import io.emeraldpay.etherjar.erc20.ERC20Token
-import io.emeraldpay.etherjar.hex.Hex32
-import io.emeraldpay.etherjar.hex.HexQuantity
 import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
@@ -35,6 +46,8 @@ class TrackERC20Address(
         private val log = LoggerFactory.getLogger(TrackERC20Address::class.java)
     }
 
+    var erc20Balance: ERC20Balance = ERC20Balance()
+
     private val ethereumAddresses = EthereumAddresses()
     private val tokens: MutableMap<TokenId, TokenDefinition> = HashMap()
 
@@ -72,13 +85,30 @@ class TrackERC20Address(
         val chain = Chain.byId(request.asset.chainValue)
         val asset = request.asset.code.lowercase(Locale.getDefault())
         val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty()
-        val head = multistreamHolder.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty()
+        val logs = getUpstream(chain)
+            .getSubscribe().logs
+            .start(
+                listOf(tokenDefinition.token.contract),
+                listOf(EventId.fromSignature("Transfer", "address", "address", "uint256"))
+            )
 
         return ethereumAddresses.extract(request.address)
             .map { TrackedAddress(chain, it, tokenDefinition.token, tokenDefinition.name) }
             .flatMap { addr ->
                 val current = getBalance(addr)
-                val updates = head.flatMap { getBalance(addr) }
+
+                val updates = logs
+                    .filter {
+                        it.topics.size >= 3 && (Address.extract(it.topics[1]) == addr.address || Address.extract(it.topics[2]) == addr.address)
+                    }
+                    .distinctUntilChanged {
+                        // check it once per block
+                        it.blockHash
+                    }
+                    .flatMap {
+                        // make sure we use actual balance, don't trust event blindly
+                        getBalance(addr)
+                    }
                 Flux.concat(current, updates)
                     .distinctUntilChanged()
                     .map { addr.withBalance(it) }
@@ -88,23 +118,7 @@ class TrackERC20Address(
 
     fun getBalance(addr: TrackedAddress): Mono<BigInteger> {
         val upstream = getUpstream(addr.chain)
-        return upstream
-            .getDirectApi(Selector.empty)
-            .flatMap { api ->
-                api.read(prepareEthCall(addr.token, addr.address, upstream.getHead()))
-                    .flatMap(JsonRpcResponse::requireStringResult)
-                    .map {
-                        Hex32.from(it).asQuantity().value
-                    }
-            }
-    }
-
-    fun prepareEthCall(token: ERC20Token, target: Address, head: Head): JsonRpcRequest {
-        val call = token
-            .readBalanceOf(target)
-            .toJson()
-        val height = head.getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
-        return JsonRpcRequest("eth_call", listOf(call, height))
+        return erc20Balance.getBalance(upstream, addr.token, addr.address)
     }
 
     fun getUpstream(chain: Chain): EthereumMultistream {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt
@@ -0,0 +1,66 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.domain.Address
+import io.emeraldpay.etherjar.erc20.ERC20Token
+import io.emeraldpay.etherjar.hex.Hex32
+import io.emeraldpay.etherjar.hex.HexQuantity
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.math.BigInteger
+
+/**
+ * Query for a ERC20 token balance for an address
+ */
+open class ERC20Balance {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ERC20Balance::class.java)
+    }
+
+    open fun getBalance(upstreams: EthereumMultistream, token: ERC20Token, address: Address): Mono<BigInteger> {
+        return upstreams
+            // use only up-to-date upstreams
+            .getApiSource(Selector.HeightMatcher(upstreams.getHead().getCurrentHeight() ?: 0))
+            .let { Flux.from(it) }
+            .flatMap {
+                getBalance(it.cast(EthereumUpstream::class.java), token, address)
+            }
+            .next()
+    }
+
+    open fun getBalance(upstream: EthereumUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
+        return upstream
+            .getApi()
+            .read(prepareEthCall(token, address, upstream.getHead()))
+            .flatMap(JsonRpcResponse::requireStringResult)
+            .map { Hex32.from(it).asQuantity().value }
+    }
+
+    fun prepareEthCall(token: ERC20Token, target: Address, head: Head): JsonRpcRequest {
+        val call = token
+            .readBalanceOf(target)
+            .toJson()
+        val height = head.getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
+        return JsonRpcRequest("eth_call", listOf(call, height))
+    }
+}
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
@@ -17,7 +17,7 @@ open class EthereumSubscribe(
     }
 
     private val newHeads = ConnectNewHeads(upstream)
-    private val logs = ConnectLogs(upstream)
+    open val logs = ConnectLogs(upstream)
     private val syncing = ConnectSyncing(upstream)
 
     @Suppress("UNCHECKED_CAST")
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
@@ -24,7 +24,7 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import java.util.function.Function
 
-class ConnectLogs(
+open class ConnectLogs(
     upstream: EthereumMultistream,
     private val connectBlockUpdates: ConnectBlockUpdates,
 ) {
@@ -44,7 +44,7 @@ class ConnectLogs(
         return produceLogs.produce(connectBlockUpdates.connect())
     }
 
-    fun start(addresses: List<Address>, topics: List<Hex32>): Flux<LogMessage> {
+    open fun start(addresses: List<Address>, topics: List<Hex32>): Flux<LogMessage> {
         // shortcut to the whole output if we don't have any filters
         if (addresses.isEmpty() && topics.isEmpty()) {
             return start()
diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
@@ -6,20 +6,31 @@ import io.emeraldpay.dshackle.config.TokensConfig
 import io.emeraldpay.dshackle.test.EthereumUpstreamMock
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
 import io.emeraldpay.dshackle.test.ReaderMock
-import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.hex.Hex32
 import io.emeraldpay.grpc.Chain
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.erc20.ERC20Token
 import io.emeraldpay.etherjar.hex.HexData
 import io.emeraldpay.etherjar.rpc.json.TransactionCallJson
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Ignore
 import spock.lang.Specification
 
+import java.time.Duration
+
 class TrackERC20AddressSpec extends Specification {
 
     def "Init with single token"() {
@@ -121,38 +132,6 @@ class TrackERC20AddressSpec extends Specification {
         supportSai
     }
 
-    def "Gets balance from upstream"() {
-        setup:
-        ReaderMock api = new ReaderMock()
-                .with(
-                        new JsonRpcRequest("eth_call", [
-                                new TransactionCallJson().tap { json ->
-                                    json.setTo(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9"))
-                                    json.setData(HexData.from("0x70a0823100000000000000000000000016c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b"))
-                                },
-                                "latest"
-                        ]),
-                        JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"')
-                )
-
-        EthereumUpstream upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
-        MultistreamHolder ups = new MultistreamHolderMock(Chain.ETHEREUM, upstream)
-        TrackERC20Address track = new TrackERC20Address(ups, new TokensConfig([]))
-
-        TrackERC20Address.TrackedAddress address = new TrackERC20Address.TrackedAddress(
-                Chain.ETHEREUM,
-                Address.from("0x16c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b"),
-                new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")),
-                "test",
-                BigInteger.valueOf(1234)
-        )
-        when:
-        def act = track.getBalance(address).block()
-
-        then:
-        act.toLong() == 0x1f28d72868
-    }
-
     def "Builds response"() {
         setup:
         TrackERC20Address track = new TrackERC20Address(Stub(MultistreamHolder), new TokensConfig([]))
@@ -175,4 +154,89 @@ class TrackERC20AddressSpec extends Specification {
                 .setBalance("1234")
                 .build()
     }
+
+    def "Check balance when event happens"() {
+        setup:
+        def events = [
+                new LogMessage(
+                        Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9"),
+                        BlockHash.from("0x0c0d2969c843d0b61fbab1b2302cf24d6681b2ae0a140a3c2908990d048f7631"),
+                        13668750,
+                        HexData.from("0x0000000000000000000000000000000000000000000000000000000048f2fc7b"),
+                        1,
+                        [
+                                Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+                                Hex32.from("0x000000000000000000000000b02f1329d6a6acef07a763258f8509c2847a0a3e"),
+                                Hex32.from("0x00000000000000000000000016c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b")
+                        ],
+                        TransactionId.from("0x5a7898e27120575c33d3d0179af3b6353c7268bbad4255df079ed26b743a21a5"),
+                        1,
+                        false
+                )
+        ]
+        def logs = Mock(ConnectLogs) {
+            1 * start(
+                    [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")],
+                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")]
+            ) >> { args ->
+                println("ConnectLogs.start $args")
+                Flux.fromIterable(events)
+            }
+        }
+        def sub = Mock(EthereumSubscribe) {
+            1 * getLogs() >> logs
+        }
+        def up = Mock(EthereumMultistream) {
+            1 * getSubscribe() >> sub
+            _ * cast(EthereumMultistream) >> { args ->
+                it
+            }
+        }
+        def mup = Mock(MultistreamHolder) {
+            _ * getUpstream(Chain.ETHEREUM) >> up
+        }
+        TokensConfig tokens = new TokensConfig([
+                new TokensConfig.Token().tap {
+                    id = "test"
+                    blockchain = Chain.ETHEREUM
+                    name = "TEST"
+                    type = TokensConfig.Type.ERC20
+                    address = Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")
+                }
+        ])
+        TrackERC20Address track = new TrackERC20Address(mup, tokens)
+        track.init()
+        track.erc20Balance = Mock(ERC20Balance) {
+            2 * it.getBalance(_, _, _) >>> [
+                    Mono.just(100000.toBigInteger()),
+                    Mono.just(150000.toBigInteger())
+            ]
+        }
+        def request = BlockchainOuterClass.BalanceRequest.newBuilder()
+                .setAddress(
+                        Common.AnyAddress.newBuilder()
+                                .setAddressSingle(Common.SingleAddress.newBuilder().setAddress("0x16c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b"))
+                )
+                .setAsset(
+                        Common.Asset.newBuilder()
+                                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
+                                .setCode("TEST")
+                )
+                .build()
+        when:
+        def act = track.subscribe(request)
+
+        then:
+        StepVerifier.create(act)
+                .expectNextMatches {
+                    println("Received: $it")
+                    it.getBalance() == "100000"
+                }
+                .expectNextMatches {
+                    println("Received: $it")
+                    it.getBalance() == "150000"
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
 }
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy

Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ open class EthereumSubscribe(`
`17`	`17`	`}`
`18`	`18`
`19`	`19`	`private val newHeads = ConnectNewHeads(upstream)`
`20`		`- private val logs = ConnectLogs(upstream)`
	`20`	`+ open val logs = ConnectLogs(upstream)`
`21`	`21`	`private val syncing = ConnectSyncing(upstream)`
`22`	`22`
`23`	`23`	`@Suppress("UNCHECKED_CAST")`