StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #21 from p2p-org/native_subscribe_sign

Browse Source 
added signature for native subscribe
This commit is contained in:
Vyacheslav Shebanov
2022-10-03 16:54:27 +03:00
committed by GitHub
parent 2978d5f604 b1482fbf5a
commit cf91b79609
57 changed files with 556 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt
View File

@@ -31,34 +31,40 @@ class BlockContainer(
json: ByteArray?,
val parsed: Any?,
val transactions: List<TxId> = emptyList(),
val nodeRating: Int = 0
val nodeRating: Int = 0,
val upstreamId: String = ""
) : SourceContainer(json, parsed) {
companion object {
@JvmStatic
fun from(block: BlockJson<*>, raw: ByteArray): BlockContainer {
val hasTransactions = block.transactions?.filterIsInstance<TransactionJson>()?.count() ?: 0 > 0
fun from(block: BlockJson<*>, raw: ByteArray, upstreamId: String): BlockContainer {
val hasTransactions = !block.transactions?.filterIsInstance<TransactionJson>().isNullOrEmpty()
return BlockContainer(
block.number,
BlockId.from(block),
block.totalDifficulty,
block.timestamp,
hasTransactions,
raw,
block,
block.transactions?.map { TxId.from(it.hash) } ?: emptyList()
height = block.number,
hash = BlockId.from(block),
difficulty = block.totalDifficulty,
timestamp = block.timestamp,
full = hasTransactions,
json = raw,
parsed = block,
transactions = block.transactions?.map { TxId.from(it.hash) } ?: emptyList(),
upstreamId = upstreamId
)
}
@JvmStatic
fun from(block: BlockJson<*>): BlockContainer {
return from(block, Global.objectMapper.writeValueAsBytes(block))
return from(block, "unknown")
}
@JvmStatic
fun from(block: BlockJson<*>, upstream: String): BlockContainer {
return from(block, Global.objectMapper.writeValueAsBytes(block), upstream)
}
@JvmStatic
fun fromEthereumJson(raw: ByteArray): BlockContainer {
fun fromEthereumJson(raw: ByteArray, upstream: String): BlockContainer {
val block = Global.objectMapper.readValue(raw, BlockJson::class.java)
return from(block, raw)
return from(block, raw, upstream)
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -141,7 +141,7 @@ class QuorumRpcReader(
src.map {
val signature = response.providedSignature
?: if (key.nonce != null) {
signer?.sign(key.nonce, response.getResult(), api)
signer?.sign(key.nonce, response.getResult(), api.getId())
} else {
null
}

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -220,7 +220,7 @@ open class NativeCall(
.flatMap(JsonRpcResponse::requireResult)
.map {
if (ctx.nonce != null) {
CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream))
CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId()))
} else {
CallResult.ok(ctx.id, null, it, null)
}

90
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
View File

@@ -17,11 +17,14 @@ package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeReplyItem
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.HasUpstream
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import io.grpc.Status
@@ -35,7 +38,8 @@ import reactor.core.publisher.Mono
@Service
open class NativeSubscribe(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val signer: ResponseSigner
) {
companion object {
@@ -44,33 +48,46 @@ open class NativeSubscribe(
private val objectMapper = Global.objectMapper
fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<NativeSubscribeReplyItem> {
return request
.flatMapMany(this@NativeSubscribe::start)
.map(this@NativeSubscribe::convertToProto)
.onErrorMap(this@NativeSubscribe::convertToStatus)
}
fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<ResponseHolder> {
val chain = Chain.byId(request.chainValue)
if (BlockchainType.from(chain) != BlockchainType.ETHEREUM_POS && BlockchainType.from(chain) != BlockchainType.ETHEREUM) {
return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
}
val method = request.method
val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
objectMapper.readValue(it.newInput(), Map::class.java)
}
val nonce = request.nonce.takeIf { it != 0L }
val matcher = Selector.convertToMatcher(request.selector)
return subscribe(chain, method, params, matcher)
/**
* Try to proxy request subscription directly to the upstream dshackle instance.
* If not possible - performs subscription logic on the current instance
* @see EthereumLikeMultistream.tryProxy
*/
val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run {
val method = request.method
val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
objectMapper.readValue(it.newInput(), Map::class.java)
}
subscribe(chain, method, params, matcher)
}
return publisher.map { ResponseHolder(it, nonce) }
}
fun convertToStatus(t: Throwable) = when (t) {
is SilentException.UnsupportedBlockchain -> StatusException(
Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}")
)
is UnsupportedOperationException -> StatusException(
Status.UNIMPLEMENTED.withDescription(t.message)
)
else -> {
log.warn("Unhandled error", t)
StatusException(
@@ -79,17 +96,54 @@ open class NativeSubscribe(
}
}
open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
return (up as EthereumLikeMultistream)
.getSubscribe()
.subscribe(method, params, matcher)
open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
getUpstream(chain)?.getSubscribe()?.subscribe(method, params, matcher)
?: Flux.error(SilentException.UnsupportedBlockchain(chain))
private fun getUpstream(chain: Chain): EthereumLikeMultistream? =
multistreamHolder.getUpstream(chain)
?.let { it as EthereumLikeMultistream }
fun convertToProto(holder: ResponseHolder): NativeSubscribeReplyItem {
if (holder.response is NativeSubscribeReplyItem) {
return holder.response
}
val result = objectMapper.writeValueAsBytes(holder.response)
val builder = NativeSubscribeReplyItem.newBuilder()
.setPayload(ByteString.copyFrom(result))
holder.nonce?.also { nonce ->
holder.getSource()?.let {
signer.sign(nonce, result, it)
}?.let {
buildSignature(nonce, it)
}?.also {
builder.signature = it
}
}
return builder.build()
}
fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem {
val result = objectMapper.writeValueAsBytes(value)
return BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder()
.setPayload(ByteString.copyFrom(result))
.build()
fun buildSignature(
nonce: Long,
signature: ResponseSigner.Signature
): BlockchainOuterClass.NativeCallReplySignature {
val msg = BlockchainOuterClass.NativeCallReplySignature.newBuilder()
msg.signature = ByteString.copyFrom(signature.value)
msg.keyId = signature.keyId
msg.upstreamId = signature.upstreamId
msg.nonce = nonce
return msg.build()
}
data class ResponseHolder(
val response: Any,
val nonce: Long?
) {
fun getSource(): String? =
if (response is HasUpstream) {
response.upstreamId.takeIf { it != "unknown" }
} else null
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.upstream
import com.google.common.annotations.VisibleForTesting
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
@@ -61,4 +62,8 @@ class MergedHead(
}
}
}
@VisibleForTesting
private fun getSources() =
sources
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
View File

@@ -29,6 +29,7 @@ import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
open class DefaultEthereumHead(
private val upstreamId: String,
forkChoice: ForkChoice,
blockValidator: BlockValidator
) : Head, AbstractHead(forkChoice, blockValidator) {
@@ -57,7 +58,7 @@ open class DefaultEthereumHead(
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
}
.map {
BlockContainer.fromEthereumJson(it.getResult())
BlockContainer.fromEthereumJson(it.getResult(), upstreamId)
}
.onErrorResume { err ->
log.debug("Failed to fetch latest block: ${err.message}")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
View File

@@ -119,7 +119,7 @@ class EthereumDirectReader(
if (block == null) {
Mono.empty<BlockContainer>()
} else {
Mono.just(BlockContainer.from(block, blockbytes))
Mono.just(BlockContainer.from(block, blockbytes, "unknown"))
}
}
.doOnNext { block ->

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
View File

@@ -1,12 +1,23 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import reactor.core.publisher.Flux
interface EthereumLikeMultistream : Upstream {
fun getReader(): EthereumReader
fun getSubscribe(): EthereumSubscribe
fun getHead(mather: Selector.Matcher): Head
/**
* Tries to proxy the native subscribe request to the managed upstreams if
* - any of them matches the matcher criteria
* - all of matching above are gRPC ones
* in this case the upstream dshackle instances can sign the results and they will just proxied as is with original signs
* Otherwise return null
*/
fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
}

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
@@ -27,12 +28,14 @@ import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
@@ -93,6 +96,22 @@ open class EthereumMultistream(
return head!!
}
override fun tryProxy(
matcher: Selector.Matcher,
request: BlockchainOuterClass.NativeSubscribeRequest
): Flux<out Any>? =
upstreams.filter {
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {
it.getBlockchainApi().nativeSubscribe(request)
}?.let {
Flux.merge(it)
}
override fun setHead(head: Head) {
this.head = head
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -33,9 +33,10 @@ import java.util.concurrent.Executors
class EthereumRpcHead(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
forkChoice: ForkChoice,
upstreamId: String,
blockValidator: BlockValidator,
private val interval: Duration = Duration.ofSeconds(10),
) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle {
) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
companion object {
val scheduler =

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt
View File

@@ -62,7 +62,7 @@ class EthereumWsFactory(
require(upstream == null || upstream.getId() == id) {
"Creating instance for different upstream. ${upstream?.getId()} != id"
}
return WsConnection(uri, origin, basicAuth, metrics, upstream, validator).also { ws ->
return WsConnection(id, uri, origin, basicAuth, metrics, upstream, validator).also { ws ->
config?.frameSize?.let {
ws.frameSize = it
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -26,9 +26,10 @@ import reactor.core.publisher.Flux
class EthereumWsHead(
private val ws: WsConnection,
upstreamId: String,
forkChoice: ForkChoice,
blockValidator: BlockValidator
) : DefaultEthereumHead(forkChoice, blockValidator), Lifecycle {
) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt
View File

@@ -65,6 +65,7 @@ import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
open class WsConnection(
private val id: String,
private val uri: URI,
private val origin: URI,
private val basicAuth: AuthConfig.ClientBasicAuth?,
@@ -352,7 +353,7 @@ open class WsConnection(
}
}
.flatMap(JsonRpcResponse::requireResult)
.map { BlockContainer.fromEthereumJson(it) }
.map { BlockContainer.fromEthereumJson(it, id) }
.subscribeOn(Schedulers.boundedElastic())
.timeout(Defaults.timeoutInternal, Mono.empty())
}.repeatWhenEmpty { n ->
@@ -368,7 +369,7 @@ open class WsConnection(
.then()
} else {
Mono.fromCallable {
blocks.tryEmitNext(BlockContainer.from(block))
blocks.tryEmitNext(BlockContainer.from(block, id))
}.then()
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
View File

@@ -16,7 +16,9 @@ open class EthereumConnectorFactory(
private val forkChoice: ForkChoice,
private val blockValidator: BlockValidator
) : ConnectorFactory {
private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java)
companion object {
private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java)
}
override fun isValid(): Boolean {
if (preferHttp && httpFactory == null) {
@@ -27,7 +29,7 @@ open class EthereumConnectorFactory(
override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector {
if (wsFactory != null && !preferHttp) {
return EthereumWsConnector(wsFactory, upstream, validator, chain, forkChoice, blockValidator)
return EthereumWsConnector(wsFactory, upstream, validator, forkChoice, blockValidator)
}
if (httpFactory == null) {
throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set")

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
View File

@@ -35,14 +35,14 @@ class EthereumRpcConnector(
if (wsFactory != null) {
// do not set upstream to the WS, since it doesn't control the RPC upstream
conn = wsFactory.create(null, null)
val wsHead = EthereumWsHead(conn, forkChoice, blockValidator)
val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator)
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(directReader, forkChoice, blockValidator, Duration.ofSeconds(60))
val rpcHead = EthereumRpcHead(directReader, forkChoice, id, blockValidator, Duration.ofSeconds(60))
head = MergedHead(listOf(rpcHead, wsHead), forkChoice)
} else {
conn = null
log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
head = EthereumRpcHead(directReader, forkChoice, blockValidator)
head = EthereumRpcHead(directReader, forkChoice, id, blockValidator)
}
}

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -4,18 +4,19 @@ import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.*
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
import io.emeraldpay.grpc.Chain
class EthereumWsConnector(
wsFactory: EthereumWsFactory,
upstream: DefaultUpstream,
validator: EthereumUpstreamValidator,
chain: Chain,
forkChoice: ForkChoice,
blockValidator: BlockValidator
) : EthereumConnector {
@@ -25,7 +26,7 @@ class EthereumWsConnector(
init {
conn = wsFactory.create(upstream, validator)
head = EthereumWsHead(conn, forkChoice, blockValidator)
head = EthereumWsHead(conn, upstream.getId(), forkChoice, blockValidator)
api = JsonRpcWsClient(conn)
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt
View File

@@ -71,7 +71,7 @@ class ConnectBlockUpdates(
val prev = findPrevious(block)
remember(block)
val removed = if (prev != null) {
whenReplaced(prev)
whenReplaced(prev, block.upstreamId)
} else {
Flux.empty()
}
@@ -103,13 +103,14 @@ class ConnectBlockUpdates(
/**
* Produce updates for transactions when a block is replaces with a different one on the same height.
*/
fun whenReplaced(prev: BlockContainer): Flux<Update> {
fun whenReplaced(prev: BlockContainer, source: String): Flux<Update> {
return Flux.fromIterable(prev.transactions).map {
Update(
prev.hash,
prev.height,
UpdateType.DROP,
it
it,
source
)
}
}
@@ -121,7 +122,8 @@ class ConnectBlockUpdates(
block.hash,
block.height,
UpdateType.NEW,
it
it,
block.upstreamId
)
}
}
@@ -131,6 +133,7 @@ class ConnectBlockUpdates(
val blockNumber: Long,
val type: UpdateType,
val transactionId: TxId,
val upstreamId: String
)
enum class UpdateType {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt
View File

@@ -94,7 +94,9 @@ class ProduceLogs(
txlog.topics,
txlog.transactionHash,
txlog.transactionIndex,
false
false,
update.upstreamId
)
}
oldMessages.put(LogReference(update.blockHash, update.transactionId), messages)

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt
View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.ethereum.subscribe
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
import io.emeraldpay.etherjar.rpc.json.BlockJson
@@ -41,13 +42,7 @@ class ProduceNewHeads(
fun start(): Flux<NewHeadMessage> {
return head.getFlux()
.map {
if (it.parsed != null) {
it.parsed as BlockJson<TransactionRefJson>
} else {
objectMapper.readValue(it.json, BlockJson::class.java)
}
}
.map { block ->
val block = extractBlock(it)
NewHeadMessage(
block.number,
block.hash,
@@ -58,8 +53,16 @@ class ProduceNewHeads(
block.gasUsed,
block.logsBloom,
block.miner,
block.baseFeePerGas?.amount
block.baseFeePerGas?.amount,
it.upstreamId
)
}
}
private fun extractBlock(blockContainer: BlockContainer): BlockJson<out TransactionRefJson> =
if (blockContainer.parsed != null) {
blockContainer.parsed as BlockJson<TransactionRefJson>
} else {
objectMapper.readValue(blockContainer.json, BlockJson::class.java)
}
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/HasUpstream.kt Normal file
View File

@@ -0,0 +1,5 @@
package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json
interface HasUpstream {
val upstreamId: String
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json
import com.fasterxml.jackson.annotation.JsonIgnore
import com.fasterxml.jackson.databind.annotation.JsonSerialize
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.etherjar.domain.BlockHash
@@ -40,5 +41,7 @@ data class LogMessage(
val transactionHash: TransactionId,
@get:JsonSerialize(using = NumberAsHexSerializer::class)
val transactionIndex: Long,
val removed: Boolean
)
val removed: Boolean,
@get:JsonIgnore
override val upstreamId: String
) : HasUpstream

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json
import com.fasterxml.jackson.annotation.JsonIgnore
import com.fasterxml.jackson.annotation.JsonInclude
import com.fasterxml.jackson.databind.annotation.JsonSerialize
import io.emeraldpay.etherjar.domain.Address
@@ -50,5 +51,7 @@ data class NewHeadMessage(
val miner: Address,
@get:JsonSerialize(using = NumberAsHexSerializer::class)
@get:JsonInclude(JsonInclude.Include.NON_NULL)
val baseFeePerGas: BigInteger?
)
val baseFeePerGas: BigInteger?,
@get:JsonIgnore
override val upstreamId: String
) : HasUpstream

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
@@ -27,12 +28,14 @@ import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
@@ -88,6 +91,19 @@ open class EthereumPosMultiStream(
return head!!
}
override fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
upstreams.filter {
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {
it.proxySubscribe(request)
}?.let {
Flux.merge(it)
}
override fun setHead(head: Head) {
this.head = head
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -38,6 +38,7 @@ import io.emeraldpay.grpc.Chain
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.time.Instant
@@ -105,6 +106,9 @@ class BitcoinGrpcUpstream(
return remote
}
override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any> =
remote.nativeSubscribe(request)
override fun getHead(): Head {
return grpcHead
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -41,6 +41,7 @@ import io.emeraldpay.grpc.Chain
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.time.Instant
@@ -84,7 +85,7 @@ open class EthereumGrpcUpstream(
defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false)))
.flatMap(JsonRpcResponse::requireResult)
.map {
BlockContainer.fromEthereumJson(it)
BlockContainer.fromEthereumJson(it, getId())
}
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
.doOnError { t ->
@@ -110,6 +111,9 @@ open class EthereumGrpcUpstream(
return remote
}
override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any> =
remote.nativeSubscribe(request)
override fun start() {
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
View File

@@ -41,6 +41,7 @@ import io.emeraldpay.grpc.Chain
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.time.Instant
@@ -84,7 +85,7 @@ open class EthereumPosGrpcUpstream(
defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false)))
.flatMap(JsonRpcResponse::requireResult)
.map {
BlockContainer.fromEthereumJson(it)
BlockContainer.fromEthereumJson(it, getId())
}
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
.doOnError { t ->
@@ -130,6 +131,9 @@ open class EthereumPosGrpcUpstream(
return remote
}
override fun proxySubscribe(request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any> =
remote.nativeSubscribe(request)
// ------------------------------------------------------------------------------------------
override fun getLabels(): Collection<UpstreamsConfig.Labels> {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
View File

@@ -16,8 +16,10 @@
package io.emeraldpay.dshackle.upstream.grpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeRequest
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.upstream.Upstream
import reactor.core.publisher.Flux
interface GrpcUpstream : Upstream {
@@ -28,4 +30,6 @@ interface GrpcUpstream : Upstream {
fun update(conf: BlockchainOuterClass.DescribeChain)
fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub
fun proxySubscribe(request: NativeSubscribeRequest): Flux<out Any>
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/EcdsaSigner.kt
View File

@@ -1,6 +1,5 @@
package io.emeraldpay.dshackle.upstream.signature
import io.emeraldpay.dshackle.upstream.Upstream
import org.apache.commons.codec.binary.Hex
import java.security.MessageDigest
import java.security.Signature
@@ -17,15 +16,13 @@ class EcdsaSigner(
const val MSG_SEPARATOR = '/'
}
override fun sign(nonce: Long, message: ByteArray, source: Upstream): ResponseSigner.Signature {
override fun sign(nonce: Long, message: ByteArray, source: String): ResponseSigner.Signature {
val sig = Signature.getInstance(SIGN_SCHEME)
sig.initSign(privateKey)
val wrapped = wrapMessage(nonce, message, source)
sig.update(wrapped.toByteArray())
val value = sig.sign()
return ResponseSigner.Signature(
value, source.getId(), keyId
)
return ResponseSigner.Signature(value, source, keyId)
}
/**
@@ -42,7 +39,7 @@ class EcdsaSigner(
* - second is the nonce value encode as decimal string
* - third is SHA256 hash of the original message encoded as hex string
*/
fun wrapMessage(nonce: Long, message: ByteArray, source: Upstream): String {
fun wrapMessage(nonce: Long, message: ByteArray, source: String): String {
val sha256 = MessageDigest.getInstance("SHA-256")
// we create it with max capacity that we expect for the result, which is total lengths of its parts
val formatterMsg = StringBuilder(11 + 1 + 18 + 1 + 64 + 1 + 64)
@@ -52,7 +49,7 @@ class EcdsaSigner(
.append(MSG_SEPARATOR)
// We expect that the id is short enough (less than 64 symbols) and also it doesn't contain the `/` symbol
// which is verified in UpstreamConfigReader and DefaultUpstream constructor
.append(source.getId())
.append(source)
.append(MSG_SEPARATOR)
.append(Hex.encodeHexString(sha256.digest(message)))
return formatterMsg.toString()

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/NoSigner.kt
View File

@@ -1,9 +1,7 @@
package io.emeraldpay.dshackle.upstream.signature
import io.emeraldpay.dshackle.upstream.Upstream
class NoSigner : ResponseSigner {
override fun sign(nonce: Long, message: ByteArray, source: Upstream): ResponseSigner.Signature? {
override fun sign(nonce: Long, message: ByteArray, source: String): ResponseSigner.Signature? {
return null
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSigner.kt
View File

@@ -1,10 +1,8 @@
package io.emeraldpay.dshackle.upstream.signature
import io.emeraldpay.dshackle.upstream.Upstream
interface ResponseSigner {
fun sign(nonce: Long, message: ByteArray, source: Upstream): Signature?
fun sign(nonce: Long, message: ByteArray, source: String): Signature?
data class Signature(
val value: ByteArray,

2
src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy vendored
View File

@@ -62,7 +62,7 @@ class BlocksRedisCacheSpec extends Specification {
false,
"test".bytes,
null,
[TxId.from(hash2), TxId.from(hash1)], 0
[TxId.from(hash2), TxId.from(hash1)], 0, "BlocksRedisCacheSpec"
)
when:

2
src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy vendored
View File

@@ -27,7 +27,7 @@ class HeightByHashAddingSpec extends Specification {
def block = new BlockContainer(
12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"),
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0, "upstream"
)
def "use memory if available"() {

4
src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashRedisCacheSpec.groovy vendored
View File

@@ -30,11 +30,11 @@ class HeightByHashRedisCacheSpec extends Specification {
def block1 = new BlockContainer(
12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"),
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0, "HeightByHashRedisCacheSpec"
)
def block2 = new BlockContainer(
12079193L, BlockId.from("0xd27944b460632699768fbfec3e5d454db590cae43d470b5f42fc4d091e372c25"),
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0
BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0, "HeightByHashRedisCacheSpec"
)
StatefulRedisConnection<String, byte[]> redis

3
src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy vendored
View File

@@ -87,7 +87,8 @@ class ReceiptMemCacheSpec extends Specification {
"{}".bytes,
null,
[TxId.from(receipt.transactionHash)],
0
0,
"unknown"
)
when:

71
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
View File

@@ -19,9 +19,9 @@ import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.test.MultistreamHolderMock
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
import io.emeraldpay.dshackle.upstream.signature.NoSigner
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
@@ -30,33 +30,46 @@ import spock.lang.Specification
import java.time.Duration
class NativeSubscribeSpec extends Specification {
def signer = new NoSigner()
def "Call with empty params when not provided"() {
setup:
def subscribe = Mock(EthereumSubscribe) {
1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
}
def up = Mock(EthereumPosMultiStream) {
1 * it.getSubscribe() >> subscribe
}
def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up))
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def subscribe = Mock(EthereumSubscribe) {
1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
}
def up = Mock(EthereumPosMultiStream) {
1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
1 * it.getSubscribe() >> subscribe
}
def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
when:
def act = nativeSubscribe.start(call)
then:
StepVerifier.create(act)
.expectNext("{}")
.expectNext(new NativeSubscribe.ResponseHolder("{}", null))
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "Call with params when provided"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("logs")
.setPayload(ByteString.copyFromUtf8(
'{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' +
'"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}'
))
.build()
def subscribe = Mock(EthereumSubscribe) {
1 * it.subscribe("logs", { params ->
println("params: $params")
@@ -69,25 +82,43 @@ class NativeSubscribeSpec extends Specification {
}, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
}
def up = Mock(EthereumPosMultiStream) {
1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
1 * it.getSubscribe() >> subscribe
}
def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up))
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("logs")
.setPayload(ByteString.copyFromUtf8(
'{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' +
'"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}'
))
.build()
def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
when:
def act = nativeSubscribe.start(call)
then:
StepVerifier.create(act)
.expectNext("{}")
.expectNext(new NativeSubscribe.ResponseHolder("{}", null))
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "Proxy call"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def up = Mock(EthereumPosMultiStream) {
1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> Flux.just("{}")
0 * it.getSubscribe()
}
def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
when:
def act = nativeSubscribe.start(call)
then:
StepVerifier.create(act)
.expectNext(new NativeSubscribe.ResponseHolder("{}", null))
.expectComplete()
.verify(Duration.ofSeconds(1))
}
}

4
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy
View File

@@ -271,7 +271,7 @@ class TrackBitcoinAddressSpec extends Specification {
Head head = Mock(Head) {
1 * getFlux() >> Flux.concat(
Flux.just(
new BlockContainer(0L, BlockId.from(hash1), BigInteger.ZERO, Instant.now(), false, null, null, [], 0)
new BlockContainer(0L, BlockId.from(hash1), BigInteger.ZERO, Instant.now(), false, null, null, [], 0, "TrackBitcoinAddressSpec")
),
blocks.asFlux()
)
@@ -312,7 +312,7 @@ class TrackBitcoinAddressSpec extends Specification {
StepVerifier.create(resp)
.expectNext("0")
.then {
blocks.tryEmitNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [], 0))
blocks.tryEmitNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [], 0, "TrackBitcoinAddressSpec"))
}
.expectNext("1230000")
.then {

6
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy
View File

@@ -142,7 +142,7 @@ class TrackBitcoinTxSpec extends Specification {
def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
// start with the current block
def next = Flux.fromIterable([10, 12, 13, 14, 15]).map { h ->
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0)
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0, "unknown")
}
Head head = Mock(Head) {
1 * getFlux() >> next
@@ -173,7 +173,7 @@ class TrackBitcoinTxSpec extends Specification {
def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9"
// start with the current block
def next = Flux.fromIterable([10, 12, 13]).map { h ->
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0)
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0, "unknown")
}
Head head = Mock(Head) {
1 * getFlux() >> next
@@ -268,7 +268,7 @@ class TrackBitcoinTxSpec extends Specification {
])
}
def next = Flux.fromIterable([10, 11, 12]).map { h ->
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0)
new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0, "unknown")
}
Head head = Mock(Head) {
_ * getFlux() >> next

3
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
View File

@@ -165,7 +165,8 @@ class TrackERC20AddressSpec extends Specification {
],
TransactionId.from("0x5a7898e27120575c33d3d0179af3b6353c7268bbad4255df079ed26b743a21a5"),
1,
false
false,
"unknown"
)
]
def logs = Mock(ConnectLogs) {

4
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy
View File

@@ -199,7 +199,7 @@ class TrackEthereumTxSpec extends Specification {
def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6)
def block = new BlockContainer(
100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes, null,
[TxId.from(txId)], 0
[TxId.from(txId)], 0, "unknown"
)
when:
@@ -222,7 +222,7 @@ class TrackEthereumTxSpec extends Specification {
def block = new BlockContainer(
100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes, null,
[TxId.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22")],
0
0, "unknown"
)
apiMock.answer("eth_getTransactionByHash", [txId], null)

10
src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy
View File

@@ -52,6 +52,10 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
this(chain, api, allMethods())
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, Map<String, String> labels) {
this(id, chain, api, allMethods(), labels)
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api) {
this(id, chain, api, allMethods())
}
@@ -61,11 +65,15 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
this(id, chain, api, methods, Collections.<String, String>emptyMap())
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels) {
super(id, chain,
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.PRIMARY,
methods,
new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),
new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)),
new ConnectorFactoryMock(api, new EthereumHeadMock()))
this.ethereumHeadMock = this.getHead() as EthereumHeadMock
setLag(0)

13
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -16,7 +16,6 @@
*/
package io.emeraldpay.dshackle.test
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesFactory
@@ -27,15 +26,12 @@ import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.MeterRegistry
import io.micrometer.core.instrument.logging.LoggingMeterRegistry
import org.apache.commons.lang3.StringUtils
@@ -56,6 +52,10 @@ class TestingCommons {
return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api())
}
static EthereumPosRpcUpstreamMock upstream(String id, String provider) {
return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api(), Collections.singletonMap("provider", provider))
}
static EthereumPosRpcUpstreamMock upstream(String id, Reader<JsonRpcRequest, JsonRpcResponse> api) {
return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM, api)
}
@@ -114,7 +114,8 @@ class TestingCommons {
null,
null,
[],
0
0,
"upstream"
)
}

4
src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy
View File

@@ -34,7 +34,7 @@ class AbstractHeadSpec extends Specification {
def blocks = [1L, 2, 3, 4].collect { i ->
byte[] hash = new byte[32]
hash[0] = i as byte
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0)
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0, "AbstractHeadSpec")
}
def "Calls beforeBlock on each block"() {
@@ -96,7 +96,7 @@ class AbstractHeadSpec extends Specification {
blocks[1].height, BlockId.from(blocks[1].hash.value.clone().tap { it[1] = 0xff as byte }),
blocks[1].difficulty - 1,
Instant.now(),
false, null, null, [], 0
false, null, null, [], 0, "AbstractHeadSpec"
)
when:
head.follow(source.asFlux())

206
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -16,25 +16,33 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock
import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
import io.emeraldpay.dshackle.upstream.grpc.EthereumPosGrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import io.emeraldpay.grpc.Chain
import org.jetbrains.annotations.NotNull
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
import spock.lang.Specification
import java.time.Duration
import java.time.Instant
import java.time.temporal.ChronoUnit
class MultistreamSpec extends Specification {
@@ -193,6 +201,147 @@ class MultistreamSpec extends Specification {
1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes)
}
def "Filter upstream matching selector single"() {
setup:
def up1 = TestingCommons.upstream("test-1", "internal")
def up2 = TestingCommons.upstream("test-2", "external")
def up3 = TestingCommons.upstream("test-3", "external")
def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default())
expect:
multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock)
multistream.getHead(new Selector.LabelMatcher("provider", ["unknown"])) in EmptyHead
def head = multistream.getHead(new Selector.LabelMatcher("provider", ["external"]))
head in MergedHead
(head as MergedHead).isRunning()
(head as MergedHead).getSources().sort() == [up2.ethereumHeadMock, up3.ethereumHeadMock].sort()
}
def "Proxy gRPC request - select one"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def up1 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "internal"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
1 * proxySubscribe(call) >> Flux.just("{}")
}
def up2 = Mock(EthereumPosGrpcUpstream) {
1 * getId() >> "external"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "external"))]
}
def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default())
when:
def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call)
then:
StepVerifier.create(act)
.expectNext("{}")
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "Proxy gRPC request - not all gRPC"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def up1 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "1"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
}
def up2 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> false
1 * getId() >> "2"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
}
def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default())
when:
def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call)
then:
!act
}
def "Proxy gRPC request - select many"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def up1 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "1"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
1 * proxySubscribe(call) >> Flux.just("{1}")
}
def up2 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "2"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
1 * proxySubscribe(call) >> Flux.just("{2}")
}
def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default())
when:
def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call)
then:
StepVerifier.create(act)
.expectNextMatches { it as String in ["{1}", "{2}"] }
.expectNextMatches { it as String in ["{1}", "{2}"] }
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "Proxy gRPC request - select many"() {
setup:
def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
.setChainValue(Chain.ETHEREUM.id)
.setMethod("newHeads")
.build()
def up1 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "1"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
1 * proxySubscribe(call) >> Flux.just("{1}")
}
def up2 = Mock(EthereumPosGrpcUpstream) {
1 * isGrpc() >> true
1 * getId() >> "2"
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(Collections.singletonMap("provider", "internal"))]
1 * proxySubscribe(call) >> Flux.just("{2}")
}
def multiStream = new TestEthereumPosMultistream(Chain.ETHEREUM, [up1, up2], Caches.default())
when:
def act = multiStream.tryProxy(new Selector.LabelMatcher("provider", ["internal"]), call)
then:
StepVerifier.create(act)
.expectNextMatches { it as String in ["{1}", "{2}"] }
.expectNextMatches { it as String in ["{1}", "{2}"] }
.expectComplete()
.verify(Duration.ofSeconds(1))
}
class TestMultistream extends Multistream {
TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
@@ -233,4 +382,57 @@ class MultistreamSpec extends Specification {
return null
}
}
class TestEthereumPosMultistream extends EthereumPosMultiStream {
TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches)
}
@Override
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(@NotNull Selector.Matcher matcher) {
return null
}
@Override
Head updateHead() {
return null
}
@Override
void setHead(@NotNull Head head) {
}
@Override
Head getHead() {
return null
}
public <T extends Upstream> T cast(Class<T> selfType) {
return this
}
@Override
ChainFees getFeeEstimation() {
return null
}
@Override
void init() {
}
}
BlockContainer createBlock(long number) {
def block = new BlockJson<TransactionRefJson>()
block.number = number
block.hash = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000" + number)
block.totalDifficulty = BigInteger.ONE
block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
block.uncles = []
block.transactions = []
return BlockContainer.from(block)
}
}

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy
View File

@@ -32,7 +32,7 @@ import java.time.Instant
class DefaultEthereumHeadSpec extends Specification {
DefaultEthereumHead head = new DefaultEthereumHead(new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
DefaultEthereumHead head = new DefaultEthereumHead("upstream", new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
ObjectMapper objectMapper = Global.objectMapper
def blocks = (10L..20L).collect { i ->

4
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockValidatorSpec.groovy
View File

@@ -52,6 +52,8 @@ class EthereumBlockValidatorSpec extends Specification {
true, bytes,
block,
Collections.emptyList(),
1)
1,
"upstream"
)
}
}

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy
View File

@@ -302,7 +302,7 @@ class EthereumFullBlocksReaderSpec extends Specification {
"extraField2": "extraValue2"
}
'''
blocks.add(BlockContainer.fromEthereumJson(blockJson.bytes))
blocks.add(BlockContainer.fromEthereumJson(blockJson.bytes, "EthereumFullBlocksReaderSpec"))
def tx1 = '''
{

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy
View File

@@ -81,7 +81,7 @@ class ConnectBlockUpdatesSpec extends Specification {
]
})
when:
def act = connectBlockUpdates.whenReplaced(block)
def act = connectBlockUpdates.whenReplaced(block, "ConnectBlockUpdatesSpec")
.collectList().block(Duration.ofSeconds(3))
then:

12
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy
View File

@@ -40,7 +40,8 @@ class ConnectLogsSpec extends Specification {
],
TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
1L,
false
false,
"upstream"
)
def log2 = new LogMessage(
@@ -54,7 +55,8 @@ class ConnectLogsSpec extends Specification {
],
TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
1L,
false
false,
"upstream"
)
def log3 = new LogMessage(
@@ -68,7 +70,8 @@ class ConnectLogsSpec extends Specification {
],
TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
1L,
false
false,
"upstream"
)
def log4 = new LogMessage(
@@ -82,7 +85,8 @@ class ConnectLogsSpec extends Specification {
],
TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
1L,
false
false,
"upstream"
)
def "Filter is empty"() {

21
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy
View File

@@ -45,7 +45,8 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"),
"upstream"
)
when:
def act = producer.produceAdded(update)
@@ -67,7 +68,8 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"),
"upstream"
)
when:
def act = producer.produceAdded(update)
@@ -112,7 +114,8 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"),
"upstream"
)
when:
def act = producer.produceAdded(update)
@@ -157,7 +160,8 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")
TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af"),
"upstream"
)
when:
def act = producer.produceAdded(update)
@@ -267,7 +271,8 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
"upstream"
)
when:
def act = producer.produceAdded(update)
@@ -345,13 +350,15 @@ class ProduceLogsSpec extends Specification {
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.NEW,
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
"upstream"
)
def update2 = new ConnectBlockUpdates.Update(
BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"),
13412871,
ConnectBlockUpdates.UpdateType.DROP,
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")
TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
"upstream"
)
when:
// first need to produce them as added, because that's when it remembers logs to "remove"

3
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessageSpec.groovy
View File

@@ -41,7 +41,8 @@ class LogMessageSpec extends Specification {
],
TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1"),
0xa3,
false)
false,
"LogMessageSpec")
ObjectMapper objectMapper = Global.getObjectMapper()
def exp = '{' +
'"address":"0x011b6e24ffb0b5f5fcc564cf4183c5bbbc96d515",' +

3
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessageSpec.groovy
View File

@@ -23,7 +23,8 @@ class NewHeadMessageSpec extends Specification {
0x7bb33e,
Bloom.from("0x012040020880820356a20b8e980a2004c19f1291800501040001180bb029d0002d8c49440e002048ca00d48581000d900a458100c90139056140880582f22a0a8050224c020c233be8c3080c0a016aa4226a2001446c800822080445a2454118139804001202068401841900840c484222420c4b2022046052c0011e81a9e450085883708545810592e40040010411442300080b0130711f602880600a30c90702cb420a0102a644820650908802840810948142541404884300acc69d000840702020224c000020200880c10858418408098a61445b0ab0480234862655a5000434311b91044849c165040411aa0400b00008222642d24313020d9022219120"),
Address.from("0x829bd824b016326a401d083b33d092293333a830"),
null
null,
"NewHeadMessageSpec"
)
ObjectMapper objectMapper = Global.getObjectMapper()
def exp = '{' +

2
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy
View File

@@ -11,7 +11,7 @@ class MostWorkForkChoiceSpec extends Specification {
def blocks = [1L, 2, 3, 4].collect { i ->
byte[] hash = new byte[32]
hash[0] = i as byte
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0)
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0, "MostWorkForkChoiceSpec")
}
def "filters blocks"() {

2
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy
View File

@@ -10,7 +10,7 @@ class NoChoiceWithPriorityForkChoiceSpec extends Specification {
def blocks = [1L, 2, 3, 4].collect { i ->
byte[] hash = new byte[32]
hash[0] = i as byte
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0)
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0, "NoChoiceWithPriorityForkChoiceSpec")
}
def "filters blocks"() {

2
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy
View File

@@ -10,7 +10,7 @@ class PriorityForkChoiceSpec extends Specification {
def blocks = [1L, 2, 3, 4].collect { i ->
byte[] hash = new byte[32]
hash[0] = i as byte
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], i.toInteger())
new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], i.toInteger(), "PriorityForkChoiceSpec")
}
def "filters blocks"() {
def choice = new PriorityForkChoice()

6
src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy
View File

@@ -95,7 +95,7 @@ class EcdsaSignerSpec extends Specification {
def signer = new EcdsaSigner(Stub(ECPrivateKey), 100L)
when:
def act = signer.wrapMessage(10, "test".bytes, up)
def act = signer.wrapMessage(10, "test".bytes, up.id)
then:
act == "DSHACKLESIG/10/infura/9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08"
@@ -118,7 +118,7 @@ class EcdsaSignerSpec extends Specification {
def signer = new EcdsaSigner((pair.getPrivate() as ECPrivateKey), 100L)
when:
def sig = signer.sign(10, result, up)
def sig = signer.sign(10, result, up.id)
then:
verifier.verify(sig.value)
@@ -148,7 +148,7 @@ class EcdsaSignerSpec extends Specification {
def signer = factory.getObject() as EcdsaSigner
when:
def sig = signer.sign(10, result, up)
def sig = signer.sign(10, result, up.id)
println("Signature: ${Hex.encodeHexString(sig.value)}")
then: