From e1170dd055a6312ff2cb7e55cec28d52c2997301 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 24 Nov 2021 19:25:43 -0500 Subject: [PATCH] problem: ERC20 Balancer subsription checks balancer on each block, even if no transfer happened solution: use ERC20 Logs to determine changes to ERC20 balance --- .../dshackle/rpc/TrackERC20Address.kt | 64 +++++---- .../upstream/ethereum/ERC20Balance.kt | 66 +++++++++ .../upstream/ethereum/EthereumSubscribe.kt | 2 +- .../ethereum/subscribe/ConnectLogs.kt | 4 +- .../dshackle/rpc/TrackERC20AddressSpec.groovy | 132 +++++++++++++----- .../upstream/ethereum/ERC20BalanceSpec.groovy | 58 ++++++++ 6 files changed, 264 insertions(+), 62 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt index 2a8d0430..5a968c44 100644 --- 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 = 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 { 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 new file mode 100644 index 00000000..496bfc93 --- /dev/null +++ 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 { + 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 { + 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 index 346a4f1c..69487b1e 100644 --- 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 index d1bda8e7..1309e4c2 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 @@ -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
, topics: List): Flux { + open fun start(addresses: List
, topics: List): Flux { // 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 index 919c3676..062820ed 100644 --- 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 new file mode 100644 index 00000000..27ecdf7d --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy @@ -0,0 +1,58 @@ +/** + * 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.test.EthereumUpstreamMock +import io.emeraldpay.dshackle.test.ReaderMock +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.HexData +import io.emeraldpay.etherjar.rpc.json.TransactionCallJson +import io.emeraldpay.grpc.Chain +import spock.lang.Specification + +import java.time.Duration + +class ERC20BalanceSpec extends Specification { + + 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) + ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")) + ERC20Balance query = new ERC20Balance() + + when: + def act = query.getBalance(upstream, token, Address.from("0x16c15c65ad00b6dfbcc2cb8a7b6c2d0103a3883b")) + .block(Duration.ofSeconds(1)) + + then: + act.toLong() == 0x1f28d72868 + } +}